Files
llm/scripts/build_1c_extension_inventory.py

122 lines
4.3 KiB
Python

#!/usr/bin/env python3
"""Build an inventory that maps _ExtensionsInfo rows to DBNames-Ext files.
The script keeps the byte conversion explicit. SQL _ExtensionsInfo._IDRRef is
stored as 16 bytes. Observed DBNames-Ext suffixes match this rearrangement:
b[12:16] b[10:12] b[8:10] b[0:2] b[2:8]
This is recorded as an observed conversion and validated against exported
DBNames-Ext file names.
"""
from __future__ import annotations
import argparse
import json
import uuid
from pathlib import Path
from typing import Any
def load_json(path: Path) -> dict[str, Any]:
return json.loads(path.read_text(encoding="utf-8-sig"))
def dbnames_ext_guid_from_idrref(data: bytes) -> str:
if len(data) != 16:
raise ValueError(f"_IDRRef must contain 16 bytes, got {len(data)}")
reordered = data[12:16] + data[10:12] + data[8:10] + data[0:2] + data[2:8]
return str(uuid.UUID(bytes=reordered))
def binary_info(value: Any) -> dict[str, Any] | None:
if isinstance(value, dict) and value.get("type") == "binary":
return value
return None
def main() -> int:
parser = argparse.ArgumentParser(description="Map _ExtensionsInfo rows to DBNames-Ext files.")
parser.add_argument("--extensions-info", type=Path, required=True)
parser.add_argument("--params-dir", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
args = parser.parse_args()
report = load_json(args.extensions_info)
dbnames_ext_files = {
path.name.removeprefix("DBNames-Ext-").lower(): path
for path in args.params_dir.iterdir()
if path.is_file() and path.name.startswith("DBNames-Ext-")
}
rows = []
matched = 0
for row in report.get("rows") or []:
columns = row.get("columns") or {}
idrref = binary_info(columns.get("_IDRRef"))
if not idrref:
continue
idrref_bytes = Path(idrref["path"]).read_bytes()
dbnames_guid = dbnames_ext_guid_from_idrref(idrref_bytes)
dbnames_file = dbnames_ext_files.get(dbnames_guid)
if dbnames_file:
matched += 1
rows.append(
{
"row_index": row.get("row_index"),
"extension_name": columns.get("_ExtName"),
"extension_order": columns.get("_ExtensionOrder"),
"update_time": columns.get("_UpdateTime"),
"use_purpose": columns.get("_ExtensionUsePurpose"),
"scope": columns.get("_ExtensionScope"),
"idrref_hex": idrref_bytes.hex(),
"dbnames_ext_guid": dbnames_guid,
"dbnames_ext_file": str(dbnames_file) if dbnames_file else None,
"dbnames_ext_file_name": dbnames_file.name if dbnames_file else None,
"dbnames_ext_file_bytes": dbnames_file.stat().st_size if dbnames_file else None,
"extension_zipped_info": binary_info(columns.get("_ExtensionZippedInfo")),
}
)
known_from_rows = {row["dbnames_ext_guid"] for row in rows}
orphan_dbnames_files = [
{
"file_name": path.name,
"guid_or_marker": guid,
"bytes": path.stat().st_size,
}
for guid, path in sorted(dbnames_ext_files.items())
if guid not in known_from_rows
]
result = {
"schema": "onec_extension_inventory.v1",
"extensions_info": str(args.extensions_info),
"params_dir": str(args.params_dir),
"observed_idrref_to_dbnames_ext_guid": "b[12:16] + b[10:12] + b[8:10] + b[0:2] + b[2:8]",
"extension_row_count": len(rows),
"matched_dbnames_ext_count": matched,
"orphan_dbnames_ext_count": len(orphan_dbnames_files),
"extensions": rows,
"orphan_dbnames_ext_files": orphan_dbnames_files,
}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(
json.dumps(
{
"output": str(args.output),
"extensions": len(rows),
"matched_dbnames_ext": matched,
"orphan_dbnames_ext": len(orphan_dbnames_files),
},
ensure_ascii=False,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())