import difflib import json from datetime import datetime, timezone from pathlib import Path STORE_PATH = Path(__file__).parent / "site_content.json" MAX_SNAPSHOTS = 10 # Multi-page store — one snapshot per URL, refreshed together MULTI_PAGE_STORE = Path(__file__).parent / "multi_page_content.json" def _load_store() -> list[dict]: if not STORE_PATH.exists(): return [] with STORE_PATH.open("r", encoding="utf-8") as f: return json.load(f) def _write_store(snapshots: list[dict]) -> None: with STORE_PATH.open("w", encoding="utf-8") as f: json.dump(snapshots, f, ensure_ascii=False, indent=2) def save_site_content(data: dict) -> None: snapshots = _load_store() entry = {**data, "saved_at": datetime.now(timezone.utc).isoformat()} snapshots.append(entry) snapshots = snapshots[-MAX_SNAPSHOTS:] _write_store(snapshots) def load_latest_content() -> dict: snapshots = _load_store() if not snapshots: return {} return snapshots[-1] def get_content_diff(snapshot_1: dict, snapshot_2: dict) -> dict: text_a = (snapshot_1.get("body_text") or "").splitlines() text_b = (snapshot_2.get("body_text") or "").splitlines() diff_lines = list( difflib.unified_diff(text_a, text_b, lineterm="", n=2) ) if not diff_lines: return {"is_changed": False, "summary_of_changes": "No changes detected."} added = [l[1:] for l in diff_lines if l.startswith("+") and not l.startswith("+++")] removed = [l[1:] for l in diff_lines if l.startswith("-") and not l.startswith("---")] parts = [] if removed: parts.append(f"Removed ({len(removed)} line(s)):\n" + "\n".join(f" - {l}" for l in removed[:5])) if added: parts.append(f"Added ({len(added)} line(s)):\n" + "\n".join(f" + {l}" for l in added[:5])) if len(removed) > 5 or len(added) > 5: parts.append("(truncated — showing first 5 lines per section)") return {"is_changed": True, "summary_of_changes": "\n".join(parts)} def is_content_stale(max_age_minutes: int = 60) -> bool: latest = load_latest_content() if not latest: return True saved_at_str = latest.get("saved_at") or latest.get("scraped_at") if not saved_at_str: return True saved_at = datetime.fromisoformat(saved_at_str) age_minutes = (datetime.now(timezone.utc) - saved_at).total_seconds() / 60 return age_minutes > max_age_minutes # ── Multi-page store ───────────────────────────────────────────────────────── def save_all_pages(pages: list[dict]) -> None: """Save a fresh multi-page snapshot keyed by URL.""" data = { "pages": {p["url"]: p for p in pages}, "saved_at": datetime.now(timezone.utc).isoformat(), } MULTI_PAGE_STORE.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def load_all_pages() -> dict[str, dict]: """Return the latest multi-page snapshot as {url: page_data}.""" if not MULTI_PAGE_STORE.exists(): return {} data = json.loads(MULTI_PAGE_STORE.read_text(encoding="utf-8")) return data.get("pages", {}) def is_multi_page_stale(max_age_minutes: int = 60) -> bool: if not MULTI_PAGE_STORE.exists(): return True data = json.loads(MULTI_PAGE_STORE.read_text(encoding="utf-8")) saved_at_str = data.get("saved_at") if not saved_at_str: return True saved_at = datetime.fromisoformat(saved_at_str) age_minutes = (datetime.now(timezone.utc) - saved_at).total_seconds() / 60 return age_minutes > max_age_minutes