integrity_scanner_fuer_statische_Webseiten/scanner/baseline.py

"""
Snapshot and baseline management.

Snapshots are timestamped crawl results.  The baseline is a manually
approved reference state.  Compromised content must never flow into the
baseline automatically — every update requires an explicit `approve` call.
"""
import hashlib
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

logger = logging.getLogger(__name__)


def url_key(url: str) -> str:
    """Stable filename-safe key for a URL (first 16 hex chars of SHA-256)."""
    return hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]


class BaselineManager:
    def __init__(self, data_dir: str | Path):
        self.data_dir = Path(data_dir)
        self.baseline_dir = self.data_dir / "baseline"
        self.pages_dir = self.baseline_dir / "pages"
        self.snapshots_dir = self.data_dir / "snapshots"

    # ------------------------------------------------------------------
    # Snapshot operations
    # ------------------------------------------------------------------

    def save_snapshot(self, pages: list[dict], errors: list[dict] | None = None) -> Path:
        """
        Persist a crawl+extraction result as a timestamped snapshot.
        Crawl errors (network failures) are recorded in the manifest so they
        survive into the comparison/report stage.
        Returns the snapshot directory path.
        """
        ts = _utc_stamp()
        snap_dir = self.snapshots_dir / ts
        snap_pages_dir = snap_dir / "pages"
        snap_pages_dir.mkdir(parents=True, exist_ok=True)

        manifest = {
            "timestamp": ts,
            "page_count": len(pages),
            "urls": [p["url"] for p in pages],
            "errors": errors or [],
        }
        (snap_dir / "manifest.json").write_text(
            json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
        )
        for page in pages:
            fname = url_key(page["url"]) + ".json"
            (snap_pages_dir / fname).write_text(
                json.dumps(page, indent=2, ensure_ascii=False), encoding="utf-8"
            )

        logger.info("Snapshot saved: %s (%d pages)", snap_dir, len(pages))
        return snap_dir

    def latest_snapshot_dir(self) -> Optional[Path]:
        """Return the most recent snapshot directory, or None."""
        if not self.snapshots_dir.exists():
            return None
        dirs = sorted(
            (d for d in self.snapshots_dir.iterdir() if d.is_dir()),
            key=lambda d: d.name,
        )
        return dirs[-1] if dirs else None

    def load_snapshot(self, snap_dir: Optional[Path] = None) -> dict:
        """
        Load a snapshot (default: latest).
        Returns {manifest, pages: {url: page_dict}, errors: list, dir: Path}.
        """
        if snap_dir is None:
            snap_dir = self.latest_snapshot_dir()
        if snap_dir is None or not snap_dir.exists():
            return {}

        manifest = json.loads((snap_dir / "manifest.json").read_text(encoding="utf-8"))
        pages: dict[str, dict] = {}
        pages_dir = snap_dir / "pages"
        for f in pages_dir.glob("*.json"):
            page = json.loads(f.read_text(encoding="utf-8"))
            pages[page["url"]] = page

        return {
            "manifest": manifest,
            "pages": pages,
            "errors": manifest.get("errors", []),
            "dir": snap_dir,
        }

    # ------------------------------------------------------------------
    # Baseline operations
    # ------------------------------------------------------------------

    def baseline_exists(self) -> bool:
        return (self.baseline_dir / "manifest.json").exists()

    def load_baseline(self) -> dict:
        """
        Load the current approved baseline.
        Returns {manifest, pages: {url: page_dict}}.
        """
        if not self.baseline_exists():
            return {"manifest": {}, "pages": {}}

        manifest = json.loads(
            (self.baseline_dir / "manifest.json").read_text(encoding="utf-8")
        )
        pages: dict[str, dict] = {}
        if self.pages_dir.exists():
            for f in self.pages_dir.glob("*.json"):
                page = json.loads(f.read_text(encoding="utf-8"))
                pages[page["url"]] = page

        return {"manifest": manifest, "pages": pages}

    # ------------------------------------------------------------------
    # Approve operations  (the only way baseline content is ever updated)
    # ------------------------------------------------------------------

    def approve_all(
        self,
        snap_dir: Optional[Path] = None,
        note: str = "",
        approved_by: str = "cli",
    ) -> list[str]:
        """
        Approve every page in the snapshot as the new baseline.
        Existing baseline pages NOT present in snapshot are kept unless
        --rebuild is requested (caller responsibility to wipe baseline_dir first).
        """
        snap = self.load_snapshot(snap_dir)
        if not snap:
            raise RuntimeError("No snapshot found to approve.")

        self.pages_dir.mkdir(parents=True, exist_ok=True)
        approved: list[str] = []

        for url, page in snap["pages"].items():
            fname = url_key(url) + ".json"
            (self.pages_dir / fname).write_text(
                json.dumps(page, indent=2, ensure_ascii=False), encoding="utf-8"
            )
            approved.append(url)

        self._write_manifest(approved, note, approved_by, snap.get("dir"))
        logger.info("Approved %d URLs into baseline.", len(approved))
        return approved

    def approve_url(
        self,
        url: str,
        snap_dir: Optional[Path] = None,
        note: str = "",
        approved_by: str = "cli",
    ) -> bool:
        """Approve a single URL from snapshot into the baseline."""
        snap = self.load_snapshot(snap_dir)
        if url not in snap.get("pages", {}):
            logger.error("URL not found in snapshot: %s", url)
            return False

        self.pages_dir.mkdir(parents=True, exist_ok=True)
        fname = url_key(url) + ".json"
        (self.pages_dir / fname).write_text(
            json.dumps(snap["pages"][url], indent=2, ensure_ascii=False),
            encoding="utf-8",
        )

        # Update manifest incrementally
        if self.baseline_exists():
            manifest = json.loads(
                (self.baseline_dir / "manifest.json").read_text(encoding="utf-8")
            )
        else:
            manifest = {"urls": []}

        if url not in manifest.get("urls", []):
            manifest.setdefault("urls", []).append(url)

        manifest["last_updated"] = datetime.now(timezone.utc).isoformat()
        manifest["last_note"] = note
        manifest["last_approved_by"] = approved_by

        (self.baseline_dir / "manifest.json").write_text(
            json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
        )
        logger.info("Approved single URL into baseline: %s", url)
        return True

    def rebuild_baseline(
        self,
        snap_dir: Optional[Path] = None,
        note: str = "",
        approved_by: str = "cli",
    ) -> list[str]:
        """
        Wipe the entire baseline and replace it with the given snapshot.
        Use when a major legitimate redesign has taken place.
        """
        import shutil
        if self.baseline_dir.exists():
            shutil.rmtree(self.baseline_dir)
        self.pages_dir.mkdir(parents=True, exist_ok=True)
        return self.approve_all(snap_dir=snap_dir, note=note, approved_by=approved_by)

    # ------------------------------------------------------------------
    # Asset hash baseline
    # ------------------------------------------------------------------

    @property
    def _asset_hashes_path(self) -> Path:
        return self.baseline_dir / "asset_hashes.json"

    def load_asset_hashes(self) -> dict:
        """Load approved asset hashes. Returns {} if none approved yet."""
        if not self._asset_hashes_path.exists():
            return {}
        return json.loads(self._asset_hashes_path.read_text(encoding="utf-8"))

    def save_asset_hashes(self, hashes: dict) -> None:
        """Persist asset hashes as the new approved state."""
        self.baseline_dir.mkdir(parents=True, exist_ok=True)
        self._asset_hashes_path.write_text(
            json.dumps(hashes, indent=2, ensure_ascii=False), encoding="utf-8"
        )
        logger.info("Asset hashes saved: %d entries", len(hashes))

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _write_manifest(
        self,
        urls: list[str],
        note: str,
        approved_by: str,
        source_snap: Optional[Path],
    ) -> None:
        self.baseline_dir.mkdir(parents=True, exist_ok=True)
        manifest = {
            "approved_at": datetime.now(timezone.utc).isoformat(),
            "approved_by": approved_by,
            "note": note,
            "source_snapshot": str(source_snap) if source_snap else "",
            "urls": urls,
        }
        (self.baseline_dir / "manifest.json").write_text(
            json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
        )


def _utc_stamp() -> str:
    return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")