#!/usr/bin/env python3 """Return one changed 1C object from a saved-state object report.""" from __future__ import annotations import argparse import base64 import json import re 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 decode_arg(value: str | None, encoded: str | None) -> str | None: if encoded: return base64.b64decode(encoded).decode("utf-8") return value def normalize(value: str | None) -> str: return re.sub(r"[\s._-]+", "", str(value or "")).casefold() def linked_path(report_path: Path, value: Any) -> Path | None: if not value: return None path = Path(str(value)) if path.is_absolute(): return path return (report_path.parent / path).resolve() def score_match(item: dict[str, Any], query: str) -> tuple[float, str] | None: wanted = normalize(query) full_name = str(item.get("full_name") or "") name = str(item.get("name") or "") synonym = str(item.get("synonym") or "") if normalize(full_name) == wanted: return 1.0, "full_name" if normalize(name) == wanted: return 0.96, "name" if normalize(synonym) == wanted: return 0.92, "synonym" if wanted and normalize(full_name).endswith(wanted): return 0.88, "full_name_suffix" if wanted and wanted in normalize(full_name): return 0.72, "full_name_contains" if wanted and wanted in normalize(name): return 0.68, "name_contains" if wanted and wanted in normalize(synonym): return 0.64, "synonym_contains" return None def compact_candidate(item: dict[str, Any], score: float, match_by: str) -> dict[str, Any]: return { "score": score, "match_by": match_by, "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"), "parts_count": item.get("parts_count"), "text_diff_parts": item.get("text_diff_parts"), "active_missing_parts": item.get("active_missing_parts"), } def matching_items(report: dict[str, Any], query: str) -> list[dict[str, Any]]: items = ((report.get("agent_summary") or {}).get("object_changes") or []) matches: list[dict[str, Any]] = [] for item in items: if not isinstance(item, dict): continue scored = score_match(item, query) if not scored: continue score, match_by = scored matches.append({"score": score, "match_by": match_by, "item": item}) matches.sort(key=lambda row: (-row["score"], str((row["item"] or {}).get("full_name") or ""))) return matches def by_full_name(items: list[dict[str, Any]], full_name: str) -> dict[str, Any] | None: wanted = normalize(full_name) for item in items: if normalize(str(item.get("full_name") or "")) == wanted: return item return None def trim_diff(detail: dict[str, Any] | None, max_diff_lines: int) -> dict[str, Any] | None: if detail is None: return None copy = json.loads(json.dumps(detail, ensure_ascii=False)) for part in copy.get("details") or []: payload = part.get("payload") or {} diff = payload.get("text_diff") or {} lines = diff.get("unified_diff") if isinstance(lines, list) and len(lines) > max_diff_lines: diff["unified_diff"] = [*lines[:max_diff_lines], f"... truncated {len(lines) - max_diff_lines} lines ..."] return copy def build_change(report_path: Path, query: str, *, max_diff_lines: int) -> dict[str, Any]: report = load_json(report_path) matches = matching_items(report, query) if not matches: return { "schema": "onec_saved_state_object_change.v1", "report": str(report_path), "query": query, "status": "not_found", "matched": False, "available_objects": (report.get("agent_summary") or {}).get("object_names") or [], } best_score = matches[0]["score"] best = [row for row in matches if row["score"] == best_score] if len(best) > 1 and best_score < 1.0: return { "schema": "onec_saved_state_object_change.v1", "report": str(report_path), "query": query, "status": "ambiguous", "matched": False, "candidates": [compact_candidate(row["item"], row["score"], row["match_by"]) for row in best], "available_objects": (report.get("agent_summary") or {}).get("object_names") or [], } selected = matches[0] summary = selected["item"] full_name = str(summary.get("full_name") or "") comparison = load_json(linked_path(report_path, report.get("comparison"))) if report.get("comparison") else {} detail = load_json(linked_path(report_path, report.get("detail"))) if report.get("detail") else {} comparison_change = by_full_name(comparison.get("object_changes") or [], full_name) object_detail = by_full_name(detail.get("object_details") or [], full_name) return { "schema": "onec_saved_state_object_change.v1", "report": str(report_path), "query": query, "status": "matched", "matched": True, "match": { "score": selected["score"], "match_by": selected["match_by"], "full_name": full_name, }, "object_summary": summary, "comparison_change": comparison_change, "detail": trim_diff(object_detail, max_diff_lines=max_diff_lines), "safety": { "read_only": True, "sql_write_performed": False, "public_terms_are_1c_objects": True, }, } def main() -> int: parser = argparse.ArgumentParser(description="Get one 1C saved-state changed object by name.") parser.add_argument("--report", type=Path, required=True) parser.add_argument("--name") parser.add_argument("--name-b64") parser.add_argument("--max-diff-lines", type=int, default=40) parser.add_argument("--output", type=Path) args = parser.parse_args() name = decode_arg(args.name, args.name_b64) if not name: raise SystemExit("Use --name or --name-b64.") result = build_change(args.report, name, max_diff_lines=args.max_diff_lines) if args.output: write_json(args.output, result) print(json.dumps({"output": str(args.output) if args.output else None, "schema": result["schema"], "status": result["status"], "matched": result["matched"]}, ensure_ascii=False)) return 0 if result["matched"] else 2 if __name__ == "__main__": raise SystemExit(main())