#!/usr/bin/env python3 """Summarize extension manifest parts against available XML extension dumps. This report is broad, not deep. It maps XML GUIDs to manifest object ids and records observable payload structure for each manifest part. It avoids assigning semantic labels from suffixes. """ 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 from inspect_1c_sql_files import GUID_RE def extension_name_from_manifest(path: Path) -> str: stem = path.stem return re.sub(r"^\d+_", "", stem).rsplit("-", 1)[0] def load_manifests(manifest_dir: Path) -> dict[str, dict[str, Any]]: result = {} for path in sorted(manifest_dir.glob("*.json")): data = json.loads(path.read_text(encoding="utf-8")) name = extension_name_from_manifest(path) result[name] = {"path": path, "data": data} return result def scan_xml_guids(paths: list[Path]) -> dict[str, list[str]]: guid_paths: dict[str, list[str]] = defaultdict(list) for root in paths: if not root.exists(): continue candidates = [root] if root.is_file() else [item for item in root.rglob("*") if item.is_file()] for path in candidates: try: text = path.read_text(encoding="utf-8-sig") except Exception: continue for guid in set(match.lower() for match in GUID_RE.findall(text)): guid_paths[guid].append(str(path)) return dict(guid_paths) def part_signature(payload: dict[str, Any]) -> str: root = payload.get("root_marker") root_len = payload.get("root_len") markers = ",".join(payload.get("payload_markers") or []) status = payload.get("parse_status") return f"{status}|root={root}|len={root_len}|markers={markers}" def summarize_part(entry: dict[str, Any], cas_dir: Path) -> dict[str, Any]: cas_key = entry["cas_key"] cas_path = Path(entry.get("cas_path") or cas_dir / cas_key) if not cas_path.is_file(): cas_path = cas_dir / cas_key payload = parse_cas_payload(cas_path) if cas_path.is_file() else {"parse_status": "missing_cas"} return { "object_id": entry["object_id"], "suffix": suffix_of(entry["object_id"]), "cas_key": cas_key, "parse_status": payload.get("parse_status"), "encoding": payload.get("encoding"), "text_offset": payload.get("text_offset"), "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 []), "signature": part_signature(payload), } def main() -> int: parser = argparse.ArgumentParser(description="Summarize 1C extension manifest/XML part structures.") parser.add_argument("--manifest-dir", type=Path, required=True) parser.add_argument("--cas-dir", type=Path, required=True) parser.add_argument("--extension-xml-inventory", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--sample-per-extension", type=int, default=12) args = parser.parse_args() manifests = load_manifests(args.manifest_dir) inventory = json.loads(args.extension_xml_inventory.read_text(encoding="utf-8")) extensions = [] global_suffix_counts: Counter[str] = Counter() global_signature_counts: Counter[str] = Counter() for ext in inventory.get("extensions") or []: name = ext["extension_name"] manifest_info = manifests.get(name) xml_dirs = [Path(item["path"]) for item in ext.get("xml_dirs") or []] xml_guid_paths = scan_xml_guids(xml_dirs) entries = (manifest_info or {}).get("data", {}).get("entries") or [] by_base: dict[str, list[dict[str, Any]]] = defaultdict(list) for entry in entries: by_base[str(entry["object_id"]).split(".", 1)[0].lower()].append(entry) matched_guids = sorted(set(xml_guid_paths) & set(by_base)) suffix_counts: Counter[str] = Counter() signature_counts: Counter[str] = Counter() marker_counts: Counter[str] = Counter() sample_objects = [] for guid in matched_guids[: args.sample_per_extension]: parts = [] for entry in sorted(by_base[guid], key=lambda item: (suffix_of(item["object_id"]), item["object_id"])): part = summarize_part(entry, args.cas_dir) parts.append(part) suffix_counts[part["suffix"]] += 1 signature_counts[part["signature"]] += 1 for marker in part["payload_markers"]: marker_counts[marker] += 1 sample_objects.append({"guid": guid, "xml_paths": xml_guid_paths[guid][:5], "parts": parts}) # Also count all manifest suffixes, even when XML is absent, without parsing all_suffix_counts = Counter(suffix_of(entry["object_id"]) for entry in entries) global_suffix_counts.update(all_suffix_counts) global_signature_counts.update(signature_counts) extensions.append( { "extension_name": name, "manifest_path": str(manifest_info["path"]) if manifest_info else None, "manifest_entry_count": len(entries), "xml_dir_count": len(xml_dirs), "xml_guid_count": len(xml_guid_paths), "matched_xml_manifest_guid_count": len(matched_guids), "all_manifest_suffix_counts": dict(sorted(all_suffix_counts.items())), "sampled_part_suffix_counts": dict(sorted(suffix_counts.items())), "sampled_part_signature_counts": dict(signature_counts.most_common()), "sampled_payload_marker_counts": dict(marker_counts.most_common()), "sample_objects": sample_objects, } ) report = { "schema": "onec_extension_manifest_xml_part_summary.v1", "manifest_dir": str(args.manifest_dir), "cas_dir": str(args.cas_dir), "extension_xml_inventory": str(args.extension_xml_inventory), "extension_count": len(extensions), "global_manifest_suffix_counts": dict(sorted(global_suffix_counts.items())), "global_sampled_signature_counts": dict(global_signature_counts.most_common()), "extensions": extensions, } 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), "extensions": len(extensions)}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())