from __future__ import annotations import hashlib import json import mimetypes import re import shutil import tempfile from io import BytesIO from copy import deepcopy from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from uuid import uuid4 from mcp.server.fastmcp import FastMCP from openpyxl import Workbook, load_workbook from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, message_looks_like_invalid_token from ..import_store import ImportJobStore, ImportVerificationStore from ..json_types import JSONObject from .app_tools import _derive_import_capability from .base import ToolBase from .file_tools import FileTools from .record_tools import RecordTools, _build_field_index, _normalize_form_schema SUPPORTED_IMPORT_EXTENSIONS = {".xlsx", ".xls"} REPAIRABLE_IMPORT_EXTENSIONS = {".xlsx"} SAFE_REPAIRS = { "normalize_headers", "trim_trailing_blank_rows", "normalize_enum_values", "normalize_date_formats", "normalize_number_formats", "normalize_url_cells", } EMAIL_PATTERN = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") IMPORT_STATUS_BY_PROCESS_STATUS = { 1: "queued", 2: "running", 3: "succeeded", 4: "failed", 5: "partially_failed", } class ImportTools(ToolBase): """导入工具（中文名：数据导入与校验）。 类型：批量数据导入工具。 主要职责： 1. 获取导入模板与导入 schema； 2. 执行导入文件校验与本地修复； 3. 启动导入任务并查询导入进度与结果。 """ def __init__( self, sessions, backend, *, verification_store: ImportVerificationStore | None = None, job_store: ImportJobStore | None = None, ) -> None: """执行内部辅助逻辑。""" super().__init__(sessions, backend) self._record_tools = RecordTools(sessions, backend) self._file_tools = FileTools(sessions, backend) self._verification_store = verification_store or ImportVerificationStore() self._job_store = job_store or ImportJobStore() def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" @mcp.tool() def record_import_schema_get( profile: str = DEFAULT_PROFILE, app_key: str = "", output_profile: str = "normal", ) -> dict[str, Any]: return self.record_import_schema_get( profile=profile, app_key=app_key, output_profile=output_profile, ) @mcp.tool(description="Get the official app import template and the expected applicant import columns.") def record_import_template_get( profile: str = DEFAULT_PROFILE, app_key: str = "", download_to_path: str | None = None, ) -> dict[str, Any]: return self.record_import_template_get( profile=profile, app_key=app_key, download_to_path=download_to_path, ) @mcp.tool(description="Verify a local Excel import file and produce the only verification_id allowed for import start.") def record_import_verify( profile: str = DEFAULT_PROFILE, app_key: str = "", file_path: str = "", ) -> dict[str, Any]: return self.record_import_verify( profile=profile, app_key=app_key, file_path=file_path, ) @mcp.tool(description="Repair a local .xlsx import file after explicit user authorization, then re-verify it.") def record_import_repair_local( profile: str = DEFAULT_PROFILE, verification_id: str = "", authorized_file_modification: bool = False, output_path: str | None = None, selected_repairs: list[str] | None = None, ) -> dict[str, Any]: return self.record_import_repair_local( profile=profile, verification_id=verification_id, authorized_file_modification=authorized_file_modification, output_path=output_path, selected_repairs=selected_repairs, ) @mcp.tool(description="Start import from a successful verification_id. being_enter_auditing must be passed explicitly.") def record_import_start( profile: str = DEFAULT_PROFILE, app_key: str = "", verification_id: str = "", being_enter_auditing: bool | None = None, view_key: str | None = None, ) -> dict[str, Any]: return self.record_import_start( profile=profile, app_key=app_key, verification_id=verification_id, being_enter_auditing=being_enter_auditing, view_key=view_key, ) @mcp.tool(description="Get import status by process_id_str, import_id, or the latest remembered import in the current app.") def record_import_status_get( profile: str = DEFAULT_PROFILE, app_key: str = "", import_id: str | None = None, process_id_str: str | None = None, ) -> dict[str, Any]: selector_count = sum( 1 for item in ( bool(_normalize_optional_text(process_id_str)), bool(_normalize_optional_text(import_id)), bool(str(app_key or "").strip()), ) if item ) if selector_count != 1: return self._failed_status_result( error_code="CONFIG_ERROR", message="record_import_status_get accepts exactly one selector: process_id_str, import_id, or app_key", extra={ "details": { "fix_hint": "Use `process_id_str` or `import_id` for a known import, or use only `app_key` to inspect the latest import in that app.", } }, ) return self.record_import_status_get( profile=profile, app_key=app_key, import_id=import_id, process_id_str=process_id_str, ) def record_import_schema_get( self, *, profile: str = DEFAULT_PROFILE, app_key: str, output_profile: str = "normal", ) -> dict[str, Any]: """执行记录相关逻辑。""" if not app_key.strip(): return { "ok": False, "status": "blocked", "app_key": app_key, "error_code": "IMPORT_SCHEMA_UNAVAILABLE", "message": "app_key is required", } def runner(session_profile, context): import_capability, import_warnings = self._fetch_import_capability(context, app_key) if import_capability.get("can_import") is False and import_capability.get("auth_source") != "unknown": return { "ok": False, "status": "failed", "app_key": app_key, "ws_id": session_profile.selected_ws_id, "request_route": self.backend.describe_route(context), "error_code": "IMPORT_AUTH_PRECHECK_FAILED", "message": "the current user does not have import permission for this app", "warnings": import_warnings, "import_capability": import_capability, "verification": { "import_auth_prechecked": True, "import_auth_precheck_passed": False, }, } _index, expected_columns, schema_fingerprint = self._resolve_import_schema_bundle( profile, context, app_key, import_capability=import_capability, ) columns: list[JSONObject] = [] for column in expected_columns: payload: JSONObject = { "title": column["title"], "kind": column["write_kind"], "required": bool(column.get("required")), } if isinstance(column.get("options"), list) and column.get("options"): payload["options"] = column["options"] if column["write_kind"] == "member": payload["import_value_format"] = "member_email" payload["format_hint"] = "Import files must use member email values." elif column["write_kind"] == "relation": payload["import_value_format"] = "target_apply_id" payload["format_hint"] = "Import files must use the target record apply_id." elif column["write_kind"] == "department": payload["import_value_format"] = "department_name" payload["format_hint"] = "Import files may use a department name within the field candidate scope." if bool(column.get("requires_upload")): payload["requires_upload"] = True if isinstance(column.get("target_app_key"), str): payload["target_app_key"] = column["target_app_key"] if isinstance(column.get("target_app_name"), str): payload["target_app_name"] = column["target_app_name"] if isinstance(column.get("searchable_fields"), list) and column.get("searchable_fields"): payload["searchable_fields"] = column["searchable_fields"] columns.append(payload) response: dict[str, Any] = { "ok": True, "status": "success", "app_key": app_key, "ws_id": session_profile.selected_ws_id, "request_route": self.backend.describe_route(context), "warnings": import_warnings, "schema_scope": "import_ready", "columns": columns, "schema_fingerprint": schema_fingerprint, } if output_profile == "verbose": response["expected_columns"] = expected_columns response["import_capability"] = import_capability return response return self._run(profile, runner, tool_name='导入 Schema') def record_import_template_get( self, *, profile: str, app_key: str, download_to_path: str | None = None, ) -> dict[str, Any]: """执行记录相关逻辑。""" if not app_key.strip(): return self._failed_template_result(app_key=app_key, error_code="IMPORT_TEMPLATE_UNAUTHORIZED", message="app_key is required") def runner(session_profile, context): import_capability, import_warnings = self._fetch_import_capability(context, app_key) if import_capability.get("can_import") is False and import_capability.get("auth_source") != "unknown": return self._failed_template_result( app_key=app_key, error_code="IMPORT_AUTH_PRECHECK_FAILED", message="the current user does not have import permission for this app", request_route=self.backend.describe_route(context), extra={ "warnings": import_warnings, "import_capability": import_capability, "verification": { "import_auth_prechecked": True, "import_auth_precheck_passed": False, }, }, ) field_index, expected_columns, schema_fingerprint = self._resolve_import_schema_bundle( profile, context, app_key, import_capability=import_capability, ) try: payload = self.backend.request("GET", context, f"/app/{app_key}/apply/excelTemplate") except QingflowApiError as exc: can_generate_local_template = bool(expected_columns) and ( import_capability.get("auth_source") == "apply_auth" or ( import_capability.get("auth_source") == "unknown" and not is_auth_like_error(exc) and backend_code_int(exc) in {40002, 40027} ) ) if can_generate_local_template: downloaded_to_path = self._write_local_template( expected_columns=expected_columns, destination_hint=download_to_path, app_key=app_key, ) template_warning = { "code": "IMPORT_TEMPLATE_LOCAL_FALLBACK", "message": "Official template download requires data management permission; MCP generated a local applicant-import template instead.", } if import_capability.get("auth_source") == "unknown": template_warning = { "code": "IMPORT_TEMPLATE_LOCAL_FALLBACK_AUTH_UNKNOWN", "message": "Official template download was permission-restricted and import permission could not be prechecked; MCP generated a local applicant-import template from readable applicant fields.", } _copy_api_error_fields(template_warning, exc) return { "ok": True, "status": "partial_success", "app_key": app_key, "ws_id": session_profile.selected_ws_id, "request_route": self.backend.describe_route(context), "template_url": None, "downloaded_to_path": downloaded_to_path, "expected_columns": expected_columns, "schema_fingerprint": schema_fingerprint, "warnings": import_warnings + [template_warning], "verification": { "schema_fingerprint": schema_fingerprint, "template_url_resolved": False, "template_downloaded": True, "template_source": "local_generated", "import_auth_prechecked": import_capability.get("auth_source") != "unknown", }, } return self._failed_template_result( app_key=app_key, error_code="IMPORT_TEMPLATE_UNAUTHORIZED", message=exc.message, request_route=self.backend.describe_route(context), ) template_url = _pick_template_url(payload) if not template_url: return self._failed_template_result( app_key=app_key, error_code="IMPORT_TEMPLATE_UNAUTHORIZED", message="template endpoint did not return excelUrl", request_route=self.backend.describe_route(context), ) downloaded_to_path = None warnings: list[JSONObject] = list(import_warnings) verification = { "schema_fingerprint": schema_fingerprint, "template_url_resolved": True, "template_downloaded": False, "template_source": "official", } if download_to_path: destination = _resolve_template_download_path(download_to_path, app_key=app_key) destination.parent.mkdir(parents=True, exist_ok=True) content = self.backend.download_binary(template_url) destination.write_bytes(content) downloaded_to_path = str(destination) verification["template_downloaded"] = True return { "ok": True, "status": "success", "app_key": app_key, "ws_id": session_profile.selected_ws_id, "request_route": self.backend.describe_route(context), "template_url": template_url, "downloaded_to_path": downloaded_to_path, "expected_columns": expected_columns, "schema_fingerprint": schema_fingerprint, "warnings": warnings, "verification": verification, } try: return self._run(profile, runner, tool_name='导入模板') except RuntimeError as exc: return self._runtime_error_as_result(exc, error_code="IMPORT_TEMPLATE_UNAUTHORIZED") def record_import_verify( self, *, profile: str, app_key: str, file_path: str, ) -> dict[str, Any]: """执行记录相关逻辑。""" if not app_key.strip(): return self._failed_verify_result(app_key=app_key, file_path=file_path, error_code="IMPORT_VERIFICATION_FAILED", message="app_key is required") path = Path(file_path).expanduser() if not path.is_file(): return self._failed_verify_result(app_key=app_key, file_path=file_path, error_code="IMPORT_VERIFICATION_FAILED", message="file_path must point to an existing file") def runner(session_profile, context): import_capability, import_warnings = self._fetch_import_capability(context, app_key) precheck_known = import_capability.get("auth_source") != "unknown" if not bool(import_capability.get("can_import")): if import_capability.get("auth_source") != "unknown": return self._failed_verify_result( app_key=app_key, file_path=file_path, error_code="IMPORT_AUTH_PRECHECK_FAILED", message="the current user does not have import permission for this app", extra={ "warnings": import_warnings, "verification": { "import_auth_prechecked": True, "import_auth_precheck_passed": False, "backend_verification_passed": False, }, "import_capability": import_capability, }, ) import_warnings = list(import_warnings) + [ { "code": "IMPORT_AUTH_PRECHECK_SKIPPED", "message": "record_import_verify could not determine import permission from app metadata; continuing with file verification only.", } ] field_index, expected_columns, schema_fingerprint = self._resolve_import_schema_bundle( profile, context, app_key, import_capability=import_capability, ) template_header_profile, header_warnings = self._load_template_header_profile( context, app_key, import_capability=import_capability, expected_columns=expected_columns, ) template_header_titles = template_header_profile.get("allowed_titles") local_check = self._local_verify( profile=profile, context=context, path=path, app_key=app_key, field_index=field_index, expected_columns=expected_columns, allowed_header_titles=template_header_titles, schema_fingerprint=schema_fingerprint, ) effective_path = path effective_local_check = local_check auto_normalization = None try: auto_normalization = self._maybe_auto_normalize_file( source_path=path, expected_columns=expected_columns, template_header_profile=template_header_profile, local_check=local_check, ) except Exception as exc: effective_local_check = deepcopy(local_check) effective_local_check["issues"].append( _issue( "IMPORT_AUTO_NORMALIZATION_FAILED", f"Workbook compatibility normalization failed before backend verification: {exc}", severity="error", ) ) effective_local_check["warnings"].append( { "code": "IMPORT_AUTO_NORMALIZATION_FAILED", "message": "Workbook compatibility normalization failed during local precheck; returning a structured verification failure instead of crashing.", } ) effective_local_check["local_precheck_passed"] = False effective_local_check["can_import"] = False effective_local_check["error_code"] = "IMPORT_VERIFICATION_FAILED" if auto_normalization is not None: effective_path = Path(str(auto_normalization["verified_file_path"])) effective_local_check = self._local_verify( profile=profile, context=context, path=effective_path, app_key=app_key, field_index=field_index, expected_columns=expected_columns, allowed_header_titles=list(auto_normalization["header_titles"]), schema_fingerprint=schema_fingerprint, ) warnings = import_warnings + deepcopy(effective_local_check["warnings"]) + header_warnings if auto_normalization is not None: warnings.extend(deepcopy(auto_normalization["warnings"])) issues = deepcopy(effective_local_check["issues"]) can_import = bool(effective_local_check["can_import"]) backend_verification = None backend_failure_error_code = None if can_import: try: payload = self.backend.request_multipart( "POST", context, f"/app/{app_key}/upload/verification", files={ "file": ( effective_path.name, effective_path.read_bytes(), mimetypes.guess_type(effective_path.name)[0] or "application/octet-stream", ) }, ) if isinstance(payload, dict): backend_verification = payload else: backend_verification = {} being_validated = backend_verification.get("beingValidated", True) if being_validated is False: can_import = False issues.append( _issue( "BACKEND_IMPORT_VERIFICATION_REJECTED", "Backend verification rejected the file for import.", severity="error", ) ) except QingflowApiError as exc: can_import = False backend_error_code = _import_permission_error_code( exc, permission_code="IMPORT_VERIFICATION_UNAUTHORIZED", default="BACKEND_IMPORT_VERIFICATION_FAILED", ) backend_failure_error_code = backend_error_code issues.append( _issue( backend_error_code, exc.message or "Backend import verification failed.", severity="error", ) ) warning = { "code": backend_error_code, "message": "Backend verification failed; the file cannot be imported until verification succeeds.", } _copy_api_error_fields(warning, exc) warnings.append(warning) verification_id = str(uuid4()) verification_payload = { "id": verification_id, "created_at": _utc_now().isoformat(), "profile": profile, "app_key": app_key, "file_path": str(path.resolve()), "source_file_path": str(path.resolve()), "verified_file_path": str(effective_path.resolve()) if effective_path != path else None, "file_name": path.name, "file_sha256": local_check["file_sha256"], "verified_file_sha256": effective_local_check["file_sha256"] if effective_path != path else None, "file_size": local_check["file_size"], "schema_fingerprint": schema_fingerprint, "can_import": can_import, "issues": issues, "warnings": warnings, "import_capability": import_capability, "apply_rows": backend_verification.get("applyRows") if isinstance(backend_verification, dict) else None, "backend_verification": backend_verification, "local_precheck": effective_local_check, "source_local_precheck": local_check, "auto_normalization": auto_normalization, } self._verification_store.put(verification_id, verification_payload) return { "ok": can_import, "status": "success" if can_import else "failed", "error_code": None if can_import else (effective_local_check.get("error_code") or local_check.get("error_code") or backend_failure_error_code or "IMPORT_VERIFICATION_FAILED"), "can_import": can_import, "verification_id": verification_id, "file_path": str(path.resolve()), "verified_file_path": str(effective_path.resolve()) if effective_path != path else None, "file_name": path.name, "file_sha256": local_check["file_sha256"], "verified_file_sha256": effective_local_check["file_sha256"] if effective_path != path else None, "file_size": local_check["file_size"], "schema_fingerprint": schema_fingerprint, "apply_rows": backend_verification.get("applyRows") if isinstance(backend_verification, dict) else None, "issues": issues, "repair_suggestions": local_check["repair_suggestions"], "warnings": warnings, "import_capability": import_capability, "verification": { "import_auth_prechecked": precheck_known, "import_auth_precheck_passed": True if precheck_known else None, "import_auth_source": import_capability.get("auth_source"), "local_precheck_passed": bool(effective_local_check["local_precheck_passed"]), "backend_verification_passed": isinstance(backend_verification, dict) and backend_verification.get("beingValidated", True) is not False, "schema_fingerprint": schema_fingerprint, "file_sha256": local_check["file_sha256"], "verified_file_sha256": effective_local_check["file_sha256"] if effective_path != path else None, "file_format": local_check["extension"], "local_precheck_limited": bool(effective_local_check["local_precheck_limited"]), "auto_normalized": effective_path != path, }, } try: return self._run(profile, runner, tool_name='导入校验') except RuntimeError as exc: return self._runtime_error_as_result( exc, error_code="IMPORT_VERIFICATION_FAILED", extra={"can_import": _runtime_import_can_import_value(exc)}, ) def record_import_repair_local( self, *, profile: str, verification_id: str, authorized_file_modification: bool, output_path: str | None = None, selected_repairs: list[str] | None = None, ) -> dict[str, Any]: """执行记录相关逻辑。""" if not verification_id.strip(): return self._failed_repair_result(error_code="IMPORT_VERIFICATION_FAILED", message="verification_id is required") if not authorized_file_modification: return self._failed_repair_result( error_code="IMPORT_REPAIR_NOT_AUTHORIZED", message="record_import_repair_local requires authorized_file_modification=true", ) unknown_repairs = sorted({item for item in (selected_repairs or []) if item not in SAFE_REPAIRS}) if unknown_repairs: return self._failed_repair_result( error_code="IMPORT_REPAIR_FORMAT_UNSUPPORTED", message=f"unknown selected_repairs: {', '.join(unknown_repairs)}", ) def runner(_session_profile, context): stored = self._verification_store.get(verification_id) if stored is None: return self._failed_repair_result(error_code="IMPORT_VERIFICATION_STALE", message="verification_id is missing or expired") source_path = Path(str(stored.get("source_file_path") or stored["file_path"])) extension = source_path.suffix.lower() if extension not in REPAIRABLE_IMPORT_EXTENSIONS: return self._failed_repair_result( error_code="IMPORT_REPAIR_FORMAT_UNSUPPORTED", message="record_import_repair_local v1 only supports .xlsx files", extra={"source_file_path": str(source_path)}, ) expected_columns, _ = self._expected_import_columns(profile, context, str(stored["app_key"])) normalized_repairs = set(selected_repairs or SAFE_REPAIRS) destination = _resolve_repaired_output_path(source_path, output_path=output_path) destination.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source_path, destination) workbook = load_workbook(destination) sheet = workbook[workbook.sheetnames[0]] applied_repairs: list[str] = [] skipped_repairs: list[str] = [] if "normalize_headers" in normalized_repairs: repair_header_columns = _repair_header_columns_from_stored_precheck(stored, expected_columns) if _repair_headers(sheet, repair_header_columns): applied_repairs.append("normalize_headers") else: skipped_repairs.append("normalize_headers") if "trim_trailing_blank_rows" in normalized_repairs: if _trim_trailing_blank_rows(sheet): applied_repairs.append("trim_trailing_blank_rows") else: skipped_repairs.append("trim_trailing_blank_rows") if "normalize_enum_values" in normalized_repairs: if _normalize_enum_values(sheet, expected_columns): applied_repairs.append("normalize_enum_values") else: skipped_repairs.append("normalize_enum_values") if "normalize_date_formats" in normalized_repairs: if _normalize_date_formats(sheet): applied_repairs.append("normalize_date_formats") else: skipped_repairs.append("normalize_date_formats") if "normalize_number_formats" in normalized_repairs: if _normalize_number_formats(sheet): applied_repairs.append("normalize_number_formats") else: skipped_repairs.append("normalize_number_formats") if "normalize_url_cells" in normalized_repairs: if _normalize_url_cells(sheet): applied_repairs.append("normalize_url_cells") else: skipped_repairs.append("normalize_url_cells") workbook.save(destination) verification_result = self.record_import_verify( profile=profile, app_key=str(stored["app_key"]), file_path=str(destination), ) new_verification_id = verification_result.get("verification_id") return { "ok": bool(verification_result.get("ok")), "status": verification_result.get("status"), "error_code": verification_result.get("error_code"), "source_file_path": str(source_path), "repaired_file_path": str(destination), "applied_repairs": applied_repairs, "skipped_repairs": skipped_repairs, "new_verification_id": new_verification_id, "can_import_after_repair": bool(verification_result.get("can_import")), "post_repair_issues": verification_result.get("issues", []), "warnings": verification_result.get("warnings", []), "verification": { "source_preserved": True, "repair_authorized": True, "reverified": True, "selected_repairs": sorted(normalized_repairs), }, } try: return self._run(profile, runner, tool_name='导入修复') except RuntimeError as exc: return self._runtime_error_as_result(exc, error_code="IMPORT_REPAIR_FORMAT_UNSUPPORTED") def record_import_start( self, *, profile: str, app_key: str, verification_id: str, being_enter_auditing: bool | None, view_key: str | None = None, ) -> dict[str, Any]: """执行记录相关逻辑。""" if being_enter_auditing is None: return self._failed_start_result(error_code="IMPORT_VERIFICATION_FAILED", message="being_enter_auditing must be passed explicitly") def runner(session_profile, context): stored = self._verification_store.get(verification_id) if stored is None: return self._failed_start_result(error_code="IMPORT_VERIFICATION_STALE", message="verification_id is missing or expired") if str(stored.get("app_key")) != app_key: return self._failed_start_result(error_code="IMPORT_VERIFICATION_STALE", message="verification_id does not belong to the requested app") if not bool(stored.get("can_import")): return self._failed_start_result(error_code="IMPORT_VERIFICATION_FAILED", message="verification_id is not importable", extra={"accepted": False}) source_local_precheck = stored.get("source_local_precheck") source_precheck_passed = isinstance(source_local_precheck, dict) and bool(source_local_precheck.get("can_import")) if source_precheck_passed: current_path = Path(str(stored.get("source_file_path") or stored["file_path"])) expected_sha256 = stored.get("file_sha256") else: current_path = Path(str(stored.get("verified_file_path") or stored.get("source_file_path") or stored["file_path"])) expected_sha256 = stored.get("verified_file_sha256") or stored.get("file_sha256") if not current_path.is_file(): return self._failed_start_result(error_code="IMPORT_VERIFICATION_STALE", message="verified import file no longer exists") current_sha256 = _sha256_file(current_path) if current_sha256 != expected_sha256: return self._failed_start_result( error_code="IMPORT_FILE_CHANGED_AFTER_VERIFY", message="the file changed after verification; run record_import_verify again", extra={"accepted": False}, ) stored_import_capability = stored.get("import_capability") _, current_schema_fingerprint = self._expected_import_columns( profile, context, app_key, import_capability=stored_import_capability if isinstance(stored_import_capability, dict) else None, ) if current_schema_fingerprint != stored.get("schema_fingerprint"): return self._failed_start_result( error_code="IMPORT_SCHEMA_CHANGED_AFTER_VERIFY", message="the applicant schema changed after verification; run record_import_verify again", extra={"accepted": False}, ) upload_result = self._file_tools.file_upload_local( profile=profile, upload_kind="login", file_path=str(current_path), ) file_url = upload_result.get("download_url") if not isinstance(file_url, str) or not file_url.strip(): return self._failed_start_result(error_code="IMPORT_VERIFICATION_FAILED", message="file upload did not return download_url") try: socket_result = self.backend.start_socket_data_import( context, app_key=app_key, being_enter_auditing=bool(being_enter_auditing), view_key=view_key, excel_url=file_url, excel_name=str(stored.get("file_name") or current_path.name), ) except QingflowApiError as exc: error_code = ( "IMPORT_SOCKET_ACK_TIMEOUT" if exc.details and exc.details.get("error_code") == "IMPORT_SOCKET_ACK_TIMEOUT" else _import_permission_error_code(exc, permission_code="IMPORT_START_UNAUTHORIZED", default="IMPORT_START_FAILED") ) details: JSONObject = {"accepted": False, "file_url": file_url} _copy_api_error_fields(details, exc) return self._failed_start_result(error_code=error_code, message=exc.message, extra=details) import_id = str(socket_result.get("import_id") or "") process_id_str = _normalize_optional_text(socket_result.get("process_id_str")) started_at = _utc_now().isoformat() self._job_store.put( import_id, { "created_at": started_at, "profile": profile, "app_key": app_key, "import_id": import_id, "process_id_str": process_id_str, "source_file_name": str(stored.get("file_name") or current_path.name), "started_at": started_at, "file_url": file_url, "verification_id": verification_id, }, ) warnings = deepcopy(socket_result.get("warnings", [])) return { "ok": True, "status": "accepted", "accepted": True, "write_executed": True, "safe_to_retry": False, "import_id": import_id, "process_id_str": process_id_str, "source_file_name": str(stored.get("file_name") or current_path.name), "file_url": file_url, "warnings": warnings, "verification": { "verification_id_valid": True, "file_hash_verified": True, "schema_fingerprint_verified": True, "upload_staged": True, "import_acknowledged": bool(import_id), }, } try: return self._run(profile, runner, tool_name='开始导入') except RuntimeError as exc: return self._runtime_error_as_result(exc, error_code="IMPORT_VERIFICATION_FAILED", extra={"accepted": False}) def record_import_status_get( self, *, profile: str, app_key: str = "", import_id: str | None = None, process_id_str: str | None = None, ) -> dict[str, Any]: """执行记录相关逻辑。""" normalized_app_key = (app_key or "").strip() normalized_import_id = _normalize_optional_text(import_id) normalized_process_id = _normalize_optional_text(process_id_str) if normalized_import_id and normalized_process_id: return self._failed_status_result( error_code="CONFIG_ERROR", message="record_import_status_get accepts import_id or process_id_str, but not both at the same time", extra={ "import_id": normalized_import_id, "process_id_str": normalized_process_id, "details": { "fix_hint": "Use only one of `import_id` or `process_id_str`. You may pass `app_key` as an optional routing hint for direct method compatibility.", } }, ) if not normalized_process_id and not normalized_import_id and not normalized_app_key: return self._failed_status_result( error_code="CONFIG_ERROR", message="record_import_status_get requires at least one selector: process_id_str, import_id, or app_key", extra={ "import_id": normalized_import_id, "process_id_str": normalized_process_id, "details": { "fix_hint": "Use `process_id_str` or `import_id` for a known import, or use only `app_key` to inspect the latest import in that app.", } }, ) def runner(_session_profile, context): local_job = None if normalized_import_id: local_job = self._job_store.get(normalized_import_id) if local_job is None and normalized_process_id: matches = [item for item in self._job_store.list() if _normalize_optional_text(item.get("process_id_str")) == normalized_process_id] local_job = matches[0] if len(matches) == 1 else None effective_process_id = normalized_process_id if effective_process_id is None and isinstance(local_job, dict): effective_process_id = _normalize_optional_text(local_job.get("process_id_str")) resolved_app_key = normalized_app_key if not resolved_app_key and isinstance(local_job, dict): resolved_app_key = str(local_job.get("app_key") or "").strip() if not resolved_app_key: return self._failed_status_result( error_code="CONFIG_ERROR", message="record_import_status_get could not determine app_key from the provided selector", extra={ "import_id": normalized_import_id, "process_id_str": effective_process_id, "details": { "fix_hint": "Use the original `app_key`, or call import status with the latest-import mode: only `app_key`.", } }, ) if local_job is None and not normalized_import_id and not normalized_process_id: recent = [item for item in self._job_store.list() if str(item.get("app_key")) == resolved_app_key] local_job = recent[0] if recent else None try: page = self.backend.request( "GET", context, "/app/apply/dataImport/record", params={"appKey": resolved_app_key, "pageNum": 1, "pageSize": 100}, ) except QingflowApiError as exc: error_code = _import_permission_error_code(exc, permission_code="IMPORT_STATUS_UNAUTHORIZED", default="IMPORT_STATUS_UNAVAILABLE") details: JSONObject = { "app_key": resolved_app_key, "import_id": normalized_import_id, "process_id_str": effective_process_id, "verification": { "status_lookup_completed": False, "process_id_verified": bool(effective_process_id), }, } _copy_api_error_fields(details, exc) if _is_import_permission_error(exc): return { "ok": True, "status": "unknown", "error_code": error_code, "app_key": resolved_app_key, "import_id": normalized_import_id or (local_job.get("import_id") if isinstance(local_job, dict) else None), "process_id_str": effective_process_id, "matched_by": "local_job" if isinstance(local_job, dict) else None, "source_file_name": local_job.get("source_file_name") if isinstance(local_job, dict) else None, "total_rows": None, "success_rows": None, "failed_rows": None, "progress": None, "error_file_urls": [], "operate_time": None, "operate_user": None, "accepted": bool(local_job or effective_process_id), "warnings": [ { "code": error_code, "message": "import history is not readable; final import status is unknown", **{ key: details.get(key) for key in ("category", "backend_code", "http_status", "request_id") if details.get(key) is not None }, } ], "verification": details["verification"], "details": {"status_readback_error": details}, "backend_code": details.get("backend_code"), "request_id": details.get("request_id"), "http_status": details.get("http_status"), "message": "import history is not readable; final import status is unknown", } return self._failed_status_result( error_code=error_code, message=exc.message, extra=details, ) records = _extract_import_records(page) matched_record, matched_by = _match_import_record( records, local_job=local_job, import_id=normalized_import_id, process_id_str=effective_process_id, ) if matched_record is None: return self._failed_status_result( error_code="IMPORT_STATUS_AMBIGUOUS", message="could not uniquely resolve an import record from the provided identifiers", extra={ "import_id": normalized_import_id, "process_id_str": effective_process_id, "matched_by": matched_by, }, ) normalized_process = _normalize_optional_text( matched_record.get("processIdStr") or matched_record.get("processId") or matched_record.get("process_id_str") ) if local_job is not None and normalized_import_id: self._job_store.put( normalized_import_id, { **local_job, "created_at": local_job.get("created_at") or _utc_now().isoformat(), "process_id_str": normalized_process, }, ) raw_process_status = matched_record.get("processStatus") total_rows = _coerce_int(matched_record.get("totalNumber") or matched_record.get("total_rows")) success_rows = _coerce_int(matched_record.get("successNum") or matched_record.get("success_rows")) failed_rows = _coerce_int(matched_record.get("errorNum") or matched_record.get("failed_rows")) progress = _coerce_int(matched_record.get("importPercentage") or matched_record.get("progress")) normalized_status = _normalize_import_status(raw_process_status) warnings: list[dict[str, str]] = [] if normalized_status in {"succeeded", "failed", "partially_failed"} and all( value is None for value in (total_rows, success_rows, failed_rows) ): warnings.append( { "code": "IMPORT_STATUS_COUNTERS_MISSING", "message": "backend import history returned a terminal process status without row counters", } ) return { "ok": True, "status": normalized_status, "process_status": _coerce_int(raw_process_status), "app_key": resolved_app_key, "import_id": normalized_import_id or (local_job.get("import_id") if isinstance(local_job, dict) else None), "process_id_str": normalized_process, "matched_by": matched_by, "source_file_name": matched_record.get("sourceFileName") or matched_record.get("source_file_name"), "total_rows": total_rows, "success_rows": success_rows, "failed_rows": failed_rows, "progress": progress, "error_file_urls": _normalize_error_file_urls(matched_record.get("errorFileUrls")), "operate_time": matched_record.get("operateTime"), "operate_user": matched_record.get("operateUser"), "warnings": warnings, "verification": { "status_lookup_completed": True, "matched_by": matched_by, "process_id_verified": bool(normalized_process), }, } try: return self._run(profile, runner, tool_name='导入状态') except RuntimeError as exc: return self._runtime_error_as_result( exc, error_code="IMPORT_STATUS_AMBIGUOUS", extra={ "app_key": normalized_app_key, "import_id": normalized_import_id, "process_id_str": normalized_process_id, "verification": { "status_lookup_completed": False, "process_id_verified": bool(normalized_process_id), }, }, ) def _resolve_import_schema_bundle( self, profile: str, context, app_key: str, *, import_capability: JSONObject | None = None, ) -> tuple[Any, list[JSONObject], str]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" auth_source = _normalize_optional_text((import_capability or {}).get("auth_source")) or "unknown" if auth_source == "data_manage_auth": schema = self.backend.request("GET", context, f"/app/{app_key}/form", params={"type": 1}) index = _build_field_index(_normalize_form_schema(schema)) else: index = self._record_tools._get_field_index(profile, context, app_key, force_refresh=False) ws_id = self.sessions.get_profile(profile).selected_ws_id expected_columns: list[JSONObject] = [] for field in index.by_id.values(): payload = self._record_tools._schema_field_payload( profile, context, field, workflow_node_id=None, ws_id=ws_id, schema_mode="applicant", ) if not bool(payload.get("writable")): continue expected_columns.append( { "field_id": payload["field_id"], "title": payload["title"], "que_type": payload["que_type"], "required": bool(field.required), "write_kind": payload["write_kind"], "options": payload.get("options", []), "requires_lookup": bool(payload.get("requires_lookup")), "requires_upload": bool(payload.get("requires_upload")), "target_app_key": payload.get("target_app_key"), "target_app_name": payload.get("target_app_name"), "searchable_fields": payload.get("searchable_fields", []), } ) expected_columns.sort(key=lambda item: int(item["field_id"])) schema_fingerprint = _stable_import_schema_fingerprint(expected_columns) return index, expected_columns, schema_fingerprint def _expected_import_columns( self, profile: str, context, app_key: str, *, import_capability: JSONObject | None = None, ) -> tuple[list[JSONObject], str]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" _, expected_columns, schema_fingerprint = self._resolve_import_schema_bundle( profile, context, app_key, import_capability=import_capability, ) return expected_columns, schema_fingerprint def _local_verify( self, *, profile: str, context, path: Path, app_key: str, field_index: Any, expected_columns: list[JSONObject], allowed_header_titles: list[str] | None, schema_fingerprint: str, ) -> dict[str, Any]: """执行内部辅助逻辑。""" extension = path.suffix.lower() file_sha256 = _sha256_file(path) base_result = { "app_key": app_key, "file_path": str(path.resolve()), "file_size": path.stat().st_size, "file_sha256": file_sha256, "schema_fingerprint": schema_fingerprint, "issues": [], "warnings": [], "repair_suggestions": [], "local_precheck_passed": True, "local_precheck_limited": False, "can_import": True, "extension": extension, "error_code": None, "expected_header_titles": list(allowed_header_titles) if allowed_header_titles else [str(item["title"]) for item in expected_columns], } if extension not in SUPPORTED_IMPORT_EXTENSIONS: base_result["issues"].append(_issue("UNSUPPORTED_FILE_FORMAT", "Only .xlsx and .xls files are supported in import v1.", severity="error")) base_result["local_precheck_passed"] = False base_result["can_import"] = False base_result["error_code"] = "IMPORT_FILE_FORMAT_UNSUPPORTED" return base_result if extension == ".xls": base_result["warnings"].append( { "code": "IMPORT_LOCAL_PRECHECK_LIMITED", "message": ".xls files are allowed for verify/start, but v1 local precheck is limited and repair is unsupported.", } ) base_result["local_precheck_limited"] = True return base_result try: workbook = load_workbook(path, read_only=True, data_only=False) except Exception as exc: base_result["issues"].append(_issue("WORKBOOK_OPEN_FAILED", f"Workbook could not be opened: {exc}", severity="error")) base_result["local_precheck_passed"] = False base_result["can_import"] = False base_result["error_code"] = "IMPORT_VERIFICATION_FAILED" return base_result if not workbook.sheetnames: base_result["issues"].append(_issue("SHEET_MISSING", "Workbook does not contain any sheets.", severity="error")) base_result["local_precheck_passed"] = False base_result["can_import"] = False base_result["error_code"] = "IMPORT_VERIFICATION_FAILED" return base_result try: sheet = workbook[workbook.sheetnames[0]] header_row = [cell.value for cell in next(sheet.iter_rows(min_row=1, max_row=1), [])] header_analysis = _analyze_headers( header_row, expected_columns, allowed_titles=allowed_header_titles, ) base_result["issues"].extend(header_analysis["issues"]) base_result["repair_suggestions"].extend(header_analysis["repair_suggestions"]) if not any(issue.get("severity") == "error" for issue in base_result["issues"]): semantic_issues, semantic_warnings = self._inspect_semantic_cells( profile=profile, context=context, sheet=sheet, expected_columns=expected_columns, field_index=field_index, ) base_result["issues"].extend(semantic_issues) base_result["warnings"].extend(semantic_warnings) trailing_blank_rows = _count_trailing_blank_rows(sheet) if trailing_blank_rows > 0: base_result["warnings"].append( { "code": "TRAILING_BLANK_ROWS", "message": f"Workbook contains {trailing_blank_rows} trailing blank rows that can be safely removed.", } ) base_result["repair_suggestions"].append("trim_trailing_blank_rows") enum_suggestions = _find_enum_repairs(sheet, expected_columns) if enum_suggestions: base_result["warnings"].append( { "code": "ENUM_VALUE_NORMALIZATION_AVAILABLE", "message": "Some enum-like cells can be normalized to exact template values without changing meaning.", } ) base_result["repair_suggestions"].append("normalize_enum_values") base_result["repair_suggestions"] = sorted(set(base_result["repair_suggestions"])) except Exception as exc: base_result["issues"].append( _issue( "IMPORT_LOCAL_PRECHECK_FAILED", f"Workbook content could not be fully inspected during local precheck: {exc}", severity="error", ) ) base_result["warnings"].append( { "code": "IMPORT_LOCAL_PRECHECK_FAILED", "message": "Workbook local precheck encountered an unexpected compatibility problem; returning a structured verification failure instead of crashing.", } ) if any(issue.get("severity") == "error" for issue in base_result["issues"]): base_result["local_precheck_passed"] = False base_result["can_import"] = False base_result["error_code"] = "IMPORT_VERIFICATION_FAILED" return base_result def _inspect_semantic_cells( self, *, profile: str, context, sheet, expected_columns: list[JSONObject], field_index: Any, ) -> tuple[list[JSONObject], list[JSONObject]]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" issues: list[JSONObject] = [] warnings: list[JSONObject] = [] header_positions = _sheet_header_positions(sheet) expected_by_key: dict[str, list[JSONObject]] = {} for column in expected_columns: key = _normalize_header_key(column.get("title")) if key: expected_by_key.setdefault(key, []).append(column) for key, columns in expected_by_key.items(): positions = header_positions.get(key, []) if len(columns) != 1 or len(positions) != 1: continue column = columns[0] column_index = positions[0] write_kind = _normalize_optional_text(column.get("write_kind")) or "scalar" if column.get("options"): issue = _inspect_enum_column(sheet, column_index=column_index, column=column) if issue is not None: issues.append(issue) continue if write_kind == "relation": issue = _inspect_relation_column(sheet, column_index=column_index, column=column) if issue is not None: issues.append(issue) continue field = field_index.by_id.get(str(column.get("field_id"))) if field_index is not None else None if ( write_kind == "member" and field is not None and ( field.member_select_scope_type is not None or field.member_select_scope is not None ) ): member_issue, member_warning = self._inspect_member_column( context=context, sheet=sheet, column_index=column_index, column=column, field=field, ) if member_issue is not None: issues.append(member_issue) continue if member_warning is not None: warnings.append(member_warning) continue if ( write_kind == "department" and field is not None and ( field.dept_select_scope_type is not None or field.dept_select_scope is not None ) ): department_issue, department_warning = self._inspect_department_column( context=context, sheet=sheet, column_index=column_index, column=column, field=field, ) if department_issue is not None: issues.append(department_issue) continue if department_warning is not None: warnings.append(department_warning) continue return issues, warnings def _inspect_member_column( self, *, context, sheet, column_index: int, column: JSONObject, field, ) -> tuple[JSONObject | None, JSONObject | None]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" invalid_email_samples: list[str] = [] scope_miss_samples: list[str] = [] checked_values: set[str] = set() for row_index in range(2, sheet.max_row + 1): text = _normalize_optional_text(sheet.cell(row=row_index, column=column_index).value) if text is None: continue normalized = text.strip() if normalized in checked_values: continue checked_values.add(normalized) if not EMAIL_PATTERN.fullmatch(normalized): invalid_email_samples.append(f"row {row_index}: {normalized}") if len(invalid_email_samples) >= 3: break continue try: candidates = self._record_tools._resolve_member_candidates(context, field, keyword=normalized) matches = self._record_tools._match_member_candidates(candidates, normalized) except QingflowApiError as exc: if exc.category == "not_supported": return None, { "code": "MEMBER_CANDIDATE_VALIDATION_SKIPPED", "message": f"Member candidate scope for column '{column['title']}' could not be resolved safely during local precheck.", } raise except RuntimeError as exc: if not _runtime_candidate_validation_skippable(exc): raise return None, { "code": "MEMBER_CANDIDATE_VALIDATION_SKIPPED", "message": f"Member candidate scope for column '{column['title']}' could not be resolved safely during local precheck.", } if len(matches) != 1: scope_miss_samples.append(f"row {row_index}: {normalized}") if len(scope_miss_samples) >= 3: break if invalid_email_samples: return _issue( "MEMBER_IMPORT_REQUIRES_EMAIL", f"Column '{column['title']}' must use member email values in import files. Samples: {', '.join(invalid_email_samples)}", severity="error", ), None if scope_miss_samples: return _issue( "MEMBER_NOT_IN_CANDIDATE_SCOPE", f"Column '{column['title']}' contains members outside the current candidate scope. Samples: {', '.join(scope_miss_samples)}", severity="error", ), None return None, None def _inspect_department_column( self, *, context, sheet, column_index: int, column: JSONObject, field, ) -> tuple[JSONObject | None, JSONObject | None]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" scope_miss_samples: list[str] = [] checked_values: set[str] = set() for row_index in range(2, sheet.max_row + 1): value = sheet.cell(row=row_index, column=column_index).value text = _normalize_optional_text(value) if text is None: continue normalized = text.strip() if normalized in checked_values: continue checked_values.add(normalized) try: candidates = self._record_tools._resolve_department_candidates(context, field, keyword=normalized) matches = self._record_tools._match_department_candidates(candidates, normalized) except QingflowApiError as exc: if exc.category == "not_supported": return None, { "code": "DEPARTMENT_CANDIDATE_VALIDATION_SKIPPED", "message": f"Department candidate scope for column '{column['title']}' could not be resolved safely during local precheck.", } raise except RuntimeError as exc: if not _runtime_candidate_validation_skippable(exc): raise return None, { "code": "DEPARTMENT_CANDIDATE_VALIDATION_SKIPPED", "message": f"Department candidate scope for column '{column['title']}' could not be resolved safely during local precheck.", } if len(matches) != 1: scope_miss_samples.append(f"row {row_index}: {normalized}") if len(scope_miss_samples) >= 3: break if scope_miss_samples: return _issue( "DEPARTMENT_NOT_IN_CANDIDATE_SCOPE", f"Column '{column['title']}' contains departments outside the current candidate scope. Samples: {', '.join(scope_miss_samples)}", severity="error", ), None return None, None def _load_template_header_profile( self, context, app_key: str, *, import_capability: JSONObject | None = None, expected_columns: list[JSONObject] | None = None, ) -> tuple[dict[str, Any], list[JSONObject]]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" warnings: list[JSONObject] = [] try: payload = self.backend.request("GET", context, f"/app/{app_key}/apply/excelTemplate") template_url = _pick_template_url(payload) if not template_url: return {"allowed_titles": None, "leaf_titles": None, "header_depth": 1}, warnings content = self.backend.download_binary(template_url) workbook = load_workbook(BytesIO(content), read_only=False, data_only=False) if not workbook.sheetnames: return {"allowed_titles": None, "leaf_titles": None, "header_depth": 1}, warnings sheet = workbook[workbook.sheetnames[0]] header_row = [cell.value for cell in next(sheet.iter_rows(min_row=1, max_row=1), [])] titles = [_normalize_optional_text(value) for value in header_row] normalized_titles = [title for title in titles if title] header_depth = _infer_header_depth(sheet) leaf_titles = [title for title in _extract_leaf_header_titles(sheet, header_depth) if title] return { "allowed_titles": normalized_titles or None, "leaf_titles": leaf_titles or None, "header_depth": header_depth, }, warnings except Exception: if ( _normalize_optional_text((import_capability or {}).get("auth_source")) == "apply_auth" and expected_columns ): warnings.append( { "code": "IMPORT_TEMPLATE_HEADER_LOCAL_FALLBACK", "message": "Official template headers require data management permission; local precheck fell back to applicant import columns.", } ) fallback_titles = [str(item["title"]) for item in expected_columns] return {"allowed_titles": fallback_titles, "leaf_titles": fallback_titles, "header_depth": 1}, warnings warnings.append( { "code": "IMPORT_TEMPLATE_HEADER_UNAVAILABLE", "message": "Official template headers could not be loaded during local precheck; falling back to applicant writable columns only.", } ) return {"allowed_titles": None, "leaf_titles": None, "header_depth": 1}, warnings def _maybe_auto_normalize_file( self, *, source_path: Path, expected_columns: list[JSONObject], template_header_profile: dict[str, Any], local_check: dict[str, Any], ) -> dict[str, Any] | None: """执行内部辅助逻辑。""" if source_path.suffix.lower() != ".xlsx": return None try: workbook = load_workbook(source_path, read_only=False, data_only=False) if not workbook.sheetnames: return None sheet = workbook[workbook.sheetnames[0]] rows = [list(row) for row in sheet.iter_rows(values_only=True)] header_depth = _infer_header_depth(sheet) return _build_auto_normalized_file( source_path=source_path, sheet_title=sheet.title, rows=rows, header_depth=header_depth, template_leaf_titles=template_header_profile.get("leaf_titles"), local_check=local_check, ) except Exception as exc: workbook = load_workbook(source_path, read_only=True, data_only=False) if not workbook.sheetnames: return None sheet = workbook[workbook.sheetnames[0]] rows = [list(row) for row in sheet.iter_rows(values_only=True)] header_depth = _infer_header_depth_from_rows( rows, template_header_profile=template_header_profile, local_check=local_check, ) normalized = _build_auto_normalized_file( source_path=source_path, sheet_title=sheet.title, rows=rows, header_depth=header_depth, template_leaf_titles=template_header_profile.get("leaf_titles"), local_check=local_check, ) if normalized is not None: normalized["warnings"].insert( 0, { "code": "IMPORT_AUTO_NORMALIZATION_COMPATIBILITY_FALLBACK", "message": f"Workbook compatibility normalization retried in compatibility mode after a workbook parsing error: {exc}", }, ) return normalized def _fetch_import_capability(self, context, app_key: str) -> tuple[JSONObject, list[JSONObject]]: # type: ignore[no-untyped-def] """执行内部辅助逻辑。""" base_info_error: QingflowApiError | None = None try: payload = self.backend.request("GET", context, f"/app/{app_key}/baseInfo") except QingflowApiError as exc: if not _is_optional_import_capability_error(exc): raise base_info_error = exc payload = None capability, warnings = _derive_import_capability(payload) if base_info_error is not None: for warning in warnings: if warning.get("code") != "IMPORT_CAPABILITY_UNAVAILABLE": continue warning["message"] = ( "import capability precheck could not read app baseInfo; continuing with applicant schema " "when available, but do not treat import permission as verified." ) if base_info_error.backend_code is not None: warning["backend_code"] = base_info_error.backend_code if base_info_error.http_status is not None: warning["http_status"] = base_info_error.http_status if base_info_error.request_id is not None: warning["request_id"] = base_info_error.request_id break return capability, warnings def _write_local_template( self, *, expected_columns: list[JSONObject], destination_hint: str | None, app_key: str, ) -> str: """执行内部辅助逻辑。""" if destination_hint: destination = _resolve_template_download_path(destination_hint, app_key=app_key) else: destination = Path(tempfile.gettempdir()) / f"qingflow-import-template-{app_key}-{uuid4().hex[:8]}.xlsx" destination.parent.mkdir(parents=True, exist_ok=True) workbook = Workbook() sheet = workbook.active sheet.title = "导入模板" sheet.append([str(item["title"]) for item in expected_columns]) workbook.save(destination) return str(destination) def _failed_template_result( self, *, app_key: str, error_code: str, message: str, request_route: JSONObject | None = None, extra: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" payload = { "ok": False, "status": "failed", "error_code": error_code, "app_key": app_key, "template_url": None, "downloaded_to_path": None, "expected_columns": [], "schema_fingerprint": None, "request_route": request_route, "warnings": [], "verification": {"template_url_resolved": False}, "message": message, } if extra: payload.update(extra) return payload def _failed_verify_result( self, *, app_key: str, file_path: str, error_code: str, message: str, extra: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" payload = { "ok": False, "status": "failed", "error_code": error_code, "app_key": app_key, "can_import": False, "verification_id": None, "file_path": str(Path(file_path).expanduser()) if file_path else file_path, "verified_file_path": None, "file_name": Path(file_path).name if file_path else None, "file_sha256": None, "verified_file_sha256": None, "file_size": None, "schema_fingerprint": None, "apply_rows": None, "issues": [_issue(error_code, message, severity="error")], "repair_suggestions": [], "warnings": [], "verification": { "import_auth_prechecked": False, "import_auth_precheck_passed": False, "local_precheck_passed": False, "backend_verification_passed": False, }, "import_capability": None, "message": message, } if extra: payload.update(extra) return payload def _failed_repair_result(self, *, error_code: str, message: str, extra: dict[str, Any] | None = None) -> dict[str, Any]: """执行内部辅助逻辑。""" payload = { "ok": False, "status": "failed", "error_code": error_code, "source_file_path": None, "repaired_file_path": None, "applied_repairs": [], "skipped_repairs": [], "new_verification_id": None, "can_import_after_repair": False, "post_repair_issues": [_issue(error_code, message, severity="error")], "warnings": [], "verification": { "repair_authorized": False, "reverified": False, }, "message": message, } if extra: payload.update(extra) return payload def _failed_start_result(self, *, error_code: str, message: str, extra: dict[str, Any] | None = None) -> dict[str, Any]: """执行内部辅助逻辑。""" payload = { "ok": False, "status": "failed", "error_code": error_code, "accepted": False, "import_id": None, "process_id_str": None, "source_file_name": None, "file_url": None, "warnings": [], "verification": { "verification_id_valid": False, "file_hash_verified": False, "schema_fingerprint_verified": False, "upload_staged": False, "import_acknowledged": False, }, "message": message, } if extra: payload.update(extra) return payload def _failed_status_result(self, *, error_code: str, message: str, extra: dict[str, Any] | None = None) -> dict[str, Any]: """执行内部辅助逻辑。""" payload = { "ok": False, "status": "failed", "error_code": error_code, "import_id": None, "process_id_str": None, "matched_by": None, "source_file_name": None, "total_rows": None, "success_rows": None, "failed_rows": None, "progress": None, "error_file_urls": [], "operate_time": None, "operate_user": None, "warnings": [], "verification": { "status_lookup_completed": False, "process_id_verified": False, }, "message": message, } if extra: payload.update(extra) return payload def _runtime_error_as_result( self, error: RuntimeError, *, error_code: str, extra: dict[str, Any] | None = None, ) -> dict[str, Any]: """执行内部辅助逻辑。""" payload = _runtime_error_payload(error) details = payload.get("details") if isinstance(payload.get("details"), dict) else {} response = { "ok": False, "status": "failed", "error_code": details.get("error_code") or _runtime_error_code(payload, default=error_code), "warnings": [], "verification": {}, "message": payload.get("message") or str(error), } for key in ("category", "backend_code", "request_id", "http_status"): if key in payload: response[key] = payload.get(key) if details: response["details"] = details if extra: response.update(extra) return response def _pick_template_url(payload: Any) -> str | None: if isinstance(payload, dict): for key in ("excelUrl", "url", "downloadUrl"): value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return None def _resolve_template_download_path(raw_path: str, *, app_key: str) -> Path: path = Path(raw_path).expanduser() if path.exists() and path.is_dir(): return path / f"{app_key}_import_template.xlsx" if path.suffix: return path return path / f"{app_key}_import_template.xlsx" def _resolve_repaired_output_path(source_path: Path, *, output_path: str | None) -> Path: if output_path: path = Path(output_path).expanduser() if path.exists() and path.is_dir(): return path / f"{source_path.stem}.repaired{source_path.suffix}" if path.suffix: return path return path / f"{source_path.stem}.repaired{source_path.suffix}" return source_path.with_name(f"{source_path.stem}.repaired{source_path.suffix}") def _resolve_verified_output_path(source_path: Path) -> Path: return Path(tempfile.gettempdir()) / f"qingflow-import-verified-{source_path.stem}-{uuid4().hex[:8]}{source_path.suffix}" def _utc_now() -> datetime: return datetime.now(timezone.utc) def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _normalize_optional_text(value: Any) -> str | None: if value is None: return None normalized = str(value).strip() return normalized or None def _normalize_header_key(value: Any) -> str: text = _normalize_optional_text(value) return (text or "").casefold() def _issue(code: str, message: str, *, severity: str, repairable: bool = False, repair_code: str | None = None) -> JSONObject: payload: JSONObject = { "code": code, "message": message, "severity": severity, "repairable": repairable, } if repair_code: payload["repair_code"] = repair_code return payload def _analyze_headers( header_row: list[Any], expected_columns: list[JSONObject], *, allowed_titles: list[str] | None = None, ) -> dict[str, Any]: expected_titles = [str(item["title"]) for item in expected_columns] allowed_title_list = allowed_titles if allowed_titles else expected_titles allowed_counts = _header_title_counts(allowed_title_list) allowed_by_key = { key: title for key, title in ( (_normalize_header_key(title), _normalize_optional_text(title)) for title in allowed_title_list ) if key and title } seen: dict[str, int] = {} actual_headers: list[str] = [] for item in header_row: text = _normalize_optional_text(item) if text is None: actual_headers.append("") continue actual_headers.append(text) key = _normalize_header_key(text) seen[key] = seen.get(key, 0) + 1 missing: list[str] = [] for key, expected_count in allowed_counts.items(): actual_count = seen.get(key, 0) if actual_count >= expected_count: continue title = allowed_by_key.get(key) or key if expected_count <= 1: missing.append(title) else: missing.append(f"{title} (need {expected_count}, got {actual_count})") extra = [text for text in actual_headers if text and _normalize_header_key(text) not in allowed_by_key] duplicates = [] for key, count in seen.items(): if not key: continue allowed_count = allowed_counts.get(key, 0) if count > max(allowed_count, 1 if allowed_count == 0 else allowed_count): duplicates.append(allowed_by_key.get(key) or key) issues: list[JSONObject] = [] repair_suggestions: list[str] = [] if missing: issues.append( _issue( "MISSING_COLUMNS", f"Missing expected columns: {', '.join(missing)}", severity="error", repairable=True, repair_code="normalize_headers", ) ) if extra: issues.append( _issue( "EXTRA_COLUMNS", f"Unexpected columns: {', '.join(extra)}", severity="error", repairable=True, repair_code="normalize_headers", ) ) if duplicates: issues.append( _issue( "DUPLICATE_COLUMNS", f"Duplicate columns: {', '.join(sorted(set(duplicates)))}", severity="error", repairable=True, repair_code="normalize_headers", ) ) normalized_changes = [] for text in actual_headers: if not text: continue canonical = allowed_by_key.get(_normalize_header_key(text)) if canonical and canonical != text: normalized_changes.append((text, canonical)) if missing or extra or duplicates or normalized_changes: repair_suggestions.append("normalize_headers") return {"issues": issues, "repair_suggestions": repair_suggestions} def _header_title_counts(titles: list[str]) -> dict[str, int]: counts: dict[str, int] = {} for title in titles: key = _normalize_header_key(title) if not key: continue counts[key] = counts.get(key, 0) + 1 return counts def _sheet_header_positions(sheet) -> dict[str, list[int]]: # type: ignore[no-untyped-def] mapping: dict[str, list[int]] = {} for index, cell in enumerate(next(sheet.iter_rows(min_row=1, max_row=1), []), start=1): key = _normalize_header_key(cell.value) if not key: continue mapping.setdefault(key, []).append(index) return mapping def _inspect_enum_column(sheet, *, column_index: int, column: JSONObject) -> JSONObject | None: # type: ignore[no-untyped-def] options = [str(item).strip() for item in column.get("options", []) if str(item).strip()] if not options: return None option_map = {_normalize_header_key(item): item for item in options} invalid_samples: list[str] = [] for row_index in range(2, sheet.max_row + 1): text = _normalize_optional_text(sheet.cell(row=row_index, column=column_index).value) if text is None: continue if _normalize_header_key(text) in option_map: continue invalid_samples.append(f"row {row_index}: {text}") if len(invalid_samples) >= 3: break if not invalid_samples: return None return _issue( "INVALID_ENUM_VALUES", f"Column '{column['title']}' contains values outside the allowed options. Samples: {', '.join(invalid_samples)}", severity="error", ) def _inspect_relation_column(sheet, *, column_index: int, column: JSONObject) -> JSONObject | None: # type: ignore[no-untyped-def] invalid_samples: list[str] = [] for row_index in range(2, sheet.max_row + 1): value = sheet.cell(row=row_index, column=column_index).value text = _normalize_optional_text(value) if text is None: continue relation_id = _coerce_positive_relation_id(value) if relation_id is not None: continue invalid_samples.append(f"row {row_index}: {text}") if len(invalid_samples) >= 3: break if not invalid_samples: return None return _issue( "RELATION_IMPORT_REQUIRES_APPLY_ID", f"Column '{column['title']}' must use target record apply_id values during import. Samples: {', '.join(invalid_samples)}", severity="error", ) def _stable_import_schema_fingerprint(expected_columns: list[JSONObject]) -> str: stable_columns = [] for item in expected_columns: stable_columns.append( { "field_id": item["field_id"], "title": item["title"], "que_type": item["que_type"], "required": item["required"], "write_kind": item["write_kind"], "options": item.get("options", []), "requires_lookup": bool(item.get("requires_lookup")), "requires_upload": bool(item.get("requires_upload")), "target_app_key": item.get("target_app_key"), } ) return hashlib.sha256( json.dumps(stable_columns, ensure_ascii=False, sort_keys=True).encode("utf-8") ).hexdigest() def _coerce_positive_relation_id(value: Any) -> int | None: if isinstance(value, bool): return None if isinstance(value, int): return value if value > 0 else None if isinstance(value, float): if value.is_integer() and value > 0: return int(value) return None text = _normalize_optional_text(value) if text is None: return None if text.isdigit(): parsed = int(text) return parsed if parsed > 0 else None return None def _infer_header_depth(sheet) -> int: # type: ignore[no-untyped-def] header_depth = 1 merged_cells = getattr(sheet, "merged_cells", None) merged_ranges = getattr(merged_cells, "ranges", merged_cells) if merged_cells is not None else [] row_one_has_merge = False for merged_range in merged_ranges or []: min_row = int(getattr(merged_range, "min_row", 1)) max_row = int(getattr(merged_range, "max_row", 1)) if min_row == 1: row_one_has_merge = True header_depth = max(header_depth, max_row) if row_one_has_merge and sheet.max_row >= 2: row_two_values = [cell.value for cell in sheet[2]] if any(_normalize_optional_text(value) for value in row_two_values): header_depth = max(header_depth, 2) return min(header_depth, max(1, int(sheet.max_row))) def _extract_leaf_header_titles(sheet, header_depth: int) -> list[str]: # type: ignore[no-untyped-def] titles: list[str] = [] max_column = max(1, int(sheet.max_column)) depth = max(1, min(header_depth, int(sheet.max_row))) for column_index in range(1, max_column + 1): selected = "" for row_index in range(depth, 0, -1): text = _normalize_optional_text(sheet.cell(row=row_index, column=column_index).value) if text: selected = text break titles.append(selected) return titles def _overlay_header_titles(actual_titles: list[str], template_leaf_titles: Any) -> list[str]: normalized = list(actual_titles) if not isinstance(template_leaf_titles, list): return normalized for index, title in enumerate(template_leaf_titles): normalized_title = _normalize_optional_text(title) if normalized_title is None: continue if index < len(normalized): normalized[index] = normalized_title return normalized def _infer_header_depth_from_rows( rows: list[list[Any]], *, template_header_profile: dict[str, Any], local_check: dict[str, Any], ) -> int: template_depth = max(1, int(template_header_profile.get("header_depth") or 1)) header_depth = min(template_depth, max(1, len(rows))) if header_depth > 1: return header_depth if "normalize_headers" in (local_check.get("repair_suggestions") or []) and len(rows) >= 2: if any(_normalize_optional_text(value) for value in rows[1]): return 2 return 1 def _extract_leaf_header_titles_from_rows(rows: list[list[Any]], header_depth: int) -> list[str]: titles: list[str] = [] max_column = max((len(row) for row in rows[: max(1, header_depth)]), default=0) depth = max(1, min(header_depth, len(rows))) for column_index in range(max_column): selected = "" for row_index in range(depth - 1, -1, -1): value = rows[row_index][column_index] if column_index < len(rows[row_index]) else None text = _normalize_optional_text(value) if text: selected = text break titles.append(selected) return titles def _count_trailing_blank_rows_from_rows(rows: list[list[Any]], *, min_data_index: int = 1) -> int: count = 0 for row in reversed(rows[min_data_index:]): if any(value not in (None, "") for value in row): break count += 1 return count def _build_auto_normalized_file( *, source_path: Path, sheet_title: str, rows: list[list[Any]], header_depth: int, template_leaf_titles: Any, local_check: dict[str, Any], ) -> dict[str, Any] | None: if not rows: return None normalized_header_depth = max(1, min(header_depth, len(rows))) trailing_blank_rows = _count_trailing_blank_rows_from_rows(rows, min_data_index=normalized_header_depth) if normalized_header_depth <= 1 and trailing_blank_rows <= 0: return None extracted_headers = _extract_leaf_header_titles_from_rows(rows, normalized_header_depth) target_headers = _overlay_header_titles(extracted_headers, template_leaf_titles) row_width = max(len(target_headers), max((len(row) for row in rows), default=0)) if row_width <= 0: return None padded_headers = list(target_headers) + [""] * max(0, row_width - len(target_headers)) verified_path = _resolve_verified_output_path(source_path) normalized_workbook = Workbook() normalized_sheet = normalized_workbook.active normalized_sheet.title = sheet_title normalized_sheet.append(padded_headers) last_nonblank_row = max(normalized_header_depth, len(rows) - trailing_blank_rows) for row in rows[normalized_header_depth:last_nonblank_row]: normalized_sheet.append(list(row) + [None] * max(0, row_width - len(row))) verified_path.parent.mkdir(parents=True, exist_ok=True) normalized_workbook.save(verified_path) warnings: list[JSONObject] = [] applied_repairs: list[str] = [] if normalized_header_depth > 1: applied_repairs.append("normalize_headers") warnings.append( { "code": "IMPORT_HEADERS_AUTO_NORMALIZED", "message": f"Workbook used {normalized_header_depth} header rows; record_import_verify normalized it to a single leaf-header row automatically.", } ) if trailing_blank_rows > 0: applied_repairs.append("trim_trailing_blank_rows") warnings.append( { "code": "TRAILING_BLANK_ROWS_AUTO_TRIMMED", "message": f"Removed {trailing_blank_rows} trailing blank rows before backend verification.", } ) return { "verified_file_path": str(verified_path.resolve()), "header_titles": target_headers or padded_headers, "warnings": warnings, "applied_repairs": applied_repairs, "header_depth": normalized_header_depth, "trailing_blank_rows": trailing_blank_rows, "source_local_check": local_check, } def _count_trailing_blank_rows(sheet) -> int: # type: ignore[no-untyped-def] count = 0 for row_index in range(sheet.max_row, 1, -1): values = [cell.value for cell in sheet[row_index]] if any(value not in (None, "") for value in values): break count += 1 return count def _find_enum_repairs(sheet, expected_columns: list[JSONObject]) -> list[str]: # type: ignore[no-untyped-def] header_map = _sheet_header_map(sheet) found: list[str] = [] for column in expected_columns: options = [str(item).strip() for item in column.get("options", []) if str(item).strip()] if not options: continue column_index = header_map.get(_normalize_header_key(column["title"])) if column_index is None: continue option_map = {_normalize_header_key(item): item for item in options} for row in range(2, min(sheet.max_row, 50) + 1): value = sheet.cell(row=row, column=column_index).value text = _normalize_optional_text(value) if text is None: continue exact = option_map.get(_normalize_header_key(text)) if exact and exact != text: found.append(column["title"]) break return found def _sheet_header_map(sheet) -> dict[str, int]: # type: ignore[no-untyped-def] mapping: dict[str, int] = {} for index, cell in enumerate(next(sheet.iter_rows(min_row=1, max_row=1), []), start=1): key = _normalize_header_key(cell.value) if key and key not in mapping: mapping[key] = index return mapping def _repair_header_columns_from_stored_precheck(stored: JSONObject, expected_columns: list[JSONObject]) -> list[JSONObject]: for key in ("source_local_precheck", "local_precheck"): precheck = stored.get(key) if not isinstance(precheck, dict): continue titles = precheck.get("expected_header_titles") if not isinstance(titles, list): continue normalized_titles = [title for title in (_normalize_optional_text(item) for item in titles) if title] if normalized_titles: return [{"title": title} for title in normalized_titles] return expected_columns def _repair_headers(sheet, expected_columns: list[JSONObject]) -> bool: # type: ignore[no-untyped-def] changed = False expected_by_key = {_normalize_header_key(item["title"]): item["title"] for item in expected_columns} header_cells = list(next(sheet.iter_rows(min_row=1, max_row=1), [])) for cell in header_cells: text = _normalize_optional_text(cell.value) if text is None: continue canonical = expected_by_key.get(_normalize_header_key(text)) if canonical and canonical != text: cell.value = canonical changed = True if changed: return True # Fallback for template-based files where headers were edited into non-canonical # values but column order is still intact. Keep any extra trailing system columns. for index, column in enumerate(expected_columns, start=1): expected_title = str(column["title"]).strip() if index > len(header_cells): sheet.cell(row=1, column=index, value=expected_title) changed = True continue current_title = _normalize_optional_text(header_cells[index - 1].value) if current_title != expected_title: header_cells[index - 1].value = expected_title changed = True return changed def _trim_trailing_blank_rows(sheet) -> bool: # type: ignore[no-untyped-def] removed = 0 while sheet.max_row > 1: values = [cell.value for cell in sheet[sheet.max_row]] if any(value not in (None, "") for value in values): break sheet.delete_rows(sheet.max_row, 1) removed += 1 return removed > 0 def _normalize_enum_values(sheet, expected_columns: list[JSONObject]) -> bool: # type: ignore[no-untyped-def] changed = False header_map = _sheet_header_map(sheet) for column in expected_columns: options = [str(item).strip() for item in column.get("options", []) if str(item).strip()] if not options: continue column_index = header_map.get(_normalize_header_key(column["title"])) if column_index is None: continue option_map = {_normalize_header_key(item): item for item in options} for row in range(2, sheet.max_row + 1): cell = sheet.cell(row=row, column=column_index) text = _normalize_optional_text(cell.value) if text is None: continue canonical = option_map.get(_normalize_header_key(text)) if canonical and canonical != text: cell.value = canonical changed = True return changed def _normalize_date_formats(sheet) -> bool: # type: ignore[no-untyped-def] changed = False for row in sheet.iter_rows(min_row=2): for cell in row: if getattr(cell, "is_date", False): if cell.number_format != "yyyy-mm-dd hh:mm:ss": cell.number_format = "yyyy-mm-dd hh:mm:ss" changed = True return changed def _normalize_number_formats(sheet) -> bool: # type: ignore[no-untyped-def] changed = False for row in sheet.iter_rows(min_row=2): for cell in row: if isinstance(cell.value, (int, float)) and not getattr(cell, "is_date", False): if cell.number_format == "General": cell.number_format = "0.00" if isinstance(cell.value, float) else "0" changed = True return changed def _normalize_url_cells(sheet) -> bool: # type: ignore[no-untyped-def] changed = False for row in sheet.iter_rows(min_row=2): for cell in row: text = _normalize_optional_text(cell.value) if text and (text.startswith("http://") or text.startswith("https://")) and text != cell.value: cell.value = text changed = True return changed def _extract_import_records(payload: Any) -> list[JSONObject]: if isinstance(payload, dict): for key in ("list", "records", "items"): value = payload.get(key) if isinstance(value, list): return [item for item in value if isinstance(item, dict)] if isinstance(payload, list): return [item for item in payload if isinstance(item, dict)] return [] def _match_import_record( records: list[JSONObject], *, local_job: dict[str, Any] | None, import_id: str | None, process_id_str: str | None, ) -> tuple[JSONObject | None, str | None]: if process_id_str: exact = [ item for item in records if _normalize_optional_text(item.get("processIdStr") or item.get("processId") or item.get("process_id_str")) == process_id_str ] if len(exact) == 1: return exact[0], "process_id_str" if len(exact) > 1: return None, "process_id_str" if import_id: exact = [ item for item in records if import_id in _extract_import_record_ids(item) ] if len(exact) == 1: return exact[0], "import_id" if len(exact) > 1: return None, "import_id" if isinstance(local_job, dict): source_file_name = _normalize_optional_text(local_job.get("source_file_name")) started_at = _parse_utc(local_job.get("started_at")) candidates = records if source_file_name: candidates = [ item for item in candidates if _normalize_optional_text(item.get("sourceFileName") or item.get("source_file_name")) == source_file_name ] if started_at is not None: window_end = started_at + timedelta(minutes=10) timed = [] for item in candidates: operate_time = _parse_utc(item.get("operateTime")) if operate_time is None: continue if started_at - timedelta(minutes=1) <= operate_time <= window_end: timed.append(item) if len(timed) == 1: return timed[0], "local_job_window" if len(timed) > 1: return None, "local_job_window" if len(candidates) == 1: return candidates[0], "source_file_name" if len(candidates) > 1: return None, "source_file_name" return None, None def _extract_import_record_ids(record: JSONObject) -> set[str]: identifiers: set[str] = set() for key in ("importId", "import_id", "dataImportId", "data_import_id"): normalized = _normalize_optional_text(record.get(key)) if normalized: identifiers.add(normalized) return identifiers def _parse_utc(value: Any) -> datetime | None: text = _normalize_optional_text(value) if text is None: return None normalized = text.replace("Z", "+00:00") try: parsed = datetime.fromisoformat(normalized) except ValueError: return None if parsed.tzinfo is None: return parsed.replace(tzinfo=timezone.utc) return parsed.astimezone(timezone.utc) def _coerce_int(value: Any) -> int | None: if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return None def _normalize_import_status(value: Any) -> str: status_code = _coerce_int(value) if status_code is not None: return IMPORT_STATUS_BY_PROCESS_STATUS.get(status_code, "unknown") text = str(value or "").strip().lower() if text in {"queued", "running", "succeeded", "failed", "partially_failed", "unknown"}: return text if text in {"line_up", "lineup"}: return "queued" if text in {"execute", "executing", "processing"}: return "running" if text in {"success", "completed"}: return "succeeded" if text in {"partly_fail", "partial_fail", "partially_fail", "partial_failed"}: return "partially_failed" if text in {"fail", "error"}: return "failed" return "unknown" def _runtime_error_payload(error: RuntimeError) -> JSONObject: try: payload = json.loads(str(error)) except json.JSONDecodeError: return {"message": str(error)} return payload if isinstance(payload, dict) else {"message": str(error)} def _runtime_import_can_import_value(error: RuntimeError) -> bool | None: payload = _runtime_error_payload(error) category = str(payload.get("category") or "").strip().lower() if category == "auth" or _coerce_int(payload.get("http_status")) == 401 or message_looks_like_invalid_token(payload.get("message")): return None backend_code = backend_code_int( QingflowApiError( category=category, message=str(payload.get("message") or ""), backend_code=payload.get("backend_code"), http_status=_coerce_int(payload.get("http_status")), ) ) return False if backend_code in {40002, 40027} else None def _runtime_candidate_validation_skippable(error: RuntimeError) -> bool: payload = _runtime_error_payload(error) category = str(payload.get("category") or "").strip().lower() if category == "not_supported": return True if category in {"auth", "workspace"}: return False if message_looks_like_invalid_token(payload.get("message")): return False if _coerce_int(payload.get("http_status")) == 401: return False backend_code = backend_code_int( QingflowApiError( category=category, message=str(payload.get("message") or ""), backend_code=payload.get("backend_code"), http_status=_coerce_int(payload.get("http_status")), ) ) return backend_code in {40002, 40027, 404} or _coerce_int(payload.get("http_status")) == 404 def _runtime_error_code(payload: JSONObject, *, default: str) -> str: category = str(payload.get("category") or "").strip().lower() http_status = _coerce_int(payload.get("http_status")) if category == "auth" or http_status == 401 or message_looks_like_invalid_token(payload.get("message")): return "AUTH_REQUIRED" if category == "workspace": return "WORKSPACE_NOT_SELECTED" return default def _is_import_permission_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return backend_code_int(error) in {40002, 40027} def _is_import_capability_auth_error(error: QingflowApiError) -> bool: return is_auth_like_error(error) def _is_optional_import_capability_error(error: QingflowApiError) -> bool: if _is_import_capability_auth_error(error): return False return _is_import_permission_error(error) or backend_code_int(error) == 404 or error.http_status == 404 def _import_permission_error_code(error: QingflowApiError, *, permission_code: str, default: str) -> str: if is_auth_like_error(error): return "AUTH_REQUIRED" if _is_import_permission_error(error): return permission_code return default def _copy_api_error_fields(payload: dict[str, Any], error: QingflowApiError) -> None: if error.category: payload["category"] = error.category if error.backend_code is not None: payload["backend_code"] = error.backend_code if error.http_status is not None: payload["http_status"] = error.http_status if error.request_id: payload["request_id"] = error.request_id def _normalize_error_file_urls(value: Any) -> list[str]: if isinstance(value, list): return [str(item).strip() for item in value if str(item).strip()] return []