deerflow-code/offline-backend-20260512/backend/skills/public/_shared/fast_patch_page.py

#!/usr/bin/env python3
"""Apply a compact JSON patch spec and render a fixed-template page once."""

from __future__ import annotations

import argparse
import copy
import json
from pathlib import Path
from types import ModuleType
from typing import Any


class FastSpecError(ValueError):
    """Raised when a compact page spec cannot be applied safely."""


def _read_spec(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise FastSpecError(f"Invalid spec JSON: {exc}") from exc
    if not isinstance(value, dict):
        raise FastSpecError("Spec JSON must be an object")
    return value


def _deep_merge(base: Any, patch: Any) -> Any:
    if not isinstance(base, dict) or not isinstance(patch, dict):
        return copy.deepcopy(patch)
    merged = copy.deepcopy(base)
    for key, value in patch.items():
        if key in merged:
            merged[key] = _deep_merge(merged[key], value)
        else:
            merged[key] = copy.deepcopy(value)
    return merged


def _set_existing_path(data: dict[str, Any], path: str, value: Any) -> None:
    parts = path.split(".")
    if not parts or any(not part for part in parts):
        raise FastSpecError(f"Invalid $set path: {path!r}")
    current: Any = data
    for index, part in enumerate(parts[:-1]):
        prefix = ".".join(parts[: index + 1])
        if isinstance(current, dict):
            if part not in current:
                raise FastSpecError(f"$set path does not exist: {prefix}")
            current = current[part]
        elif isinstance(current, list):
            if not part.isdigit() or int(part) >= len(current):
                raise FastSpecError(f"$set list index does not exist: {prefix}")
            current = current[int(part)]
        else:
            raise FastSpecError(f"$set path is not traversable: {prefix}")
    final = parts[-1]
    if isinstance(current, dict):
        if final not in current:
            raise FastSpecError(f"$set path does not exist: {path}")
        current[final] = copy.deepcopy(value)
    elif isinstance(current, list):
        if not final.isdigit() or int(final) >= len(current):
            raise FastSpecError(f"$set list index does not exist: {path}")
        current[int(final)] = copy.deepcopy(value)
    else:
        raise FastSpecError(f"$set parent is not assignable: {path}")


def apply_spec(
    data: dict[str, Any],
    spec: dict[str, Any],
    modules: tuple[str, ...],
) -> dict[str, Any]:
    allowed = set(modules) | {"$set"}
    unknown = sorted(set(spec) - allowed)
    if unknown:
        raise FastSpecError("Unknown top-level spec keys: " + ", ".join(unknown))
    result = copy.deepcopy(data)
    for module in modules:
        if module in spec:
            patch = spec[module]
            if not isinstance(patch, dict):
                raise FastSpecError(f"{module} must be an object")
            result[module] = _deep_merge(result[module], patch)
    assignments = spec.get("$set", {})
    if not isinstance(assignments, dict):
        raise FastSpecError("$set must be an object mapping paths to values")
    for path, value in assignments.items():
        if not isinstance(path, str):
            raise FastSpecError("Every $set key must be a string path")
        if path.split(".", 1)[0] not in modules:
            raise FastSpecError(f"$set path must start with a page module: {path}")
        _set_existing_path(result, path, value)
    return result


def _next_paths(output_dir: Path, stem: str) -> tuple[Path, Path]:
    for index in range(1, 1001):
        candidate = stem if index == 1 else f"{stem}-{index}"
        data_path = output_dir / f"{candidate}-data.json"
        html_path = output_dir / f"{candidate}.html"
        if not data_path.exists() and not html_path.exists():
            return data_path, html_path
    raise FastSpecError("Unable to allocate output filenames")


def _emit(renderer: ModuleType, payload: dict[str, Any]) -> None:
    emitter = getattr(renderer, "_emit", None)
    if callable(emitter):
        emitter(payload)
    else:
        print(json.dumps(payload, ensure_ascii=False))


def run(
    renderer: ModuleType,
    *,
    output_stem: str,
    argv: list[str] | None = None,
) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--spec", type=Path, required=True)
    parser.add_argument("--output-dir", type=Path, required=True)
    args = parser.parse_args(argv)
    validation_error = getattr(renderer, "DataValidationError", ValueError)
    try:
        default_path = getattr(
            renderer,
            "DEFAULT_DATA_PATH",
            getattr(renderer, "DEFAULT_DATA", None),
        )
        if default_path is None:
            raise FastSpecError("Renderer does not expose a default data path")
        data = apply_spec(
            renderer._read_json(default_path),
            _read_spec(args.spec),
            tuple(renderer.MODULES),
        )
        summary = renderer.validate_document(data)
        rendered = renderer.render_document(
            data,
            renderer.TEMPLATE_PATH.read_text(encoding="utf-8"),
            renderer.RUNTIME_PATH.read_text(encoding="utf-8"),
        )
        data_path, html_path = _next_paths(args.output_dir, output_stem)
        renderer._write_new(
            data_path,
            json.dumps(data, ensure_ascii=False, indent=2) + "\n",
        )
        renderer._write_new(html_path, rendered)
    except (FastSpecError, validation_error, OSError, UnicodeError) as exc:
        _emit(
            renderer,
            {
                "status": "error",
                "action": "fast-page-spec",
                "error": str(exc),
            },
        )
        return 2
    _emit(
        renderer,
        {
            "status": "ok",
            "action": "fast-page-spec",
            "data": str(data_path),
            "html": str(html_path),
            **summary,
        },
    )
    return 0