297 lines
10 KiB
Python
297 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Resolve a 1C metadata object by configurator-visible kind/name."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
KIND_ALIASES = {
|
|
"документ": "Document",
|
|
"documents": "Document",
|
|
"document": "Document",
|
|
"справочник": "Catalog",
|
|
"catalogs": "Catalog",
|
|
"catalog": "Catalog",
|
|
"перечисление": "Enum",
|
|
"enums": "Enum",
|
|
"enum": "Enum",
|
|
"регистрсведений": "InformationRegister",
|
|
"регистр сведений": "InformationRegister",
|
|
"informationregister": "InformationRegister",
|
|
"informationregisters": "InformationRegister",
|
|
"регистрнакопления": "AccumulationRegister",
|
|
"регистр накопления": "AccumulationRegister",
|
|
"accumulationregister": "AccumulationRegister",
|
|
"accumulationregisters": "AccumulationRegister",
|
|
"регистрбухгалтерии": "AccountingRegister",
|
|
"регистр бухгалтерии": "AccountingRegister",
|
|
"accountingregister": "AccountingRegister",
|
|
"accountingregisters": "AccountingRegister",
|
|
"отчет": "Report",
|
|
"reports": "Report",
|
|
"report": "Report",
|
|
"обработка": "DataProcessor",
|
|
"dataprocessor": "DataProcessor",
|
|
"dataprocessors": "DataProcessor",
|
|
"общиймодуль": "CommonModule",
|
|
"общий модуль": "CommonModule",
|
|
"commonmodule": "CommonModule",
|
|
"commonmodules": "CommonModule",
|
|
"форма": "Form",
|
|
"forms": "Form",
|
|
"form": "Form",
|
|
"макет": "Template",
|
|
"templates": "Template",
|
|
"template": "Template",
|
|
"команда": "Command",
|
|
"commands": "Command",
|
|
"command": "Command",
|
|
}
|
|
|
|
|
|
GENERATED_TYPE_KIND = {
|
|
"DocumentObject": "Document",
|
|
"DocumentRef": "Document",
|
|
"DocumentManager": "Document",
|
|
"DocumentSelection": "Document",
|
|
"DocumentList": "Document",
|
|
"CatalogObject": "Catalog",
|
|
"CatalogRef": "Catalog",
|
|
"CatalogManager": "Catalog",
|
|
"CatalogSelection": "Catalog",
|
|
"CatalogList": "Catalog",
|
|
"EnumRef": "Enum",
|
|
"EnumManager": "Enum",
|
|
"InformationRegisterRecordSet": "InformationRegister",
|
|
"InformationRegisterManager": "InformationRegister",
|
|
"AccumulationRegisterRecordSet": "AccumulationRegister",
|
|
"AccumulationRegisterManager": "AccumulationRegister",
|
|
"AccountingRegisterRecordSet": "AccountingRegister",
|
|
"AccountingRegisterManager": "AccountingRegister",
|
|
"ReportManager": "Report",
|
|
"DataProcessorManager": "DataProcessor",
|
|
}
|
|
|
|
|
|
TABLE_PREFIX_BY_ROLE = {
|
|
"Document": "_Document",
|
|
"DocumentChngR": "_DocumentChngR",
|
|
"Catalog": "_Reference",
|
|
"Reference": "_Reference",
|
|
"Enum": "_Enum",
|
|
"InfoRg": "_InfoRg",
|
|
"InfoRgChngR": "_InfoRgChngR",
|
|
"AccumRg": "_AccumRg",
|
|
"AccumRgT": "_AccumRgT",
|
|
"AccumRgOpt": "_AccumRgOpt",
|
|
"AccRg": "_AccRg",
|
|
"AccRgAT": "_AccRgAT",
|
|
"AccRgCT": "_AccRgCT",
|
|
"AccRgOpt": "_AccRgOpt",
|
|
"BPr": "_BPr",
|
|
"Task": "_Task",
|
|
}
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
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 canonical_kind(value: str | None) -> str | None:
|
|
if not value:
|
|
return None
|
|
stripped = value.strip()
|
|
return KIND_ALIASES.get(normalize(stripped), stripped)
|
|
|
|
|
|
def parse_query(kind: str | None, name: str) -> tuple[str | None, str]:
|
|
kind = canonical_kind(kind)
|
|
query = name.strip()
|
|
if "." not in query:
|
|
return kind, query
|
|
left, right = query.split(".", 1)
|
|
if left.startswith("cfg:"):
|
|
left = left[4:]
|
|
generated_kind = GENERATED_TYPE_KIND.get(left)
|
|
if generated_kind:
|
|
return kind or generated_kind, right
|
|
parsed_kind = canonical_kind(left)
|
|
return kind or parsed_kind, right
|
|
|
|
|
|
def physical_name(route: dict[str, Any]) -> str | None:
|
|
role = str(route.get("storage_role") or "")
|
|
number = route.get("sql_number")
|
|
if number is None:
|
|
return None
|
|
prefix = TABLE_PREFIX_BY_ROLE.get(role)
|
|
if prefix:
|
|
return f"{prefix}{number}"
|
|
if role == "VT":
|
|
return None
|
|
if role == "Fld":
|
|
return f"_Fld{number}"
|
|
if role == "LineNo":
|
|
return f"_LineNo{number}"
|
|
return None
|
|
|
|
|
|
def is_extension_path(relative_path: str | None, path: str | None = None) -> bool:
|
|
relative_parts = str(relative_path or "").replace("/", "\\").split("\\")
|
|
if relative_parts and relative_parts[0].casefold() in {"расширения", "extensions"}:
|
|
return True
|
|
full_parts = [part.casefold() for part in str(path or "").replace("/", "\\").split("\\")]
|
|
return "расширения" in full_parts or "extensions" in full_parts
|
|
|
|
|
|
def top_objects(item: dict[str, Any]) -> list[dict[str, Any]]:
|
|
return item.get("xml_top_objects") or []
|
|
|
|
|
|
def match_top(top: dict[str, Any], *, kind: str | None, wanted: str) -> tuple[float, str] | None:
|
|
top_kind = canonical_kind(str(top.get("xml_kind") or ""))
|
|
if kind and top_kind != kind:
|
|
return None
|
|
wanted_norm = normalize(wanted)
|
|
name = str(top.get("name") or "")
|
|
synonym = str(top.get("synonym") or "")
|
|
relative_path = str(top.get("relative_path") or "")
|
|
if normalize(name) == wanted_norm:
|
|
return (1.0, "xml_top_object.name")
|
|
if normalize(synonym) == wanted_norm:
|
|
return (0.95, "xml_top_object.synonym")
|
|
if wanted_norm and wanted_norm in normalize(name):
|
|
return (0.82, "xml_top_object.name_contains")
|
|
if wanted_norm and wanted_norm in normalize(synonym):
|
|
return (0.78, "xml_top_object.synonym_contains")
|
|
if wanted_norm and wanted_norm in normalize(relative_path):
|
|
return (0.65, "xml_top_object.relative_path")
|
|
return None
|
|
|
|
|
|
def enrich_dbnames(routes: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
result = []
|
|
for route in routes:
|
|
copy = dict(route)
|
|
name = physical_name(copy)
|
|
if name:
|
|
copy["physical_name"] = name
|
|
result.append(copy)
|
|
return result
|
|
|
|
|
|
def compact_match(guid: str, item: dict[str, Any], top: dict[str, Any], *, score: float, match_by: str) -> dict[str, Any]:
|
|
return {
|
|
"score": score,
|
|
"match_by": match_by,
|
|
"guid": guid,
|
|
"kind": canonical_kind(str(top.get("xml_kind") or "")),
|
|
"name": top.get("name"),
|
|
"synonym": top.get("synonym"),
|
|
"source": "extension" if is_extension_path(top.get("relative_path"), top.get("path")) else "base",
|
|
"relative_path": top.get("relative_path"),
|
|
"path": top.get("path"),
|
|
"route_kind": item.get("route_kind") or [],
|
|
"storage": {
|
|
"dbnames": enrich_dbnames(item.get("dbnames") or []),
|
|
"config_routes": item.get("config_routes") or [],
|
|
"extension_routes": item.get("extension_routes") or [],
|
|
},
|
|
"xml_occurrence_count": item.get("xml_occurrence_count"),
|
|
}
|
|
|
|
|
|
def find_matches(index: dict[str, Any], *, kind: str | None, name: str, limit: int) -> list[dict[str, Any]]:
|
|
wanted_kind, wanted_name = parse_query(kind, name)
|
|
matches = []
|
|
for guid, item in (index.get("objects") or {}).items():
|
|
for top in top_objects(item):
|
|
match = match_top(top, kind=wanted_kind, wanted=wanted_name)
|
|
if not match:
|
|
continue
|
|
score, match_by = match
|
|
matches.append(compact_match(guid, item, top, score=score, match_by=match_by))
|
|
matches.sort(
|
|
key=lambda row: (
|
|
-row["score"],
|
|
row["source"] != "base",
|
|
row.get("relative_path") or "",
|
|
row.get("guid") or "",
|
|
)
|
|
)
|
|
return matches[:limit]
|
|
|
|
|
|
def resolve_object(index: dict[str, Any], *, kind: str | None, name: str, limit: int = 20) -> dict[str, Any]:
|
|
wanted_kind, wanted_name = parse_query(kind, name)
|
|
matches = find_matches(index, kind=wanted_kind, name=wanted_name, limit=limit)
|
|
exact_base = [row for row in matches if row["score"] >= 0.95 and row["source"] == "base"]
|
|
canonical = exact_base[0] if exact_base else (matches[0] if matches else None)
|
|
extension_overlays = [
|
|
row
|
|
for row in matches
|
|
if canonical
|
|
and row.get("source") == "extension"
|
|
and normalize(row.get("name")) == normalize(canonical.get("name"))
|
|
and row.get("kind") == canonical.get("kind")
|
|
]
|
|
return {
|
|
"schema": "onec_object_resolution.v1",
|
|
"query": {"kind": wanted_kind, "name": wanted_name, "raw_kind": kind, "raw_name": name},
|
|
"canonical": canonical,
|
|
"extension_overlays": extension_overlays,
|
|
"matches": matches,
|
|
"summary": {
|
|
"match_count": len(matches),
|
|
"has_canonical": canonical is not None,
|
|
"extension_overlay_count": len(extension_overlays),
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Resolve 1C metadata object by visible kind/name.")
|
|
parser.add_argument("--index", type=Path, required=True)
|
|
parser.add_argument("--kind")
|
|
parser.add_argument("--name")
|
|
parser.add_argument("--kind-b64")
|
|
parser.add_argument("--name-b64")
|
|
parser.add_argument("--limit", type=int, default=20)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
|
|
kind = decode_arg(args.kind, args.kind_b64)
|
|
name = decode_arg(args.name, args.name_b64)
|
|
if not name:
|
|
raise SystemExit("Use --name or --name-b64.")
|
|
|
|
result = resolve_object(load_json(args.index), kind=kind, name=name, limit=args.limit)
|
|
text = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(text, encoding="utf-8")
|
|
print(json.dumps({"output": str(args.output), "summary": result["summary"]}, ensure_ascii=False))
|
|
else:
|
|
print(text)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|