Load EDT form elements on demand

services/api-server/src/api_server/main.py

@@ -227,8 +227,9 @@ from semantic_versioning import (
     SemanticObjectVersion,
     diff_versions,
 )
+from one_c_normalizer import parse_one_c_xml_file
 from semantic_kernel import index_project
-from sir import DiagnosticSeverity, EdgeKind, NodeKind, SirDelta, SirSnapshot, stable_hash
+from sir import DiagnosticSeverity, EdgeKind, NodeKind, SirDelta, SirSnapshot, make_lineage_id, stable_hash
 from storage_core import FileStorage, StoredSnapshotInfo
 from transaction_topology import routines_touching_target, transaction_write_sets
 from ui_semantics import form_semantics
@@ -4755,7 +4756,11 @@ async def get_object_ui(project_id: str, object_name: str) -> ObjectUiResponse:
     }
     return ObjectUiResponse(
         object=_named_node(object_node),
-        forms=_form_semantics_for_lineages(snapshot, form_lineages),
+        forms=_hydrate_object_ui_forms_from_source(
+            project_id,
+            object_node,
+            _form_semantics_for_lineages(snapshot, form_lineages),
+        ),
     )
 
 
@@ -7884,6 +7889,110 @@ def _form_semantics_for_lineages(snapshot: SirSnapshot, form_lineages: set[str])
     ]
 
 
+_EDT_SOURCE_DIRECTORY_BY_PREFIX = {
+    "РегистрБухгалтерии": "AccountingRegisters",
+    "РегистрНакопления": "AccumulationRegisters",
+    "БизнесПроцесс": "BusinessProcesses",
+    "РегистрРасчета": "CalculationRegisters",
+    "Справочник": "Catalogs",
+    "ПланСчетов": "ChartsOfAccounts",
+    "ПланВидовРасчета": "ChartsOfCalculationTypes",
+    "ПланВидовХарактеристик": "ChartsOfCharacteristicTypes",
+    "Обработка": "DataProcessors",
+    "ЖурналДокументов": "DocumentJournals",
+    "Документ": "Documents",
+    "Перечисление": "Enums",
+    "ПланОбмена": "ExchangePlans",
+    "ВнешнийИсточникДанных": "ExternalDataSources",
+    "РегистрСведений": "InformationRegisters",
+    "Отчет": "Reports",
+    "Задача": "Tasks",
+}
+
+
+def _hydrate_object_ui_forms_from_source(
+    project_id: str,
+    object_node,
+    forms: list[FormSemanticsResponse],
+) -> list[FormSemanticsResponse]:
+    source_root = _current_project_source_root(project_id)
+    if source_root is None:
+        return forms
+    return [
+        _hydrate_form_elements_from_source(project_id, source_root, object_node, form)
+        for form in forms
+    ]
+
+
+def _hydrate_form_elements_from_source(
+    project_id: str,
+    source_root: Path,
+    object_node,
+    form: FormSemanticsResponse,
+) -> FormSemanticsResponse:
+    if form.elements:
+        return form
+    form_file = _edt_form_file_for_object(source_root, object_node.qualified_name, form.form.name)
+    if form_file is None:
+        return form
+    try:
+        xml_objects = parse_one_c_xml_file(form_file)
+    except (OSError, UnicodeDecodeError, ET.ParseError):
+        return form
+    source_path = form_file.as_posix()
+    elements = [
+        NamedNode(
+            lineage_id=make_lineage_id(
+                NodeKind.FORM_ELEMENT.value,
+                f"{project_id}:{source_path}:ELEMENT:{item.qualified_name}",
+            ),
+            kind=NodeKind.FORM_ELEMENT.value,
+            name=item.name,
+            qualified_name=item.qualified_name,
+            attributes=item.attributes,
+        )
+        for item in xml_objects
+        if item.object_kind == "ELEMENT" and item.qualified_name.startswith(f"{form.form.qualified_name}.")
+    ]
+    if not elements:
+        return form
+    return FormSemanticsResponse(
+        form=form.form,
+        commands=form.commands,
+        elements=elements,
+        command_handlers=form.command_handlers,
+    )
+
+
+def _current_project_source_root(project_id: str) -> Path | None:
+    state = _project_setup.get(project_id, {})
+    last_import = state.get("last_import") or {}
+    source_path = last_import.get("source_path")
+    if not source_path:
+        return None
+    root = Path(source_path)
+    return root if root.exists() else None
+
+
+def _edt_form_file_for_object(source_root: Path, object_qualified_name: str, form_name: str) -> Path | None:
+    parts = object_qualified_name.split(".")
+    if len(parts) < 2:
+        return None
+    prefix, object_name = parts[0], parts[1]
+    source_dirs = [source_root / "src", *(path / "src" for path in source_root.iterdir() if path.is_dir())]
+    if prefix == "ОбщаяФорма":
+        candidates = [source_dir / "CommonForms" / object_name / "Form.form" for source_dir in source_dirs]
+    else:
+        directory = _EDT_SOURCE_DIRECTORY_BY_PREFIX.get(prefix)
+        if directory is None:
+            return None
+        candidates = [
+            source_dir / directory / object_name / "Forms" / form_name / "Form.form"
+            for source_dir in source_dirs
+        ]
+    return next((candidate for candidate in candidates if candidate.exists()), None)
+
+
 def _persist_job(job: OperationJob) -> OperationJob:
     _storage.write_document("operations_jobs", job.job_id, job.model_dump(mode="json"))
     return job