422 lines
15 KiB
Python
422 lines
15 KiB
Python
"""Universal semantic profile helpers for 1C Config brace trees.
|
|
|
|
The rules here name repeatedly observed section paths, but every returned item
|
|
keeps structural evidence. Runtime data still comes from the decoded Config
|
|
tree and DBNames records of the requested base.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass
|
|
from typing import Any
|
|
|
|
from .child_records import ChildRecord, collect_evidence, declared_child_records
|
|
from .config_object import find_identity
|
|
from .config_sections import summarize_sections
|
|
from .payload import GUID_RE, root_signature, scalar
|
|
from .structured_metadata import get_by_path
|
|
|
|
|
|
SECTION_RULES: dict[str, list[dict[str, Any]]] = {
|
|
"AccountingRegister": [
|
|
{"path": "3", "category": "Dimension"},
|
|
{"path": "5", "category": "Resource"},
|
|
{"path": "7", "category": "Attribute"},
|
|
],
|
|
"AccumulationRegister": [
|
|
{"path": "5", "category": "Resource"},
|
|
{"path": "6", "category": "Attribute"},
|
|
{"path": "7", "category": "Dimension"},
|
|
],
|
|
"BusinessProcess": [{"path": "6", "category": "Attribute"}],
|
|
"Catalog": [
|
|
{"path": "5", "category": "TabularSection"},
|
|
{"path": "6", "category": "Attribute"},
|
|
],
|
|
"ChartOfAccounts": [
|
|
{"path": "5", "category": "TabularSection"},
|
|
{"path": "7", "category": "Attribute"},
|
|
{"path": "8", "category": "AccountingFlag"},
|
|
],
|
|
"ChartOfCalculationTypes": [
|
|
{"path": "3", "category": "TabularSection"},
|
|
{"path": "4", "category": "Attribute"},
|
|
],
|
|
"CalculationRegister": [
|
|
{"path": "3", "category": "Attribute"},
|
|
{"path": "4", "category": "Recalculation"},
|
|
{"path": "6", "category": "Resource"},
|
|
{"path": "9", "category": "Dimension"},
|
|
],
|
|
"Document": [
|
|
{"path": "3", "category": "TabularSection"},
|
|
{"path": "5", "category": "Attribute"},
|
|
],
|
|
"Enum": [{"path": "6", "category": "EnumValue"}],
|
|
"InformationRegister": [
|
|
{"path": "3", "category": "Resource"},
|
|
{"path": "4", "category": "Dimension"},
|
|
{"path": "5", "category": "Attribute"},
|
|
],
|
|
"Report": [{"path": "4", "category": "Attribute"}],
|
|
"Task": [
|
|
{"path": "5", "category": "Attribute"},
|
|
{"path": "6", "category": "AddressingAttribute"},
|
|
{"path": "8", "category": "Command"},
|
|
],
|
|
}
|
|
|
|
ROLE_ROUTE_KIND = {
|
|
"Fld": "field",
|
|
"TabularSection": "tabular_section",
|
|
"VT": "tabular_section",
|
|
"EnumValue": "enum_value",
|
|
"Dimension": "dimension",
|
|
"Resource": "resource",
|
|
"Document": "object",
|
|
"Reference": "object",
|
|
"Enum": "object",
|
|
"InfoRg": "object",
|
|
"AccumRg": "object",
|
|
"AccRg": "object",
|
|
"BPr": "object",
|
|
"Task": "object",
|
|
}
|
|
|
|
CATEGORY_ROUTE_KINDS = {
|
|
"Attribute": {"field"},
|
|
"AddressingAttribute": {"field"},
|
|
"AccountingFlag": {"field"},
|
|
"Column": {"field"},
|
|
"Dimension": {"dimension", "field"},
|
|
"Resource": {"resource", "field"},
|
|
"TabularSection": {"tabular_section"},
|
|
"EnumValue": {"enum_value"},
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DBNamesRoute:
|
|
guid: str
|
|
storage_role: str
|
|
sql_number: int
|
|
source: str
|
|
route_kind: str
|
|
physical_name_candidate: str | None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
def _children(node: Any) -> list[Any]:
|
|
if isinstance(node, dict) and node.get("type") in {"list", "sequence"}:
|
|
return node.get("items") or []
|
|
return []
|
|
|
|
|
|
def _ordered_scalars(node: Any, *, limit: int = 200) -> list[str]:
|
|
values: list[str] = []
|
|
|
|
def walk(value: Any) -> None:
|
|
if len(values) >= limit:
|
|
return
|
|
text = scalar(value)
|
|
if text:
|
|
values.append(text)
|
|
return
|
|
for child in _children(value):
|
|
walk(child)
|
|
|
|
walk(node)
|
|
return values
|
|
|
|
|
|
def _record_type(node: Any) -> dict[str, Any] | None:
|
|
values = _ordered_scalars(node)
|
|
try:
|
|
index = values.index("Pattern")
|
|
except ValueError:
|
|
return None
|
|
if index + 1 >= len(values):
|
|
return None
|
|
code = values[index + 1]
|
|
result: dict[str, Any] = {"code": code}
|
|
if code == "D":
|
|
result.update({"kind": "date", "presentation": "Дата"})
|
|
elif code == "B":
|
|
result.update({"kind": "boolean", "presentation": "Булево"})
|
|
elif code == "S":
|
|
result.update({"kind": "string", "presentation": "Строка"})
|
|
if index + 2 < len(values):
|
|
try:
|
|
length = int(values[index + 2])
|
|
result["length"] = length
|
|
if length > 0:
|
|
result["presentation"] = f"Строка({length})"
|
|
except ValueError:
|
|
pass
|
|
elif code == "N":
|
|
result.update({"kind": "number", "presentation": "Число"})
|
|
if index + 2 < len(values):
|
|
try:
|
|
precision = int(values[index + 2])
|
|
result["precision"] = precision
|
|
except ValueError:
|
|
pass
|
|
if index + 3 < len(values):
|
|
try:
|
|
scale = int(values[index + 3])
|
|
result["scale"] = scale
|
|
except ValueError:
|
|
pass
|
|
if "precision" in result:
|
|
scale = result.get("scale")
|
|
result["presentation"] = f"Число({result['precision']}, {scale})" if scale is not None else f"Число({result['precision']})"
|
|
elif code == "#":
|
|
result.update({"kind": "reference", "presentation": "Ссылка"})
|
|
if index + 2 < len(values) and GUID_RE.fullmatch(values[index + 2]):
|
|
result["type_guid"] = values[index + 2].lower()
|
|
else:
|
|
result.update({"kind": "unknown"})
|
|
return result
|
|
|
|
|
|
def _record_title(record: ChildRecord, *, include_samples: bool = True) -> dict[str, Any]:
|
|
identity = find_identity(record.node)
|
|
strings = sorted(record.evidence.get("strings") or [])
|
|
guids = sorted(record.evidence.get("guids") or [])
|
|
likely_name = identity.name if identity else next((value for value in strings if value and not GUID_RE.fullmatch(value)), None)
|
|
result = {
|
|
"index": record.index,
|
|
"path": record.path,
|
|
"identity": identity.to_dict() if identity else None,
|
|
"likely_name": likely_name,
|
|
"type": _record_type(record.node),
|
|
}
|
|
if include_samples:
|
|
result.update(
|
|
{
|
|
"strings_sample": strings[:12],
|
|
"guids_sample": guids[:12],
|
|
"string_count": len(strings),
|
|
"guid_count": len(guids),
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def dbnames_routes(records: list[Any] | None) -> dict[str, list[DBNamesRoute]]:
|
|
result: dict[str, list[DBNamesRoute]] = {}
|
|
for record in records or []:
|
|
guid = str(getattr(record, "guid", "") or "").lower()
|
|
if not guid:
|
|
continue
|
|
role = str(getattr(record, "storage_role", "") or "")
|
|
sql_number = int(getattr(record, "sql_number", 0) or 0)
|
|
route_kind = ROLE_ROUTE_KIND.get(role, "storage")
|
|
physical = f"_{role}{sql_number}" if role and sql_number and route_kind != "object" else None
|
|
result.setdefault(guid, []).append(
|
|
DBNamesRoute(
|
|
guid=guid,
|
|
storage_role=role,
|
|
sql_number=sql_number,
|
|
source=str(getattr(record, "source", "") or ""),
|
|
route_kind=route_kind,
|
|
physical_name_candidate=physical,
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def _routes_for_evidence(
|
|
evidence: dict[str, set[str]],
|
|
routes_by_guid: dict[str, list[DBNamesRoute]],
|
|
*,
|
|
category: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
routes: list[dict[str, Any]] = []
|
|
seen: set[tuple[str, str, int]] = set()
|
|
allowed_route_kinds = CATEGORY_ROUTE_KINDS.get(str(category or ""))
|
|
for guid in sorted(evidence.get("guids") or []):
|
|
for route in routes_by_guid.get(guid.lower(), []):
|
|
if allowed_route_kinds and route.route_kind not in allowed_route_kinds:
|
|
continue
|
|
key = (route.guid, route.storage_role, route.sql_number)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
routes.append(route.to_dict())
|
|
return routes
|
|
|
|
|
|
def _routes_for_record(
|
|
record: ChildRecord,
|
|
routes_by_guid: dict[str, list[DBNamesRoute]],
|
|
*,
|
|
category: str | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
identity = find_identity(record.node)
|
|
if identity:
|
|
direct = _routes_for_evidence({"guids": {identity.guid}, "strings": set()}, routes_by_guid, category=category)
|
|
if direct:
|
|
return direct
|
|
return _routes_for_evidence(record.evidence, routes_by_guid, category=category)
|
|
|
|
|
|
def _section_profile(
|
|
tree: Any,
|
|
rule: dict[str, Any],
|
|
routes_by_guid: dict[str, list[DBNamesRoute]],
|
|
*,
|
|
lightweight: bool = False,
|
|
) -> dict[str, Any]:
|
|
path = str(rule["path"])
|
|
category = str(rule["category"])
|
|
node = get_by_path(tree, path)
|
|
if node is None:
|
|
return {
|
|
"path": path,
|
|
"category": category,
|
|
"status": "missing",
|
|
"declared_record_count": 0,
|
|
"records": [],
|
|
}
|
|
records = declared_child_records(node, path, include_evidence=not lightweight)
|
|
section_evidence = collect_evidence(node) if not lightweight else {"strings": set(), "guids": set()}
|
|
def record_profile(record: ChildRecord) -> dict[str, Any]:
|
|
item = {
|
|
**_record_title(record, include_samples=not lightweight),
|
|
"storage_routes": [] if lightweight else _routes_for_record(record, routes_by_guid, category=category),
|
|
}
|
|
if category == "TabularSection":
|
|
item["columns"] = _tabular_section_columns(record, routes_by_guid)
|
|
return item
|
|
|
|
return {
|
|
"path": path,
|
|
"category": category,
|
|
"status": "ok",
|
|
"list_len": len(_children(node)),
|
|
"declared_record_count": len(records),
|
|
"storage_routes": _routes_for_evidence(section_evidence, routes_by_guid, category=category),
|
|
"records": [record_profile(record) for record in records],
|
|
}
|
|
|
|
|
|
def _nested_record_containers(node: Any, path: str, *, max_depth: int = 5, include_root: bool = False) -> list[list[ChildRecord]]:
|
|
containers: list[list[ChildRecord]] = []
|
|
|
|
def walk(value: Any, current_path: str, depth: int) -> None:
|
|
records = declared_child_records(value, current_path)
|
|
if records and (include_root or depth > 0):
|
|
containers.append(records)
|
|
if depth >= max_depth:
|
|
return
|
|
for index, child in enumerate(_children(value)):
|
|
walk(child, f"{current_path}.{index}", depth + 1)
|
|
|
|
walk(node, path, 0)
|
|
return containers
|
|
|
|
|
|
def _container_score(records: list[ChildRecord]) -> tuple[int, int, int]:
|
|
titles = [_record_title(record) for record in records]
|
|
identities = sum(1 for title in titles if title.get("identity"))
|
|
names = sum(1 for title in titles if title.get("likely_name"))
|
|
return identities, names, len(records)
|
|
|
|
|
|
def _tabular_section_columns(record: ChildRecord, routes_by_guid: dict[str, list[DBNamesRoute]]) -> list[dict[str, Any]]:
|
|
containers = _nested_record_containers(record.node, record.path, include_root=False)
|
|
candidates = [records for records in containers if len(records) >= 1]
|
|
if not candidates:
|
|
return []
|
|
best = max(candidates, key=_container_score)
|
|
columns = []
|
|
for column_record in best:
|
|
title = _record_title(column_record)
|
|
if not title.get("likely_name") and not title.get("identity"):
|
|
continue
|
|
columns.append(
|
|
{
|
|
**title,
|
|
"storage_routes": _routes_for_record(column_record, routes_by_guid, category="Column"),
|
|
}
|
|
)
|
|
return columns
|
|
|
|
|
|
def _generic_record_containers(tree: Any, *, max_depth: int = 3, limit: int = 200) -> list[dict[str, Any]]:
|
|
containers: list[dict[str, Any]] = []
|
|
|
|
def walk(node: Any, path: list[int], depth: int) -> None:
|
|
if len(containers) >= limit:
|
|
return
|
|
current_path = ".".join(str(part) for part in path)
|
|
records = declared_child_records(node, current_path)
|
|
if records:
|
|
containers.append(
|
|
{
|
|
"path": current_path,
|
|
"declared_record_count": len(records),
|
|
"record_paths_sample": [record.path for record in records[:20]],
|
|
}
|
|
)
|
|
if depth >= max_depth:
|
|
return
|
|
for index, child in enumerate(_children(node)):
|
|
walk(child, [*path, index], depth + 1)
|
|
|
|
walk(tree, [], 0)
|
|
return containers
|
|
|
|
|
|
def decode_config_semantic(
|
|
tree: Any,
|
|
*,
|
|
kind: str | None = None,
|
|
dbnames_records: list[Any] | None = None,
|
|
max_depth: int = 3,
|
|
section_sample_limit: int = 200,
|
|
include_generic: bool = True,
|
|
categories: set[str] | list[str] | tuple[str, ...] | None = None,
|
|
lightweight: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Return a structured, evidence-first profile for a Config object tree."""
|
|
|
|
identity = find_identity(tree)
|
|
routes_by_guid = dbnames_routes(dbnames_records)
|
|
object_routes = routes_by_guid.get((identity.guid if identity else "").lower(), [])
|
|
wanted_categories = {str(category) for category in (categories or [])}
|
|
rules = [
|
|
rule
|
|
for rule in SECTION_RULES.get(str(kind or ""), [])
|
|
if not wanted_categories or str(rule.get("category") or "") in wanted_categories
|
|
]
|
|
sections = [_section_profile(tree, rule, routes_by_guid, lightweight=lightweight) for rule in rules]
|
|
generic_sections = [item.to_dict() for item in summarize_sections(tree, max_depth=max_depth, limit=section_sample_limit)] if include_generic else []
|
|
generic_record_containers = _generic_record_containers(tree, max_depth=max_depth) if include_generic else []
|
|
return {
|
|
"schema": "onec_config_semantic_profile.v1",
|
|
"kind": kind,
|
|
"root": root_signature(tree),
|
|
"identity": identity.to_dict() if identity else None,
|
|
"object_storage_routes": [route.to_dict() for route in object_routes],
|
|
"section_rules": [
|
|
{
|
|
**rule,
|
|
"source": "built_in_observed_rules",
|
|
"note": "Rule names an observed section path; returned records are decoded from the current live Config payload.",
|
|
}
|
|
for rule in rules
|
|
],
|
|
"sections": sections,
|
|
"generic_sections": generic_sections,
|
|
"generic_record_containers": generic_record_containers,
|
|
"counts": {
|
|
"sections": len(sections),
|
|
"ok_sections": sum(1 for section in sections if section.get("status") == "ok"),
|
|
"generic_record_containers": len(generic_record_containers),
|
|
},
|
|
}
|