Spaces:
Running on Zero
Running on Zero
File size: 1,917 Bytes
9936912 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 | #!/usr/bin/env python3
"""Backfill publisher metadata and demote bulk-discovered books to supplemental."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CANDIDATES = PROJECT_ROOT / "data" / "sources" / "discovered_open_books.jsonl"
DEFAULT_LOCK = PROJECT_ROOT / "data" / "raw" / "open_books" / "source_lock.json"
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--candidates", type=Path, default=DEFAULT_CANDIDATES)
parser.add_argument("--lock", type=Path, default=DEFAULT_LOCK)
args = parser.parse_args()
candidates = {
row["pdf_url"]: row
for line in args.candidates.read_text(encoding="utf-8").splitlines()
if line.strip()
for row in [json.loads(line)]
}
records = json.loads(args.lock.read_text(encoding="utf-8"))
publisher_counts: dict[str, int] = {}
matched = 0
for record in records:
candidate = candidates.get(record["url"])
publishers = (candidate or {}).get("publishers", [])
record["publishers"] = publishers
record["corpus_tier"] = "supplemental_open_book"
if candidate:
matched += 1
for publisher in publishers or ["unknown"]:
publisher_counts[publisher] = publisher_counts.get(publisher, 0) + 1
args.lock.write_text(
json.dumps(records, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
)
print(
json.dumps(
{
"records": len(records),
"matched_to_candidates": matched,
"corpus_tier": "supplemental_open_book",
"publisher_counts": dict(sorted(publisher_counts.items())),
},
indent=2,
ensure_ascii=False,
)
)
if __name__ == "__main__":
main()
|