"""BE parity: run the current parser on stored BE exports and compare EVERY field with manually created
FTWZ BOE Import documents on the eap Frappe site (read-only)."""
import json
import re
import subprocess
import sys
import urllib.parse
import urllib.request
from collections import Counter, defaultdict
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import pandas as pd

S = Path(sys.argv[1])
CLIENT = sys.argv[2] if len(sys.argv) > 2 else "eap"
sys.path.insert(0, "app/phase_a_fetch")
sys.path.insert(0, ".")
import nsdl_parser  # noqa: E402

secret = json.loads(
    subprocess.check_output(
        ["aws", "secretsmanager", "get-secret-value", "--secret-id", "nsdl-boe-automation-secrets-TEST",
         "--region", "ap-south-1", "--query", "SecretString", "--output", "text"], text=True))
c = secret["CLIENTS"][CLIENT]
base = c["FRAPPE_URL"].rstrip("/")
H = {"Authorization": f"token {c['api_key']}:{c['api_secret']}"}


def get(path):
    return json.load(urllib.request.urlopen(urllib.request.Request(base + path, headers=H), timeout=120))


frames, latest = [], {}
for i, f in enumerate(sorted((S / "rawbe").glob("*/be_export_v1.xlsx"))):
    df = nsdl_parser.read_nsdl_export(str(f))
    df["__file"] = i
    frames.append(df)
    for rid in df["REQUEST ID"].astype(str).str.strip().unique():
        latest[rid] = i
all_df = pd.concat(frames, ignore_index=True)
all_df["__rid"] = all_df["REQUEST ID"].astype(str).str.strip()
all_df = all_df[all_df.apply(lambda r: latest[r["__rid"]] == r["__file"], axis=1)]
print("BE requests in stored exports:", all_df["__rid"].nunique())

DT = "FTWZ BOE Import"
q = urllib.parse.urlencode({"fields": json.dumps(["name", "owner", "creation", "request_id"]), "limit_page_length": "9000", "order_by": "creation asc"})
lst = get("/api/resource/" + urllib.parse.quote(DT) + "?" + q)["data"]
print("site", base.split("//")[1], "| docs:", len(lst), "| owners:", Counter(r["owner"] for r in lst).most_common(4))
manual = [r for r in lst if r["owner"] != "<automation-user>" and r["creation"] >= "2026-09-05" and r.get("request_id") in latest]
print("manual docs since 2026-09-05 matching a stored BE request:", len(manual))
if not manual:
    sys.exit(0)
with ThreadPoolExecutor(6) as pool:
    docs = list(pool.map(lambda n: get("/api/resource/" + urllib.parse.quote(DT) + "/" + urllib.parse.quote(n))["data"], [r["name"] for r in manual]))

sel = all_df[all_df["__rid"].isin({d["request_id"] for d in docs})].drop(columns=["__file", "__rid"])
res = nsdl_parser.parse_be_records(sel, len(sel))
parsed = {r["request_id"]: r for r in res.records}
print("parser produced:", len(parsed), "records | QC passed:", nsdl_parser.qc_all_passed(res.qc)[0])

SYS = {"name", "owner", "creation", "modified", "modified_by", "docstatus", "idx", "parent", "parentfield", "parenttype",
       "doctype", "system_doc_no", "workflow_state", "naming_series", "amended_from", "_comments", "_assign", "_liked_by", "_user_tags"}


def empty(v):
    return v is None or v == "" or v == [] or v == {}


def norm(v):
    return re.sub(r"\s+", " ", str(v).strip()).upper()


def same(a, b):
    if norm(a) == norm(b):
        return True
    try:
        return abs(float(a) - float(b)) < 0.0051
    except (TypeError, ValueError):
        return False


miss = defaultdict(Counter)   # scope -> field -> manual filled, parser empty
diff = defaultdict(Counter)   # scope -> field -> both filled, differ
ex = defaultdict(list)
extra_in_parser = defaultdict(Counter)
n_items = 0
for d in docs:
    p = parsed.get(d["request_id"])
    if not p:
        continue
    for k, mv in d.items():
        if k in SYS or isinstance(mv, (list, dict)):
            continue
        pv = p.get(k)
        if not empty(mv) and empty(pv):
            miss["header"][k] += 1
        elif not empty(mv) and not empty(pv) and not same(mv, pv):
            diff["header"][k] += 1
            if len(ex[("header", k)]) < 2:
                ex[("header", k)].append((d["name"], str(mv)[:40], str(pv)[:40]))
    for k, pv in p.items():
        if k not in SYS and not isinstance(pv, (list, dict)) and not empty(pv) and empty(d.get(k)):
            extra_in_parser["header"][k] += 1
    for item in d.get("items", []):
        cand = [i for i in p.get("items", []) if norm(i.get("invoice_no")) == norm(item.get("invoice_no"))
                and norm(i.get("item_description"))[:40] == norm(item.get("item_description"))[:40]]
        if len(cand) != 1:
            continue
        n_items += 1
        pi = cand[0]
        for k, mv in item.items():
            if k in SYS or isinstance(mv, (list, dict)):
                continue
            pv = pi.get(k)
            if not empty(mv) and empty(pv):
                miss["item"][k] += 1
            elif not empty(mv) and not empty(pv) and not same(mv, pv):
                diff["item"][k] += 1
                if len(ex[("item", k)]) < 2:
                    ex[("item", k)].append((d["name"], str(mv)[:30], str(pv)[:30]))
        for k, pv in pi.items():
            if k not in SYS and not isinstance(pv, (list, dict)) and not empty(pv) and empty(item.get(k)):
                extra_in_parser["item"][k] += 1

print(f"\ncompared {len(docs)} manual docs and {n_items} item rows\n")
for scope in ("header", "item"):
    print(f"== {scope}: manual value present, parser value EMPTY")
    for k, n in miss[scope].most_common():
        print(f"   {k:55s} {n}")
    if not miss[scope]:
        print("   (none)")
    print(f"== {scope}: both present but DIFFERENT")
    for k, n in diff[scope].most_common():
        print(f"   {k:55s} {n}   e.g. {ex[(scope, k)][:1]}")
    if not diff[scope]:
        print("   (none)")
    print(f"== {scope}: parser sends a value, manual record has none")
    for k, n in extra_in_parser[scope].most_common(8):
        print(f"   {k:55s} {n}")
    if not extra_in_parser[scope]:
        print("   (none)")