from __future__ import annotations import json import re from copy import deepcopy from datetime import datetime, timedelta, timezone from pathlib import Path from time import monotonic, sleep from typing import Any, cast from uuid import uuid4 from mcp.server.fastmcp import FastMCP from ..backend_client import BackendRequestContext from ..config import DEFAULT_PROFILE, DEFAULT_RECORD_LIST_TYPE from ..errors import QingflowApiError, backend_code_int, is_auth_like_error, message_looks_like_invalid_token from ..export_store import ExportJobStore from ..json_types import JSONObject from .base import ToolBase from .record_tools import ( AccessibleViewRoute, DEFAULT_LIST_PAGE_SIZE, FormField, LAYOUT_ONLY_QUE_TYPES, RecordTools, _build_top_level_field_index, _normalize_data_list_base_info_schema, _normalize_public_column_selectors, ) EXPORT_STATUS_BY_PROCESS_STATUS = { 1: "queued", 2: "running", 3: "succeeded", 4: "failed", 5: "failed", } DEFAULT_EXPORT_TIMEOUT_SECONDS = 60.0 EXPORT_POLL_INTERVAL_SECONDS = 1.5 EXPORT_ROWS_LIMIT = 20_000 EXPORT_CELL_LIMIT = 4_000_000 _SAFE_FILE_CHARS = re.compile(r"[^0-9A-Za-z._-]+") LOCAL_TIMEZONE = datetime.now().astimezone().tzinfo or timezone.utc class ExportTools(ToolBase): def __init__( self, sessions, backend, *, job_store: ExportJobStore | None = None, ) -> None: super().__init__(sessions, backend) self._record_tools = RecordTools(sessions, backend) self._job_store = job_store or ExportJobStore() def register(self, mcp: FastMCP) -> None: @mcp.tool(description="Start an asynchronous xlsx export for a Qingflow record view and return an opaque export_handle.") def record_export_start( profile: str = DEFAULT_PROFILE, app_key: str = "", view_id: str = "", columns: list[JSONObject | int] | None = None, where: list[JSONObject] | None = None, order_by: list[JSONObject] | None = None, record_id: str | int | None = None, record_ids: list[str | int] | None = None, include_workflow_log: bool = False, ) -> dict[str, Any]: return self.record_export_start( profile=profile, app_key=app_key, view_id=view_id, columns=columns or [], where=where or [], order_by=order_by or [], record_id=record_id, record_ids=record_ids or [], include_workflow_log=include_workflow_log, ) @mcp.tool(description="Get asynchronous record export status by opaque export_handle.") def record_export_status_get( profile: str = DEFAULT_PROFILE, export_handle: str = "", ) -> dict[str, Any]: return self.record_export_status_get( profile=profile, export_handle=export_handle, ) @mcp.tool(description="Get completed record export result, returning remote file links and optionally downloading files locally.") def record_export_get( profile: str = DEFAULT_PROFILE, export_handle: str = "", download_to_path: str | None = None, ) -> dict[str, Any]: return self.record_export_get( profile=profile, export_handle=export_handle, download_to_path=download_to_path, ) @mcp.tool(description="Start a record export, wait for completion, and download the xlsx locally while also returning remote download links.") def record_export_direct( profile: str = DEFAULT_PROFILE, app_key: str = "", view_id: str = "", columns: list[JSONObject | int] | None = None, where: list[JSONObject] | None = None, order_by: list[JSONObject] | None = None, record_id: str | int | None = None, record_ids: list[str | int] | None = None, include_workflow_log: bool = False, download_to_path: str | None = None, wait_timeout_seconds: float | None = None, ) -> dict[str, Any]: return self.record_export_direct( profile=profile, app_key=app_key, view_id=view_id, columns=columns or [], where=where or [], order_by=order_by or [], record_id=record_id, record_ids=record_ids or [], include_workflow_log=include_workflow_log, download_to_path=download_to_path, wait_timeout_seconds=wait_timeout_seconds, ) def record_export_start( self, *, profile: str = DEFAULT_PROFILE, app_key: str, view_id: str = "", columns: list[JSONObject | int] | None = None, where: list[JSONObject] | None = None, order_by: list[JSONObject] | None = None, record_id: str | int | None = None, record_ids: list[str | int] | None = None, include_workflow_log: bool = False, ) -> dict[str, Any]: normalized_app_key = str(app_key or "").strip() normalized_view_id = str(view_id or "").strip() normalized_columns = _normalize_export_columns(columns or []) normalized_where = self._record_tools._normalize_record_list_where(where or []) normalized_order_by = self._record_tools._normalize_record_list_order_by(order_by or []) normalized_record_ids = _normalize_export_record_ids(_merge_single_export_record_id(record_id, record_ids or [])) if not normalized_app_key: return self._failed_export_result( error_code="EXPORT_START_FAILED", message="app_key is required", extra={"view_id": normalized_view_id, "status": "failed"}, ) if not normalized_view_id: return self._failed_export_result( error_code="EXPORT_VIEW_REQUIRED", message="view_id is required; call app_get first and pass accessible_views[].view_id or use the view_id from the frontend URL", extra={"app_key": normalized_app_key, "view_id": normalized_view_id, "status": "failed"}, ) def runner(session_profile, context): resolved_view, compatibility_warnings = self._record_tools._resolve_accessible_view_route( profile, context, normalized_app_key, view_id=normalized_view_id, list_type=None, view_key=None, view_name=None, allow_default=False, ) export_config, export_config_warnings = self._build_export_config( profile=profile, context=context, app_key=normalized_app_key, resolved_view=resolved_view, column_selectors=normalized_columns, ) effective_record_ids, row_scope = self._resolve_export_record_scope( profile=profile, context=context, app_key=normalized_app_key, resolved_view=resolved_view, selected_record_ids=normalized_record_ids, where_filters=normalized_where, order_by=normalized_order_by, select_columns=cast(list[JSONObject], export_config.get("questionExportConfigList") or []), ) result_amount = self._estimate_export_result_amount( context, app_key=normalized_app_key, resolved_view=resolved_view, selected_record_ids=effective_record_ids, ) self._validate_export_limits( result_amount=result_amount, selected_field_count=len(cast(list[JSONObject], export_config.get("questionExportConfigList") or [])), include_workflow_log=include_workflow_log, ) filter_bean = self._build_export_filter_bean( resolved_view, selected_record_ids=effective_record_ids, order_by=normalized_order_by, row_scope=row_scope, include_workflow_log=include_workflow_log, ) started_at = _utc_now().replace(microsecond=0).isoformat() socket_result = self.backend.start_socket_record_export( context, app_key=normalized_app_key, view_id=resolved_view.view_id, view_key=resolved_view.view_selection.view_key if resolved_view.view_selection is not None else None, filter_bean=filter_bean, export_config=export_config, result_amount=result_amount, ) column_scope = "selected" if normalized_columns else "all" export_handle = uuid4().hex self._job_store.put( export_handle, { "created_at": started_at, "profile": profile, "base_url": context.base_url, "ws_id": context.ws_id, "qf_version": context.qf_version, "qf_version_source": context.qf_version_source, "app_key": normalized_app_key, "view_id": resolved_view.view_id, "backend_export_id": str(socket_result.get("backend_export_id") or ""), "started_at": started_at, "uid": session_profile.uid, "row_scope": row_scope, "selected_record_count": len(effective_record_ids), "field_scope": column_scope, "selected_field_count": len(cast(list[JSONObject], export_config.get("questionExportConfigList") or [])), "include_workflow_log": bool(include_workflow_log), }, ) warnings = [ *deepcopy(compatibility_warnings), *deepcopy(export_config_warnings), *deepcopy(cast(list[JSONObject], socket_result.get("warnings") or [])), ] return { "ok": True, "status": "accepted", "export_executed": True, "safe_to_retry_export": False, "app_key": normalized_app_key, "view_id": resolved_view.view_id, "export_handle": export_handle, "row_scope": row_scope, "selected_record_count": len(effective_record_ids), "field_scope": column_scope, "selected_field_count": len(cast(list[JSONObject], export_config.get("questionExportConfigList") or [])), "include_workflow_log": bool(include_workflow_log), "file_urls": [], "file_names": [], "downloaded_files": [], "warnings": warnings, "verification": { "export_acknowledged": bool(socket_result.get("backend_export_id")), "view_route_supported": True, "row_selection_applied": bool(effective_record_ids), "query_filter_applied": bool(normalized_where), "query_sort_applied": bool(normalized_order_by), "field_selection_applied": bool(normalized_columns), }, "request_route": self.backend.describe_route(context), } try: return self._run(profile, runner, tool_name='记录导出启动') except RuntimeError as exc: return self._runtime_error_as_result( exc, error_code="EXPORT_START_FAILED", extra={"app_key": normalized_app_key, "view_id": normalized_view_id}, ) def record_export_status_get( self, *, profile: str = DEFAULT_PROFILE, export_handle: str, ) -> dict[str, Any]: normalized_handle = str(export_handle or "").strip() if not normalized_handle: return self._failed_export_result( error_code="CONFIG_ERROR", message="export_handle is required", extra={"status": "failed"}, ) def runner(session_profile, context): local_job = self._job_store.get(normalized_handle) if local_job is None: return self._failed_export_result( error_code="EXPORT_HANDLE_UNKNOWN", message="export_handle is missing or expired", extra={"export_handle": normalized_handle, "status": "failed"}, ) lookup_context = self._build_export_lookup_context( profile=profile, session_profile=session_profile, current_context=context, local_job=local_job, ) snapshot = self._resolve_export_snapshot(lookup_context, local_job) return self._status_payload_from_snapshot(local_job, normalized_handle, snapshot) try: return self._run(profile, runner, tool_name='记录导出状态') except RuntimeError as exc: return self._runtime_error_as_result( exc, error_code="EXPORT_STATUS_FAILED", extra={"export_handle": normalized_handle}, ) def record_export_get( self, *, profile: str = DEFAULT_PROFILE, export_handle: str, download_to_path: str | None = None, ) -> dict[str, Any]: normalized_handle = str(export_handle or "").strip() if not normalized_handle: return self._failed_export_result( error_code="CONFIG_ERROR", message="export_handle is required", extra={"status": "failed"}, ) def runner(session_profile, context): local_job = self._job_store.get(normalized_handle) if local_job is None: return self._failed_export_result( error_code="EXPORT_HANDLE_UNKNOWN", message="export_handle is missing or expired", extra={"export_handle": normalized_handle, "status": "failed"}, ) lookup_context = self._build_export_lookup_context( profile=profile, session_profile=session_profile, current_context=context, local_job=local_job, ) snapshot = self._resolve_export_snapshot(lookup_context, local_job) normalized_status = str(snapshot.get("status") or "unknown") if normalized_status not in {"succeeded", "failed"}: return self._failed_export_result( error_code="EXPORT_NOT_READY", message="export is not ready yet", extra={ "status": "blocked", "export_handle": normalized_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "row_scope": str(local_job.get("row_scope") or "all"), "selected_record_count": _coerce_int(local_job.get("selected_record_count")) or 0, "field_scope": str(local_job.get("field_scope") or "all"), "selected_field_count": _coerce_int(local_job.get("selected_field_count")), "include_workflow_log": bool(local_job.get("include_workflow_log")), "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], "downloaded_files": [], "warnings": snapshot.get("warnings") or [], "verification": snapshot.get("verification") or {}, }, ) if normalized_status == "failed": return { **self._failed_export_result( error_code=str(snapshot.get("error_code") or "EXPORT_FAILED"), message=str(snapshot.get("message") or "export failed"), extra={ "status": "failed", "export_handle": normalized_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "row_scope": str(local_job.get("row_scope") or "all"), "selected_record_count": _coerce_int(local_job.get("selected_record_count")) or 0, "field_scope": str(local_job.get("field_scope") or "all"), "selected_field_count": _coerce_int(local_job.get("selected_field_count")), "include_workflow_log": bool(local_job.get("include_workflow_log")), "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], "downloaded_files": [], "warnings": snapshot.get("warnings") or [], "verification": snapshot.get("verification") or {}, "num": snapshot.get("num"), "process_status": snapshot.get("process_status"), "audit_record_status": snapshot.get("audit_record_status"), }, ), } file_infos = cast(list[JSONObject], snapshot.get("file_infos") or []) if not file_infos: return self._failed_export_result( error_code="EXPORT_FILE_UNAVAILABLE", message="export completed but did not return downloadable files", extra={ "status": "failed", "export_handle": normalized_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "warnings": snapshot.get("warnings") or [], "verification": snapshot.get("verification") or {}, }, ) downloaded_files, download_warnings = self._download_export_files( file_infos=file_infos, download_to_path=download_to_path, default_directory=None, app_key=str(local_job.get("app_key") or ""), view_id=str(local_job.get("view_id") or ""), ) warnings = [ *cast(list[JSONObject], snapshot.get("warnings") or []), *download_warnings, ] return { "ok": True, "status": "succeeded", "export_handle": normalized_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "row_scope": str(local_job.get("row_scope") or "all"), "selected_record_count": _coerce_int(local_job.get("selected_record_count")) or 0, "field_scope": str(local_job.get("field_scope") or "all"), "selected_field_count": _coerce_int(local_job.get("selected_field_count")), "include_workflow_log": bool(local_job.get("include_workflow_log")), "num": snapshot.get("num"), "process_status": snapshot.get("process_status"), "error_code": snapshot.get("error_code"), "audit_record_status": snapshot.get("audit_record_status"), "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], "downloaded_files": downloaded_files, "warnings": warnings, "verification": snapshot.get("verification") or {}, "request_route": self.backend.describe_route(lookup_context), } try: return self._run(profile, runner, tool_name='记录导出结果') except RuntimeError as exc: return self._runtime_error_as_result( exc, error_code="EXPORT_GET_FAILED", extra={"export_handle": normalized_handle}, ) def record_export_direct( self, *, profile: str = DEFAULT_PROFILE, app_key: str, view_id: str = "", columns: list[JSONObject | int] | None = None, where: list[JSONObject] | None = None, order_by: list[JSONObject] | None = None, record_id: str | int | None = None, record_ids: list[str | int] | None = None, include_workflow_log: bool = False, download_to_path: str | None = None, wait_timeout_seconds: float | None = None, ) -> dict[str, Any]: normalized_app_key = str(app_key or "").strip() normalized_view_id = str(view_id or "").strip() if not normalized_app_key: return self._failed_export_result( error_code="EXPORT_START_FAILED", message="app_key is required", extra={"status": "failed", "view_id": normalized_view_id}, ) if not normalized_view_id: return self._failed_export_result( error_code="EXPORT_VIEW_REQUIRED", message="view_id is required; call app_get first and pass accessible_views[].view_id or use the view_id from the frontend URL", extra={"status": "failed", "app_key": normalized_app_key, "view_id": normalized_view_id}, ) timeout_seconds = self._normalize_timeout_seconds(wait_timeout_seconds) def runner(session_profile, context): start_result = self.record_export_start( profile=profile, app_key=normalized_app_key, view_id=normalized_view_id, columns=columns or [], where=where or [], order_by=order_by or [], record_id=record_id, record_ids=record_ids or [], include_workflow_log=include_workflow_log, ) if not bool(start_result.get("ok")): return start_result export_handle = str(start_result.get("export_handle") or "") deadline = monotonic() + timeout_seconds last_snapshot: dict[str, Any] | None = None while monotonic() < deadline: local_job = self._job_store.get(export_handle) if local_job is None: return self._failed_export_result( error_code="EXPORT_HANDLE_UNKNOWN", message="export_handle is missing or expired", extra={"status": "failed", "export_handle": export_handle}, ) lookup_context = self._build_export_lookup_context( profile=profile, session_profile=session_profile, current_context=context, local_job=local_job, ) snapshot = self._resolve_export_snapshot(lookup_context, local_job) last_snapshot = snapshot normalized_status = str(snapshot.get("status") or "unknown") if normalized_status == "succeeded": effective_download_path = download_to_path or str(Path.cwd()) get_result = self.record_export_get( profile=profile, export_handle=export_handle, download_to_path=effective_download_path, ) if bool(get_result.get("ok")): get_result = dict(get_result) get_result.setdefault("export_executed", True) get_result.setdefault("safe_to_retry_export", False) get_result.setdefault("export_handle", export_handle) return get_result return { **get_result, "export_executed": True, "safe_to_retry_export": False, "export_handle": export_handle, "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], } if normalized_status == "failed": return { "ok": False, "status": "failed", "error_code": str(snapshot.get("error_code") or "EXPORT_FAILED"), "message": str(snapshot.get("message") or "export failed"), "export_handle": export_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "row_scope": str(local_job.get("row_scope") or "all"), "selected_record_count": _coerce_int(local_job.get("selected_record_count")) or 0, "field_scope": str(local_job.get("field_scope") or "all"), "selected_field_count": _coerce_int(local_job.get("selected_field_count")), "include_workflow_log": bool(local_job.get("include_workflow_log")), "num": snapshot.get("num"), "process_status": snapshot.get("process_status"), "audit_record_status": snapshot.get("audit_record_status"), "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], "downloaded_files": [], "warnings": snapshot.get("warnings") or [], "verification": snapshot.get("verification") or {}, } if normalized_status == "unknown": warning_codes = { str(item.get("code") or "") for item in cast(list[JSONObject], snapshot.get("warnings") or []) if isinstance(item, dict) } if "EXPORT_HISTORY_AMBIGUOUS" in warning_codes: return { "ok": False, "status": "unknown", "error_code": "EXPORT_STATUS_UNKNOWN", "export_executed": True, "safe_to_retry_export": False, "export_handle": export_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "row_scope": str(local_job.get("row_scope") or "all"), "selected_record_count": _coerce_int(local_job.get("selected_record_count")) or 0, "field_scope": str(local_job.get("field_scope") or "all"), "selected_field_count": _coerce_int(local_job.get("selected_field_count")), "include_workflow_log": bool(local_job.get("include_workflow_log")), "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], "downloaded_files": [], "warnings": snapshot.get("warnings") or [], "verification": snapshot.get("verification") or {}, "message": "export result could not be matched uniquely", } remaining = deadline - monotonic() if remaining <= 0: break sleep(min(EXPORT_POLL_INTERVAL_SECONDS, remaining)) timeout_status = "running" if isinstance(last_snapshot, dict) and str(last_snapshot.get("status") or "").strip(): timeout_status = str(last_snapshot.get("status") or timeout_status) warnings = list(cast(list[JSONObject], start_result.get("warnings") or [])) warnings.append( { "code": "EXPORT_WAIT_TIMEOUT", "message": "export is still running; reuse export_handle with record_export_status_get or record_export_get", } ) return { "ok": False, "status": timeout_status, "error_code": "EXPORT_WAIT_TIMEOUT", "export_executed": True, "safe_to_retry_export": False, "export_handle": str(start_result.get("export_handle") or ""), "app_key": normalized_app_key, "view_id": str(start_result.get("view_id") or normalized_view_id), "row_scope": start_result.get("row_scope") or ("selected" if record_ids else "all"), "selected_record_count": start_result.get("selected_record_count") or 0, "field_scope": start_result.get("field_scope") or ("selected" if columns else "all"), "selected_field_count": start_result.get("selected_field_count"), "include_workflow_log": start_result.get("include_workflow_log"), "file_urls": (last_snapshot or {}).get("file_urls") or [], "file_names": (last_snapshot or {}).get("file_names") or [], "downloaded_files": [], "warnings": warnings, "verification": (last_snapshot or {}).get("verification") or start_result.get("verification") or {}, "message": "export did not finish before wait_timeout_seconds", } try: return self._run(profile, runner, tool_name='记录直接导出') except RuntimeError as exc: return self._runtime_error_as_result( exc, error_code="EXPORT_DIRECT_FAILED", extra={"app_key": normalized_app_key, "view_id": normalized_view_id}, ) def _build_export_lookup_context( self, *, profile: str, session_profile, current_context: BackendRequestContext, local_job: dict[str, Any], ) -> BackendRequestContext: stored_profile = str(local_job.get("profile") or "").strip() if stored_profile and stored_profile != profile: raise QingflowApiError.config_error( "export_handle was created under a different profile", details={ "error_code": "EXPORT_HANDLE_PROFILE_MISMATCH", "expected_profile": stored_profile, "received_profile": profile, }, ) stored_uid = _coerce_positive_int(local_job.get("uid")) if stored_uid is not None and stored_uid != session_profile.uid: raise QingflowApiError.config_error( "export_handle belongs to a different authenticated user", details={ "error_code": "EXPORT_HANDLE_OWNER_MISMATCH", "expected_uid": stored_uid, "current_uid": session_profile.uid, }, ) stored_base_url = str(local_job.get("base_url") or "").strip() or current_context.base_url stored_ws_id = _coerce_positive_int(local_job.get("ws_id")) stored_qf_version = str(local_job.get("qf_version") or "").strip() or current_context.qf_version stored_qf_version_source = ( str(local_job.get("qf_version_source") or "").strip() or current_context.qf_version_source ) return BackendRequestContext( base_url=stored_base_url, token=current_context.token, ws_id=stored_ws_id if stored_ws_id is not None else current_context.ws_id, qf_request_id=current_context.qf_request_id, qf_version=stored_qf_version, qf_version_source=stored_qf_version_source, ) def _resolve_export_record_scope( self, *, profile: str, context, app_key: str, resolved_view: AccessibleViewRoute, selected_record_ids: list[int], where_filters: list[JSONObject], order_by: list[JSONObject], select_columns: list[JSONObject], ) -> tuple[list[int], str]: if selected_record_ids and (where_filters or order_by): raise QingflowApiError( category="config", message="record export does not allow record_ids together with query selectors", details={ "error_code": "EXPORT_ROW_SCOPE_CONFLICT", "fix_hint": "Use record_ids for explicit selected rows, or use where/order_by for internal query selection, but not both.", }, ) if selected_record_ids: return selected_record_ids, "selected" if not where_filters and not order_by: return [], "all" resolved_ids = self._collect_record_ids_from_query( profile=profile, context=context, app_key=app_key, resolved_view=resolved_view, where_filters=where_filters, order_by=order_by, select_columns=select_columns, ) if not resolved_ids: raise QingflowApiError.config_error( "record export query did not match any records", details={ "error_code": "EXPORT_NO_MATCHED_RECORDS", "view_id": resolved_view.view_id, }, ) return resolved_ids, "queried" def _collect_record_ids_from_query( self, *, profile: str, context, app_key: str, resolved_view: AccessibleViewRoute, where_filters: list[JSONObject], order_by: list[JSONObject], select_columns: list[JSONObject], ) -> list[int]: browse_scope = self._build_export_read_scope( profile, context, app_key, resolved_view, force_refresh=False, ) index = browse_scope["index"] match_rules = self._record_tools._resolve_match_rules(context, where_filters, index) query_sorts = [ { "queId": _coerce_int(item.get("field_id")), "direction": str(item.get("direction") or "asc"), } for item in order_by if isinstance(item, dict) and _coerce_int(item.get("field_id")) is not None ] dept_member_cache: dict[int, set[int]] = {} current_page = 1 selected_ids: list[int] = [] seen: set[int] = set() primary_field_ids = [ _coerce_int(item.get("queId")) for item in select_columns if isinstance(item, dict) ] primary_search_que_ids = [item for item in primary_field_ids if item is not None][:1] or None while True: page = self._record_tools._search_page( context, app_key=app_key, view_selection=resolved_view.view_selection, page_num=current_page, page_size=DEFAULT_LIST_PAGE_SIZE, query_key=None, match_rules=match_rules, sorts=cast(list[JSONObject], query_sorts), search_que_ids=primary_search_que_ids, list_type=resolved_view.list_type if resolved_view.list_type is not None else DEFAULT_RECORD_LIST_TYPE, ) raw_rows = page.get("list") items = raw_rows if isinstance(raw_rows, list) else [] for item in items: if not isinstance(item, dict): continue answers = item.get("answers") answer_list = answers if isinstance(answers, list) else [] if not self._record_tools._matches_view_selection( context, answer_list, view_selection=resolved_view.view_selection, dept_member_cache=dept_member_cache, ): continue record_id = _coerce_int(item.get("applyId")) if record_id is None: record_id = _coerce_int(item.get("apply_id")) if record_id is None: record_id = _coerce_int(item.get("id")) if record_id is None: record_id = _coerce_int(item.get("record_id")) if record_id is None or record_id in seen or record_id <= 0: continue seen.add(record_id) selected_ids.append(record_id) if len(selected_ids) > EXPORT_ROWS_LIMIT: raise QingflowApiError.config_error( f"record export exceeds the native row limit of {EXPORT_ROWS_LIMIT}", details={ "error_code": "EXPORT_ROWS_LIMIT_EXCEEDED", "result_amount": len(selected_ids), "row_limit": EXPORT_ROWS_LIMIT, }, ) if current_page == 1: reported_total = _effective_total(page, page_size=DEFAULT_LIST_PAGE_SIZE) if reported_total > EXPORT_ROWS_LIMIT: raise QingflowApiError.config_error( f"record export exceeds the native row limit of {EXPORT_ROWS_LIMIT}", details={ "error_code": "EXPORT_ROWS_LIMIT_EXCEEDED", "result_amount": reported_total, "row_limit": EXPORT_ROWS_LIMIT, }, ) if not _page_has_more(page, current_page=current_page, page_size=DEFAULT_LIST_PAGE_SIZE, returned_rows=len(items)): break current_page += 1 return selected_ids def _build_export_filter_bean( self, resolved_view: AccessibleViewRoute, *, selected_record_ids: list[int], order_by: list[JSONObject], row_scope: str, include_workflow_log: bool, ) -> JSONObject: filter_payload: JSONObject = {} if resolved_view.kind == "system" and resolved_view.list_type is not None: filter_payload["type"] = resolved_view.list_type elif resolved_view.kind == "custom": # Custom-view native export later flows through shared data-export code that # expects a list semantics integer. Frontend export treats custom views as # creator-all style export, so mirror LIST_CREATOR_ALL here. filter_payload["type"] = DEFAULT_RECORD_LIST_TYPE if selected_record_ids: filter_payload["applyIds"] = selected_record_ids if row_scope == "queried" and order_by: normalized_sorts = [ { "queId": field_id, "isAscend": str(item.get("direction") or "asc").strip().lower() != "desc", } for item in order_by if isinstance(item, dict) and (field_id := _coerce_int(item.get("field_id"))) is not None ] if normalized_sorts: filter_payload["sorts"] = normalized_sorts return { "filter": filter_payload, # Backend export code later auto-unboxes this field to primitive boolean. # Always send an explicit boolean to avoid a null -> NPE -> 41100 failure path. "auditRecordStatus": bool(include_workflow_log), } def _build_export_config( self, *, profile: str, context, app_key: str, resolved_view: AccessibleViewRoute, column_selectors: list[int], ) -> tuple[JSONObject, list[JSONObject]]: # type: ignore[no-untyped-def] browse_scope = self._build_export_read_scope( profile, context, app_key, resolved_view, force_refresh=False, ) index = browse_scope["index"] visible_question_ids = cast(set[int], browse_scope.get("visible_question_ids") or set()) ordered_visible_fields, warnings = self._resolve_exportable_fields( profile=profile, context=context, app_key=app_key, resolved_view=resolved_view, index=index, visible_question_ids=visible_question_ids, ) if not ordered_visible_fields: ordered_visible_fields = [ field for field in cast(Any, index).by_id.values() if field.que_type not in LAYOUT_ONLY_QUE_TYPES ] exportable_by_id = {field.que_id: field for field in ordered_visible_fields} selected_fields: list[Any] if column_selectors: selected_fields = [] seen: set[int] = set() for selector in column_selectors: field = self._record_tools._resolve_field_selector(selector, index, location="export_columns") if field.que_id in seen: continue exportable_field = exportable_by_id.get(field.que_id) if exportable_field is None: raise QingflowApiError.config_error( f"field '{field.que_title}' is not exportable in the selected view", details={ "error_code": "EXPORT_FIELD_NOT_VISIBLE", "field_id": field.que_id, "view_id": resolved_view.view_id, }, ) selected_fields.append(exportable_field) seen.add(field.que_id) else: selected_fields = ordered_visible_fields question_export_config_list: list[JSONObject] = [] for field in selected_fields: if field.que_type is None: warnings.append( { "code": "EXPORT_FIELD_TYPE_UNAVAILABLE", "message": f"Skipped field '{field.que_title}' because its field type could not be resolved.", } ) continue question_export_config_list.append( { "queId": field.que_id, "queTitle": field.que_title, "queType": field.que_type, "exportStyle": "default", } ) if not question_export_config_list: raise QingflowApiError.config_error( "record export could not determine exportable fields for the selected view", details={"error_code": "EXPORT_CONFIG_UNAVAILABLE"}, ) return {"questionExportConfigList": question_export_config_list}, warnings def _build_export_read_scope( self, profile: str, context, app_key: str, resolved_view: AccessibleViewRoute, *, force_refresh: bool, ) -> JSONObject: try: scope = self._record_tools._build_browse_read_scope( profile, context, app_key, resolved_view, force_refresh=force_refresh, ) except QingflowApiError as exc: if not _is_optional_export_lookup_error(exc): raise scope = {} index = scope.get("index") if isinstance(scope, dict) else None if getattr(index, "by_id", None): return scope if resolved_view.kind == "system" and resolved_view.list_type is not None: try: list_base_scope = self._build_system_export_list_base_info_scope(context, app_key) except QingflowApiError as exc: if not _is_optional_export_lookup_error(exc): raise else: list_base_index = list_base_scope.get("index") if getattr(list_base_index, "by_id", None): return list_base_scope try: applicant_index = self._record_tools._get_applicant_top_level_field_index( profile, context, app_key, force_refresh=force_refresh, ) except QingflowApiError as exc: if not _is_optional_export_lookup_error(exc): raise applicant_index = None if applicant_index is not None and applicant_index.by_id: visible_question_ids = {field.que_id for field in applicant_index.by_id.values()} return { "index": applicant_index, "writable_field_ids": set(), "visible_question_ids": visible_question_ids, } return scope or {"index": _build_top_level_field_index({}), "writable_field_ids": set(), "visible_question_ids": set()} def _build_system_export_list_base_info_scope(self, context, app_key: str) -> JSONObject: payload = self.backend.request("GET", context, f"/app/{app_key}/data/listBaseInfo") schema = _normalize_data_list_base_info_schema(payload) index = _build_top_level_field_index(schema) visible_question_ids = {field.que_id for field in index.by_id.values()} return { "index": index, "writable_field_ids": set(), "visible_question_ids": visible_question_ids, } def _resolve_exportable_fields( self, *, profile: str, context, app_key: str, resolved_view: AccessibleViewRoute, index, visible_question_ids: set[int], ) -> tuple[list[FormField], list[JSONObject]]: # type: ignore[no-untyped-def] warnings: list[JSONObject] = [] if resolved_view.kind == "custom" and resolved_view.view_selection is not None: custom_fields = self._resolve_custom_view_exportable_fields( profile=profile, context=context, app_key=app_key, view_key=resolved_view.view_selection.view_key, index=index, visible_question_ids=visible_question_ids, ) if custom_fields is not None: return custom_fields, warnings warnings.append( { "code": "EXPORT_VIEW_CONFIG_PARTIAL", "message": "custom view export fields fell back to schema order because viewConfig could not provide exportable field entries", } ) ordered_visible_fields = [ field for field in self._record_tools._schema_fields_for_mode( profile, context, app_key, index, schema_mode="browse", resolved_view=resolved_view, ) if field.que_id in visible_question_ids and field.que_type not in LAYOUT_ONLY_QUE_TYPES ] return ordered_visible_fields, warnings def _resolve_custom_view_exportable_fields( self, *, profile: str, context, app_key: str, view_key: str, index, visible_question_ids: set[int], ) -> list[FormField] | None: # type: ignore[no-untyped-def] view_config = self._record_tools._get_view_config(profile, context, view_key) if not isinstance(view_config, dict): return None entries = _extract_export_view_question_entries(view_config.get("viewgraphQuestions")) if not entries: return [] ordered_fields: list[FormField] = [] seen: set[int] = set() for entry in entries: if not bool(entry.get("downloadable", True)): continue field_id = _coerce_int(entry.get("field_id")) if field_id is None or field_id in seen: continue field = cast(FormField | None, cast(Any, index).by_id.get(str(field_id))) if field is None or field.que_type in LAYOUT_ONLY_QUE_TYPES: continue ordered_fields.append(field) seen.add(field_id) return ordered_fields def _estimate_export_result_amount( self, context, *, app_key: str, resolved_view: AccessibleViewRoute, selected_record_ids: list[int], ) -> int: if selected_record_ids: return len(selected_record_ids) page = self._record_tools._search_page( context, app_key=app_key, view_selection=resolved_view.view_selection, page_num=1, page_size=1, query_key=None, match_rules=[], sorts=[], search_que_ids=None, list_type=resolved_view.list_type if resolved_view.list_type is not None else DEFAULT_RECORD_LIST_TYPE, ) result_amount = _effective_total(page, page_size=1) if result_amount < 0: result_amount = 0 return result_amount def _validate_export_limits( self, *, result_amount: int, selected_field_count: int, include_workflow_log: bool, ) -> None: if result_amount > EXPORT_ROWS_LIMIT: raise QingflowApiError.config_error( f"record export exceeds the native row limit of {EXPORT_ROWS_LIMIT}", details={ "error_code": "EXPORT_ROWS_LIMIT_EXCEEDED", "result_amount": result_amount, "row_limit": EXPORT_ROWS_LIMIT, }, ) estimated_cells = max(result_amount, 0) * max(selected_field_count, 0) if estimated_cells > EXPORT_CELL_LIMIT: raise QingflowApiError.config_error( f"record export exceeds the native cell limit of {EXPORT_CELL_LIMIT}", details={ "error_code": "EXPORT_CELLS_LIMIT_EXCEEDED", "estimated_cells": estimated_cells, "cell_limit": EXPORT_CELL_LIMIT, "include_workflow_log": include_workflow_log, }, ) def _resolve_export_snapshot( self, context, local_job: dict[str, Any], ) -> dict[str, Any]: # type: ignore[no-untyped-def] app_key = str(local_job.get("app_key") or "").strip() process_payload = self._lookup_process_details(context, app_key=app_key) history_unavailable_warning: JSONObject | None = None try: history_page = self.backend.request( "GET", context, "/app/apply/dataExport/record", params={"appKey": app_key, "pageNum": 1, "pageSize": 100}, ) except QingflowApiError as exc: if not _is_optional_export_lookup_error(exc): raise history_page = {"list": []} history_unavailable_warning = { "code": "EXPORT_HISTORY_UNAVAILABLE", "message": "export history is not readable for the current user; using current process details when available.", } if exc.category: history_unavailable_warning["category"] = exc.category if exc.backend_code is not None: history_unavailable_warning["backend_code"] = exc.backend_code if exc.http_status is not None: history_unavailable_warning["http_status"] = exc.http_status if exc.request_id: history_unavailable_warning["request_id"] = exc.request_id history_records = _extract_export_records(history_page) matched_record, matched_by = _match_export_history_record(history_records, local_job=local_job) if process_payload is not None: normalized_status = _normalize_export_status(process_payload.get("processStatus") or process_payload.get("status")) process_file_infos = _normalize_export_file_infos(process_payload.get("fileUrls")) if normalized_status in {"queued", "running"}: return { "status": normalized_status, "process_status": _coerce_int(process_payload.get("processStatus") or process_payload.get("status")), "num": _coerce_int(process_payload.get("num")), "error_code": process_payload.get("errorCode"), "audit_record_status": process_payload.get("auditRecordStatus"), "file_infos": process_file_infos, "file_urls": [item.get("url") for item in process_file_infos if isinstance(item.get("url"), str)], "file_names": [item.get("name") for item in process_file_infos if isinstance(item.get("name"), str)], "warnings": [history_unavailable_warning] if history_unavailable_warning is not None else [], "verification": { "current_process_visible": True, "history_match_resolved": matched_record is not None, "history_readable": history_unavailable_warning is None, }, "message": None, } if normalized_status in {"succeeded", "failed"} and (process_file_infos or normalized_status == "failed"): message = "export failed" if normalized_status == "failed" else None return { "status": normalized_status, "process_status": _coerce_int(process_payload.get("processStatus") or process_payload.get("status")), "num": _coerce_int(process_payload.get("num")), "error_code": process_payload.get("errorCode"), "audit_record_status": process_payload.get("auditRecordStatus"), "file_infos": process_file_infos, "file_urls": [item.get("url") for item in process_file_infos if isinstance(item.get("url"), str)], "file_names": [item.get("name") for item in process_file_infos if isinstance(item.get("name"), str)], "warnings": [history_unavailable_warning] if history_unavailable_warning is not None else [], "verification": { "current_process_visible": True, "history_match_resolved": matched_record is not None, "history_readable": history_unavailable_warning is None, "matched_by": "current_process", }, "message": message, } if matched_record is None: warning_code = "EXPORT_HISTORY_PENDING" warning_message = "export has not appeared in export history yet" if matched_by == "ambiguous": warning_code = "EXPORT_HISTORY_AMBIGUOUS" warning_message = "export result could not be matched uniquely in export history" warnings = [{"code": warning_code, "message": warning_message}] if history_unavailable_warning is not None: warnings = [history_unavailable_warning] warning_message = str(history_unavailable_warning["message"]) return { "status": "unknown", "process_status": None, "num": None, "error_code": None, "audit_record_status": None, "file_infos": [], "file_urls": [], "file_names": [], "warnings": warnings, "verification": { "current_process_visible": process_payload is not None, "history_match_resolved": False, "history_readable": history_unavailable_warning is None, }, "message": warning_message, } file_infos = _normalize_export_file_infos(matched_record.get("fileUrls")) raw_process_status = _coerce_int(matched_record.get("processStatus")) normalized_status = _normalize_export_status(raw_process_status) message = None if normalized_status == "failed": message = "export failed" return { "status": normalized_status, "process_status": raw_process_status, "num": _coerce_int(matched_record.get("num")), "error_code": matched_record.get("errorCode"), "audit_record_status": matched_record.get("auditRecordStatus"), "file_infos": file_infos, "file_urls": [item.get("url") for item in file_infos if isinstance(item.get("url"), str)], "file_names": [item.get("name") for item in file_infos if isinstance(item.get("name"), str)], "warnings": [], "verification": { "current_process_visible": process_payload is not None, "history_match_resolved": True, "matched_by": matched_by, }, "message": message, } def _lookup_process_details(self, context, *, app_key: str) -> JSONObject | None: # type: ignore[no-untyped-def] try: payload = self.backend.request( "GET", context, f"/process/{app_key}/details", params={"taskType": 2}, ) except QingflowApiError as exc: if _is_optional_export_lookup_error(exc): return None raise if isinstance(payload, dict): nested = payload.get("data") if isinstance(nested, dict): payload = nested if isinstance(payload.get("detail"), dict): return cast(JSONObject, payload.get("detail")) if any(key in payload for key in ("processStatus", "progress", "num", "fileUrls", "errorCode")): return cast(JSONObject, payload) return None def _status_payload_from_snapshot( self, local_job: dict[str, Any], export_handle: str, snapshot: dict[str, Any], ) -> dict[str, Any]: normalized_status = str(snapshot.get("status") or "unknown") ok = normalized_status not in {"failed"} or snapshot.get("error_code") in (None, "", 0) if normalized_status == "failed": ok = False return { "ok": ok, "status": normalized_status, "export_handle": export_handle, "app_key": str(local_job.get("app_key") or ""), "view_id": str(local_job.get("view_id") or ""), "row_scope": str(local_job.get("row_scope") or "all"), "selected_record_count": _coerce_int(local_job.get("selected_record_count")) or 0, "field_scope": str(local_job.get("field_scope") or "all"), "selected_field_count": _coerce_int(local_job.get("selected_field_count")), "include_workflow_log": bool(local_job.get("include_workflow_log")), "num": snapshot.get("num"), "process_status": snapshot.get("process_status"), "error_code": snapshot.get("error_code"), "audit_record_status": snapshot.get("audit_record_status"), "file_urls": snapshot.get("file_urls") or [], "file_names": snapshot.get("file_names") or [], "downloaded_files": [], "warnings": snapshot.get("warnings") or [], "verification": snapshot.get("verification") or {}, "message": snapshot.get("message"), } def _download_export_files( self, *, file_infos: list[JSONObject], download_to_path: str | None, default_directory: str | None, app_key: str, view_id: str, ) -> tuple[list[JSONObject], list[JSONObject]]: if download_to_path is None and default_directory is None: return [], [] effective_hint = download_to_path or default_directory assert effective_hint is not None targets = _resolve_download_targets( effective_hint, file_infos=file_infos, app_key=app_key, view_id=view_id, ) downloaded_files: list[JSONObject] = [] warnings: list[JSONObject] = [] for file_info, target in zip(file_infos, targets, strict=False): url = str(file_info.get("url") or "").strip() if not url: continue try: content = self.backend.download_binary(url) target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(content) except QingflowApiError as exc: warning: JSONObject = { "code": "EXPORT_FILE_DOWNLOAD_UNAVAILABLE", "message": "export file link is available, but local download failed; use file_urls or retry download later.", "file_name": str(file_info.get("name") or target.name), "url": url, "category": exc.category, "backend_code": exc.backend_code, "request_id": exc.request_id, "http_status": exc.http_status, } if is_auth_like_error(exc): warning["auth_like"] = True warning["error_code"] = "AUTH_REQUIRED" if exc.details: warning["details"] = exc.details warnings.append(warning) continue except OSError as exc: warnings.append( { "code": "EXPORT_FILE_WRITE_UNAVAILABLE", "message": "export file link is available, but writing the local file failed; use file_urls or retry with another download_to_path.", "file_name": str(file_info.get("name") or target.name), "url": url, "path": str(target), "error": str(exc), } ) continue downloaded_files.append( { "file_name": str(file_info.get("name") or target.name), "path": str(target), "url": url, } ) return downloaded_files, warnings def _normalize_timeout_seconds(self, wait_timeout_seconds: float | None) -> float: if wait_timeout_seconds is None: return DEFAULT_EXPORT_TIMEOUT_SECONDS try: value = float(wait_timeout_seconds) except (TypeError, ValueError): return DEFAULT_EXPORT_TIMEOUT_SECONDS return value if value > 0 else DEFAULT_EXPORT_TIMEOUT_SECONDS def _failed_export_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, "export_handle": None, "app_key": None, "view_id": None, "num": None, "process_status": None, "audit_record_status": None, "file_urls": [], "file_names": [], "downloaded_files": [], "warnings": [], "verification": {}, "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]: try: payload = json.loads(str(error)) except json.JSONDecodeError: payload = {"message": str(error)} details = payload.get("details") if isinstance(payload.get("details"), dict) else {} response = self._failed_export_result( error_code=details.get("error_code") or _runtime_error_code(payload, default=error_code), message=str(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 _extract_export_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 _is_optional_export_lookup_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) return backend_code in {40002, 40027, 404} or error.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 _match_export_history_record( records: list[JSONObject], *, local_job: dict[str, Any], ) -> tuple[JSONObject | None, str | None]: uid = _extract_operate_user_uid(local_job) started_at = _parse_utc(local_job.get("started_at")) candidates = records if uid is not None: candidates = [item for item in candidates if _extract_operate_user_uid(item.get("operateUser")) == uid] if started_at is not None: started_at = started_at.replace(microsecond=0) parsed_candidates: list[tuple[JSONObject, datetime]] = [] for item in candidates: operate_time = _parse_utc(item.get("operateTime")) or _parse_utc(item.get("operate_time")) if operate_time is None: continue parsed_candidates.append((item, operate_time)) exact_or_after = [item for item, operate_time in parsed_candidates if operate_time >= started_at] if len(exact_or_after) == 1: return exact_or_after[0], "operate_user_started_at" if len(exact_or_after) > 1: return None, "ambiguous" skew_tolerant = [ item for item, operate_time in parsed_candidates if operate_time >= (started_at - timedelta(seconds=5)) ] if len(skew_tolerant) == 1: return skew_tolerant[0], "operate_user_started_at_skew" if len(skew_tolerant) > 1: return None, "ambiguous" candidates = [] if len(candidates) == 1: return candidates[0], "operate_user_started_at" if len(candidates) > 1: return None, "ambiguous" return None, None def _normalize_export_status(value: Any) -> str: status_code = _coerce_int(value) if status_code is not None: return EXPORT_STATUS_BY_PROCESS_STATUS.get(status_code, "unknown") text = str(value or "").strip().lower() if text in {"queued", "running", "succeeded", "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 {"fail", "partly_fail", "partial_fail"}: return "failed" return "unknown" def _normalize_export_file_infos(value: Any) -> list[JSONObject]: if not isinstance(value, list): return [] items: list[JSONObject] = [] for item in value: if not isinstance(item, dict): continue url = str(item.get("url") or "").strip() name = str(item.get("name") or item.get("fileName") or "").strip() payload: JSONObject = {} if url: payload["url"] = url if name: payload["name"] = name if payload: items.append(payload) return items def _extract_export_file_urls(value: Any) -> list[str]: return [str(item.get("url") or "").strip() for item in _normalize_export_file_infos(value) if str(item.get("url") or "").strip()] def _extract_export_file_names(value: Any) -> list[str]: return [str(item.get("name") or "").strip() for item in _normalize_export_file_infos(value) if str(item.get("name") or "").strip()] def _extract_operate_user_uid(value: Any) -> int | None: if isinstance(value, dict): for key in ("uid", "userId", "id"): uid = _coerce_int(value.get(key)) if uid is not None: return uid return _coerce_int(value) def _parse_utc(value: Any) -> datetime | None: text = str(value or "").strip() if not text: 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=LOCAL_TIMEZONE).astimezone(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 _coerce_positive_int(value: Any) -> int | None: parsed = _coerce_int(value) if parsed is None or parsed <= 0: return None return parsed def _extract_export_view_question_entries(questions: Any) -> list[JSONObject]: if not isinstance(questions, list): return [] entries: list[JSONObject] = [] fallback_order = 0 def walk(nodes: Any) -> None: nonlocal fallback_order if not isinstance(nodes, list): return for item in nodes: if not isinstance(item, dict): continue children: list[Any] = [] for child_key in ("innerQues", "subQues", "innerQuestions", "subQuestions"): child_value = item.get(child_key) if isinstance(child_value, list) and child_value: children.extend(child_value) if children: walk(children) continue field_id = _coerce_int(item.get("queId")) if field_id is None: continue fallback_order += 1 downloadable_raw = item.get("beingDownload") entries.append( { "field_id": field_id, "name": str(item.get("queTitle") or "").strip(), "display_order": _coerce_positive_int(item.get("displayOrdinal")) or fallback_order, "downloadable": bool(downloadable_raw) if downloadable_raw is not None else True, } ) walk(questions) return sorted( entries, key=lambda entry: ( _coerce_positive_int(entry.get("display_order")) if _coerce_positive_int(entry.get("display_order")) is not None else 10**9, str(entry.get("name") or ""), ), ) def _normalize_export_columns(columns: list[JSONObject | int]) -> list[int]: normalized: list[int] = [] for field_id in _normalize_public_column_selectors(columns): if field_id not in normalized: normalized.append(field_id) return normalized def _normalize_export_record_ids(record_ids: list[str | int]) -> list[int]: normalized: list[int] = [] seen: set[int] = set() for item in record_ids: record_id = _coerce_int(item) if record_id is None or record_id <= 0 or record_id in seen: continue normalized.append(record_id) seen.add(record_id) return normalized def _merge_single_export_record_id(record_id: str | int | None, record_ids: list[str | int]) -> list[str | int]: if record_id is None or str(record_id).strip() == "": return list(record_ids) return [record_id, *record_ids] def _effective_total(page: JSONObject, *, page_size: int) -> int: rows = page.get("list") returned_rows = len(rows) if isinstance(rows, list) else 0 reported = _coerce_int(page.get("total")) if reported is None: reported = _coerce_int(page.get("count")) if reported is not None: return max(reported, returned_rows) page_amount = _coerce_int(page.get("pageAmount")) if page_amount is not None: return page_amount * page_size return returned_rows def _page_has_more(page: JSONObject, *, current_page: int, page_size: int, returned_rows: int) -> bool: page_amount = _coerce_int(page.get("pageAmount")) if page_amount is not None: return current_page < page_amount return returned_rows >= page_size def _resolve_download_targets( destination_hint: str, *, file_infos: list[JSONObject], app_key: str, view_id: str, ) -> list[Path]: path = Path(destination_hint).expanduser() timestamp = _utc_now().strftime("%Y%m%dT%H%M%SZ") if len(file_infos) > 1: if path.exists() and not path.is_dir(): raise QingflowApiError.config_error("download_to_path must be a directory when multiple export files are returned") if not path.exists() and path.suffix: raise QingflowApiError.config_error("download_to_path must be a directory when multiple export files are returned") path.mkdir(parents=True, exist_ok=True) return [ path / _choose_export_file_name( file_info, app_key=app_key, view_id=view_id, timestamp=timestamp, index=index, ) for index, file_info in enumerate(file_infos, start=1) ] if path.exists() and path.is_dir(): return [ path / _choose_export_file_name( file_infos[0], app_key=app_key, view_id=view_id, timestamp=timestamp, index=1, ) ] if path.suffix or path.exists(): return [path] path.mkdir(parents=True, exist_ok=True) return [ path / _choose_export_file_name( file_infos[0], app_key=app_key, view_id=view_id, timestamp=timestamp, index=1, ) ] def _choose_export_file_name( file_info: JSONObject, *, app_key: str, view_id: str, timestamp: str, index: int, ) -> str: remote_name = str(file_info.get("name") or "").strip() if remote_name: sanitized = _sanitize_filename(remote_name) if sanitized: return sanitized suffix = ".xlsx" safe_view = _sanitize_filename(view_id.replace(":", "_")) or "view" return f"{_sanitize_filename(app_key) or 'app'}_{safe_view}_{timestamp}_{index}{suffix}" def _sanitize_filename(value: str) -> str: base = _SAFE_FILE_CHARS.sub("_", value).strip("._") return base or "" def _utc_now() -> datetime: return datetime.now(timezone.utc)