49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""Provenance of shared answer content, independent of Gemini corrections."""
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
|
|
from .json_io import atomic_update_json, read_json
|
|
|
|
|
|
def _path(root):
|
|
return Path(root) / ".copienator" / "answer_origins.json"
|
|
|
|
|
|
def origins_for(root, copy_id, label, pdf):
|
|
entry = read_json(_path(root), default={}).get(str(copy_id), {}).get(label, {})
|
|
return _valid_origins(entry, pdf)
|
|
|
|
|
|
def _valid_origins(entry, pdf):
|
|
if not entry or not Path(pdf).is_file():
|
|
return []
|
|
if entry.get("sha256") != sha256(Path(pdf).read_bytes()).hexdigest():
|
|
return []
|
|
return entry.get("labels", [])
|
|
|
|
|
|
def populated_labels(root, copy_id, source_label):
|
|
"""Find current answer PDFs populated from this source's recorded content."""
|
|
entries = read_json(_path(root), default={}).get(str(copy_id), {})
|
|
copies = Path(root) / "Copies" / f"Copie{copy_id}"
|
|
populated = []
|
|
for label, entry in entries.items():
|
|
if label == source_label or source_label not in entry.get("labels", []):
|
|
continue
|
|
pdf = copies / f"{label}.pdf"
|
|
if not pdf.is_file():
|
|
pdf = copies / f"{label}_new.pdf"
|
|
if source_label in _valid_origins(entry, pdf):
|
|
populated.append(label)
|
|
return sorted(populated)
|
|
|
|
|
|
def save_origins(root, copy_id, label, pdf, origins):
|
|
entry = {"labels": list(dict.fromkeys(origins)),
|
|
"sha256": sha256(Path(pdf).read_bytes()).hexdigest()}
|
|
|
|
def update(data):
|
|
data.setdefault(str(copy_id), {})[label] = entry
|
|
|
|
atomic_update_json(_path(root), update, default_factory=dict)
|