#!/usr/bin/env python3
"""Daily datalake profile snapshot (Ahel).

Runs on M1 where the arkenstone-pf daemon keeps a loopback port-forward to the
Arkenstone OpenSearch (127.0.0.1:9200) — this avoids the in-cluster
NetworkPolicy that isolates the arkenstone data tier from other namespaces.

Writes one doc per index per day into `ahel-profiles`, giving growth history
(docs/bytes over time), new-corpus detection, and staleness signals. Metadata
only — never document contents.
"""
from __future__ import annotations

import datetime
import json
import urllib.error
import urllib.request

OS_URL = "http://127.0.0.1:9200"
PROFILE_INDEX = "ahel-profiles"
TIMEOUT = 120


def req(method: str, path: str, body=None, ndjson: bool = False):
    data = None
    ctype = "application/json"
    if ndjson:
        data = body.encode()
        ctype = "application/x-ndjson"
    elif body is not None:
        data = json.dumps(body).encode()
    request = urllib.request.Request(
        OS_URL + path, data=data, method=method, headers={"Content-Type": ctype}
    )
    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
    with opener.open(request, timeout=TIMEOUT) as resp:
        return json.loads(resp.read())


def ensure_index() -> None:
    try:
        req("PUT", "/" + PROFILE_INDEX, {
            "settings": {"number_of_replicas": 0},
            "mappings": {"properties": {
                "index": {"type": "keyword"},
                "docsCount": {"type": "long"},
                "sizeBytes": {"type": "long"},
                "health": {"type": "keyword"},
                "date": {"type": "keyword"},
                "ts": {"type": "date"},
            }},
        })
        print("created", PROFILE_INDEX)
    except urllib.error.HTTPError as exc:
        if exc.code != 400:  # 400 = already exists
            raise
        print(PROFILE_INDEX, "exists")


def main() -> int:
    ensure_index()
    now = datetime.datetime.now(datetime.timezone.utc)
    stamp, day = now.isoformat(), now.strftime("%Y-%m-%d")

    indices = req("GET", "/_cat/indices?format=json&bytes=b")
    lines: list[str] = []
    count = 0
    for item in indices:
        name = str(item.get("index", ""))
        if name.startswith(".") or name == PROFILE_INDEX:
            continue
        lines.append(json.dumps({"index": {"_index": PROFILE_INDEX, "_id": f"{name}@{day}"}}))
        lines.append(json.dumps({
            "index": name,
            "docsCount": int(item.get("docs.count") or 0),
            "sizeBytes": int(item.get("store.size") or 0),
            "health": item.get("health"),
            "date": day,
            "ts": stamp,
        }))
        count += 1

    if not lines:
        print("no data indices found")
        return 1
    result = req("POST", "/_bulk", "\n".join(lines) + "\n", ndjson=True)
    print(f"profiled {count} indices; bulk errors: {result.get('errors')}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
