from __future__ import annotations from typing import cast from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE, DEFAULT_RECORD_LIST_TYPE from ..errors import QingflowApiError, backend_code_int, backend_code_value_int, is_auth_like_error, message_looks_like_invalid_token, raise_tool_error from ..json_types import JSONObject, JSONValue from .record_tools import ( ATTACHMENT_QUE_TYPES, DEPARTMENT_QUE_TYPES, MEMBER_QUE_TYPES, MULTI_SELECT_QUE_TYPES, RELATION_QUE_TYPES, SINGLE_SELECT_QUE_TYPES, SUBTABLE_QUE_TYPES, FieldIndex, FormField, RecordTools, _build_answer_backed_field_index, _coerce_count, _collect_question_relations, _field_ref_payload, _merge_field_indexes, _normalize_optional_text, _relation_ids_from_answer, _stringify_json, ) CODE_BLOCK_QUE_TYPE = 26 CODE_BLOCK_RELATION_TYPE = 3 SUPPORTED_CODE_BLOCK_ROLES = {1, 2, 3, 5} _CODE_BLOCK_SCHEMA_PERMISSION_CODES = {40002, 40027, 404} class CodeBlockTools(RecordTools): """代码块工具（中文名：代码块运行与映射）。 类型：记录增强工具。 主要职责： 1. 读取代码块字段配置与关联关系； 2. 执行代码块并处理返回别名； 3. 按规则回写目标字段并输出执行摘要。 """ def _get_code_block_relation_schema( self, profile: str, context, # type: ignore[no-untyped-def] app_key: str, *, force_refresh: bool, ) -> JSONObject: """执行内部辅助逻辑。""" cache_key = (profile, app_key, "code_block_relation_form", 1) if not force_refresh and cache_key in self._form_cache: return self._form_cache[cache_key] schema = self.backend.request( "GET", context, f"/app/{app_key}/form", params={"type": 1}, ) normalized = schema if isinstance(schema, dict) else {} self._form_cache[cache_key] = normalized return normalized def _get_code_block_relation_schema_optional( self, profile: str, context, # type: ignore[no-untyped-def] app_key: str, *, force_refresh: bool, warnings: list[JSONObject], ) -> JSONObject: try: return self._get_code_block_relation_schema( profile, context, app_key, force_refresh=force_refresh, ) except QingflowApiError as exc: if not _is_optional_code_block_schema_error(exc): raise warnings.append( { "code": "CODE_BLOCK_SCHEMA_UNAVAILABLE", "message": "applicant form schema was not readable in this permission context; code-block execution will use record/task answers and skip schema-bound relation writeback.", "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) return {} def register(self, mcp: FastMCP) -> None: """注册当前工具到 MCP 服务。""" super().register(mcp) @mcp.tool() def record_code_block_schema_get( profile: str = DEFAULT_PROFILE, app_key: str = "", output_profile: str = "normal", ) -> JSONObject: return self.record_code_block_schema_get_public( profile=profile, app_key=app_key, output_profile=output_profile, ) @mcp.tool( description=( "Run a form code-block field against the current record data, parse alias results, and optionally " "reuse Qingflow's existing relation-calculation chain to compute bound outputs and write them back. " "Use record_code_block_schema_get when field selection or binding diagnostics are unclear; " "if the exact code-block field id is known from record/task detail, run directly. " "For safe debugging, pass apply_writeback=false to inspect parsed results without writing back." ) ) def record_code_block_run( profile: str = DEFAULT_PROFILE, app_key: str = "", record_id: str = "", code_block_field: str = "", view_id: str | None = None, role: int = 1, workflow_node_id: int | None = None, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, manual: bool = True, apply_writeback: bool = True, verify_writeback: bool = True, force_refresh_form: bool = False, output_profile: str = "normal", ) -> JSONObject: return self.record_code_block_run( profile=profile, app_key=app_key, record_id=record_id, code_block_field=code_block_field, view_id=view_id, role=role, workflow_node_id=workflow_node_id, answers=answers or [], fields=fields or {}, manual=manual, apply_writeback=apply_writeback, verify_writeback=verify_writeback, force_refresh_form=force_refresh_form, output_profile=output_profile, ) def record_code_block_schema_get_public( self, *, profile: str = DEFAULT_PROFILE, app_key: str, output_profile: str = "normal", ) -> JSONObject: """执行记录相关逻辑。""" if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) normalized_output_profile = self._normalize_public_output_profile(output_profile) def runner(session_profile, context): try: relation_schema = self._get_code_block_relation_schema(profile, context, app_key, force_refresh=False) index = self._get_applicant_top_level_field_index(profile, context, app_key, force_refresh=False) except QingflowApiError as exc: if not _is_optional_code_block_schema_error(exc): raise return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": False, "status": "failed", "error_code": "CODE_BLOCK_SCHEMA_UNAVAILABLE", "message": ( "applicant form schema was not readable in this permission context; " "record schema code-block is only a diagnostic helper. " "If the code-block field is known from record/task detail, run record code-block-run directly." ), "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, "request_route": self._request_route_payload(context), "warnings": [ { "code": "CODE_BLOCK_SCHEMA_UNAVAILABLE", "message": "schema diagnostic unavailable; code-block run can still use record/task answers when code_block_field is known.", "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ], "app_key": app_key, "schema_scope": "code_block_ready", "code_block_fields": [], "input_fields": [], "suggested_next_call": { "tool_name": "record_code_block_run", "required": ["app_key", "record_id", "code_block_field"], }, } input_fields = [ self._ready_schema_field_payload( profile, context, field, ws_id=session_profile.selected_ws_id, required_override=None, ) for field in index.by_id.values() if field.que_type != CODE_BLOCK_QUE_TYPE and bool(self._schema_write_hints(field).get("writable")) ] code_block_fields: list[JSONObject] = [] for field in index.by_id.values(): if field.que_type != CODE_BLOCK_QUE_TYPE: continue targets = _collect_code_block_relation_targets( _collect_question_relations(relation_schema), code_block_que_id=field.que_id, ) bound_output_fields = [ target_field.que_title for target in targets for target_field in [index.by_id.get(str(_coerce_count(target.get("que_id")) or -1))] if target_field is not None and isinstance(target_field.que_title, str) ] code_block_fields.append( { "title": field.que_title, "selector": field.que_title, "bound_output_fields": bound_output_fields, "configured_aliases": _extract_code_block_configured_aliases(field), } ) response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "status": "success", "request_route": self._request_route_payload(context), "warnings": [], "app_key": app_key, "schema_scope": "code_block_ready", "code_block_fields": code_block_fields, "input_fields": input_fields, } if normalized_output_profile == "verbose": response["legacy_schema"] = self.record_schema_get( profile=profile, app_key=app_key, schema_mode="applicant", output_profile="verbose", ) return response return self._run_record_tool(profile, runner, tool_name="代码块 Schema") def record_code_block_run( self, *, profile: str, app_key: str, record_id: int | str, code_block_field: str, view_id: str | None = None, role: int = 1, workflow_node_id: int | None = None, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, manual: bool = True, apply_writeback: bool = True, verify_writeback: bool = True, force_refresh_form: bool = False, output_profile: str = "normal", ) -> JSONObject: """执行记录相关逻辑。""" normalized_record_id = self._validate_app_and_record(app_key, record_id) normalized_output_profile = self._normalize_public_output_profile(output_profile) if role not in SUPPORTED_CODE_BLOCK_ROLES: raise_tool_error(QingflowApiError.config_error("role must be one of 1, 2, 3, or 5")) if role == 3 and (workflow_node_id is None or workflow_node_id <= 0): raise_tool_error(QingflowApiError.config_error("workflow_node_id is required when role=3")) if not code_block_field: raise_tool_error(QingflowApiError.config_error("code_block_field is required")) def runner(session_profile, context): warnings: list[JSONObject] = [] current_answers = self._load_record_answers_for_code_block( context, profile=profile, app_key=app_key, apply_id=normalized_record_id, view_id=view_id, role=role, audit_node_id=workflow_node_id, ) answer_index = _build_answer_backed_field_index(current_answers) schema_index: FieldIndex | None = None relation_schema = self._get_code_block_relation_schema_optional( profile, context, app_key, force_refresh=force_refresh_form, warnings=warnings, ) try: schema_index = self._get_field_index(profile, context, app_key, force_refresh=force_refresh_form) except QingflowApiError as exc: if not _is_optional_code_block_schema_error(exc): raise if not any(item.get("code") == "CODE_BLOCK_SCHEMA_UNAVAILABLE" for item in warnings): warnings.append( { "code": "CODE_BLOCK_SCHEMA_UNAVAILABLE", "message": "applicant form schema was not readable in this permission context; code-block execution will use record/task answers and skip schema-bound relation writeback.", "backend_code": exc.backend_code, "http_status": exc.http_status, "request_id": exc.request_id, } ) index = _merge_field_indexes(schema_index, answer_index) if schema_index is not None else answer_index code_block = self._resolve_code_block_field_for_run(code_block_field, index) if code_block.que_type != CODE_BLOCK_QUE_TYPE: raise_tool_error( QingflowApiError( category="config", message=f"field '{code_block.que_title}' is not a code-block field", backend_code="CODE_BLOCK_FIELD_REQUIRED", details={ "error_code": "CODE_BLOCK_FIELD_REQUIRED", "field": _field_ref_payload(code_block), "expected_que_type": CODE_BLOCK_QUE_TYPE, }, ) ) override_answers = ( self._resolve_answers( profile, context, app_key, answers=answers or [], fields=fields or {}, force_refresh_form=force_refresh_form, field_index_override=index, ) if answers or fields else [] ) merged_answers = self._merge_record_answers(current_answers, override_answers) if override_answers else current_answers key_que_values = self._answers_to_open_match_values(merged_answers, index, exclude_que_ids={code_block.que_id}) run_body: JSONObject = { "role": role, "manual": bool(manual), "applyId": normalized_record_id, "appKey": app_key, "queryQuestions": [{"queId": code_block.que_id, "ordinal": None}], "keyQueValues": key_que_values, } if workflow_node_id is not None: run_body["auditNodeId"] = workflow_node_id run_result = self.backend.request( "POST", context, f"/data/{app_key}/codeBlock/working", json_body=run_body, ) alias_results = _normalize_code_block_alias_results(run_result) configured_aliases = _extract_code_block_configured_aliases(code_block) relation_target_fields = _collect_code_block_relation_targets( _collect_question_relations(relation_schema), code_block_que_id=code_block.que_id, ) relation_errors: list[JSONObject] = [] relation_items: list[JSONObject] = [] calculated_answers: list[JSONObject] = [] relation_result: JSONObject | None = None relation_transport_error: JSONObject | None = None if relation_target_fields: relation_body: JSONObject = { "role": role, "manual": bool(manual), "applyId": normalized_record_id, "appKey": app_key, "queryQuestions": [{"queId": target["que_id"], "ordinal": None} for target in relation_target_fields], "keyQueValues": key_que_values, "codeBlockValues": [{"queId": code_block.que_id, "values": alias_results}], } if workflow_node_id is not None: relation_body["auditNodeId"] = workflow_node_id relation_route = "/data/que/actuator" try: relation_result = self.backend.request("POST", context, relation_route, json_body=relation_body) relation_items = _relation_result_items(relation_result) except QingflowApiError as exc: if exc.http_status == 404: relation_route = "/que/actuator" try: relation_result = self.backend.request("POST", context, relation_route, json_body=relation_body) relation_items = _relation_result_items(relation_result) except QingflowApiError as fallback_exc: relation_transport_error = _code_block_transport_error(fallback_exc) else: relation_transport_error = _code_block_transport_error(exc) if relation_transport_error is None and not relation_items: # Keep compatibility with legacy runtime deployments and lightweight test doubles # that still stub the older relation-calculation route only. relation_route = "/que/actuator" try: relation_result = self.backend.request("POST", context, relation_route, json_body=relation_body) relation_items = _relation_result_items(relation_result) relation_transport_error = None except QingflowApiError as exc: relation_transport_error = relation_transport_error or _code_block_transport_error(exc) relation_errors = _relation_result_errors(relation_items) calculated_answers = _relation_result_answers(relation_items) write_result: JSONObject | None = None write_error: JSONObject | None = None verification: JSONObject | None = None writeback_attempted = False writeback_applied = False status = "completed" ok = True if relation_transport_error is not None: status = "relation_failed" ok = False elif relation_errors: status = "relation_failed" ok = False elif not apply_writeback: status = "debug_completed" elif relation_target_fields and calculated_answers: write_body: JSONObject = {"role": role, "answers": calculated_answers} if workflow_node_id is not None: write_body["auditNodeId"] = workflow_node_id writeback_attempted = True try: write_result = cast( JSONObject, self.backend.request( "POST", context, f"/app/{app_key}/apply/{normalized_record_id}", json_body=write_body, ), ) writeback_applied = True except QingflowApiError as exc: write_error = _code_block_transport_error(exc) status = "writeback_failed" ok = False if writeback_applied and verify_writeback: verification = self._verify_code_block_writeback_result( context, profile=profile, app_key=app_key, apply_id=normalized_record_id, view_id=view_id, expected_answers=calculated_answers, index=index, role=role, audit_node_id=workflow_node_id, ) if not bool(verification.get("verified")): status = "verification_failed" ok = True warnings.append( { "code": "CODE_BLOCK_WRITEBACK_VERIFICATION_FAILED", "message": ( "code-block execution and writeback completed, but field-level readback " "could not verify the written values; do not treat this as writeback denial." ), } ) else: status = "no_writeback" response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "record_id": normalized_record_id, "apply_id": normalized_record_id, "status": status, "ok": ok, "write_executed": writeback_applied, "write_succeeded": writeback_applied, "safe_to_retry": not writeback_applied, "warnings": warnings, "code_block_field": _field_ref_payload(code_block), "execution": { "executed": True, "view_id": _normalize_optional_text(view_id), "role": role, "workflow_node_id": workflow_node_id, "manual": bool(manual), "apply_writeback": bool(apply_writeback), "result_count": len(alias_results), }, "outputs": { "configured_aliases": configured_aliases, "alias_results": alias_results, "alias_map": _build_alias_result_map(alias_results), }, "relation": { "target_fields": relation_target_fields, "result_item_count": len(relation_items), "calculated_answer_count": len(calculated_answers), "calculated_answers_preview": calculated_answers, "errors": relation_errors, "transport_error": relation_transport_error, }, "writeback": { "enabled": bool(apply_writeback), "attempted": writeback_attempted, "applied": writeback_applied, "skipped_reason": "apply_writeback_disabled" if not apply_writeback else None, "verify_writeback": verify_writeback, "write_verified": bool(verification.get("verified")) if verification is not None else None, "result": write_result, "error": write_error, "verification": verification, }, "resource": {"apply_id": normalized_record_id}, } if not ok: failure_error = relation_transport_error if relation_transport_error is not None else write_error failure_context = "relation" if relation_transport_error is not None else "writeback" response.update(_code_block_failure_fields(failure_error, context=failure_context)) if normalized_output_profile == "verbose": response["debug"] = { "run_body": run_body, "relation_route": relation_route if relation_target_fields else None, "relation_result": relation_result, "calculated_answers": calculated_answers, "merged_answers": merged_answers, "key_que_values": key_que_values, } return response return self._run_record_tool(profile, runner, tool_name="运行代码块") def _resolve_code_block_field_for_run(self, selector: str | int, index: FieldIndex) -> FormField: field_id = _coerce_count(selector) if field_id is not None and str(field_id) not in index.by_id: return FormField( que_id=field_id, que_title=str(field_id), que_type=CODE_BLOCK_QUE_TYPE, required=False, readonly=False, system=False, options=[], aliases=[], target_app_key=None, target_app_name_hint=None, member_select_scope_type=None, member_select_scope=None, dept_select_scope_type=None, dept_select_scope=None, raw={"queId": field_id, "queTitle": str(field_id), "queType": CODE_BLOCK_QUE_TYPE}, ) return self._resolve_field_selector(selector, index, location="code_block_field") def _load_record_answers_for_code_block( self, context, # type: ignore[no-untyped-def] *, profile: str, app_key: str, apply_id: int, view_id: str | None, role: int, audit_node_id: int | None, ) -> list[JSONObject]: """执行内部辅助逻辑。""" normalized_view_id = _normalize_optional_text(view_id) if normalized_view_id: try: resolved_view, _warnings = self._resolve_accessible_view_route( profile, context, app_key, view_id=normalized_view_id, list_type=None, view_key=None, view_name=None, allow_default=False, ) record, _used_list_type, _used_role = self._record_get_apply_detail( context, app_key=app_key, record_id=apply_id, resolved_view=resolved_view, audit_node_id=audit_node_id, ) answers = record.get("answers") if isinstance(record, dict) else None return [item for item in answers if isinstance(item, dict)] if isinstance(answers, list) else [] except QingflowApiError as exc: if not _is_optional_code_block_record_read_error(exc): raise last_error: QingflowApiError | None = None for list_type in self._INTERNAL_GET_LIST_TYPE_FALLBACKS: params: JSONObject = {"role": role, "listType": list_type} if audit_node_id is not None: params["auditNodeId"] = audit_node_id try: record = self.backend.request("GET", context, f"/app/{app_key}/apply/{apply_id}", params=params) answers = record.get("answers") if isinstance(record, dict) else None return [item for item in answers if isinstance(item, dict)] if isinstance(answers, list) else [] except QingflowApiError as exc: last_error = exc if _is_code_block_permission_error(exc): continue raise if last_error is not None: raise last_error raise_tool_error(QingflowApiError.config_error("record answers could not be loaded for code-block execution")) def _answers_to_open_match_values( self, answers: list[JSONObject], index: FieldIndex, *, exclude_que_ids: set[int] | None = None, ) -> list[JSONObject]: """执行内部辅助逻辑。""" values: list[JSONObject] = [] for answer in answers: if not isinstance(answer, dict): continue que_id = _coerce_count(answer.get("queId", answer.get("que_id"))) if que_id is not None and exclude_que_ids is not None and que_id in exclude_que_ids: continue open_match = self._answer_to_open_match_value(answer, index) if open_match is None: continue values.append(open_match) return values def _answer_to_open_match_value(self, answer: JSONObject, index: FieldIndex) -> JSONObject | None: """执行内部辅助逻辑。""" que_id = _coerce_count(answer.get("queId", answer.get("que_id"))) if que_id is None or que_id <= 0: return None field = index.by_id.get(str(que_id)) if field is None: return None ordinal = _coerce_count(answer.get("ordinal")) if field.que_type in SUBTABLE_QUE_TYPES: rows = answer.get("tableValues") row_values: list[list[JSONObject]] = [] subtable_index = self._subtable_field_index_optional(field) if isinstance(rows, list) and subtable_index is not None: for row in rows: if not isinstance(row, list): continue normalized_row: list[JSONObject] = [] for cell in row: if not isinstance(cell, dict): continue converted = self._answer_to_open_match_value(cell, subtable_index) if converted is not None: normalized_row.append(converted) row_values.append(normalized_row) payload: JSONObject = {"keyQueId": field.que_id, "ordinal": ordinal, "values": [], "tableValues": row_values} return payload return { "keyQueId": field.que_id, "ordinal": ordinal, "values": self._answer_values_to_code_block_values(answer, field), } def _answer_values_to_code_block_values(self, answer: JSONObject, field: FormField) -> list[str]: """执行内部辅助逻辑。""" if field.que_type in RELATION_QUE_TYPES: return _relation_ids_from_answer(answer) raw_values = answer.get("values") if not isinstance(raw_values, list): return [] normalized: list[str] = [] for item in raw_values: normalized_value = _normalize_code_block_value_item(item, field) if normalized_value is None: continue normalized.append(normalized_value) return normalized def _verify_code_block_writeback_result( self, context, # type: ignore[no-untyped-def] *, profile: str, app_key: str, apply_id: int, view_id: str | None, expected_answers: list[JSONObject], index: FieldIndex, role: int, audit_node_id: int | None, ) -> JSONObject: """执行内部辅助逻辑。""" if role == 1 and audit_node_id is None: return self._verify_record_write_result( context, app_key=app_key, apply_id=apply_id, normalized_answers=expected_answers, index=index, verify_list_type=DEFAULT_RECORD_LIST_TYPE, ) actual_answers = self._load_record_answers_for_code_block( context, profile=profile, app_key=app_key, apply_id=apply_id, view_id=view_id, role=role, audit_node_id=audit_node_id, ) actual_by_id = { que_id: item for item in actual_answers if isinstance(item, dict) and (que_id := _coerce_count(item.get("queId"))) is not None } missing_fields: list[JSONObject] = [] empty_fields: list[JSONObject] = [] count_mismatches: list[JSONObject] = [] for answer in expected_answers: que_id = _coerce_count(answer.get("queId")) if que_id is None or que_id <= 0: continue actual = actual_by_id.get(que_id) field = index.by_id.get(str(que_id)) field_payload = _field_ref_payload(field) if field is not None else {"que_id": que_id} if actual is None: missing_fields.append(field_payload) continue expected_rows = answer.get("tableValues") if isinstance(answer.get("tableValues"), list) else [] if expected_rows: actual_rows = actual.get("tableValues") if isinstance(actual.get("tableValues"), list) else [] self._verify_subtable_write_result( field=field, expected_rows=expected_rows, actual_rows=actual_rows, missing_fields=missing_fields, empty_fields=empty_fields, count_mismatches=count_mismatches, ) continue if field is not None and field.que_type in RELATION_QUE_TYPES: expected_relation_ids = _relation_ids_from_answer(answer) actual_relation_ids = _relation_ids_from_answer(actual) if expected_relation_ids and not actual_relation_ids: empty_fields.append(field_payload) continue if expected_relation_ids: actual_id_set = set(actual_relation_ids) missing_ids = [value for value in expected_relation_ids if value not in actual_id_set] if missing_ids: count_mismatches.append( { **field_payload, "expected_ids": expected_relation_ids, "actual_ids": actual_relation_ids, "missing_ids": missing_ids, } ) continue actual_values = actual.get("values") if isinstance(actual.get("values"), list) else [] if not actual_values: empty_fields.append(field_payload) continue expected_values = answer.get("values") if isinstance(answer.get("values"), list) else [] if expected_values and len(actual_values) < len(expected_values): count_mismatches.append( { **field_payload, "expected_count": len(expected_values), "actual_count": len(actual_values), } ) return { "verified": not missing_fields and not empty_fields and not count_mismatches, "verification_mode": "role_record_view", "field_level_verified": True, "missing_fields": missing_fields, "empty_fields": empty_fields, "count_mismatches": count_mismatches, } def _normalize_code_block_value_item(value: JSONValue, field: FormField) -> str | None: if field.que_type in SINGLE_SELECT_QUE_TYPES | MULTI_SELECT_QUE_TYPES: return _selector_numeric_or_text(value, ("optionId", "optId", "id"), allow_text=False) if field.que_type in MEMBER_QUE_TYPES: return _selector_numeric_or_text(value, ("id", "uid"), allow_text=False) if field.que_type in DEPARTMENT_QUE_TYPES: return _selector_numeric_or_text(value, ("id", "deptId"), allow_text=False) if field.que_type in ATTACHMENT_QUE_TYPES: return _selector_numeric_or_text(value, ("value", "url", "otherInfo", "name", "fileName"), allow_text=True) if isinstance(value, dict): scalar = value.get("value") return _stringify_json(scalar) if scalar is not None else None text = _normalize_optional_text(value) return text if text is not None else None def _code_block_transport_error(error: QingflowApiError) -> JSONObject: payload: JSONObject = { "category": error.category, "message": error.message, } 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 if error.details: payload["details"] = error.details return payload def _code_block_failure_fields(error: JSONObject | None, *, context: str) -> JSONObject: default_code = "CODE_BLOCK_RELATION_FAILED" if context == "relation" else "CODE_BLOCK_WRITEBACK_FAILED" permission_code = "CODE_BLOCK_RELATION_PERMISSION_DENIED" if context == "relation" else "CODE_BLOCK_WRITEBACK_PERMISSION_DENIED" payload: JSONObject = { "error_code": default_code, } if not isinstance(error, dict): return payload category = str(error.get("category") or "").strip().lower() http_status = backend_code_value_int(error.get("http_status")) if category == "auth" or http_status == 401 or message_looks_like_invalid_token(error.get("message")): payload["error_code"] = "AUTH_REQUIRED" elif backend_code_value_int(error.get("backend_code")) in {40002, 40027}: payload["error_code"] = permission_code for key in ("category", "backend_code", "http_status", "request_id"): if key in error: payload[key] = error.get(key) return payload def _selector_numeric_or_text(value: JSONValue, keys: tuple[str, ...], *, allow_text: bool) -> str | None: numeric = _coerce_count(value) if numeric is not None: return str(numeric) if not isinstance(value, dict): if not allow_text: return None text = _normalize_optional_text(value) return text if text is not None else None for key in keys: if key not in value: continue candidate = value.get(key) if candidate is None: continue numeric = _coerce_count(candidate) if numeric is not None: return str(numeric) if allow_text: text = _normalize_optional_text(candidate) if text is not None: return text return _normalize_optional_text(value.get("value")) if allow_text and isinstance(value.get("value"), (str, int, float)) else None def _normalize_code_block_alias_results(payload: JSONValue) -> list[JSONObject]: if not isinstance(payload, dict): return [] raw_results = payload.get("result") if not isinstance(raw_results, list): return [] results: list[JSONObject] = [] for item in raw_results: if not isinstance(item, dict): continue values = item.get("value") result: JSONObject = { "parentAliasId": _coerce_count(item.get("parentAliasId")), "parentAlias": _normalize_optional_text(item.get("parentAlias")), "aliasId": _coerce_count(item.get("aliasId")), "alias": _normalize_optional_text(item.get("alias")), "value": [_stringify_json(entry) for entry in values] if isinstance(values, list) else [], } results.append(result) return results def _extract_code_block_configured_aliases(field: FormField) -> list[JSONObject]: raw_config = field.raw.get("codeBlockConfig") if not isinstance(raw_config, dict): return [] raw_aliases = raw_config.get("resultAliasPath") if not isinstance(raw_aliases, list): return [] return [item for item in (_normalize_code_block_alias_item(alias) for alias in raw_aliases) if item is not None] def _normalize_code_block_alias_item(value: JSONValue) -> JSONObject | None: if not isinstance(value, dict): return None alias_name = _normalize_optional_text(value.get("aliasName", value.get("alias_name"))) alias_path = _normalize_optional_text(value.get("aliasPath", value.get("alias_path"))) if alias_name is None and alias_path is None: return None raw_sub_alias = value.get("subAlias", value.get("sub_alias")) sub_alias = ( [item for item in (_normalize_code_block_alias_item(entry) for entry in raw_sub_alias) if item is not None] if isinstance(raw_sub_alias, list) else [] ) return { "alias_id": _coerce_count(value.get("aliasId", value.get("alias_id"))), "alias_name": alias_name, "alias_path": alias_path, "alias_type": _coerce_count(value.get("aliasType", value.get("alias_type"))) or 1, "sub_alias": sub_alias, } def _build_alias_result_map(alias_results: list[JSONObject]) -> JSONObject: alias_map: JSONObject = {} for item in alias_results: alias = _normalize_optional_text(item.get("alias")) if alias is None: continue parent_alias = _normalize_optional_text(item.get("parentAlias")) key = f"{parent_alias}.{alias}" if parent_alias else alias values = item.get("value") alias_map[key] = values if isinstance(values, list) else [] return alias_map def _collect_code_block_relation_targets(question_relations: list[JSONObject], *, code_block_que_id: int) -> list[JSONObject]: targets: list[JSONObject] = [] seen: set[int] = set() for relation in question_relations: if _coerce_count(relation.get("relationType")) != CODE_BLOCK_RELATION_TYPE: continue alias_config = relation.get("aliasConfig") if isinstance(relation.get("aliasConfig"), dict) else {} relation_code_block_que_id = _coerce_count(relation.get("qlinkerQueId")) if relation_code_block_que_id is None: relation_code_block_que_id = _coerce_count(alias_config.get("queId")) if relation_code_block_que_id != code_block_que_id: continue target_id = _coerce_count( relation.get("queId", relation.get("targetQueId", relation.get("displayedQueId"))) ) if target_id is None or target_id in seen: continue seen.add(target_id) targets.append( { "que_id": target_id, "alias_id": _coerce_count(relation.get("aliasId")) or _coerce_count(alias_config.get("aliasId")), "alias_name": _normalize_optional_text(relation.get("qlinkerAlias")) or _normalize_optional_text(alias_config.get("qlinkerAlias")), "qlinker_que_id": relation_code_block_que_id, } ) return targets def _relation_result_items(payload: JSONValue) -> list[JSONObject]: if not isinstance(payload, dict): return [] result = payload.get("result") return [item for item in result if isinstance(item, dict)] if isinstance(result, list) else [] def _relation_result_answers(items: list[JSONObject]) -> list[JSONObject]: answers: list[JSONObject] = [] for item in items: raw_answers = item.get("answers") if not isinstance(raw_answers, list): continue for answer in raw_answers: if isinstance(answer, dict): answers.append(answer) return answers def _relation_result_errors(items: list[JSONObject]) -> list[JSONObject]: errors: list[JSONObject] = [] for item in items: message = _normalize_optional_text(item.get("errorMsg")) if message is None: continue errors.append( { "que_id": _coerce_count(item.get("queId")), "que_title": _normalize_optional_text(item.get("queTitle")), "ordinal": _coerce_count(item.get("ordinal")), "message": message, } ) return errors def _is_optional_code_block_record_read_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = _code_block_backend_code(error) return backend_code in {40002, 40027, 404} or error.http_status == 404 def _is_optional_code_block_schema_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return _code_block_backend_code(error) in _CODE_BLOCK_SCHEMA_PERMISSION_CODES or error.http_status == 404 def _is_code_block_permission_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return _code_block_backend_code(error) in {40002, 40027} def _code_block_backend_code(error: QingflowApiError) -> int | None: return backend_code_int(error)