"""Page lifecycle tracking across wiki exports.""" from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from pathlib import Path from typing import Any @dataclass class PageRecord: meta_key: str title: str filename: str type_dir: str object: str form: str | None source: str content_hash: str created_in: str edited_in: str deleted_in: str | None = None status: str = "active" # active | deleted def to_dict(self) -> dict[str, Any]: d: dict[str, Any] = { "meta_key": self.meta_key, "title": self.title, "filename": self.filename, "type_dir": self.type_dir, "object": self.object, "form": self.form, "source": self.source, "content_hash": self.content_hash, "created_in": self.created_in, "edited_in": self.edited_in, "status": self.status, } if self.deleted_in: d["deleted_in"] = self.deleted_in return d @classmethod def from_dict(cls, d: dict[str, Any]) -> PageRecord: return cls( meta_key=d["meta_key"], title=d.get("title") or d["meta_key"], filename=d.get("filename") or "", type_dir=d.get("type_dir") or "", object=d.get("object") or "", form=d.get("form"), source=d.get("source") or "", content_hash=d.get("content_hash") or "", created_in=d.get("created_in") or d.get("config_version") or "unknown", edited_in=d.get("edited_in") or d.get("created_in") or "unknown", deleted_in=d.get("deleted_in"), status=d.get("status") or ("deleted" if d.get("deleted_in") else "active"), ) @dataclass class LifecycleResult: active: list[PageRecord] = field(default_factory=list) deleted: list[PageRecord] = field(default_factory=list) created_count: int = 0 edited_count: int = 0 unchanged_count: int = 0 deleted_count: int = 0 def content_hash(text: str) -> str: return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] def load_prev_manifest(path: Path | None) -> dict[str, Any] | None: if path is None or not path.is_file(): return None try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None def prev_records(manifest: dict[str, Any] | None) -> dict[str, PageRecord]: if not manifest: return {} out: dict[str, PageRecord] = {} for key in ("pages", "deleted_pages"): for item in manifest.get(key) or []: if not isinstance(item, dict) or "meta_key" not in item: continue rec = PageRecord.from_dict(item) # prefer active over deleted if both present if rec.meta_key in out and out[rec.meta_key].status == "active": continue out[rec.meta_key] = rec return out def merge_lifecycle( *, current: list[PageRecord], previous: dict[str, PageRecord], config_version: str, ) -> LifecycleResult: """ current — new active pages (created_in/edited_in placeholders ignored; content_hash must be set). """ result = LifecycleResult() current_keys = {r.meta_key for r in current} for cur in current: old = previous.get(cur.meta_key) if old is None or old.status == "deleted": cur.created_in = config_version cur.edited_in = config_version cur.deleted_in = None cur.status = "active" result.created_count += 1 elif not old.content_hash: # legacy manifest without hashes — baseline this export cur.created_in = ( old.created_in if old.created_in and old.created_in != "unknown" else config_version ) cur.edited_in = config_version cur.deleted_in = None cur.status = "active" result.unchanged_count += 1 elif old.content_hash == cur.content_hash: cur.created_in = old.created_in cur.edited_in = old.edited_in cur.deleted_in = None cur.status = "active" result.unchanged_count += 1 else: cur.created_in = old.created_in or config_version cur.edited_in = config_version cur.deleted_in = None cur.status = "active" result.edited_count += 1 result.active.append(cur) for key, old in previous.items(): if key in current_keys: continue if old.status == "deleted" and old.deleted_in: # already tombstoned — keep result.deleted.append(old) continue tomb = PageRecord( meta_key=old.meta_key, title=old.title, filename=old.filename, type_dir=old.type_dir, object=old.object, form=old.form, source=old.source, content_hash=old.content_hash, created_in=old.created_in, edited_in=old.edited_in, deleted_in=config_version, status="deleted", ) result.deleted.append(tomb) result.deleted_count += 1 result.active.sort(key=lambda r: r.meta_key) result.deleted.sort(key=lambda r: r.meta_key) return result