258 lines
9.4 KiB
Python
258 lines
9.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import xml.etree.ElementTree as ET
|
|
import sys
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from config.client import ODataClient, utc_now_iso
|
|
from config.settings import LOGS_DIR, load_settings
|
|
|
|
|
|
def _tag_ends(tag: str, suffix: str) -> bool:
|
|
return tag.endswith(suffix)
|
|
|
|
|
|
def _parse_metadata(metadata_path: Path) -> tuple[dict[str, list[str]], dict[str, str]]:
|
|
root = ET.fromstring(metadata_path.read_text(encoding="utf-8"))
|
|
|
|
entity_type_props: dict[str, list[str]] = {}
|
|
for entity_type in root.iter():
|
|
if not _tag_ends(entity_type.tag, "EntityType"):
|
|
continue
|
|
name = entity_type.attrib.get("Name", "")
|
|
if not name:
|
|
continue
|
|
props: list[str] = []
|
|
for child in entity_type:
|
|
if _tag_ends(child.tag, "Property"):
|
|
prop_name = child.attrib.get("Name", "")
|
|
if prop_name:
|
|
props.append(prop_name)
|
|
entity_type_props[name] = props
|
|
|
|
entity_set_to_type: dict[str, str] = {}
|
|
for entity_set in root.iter():
|
|
if not _tag_ends(entity_set.tag, "EntitySet"):
|
|
continue
|
|
set_name = entity_set.attrib.get("Name", "")
|
|
et = entity_set.attrib.get("EntityType", "")
|
|
if not set_name or not et:
|
|
continue
|
|
entity_type_name = et.split(".")[-1]
|
|
entity_set_to_type[set_name] = entity_type_name
|
|
|
|
return entity_type_props, entity_set_to_type
|
|
|
|
|
|
def _entity_sets_with_subconto(
|
|
entity_type_props: dict[str, list[str]],
|
|
entity_set_to_type: dict[str, str],
|
|
) -> list[dict[str, Any]]:
|
|
results: list[dict[str, Any]] = []
|
|
for entity_set, entity_type in entity_set_to_type.items():
|
|
props = entity_type_props.get(entity_type, [])
|
|
subconto_props = [p for p in props if "Субконто" in p or "Subconto" in p]
|
|
if not subconto_props:
|
|
continue
|
|
account_props = [p for p in props if p.startswith("Account") or "Счет" in p]
|
|
results.append(
|
|
{
|
|
"entity_set": entity_set,
|
|
"entity_type": entity_type,
|
|
"subconto_properties": subconto_props,
|
|
"account_properties": account_props,
|
|
"has_account_and_subconto": bool(account_props),
|
|
}
|
|
)
|
|
|
|
results.sort(key=lambda x: x["entity_set"].lower())
|
|
return results
|
|
|
|
|
|
def _safe_read_selected(
|
|
client: ODataClient,
|
|
entity_set: str,
|
|
select_fields: list[str] | None = None,
|
|
top: int = 20,
|
|
) -> dict[str, Any]:
|
|
params: dict[str, Any] = {}
|
|
if select_fields:
|
|
params["$select"] = ",".join(select_fields)
|
|
try:
|
|
response = client.read_entity_set(entity_set, top=top, extra_params=params or None)
|
|
payload = response.payload
|
|
rows = payload.get("value")
|
|
if rows is None and isinstance(payload.get("d"), dict):
|
|
rows = payload["d"].get("results")
|
|
if rows is None:
|
|
rows = []
|
|
if not isinstance(rows, list):
|
|
rows = [rows]
|
|
return {"status": "ok", "rows": rows}
|
|
except Exception as exc:
|
|
return {"status": "error", "error": str(exc), "rows": []}
|
|
|
|
|
|
def _non_null_subconto_counts(rows: list[dict[str, Any]]) -> dict[str, int]:
|
|
counts: dict[str, int] = {}
|
|
for row in rows:
|
|
for key, value in row.items():
|
|
if "Субконто" not in key and "Subconto" not in key:
|
|
continue
|
|
if value is None or value == "":
|
|
continue
|
|
counts[key] = counts.get(key, 0) + 1
|
|
return counts
|
|
|
|
|
|
def _load_join_probe_override(logs_dir: Path) -> dict[str, Any]:
|
|
path = logs_dir / "deep_subconto_join_probe.json"
|
|
if not path.exists():
|
|
return {"available": False}
|
|
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
except Exception as exc:
|
|
return {"available": False, "error": str(exc)}
|
|
|
|
chain_a = payload.get("chain_A_status")
|
|
chain_f = payload.get("chain_F_status")
|
|
valid = {"derivable", "opaque"}
|
|
return {
|
|
"available": True,
|
|
"path": str(path),
|
|
"chain_A_status": chain_a if chain_a in valid else None,
|
|
"chain_F_status": chain_f if chain_f in valid else None,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
settings = load_settings()
|
|
client = ODataClient(settings)
|
|
|
|
metadata_path = LOGS_DIR / "metadata.xml"
|
|
if not metadata_path.exists():
|
|
print("[error] metadata.xml not found. Run fetch_metadata first.")
|
|
return 1
|
|
|
|
entity_type_props, entity_set_to_type = _parse_metadata(metadata_path)
|
|
subconto_sets = _entity_sets_with_subconto(entity_type_props, entity_set_to_type)
|
|
|
|
# Targeted probes for A/F chains
|
|
target_sets = [
|
|
"AccountingRegister_Хозрасчетный_RecordType",
|
|
"AccountingRegister_Хозрасчетный",
|
|
"ChartOfAccounts_Хозрасчетный",
|
|
"ChartOfCharacteristicTypes_ВидыСубконтоХозрасчетные",
|
|
"Document_ОперацияБух",
|
|
"Document_ОперацияБух_ТаблицаРегистровБухгалтерии",
|
|
"Document_РеализацияТоваровУслуг",
|
|
"Document_ПоступлениеТоваровУслуг",
|
|
]
|
|
|
|
target_results: list[dict[str, Any]] = []
|
|
for entity_set in target_sets:
|
|
if entity_set not in entity_set_to_type:
|
|
target_results.append(
|
|
{
|
|
"entity_set": entity_set,
|
|
"status": "missing_in_metadata",
|
|
}
|
|
)
|
|
continue
|
|
|
|
et = entity_set_to_type[entity_set]
|
|
props = entity_type_props.get(et, [])
|
|
key_fields = [
|
|
p
|
|
for p in props
|
|
if p in {"Ref_Key", "Recorder", "Recorder_Type", "AccountDr_Key", "AccountCr_Key", "Организация_Key"}
|
|
or "Субконто" in p
|
|
]
|
|
probe = _safe_read_selected(client, entity_set, select_fields=key_fields or None, top=30)
|
|
rows = probe.get("rows", [])
|
|
target_results.append(
|
|
{
|
|
"entity_set": entity_set,
|
|
"entity_type": et,
|
|
"status": probe.get("status"),
|
|
"error": probe.get("error"),
|
|
"rows_fetched": len(rows),
|
|
"selected_fields": key_fields,
|
|
"non_null_subconto_counts": _non_null_subconto_counts(rows),
|
|
"sample_rows": rows[:3],
|
|
}
|
|
)
|
|
|
|
# Chain A/F compact verdict helpers
|
|
rec_type = next((x for x in target_results if x["entity_set"] == "AccountingRegister_Хозрасчетный_RecordType"), None)
|
|
chart_accounts = next((x for x in target_results if x["entity_set"] == "ChartOfAccounts_Хозрасчетный"), None)
|
|
chart_subconto = next((x for x in target_results if x["entity_set"] == "ChartOfCharacteristicTypes_ВидыСубконтоХозрасчетные"), None)
|
|
op_buh_tbl = next((x for x in target_results if x["entity_set"] == "Document_ОперацияБух_ТаблицаРегистровБухгалтерии"), None)
|
|
|
|
chain_a_status = "opaque"
|
|
if rec_type and rec_type.get("status") == "ok" and rec_type.get("rows_fetched", 0) > 0:
|
|
if rec_type.get("non_null_subconto_counts"):
|
|
chain_a_status = "derivable"
|
|
elif op_buh_tbl and op_buh_tbl.get("non_null_subconto_counts"):
|
|
chain_a_status = "derivable"
|
|
|
|
chain_f_status = "opaque"
|
|
if (
|
|
chart_accounts
|
|
and chart_accounts.get("status") == "ok"
|
|
and chart_accounts.get("rows_fetched", 0) > 0
|
|
and chart_subconto
|
|
and chart_subconto.get("status") == "ok"
|
|
and chart_subconto.get("rows_fetched", 0) > 0
|
|
and rec_type
|
|
and rec_type.get("status") == "ok"
|
|
and rec_type.get("rows_fetched", 0) > 0
|
|
):
|
|
if rec_type.get("non_null_subconto_counts"):
|
|
chain_f_status = "derivable"
|
|
|
|
join_probe_override = _load_join_probe_override(LOGS_DIR)
|
|
if join_probe_override.get("available"):
|
|
override_a = join_probe_override.get("chain_A_status")
|
|
override_f = join_probe_override.get("chain_F_status")
|
|
if override_a == "derivable":
|
|
chain_a_status = "derivable"
|
|
if override_f == "derivable":
|
|
chain_f_status = "derivable"
|
|
|
|
report = {
|
|
"generated_at": utc_now_iso(),
|
|
"endpoint": settings.service_root,
|
|
"entity_sets_with_subconto_total": len(subconto_sets),
|
|
"entity_sets_with_subconto": subconto_sets,
|
|
"targeted_results": target_results,
|
|
"chain_assessment": {
|
|
"A_document_to_posting_account_subconto": chain_a_status,
|
|
"F_chart_to_subconto_to_posting": chain_f_status,
|
|
},
|
|
"override_from_join_probe": join_probe_override,
|
|
}
|
|
|
|
out_path = LOGS_DIR / "deep_subconto_probe.json"
|
|
out_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"[ok] saved: {out_path}")
|
|
print(f"[ok] entity_sets_with_subconto_total={len(subconto_sets)}")
|
|
print(
|
|
"[ok] chain A="
|
|
+ report["chain_assessment"]["A_document_to_posting_account_subconto"]
|
|
+ ", chain F="
|
|
+ report["chain_assessment"]["F_chart_to_subconto_to_posting"]
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|