1"""Phase 1 — corpus inventory, validation, duplicate detection, bibliographic join, frozen manifest."""2from __future__ import annotations3import csv, json, re, unicodedata4from pathlib import Path5import pandas as pd6import pymupdf7from .provenance import sha256_file, stamp, Ledger, now_iso8from . import config910DOI_RE = re.compile(r"10\.1111/jofi\.\d{5}")1112def pdf_probe(path: Path) -> dict:13 """Open with PyMuPDF; return validity, page count, metadata and first-page DOI/title candidates."""14 out = {"pdf_valid": False, "pages": None, "pdf_title": None, "pdf_author": None, "doi_in_pdf": None, "first_page_chars": 0, "error": None}15 try:16 doc = pymupdf.open(path)17 out["pages"] = doc.page_count18 md = doc.metadata or {}19 out["pdf_title"] = (md.get("title") or "").strip() or None20 out["pdf_author"] = (md.get("author") or "").strip() or None21 txt = unicodedata.normalize("NFKC", "\n".join(doc[i].get_text() for i in range(min(3, doc.page_count))))22 out["first_page_chars"] = len(txt) # chars over the first three pages (presidential addresses open with a portrait page)23 m = DOI_RE.search(txt.replace("\n", "")) or DOI_RE.search(" ".join((md.get("subject") or "", md.get("keywords") or "")))24 out["doi_in_pdf"] = m.group(0) if m else None25 out["pdf_valid"] = doc.page_count > 0 and len(txt) > 5026 doc.close()27 except Exception as e:28 out["error"] = f"{type(e).__name__}: {e}"[:200]29 return out3031def scan(root: Path) -> pd.DataFrame:32 rows = []33 for p in sorted(root.rglob("*")):34 if p.is_dir() or p.name.startswith(".") or "_outils" in p.parts: continue35 rel = str(p.relative_to(root))36 r = {"relpath": rel, "filename": p.name, "ext": p.suffix.lower(), "size_bytes": p.stat().st_size, "sha256": sha256_file(p)}37 if p.suffix.lower() == ".pdf":38 r.update(pdf_probe(p))39 else:40 r.update({"pdf_valid": None, "pages": None})41 rows.append(r)42 return pd.DataFrame(rows)4344def join_bibliography(df: pd.DataFrame) -> pd.DataFrame:45 """Join the download index (title/authors/pages/DOI/section/year/issue) on relative path."""46 idx_path = config.expand(config.load("corpus")["bibliographic_index"])47 if not idx_path.exists():48 df["bib_joined"] = False; return df49 idx = pd.read_csv(idx_path)50 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"})51 idx = idx.drop(columns=[c for c in ["statut"] if c in idx.columns])52 out = df.merge(idx, on="relpath", how="left")53 out["bib_joined"] = out["bib_doi"].notna()54 return out5556def classify(df: pd.DataFrame) -> pd.DataFrame:57 cfg = config.load("corpus")58 years = set(cfg["years_in_scope"]); secs = set(cfg["article_sections_in_scope"])59 dec, reason = [], []60 dup_hash = df["sha256"].duplicated(keep=False)61 dup_doi = df["bib_doi"].notna() & df["bib_doi"].duplicated(keep=False)62 for i, r in df.iterrows():63 if r["ext"] != ".pdf": dec.append("exclude"); reason.append("non-PDF file"); continue64 if r["size_bytes"] < cfg["min_pdf_bytes"]: dec.append("exclude"); reason.append("file too small"); continue65 if not r["pdf_valid"]: dec.append("exclude"); reason.append(f"broken PDF: {r.get('error')}"); continue66 if dup_hash[i]: dec.append("exclude"); reason.append("duplicate content (same SHA-256)"); continue67 if dup_doi[i]: dec.append("review"); reason.append("duplicate DOI (alternate version?)"); continue68 if not r["bib_joined"]: dec.append("review"); reason.append("no bibliographic record in index.csv"); continue69 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']}"); continue70 if int(r["bib_year"]) not in years: dec.append("exclude"); reason.append("year out of scope"); continue71 if r["bib_section"] not in secs: dec.append("exclude"); reason.append(f"section out of scope: {r['bib_section']}"); continue72 kind = "article"73 if r["bib_section"] in ("REPLICATIONS AND CORRIGENDA", "CORRIGENDUM"): kind = "corrigendum_or_replication"74 if r["bib_section"] == "THIS ARTICLE HAS BEEN RETRACTED": kind = "retracted_article"75 dec.append("include"); reason.append(kind)76 df["decision"] = dec; df["decision_reason"] = reason77 df["article_type"] = [r if d == "include" else None for d, r in zip(dec, reason)]78 return df7980def crossref_verify(df: pd.DataFrame, limit: int | None = None) -> pd.DataFrame:81 """Verify DOI records against Crossref (public API, polite pool). Stores title/year/container; never infers missing metadata."""82 import httpx, time83 mailto = config.load("models")["crossref_mailto"]84 cache_p = config.manifests_dir() / "crossref_cache.json"85 cache = json.loads(cache_p.read_text()) if cache_p.exists() else {}86 dois = [d for d in df["bib_doi"].dropna().unique().tolist() if d not in cache]87 if limit: dois = dois[:limit]88 with httpx.Client(timeout=30, headers={"User-Agent": f"jof250/0.1 (mailto:{mailto})"}) as cl:89 for d in dois:90 try:91 r = cl.get(f"https://api.crossref.org/works/{d}", params={"mailto": mailto})92 if r.status_code == 200:93 m = r.json()["message"]94 cache[d] = {"status": 200, "title": (m.get("title") or [None])[0], "container": (m.get("container-title") or [None])[0],95 "year": (m.get("issued", {}).get("date-parts") or [[None]])[0][0], "volume": m.get("volume"), "issue": m.get("issue"),96 "page": m.get("page"), "type": m.get("type"), "authors": "; ".join(f"{a.get('given','')} {a.get('family','')}".strip() for a in m.get("author", [])),97 "references_count": m.get("references-count"), "is_referenced_by_count": m.get("is-referenced-by-count"), "fetched": now_iso()}98 else:99 cache[d] = {"status": r.status_code, "fetched": now_iso()}100 except Exception as e:101 cache[d] = {"status": "error", "error": str(e)[:200], "fetched": now_iso()}102 time.sleep(0.15)103 cache_p.write_text(json.dumps(cache, indent=1, ensure_ascii=False))104 def norm(t): return re.sub(r"[^a-z0-9]+", " ", str(t or "").lower()).strip()105 df["crossref_status"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("status"))106 df["crossref_title"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("title"))107 df["crossref_year"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("year"))108 df["crossref_container"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("container"))109 df["crossref_authors"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("authors"))110 df["crossref_refcount"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("references_count"))111 df["crossref_cited_by"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("is_referenced_by_count"))112 from rapidfuzz import fuzz113 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"])]114 df["crossref_ok"] = (df["crossref_status"] == 200) & (df["crossref_container"].fillna("").str.contains("Journal of Finance")) & (df["title_match_ratio"].fillna(0) >= 85)115 return df116117def build(verify: bool = True, freeze: bool = True) -> tuple[pd.DataFrame, dict]:118 root = config.corpus_root()119 if not root.exists(): raise FileNotFoundError(f"corpus root not found: {root}")120 df = scan(root)121 df = join_bibliography(df)122 df = classify(df)123 if verify and config.load("corpus").get("crossref_verify"):124 df = crossref_verify(df)125 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"])]126 man = config.manifests_dir()127 df.to_parquet(man / "corpus_manifest.parquet", index=False)128 df.to_csv(man / "corpus_manifest.csv", index=False)129 summary = audit_report(df, man / "corpus_audit.md", root)130 if freeze:131 inc = df[df["decision"] == "include"][["doc_id", "relpath", "sha256", "bib_doi", "bib_year", "bib_volume", "bib_issue", "bib_title", "article_type", "pages"]]132 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")}133 (man / "corpus_v1_frozen.json").write_text(json.dumps(frozen, indent=1, ensure_ascii=False, default=str))134 Ledger(man / "inventory_ledger.jsonl").append({"event": "inventory_build", "n_files": len(df), "summary": summary, **stamp()})135 return df, summary136137def audit_report(df: pd.DataFrame, out: Path, root: Path) -> dict:138 inc = df[df["decision"] == "include"]139 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()),140 "by_year": {int(k): int(v) for k, v in inc["bib_year"].value_counts().sort_index().items()},141 "by_type": {str(k): int(v) for k, v in inc["article_type"].value_counts().items()},142 "pages_total": int(inc["pages"].sum()), "pages_median": float(inc["pages"].median()), "size_mb": round(float(inc["size_bytes"].sum()) / 1e6, 1),143 "crossref_ok": int(inc["crossref_ok"].sum()) if "crossref_ok" in inc else None, "duplicates_sha": int(df["sha256"].duplicated().sum())}144 lines = [f"# Corpus audit — {root}", "", f"Generated {now_iso()} by `jof250 corpus audit` (code {stamp()['code_version']}).", "",145 f"- Files scanned: {s['n_files']} (PDF: {s['n_pdf']})", f"- Included: **{s['n_included']}**; excluded: {s['n_excluded']}; needs review: {s['n_review']}",146 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']}",147 f"- Pages: total {s['pages_total']}, median {s['pages_median']}; size {s['size_mb']} MB", "", "## Included by year", ""]148 lines += [f"| {k} | {v} |" for k, v in s["by_year"].items()]149 lines += ["", "## Included by article type", ""] + [f"| {k} | {v} |" for k, v in s["by_type"].items()]150 ex = df[df["decision"] != "include"]151 lines += ["", "## Excluded / review records", "", "| decision | reason | file |", "|---|---|---|"] + [f"| {r.decision} | {r.decision_reason} | {r.relpath} |" for r in ex.itertuples()]152 if "crossref_ok" in df:153 bad = inc[~inc["crossref_ok"]]154 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()]155 out.write_text("\n".join(lines) + "\n")156 return s157