Extract import summary service

2026-05-17 22:32:38 +03:00
parent 588edfcf24
commit c979428d90
2 changed files with 141 additions and 115 deletions
@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from api_server.import_models import ImportSummary, SnapshotSummary
+from api_server.import_sync_models import ImportSyncPreview
+from api_server.normalized_project_models import NormalizedProjectSummary
+from api_server.normalized_project_service import normalized_project_summary
+from one_c_normalizer import NormalizedProject
+from sir import NodeKind, SirSnapshot
+
+
+def snapshot_summary(snapshot: SirSnapshot) -> SnapshotSummary:
+    return SnapshotSummary(
+        snapshot_id=snapshot.snapshot_id,
+        project_id=snapshot.project_id,
+        snapshot_hash=snapshot.snapshot_hash,
+        node_count=len(snapshot.nodes),
+        edge_count=len(snapshot.edges),
+        diagnostics_count=len(snapshot.diagnostics),
+        unresolved_references_count=len(snapshot.unresolved_references),
+    )
+
+
+def import_summary_from_snapshot(
+    *,
+    project_id: str,
+    source: str,
+    status: str,
+    snapshot: SirSnapshot | None,
+    errors: list[str],
+    metadata: dict,
+    runtime_mode: str,
+    runtime_diagnostics: list[str],
+    normalized: NormalizedProject | None,
+    mode: str = "FULL_REPLACE",
+    applied: bool = True,
+    sync_preview: ImportSyncPreview | None = None,
+) -> ImportSummary:
+    normalized_summary = normalized_project_summary(normalized) if normalized is not None else None
+    empty_counts = snapshot is None and normalized_summary is None and not applied
+    if snapshot is None:
+        return ImportSummary(
+            source=source,
+            mode=mode,
+            applied=applied,
+            status=status,
+            last_import=_current_timestamp(),
+            source_path=None,
+            runtime_mode=runtime_mode,
+            runtime_diagnostics=runtime_diagnostics,
+            errors=errors,
+            diagnostics_count=0,
+            diagnostics=[],
+            object_count=normalized_summary.object_count if normalized_summary is not None else 0 if empty_counts else 12,
+            module_count=normalized_summary.module_count if normalized_summary is not None else 0 if empty_counts else 4,
+            form_count=normalized_summary.form_count if normalized_summary is not None else 0 if empty_counts else 3,
+            role_count=normalized_summary.role_count if normalized_summary is not None else 0 if empty_counts else 2,
+            extensions=_summary_extensions(normalized_summary, metadata, empty_counts, default=["DemoExtension"]),
+            platform_version=metadata.get("platform_version", "mock-8.3"),
+            compatibility_mode=metadata.get("compatibility_mode", "mock"),
+            normalized_summary=normalized_summary,
+            sync_preview=sync_preview,
+        )
+    object_kinds = {
+        NodeKind.CATALOG,
+        NodeKind.DOCUMENT,
+        NodeKind.REGISTER,
+        NodeKind.REPORT,
+        NodeKind.DATA_PROCESSOR,
+        NodeKind.COMMON_MODULE,
+        NodeKind.EXCHANGE_PLAN,
+        NodeKind.BUSINESS_PROCESS,
+        NodeKind.TASK,
+        NodeKind.SUBSYSTEM,
+        NodeKind.HTTP_SERVICE,
+        NodeKind.XDTO_PACKAGE,
+        NodeKind.EXTENSION,
+    }
+    return ImportSummary(
+        source=source,
+        mode=mode,
+        applied=applied,
+        status=status,
+        last_import=_current_timestamp(),
+        source_path=snapshot.metadata.source_root,
+        runtime_mode=runtime_mode,
+        runtime_diagnostics=runtime_diagnostics,
+        errors=errors,
+        diagnostics_count=len(snapshot.diagnostics),
+        diagnostics=[
+            f"{diagnostic.severity.value} {diagnostic.code}: {diagnostic.message}"
+            for diagnostic in snapshot.diagnostics[:20]
+        ],
+        object_count=normalized_summary.object_count
+        if normalized_summary is not None
+        else sum(1 for node in snapshot.nodes if node.kind in object_kinds),
+        module_count=normalized_summary.module_count
+        if normalized_summary is not None
+        else sum(1 for node in snapshot.nodes if node.kind == NodeKind.MODULE),
+        form_count=normalized_summary.form_count
+        if normalized_summary is not None
+        else sum(1 for node in snapshot.nodes if node.kind == NodeKind.FORM),
+        role_count=normalized_summary.role_count
+        if normalized_summary is not None
+        else sum(1 for node in snapshot.nodes if node.kind == NodeKind.ROLE),
+        extensions=normalized_summary.extensions if normalized_summary is not None else list(metadata.get("extensions", [])),
+        platform_version=metadata.get("platform_version"),
+        compatibility_mode=metadata.get("compatibility_mode"),
+        snapshot=snapshot_summary(snapshot),
+        normalized_summary=normalized_summary,
+        sync_preview=sync_preview,
+    )
+
+
+def _summary_extensions(
+    normalized_summary: NormalizedProjectSummary | None,
+    metadata: dict,
+    empty_counts: bool,
+    *,
+    default: list[str],
+) -> list[str]:
+    if normalized_summary is not None:
+        return normalized_summary.extensions
+    if empty_counts:
+        return []
+    return list(metadata.get("extensions", default))
+
+
+def _current_timestamp() -> str:
+    return datetime.now(timezone.utc).isoformat()