llm/scripts/list_1c_saved_state_object_changes.py

#!/usr/bin/env python3
"""List changed 1C objects from a saved-state object report."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Any


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8-sig"))


def write_json(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def has_payload_role(item: dict[str, Any], role: str | None) -> bool:
    if not role:
        return True
    return any((part or {}).get("payload_role") == role for part in item.get("parts") or [])


def has_text_diff(item: dict[str, Any], required: bool | None) -> bool:
    if required is None:
        return True
    value = int(item.get("text_diff_parts") or 0) > 0
    return value is required


def has_active_missing(item: dict[str, Any], required: bool | None) -> bool:
    if required is None:
        return True
    value = int(item.get("active_missing_parts") or 0) > 0
    return value is required


def compact_object(item: dict[str, Any]) -> dict[str, Any]:
    return {
        "full_name": item.get("full_name"),
        "layer": item.get("layer"),
        "extension": item.get("extension"),
        "kind": item.get("kind"),
        "kind_ru": item.get("kind_ru"),
        "name": item.get("name"),
        "synonym": item.get("synonym"),
        "change_state": item.get("change_state"),
        "parts_count": item.get("parts_count"),
        "text_diff_parts": item.get("text_diff_parts"),
        "active_missing_parts": item.get("active_missing_parts"),
        "added_terms": item.get("added_terms") or [],
        "removed_terms": item.get("removed_terms") or [],
        "payload_roles": sorted({
            str((part or {}).get("payload_role"))
            for part in item.get("parts") or []
            if (part or {}).get("payload_role")
        }),
        "parts": [
            {
                "file_name": part.get("file_name"),
                "payload_role": part.get("payload_role"),
                "active_exists": part.get("active_exists"),
                "summary": part.get("summary"),
                "delta_chars": part.get("delta_chars"),
            }
            for part in item.get("parts") or []
            if isinstance(part, dict)
        ],
    }


def build_list(
    report_path: Path,
    *,
    layer: str | None,
    kind: str | None,
    extension: str | None,
    payload_role: str | None,
    active_missing: bool | None,
    text_diff: bool | None,
) -> dict[str, Any]:
    report = load_json(report_path)
    summary = report.get("agent_summary") or {}
    source_items = summary.get("object_changes") or []
    items: list[dict[str, Any]] = []
    for item in source_items:
        if not isinstance(item, dict):
            continue
        if layer and str(item.get("layer") or "").casefold() != layer.casefold():
            continue
        if kind:
            wanted = kind.casefold()
            if str(item.get("kind") or "").casefold() != wanted and str(item.get("kind_ru") or "").casefold() != wanted:
                continue
        if extension and str(item.get("extension") or "").casefold() != extension.casefold():
            continue
        if not has_payload_role(item, payload_role):
            continue
        if not has_active_missing(item, active_missing):
            continue
        if not has_text_diff(item, text_diff):
            continue
        items.append(compact_object(item))

    return {
        "schema": "onec_saved_state_object_change_list.v1",
        "report": str(report_path),
        "database": report.get("database"),
        "filters": {
            "layer": layer,
            "kind": kind,
            "extension": extension,
            "payload_role": payload_role,
            "active_missing": active_missing,
            "text_diff": text_diff,
        },
        "objects": items,
        "system_changes": summary.get("system_changes") or [],
        "counts": {
            "objects": len(items),
            "total_objects": len(source_items),
            "system_changes": len(summary.get("system_changes") or []),
        },
        "safety": {
            "read_only": True,
            "sql_write_performed": False,
            "public_terms_are_1c_objects": True,
        },
    }


def optional_bool(value: str | None) -> bool | None:
    if value is None:
        return None
    normalized = value.strip().casefold()
    if normalized in {"1", "true", "yes", "y", "да"}:
        return True
    if normalized in {"0", "false", "no", "n", "нет"}:
        return False
    raise argparse.ArgumentTypeError(f"Expected boolean value, got: {value}")


def main() -> int:
    parser = argparse.ArgumentParser(description="List changed 1C objects from a saved-state object report.")
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--layer")
    parser.add_argument("--kind")
    parser.add_argument("--extension")
    parser.add_argument("--payload-role")
    parser.add_argument("--active-missing", type=optional_bool)
    parser.add_argument("--text-diff", type=optional_bool)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()

    result = build_list(
        args.report,
        layer=args.layer,
        kind=args.kind,
        extension=args.extension,
        payload_role=args.payload_role,
        active_missing=args.active_missing,
        text_diff=args.text_diff,
    )
    if args.output:
        write_json(args.output, result)
    print(json.dumps({"output": str(args.output) if args.output else None, "schema": result["schema"], "counts": result["counts"]}, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())