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 if re.search(r"Participant Schedule|Preliminary Program|Report of the Editor|Minutes of the", str(r["bib_title"]), re.I):73 dec.append("exclude"); reason.append("not a research article (TOC mislabel: program/report)"); continue74 kind = "article"75 if str(r["bib_title"]).startswith("Presidential Address"): kind = "presidential_address"76 if r["bib_section"] in ("REPLICATIONS AND CORRIGENDA", "CORRIGENDUM"): kind = "corrigendum_or_replication"77 if r["bib_section"] == "THIS ARTICLE HAS BEEN RETRACTED": kind = "retracted_article"78 dec.append("include"); reason.append(kind)79 df["decision"] = dec; df["decision_reason"] = reason80 df["article_type"] = [r if d == "include" else None for d, r in zip(dec, reason)]81 return df8283def crossref_verify(df: pd.DataFrame, limit: int | None = None) -> pd.DataFrame:84 """Verify DOI records against Crossref (public API, polite pool). Stores title/year/container; never infers missing metadata."""85 import httpx, time86 mailto = config.load("models")["crossref_mailto"]87 cache_p = config.manifests_dir() / "crossref_cache.json"88 cache = json.loads(cache_p.read_text()) if cache_p.exists() else {}89 dois = [d for d in df["bib_doi"].dropna().unique().tolist() if d not in cache]90 if limit: dois = dois[:limit]91 with httpx.Client(timeout=30, headers={"User-Agent": f"jof250/0.1 (mailto:{mailto})"}) as cl:92 for d in dois:93 try:94 r = cl.get(f"https://api.crossref.org/works/{d}", params={"mailto": mailto})95 if r.status_code == 200:96 m = r.json()["message"]97 cache[d] = {"status": 200, "title": (m.get("title") or [None])[0], "container": (m.get("container-title") or [None])[0],98 "year": (m.get("issued", {}).get("date-parts") or [[None]])[0][0], "volume": m.get("volume"), "issue": m.get("issue"),99 "page": m.get("page"), "type": m.get("type"), "authors": "; ".join(f"{a.get('given','')} {a.get('family','')}".strip() for a in m.get("author", [])),100 "references_count": m.get("references-count"), "is_referenced_by_count": m.get("is-referenced-by-count"), "fetched": now_iso()}101 else:102 cache[d] = {"status": r.status_code, "fetched": now_iso()}103 except Exception as e:104 cache[d] = {"status": "error", "error": str(e)[:200], "fetched": now_iso()}105 time.sleep(0.15)106 cache_p.write_text(json.dumps(cache, indent=1, ensure_ascii=False))107 def norm(t): return re.sub(r"[^a-z0-9]+", " ", str(t or "").lower()).strip()108 df["crossref_status"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("status"))109 df["crossref_title"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("title"))110 df["crossref_year"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("year"))111 df["crossref_container"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("container"))112 df["crossref_authors"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("authors"))113 df["crossref_refcount"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("references_count"))114 df["crossref_cited_by"] = df["bib_doi"].map(lambda d: cache.get(d, {}).get("is_referenced_by_count"))115 from rapidfuzz import fuzz116 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"])]117 df["crossref_ok"] = (df["crossref_status"] == 200) & (df["crossref_container"].fillna("").str.contains("Journal of Finance")) & (df["title_match_ratio"].fillna(0) >= 85)118 return df119120def build(verify: bool = True, freeze: bool = True) -> tuple[pd.DataFrame, dict]:121 root = config.corpus_root()122 if not root.exists(): raise FileNotFoundError(f"corpus root not found: {root}")123 df = scan(root)124 df = join_bibliography(df)125 df = classify(df)126 if verify and config.load("corpus").get("crossref_verify"):127 df = crossref_verify(df)128 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"])]129 man = config.manifests_dir()130 df.to_parquet(man / "corpus_manifest.parquet", index=False)131 df.to_csv(man / "corpus_manifest.csv", index=False)132 summary = audit_report(df, man / "corpus_audit.md", root)133 if freeze:134 inc = df[df["decision"] == "include"][["doc_id", "relpath", "sha256", "bib_doi", "bib_year", "bib_volume", "bib_issue", "bib_title", "article_type", "pages"]]135 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")}136 (man / "corpus_v1_frozen.json").write_text(json.dumps(frozen, indent=1, ensure_ascii=False, default=str))137 Ledger(man / "inventory_ledger.jsonl").append({"event": "inventory_build", "n_files": len(df), "summary": summary, **stamp()})138 return df, summary139140def audit_report(df: pd.DataFrame, out: Path, root: Path) -> dict:141 inc = df[df["decision"] == "include"]142 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()),143 "by_year": {int(k): int(v) for k, v in inc["bib_year"].value_counts().sort_index().items()},144 "by_type": {str(k): int(v) for k, v in inc["article_type"].value_counts().items()},145 "pages_total": int(inc["pages"].sum()), "pages_median": float(inc["pages"].median()), "size_mb": round(float(inc["size_bytes"].sum()) / 1e6, 1),146 "crossref_ok": int(inc["crossref_ok"].sum()) if "crossref_ok" in inc else None, "duplicates_sha": int(df["sha256"].duplicated().sum())}147 lines = [f"# Corpus audit — {root}", "", f"Generated {now_iso()} by `jof250 corpus audit` (code {stamp()['code_version']}).", "",148 f"- Files scanned: {s['n_files']} (PDF: {s['n_pdf']})", f"- Included: **{s['n_included']}**; excluded: {s['n_excluded']}; needs review: {s['n_review']}",149 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']}",150 f"- Pages: total {s['pages_total']}, median {s['pages_median']}; size {s['size_mb']} MB", "", "## Included by year", ""]151 lines += [f"| {k} | {v} |" for k, v in s["by_year"].items()]152 lines += ["", "## Included by article type", ""] + [f"| {k} | {v} |" for k, v in s["by_type"].items()]153 ex = df[df["decision"] != "include"]154 lines += ["", "## Excluded / review records", "", "| decision | reason | file |", "|---|---|---|"] + [f"| {r.decision} | {r.decision_reason} | {r.relpath} |" for r in ex.itertuples()]155 if "crossref_ok" in df:156 bad = inc[~inc["crossref_ok"]]157 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()]158 out.write_text("\n".join(lines) + "\n")159 return s160