258 lines
11 KiB
Python
258 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import xml.etree.ElementTree as ET
|
||
from collections import Counter, defaultdict
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Iterable
|
||
|
||
|
||
DEFAULT_ROOT = Path(r"Z:\codex\1C\XML\UPO\Структура базы 1с")
|
||
|
||
|
||
def local_name(tag: str) -> str:
|
||
return tag.rsplit("}", 1)[-1]
|
||
|
||
|
||
def direct_child(element: ET.Element, name: str) -> ET.Element | None:
|
||
return next((child for child in element if local_name(child.tag) == name), None)
|
||
|
||
|
||
def property_text(properties: ET.Element | None, name: str) -> str:
|
||
if properties is None:
|
||
return ""
|
||
node = next((child for child in properties if local_name(child.tag) == name), None)
|
||
return str(node.text or "").strip() if node is not None else ""
|
||
|
||
|
||
def parse_metadata_file(path: Path, layer: str) -> dict[str, Any]:
|
||
try:
|
||
root = ET.parse(path).getroot()
|
||
metadata = next(iter(root), None) if local_name(root.tag) == "MetaDataObject" else root
|
||
if metadata is None:
|
||
raise ValueError("metadata object element is missing")
|
||
kind = local_name(metadata.tag)
|
||
properties = direct_child(metadata, "Properties")
|
||
children = direct_child(metadata, "ChildObjects")
|
||
child_schemas: dict[str, dict[str, Any]] = {}
|
||
if children is not None:
|
||
grouped: dict[str, list[ET.Element]] = defaultdict(list)
|
||
for child in children:
|
||
grouped[local_name(child.tag)].append(child)
|
||
for child_kind, values in grouped.items():
|
||
child_properties: set[str] = set()
|
||
for value in values:
|
||
value_properties = direct_child(value, "Properties")
|
||
if value_properties is not None:
|
||
child_properties.update(local_name(item.tag) for item in value_properties)
|
||
child_schemas[child_kind] = {"count": len(values), "properties": sorted(child_properties)}
|
||
return {
|
||
"status": "ok",
|
||
"layer": layer,
|
||
"path": str(path),
|
||
"kind": kind,
|
||
"name": property_text(properties, "Name"),
|
||
"uuid": str(metadata.attrib.get("uuid") or "").lower(),
|
||
"properties": sorted(local_name(child.tag) for child in properties) if properties is not None else [],
|
||
"child_schemas": child_schemas,
|
||
}
|
||
except Exception as exc:
|
||
return {"status": "error", "layer": layer, "path": str(path), "message": str(exc)}
|
||
|
||
|
||
def layer_files(root: Path) -> list[Path]:
|
||
files = [root / "Configuration.xml"] if (root / "Configuration.xml").is_file() else []
|
||
for folder in root.iterdir():
|
||
if folder.is_dir() and folder.name != "Ext":
|
||
files.extend(sorted(folder.glob("*.xml")))
|
||
return files
|
||
|
||
|
||
def artifact_files(root: Path, max_depth: int = 5) -> list[Path]:
|
||
result: set[Path] = set()
|
||
for folder in (path for path in root.iterdir() if path.is_dir()):
|
||
for depth in range(1, max_depth + 1):
|
||
pattern = "/".join(["*"] * depth + ["Ext", "*.xml"])
|
||
result.update(path for path in folder.glob(pattern) if path.is_file())
|
||
return sorted(result)
|
||
|
||
|
||
def parse_artifact(path: Path) -> dict[str, Any]:
|
||
tags: set[str] = set()
|
||
attributes: dict[str, set[str]] = defaultdict(set)
|
||
root_tag = ""
|
||
try:
|
||
for _event, element in ET.iterparse(path, events=("start",)):
|
||
tag = local_name(element.tag)
|
||
if not root_tag:
|
||
root_tag = tag
|
||
tags.add(tag)
|
||
attributes[tag].update(local_name(name) for name in element.attrib)
|
||
return {
|
||
"status": "ok",
|
||
"path": str(path),
|
||
"artifact": path.name,
|
||
"root_tag": root_tag,
|
||
"tags": sorted(tags),
|
||
"attributes": {key: sorted(value) for key, value in sorted(attributes.items()) if value},
|
||
}
|
||
except Exception as exc:
|
||
return {"status": "error", "path": str(path), "artifact": path.name, "message": str(exc)}
|
||
|
||
|
||
def scan_artifacts(root: Path, workers: int) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
||
files = artifact_files(root)
|
||
with ThreadPoolExecutor(max_workers=max(1, workers)) as executor:
|
||
parsed = list(executor.map(parse_artifact, files))
|
||
errors = [item for item in parsed if item["status"] != "ok"]
|
||
grouped: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
|
||
for item in parsed:
|
||
if item["status"] == "ok":
|
||
grouped[(str(item["artifact"]), str(item["root_tag"]))].append(item)
|
||
schemas: dict[str, Any] = {}
|
||
for (artifact, root_tag), values in sorted(grouped.items()):
|
||
tags: set[str] = set()
|
||
attributes: dict[str, set[str]] = defaultdict(set)
|
||
for value in values:
|
||
tags.update(value["tags"])
|
||
for tag, names in value["attributes"].items():
|
||
attributes[tag].update(names)
|
||
key = f"{artifact}:{root_tag}"
|
||
schemas[key] = {
|
||
"files": len(values),
|
||
"root_tag": root_tag,
|
||
"tags": sorted(tags),
|
||
"attributes": {tag: sorted(names) for tag, names in sorted(attributes.items())},
|
||
"samples": [value["path"] for value in values[:3]],
|
||
}
|
||
return {"files": len(files), "schemas": schemas}, errors
|
||
|
||
|
||
def scan_layer(root: Path, layer: str, workers: int) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||
files = layer_files(root)
|
||
with ThreadPoolExecutor(max_workers=max(1, workers)) as executor:
|
||
parsed = list(executor.map(lambda path: parse_metadata_file(path, layer), files))
|
||
return [item for item in parsed if item["status"] == "ok"], [item for item in parsed if item["status"] != "ok"]
|
||
|
||
|
||
def merge_kind_schemas(objects: Iterable[dict[str, Any]]) -> dict[str, Any]:
|
||
kinds: dict[str, dict[str, Any]] = {}
|
||
grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
||
for item in objects:
|
||
grouped[str(item["kind"])].append(item)
|
||
for kind, values in sorted(grouped.items()):
|
||
properties: set[str] = set()
|
||
child_counts: Counter[str] = Counter()
|
||
child_properties: dict[str, set[str]] = defaultdict(set)
|
||
for value in values:
|
||
properties.update(value.get("properties") or [])
|
||
for child_kind, schema in (value.get("child_schemas") or {}).items():
|
||
child_counts[child_kind] += int(schema.get("count") or 0)
|
||
child_properties[child_kind].update(schema.get("properties") or [])
|
||
kinds[kind] = {
|
||
"objects": len(values),
|
||
"properties": sorted(properties),
|
||
"children": {
|
||
child_kind: {"objects": child_counts[child_kind], "properties": sorted(child_properties[child_kind])}
|
||
for child_kind in sorted(child_counts)
|
||
},
|
||
"samples": [
|
||
{"ref": f"{kind}.{value['name']}" if value.get("name") else kind, "uuid": value.get("uuid")}
|
||
for value in values[:3]
|
||
],
|
||
}
|
||
return kinds
|
||
|
||
|
||
def object_ref(item: dict[str, Any]) -> str:
|
||
return f"{item.get('kind')}.{item.get('name')}" if item.get("name") else f"{item.get('kind')}#{item.get('uuid')}"
|
||
|
||
|
||
def extension_summary(name: str, objects: list[dict[str, Any]], base_refs: set[str]) -> dict[str, Any]:
|
||
refs = {object_ref(item) for item in objects}
|
||
return {
|
||
"name": name,
|
||
"objects": len(objects),
|
||
"kinds": merge_kind_schemas(objects),
|
||
"overrides": sorted(refs & base_refs),
|
||
"extension_only": sorted(refs - base_refs),
|
||
"counts": {"overrides": len(refs & base_refs), "extension_only": len(refs - base_refs)},
|
||
}
|
||
|
||
|
||
def build_report(root: Path, workers: int, include_artifacts: bool) -> dict[str, Any]:
|
||
configuration_root = root / "Конфигурация"
|
||
extensions_root = root / "Расширения"
|
||
base_objects, errors = scan_layer(configuration_root, "configuration", workers)
|
||
base_refs = {object_ref(item) for item in base_objects}
|
||
extensions: list[dict[str, Any]] = []
|
||
configuration_artifacts: dict[str, Any] = {"status": "not_requested", "files": 0, "schemas": {}}
|
||
if include_artifacts:
|
||
configuration_artifacts, artifact_errors = scan_artifacts(configuration_root, workers)
|
||
configuration_artifacts["status"] = "ok" if not artifact_errors else "partial"
|
||
errors.extend(artifact_errors)
|
||
if extensions_root.is_dir():
|
||
for extension_root in sorted(path for path in extensions_root.iterdir() if path.is_dir()):
|
||
objects, extension_errors = scan_layer(extension_root, f"extension:{extension_root.name}", workers)
|
||
errors.extend(extension_errors)
|
||
summary = extension_summary(extension_root.name, objects, base_refs)
|
||
if include_artifacts:
|
||
artifacts, artifact_errors = scan_artifacts(extension_root, workers)
|
||
artifacts["status"] = "ok" if not artifact_errors else "partial"
|
||
summary["artifacts"] = artifacts
|
||
errors.extend(artifact_errors)
|
||
extensions.append(summary)
|
||
kinds = merge_kind_schemas(base_objects)
|
||
return {
|
||
"schema": "onec_xml_metadata_analysis.v1",
|
||
"status": "ok" if not errors else "partial",
|
||
"generated_at": datetime.now(timezone.utc).isoformat(),
|
||
"source_root": str(root),
|
||
"assumption": "The exported base configuration is equivalent to the SQL base; extensions are independent overlays and may differ.",
|
||
"configuration": {
|
||
"objects": len(base_objects),
|
||
"kinds": kinds,
|
||
"counts": {"kinds": len(kinds), "objects": len(base_objects)},
|
||
"artifacts": configuration_artifacts,
|
||
},
|
||
"extensions": extensions,
|
||
"counts": {
|
||
"configuration_kinds": len(kinds),
|
||
"configuration_objects": len(base_objects),
|
||
"extensions": len(extensions),
|
||
"extension_objects": sum(item["objects"] for item in extensions),
|
||
"artifact_files": int(configuration_artifacts.get("files") or 0) + sum(int((item.get("artifacts") or {}).get("files") or 0) for item in extensions),
|
||
"parse_errors": len(errors),
|
||
},
|
||
"errors": errors[:100],
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Extract complete declared metadata property schemas from a 1C XML configuration export.")
|
||
parser.add_argument("--root", type=Path, default=DEFAULT_ROOT)
|
||
parser.add_argument("--workers", type=int, default=8)
|
||
parser.add_argument("--include-artifacts", action="store_true", help="Also scan nested Ext XML files such as forms, rights, and templates.")
|
||
parser.add_argument("--output", type=Path)
|
||
parser.add_argument("--json", action="store_true", help="Print the full JSON report instead of a compact summary.")
|
||
args = parser.parse_args()
|
||
report = build_report(args.root, args.workers, args.include_artifacts)
|
||
rendered = json.dumps(report, ensure_ascii=False, indent=2)
|
||
if args.output:
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(rendered + "\n", encoding="utf-8")
|
||
if args.json:
|
||
print(rendered)
|
||
else:
|
||
print(json.dumps({"status": report["status"], **report["counts"]}, ensure_ascii=False))
|
||
return 0 if report["status"] == "ok" else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|