#!/usr/bin/env python3 """Build a read-projection metadata card from resolved XML object evidence.""" from __future__ import annotations import argparse import json import sys import xml.etree.ElementTree as ET from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parent)) from resolve_1c_object import load_json, physical_name, resolve_object # noqa: E402 REFERENCE_MARKERS = ("Ref.",) TYPE_PRESENTATION_RU = { "xs:string": "Строка", "xs:decimal": "Число", "xs:boolean": "Булево", "xs:dateTime": "Дата", "v8:UUID": "УникальныйИдентификатор", "cfg:AnyRef": "ЛюбаяСсылка", "cfg:AnyIBRef": "ЛюбаяСсылка", } REFERENCE_PRESENTATION_RU = { "CatalogRef": "СправочникСсылка", "DocumentRef": "ДокументСсылка", "EnumRef": "ПеречислениеСсылка", "ChartOfAccountsRef": "ПланСчетовСсылка", "ChartOfCalculationTypesRef": "ПланВидовРасчетаСсылка", "ChartOfCharacteristicTypesRef": "ПланВидовХарактеристикСсылка", "BusinessProcessRef": "БизнесПроцессСсылка", "TaskRef": "ЗадачаСсылка", } def local_name(tag: str) -> str: return tag.rsplit("}", 1)[-1] if "}" in tag else tag def direct_child(node: ET.Element, name: str) -> ET.Element | None: return next((child for child in list(node) if local_name(child.tag) == name), None) def text_child(node: ET.Element | None, name: str) -> str | None: if node is None: return None child = direct_child(node, name) if child is None or child.text is None: return None return child.text.strip() def synonym(properties: ET.Element | None) -> str | None: if properties is None: return None syn = direct_child(properties, "Synonym") if syn is None: return None for node in syn.iter(): if local_name(node.tag) == "content" and node.text: return node.text.strip() return None def value_types(properties: ET.Element | None) -> list[str]: if properties is None: return [] type_node = direct_child(properties, "Type") if type_node is None: return [] result = [] for node in type_node.iter(): if local_name(node.tag) == "Type" and node.text and ":" in node.text: result.append(node.text.strip()) elif local_name(node.tag) == "TypeSet" and node.text and ":" in node.text: result.append(node.text.strip()) return result def type_presentation_ru(value: str) -> str: if value in TYPE_PRESENTATION_RU: return TYPE_PRESENTATION_RU[value] if value.startswith("cfg:"): payload = value[4:] if "." in payload: family, name = payload.split(".", 1) prefix = REFERENCE_PRESENTATION_RU.get(family) if prefix: return f"{prefix}.{name}" return f"{family}.{name}" return value def route_kind(role: str) -> str: if role == "Fld": return "field" if role in {"VT", "LineNo"}: return "structural" return "table" def storage_routes(index: dict[str, Any], guid: str | None) -> list[dict[str, Any]]: if not guid: return [] item = (index.get("objects") or {}).get(guid.lower()) or {} result = [] for route in item.get("dbnames") or []: copy = { "guid": guid, "storage_role": route.get("storage_role"), "sql_number": route.get("sql_number"), "source": route.get("source_file") or "DBNames", "route_kind": route_kind(str(route.get("storage_role") or "")), "physical_name_candidate": physical_name(route), } result.append(copy) return result def value_type_payload(types: list[str]) -> dict[str, Any] | None: if not types: return None return { "types": types, "presentation": { "default_language": "ru", "ru": [type_presentation_ru(item) for item in types], }, "qualifiers": {}, "is_composite": len(types) > 1, } def field_columns(routes: list[dict[str, Any]], types: list[str]) -> list[dict[str, Any]]: fld = next((route for route in routes if route.get("storage_role") == "Fld" and route.get("sql_number") is not None), None) if not fld: return [{"status": "no_storage_route"}] base = f"_Fld{fld['sql_number']}" if len(types) > 1: return [ {"column": f"{base}_TYPE", "reason": "composite value discriminator", "value_types": types}, {"column": f"{base}_S", "reason": "composite string value", "value_types": types}, {"column": f"{base}_N", "reason": "composite numeric value", "value_types": types}, {"column": f"{base}_L", "reason": "composite boolean value", "value_types": types}, {"column": f"{base}_T", "reason": "composite datetime value", "value_types": types}, {"column": f"{base}_RTRef", "reason": "composite reference type id", "value_types": types}, {"column": f"{base}_RRRef", "reason": "composite reference value", "value_types": types}, ] value_type = types[0] if types else None if value_type in {"cfg:AnyRef", "cfg:AnyIBRef"}: return [ {"column": f"{base}_TYPE", "reason": "any reference discriminator", "value_types": types}, {"column": f"{base}_RTRef", "reason": "any reference type id", "value_types": types}, {"column": f"{base}_RRRef", "reason": "any reference value", "value_types": types}, ] if value_type and any(marker in value_type for marker in REFERENCE_MARKERS): return [{"column": f"{base}RRef", "reason": "single 1C reference type", "value_type": value_type}] return [{"column": base, "reason": "single primitive value", "value_type": value_type}] def extension_name_from_path(path: str | None) -> str | None: parts = str(path or "").replace("/", "\\").split("\\") lowered = [part.casefold() for part in parts] if "расширения" in lowered: index = lowered.index("расширения") if index + 1 < len(parts): return parts[index + 1] if "extensions" in lowered: index = lowered.index("extensions") if index + 1 < len(parts): return parts[index + 1] return None def metadata_item( node: ET.Element, *, index: dict[str, Any], category: str, parent_category: str, parent_name: str, parent_uuid: str, tabular_section_name: str | None = None, tabular_section_uuid: str | None = None, record_index: int, ) -> dict[str, Any]: properties = direct_child(node, "Properties") guid = (node.get("uuid") or "").lower() name = text_child(properties, "Name") types = value_types(properties) routes = storage_routes(index, guid) item = { "category": category, "name": name, "synonym": synonym(properties), "uuid": guid, "value_type": value_type_payload(types), "parent_category": parent_category, "parent_name": parent_name, "parent_uuid": parent_uuid, "record_index": record_index, "evidence": {"name": bool(name), "synonym": bool(synonym(properties)), "uuid": bool(guid)}, "storage_routes": routes, "storage_route_count": len(routes), "physical_columns": field_columns(routes, types) if category == "Attribute" else [{"status": "no_value_type"}], } object_belonging = text_child(properties, "ObjectBelonging") extended_object = text_child(properties, "ExtendedConfigurationObject") if object_belonging: item["object_belonging"] = object_belonging if extended_object: item["extended_configuration_object"] = extended_object.lower() if tabular_section_name: item["tabular_section_name"] = tabular_section_name item["tabular_section_uuid"] = tabular_section_uuid return item def object_node(root: ET.Element, kind: str) -> ET.Element: for node in root.iter(): if local_name(node.tag) == kind: return node raise SystemExit(f"XML object node not found: {kind}") def merge_attribute_overlay(base: list[dict[str, Any]], overlay: dict[str, Any]) -> None: extended_uuid = overlay.get("extended_configuration_object") target = None if extended_uuid: target = next((item for item in base if item.get("uuid") == extended_uuid), None) if target is None and overlay.get("name"): target = next((item for item in base if item.get("name") == overlay.get("name")), None) if target is None: base.append(overlay) return record = { "source": overlay.get("source"), "extension_name": overlay.get("extension_name"), "path": overlay.get("source_path"), "uuid": overlay.get("uuid"), "object_belonging": overlay.get("object_belonging"), "extended_configuration_object": overlay.get("extended_configuration_object"), "value_type": overlay.get("value_type"), "synonym": overlay.get("synonym"), } target.setdefault("extension_overrides", []).append(record) target["effective_source"] = "base+extension" if overlay.get("value_type"): target["base_value_type"] = target.get("base_value_type") or target.get("value_type") target["value_type"] = overlay["value_type"] target["physical_columns"] = field_columns(target.get("storage_routes") or [], overlay["value_type"].get("types") or []) if overlay.get("synonym"): target["synonym"] = overlay["synonym"] def apply_object_overlay( *, index: dict[str, Any], overlay: dict[str, Any], kind: str, attributes: list[dict[str, Any]], tabular_sections: list[dict[str, Any]], tabular_section_attributes: list[dict[str, Any]], ) -> dict[str, int]: path = Path(str(overlay.get("path") or "")) if not path.is_file(): return {"missing": 1, "attributes_added": 0, "attributes_changed": 0} root = ET.parse(path).getroot() node = object_node(root, kind) children = direct_child(node, "ChildObjects") if children is None: return {"missing": 0, "attributes_added": 0, "attributes_changed": 0} added = 0 changed = 0 extension_name = extension_name_from_path(str(path)) for child in list(children): if local_name(child.tag) != "Attribute": continue before = len(attributes) item = metadata_item( child, index=index, category="Attribute", parent_category=kind, parent_name=str(overlay.get("name") or ""), parent_uuid=str(overlay.get("guid") or ""), record_index=len(attributes), ) item["source"] = "extension" item["extension_name"] = extension_name item["source_path"] = str(path) merge_attribute_overlay(attributes, item) if len(attributes) > before: added += 1 else: changed += 1 return {"missing": 0, "attributes_added": added, "attributes_changed": changed} def build_metadata(index: dict[str, Any], *, kind: str, name: str) -> dict[str, Any]: resolution = resolve_object(index, kind=kind, name=name, limit=50) canonical = resolution.get("canonical") if not canonical: raise SystemExit(f"Object not found: {kind}.{name}") xml_path = Path(str(canonical.get("path") or "")) if not xml_path.is_file(): raise SystemExit(f"Object XML file not found: {xml_path}") root = ET.parse(xml_path).getroot() node = object_node(root, str(canonical["kind"])) properties = direct_child(node, "Properties") object_name = text_child(properties, "Name") or str(canonical["name"]) object_uuid = str(canonical["guid"]).lower() main_routes = storage_routes(index, object_uuid) main_table = next((physical_name(route) for route in main_routes if route.get("storage_role") == canonical["kind"]), None) if not main_table: main_table = next((route.get("physical_name_candidate") for route in main_routes if route.get("route_kind") == "table"), None) attributes = [] tabular_sections = [] tabular_section_attributes = [] children = direct_child(node, "ChildObjects") if children is not None: attr_index = 0 ts_index = 0 for child in list(children): child_kind = local_name(child.tag) if child_kind == "Attribute": attributes.append( { **metadata_item( child, index=index, category="Attribute", parent_category=str(canonical["kind"]), parent_name=object_name, parent_uuid=object_uuid, record_index=attr_index, ), "source": "base", } ) attr_index += 1 elif child_kind == "TabularSection": section = metadata_item( child, index=index, category="TabularSection", parent_category=str(canonical["kind"]), parent_name=object_name, parent_uuid=object_uuid, record_index=ts_index, ) vt_route = next((route for route in section["storage_routes"] if route.get("storage_role") == "VT"), None) line_numbers = [route.get("sql_number") for route in section["storage_routes"] if route.get("storage_role") == "LineNo"] if main_table and vt_route and vt_route.get("sql_number") is not None: section["physical_tables"] = [ { "table": f"{main_table}_VT{vt_route['sql_number']}", "reason": "tabular section VT route under object table", "vt_sql_number": vt_route["sql_number"], "line_no_sql_numbers": line_numbers, } ] tabular_sections.append(section) section_children = direct_child(child, "ChildObjects") if section_children is not None: for record_index, section_child in enumerate([item for item in list(section_children) if local_name(item.tag) == "Attribute"]): attr = metadata_item( section_child, index=index, category="Attribute", parent_category="TabularSection", parent_name=section["name"], parent_uuid=section["uuid"], tabular_section_name=section["name"], tabular_section_uuid=section["uuid"], record_index=record_index, ) attr["parent_physical_tables"] = section.get("physical_tables") or [] attr["source"] = "base" tabular_section_attributes.append(attr) ts_index += 1 overlay_stats = [] for overlay in resolution.get("extension_overlays") or []: stats = apply_object_overlay( index=index, overlay=overlay, kind=str(canonical["kind"]), attributes=attributes, tabular_sections=tabular_sections, tabular_section_attributes=tabular_section_attributes, ) overlay_stats.append( { "extension_name": extension_name_from_path(overlay.get("path")), "path": overlay.get("path"), **stats, } ) return { "schema": "onec_structured_metadata_from_resolved_xml.v1", "kind": canonical["kind"], "xml_file": str(xml_path), "identity": { "guid": object_uuid, "name": object_name, "synonyms": {"ru": synonym(properties)} if synonym(properties) else {}, }, "resolution": {"schema": resolution.get("schema"), "canonical": canonical, "summary": resolution.get("summary")}, "effective_metadata": { "base_path": str(xml_path), "extension_overlays_applied": overlay_stats, }, "attributes": attributes, "tabular_sections": tabular_sections, "tabular_section_attributes": tabular_section_attributes, "dimensions": [], "resources": [], "forms": [], "templates": [], "commands": [], "addressing_attributes": [], "accounting_flags": [], "columns": [], "enum_values": [], "object_storage_routes": main_routes, "storage_route_summary": { "metadata_items_with_routes": sum(1 for item in attributes + tabular_sections + tabular_section_attributes if item.get("storage_routes")), "object_routes": len(main_routes), }, "counts": { "attributes": len(attributes), "tabular_sections": len(tabular_sections), "tabular_section_attributes": len(tabular_section_attributes), }, } def main() -> int: parser = argparse.ArgumentParser(description="Build metadata card from resolved 1C object XML.") parser.add_argument("--index", type=Path, required=True) parser.add_argument("--kind", required=True) parser.add_argument("--name", required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() result = build_metadata(load_json(args.index), kind=args.kind, name=args.name) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"output": str(args.output), "counts": result["counts"]}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())