1"""Phase B — deduplication of discovery records and stage-1 (title/abstract) rule-based screening.2Inputs: data/metadata/s2_records.jsonl (+ openalex_records.jsonl if present). Outputs: data/metadata/records_dedup.csv,3data/metadata/stage1_candidates.csv, data/metadata/screening_stage1_rules.csv, reports/search_audit.md (counts section).4"""5import csv, json, re, sys, unicodedata6from collections import Counter7from pathlib import Path8import pandas as pd, yaml910ROOT = Path(__file__).resolve().parents[2]; META = ROOT / "data" / "metadata"111213def norm_title(t):14 t = unicodedata.normalize("NFKD", t or "").encode("ascii", "ignore").decode().lower()15 return re.sub(r"[^a-z0-9]", "", t)161718def load():19 rows = []20 f = META / "s2_records.jsonl"21 if f.exists():22 for line in open(f, encoding="utf-8"):23 p = json.loads(line)24 ext = p.get("externalIds") or {}25 rows.append({"source": p.get("_source"), "query": p.get("_query"), "sid": p.get("paperId"), "doi": (ext.get("DOI") or "").lower(),26 "arxiv": ext.get("ArXiv"), "title": p.get("title") or "", "year": p.get("year"), "venue": (p.get("venue") or (p.get("publicationVenue") or {}).get("name") or ""),27 "authors": "; ".join(a.get("name", "") for a in (p.get("authors") or [])), "abstract": p.get("abstract") or "",28 "citations": p.get("citationCount"), "oa_pdf": ((p.get("openAccessPdf") or {}).get("url") or ""), "is_oa": p.get("isOpenAccess"),29 "pub_types": ";".join(p.get("publicationTypes") or []), "fields": ";".join(p.get("fieldsOfStudy") or []), "pub_date": p.get("publicationDate")})30 f2 = META / "openalex_records.jsonl"31 if f2.exists():32 for line in open(f2, encoding="utf-8"):33 w = json.loads(line)34 rows.append({"source": "openalex", "query": w.get("_query"), "sid": w.get("id"), "doi": (w.get("doi") or "").replace("https://doi.org/", "").lower(),35 "arxiv": None, "title": w.get("title") or "", "year": w.get("publication_year"), "venue": ((w.get("primary_location") or {}).get("source") or {}).get("display_name", ""),36 "authors": "; ".join((a.get("author") or {}).get("display_name", "") for a in w.get("authorships") or []), "abstract": w.get("abstract") or "",37 "citations": w.get("cited_by_count"), "oa_pdf": ((w.get("best_oa_location") or {}).get("pdf_url") or ""), "is_oa": (w.get("open_access") or {}).get("is_oa"),38 "pub_types": w.get("type"), "fields": "", "pub_date": w.get("publication_date")})39 f3 = META / "crossref_records.jsonl"40 if f3.exists():41 for line in open(f3, encoding="utf-8"):42 it = json.loads(line)43 yr = ((it.get("issued") or {}).get("date-parts") or [[None]])[0][0]44 abst = re.sub(r"<[^>]+>", " ", it.get("abstract") or ""); abst = re.sub(r"\s+", " ", abst).strip()45 rows.append({"source": "crossref", "query": it.get("_query"), "sid": None, "doi": (it.get("DOI") or "").lower(), "arxiv": None,46 "title": " ".join(it.get("title") or []), "year": yr, "venue": " ".join(it.get("container-title") or []),47 "authors": "; ".join(f"{a.get('given','')} {a.get('family','')}".strip() for a in it.get("author") or []), "abstract": abst,48 "citations": it.get("is-referenced-by-count"), "oa_pdf": "", "is_oa": None, "pub_types": it.get("type"), "fields": ";".join(it.get("subject") or []), "pub_date": None})49 f4 = META / "arxiv_records.jsonl"50 if f4.exists():51 for line in open(f4, encoding="utf-8"):52 it = json.loads(line); aid = it["id"].split("/abs/")[-1]53 rows.append({"source": "arxiv", "query": it.get("_query"), "sid": None, "doi": (it.get("doi") or "").lower(), "arxiv": re.sub(r"v\d+$", "", aid),54 "title": it["title"], "year": int(it["published"][:4]), "venue": "arXiv", "authors": "; ".join(it["authors"]), "abstract": it["abstract"],55 "citations": None, "oa_pdf": f"https://arxiv.org/pdf/{aid}", "is_oa": True, "pub_types": "preprint", "fields": ";".join(it["categories"]), "pub_date": it["published"][:10]})56 df = pd.DataFrame(rows)57 df["record_id"] = [r.sid if isinstance(r.sid, str) and r.sid else ("doi_" + re.sub(r"[^a-z0-9]", "_", r.doi) if r.doi else ("arxiv_" + str(r.arxiv) if r.arxiv else f"row_{i}")) for i, r in df.iterrows()]58 return df596061def dedup(df):62 df = df.copy(); df["ntitle"] = df.title.map(norm_title); df["first_author"] = df.authors.str.split(";").str[0].str.strip().str.split().str[-1].str.lower()63 df["key_doi"] = df.doi.where(df.doi != "", None)64 df["key_title"] = df.ntitle + "|" + df.year.fillna(0).astype(int).astype(str)65 df["raw_n"] = 166 # prefer records with abstract, DOI, OA pdf67 df["score_rec"] = df.abstract.str.len().fillna(0) + 1000 * (df.doi != "") + 500 * (df.oa_pdf != "")68 df = df.sort_values("score_rec", ascending=False)69 # group by DOI first, then by normalised title (+year within 1 handled by exact title match)70 seen_doi = {}; seen_title = {}; keep = []; queries = {}71 for i, r in df.iterrows():72 k = None73 if r.key_doi and r.key_doi in seen_doi:74 k = seen_doi[r.key_doi]75 elif r.ntitle and len(r.ntitle) > 15 and r.ntitle in seen_title:76 k = seen_title[r.ntitle]77 if k is None:78 k = i; keep.append(i)79 if r.key_doi: seen_doi[r.key_doi] = k80 if r.ntitle: seen_title[r.ntitle] = k81 queries.setdefault(k, set()).add(str(r.query)); 82 out = df.loc[keep].copy(); out["n_queries"] = out.index.map(lambda k: len(queries[k])); out["queries"] = out.index.map(lambda k: " || ".join(sorted(queries[k])))83 return out.reset_index(drop=True)848586def screen(df, kw):87 text = (df.title.fillna("") + " . " + df.abstract.fillna("")).str.lower()88 def hits(group):89 pats = [k.lower() for k in kw[group]]90 return text.map(lambda t: sum(1 for p_ in pats if p_ in t))91 df = df.copy()92 df["h_hedonic"] = hits("hedonic_strong"); df["h_price"] = hits("price_outcome"); df["h_res"] = hits("residential"); df["h_reg"] = hits("regression_signal"); df["h_excl"] = hits("exclude_hard")93 df["score"] = 3 * (df.h_hedonic > 0) + 2 * (df.h_price > 0) + 1 * (df.h_res > 0) + 1 * (df.h_reg > 0) - 5 * ((df.h_excl > 0) & (df.h_price == 0))94 df["stage1_candidate"] = ((df.h_price > 0) & (df.h_res > 0) & ((df.h_hedonic > 0) | (df.h_reg > 0))) | ((df.h_hedonic > 0) & (df.h_res > 0))95 df["stage1_reason"] = df.apply(lambda r: ("candidate" if r.stage1_candidate else ("no residential price outcome" if r.h_price == 0 else "no hedonic/regression signal")), axis=1)96 df["no_abstract"] = df.abstract.fillna("").str.len() < 5097 return df9899100def main():101 kw = yaml.safe_load((ROOT / "scripts" / "screening" / "keywords.yaml").read_text())102 raw = load(); dd = dedup(raw); sc = screen(dd, kw)103 sc.to_csv(META / "records_dedup.csv", index=False)104 cand = sc[sc.stage1_candidate].sort_values(["score", "citations"], ascending=False)105 cand.to_csv(META / "stage1_candidates.csv", index=False)106 sc[["sid", "doi", "title", "year", "score", "stage1_candidate", "stage1_reason", "no_abstract"]].to_csv(META / "screening_stage1_rules.csv", index=False)107 summary = {"raw_records": int(len(raw)), "unique_records": int(len(dd)), "with_doi": int((dd.doi != "").sum()), "with_abstract": int((~sc.no_abstract).sum()),108 "stage1_candidates": int(len(cand)), "candidates_with_oa_pdf": int((cand.oa_pdf != "").sum()), "by_source": raw.source.value_counts().to_dict(),109 "years": {"min": int(dd.year.min()), "max": int(dd.year.max())}}110 (META / "screening_summary.json").write_text(json.dumps(summary, indent=2)); print(json.dumps(summary, indent=2))111112113if __name__ == "__main__":114 main()115