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