export_1c_help/export1c_help/lifecycle.py

"""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