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