From 58afd3932ec608d6ec32ce4a048296321e9fd1e1 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Fri, 22 May 2026 15:08:36 +0300 Subject: [PATCH] Add XML export preflight for AI structure --- .../html5_ai_structure_controller.py | 128 +++++++++++++++++- services/api-server/src/api_server/main.py | 2 + services/api-server/tests/test_api.py | 26 ++++ 3 files changed, 155 insertions(+), 1 deletion(-) diff --git a/services/api-server/src/api_server/html5_ai_structure_controller.py b/services/api-server/src/api_server/html5_ai_structure_controller.py index 1a6dbeb..e17266e 100644 --- a/services/api-server/src/api_server/html5_ai_structure_controller.py +++ b/services/api-server/src/api_server/html5_ai_structure_controller.py @@ -168,11 +168,35 @@ async def html5_ai_structure_check_path( project_id: str, form: dict[str, list[str]], check_path: Callable[..., dict[str, Any]], + work_root: Path, + load_credentials: Callable[[str], SmbCredentials | None] | None = None, ) -> str: input_path = form_value(form, "input_path") if not input_path: return render_html5_ai_structure_path_check({"status": "error", "message": "Сначала укажите входной путь."}) - result = await check_path(project_id=project_id, input_path=input_path) + saved = load_credentials(project_id) if load_credentials else None + username = form_value(form, "smb_username") or (saved or {}).get("username", "") + password = form_value(form, "smb_password") or (saved or {}).get("password", "") + domain = form_value(form, "smb_domain") or (saved or {}).get("domain", "") + if _detect_binary_input(input_path): + result = await check_path(project_id=project_id, input_path=input_path) + return render_html5_ai_structure_path_check(result) + try: + result = _inspect_ai_structure_input( + raw_input_path=input_path, + work_root=work_root, + username=username, + password=password, + domain=domain, + ) + except FileNotFoundError as error: + result = {"status": "error", "message": str(error)} + except PermissionError as error: + result = {"status": "error", "message": f"Нет доступа к папке: {error}"} + except OSError as error: + result = {"status": "error", "message": f"Ошибка файловой системы: {error}"} + except RuntimeError as error: + result = {"status": "error", "message": str(error)} return render_html5_ai_structure_path_check(result) @@ -367,3 +391,105 @@ def _compose_ai_structure_source_root(target_root: Path, source_roots: list[Path def _enum_text(value: object) -> str: return str(getattr(value, "value", value or "")) + + +def _inspect_ai_structure_input( + *, + raw_input_path: str, + work_root: Path, + username: str, + password: str, + domain: str, +) -> dict[str, Any]: + input_path = str(raw_input_path or "").strip() + temp_root = work_root / f"inspect-{uuid4().hex}" + local_root = Path(input_path) + copied_from_unc = False + try: + if is_unc_path(input_path): + if not username or not password: + return { + "status": "error", + "message": "Для проверки сетевой XML-выгрузки укажите логин и пароль SMB.", + } + temp_root.mkdir(parents=True, exist_ok=True) + local_root = temp_root / "input" + copy_smb_tree_to_local( + source=input_path, + target=local_root, + username=username, + password=password, + domain=domain or None, + ) + copied_from_unc = True + if not local_root.exists(): + raise FileNotFoundError(f"Входная папка не найдена: {input_path}") + files = [local_root] if local_root.is_file() else sorted(path for path in local_root.rglob("*") if path.is_file()) + parseable = [path for path in files if path.suffix.casefold() in {".xml", ".mdo", ".bsl"}] + binaries = [path for path in files if path.suffix.casefold() in {".cf", ".cfe"}] + layout = _inspect_source_layout(local_root) + xml_count = sum(1 for path in files if path.suffix.casefold() == ".xml") + mdo_count = sum(1 for path in files if path.suffix.casefold() == ".mdo") + bsl_count = sum(1 for path in files if path.suffix.casefold() == ".bsl") + details = [ + f"Тип раскладки: {_layout_kind_text(layout['kind'])}", + f"Главная папка: {layout['main_configuration_root']}", + f"Папки расширений: {', '.join(layout['extension_roots']) or 'нет'}", + f"Файлов XML: {xml_count}", + f"Файлов MDO: {mdo_count}", + f"Файлов BSL: {bsl_count}", + ] + if copied_from_unc: + details.append("Проверка выполнена сервером после чтения UNC-пути по SMB.") + if parseable: + warnings: list[str] = [] + if layout["kind"] == "flat_or_mixed": + warnings.append("Папка `Конфигурация` не найдена. Сервер все равно попытается собрать проект по имеющимся XML/MDO/BSL.") + if binaries: + warnings.append("Во входной папке есть и бинарные .cf/.cfe, и XML-выгрузка. Для server-side подготовки будут использованы XML/MDO/BSL-файлы.") + return { + "status": "ok", + "message": "Сервер видит выгрузку 1С и может готовить пакет для Codex без Windows Agent.", + "details": details + warnings, + } + if binaries: + return { + "status": "info", + "message": "Во входной папке найдены только бинарные .cf/.cfe. Для них потребуется Windows Agent или runtime 1С.", + "details": details + [f"Бинарных файлов: {len(binaries)}"], + } + return { + "status": "error", + "message": "Во входной папке не найдены XML/MDO/BSL-файлы выгрузки 1С.", + "details": details, + } + finally: + if copied_from_unc and temp_root.exists(): + remove_tree(temp_root, expected_parent=work_root) + + +def _inspect_source_layout(root: Path) -> dict[str, Any]: + if root.is_file(): + return {"kind": "file", "main_configuration_root": root.name, "extension_roots": []} + children = [path for path in sorted(root.iterdir()) if path.is_dir()] + config_dir = next((path for path in children if path.name.casefold() in {"configuration", "конфигурация"}), None) + extension_roots = [ + path.name + for path in children + if path != config_dir and any(item.suffix.casefold() in {".xml", ".mdo", ".bsl"} for item in path.rglob("*") if item.is_file()) + ] + kind = "configuration_with_extensions" if config_dir else "flat_or_mixed" + return { + "kind": kind, + "main_configuration_root": config_dir.name if config_dir else root.name, + "extension_roots": extension_roots, + } + + +def _layout_kind_text(value: str) -> str: + mapping = { + "configuration_with_extensions": "Конфигурация + отдельные папки расширений", + "flat_or_mixed": "Плоская или смешанная выгрузка", + "file": "Отдельный входной файл", + } + return mapping.get(value, value) diff --git a/services/api-server/src/api_server/main.py b/services/api-server/src/api_server/main.py index b133e10..9978c03 100644 --- a/services/api-server/src/api_server/main.py +++ b/services/api-server/src/api_server/main.py @@ -2167,6 +2167,8 @@ async def html5_project_ai_structure_check_path(project_id: str, request: Reques project_id=project_id, form=form, check_path=_check_ai_structure_agent_path, + work_root=_storage.root / "ai_structure_work", + load_credentials=_load_ai_structure_smb_credentials, ) ) diff --git a/services/api-server/tests/test_api.py b/services/api-server/tests/test_api.py index f058006..0c6a7ea 100644 --- a/services/api-server/tests/test_api.py +++ b/services/api-server/tests/test_api.py @@ -2365,6 +2365,32 @@ def test_html5_ai_structure_check_path_reports_root_mismatch(): assert "доступны только корни" in checked.text +def test_html5_ai_structure_check_path_reports_xml_export_layout(tmp_path: Path): + source = tmp_path / "xml-export" + (source / "Конфигурация").mkdir(parents=True) + (source / "CRM").mkdir(parents=True) + (source / "Конфигурация" / "metadata.xml").write_text("", encoding="utf-8") + (source / "CRM" / "РасширениеCRM.mdo").write_text("", encoding="utf-8") + client = TestClient(app) + project_id = f"ai-xml-check-{uuid4()}" + + settings = client.post( + f"/projects/{project_id}/settings", + json={"name": "AI XML Check", "structure_source": "XML_DUMP"}, + ) + assert settings.status_code == 200 + + checked = client.post( + f"/html5/projects/{project_id}/ai-structure/check-path", + data={"input_path": str(source)}, + ) + assert checked.status_code == 200 + assert "Сервер видит выгрузку 1С" in checked.text + assert "Конфигурация + отдельные папки расширений" in checked.text + assert "Главная папка: Конфигурация" in checked.text + assert "Папки расширений: CRM" in checked.text + + def test_import_full_replace_replaces_current_normalized_project(tmp_path: Path): first = tmp_path / "first" second = tmp_path / "second"