195 lines
6.9 KiB
Python
195 lines
6.9 KiB
Python
#!/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())
|