"""Merge the verified literature parts into research/literature_matrix.csv and generate paper/references.bib. Every entry is built from the verified CSV rows and cross-checked against the raw Crossref JSON saved in research/verification/ (title similarity, year, DOI). Mismatches are reported, never silently fixed. """ from __future__ import annotations import csv import json import re import sys import unicodedata from difflib import SequenceMatcher from pathlib import Path ROOT = Path(__file__).resolve().parents[1] RES = ROOT / "research" VER = RES / "verification" # B-side keys that duplicate an A-side DOI -> canonical key ALIASES = {"fisher2003liquidity": "fisher2003controlling", "glower1998selling": "glower1998motivation", "han2024cashisking": "han2024cash", "reher2024mortgagecash": "reher2024mortgage"} BOOK_HINTS = ("MIT Press", "Princeton", "Cambridge University Press", "Oxford University Press", "Elsevier", "North-Holland") def norm(s: str) -> str: s = unicodedata.normalize("NFKD", s or "").encode("ascii", "ignore").decode() return re.sub(r"[^a-z0-9 ]", " ", s.lower()).strip() def tex_escape(s: str) -> str: return (s.replace("&", r"\&").replace("%", r"\%").replace("#", r"\#").replace("_", r"\_") .replace("$", r"\$")) def protect_caps(title: str) -> str: """Brace words with internal capitals / acronyms so aer.bst keeps them.""" out = [] for w in title.split(" "): core = re.sub(r"[^A-Za-z]", "", w) if len(core) > 1 and (core.isupper() or (core[0].isupper() and any(ch.isupper() for ch in core[1:]))): out.append("{" + w + "}") else: out.append(w) return " ".join(out) def load_parts(): rows = {} order = [] for part in ("A", "B"): with open(RES / f"lit_part_{part}.csv", newline="", encoding="utf-8") as fh: for r in csv.DictReader(fh): key = ALIASES.get(r["key"], r["key"]) r = dict(r); r["key"] = key; r["part"] = part if key in rows: # keep A's row, merge B's overlap statement rows[key]["overlap_with_wp19"] += " | B: " + r["overlap_with_wp19"] rows[key]["part"] = "A+B" continue rows[key] = r; order.append(key) return [rows[k] for k in order] def crossref_json(key: str, part: str): for p in ("A", "B"): f = VER / f"{p}_{key}.crossref.json" if f.exists(): return json.loads(f.read_text()) for alias, canon in ALIASES.items(): if canon == key: for p in ("A", "B"): f = VER / f"{p}_{alias}.crossref.json" if f.exists(): return json.loads(f.read_text()) return None def cr_item(j): if j is None: return None if "message" in j: m = j["message"] if "items" in m: return m["items"][0] if m["items"] else None return m if "items" in j: return j["items"][0] if j["items"] else None return j def check_against_crossref(row, item): issues = [] if item is None: return ["no crossref json (evidence file only)"] t = " ".join(item.get("title", []) or []) sim = SequenceMatcher(None, norm(t), norm(row["title"])).ratio() if sim < 0.6: issues.append(f"title similarity {sim:.2f}: crossref='{t[:60]}'") yr = None for k in ("published-print", "published-online", "issued", "created"): dp = item.get(k, {}).get("date-parts", [[None]]) if dp and dp[0] and dp[0][0]: yr = yr or dp[0][0] yrs = set() for k in ("published-print", "published-online", "issued", "created"): dp = item.get(k, {}).get("date-parts", [[None]]) if dp and dp[0] and dp[0][0]: yrs.add(int(dp[0][0])) if yrs and int(row["year"]) not in yrs and abs(int(row["year"]) - min(yrs)) > 1: issues.append(f"year {row['year']} vs crossref {sorted(yrs)}") doi = (item.get("DOI") or "").lower() if doi and doi not in row["doi_url"].lower(): issues.append(f"doi {doi} vs {row['doi_url']}") return issues def parse_vp(vp: str): """'48(2): 641-664' -> volume, number, pages.""" vol = num = pages = "" m = re.match(r"\s*([\w\-]+)\s*(?:\(([^)]*)\))?\s*[:,]?\s*(.*)$", vp or "") if m: vol, num, pages = m.group(1), m.group(2) or "", (m.group(3) or "").strip() pages = pages.replace("--", "-").replace("–", "-").replace("-", "--") return vol, num, pages def to_bib(row, item): key = row["key"]; outlet = row["outlet"]; year = row["year"] authors = tex_escape(row["authors"]) title = protect_caps(tex_escape(row["title"])) doi = row["doi_url"].replace("https://doi.org/", "") if "doi.org" in row["doi_url"] else "" url = "" if doi else row["doi_url"] typ = (item or {}).get("type", "") vol, num, pages = parse_vp(row.get("volume_pages", "")) lines = [] if typ == "book" or (typ in ("monograph",) ) or any(h in outlet for h in BOOK_HINTS) and typ not in ("journal-article", "book-chapter"): lines.append(f"@book{{{key},") lines.append(f" author = {{{authors}}},"); lines.append(f" title = {{{title}}},") lines.append(f" publisher = {{{tex_escape(outlet)}}},"); lines.append(f" year = {{{year}}},") elif typ == "book-chapter" or "Handbook" in outlet: lines.append(f"@incollection{{{key},") lines.append(f" author = {{{authors}}},"); lines.append(f" title = {{{title}}},") lines.append(f" booktitle = {{{tex_escape(outlet)}}},"); lines.append(f" year = {{{year}}},") if vol: lines.append(f" volume = {{{vol}}},") if pages: lines.append(f" pages = {{{pages}}},") pub = ((item or {}).get("publisher") or "") if pub: lines.append(f" publisher = {{{tex_escape(pub)}}},") elif typ in ("posted-content", "report") or re.search(r"working paper|discussion paper|staff|NBER|SSRN|CEPR|mimeo", outlet, re.I): lines.append(f"@techreport{{{key},") lines.append(f" author = {{{authors}}},"); lines.append(f" title = {{{title}}},") lines.append(f" institution = {{{tex_escape(outlet)}}},"); lines.append(f" year = {{{year}}},") if vol: lines.append(f" number = {{{vol}}},") else: lines.append(f"@article{{{key},") lines.append(f" author = {{{authors}}},"); lines.append(f" title = {{{title}}},") lines.append(f" journal = {{{tex_escape(outlet)}}},"); lines.append(f" year = {{{year}}},") if vol: lines.append(f" volume = {{{vol}}},") if num: lines.append(f" number = {{{num}}},") if pages: lines.append(f" pages = {{{pages}}},") if doi: lines.append(f" doi = {{{doi}}},") if url: lines.append(f" url = {{{url}}},") lines[-1] = lines[-1].rstrip(",") lines.append("}") return "\n".join(lines) def main(): rows = load_parts() report = [] bib = [] seen_doi = {} with open(RES / "literature_matrix.csv", "w", newline="", encoding="utf-8") as fh: cols = ["key", "authors", "year", "title", "outlet", "volume_pages", "doi_url", "mechanism", "approach", "key_finding", "overlap_with_wp19", "gaps", "confidence", "verified_via", "part"] w = csv.DictWriter(fh, fieldnames=cols, quoting=csv.QUOTE_ALL, extrasaction="ignore") w.writeheader() for r in rows: item = cr_item(crossref_json(r["key"], r["part"])) issues = check_against_crossref(r, item) d = r["doi_url"].lower() if d in seen_doi: issues.append(f"duplicate DOI with {seen_doi[d]}") seen_doi[d] = r["key"] if issues: report.append((r["key"], issues)) w.writerow(r) bib.append(to_bib(r, item)) (ROOT / "paper" / "references.bib").write_text("\n\n".join(bib) + "\n", encoding="utf-8") (RES / "literature_merge_report.md").write_text( "# Literature merge report\n\n" + f"{len(rows)} references ({sum(r['part']=='A' for r in rows)} A, " f"{sum(r['part']=='B' for r in rows)} B, {sum(r['part']=='A+B' for r in rows)} both; aliases {ALIASES}).\n\n" + "## Cross-check issues (CSV vs raw Crossref JSON)\n\n" + ("\n".join(f"- `{k}`: " + "; ".join(v) for k, v in report) if report else "- none") + "\n") print(f"{len(rows)} references merged; {len(report)} rows with cross-check notes") for k, v in report: print(" ", k, "->", v) if __name__ == "__main__": main()