122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
"""Parser for Params/DBNames files."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any
|
|
|
|
from .payload import parse_brace_text, payload_to_text, scalar
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DBNamesRecord:
|
|
guid: str
|
|
storage_role: str
|
|
sql_number: int
|
|
index: int
|
|
source: str
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
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 parse_dbnames_bytes(data: bytes, *, source: str = "DBNames") -> dict[str, Any]:
|
|
decoded = payload_to_text(data)
|
|
text = decoded.get("text")
|
|
if text is None:
|
|
raise ValueError(f"{source}: cannot decode DBNames text")
|
|
parsed = _unwrap_bom_sequence(parse_brace_text(text))
|
|
if not (isinstance(parsed, dict) and parsed.get("type") == "list"):
|
|
raise ValueError(f"{source}: expected root list")
|
|
items = parsed.get("items") or []
|
|
if len(items) != 2:
|
|
raise ValueError(f"{source}: expected 2 root items, got {len(items)}")
|
|
records_node = items[1]
|
|
if not (isinstance(records_node, dict) and records_node.get("type") == "list"):
|
|
raise ValueError(f"{source}: expected records list")
|
|
record_items = records_node.get("items") or []
|
|
records: list[DBNamesRecord] = []
|
|
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:
|
|
continue
|
|
records.append(
|
|
DBNamesRecord(
|
|
guid=scalar(fields[0]).lower(),
|
|
storage_role=scalar(fields[1]),
|
|
sql_number=int(scalar(fields[2])),
|
|
index=index,
|
|
source=source,
|
|
)
|
|
)
|
|
return {
|
|
"source": source,
|
|
"compression": decoded["compression"],
|
|
"encoding": decoded["encoding"],
|
|
"root_number": int(scalar(items[0])),
|
|
"declared_count": int(scalar(record_items[0]) or "0") if record_items else 0,
|
|
"records": records,
|
|
}
|
|
|
|
|
|
def parse_dbnames_file(path: Path) -> dict[str, Any]:
|
|
return parse_dbnames_bytes(path.read_bytes(), source=path.name)
|
|
|
|
|
|
_GUID_RE = re.compile(
|
|
r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
|
|
)
|
|
|
|
|
|
def parse_dbnames_version_bytes(data: bytes, *, source: str = "DBNamesVersion") -> dict[str, Any]:
|
|
"""Parse the version marker stored separately from DBNames records."""
|
|
decoded = payload_to_text(data)
|
|
text = decoded.get("text")
|
|
if text is None:
|
|
raise ValueError(f"{source}: cannot decode DBNamesVersion text")
|
|
parsed = _unwrap_bom_sequence(parse_brace_text(text))
|
|
if not (isinstance(parsed, dict) and parsed.get("type") == "list"):
|
|
raise ValueError(f"{source}: expected root list")
|
|
items = parsed.get("items") or []
|
|
if len(items) != 2:
|
|
raise ValueError(f"{source}: expected 2 root items, got {len(items)}")
|
|
if not all(isinstance(item, dict) and item.get("type") == "atom" for item in items):
|
|
actual_types = [item.get("type") if isinstance(item, dict) else type(item).__name__ for item in items]
|
|
raise ValueError(f"{source}: expected scalar marker and version, got {actual_types}")
|
|
marker_text = scalar(items[0]).strip()
|
|
try:
|
|
marker = int(marker_text)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError(f"{source}: marker must be an integer, got {marker_text!r}") from exc
|
|
version = scalar(items[1]).strip().lower()
|
|
if not _GUID_RE.fullmatch(version):
|
|
raise ValueError(f"{source}: version must be a GUID string, got {version!r}")
|
|
return {
|
|
"source": source,
|
|
"compression": decoded["compression"],
|
|
"encoding": decoded["encoding"],
|
|
"marker": marker,
|
|
"version": version,
|
|
}
|
|
|
|
|
|
def parse_dbnames_version_file(path: Path) -> dict[str, Any]:
|
|
return parse_dbnames_version_bytes(path.read_bytes(), source=path.name)
|