"""Phase 1 — corpus inventory, validation, duplicate detection, bibliographic join, frozen manifest."""
from __future__ import annotations
import csv, json, re, unicodedata
from pathlib import Path
import pandas as pd
import pymupdf
from .provenance import sha256_file, stamp, Ledger, now_iso
from . import config

DOI_RE = re.compile(r"10\.1111/jofi\.\d{5}")

def pdf_probe(path: Path) -> dict:
    """Open with PyMuPDF; return validity, page count, metadata and first-page DOI/title candidates."""
    out = {"pdf_valid": False, "pages": None, "pdf_title": None, "pdf_author": None, "doi_in_pdf": None, "first_page_chars": 0, "error": None}
    try:
        doc = pymupdf.open(path)
        out["pages"] = doc.page_count
        md = doc.metadata or {}
        out["pdf_title"] = (md.get("title") or "").strip() or None
        out["pdf_author"] = (md.get("author") or "").strip() or None
        txt = unicodedata.normalize("NFKC", "\n".join(doc[i].get_text() for i in range(min(3, doc.page_count))))
        out["first_page_chars"] = len(txt)  # chars over the first three pages (presidential addresses open with a portrait page)
        m = DOI_RE.search(txt.replace("\n", "")) or DOI_RE.search(" ".join((md.get("subject") or "", md.get("keywords") or "")))
        out["doi_in_pdf"] = m.group(0) if m else None
        out["pdf_valid"] = doc.page_count > 0 and len(txt) > 50
        doc.close()
    except Exception as e:
        out["error"] = f"{type(e).__name__}: {e}"[:200]
    return out

def scan(root: Path) -> pd.DataFrame:
    rows = []
    for p in sorted(root.rglob("*")):
        if p.is_dir() or p.name.startswith(".") or "_outils" in p.parts: continue
        rel = str(p.relative_to(root))
        r = {"relpath": rel, "filename": p.name, "ext": p.suffix.lower(), "size_bytes": p.stat().st_size, "sha256": sha256_file(p)}
        if p.suffix.lower() == ".pdf":
            r.update(pdf_probe(p))
        else:
            r.update({"pdf_valid": None, "pages": None})
        rows.append(r)
    return pd.DataFrame(rows)

def join_bibliography(df: pd.DataFrame) -> pd.DataFrame:
    """Join the download index (title/authors/pages/DOI/section/year/issue) on relative path."""
    idx_path = config.expand(config.load("corpus")["bibliographic_index"])
    if not idx_path.exists():
        df["bib_joined"] = False; return df
    idx = pd.read_csv(idx_path)
    idx = idx.rename(columns={"fichier": "relpath", "title": "bib_title", "authors": "bib_authors", "pages": "bib_pages", "published": "bib_published", "doi": "bib_doi", "section": "bib_section", "year": "bib_year", "volume": "bib_volume", "issue": "bib_issue", "access": "bib_access"})
    idx = idx.drop(columns=[c for c in ["statut"] if c in idx.columns])
    out = df.merge(idx, on="relpath", how="left")
    out["bib_joined"] = out["bib_doi"].notna()
    return out

def classify(df: pd.DataFrame) -> pd.DataFrame:
    cfg = config.load("corpus")
    years = set(cfg["years_in_scope"]); secs = set(cfg["article_sections_in_scope"])
    dec, reason = [], []
    dup_hash = df["sha256"].duplicated(keep=False)
    dup_doi = df["bib_doi"].notna() & df["bib_doi"].duplicated(keep=False)
    for i, r in df.iterrows():
        if r["ext"] != ".pdf": dec.append("exclude"); reason.append("non-PDF file"); continue
        if r["size_bytes"] < cfg["min_pdf_bytes"]: dec.append("exclude"); reason.append("file too small"); continue
        if not r["pdf_valid"]: dec.append("exclude"); reason.append(f"broken PDF: {r.get('error')}"); continue
        if dup_hash[i]: dec.append("exclude"); reason.append("duplicate content (same SHA-256)"); continue
        if dup_doi[i]: dec.append("review"); reason.append("duplicate DOI (alternate version?)"); continue
        if not r["bib_joined"]: dec.append("review"); reason.append("no bibliographic record in index.csv"); continue
        if isinstance(r["doi_in_pdf"], str) and isinstance(r["bib_doi"], str) and r["doi_in_pdf"] != r["bib_doi"]: dec.append("review"); reason.append(f"DOI mismatch pdf={r['doi_in_pdf']} index={r['bib_doi']}"); continue
        if int(r["bib_year"]) not in years: dec.append("exclude"); reason.append("year out of scope"); continue
        if r["bib_section"] not in secs: dec.append("exclude"); reason.append(f"section out of scope: {r['bib_section']}"); continue
        if re.search(r"Participant Schedule|Preliminary Program|Report of the Editor|Minutes of the", str(r["bib_title"]), re.I):
            dec.append("exclude"); reason.append("not a research article (TOC mislabel: program/report)"); continue
        kind = "article"
        if str(r["bib_title"]).startswith("Presidential Address"): kind = "presidential_address"
        if r["bib_section"] in ("REPLICATIONS AND CORRIGENDA", "CORRIGENDUM"): kind = "corrigendum_or_replication"
        if r["bib_section"] == "THIS ARTICLE HAS BEEN RETRACTED": kind = "retracted_article"
        dec.append("include"); reason.append(kind)
    df["decision"] = dec; df["decision_reason"] = reason
    df["article_type"] = [r if d == "include" else None for d, r in zip(dec, reason)]
    return df

def crossref_verify(df: pd.DataFrame, limit: int | None = None) -> pd.DataFrame:
    """Verify DOI records against Crossref (public API, polite pool). Stores title/year/container; never infers missing metadata."""
    import httpx, time
    mailto = config.load("models")["crossref_mailto"]
    cache_p = config.manifests_dir() / "crossref_cache.json"
    cache = json.loads(cache_p.read_text()) if cache_p.exists() else {}
    dois = [d for d in df["bib_doi"].dropna().unique().tolist() if d not in cache]
    if limit: dois = dois[:limit]
    with httpx.Client(timeout=30, headers={"User-Agent": f"jof250/0.1 (mailto:{mailto})"}) as cl:
        for d in dois:
            try:
                r = cl.get(f"https://api.crossref.org/works/{d}", params={"mailto": mailto})
                if r.status_code == 200:
                    m = r.json()["message"]
                    cache[d] = {"status": 200, "title": (m.get("title") or [None])[0], "container": (m.get("container-title") or [None])[0],
                                "year": (m.get("issued", {}).get("date-parts") or [[None]])[0][0], "volume": m.get("volume"), "issue": m.get("issue"),
                                "page": m.get("page"), "type": m.get("type"), "authors": "; ".join(f"{a.get('given','')} {a.get('family','')}".strip() for a in m.get("author", [])),
                                "references_count": m.get("references-count"), "is_referenced_by_count": m.get("is-referenced-by-count"), "fetched": now_iso()}
                else:
                    cache[d] = {"status": r.status_code, "fetched": now_iso()}
            except Exception as e:
                cache[d] = {"status": "error", "error": str(e)[:200], "fetched": now_iso()}
            time.sleep(0.15)
    cache_p.write_text(json.dumps(cache, indent=1, ensure_ascii=False))
    def norm(t): return re.sub(r"[^a-z0-9]+", " ", str(t or "").lower()).strip()
    df["crossref_status"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("status"))
    df["crossref_title"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("title"))
    df["crossref_year"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("year"))
    df["crossref_container"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("container"))
    df["crossref_authors"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("authors"))
    df["crossref_refcount"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("references_count"))
    df["crossref_cited_by"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("is_referenced_by_count"))
    from rapidfuzz import fuzz
    df["title_match_ratio"] = [fuzz.ratio(norm(a), norm(b)) if a and b else None for a, b in zip(df["bib_title"], df["crossref_title"])]
    df["crossref_ok"] = (df["crossref_status"] == 200) & (df["crossref_container"].fillna("").str.contains("Journal of Finance")) & (df["title_match_ratio"].fillna(0) >= 85)
    return df

def build(verify: bool = True, freeze: bool = True) -> tuple[pd.DataFrame, dict]:
    root = config.corpus_root()
    if not root.exists(): raise FileNotFoundError(f"corpus root not found: {root}")
    df = scan(root)
    df = join_bibliography(df)
    df = classify(df)
    if verify and config.load("corpus").get("crossref_verify"):
        df = crossref_verify(df)
    df["doc_id"] = ["jof_" + str(d).split("jofi.")[-1] if isinstance(d, str) else "sha_" + s[:12] for d, s in zip(df["bib_doi"], df["sha256"])]
    man = config.manifests_dir()
    df.to_parquet(man / "corpus_manifest.parquet", index=False)
    df.to_csv(man / "corpus_manifest.csv", index=False)
    summary = audit_report(df, man / "corpus_audit.md", root)
    if freeze:
        inc = df[df["decision"] == "include"][["doc_id", "relpath", "sha256", "bib_doi", "bib_year", "bib_volume", "bib_issue", "bib_title", "article_type", "pages"]]
        frozen = {"version": "corpus_v1", "frozen_at": now_iso(), "n_included": int(len(inc)), "manifest_sha256": sha256_file(man / "corpus_manifest.parquet"), **stamp(), "documents": inc.to_dict(orient="records")}
        (man / "corpus_v1_frozen.json").write_text(json.dumps(frozen, indent=1, ensure_ascii=False, default=str))
    Ledger(man / "inventory_ledger.jsonl").append({"event": "inventory_build", "n_files": len(df), "summary": summary, **stamp()})
    return df, summary

def audit_report(df: pd.DataFrame, out: Path, root: Path) -> dict:
    inc = df[df["decision"] == "include"]
    s = {"n_files": int(len(df)), "n_pdf": int((df["ext"] == ".pdf").sum()), "n_included": int(len(inc)), "n_excluded": int((df["decision"] == "exclude").sum()), "n_review": int((df["decision"] == "review").sum()),
         "by_year": {int(k): int(v) for k, v in inc["bib_year"].value_counts().sort_index().items()},
         "by_type": {str(k): int(v) for k, v in inc["article_type"].value_counts().items()},
         "pages_total": int(inc["pages"].sum()), "pages_median": float(inc["pages"].median()), "size_mb": round(float(inc["size_bytes"].sum()) / 1e6, 1),
         "crossref_ok": int(inc["crossref_ok"].sum()) if "crossref_ok" in inc else None, "duplicates_sha": int(df["sha256"].duplicated().sum())}
    lines = [f"# Corpus audit — {root}", "", f"Generated {now_iso()} by `jof250 corpus audit` (code {stamp()['code_version']}).", "",
             f"- Files scanned: {s['n_files']} (PDF: {s['n_pdf']})", f"- Included: **{s['n_included']}**; excluded: {s['n_excluded']}; needs review: {s['n_review']}",
             f"- Exact duplicates (SHA-256): {s['duplicates_sha']}", f"- Crossref-verified included records (status 200, container = Journal of Finance, title match ≥ 85): {s['crossref_ok']}",
             f"- Pages: total {s['pages_total']}, median {s['pages_median']}; size {s['size_mb']} MB", "", "## Included by year", ""]
    lines += [f"| {k} | {v} |" for k, v in s["by_year"].items()]
    lines += ["", "## Included by article type", ""] + [f"| {k} | {v} |" for k, v in s["by_type"].items()]
    ex = df[df["decision"] != "include"]
    lines += ["", "## Excluded / review records", "", "| decision | reason | file |", "|---|---|---|"] + [f"| {r.decision} | {r.decision_reason} | {r.relpath} |" for r in ex.itertuples()]
    if "crossref_ok" in df:
        bad = inc[~inc["crossref_ok"]]
        lines += ["", "## Included records failing Crossref check", "", "| doc_id | status | title match | file |", "|---|---|---|---|"] + [f"| {r.doc_id} | {r.crossref_status} | {r.title_match_ratio} | {r.relpath} |" for r in bad.itertuples()]
    out.write_text("\n".join(lines) + "\n")
    return s