127 lines
5.0 KiB
Python
127 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
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 api_server.time_utils import current_timestamp
|
|
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))
|