Initial SQL-only 1C adapter baseline

scripts/extract_1c_extension_cas_manifest.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+"""Extract object-id -> ConfigCAS key pairs from an extension root CAS file."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import re
+from pathlib import Path
+from typing import Any
+
+from inspect_1c_sql_files import Lexer, Parser, try_decode, try_decompress
+
+
+def scalar(node: Any) -> str:
+    if isinstance(node, dict) and node.get("type") in {"atom", "string"}:
+        return str(node.get("value") or "")
+    return ""
+
+
+def parse_root(path: Path) -> Any:
+    payload, _ = try_decompress(path.read_bytes())
+    text, _ = try_decode(payload)
+    if text is None:
+        raise ValueError(f"Cannot decode {path}")
+    clean = text.replace("\x00", "").replace("\ufeff", "")
+    clean = clean.lstrip("ï»¿п»ї")
+    return Parser(Lexer(clean).tokens()).parse()
+
+
+def base64_to_sha1(value: str) -> str | None:
+    if not re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", value):
+        return None
+    try:
+        data = base64.b64decode(value, validate=True)
+    except Exception:
+        return None
+    if len(data) != 20:
+        return None
+    return data.hex()
+
+
+def extract_manifest(root_path: Path, cas_dir: Path | None) -> dict[str, Any]:
+    parsed = parse_root(root_path)
+    if not (isinstance(parsed, dict) and parsed.get("type") == "sequence"):
+        raise ValueError(f"{root_path.name}: expected sequence root")
+    items = parsed.get("items") or []
+    if items and isinstance(items[0], dict) and items[0].get("type") == "atom" and scalar(items[0]).strip("ï»¿п»ї") == "":
+        items = items[1:]
+    if len(items) == 4 and scalar(items[0]) in {"", "п»ї"}:
+        items = items[1:]
+    if len(items) < 3:
+        raise ValueError(f"{root_path.name}: expected at least 3 sequence items")
+    header = items[0]
+    payload_block = items[1]
+    manifest_block = items[2]
+    extension_guid = ""
+    if isinstance(payload_block, dict) and payload_block.get("type") == "list":
+        block_items = payload_block.get("items") or []
+        if len(block_items) > 1:
+            extension_guid = scalar(block_items[1]).lower()
+
+    manifest_items = manifest_block.get("items") if isinstance(manifest_block, dict) else []
+    declared_count = int(scalar(manifest_items[0]) or "0") if manifest_items else 0
+    entries = []
+    for index in range(1, len(manifest_items or []), 2):
+        object_id = scalar(manifest_items[index])
+        value = scalar(manifest_items[index + 1]) if index + 1 < len(manifest_items) else ""
+        cas_key = base64_to_sha1(value)
+        if not object_id or not cas_key:
+            continue
+        cas_path = cas_dir / cas_key if cas_dir else None
+        entries.append(
+            {
+                "object_id": object_id,
+                "cas_key": cas_key,
+                "cas_exists": bool(cas_path and cas_path.is_file()),
+                "cas_path": str(cas_path) if cas_path and cas_path.is_file() else None,
+            }
+        )
+    return {
+        "root_cas_file": root_path.name,
+        "root_cas_path": str(root_path),
+        "extension_configuration_guid": extension_guid,
+        "declared_count": declared_count,
+        "entry_count": len(entries),
+        "entries": entries,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Extract extension root CAS manifest.")
+    parser.add_argument("root_cas_file", type=Path)
+    parser.add_argument("--cas-dir", type=Path)
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--target-object-id", action="append", default=[])
+    args = parser.parse_args()
+
+    manifest = extract_manifest(args.root_cas_file, args.cas_dir)
+    targets = {value.lower() for value in args.target_object_id}
+    target_entries = [
+        entry
+        for entry in manifest["entries"]
+        if entry["object_id"].lower() in targets or entry["object_id"].lower().split(".", 1)[0] in targets
+    ]
+    report = {
+        "schema": "onec_extension_cas_manifest.v1",
+        **manifest,
+        "target_object_ids": sorted(targets),
+        "target_entries": target_entries,
+    }
+    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),
+                "entries": manifest["entry_count"],
+                "declared": manifest["declared_count"],
+                "target_entries": len(target_entries),
+            },
+            ensure_ascii=False,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())