Repository Public Research preview
1"""Corpus assembly: raw files + ``*.meta.json`` stubs (written by source adapters) → Document records,2passages, dedup clusters and statistics. Every source adapter must write, next to each raw file:34 <stem>.meta.json with keys: doc_id, source, source_ids, title, source_url, license{code,uri,evidence,raw},5 acquired_at, raw_path, raw_sha256, raw_bytes, raw_format (jats|bits|latex|pdf|txt|html),6 optional: pub_date, pub_year, pub_date_evidence, authors, venue, publisher, subject_area,7 subject_raw, doc_type, provider_md5, provider_md5_ok, retracted, provider_meta, notes89The parser dispatch (``parse_raw``) returns a ``ParsedDocument``; stub values are used when the parser cannot10determine a field (e.g. publication date of a LaTeX source comes from the arXiv metadata, with evidence).11"""12from __future__ import annotations1314import datetime as _dt15import json16from dataclasses import dataclass, field17from pathlib import Path1819from htr.data.schema import Document, LicenseInfo, Passage20from htr.data.taxonomy import discipline_from_pmc21from htr.dedup.minhash import Deduper22from htr.licenses.policy import LicenseRecord23from htr.parse.jats import ParsedDocument, parse_jats24from htr.segment.passages import SegmentConfig, passages_from_parsed25from htr.utils.io import append_jsonl, ensure_dir, read_json, write_json, write_jsonl26from htr.utils.provenance import run_header272829def parse_raw(stub: dict) -> ParsedDocument:30 fmt = stub.get("raw_format", "jats")31 raw = Path(stub["raw_path"]).read_bytes()32 if fmt in ("jats", "bits"):33 return parse_jats(raw, stub.get("source", "pmc"))34 if fmt == "latex":35 from htr.parse.latex import parse_latex_archive3637 return parse_latex_archive(Path(stub["raw_path"]), stub)38 if fmt == "pdf":39 from htr.parse.pdf import parse_pdf4041 return parse_pdf(Path(stub["raw_path"]), stub)42 if fmt in ("txt", "html", "epub"):43 from htr.parse.books import parse_book_file4445 return parse_book_file(Path(stub["raw_path"]), stub)46 raise ValueError(f"unknown raw_format {fmt}")474849def _pre2018(pub_date: str | None) -> bool:50 if not pub_date:51 return False52 iso = pub_date + ("-01" if len(pub_date) == 7 else "-01-01" if len(pub_date) == 4 else "")53 try:54 return _dt.date.fromisoformat(iso) < _dt.date(2018, 1, 1)55 except ValueError:56 return False575859def document_from(stub: dict, parsed: ParsedDocument) -> Document:60 lic_code = parsed.license_code if parsed.license_code != "UNKNOWN" else stub["license"]["code"]61 if parsed.license_code != "UNKNOWN" and parsed.license_code != stub["license"]["code"]:62 note = f"license mismatch: provider={stub['license']['code']} document={parsed.license_code}; the document statement is used"63 else:64 note = None65 lic = LicenseRecord(code=lic_code, uri=parsed.license_uri or stub["license"].get("uri"), evidence=parsed.license_evidence or stub["license"].get("evidence"), raw=stub["license"].get("raw"))66 pub_date = parsed.pub_date or stub.get("pub_date")67 pub_year = parsed.pub_year or stub.get("pub_year") or (int(pub_date[:4]) if pub_date else None)68 evidence = parsed.pub_date_evidence or stub.get("pub_date_evidence")69 subj = stub.get("subject_area") or (discipline_from_pmc(parsed.venue, parsed.subjects) if stub.get("source") == "pmc" else "other")70 notes = list(stub.get("notes", [])) + list(parsed.warnings) + ([note] if note else [])71 return Document(72 doc_id=stub["doc_id"],73 source=stub["source"],74 source_ids={**{k: str(v) for k, v in parsed.ids.items() if k in ("doi", "pmid", "pmcid")}, **stub.get("source_ids", {})},75 title=parsed.title or stub.get("title"),76 authors=parsed.authors or stub.get("authors", []),77 author_group=parsed.author_group or stub.get("author_group"),78 venue=parsed.venue or stub.get("venue"),79 publisher=parsed.publisher or stub.get("publisher"),80 doc_type=stub.get("doc_type") or parsed.doc_type, # type: ignore[arg-type]81 pub_date=pub_date,82 pub_year=pub_year,83 pub_date_evidence=evidence,84 pre2018_verified=_pre2018(pub_date) and evidence is not None,85 subject_area=subj,86 subject_raw=parsed.subjects or stub.get("subject_raw", []),87 language=parsed.language or stub.get("language"),88 source_url=stub["source_url"],89 license=LicenseInfo(**lic.to_dict()),90 acquired_at=stub["acquired_at"],91 raw_path=stub.get("raw_path"),92 raw_sha256=stub.get("raw_sha256"),93 raw_bytes=stub.get("raw_bytes"),94 raw_format=stub.get("raw_format", "jats"),95 provider_md5=stub.get("provider_md5"),96 provider_md5_ok=stub.get("provider_md5_ok"),97 retracted=stub.get("retracted"),98 notes=notes,99 )100101102@dataclass103class BuildStats:104 stubs: int = 0105 parsed: int = 0106 parse_errors: int = 0107 not_pre2018: int = 0108 no_date: int = 0109 license_excluded: int = 0110 not_english: int = 0111 exact_dupes: int = 0112 near_dupes: int = 0113 documents_kept: int = 0114 passages_total: int = 0115 passages_eligible: int = 0116 by_source: dict = field(default_factory=dict)117 by_license: dict = field(default_factory=dict)118 by_discipline: dict = field(default_factory=dict)119 by_year: dict = field(default_factory=dict)120 by_role_eligible: dict = field(default_factory=dict)121 errors: list = field(default_factory=list)122123124def _inc(d: dict, k, n: int = 1) -> None:125 k = str(k)126 d[k] = d.get(k, 0) + n127128129def build_corpus(raw_dirs: list[Path], out_dir: Path, allowed_codes: set[str], seg: SegmentConfig | None = None, limit: int | None = None, require_pre2018: bool = True, run_tag: str = "corpus") -> BuildStats:130 out_dir = ensure_dir(out_dir)131 seg = seg or SegmentConfig()132 stats = BuildStats()133 deduper = Deduper(threshold=0.8)134 docs_path, psg_path, excl_path = out_dir / "documents.jsonl", out_dir / "passages.jsonl", out_dir / "excluded.jsonl"135 for pth in (docs_path, psg_path, excl_path):136 pth.unlink(missing_ok=True)137 stubs = sorted(p for d in raw_dirs for p in Path(d).rglob("*.meta.json"))138 if limit:139 stubs = stubs[:limit]140 for sp in stubs:141 stats.stubs += 1142 stub = read_json(sp)143 try:144 parsed = parse_raw(stub)145 except Exception as e: # noqa: BLE001146 stats.parse_errors += 1147 stats.errors.append({"stub": str(sp), "error": repr(e)[:300]})148 append_jsonl(excl_path, {"doc_id": stub.get("doc_id"), "reason": "parse_error", "error": repr(e)[:300]})149 continue150 doc = document_from(stub, parsed)151 stats.parsed += 1152 reason = None153 if doc.license.code not in allowed_codes:154 reason = "license_excluded"155 stats.license_excluded += 1156 elif require_pre2018 and not doc.pre2018_verified:157 reason = "no_date" if not doc.pub_date else "not_pre2018"158 if reason == "no_date":159 stats.no_date += 1160 else:161 stats.not_pre2018 += 1162 elif doc.language and not doc.language.lower().startswith("en"):163 reason = "not_english"164 stats.not_english += 1165 if reason:166 append_jsonl(excl_path, {"doc_id": doc.doc_id, "reason": reason, "license": doc.license.code, "pub_date": doc.pub_date, "source": doc.source})167 continue168 canon = deduper.add(doc.doc_id, parsed.body_text[:20000] if parsed.body_text else (doc.title or doc.doc_id))169 if canon != doc.doc_id:170 if doc.doc_id in deduper.result.exact_dupes:171 stats.exact_dupes += 1172 r = "exact_duplicate"173 else:174 stats.near_dupes += 1175 r = "near_duplicate"176 append_jsonl(excl_path, {"doc_id": doc.doc_id, "reason": r, "canonical": canon, "source": doc.source})177 continue178 passages = passages_from_parsed(doc.doc_id, parsed, seg)179 # passage-level exact dedup within the document and across the corpus is handled at pair time; here we180 # only count.181 append_jsonl(docs_path, doc)182 write_jsonl(psg_path, passages, append=True)183 (ensure_dir(out_dir / "text") / f"{doc.doc_id}.txt").write_text(parsed.body_text, encoding="utf-8")184 stats.documents_kept += 1185 stats.passages_total += len(passages)186 n_el = sum(p.eligible for p in passages)187 stats.passages_eligible += n_el188 _inc(stats.by_source, doc.source)189 _inc(stats.by_license, doc.license.code)190 _inc(stats.by_discipline, doc.subject_area)191 _inc(stats.by_year, doc.pub_year)192 for p in passages:193 if p.eligible:194 _inc(stats.by_role_eligible, p.section_role)195 write_json(out_dir / "clusters.json", deduper.result.clusters)196 summary = {**run_header({"run_tag": run_tag}), "stats": stats.__dict__, "segment_config": seg.__dict__, "allowed_codes": sorted(allowed_codes), "raw_dirs": [str(d) for d in raw_dirs]}197 write_json(out_dir / "build_summary.json", summary)198 return stats199