46 lines
1.9 KiB
Python
46 lines
1.9 KiB
Python
"""Restore source passages from a private archived review without changing the cohort."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def restore(items: list[dict], review: dict) -> list[dict]:
|
|
archived = review["items"]
|
|
by_id = {row["id"]: row for row in archived}
|
|
if len(by_id) != len(archived) or len({row["id"] for row in items}) != len(items):
|
|
raise ValueError("Duplicate item IDs")
|
|
restored = []
|
|
for item in items:
|
|
original = by_id[item["id"]]
|
|
comparable = dict(original)
|
|
if isinstance(comparable.get("imports"), str):
|
|
comparable["imports"] = json.loads(comparable["imports"])
|
|
if any(comparable.get(k) != v for k, v in item.items()):
|
|
raise ValueError(f"Archived review does not match item {item['id']}")
|
|
passage = original.get("source_text", "")
|
|
if not passage.strip():
|
|
raise ValueError(f"Missing archived passage for {item['id']}")
|
|
restored.append({**item, "source_text": passage})
|
|
return restored
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--items", required=True, type=Path)
|
|
parser.add_argument("--review", required=True, type=Path)
|
|
parser.add_argument("--out", required=True, type=Path)
|
|
args = parser.parse_args()
|
|
if args.out.parent.name != "data":
|
|
parser.error("Source-inclusive output must go in a gitignored data/ directory")
|
|
items = [json.loads(line) for line in args.items.read_text().splitlines() if line.strip()]
|
|
rows = restore(items, json.loads(args.review.read_text()))
|
|
args.out.parent.mkdir(parents=True, exist_ok=True)
|
|
args.out.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows))
|
|
print(f"Restored {len(rows)} passages; all exported fields are unchanged")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|