117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Extract DBNames records from exported 1C SQL Params files.
|
|
|
|
DBNames is the platform-maintained map between metadata GUIDs, storage roles
|
|
such as Reference/Fld/Document/InfoRg, and numeric SQL suffixes. This extractor
|
|
keeps those platform labels as-is and does not translate them to 1C metadata
|
|
object kinds.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from inspect_1c_sql_files import Lexer, Parser, try_decode, try_decompress
|
|
|
|
|
|
def unwrap_bom_sequence(value: Any) -> Any:
|
|
if (
|
|
isinstance(value, dict)
|
|
and value.get("type") == "sequence"
|
|
and len(value.get("items") or []) == 2
|
|
and isinstance(value["items"][0], dict)
|
|
and value["items"][0].get("type") == "atom"
|
|
and str(value["items"][0].get("value") or "").strip("\ufeff") == ""
|
|
):
|
|
return value["items"][1]
|
|
return value
|
|
|
|
|
|
def scalar(value: Any) -> str:
|
|
if isinstance(value, dict) and value.get("type") in {"atom", "string"}:
|
|
return str(value.get("value") or "")
|
|
return ""
|
|
|
|
|
|
def parse_dbnames(path: Path) -> dict[str, Any]:
|
|
payload, compression = try_decompress(path.read_bytes())
|
|
text, encoding = try_decode(payload)
|
|
if text is None:
|
|
raise ValueError(f"{path.name}: cannot decode text")
|
|
text = text.replace("\x00", "")
|
|
first_brace = text.find("{")
|
|
if first_brace > 0:
|
|
text = text[first_brace:]
|
|
else:
|
|
text = text.lstrip("\ufeff")
|
|
parsed = unwrap_bom_sequence(Parser(Lexer(text).tokens()).parse())
|
|
if not (isinstance(parsed, dict) and parsed.get("type") == "list"):
|
|
raise ValueError(f"{path.name}: expected root list")
|
|
items = parsed.get("items") or []
|
|
if len(items) != 2:
|
|
raise ValueError(f"{path.name}: expected 2 root items, got {len(items)}")
|
|
root_number = scalar(items[0])
|
|
records_node = items[1]
|
|
if not (isinstance(records_node, dict) and records_node.get("type") == "list"):
|
|
raise ValueError(f"{path.name}: expected records list")
|
|
record_items = records_node.get("items") or []
|
|
declared_count = int(scalar(record_items[0]) or "0") if record_items else 0
|
|
records = []
|
|
for index, node in enumerate(record_items[1:], start=1):
|
|
if not (isinstance(node, dict) and node.get("type") == "list"):
|
|
continue
|
|
fields = node.get("items") or []
|
|
if len(fields) != 3:
|
|
records.append({"index": index, "status": "unexpected_shape", "field_count": len(fields)})
|
|
continue
|
|
records.append(
|
|
{
|
|
"index": index,
|
|
"guid": scalar(fields[0]).lower(),
|
|
"storage_role": scalar(fields[1]),
|
|
"sql_number": int(scalar(fields[2])),
|
|
"status": "parsed",
|
|
}
|
|
)
|
|
return {
|
|
"file_name": path.name,
|
|
"compression": compression,
|
|
"encoding": encoding,
|
|
"root_number": int(root_number),
|
|
"declared_count": declared_count,
|
|
"record_count": len(records),
|
|
"records": records,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Extract DBNames records from exported Params files.")
|
|
parser.add_argument("params_dir", type=Path)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
|
|
files = sorted(path for path in args.params_dir.iterdir() if path.is_file() and path.name.startswith("DBNames") and not path.name.startswith("DBNamesVersion-"))
|
|
result = {
|
|
"schema": "onec_sql_dbnames.v1",
|
|
"source": str(args.params_dir),
|
|
"file_count": len(files),
|
|
"dbnames": [parse_dbnames(path) for path in files],
|
|
}
|
|
summary: dict[str, int] = {}
|
|
for dbnames in result["dbnames"]:
|
|
for record in dbnames["records"]:
|
|
role = record.get("storage_role") or "<unknown>"
|
|
summary[role] = summary.get(role, 0) + 1
|
|
result["storage_role_counts"] = dict(sorted(summary.items(), key=lambda item: (-item[1], item[0])))
|
|
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), "files": len(files), "roles": len(summary)}, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|