Spaces:
Sleeping
Sleeping
| 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 | |