#!/usr/bin/env python3
"""NPCI UPI Ecosystem Statistics fetcher.

Source: /product/ecosystem-statistics/upi -> /api/ecosystem-statistics/get-statistics
Tabs: upi-apps, chargeback, top50-member (remitter/beneficiary), top-15-psps
(payer/payee), mcc, p2p-and-p2m-transactions, statewise-statistic,
top-50-mem-vol-val. Months walked backward; 3 consecutive empty months stops.
Raw rows stored as JSON; manifest appended to raw/npci/manifest.jsonl.
"""
import asyncio, json, sys, time
from datetime import date
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
from npci_fetch import NpciSession  # reuse session/api helper
from playwright.async_api import async_playwright

ROOT = Path("/home/workspace/Projects/payments-stat-hub")
RAW = ROOT / "raw" / "npci" / "eco"
MANIFEST = ROOT / "raw" / "npci" / "manifest.jsonl"

PRODUCT = "UPI"
API_PATH = "ecosystem-statistics/get-statistics"
SLEEP = 0.6
STOP_EMPTY = 3

MONTHS_ABBR = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
MONTHS = ["January","February","March","April","May","June","July",
          "August","September","October","November","December"]

# tab -> (list of type_names or [None], page_size)
TABS = {
    "upi-apps":                  ([None], 200),
    "chargeback":                ([None], 200),
    "top50-member":              (["remitter", "beneficiary"], 200),
    "top-15-psps":               (["payer", "payee"], 100),
    "mcc":                       ([None], 100),
    "p2p-and-p2m-transactions":  ([None], 10),
    "statewise-statistic":       ([None], 1000),
    "top-50-mem-vol-val":        ([None], 100),
}

def month_iter(start_y, start_m):
    y, m = start_y, start_m
    while True:
        yield y, m
        m -= 1
        if m == 0:
            y, m = y - 1, 12

def log(msg):
    print(time.strftime("%H:%M:%S"), msg, flush=True)

async def eco_api(sess, params):
    import json as _json
    query = "&".join(f"{k}={v}" for k, v in params.items() if v is not None)
    url = f"https://www.npci.org.in/api/ecosystem-statistics/get-statistics?{query}"
    expr = ("(async () => { try { const r = await fetch(" + _json.dumps(url) +
            ", {headers: {'Accept': 'application/json'}});"
            " return [r.status, await r.text()]; }"
            " catch (e) { return [-1, String(e)]; } })()")
    try:
        result = await sess.page.evaluate(expr)
    except Exception as e:
        return -1, f"evaluate-failed: {e}"
    return result


async def fetch_tab(sess, tab, type_name, start_y, start_m, seen):
    counts = {"ok": 0, "empty": 0, "fail": 0}
    consec = 0
    all_rows = None  # session-cached cumulative rows for this tab
    tdir = RAW / tab / (type_name or "_") 
    tdir.mkdir(parents=True, exist_ok=True)
    for y, m in month_iter(start_y, start_m):
        key = f"{y}-{MONTHS[m-1]}"
        fname = tdir / f"{y}-{m:02d}.json"
        if fname.exists():
            counts["ok"] += 1
            consec = 0
            continue
        params = {"product_name": PRODUCT, "tab_name": tab, "year": y,
                  "month": MONTHS_ABBR[m-1], "page_no": 1, "sort_by": "asc",
                  "size": TABS[tab][1], "locale": "en"}
        if type_name:
            params["type_name"] = type_name
        status, text = await eco_api(sess, params)
        rows, tc = [], None
        if status == 200:
            try:
                body = json.loads(text)
                d = body.get("data") or {}
                rows = d.get("results") or []
                tc = d.get("totalCount")
            except json.JSONDecodeError:
                status = -2
        if status == 200 and rows and rows != all_rows:
            all_rows = rows
            fname.write_text(json.dumps(body, ensure_ascii=False))
            counts["ok"] += 1
            consec = 0
            with MANIFEST.open("a") as f:
                f.write(json.dumps({"url": f"{API_PATH} {tab}/{type_name}",
                                    "status": status, "rows": len(rows),
                                    "path": str(fname.relative_to(ROOT)),
                                    "ts": time.strftime("%Y-%m-%dT%H:%M:%S")}) + "\n")
            if tc is not None and tab == "statewise-statistic" and tc != len(rows):
                log(f"    note: statewise {key} tc={tc} rows={len(rows)}")
            log(f"  {tab}/{type_name or '-'} {key}: {len(rows)} rows")
        elif status == 200:
            counts["empty"] += 1
            consec += 1
            log(f"  {tab}/{type_name or '-'} {key}: empty (streak {consec})")
            if consec >= STOP_EMPTY:
                log(f"  {tab}/{type_name or '-'}: stop after {STOP_EMPTY} empties")
                break
        else:
            counts["fail"] += 1
            log(f"  {tab}/{type_name or '-'} {key}: status={status}")
            if status == 404:
                break
        await asyncio.sleep(SLEEP)
    return counts

async def main():
    months = None
    args = sys.argv[1:]
    if "--months" in args:
        i = args.index("--months")
        months = [a for a in args[i+1:] if not a.startswith("--")]
        args = args[:i] + args[i+1+len(months):]
    only = None
    if "--tabs" in args:
        i = args.index("--tabs")
        only = args[i+1].split(",")
    start = (2026, 8)
    if months:
        yy, mm = months[0].split("-")
        start = (int(yy), int(mm))
    async with async_playwright() as pw:
        browser = await pw.chromium.launch(headless=True)
        sess = NpciSession(browser)
        await sess.start()
        log("session up")
        total = {"ok": 0, "empty": 0, "fail": 0}
        for tab, (types, _) in TABS.items():
            if only and tab not in only:
                continue
            for tn in types:
                c = await fetch_tab(sess, tab, tn, start[0], start[1], seen=set())
                for k in total:
                    total[k] += c[k]
        await browser.close()
        log(f"DONE ok={total['ok']} empty={total['empty']} fail={total['fail']}")

if __name__ == "__main__":
    asyncio.run(main())