252 lines
9.7 KiB
Python
252 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Attach DBNames storage routes to structured metadata projection reports."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
PLUGIN_ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(PLUGIN_ROOT))
|
|
|
|
from parser.dbnames import DBNamesRecord # noqa: E402
|
|
from parser.storage import storage_routes # noqa: E402
|
|
|
|
|
|
METADATA_FIELDS = (
|
|
"attributes",
|
|
"tabular_sections",
|
|
"dimensions",
|
|
"resources",
|
|
"forms",
|
|
"templates",
|
|
"commands",
|
|
"addressing_attributes",
|
|
"accounting_flags",
|
|
"columns",
|
|
"enum_values",
|
|
"integration_service_channels",
|
|
"operations",
|
|
"url_templates",
|
|
"tabular_section_attributes",
|
|
)
|
|
|
|
|
|
def load_json(path: Path) -> dict[str, Any]:
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
def write_json(path: Path, data: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def dbnames_records(report: dict[str, Any]) -> list[DBNamesRecord]:
|
|
records = []
|
|
for db_file in report.get("dbnames") or []:
|
|
source = db_file.get("file_name") or db_file.get("source") or "DBNames"
|
|
for row in db_file.get("records") or []:
|
|
guid = (row.get("guid") or "").lower()
|
|
role = row.get("storage_role") or ""
|
|
number = row.get("sql_number")
|
|
if not guid or not role or number is None:
|
|
continue
|
|
records.append(
|
|
DBNamesRecord(
|
|
guid=guid,
|
|
storage_role=role,
|
|
sql_number=int(number),
|
|
index=int(row.get("index") or 0),
|
|
source=source,
|
|
)
|
|
)
|
|
return records
|
|
|
|
|
|
def route_index(records: list[DBNamesRecord]) -> dict[str, list[dict[str, Any]]]:
|
|
grouped: dict[str, list[DBNamesRecord]] = {}
|
|
for record in records:
|
|
grouped.setdefault(record.guid, []).append(record)
|
|
return {
|
|
guid: [route.to_dict() for route in storage_routes(rows)]
|
|
for guid, rows in grouped.items()
|
|
}
|
|
|
|
|
|
def enrich_report(report: dict[str, Any], routes_by_guid: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
|
|
result = dict(report)
|
|
object_guid = ((report.get("identity") or {}).get("guid") or "").lower()
|
|
result["object_storage_routes"] = routes_by_guid.get(object_guid, [])
|
|
object_tables = [
|
|
route["physical_name_candidate"]
|
|
for route in result["object_storage_routes"]
|
|
if route.get("route_kind") == "table"
|
|
and route.get("physical_name_candidate")
|
|
and is_primary_object_table(route)
|
|
]
|
|
matched = 0
|
|
total = 0
|
|
for field in METADATA_FIELDS:
|
|
enriched_items = []
|
|
for item in report.get(field) or []:
|
|
total += 1
|
|
copy = dict(item)
|
|
guid = (copy.get("uuid") or "").lower()
|
|
routes = routes_by_guid.get(guid, [])
|
|
copy["storage_routes"] = routes
|
|
copy["storage_route_count"] = len(routes)
|
|
copy["physical_columns"] = predicted_columns(copy, routes)
|
|
if field == "tabular_sections":
|
|
copy["physical_tables"] = predicted_tabular_section_tables(copy, routes, object_tables)
|
|
if field == "tabular_section_attributes":
|
|
copy["parent_physical_tables"] = parent_tabular_section_tables(result, copy)
|
|
if routes:
|
|
matched += 1
|
|
enriched_items.append(copy)
|
|
result[field] = enriched_items
|
|
result["storage_route_summary"] = {
|
|
"metadata_item_count": total,
|
|
"metadata_items_with_routes": matched,
|
|
"object_route_count": len(result["object_storage_routes"]),
|
|
}
|
|
return result
|
|
|
|
|
|
def parent_tabular_section_tables(report: dict[str, Any], item: dict[str, Any]) -> list[dict[str, Any]]:
|
|
parent_uuid = (item.get("tabular_section_uuid") or item.get("parent_uuid") or "").lower()
|
|
if not parent_uuid:
|
|
return []
|
|
for tabular_section in report.get("tabular_sections") or []:
|
|
if (tabular_section.get("uuid") or "").lower() == parent_uuid:
|
|
return tabular_section.get("physical_tables") or []
|
|
return []
|
|
|
|
|
|
def predicted_tabular_section_tables(
|
|
item: dict[str, Any],
|
|
routes: list[dict[str, Any]],
|
|
object_tables: list[str],
|
|
) -> list[dict[str, Any]]:
|
|
result = []
|
|
vt_routes = [route for route in routes if route.get("storage_role") == "VT" and route.get("sql_number") is not None]
|
|
line_routes = [route for route in routes if route.get("storage_role") == "LineNo"]
|
|
for vt in vt_routes:
|
|
for table in object_tables:
|
|
result.append(
|
|
{
|
|
"table": f"{table}_VT{vt['sql_number']}",
|
|
"reason": "tabular section VT route under object table",
|
|
"vt_sql_number": vt["sql_number"],
|
|
"line_no_sql_numbers": [route["sql_number"] for route in line_routes],
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def is_primary_object_table(route: dict[str, Any]) -> bool:
|
|
role = route.get("storage_role") or ""
|
|
if role.endswith("ChngR") or role.endswith("SInf"):
|
|
return False
|
|
if role in {"BPrPoints", "AccumRgT", "AccumRgOpt", "AccRgAT0", "AccRgCT", "AccRgOpt"}:
|
|
return False
|
|
return True
|
|
|
|
|
|
def predicted_columns(item: dict[str, Any], routes: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Predict physical SQL column names for proven simple type cases."""
|
|
|
|
value_type = item.get("value_type") or {}
|
|
types = value_type.get("types") or []
|
|
if len(types) > 1:
|
|
return predicted_composite_columns(types, routes)
|
|
if len(types) == 0:
|
|
return [{"status": "no_value_type"}] if routes else []
|
|
|
|
value = types[0]
|
|
result = []
|
|
for route in routes:
|
|
if route.get("storage_role") != "Fld" or not route.get("physical_name_candidate"):
|
|
continue
|
|
base = route["physical_name_candidate"]
|
|
if value.startswith("cfg:") and "Ref." in value:
|
|
result.append({"column": f"{base}RRef", "reason": "single 1C reference type", "value_type": value})
|
|
elif value in {"xs:string", "xs:decimal", "xs:boolean", "xs:dateTime"}:
|
|
result.append({"column": base, "reason": "single primitive XML type", "value_type": value})
|
|
else:
|
|
result.append({"status": "unmapped_single_type", "value_type": value, "base": base})
|
|
return result
|
|
|
|
|
|
def predicted_composite_columns(types: list[str], routes: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
result = []
|
|
has_reference = any(value.startswith("cfg:") and "Ref." in value for value in types)
|
|
reference_types = [value for value in types if value.startswith("cfg:") and "Ref." in value]
|
|
has_string = "xs:string" in types
|
|
has_decimal = "xs:decimal" in types
|
|
has_boolean = "xs:boolean" in types
|
|
has_datetime = "xs:dateTime" in types
|
|
for route in routes:
|
|
if route.get("storage_role") != "Fld" or not route.get("physical_name_candidate"):
|
|
continue
|
|
base = route["physical_name_candidate"]
|
|
result.append({"column": f"{base}_TYPE", "reason": "composite value discriminator", "value_types": types})
|
|
if has_reference:
|
|
if len(reference_types) > 1:
|
|
result.append({"column": f"{base}_RTRef", "reason": "composite reference type id", "value_types": types})
|
|
result.append({"column": f"{base}_RRRef", "reason": "composite reference value", "value_types": types})
|
|
if has_string:
|
|
result.append({"column": f"{base}_S", "reason": "composite string value", "value_types": types})
|
|
if has_decimal:
|
|
result.append({"column": f"{base}_N", "reason": "composite numeric value", "value_types": types})
|
|
if has_boolean:
|
|
result.append({"column": f"{base}_B", "reason": "composite boolean value", "value_types": types})
|
|
if has_datetime:
|
|
result.append({"column": f"{base}_T", "reason": "composite datetime value", "value_types": types})
|
|
return result or [{"status": "unmapped_composite_type", "value_types": types}]
|
|
|
|
|
|
def enrich_batch(args: argparse.Namespace) -> dict[str, Any]:
|
|
db_report = load_json(args.dbnames)
|
|
routes_by_guid = route_index(dbnames_records(db_report))
|
|
batch = load_json(args.summary)
|
|
outputs = []
|
|
totals = {"metadata_item_count": 0, "metadata_items_with_routes": 0, "object_route_count": 0}
|
|
for row in batch.get("outputs") or []:
|
|
source = Path(row["output"])
|
|
report = load_json(source)
|
|
enriched = enrich_report(report, routes_by_guid)
|
|
output = args.output_dir / source.name
|
|
write_json(output, enriched)
|
|
summary = enriched["storage_route_summary"]
|
|
for key in totals:
|
|
totals[key] += summary[key]
|
|
outputs.append({**row, "output": str(output), "storage_route_summary": summary})
|
|
return {
|
|
"schema": "onec_structured_metadata_dbnames_enrichment_batch.v1",
|
|
"summary": str(args.summary),
|
|
"dbnames": str(args.dbnames),
|
|
"outputs": outputs,
|
|
"totals": totals,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Attach DBNames storage routes to structured metadata reports.")
|
|
parser.add_argument("--dbnames", type=Path, required=True)
|
|
parser.add_argument("--summary", type=Path, required=True)
|
|
parser.add_argument("--output-dir", type=Path, required=True)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
|
|
result = enrich_batch(args)
|
|
write_json(args.output, result)
|
|
print(json.dumps({"output": str(args.output), "totals": result["totals"]}, ensure_ascii=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|