#!/usr/bin/env python3 """Build a deterministic route index for 1C metadata objects. The index answers where an object or object part can be retrieved from: DBNames storage roles, base Config direct files, XML names, and extension manifest/CAS entries. It stores routes and observed payload signatures, not semantic guesses. """ from __future__ import annotations import argparse import json import re from collections import Counter, defaultdict from pathlib import Path from typing import Any from analyze_1c_manifest_object_parts import parse_cas_payload, suffix_of def load_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8-sig")) def extension_name_from_manifest(path: Path) -> str: stem = path.stem return re.sub(r"^\d+_", "", stem).rsplit("-", 1)[0] def ensure_object(objects: dict[str, dict[str, Any]], guid: str) -> dict[str, Any]: guid = guid.lower() item = objects.get(guid) if item is None: item = { "guid": guid, "dbnames": [], "xml_top_objects": [], "xml_occurrence_count": 0, "config_routes": [], "extension_routes": [], } objects[guid] = item return item def add_dbnames(objects: dict[str, dict[str, Any]], dbnames: dict[str, Any]) -> Counter[str]: role_counts: Counter[str] = Counter() for source in dbnames.get("dbnames") or []: file_name = source.get("file_name") for record in source.get("records") or []: if record.get("status") != "parsed": continue guid = str(record.get("guid") or "").lower() role = record.get("storage_role") role_counts[str(role)] += 1 ensure_object(objects, guid)["dbnames"].append( { "source_file": file_name, "storage_role": role, "sql_number": record.get("sql_number"), "index": record.get("index"), } ) return role_counts def add_xml(objects: dict[str, dict[str, Any]], xml_index: dict[str, Any]) -> None: for guid, item in (xml_index.get("guid_map") or {}).items(): obj = ensure_object(objects, guid) obj["xml_top_objects"] = item.get("top_objects") or [] obj["xml_occurrence_count"] = item.get("total_occurrences", 0) def payload_signature(payload: dict[str, Any]) -> dict[str, Any]: return { "parse_status": payload.get("parse_status"), "encoding": payload.get("encoding"), "text_offset": payload.get("text_offset"), "payload_bytes": payload.get("payload_bytes"), "compression": payload.get("compression"), "root_marker": payload.get("root_marker"), "root_len": payload.get("root_len"), "payload_markers": payload.get("payload_markers") or [], "base64_block_count": len(payload.get("_base64_blocks_decoded") or []), "stream_block_count": len(payload.get("_stream_blocks") or []), } def add_config_routes(objects: dict[str, dict[str, Any]], config_dirs: list[Path]) -> None: for directory in config_dirs: if not directory.exists(): continue for path in sorted(item for item in directory.iterdir() if item.is_file()): guid = path.name.lower() if not re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", guid): continue payload = parse_cas_payload(path) ensure_object(objects, guid)["config_routes"].append( { "route_type": "base_config_direct", "table": directory.name, "file_name": guid, "path": str(path), "payload_signature": payload_signature(payload), } ) def load_extension_inventory(path: Path | None) -> dict[str, dict[str, Any]]: if not path or not path.is_file(): return {} inventory = load_json(path) result = {} for ext in inventory.get("extensions") or []: name = ext.get("extension_name") if name: result[name] = ext return result def add_extension_routes( objects: dict[str, dict[str, Any]], manifest_dir: Path, cas_dir: Path, extension_inventory: dict[str, dict[str, Any]], ) -> Counter[str]: signature_counts: Counter[str] = Counter() payload_cache: dict[str, dict[str, Any]] = {} for manifest_path in sorted(manifest_dir.glob("*.json")): manifest = load_json(manifest_path) extension_name = extension_name_from_manifest(manifest_path) ext = extension_inventory.get(extension_name, {}) for entry in manifest.get("entries") or []: object_id = str(entry.get("object_id") or "").lower() if not object_id: continue guid = object_id.split(".", 1)[0] cas_key = entry.get("cas_key") cas_path = Path(entry.get("cas_path") or cas_dir / str(cas_key)) if not cas_path.is_file(): cas_path = cas_dir / str(cas_key) if cas_path.is_file(): if str(cas_path) not in payload_cache: payload_cache[str(cas_path)] = payload_signature(parse_cas_payload(cas_path)) signature = payload_cache[str(cas_path)] else: signature = {"parse_status": "missing_cas"} signature_key = ( f"{signature.get('parse_status')}|root={signature.get('root_marker')}|" f"len={signature.get('root_len')}|markers={','.join(signature.get('payload_markers') or [])}" ) signature_counts[signature_key] += 1 ensure_object(objects, guid)["extension_routes"].append( { "route_type": "extension_manifest_cas", "extension_name": extension_name, "extension_order": ext.get("extension_order"), "dbnames_ext_guid": ext.get("dbnames_ext_guid"), "manifest_path": str(manifest_path), "root_cas_file": manifest.get("root_cas_file"), "extension_configuration_guid": manifest.get("extension_configuration_guid"), "object_id": object_id, "suffix": suffix_of(object_id), "cas_key": cas_key, "cas_path": str(cas_path) if cas_path.is_file() else None, "payload_signature": signature, } ) return signature_counts def compact_object(item: dict[str, Any]) -> dict[str, Any]: route_kind = [] if item["config_routes"]: route_kind.append("base_config_direct") if item["extension_routes"]: route_kind.append("extension_manifest_cas") if item["dbnames"]: route_kind.append("dbnames_storage") if item["xml_top_objects"]: route_kind.append("xml_top_object") item["route_kind"] = route_kind return item def main() -> int: parser = argparse.ArgumentParser(description="Build a unified 1C object route index.") parser.add_argument("--dbnames", type=Path, required=True) parser.add_argument("--xml-index", type=Path, required=True) parser.add_argument("--manifest-dir", type=Path, required=True) parser.add_argument("--cas-dir", type=Path, required=True) parser.add_argument("--extension-inventory", type=Path) parser.add_argument("--config-dir", type=Path, action="append", default=[]) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() objects: dict[str, dict[str, Any]] = {} role_counts = add_dbnames(objects, load_json(args.dbnames)) add_xml(objects, load_json(args.xml_index)) add_config_routes(objects, args.config_dir) signature_counts = add_extension_routes( objects, args.manifest_dir, args.cas_dir, load_extension_inventory(args.extension_inventory), ) compacted = {guid: compact_object(item) for guid, item in sorted(objects.items())} route_counts = Counter() for item in compacted.values(): for kind in item["route_kind"]: route_counts[kind] += 1 report = { "schema": "onec_unified_object_route_index.v1", "dbnames": str(args.dbnames), "xml_index": str(args.xml_index), "manifest_dir": str(args.manifest_dir), "cas_dir": str(args.cas_dir), "config_dirs": [str(path) for path in args.config_dir], "object_count": len(compacted), "route_kind_counts": dict(route_counts.most_common()), "dbnames_role_counts": dict(role_counts.most_common()), "extension_payload_signature_counts": dict(signature_counts.most_common()), "objects": compacted, } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"output": str(args.output), "objects": len(compacted)}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())