from __future__ import annotations import json from copy import deepcopy from typing import Any from uuid import uuid4 from ..errors import QingflowApiError, backend_code_int, is_auth_like_error from ..tools.app_tools import AppTools from ..tools.navigation_tools import NavigationTools from ..tools.package_tools import PackageTools from ..tools.portal_tools import PortalTools from ..tools.qingbi_report_tools import QingbiReportTools from ..tools.record_tools import RecordTools from ..tools.role_tools import RoleTools from ..tools.view_tools import ViewTools from ..tools.workspace_tools import WorkspaceTools from .compiler import CompiledEntity, CompiledRole, CompiledSolution from .compiler.form_compiler import QUESTION_TYPE_MAP from .run_store import RunArtifactStore, fingerprint_payload from .spec_models import FieldType NAVIGATION_PLUGIN_ID = 45 GRID_COMPONENT_PLUGIN_ID = 34 BI_CHART_COMPONENT_TYPE = 9 PACKAGE_ITEM_TYPE_FORM = 1 class SolutionExecutor: def __init__( self, *, workspace_tools: WorkspaceTools, package_tools: PackageTools, role_tools: RoleTools, app_tools: AppTools, record_tools: RecordTools, view_tools: ViewTools, chart_tools: QingbiReportTools, portal_tools: PortalTools, navigation_tools: NavigationTools, workflow_tools: Any | None = None, ) -> None: self.workspace_tools = workspace_tools self.package_tools = package_tools self.role_tools = role_tools self.app_tools = app_tools self.record_tools = record_tools self.view_tools = view_tools self.chart_tools = chart_tools self.portal_tools = portal_tools self.navigation_tools = navigation_tools self.workflow_tools = workflow_tools def execute( self, *, profile: str, compiled: CompiledSolution, store: RunArtifactStore, publish: bool, mode: str, ) -> dict[str, Any]: self._current_store = store force_from_index = self._repair_start_index(compiled, store) if mode == "repair" else None for index, step in enumerate(compiled.execution_plan.steps): force = force_from_index is not None and index >= force_from_index if not store.should_run(step.step_name, force=force): continue debug_context = self._step_debug_context(compiled, step.step_name) try: store.record_step_started(step.step_name, store.data["request_fingerprint"], debug_context=debug_context) self._execute_step(profile=profile, compiled=compiled, store=store, step_name=step.step_name, publish=publish) store.record_step_completed(step.step_name, debug_context=debug_context) except Exception as exc: # noqa: BLE001 store.record_step_failed(step.step_name, str(exc), debug_context=debug_context) return store.summary() final_status = "partial_success" if _artifacts_have_post_write_readback_pending(store.data.get("artifacts", {})) else "success" store.mark_finished(status=final_status) return store.summary() def _repair_start_index(self, compiled: CompiledSolution, store: RunArtifactStore) -> int: for index, step in enumerate(compiled.execution_plan.steps): if store.get_step_status(step.step_name) != "completed": return index return len(compiled.execution_plan.steps) def _step_debug_context(self, compiled: CompiledSolution, step_name: str) -> dict[str, Any]: context: dict[str, Any] = {"step_name": step_name} if step_name == "package.create": context["resource"] = "package" context["package_name"] = compiled.normalized_spec.package.name or compiled.normalized_spec.solution_name return context if step_name.startswith("package.attach."): entity = self._entity_from_step(compiled, step_name) context.update( { "resource": "package_attach", "entity_id": entity.entity_id, "display_name": entity.display_name, "package_tag_id": self._current_store.get_artifact("package", "tag_id") if hasattr(self, "_current_store") else None, } ) return context if step_name == "portal.create": context["resource"] = "portal" context["portal_name"] = compiled.normalized_spec.portal.name context["section_count"] = len(compiled.normalized_spec.portal.sections) return context if step_name == "navigation.create": context["resource"] = "navigation" context["item_count"] = len(compiled.normalized_spec.navigation.items) return context if ".create." in step_name and step_name.startswith("role.create."): role = self._role_from_step(compiled, step_name) context["resource"] = "role" context["role_id"] = role.role_id context["role_name"] = role.name return context if any(step_name.startswith(prefix) for prefix in ("app.create.", "form.base.", "form.relations.", "workflow.", "views.", "charts.", "seed_data.", "publish.form.", "publish.workflow.", "publish.app.")): entity = self._entity_from_step(compiled, step_name) context.update( { "resource": "entity", "entity_id": entity.entity_id, "display_name": entity.display_name, "workflow_action_count": len((entity.workflow_plan or {}).get("actions", [])), "view_count": len(entity.view_plans), "chart_count": len(entity.chart_plans), "sample_record_count": len(entity.sample_records), } ) return context return context def _execute_step(self, *, profile: str, compiled: CompiledSolution, store: RunArtifactStore, step_name: str, publish: bool) -> None: if step_name == "package.create": self._create_package(profile, compiled, store) return if step_name.startswith("package.attach."): entity = self._entity_from_step(compiled, step_name) self._attach_app_to_package(profile, entity, store) return if step_name.startswith("role.create."): role = self._role_from_step(compiled, step_name) self._create_role(profile, role, store) return if step_name.startswith("app.create."): entity = self._entity_from_step(compiled, step_name) self._create_app(profile, entity, store) return if step_name.startswith("form.base."): entity = self._entity_from_step(compiled, step_name) self._update_form_base(profile, entity, store) return if step_name.startswith("form.relations."): entity = self._entity_from_step(compiled, step_name) self._update_form_relations(profile, entity, store) return if step_name.startswith("workflow."): entity = self._entity_from_step(compiled, step_name) self._build_workflow(profile, entity, store) return if step_name.startswith("views."): entity = self._entity_from_step(compiled, step_name) self._build_views(profile, entity, store) return if step_name.startswith("charts."): entity = self._entity_from_step(compiled, step_name) self._build_charts(profile, entity, store) return if step_name.startswith("seed_data."): entity = self._entity_from_step(compiled, step_name) self._seed_records(profile, entity, store) return if step_name == "portal.create": self._build_portal(profile, compiled, store) return if step_name == "navigation.create": self._build_navigation(profile, compiled, store) return if ( step_name.startswith("publish.form.") or step_name.startswith("publish.workflow.") or step_name.startswith("publish.app.") ) and publish and compiled.normalized_spec.publish_policy.apps: entity = self._entity_from_step(compiled, step_name) app_key = self._get_app_key(store, entity.entity_id) edit_version_no = self._get_edit_version_no(store, entity.entity_id) self.app_tools.app_publish( profile=profile, app_key=app_key, payload=self._publish_payload(store, entity.entity_id), ) if edit_version_no is not None: self.app_tools.app_edit_finished( profile=profile, app_key=app_key, payload={"editVersionNo": int(edit_version_no)}, ) return if step_name == "publish.portal" and publish and compiled.normalized_spec.publish_policy.portal: dash_key = store.get_artifact("portal", "dash_key") if dash_key: self.portal_tools.portal_publish(profile=profile, dash_key=dash_key) store.set_artifact( "portal", "publish", {"published": True, "write_executed": True, "safe_to_retry": False}, ) self._refresh_portal_artifact(profile=profile, store=store, being_draft=False, artifact_key="published_result") return if step_name == "publish.navigation" and publish and compiled.normalized_spec.publish_policy.navigation: if store.get_artifact("navigation", "skipped"): return status = self.navigation_tools.navigation_get_status(profile=profile) navigation_status = (status.get("result") or {}) if isinstance(status.get("result"), dict) else {} navigation_id = navigation_status.get("navigationId") or navigation_status.get("id") if navigation_id: self.navigation_tools.navigation_publish(profile=profile, navigation_id=navigation_id) return def _create_package(self, profile: str, compiled: CompiledSolution, store: RunArtifactStore) -> None: if compiled.package_payload is None: return existing_package = store.get_artifact("package", "result", {}) or {} existing_tag_id = store.get_artifact("package", "tag_id") if isinstance(existing_tag_id, int) and existing_tag_id > 0: return if isinstance(existing_package, dict) and existing_package.get("tagId"): return result = self.package_tools.package_create(profile=profile, payload=compiled.package_payload) package_result = result.get("result") or {} tag_id = package_result.get("tagId") or package_result.get("id") or package_result.get("tag_id") store.set_artifact("package", "result", result) if tag_id is not None: store.set_artifact("package", "tag_id", tag_id) def _create_role(self, profile: str, role: CompiledRole, store: RunArtifactStore) -> None: existing = store.get_artifact("roles", role.role_id, {}) or {} existing_role_id = existing.get("role_id") if existing_role_id: return try: page = self.role_tools.role_search(profile=profile, keyword=role.name, page_num=1, page_size=50).get("page") or {} role_list = page.get("list") if isinstance(page, dict) else [] except Exception as exc: # noqa: BLE001 api_error = _coerce_qingflow_error(exc) if api_error is None or not _is_permission_restricted_error(api_error): raise role_list = [] matched_role = next( ( item for item in (role_list or []) if isinstance(item, dict) and item.get("roleName") == role.name and item.get("roleId") ), None, ) if matched_role is None: payload = deepcopy(role.payload) if not payload.get("users"): session_profile = self.role_tools.sessions.get_profile(profile) if session_profile is not None: payload["users"] = [session_profile.uid] result = self.role_tools.role_create(profile=profile, payload=payload) role_result = result.get("result") or {} role_id = role_result.get("roleId") or role_result.get("id") store.set_artifact( "roles", role.role_id, {"role_id": role_id, "result": result, "role_name": role.name, "role_icon": payload.get("roleIcon")}, ) return store.set_artifact( "roles", role.role_id, { "role_id": matched_role.get("roleId"), "result": {"result": matched_role}, "role_name": role.name, "role_icon": matched_role.get("roleIcon"), "reused": True, }, ) def _create_app(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: existing_artifact = store.get_artifact("apps", entity.entity_id, {}) or {} existing_app_key = existing_artifact.get("app_key") if isinstance(existing_app_key, str) and existing_app_key: next_artifact = deepcopy(existing_artifact) next_artifact["reused"] = True next_artifact.setdefault("target_mode", "update") store.set_artifact("apps", entity.entity_id, next_artifact) return payload = self._resolve_app_payload(entity.app_create_payload, store) result = self.app_tools.app_create(profile=profile, payload=payload) app_result = result.get("result") or {} app_key = None if isinstance(app_result, dict): app_key = app_result.get("appKey") if app_key is None and isinstance(app_result.get("appKeys"), list) and app_result["appKeys"]: app_key = app_result["appKeys"][0] app_artifact = {"app_key": app_key, "result": result} store.set_artifact("apps", entity.entity_id, app_artifact) if app_key: try: version_result = self.app_tools.app_get_edit_version_no(profile=profile, app_key=app_key).get("result") or {} edit_version_no = version_result.get("editVersionNo") or version_result.get("versionNo") if edit_version_no is not None: app_artifact["edit_version_no"] = int(edit_version_no) store.set_artifact("apps", entity.entity_id, app_artifact) except Exception: store.set_artifact("apps", entity.entity_id, app_artifact) raise def _attach_app_to_package(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: tag_id = store.get_artifact("package", "tag_id") if not isinstance(tag_id, int) or tag_id <= 0: return app_artifact = store.get_artifact("apps", entity.entity_id, {}) or {} app_key = app_artifact.get("app_key") if not isinstance(app_key, str) or not app_key: raise ValueError(f"missing app_key for package attach on entity '{entity.entity_id}'") try: package_detail = self.package_tools.package_get(profile=profile, tag_id=tag_id, include_raw=True) except Exception as exc: # noqa: BLE001 api_error = _coerce_qingflow_error(exc) if api_error is None or not _is_permission_restricted_error(api_error): raise raise _required_state_read_blocked_error( resource="package_attach", message=f"package attach requires readable package state before sorting items for tag '{tag_id}'", error=api_error, details={"tag_id": tag_id, "app_key": app_key}, ) from exc package_result = package_detail.get("result") if isinstance(package_detail.get("result"), dict) else {} tag_items = [deepcopy(item) for item in package_result.get("tagItems", []) if isinstance(item, dict)] if any(_package_item_app_key(item) == app_key for item in tag_items): self._record_package_attachment(store, entity.entity_id, app_artifact, tag_id=tag_id, attached=True, reused=True) return item = { "itemType": PACKAGE_ITEM_TYPE_FORM, "appKey": app_key, "title": entity.display_name, "iconUrl": entity.app_create_payload.get("appIcon"), } insert_at = _resolve_package_item_insert_index(tag_items, entity.app_create_payload.get("ordinal")) updated_items = list(tag_items) updated_items.insert(insert_at, item) self.package_tools.package_sort_items(profile=profile, tag_id=tag_id, tag_items=updated_items) try: verified_detail = _verify_package_attachment(self.package_tools, profile=profile, tag_id=tag_id, app_key=app_key) except Exception as exc: # noqa: BLE001 api_error = _coerce_qingflow_error(exc) if api_error is None or not _is_permission_restricted_error(api_error): raise store.set_artifact( "package", "attachment_readback", _post_write_readback_artifact( resource="package_attach", target={"tag_id": tag_id, "app_key": app_key}, error=api_error, ), ) self._record_package_attachment( store, entity.entity_id, app_artifact, tag_id=tag_id, attached=True, reused=False, readback_status="unavailable", readback_verified=False, ) return verified_result = verified_detail.get("result") if isinstance(verified_detail.get("result"), dict) else {} verified_items = [deepcopy(existing) for existing in verified_result.get("tagItems", []) if isinstance(existing, dict)] if not any(_package_item_app_key(existing) == app_key for existing in verified_items): raise QingflowApiError( category="runtime", message=f"failed to attach app '{app_key}' to package '{tag_id}'", details={"tag_id": tag_id, "app_key": app_key}, ) store.set_artifact("package", "result", verified_detail) store.set_artifact("package", "tag_id", tag_id) self._record_package_attachment(store, entity.entity_id, app_artifact, tag_id=tag_id, attached=True, reused=False) def _record_package_attachment( self, store: RunArtifactStore, entity_id: str, app_artifact: dict[str, Any], *, tag_id: int, attached: bool, reused: bool, readback_status: str = "verified", readback_verified: bool = True, ) -> None: next_artifact = deepcopy(app_artifact) next_artifact["package_attachment"] = { "tag_id": tag_id, "attached": attached, "reused": reused, "readback_status": readback_status, "readback_verified": readback_verified, "write_executed": not reused, "safe_to_retry": reused or not attached, } store.set_artifact("apps", entity_id, next_artifact) def _update_form_base(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: app_key = self._get_app_key(store, entity.entity_id) payload = deepcopy(entity.form_base_payload) edit_version_no = self._ensure_edit_version(profile, entity.entity_id, store, app_key=app_key) if edit_version_no is not None: payload["editVersionNo"] = int(edit_version_no) try: self.app_tools.app_update_form_schema(profile=profile, app_key=app_key, payload=payload) except Exception as exc: # noqa: BLE001 raise self._form_stage_error( exc, stage_name="form.base", entity=entity, app_key=app_key, payload=payload, ) from exc self._refresh_field_map(profile, entity, store) def _update_form_relations(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: if entity.form_relation_payload is None: return app_key = self._get_app_key(store, entity.entity_id) payload = self._resolve_relation_payload(entity, store) edit_version_no = self._ensure_edit_version(profile, entity.entity_id, store, app_key=app_key) if edit_version_no is not None: payload["editVersionNo"] = int(edit_version_no) try: self.app_tools.app_update_form_schema(profile=profile, app_key=app_key, payload=payload) except Exception as exc: # noqa: BLE001 raise self._form_stage_error( exc, stage_name="form.relations", entity=entity, app_key=app_key, payload=payload, ) from exc self._refresh_field_map(profile, entity, store) def _build_workflow(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: if entity.workflow_plan is None: return raise RuntimeError( "Legacy auditNode workflow execution was removed. " "Pass {app_key, spec} to solution_build_flow or use qingflow builder flow apply." ) def _build_views(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: app_key = self._get_app_key(store, entity.entity_id) field_meta = store.get_artifact("field_maps", entity.entity_id, {}) field_map = field_meta.get("by_field_id", {}) schema = field_meta.get("schema", {}) created_view_keys: list[str] = [] for view_plan in entity.view_plans: create_payload = self._resolve_view_payload( view_plan["create_payload"], app_key=app_key, field_map=field_map, schema=schema, group_by_field_id=view_plan["group_by_field_id"], ) result = self.view_tools.view_create(profile=profile, payload=create_payload) view_result = result.get("result") view_key = ((view_result or {}) if isinstance(view_result, dict) else {}).get("viewgraphKey") if not view_key and isinstance(view_result, str): view_key = view_result if not view_key: continue created_view_keys.append(view_key) store.set_artifact("views", f"{entity.entity_id}:{view_plan['view_id']}", {"viewgraph_key": view_key, "result": result}) if view_plan["column_widths"]: que_width_info_list = [ {"queId": que_id, "queWidth": width} for field_id, width in view_plan["column_widths"].items() if (que_id := field_map.get(field_id)) is not None ] if que_width_info_list: self.view_tools.view_set_column_width( profile=profile, app_key=app_key, payload={ "viewgraphKey": view_key, "beingCustomViewgraph": True, "beingReferenceQueWidth": False, "queWidthInfoList": que_width_info_list, }, ) if view_plan["member_config"]: self.view_tools.view_update_member_config(profile=profile, viewgraph_key=view_key, payload=view_plan["member_config"]) if view_plan["apply_config"]: self.view_tools.view_update_apply_config(profile=profile, viewgraph_key=view_key, payload=view_plan["apply_config"]) if view_plan["type"] == "board" and view_plan["group_by_field_id"] and view_plan["config"]: que_id = field_map.get(view_plan["group_by_field_id"]) payload = deepcopy(view_plan["config"]) if que_id is not None and "queId" not in payload: payload["queId"] = que_id self.view_tools.view_board_set_lane_config(profile=profile, viewgraph_key=view_key, payload=payload) if view_plan["type"] == "gantt": if "being_auto_calibration" in view_plan["config"]: self.view_tools.view_gantt_switch_auto_calibration( profile=profile, viewgraph_key=view_key, payload={"userAutoCalibration": view_plan["config"]["being_auto_calibration"]}, ) if created_view_keys: self.view_tools.view_reorder(profile=profile, app_key=app_key, payload=self._resolve_view_reorder_payload(profile, app_key, created_view_keys)) def _build_charts(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: app_key = self._get_app_key(store, entity.entity_id) field_map = self._get_field_map(store, entity.entity_id) bi_field_map = store.get_artifact("field_maps", entity.entity_id, {}).get("bi_by_field_id", {}) qingbi_fields = self.chart_tools.qingbi_report_list_fields(profile=profile, app_key=app_key).get("items") or [] qingbi_fields_by_id = { field.get("fieldId"): field for field in qingbi_fields if isinstance(field, dict) and field.get("fieldId") } existing_charts = self.chart_tools.qingbi_report_list(profile=profile, app_key=app_key).get("items") or [] existing_chart_keys_by_name = { chart.get("chartName"): chart.get("chartId") or chart.get("chartKey") for chart in existing_charts if isinstance(chart, dict) and chart.get("chartName") and (chart.get("chartId") or chart.get("chartKey")) } created_chart_keys: list[str] = [] for chart_plan in entity.chart_plans: chart_artifact_key = f"{entity.entity_id}:{chart_plan['chart_id']}" chart_artifact = store.get_artifact("charts", chart_artifact_key, {}) or {} chart_key = ( chart_artifact.get("bi_chart_id") or chart_artifact.get("chart_key") or existing_chart_keys_by_name.get(chart_plan["name"]) ) result = chart_artifact.get("result") if not chart_key: create_payload = deepcopy(chart_plan["create_payload"]) create_payload["dataSourceId"] = app_key create_payload["chartId"] = create_payload.get("chartId") if create_payload.get("chartId") != "__BI_CHART_ID__" else f"mcp_{uuid4().hex[:16]}" result = self.chart_tools.qingbi_report_create(profile=profile, payload=create_payload) chart_key = ((result.get("result") or {}) if isinstance(result.get("result"), dict) else {}).get("chartId") if not chart_key: continue created_chart_keys.append(chart_key) config_payload = self._resolve_qingbi_chart_config_payload( chart_plan["config_payload"], entity=entity, field_map=field_map, bi_field_map=bi_field_map, qingbi_fields_by_id=qingbi_fields_by_id, app_key=app_key, ) self.chart_tools.qingbi_report_update_config(profile=profile, chart_id=chart_key, payload=config_payload) chart_info = { "chartName": chart_plan["name"], "chartType": chart_plan["chart_type"], "dataSourceId": app_key, "dataSourceType": "qingflow", } chart_info.update(config_payload) chart_info.setdefault("chartName", chart_plan["name"]) chart_info["biChartId"] = chart_key store.set_artifact( "charts", chart_artifact_key, { "bi_chart_id": chart_key, "chart_key": chart_key, "result": result, "chart_info": chart_info, }, ) if created_chart_keys: self.chart_tools.qingbi_report_reorder(profile=profile, app_key=app_key, chart_ids=created_chart_keys) def _build_portal(self, profile: str, compiled: CompiledSolution, store: RunArtifactStore) -> None: if compiled.portal_plan is None: return if _portal_plan_has_source_type(compiled.portal_plan, "grid"): self.workspace_tools.workspace_set_plugin_status( profile=profile, plugin_id=GRID_COMPONENT_PLUGIN_ID, being_installed=True, ) store.set_artifact( "portal", "grid_plugin_install", { "plugin_id": GRID_COMPONENT_PLUGIN_ID, "being_installed": True, }, ) portal_artifact = store.get_artifact("portal", "result", {}) or {} dash_key = store.get_artifact("portal", "dash_key") result = portal_artifact if not dash_key: create_payload = self._resolve_portal_payload(compiled.portal_plan["create_payload"], store, base_payload=None) result = self.portal_tools.portal_create(profile=profile, payload=create_payload) portal_result = result.get("result") or {} dash_key = portal_result.get("dashKey") store.set_artifact("portal", "result", result) if dash_key: store.set_artifact("portal", "dash_key", dash_key) if dash_key: try: base_payload = self.portal_tools.portal_get(profile=profile, dash_key=dash_key, being_draft=True).get("result") or {} except Exception as exc: # noqa: BLE001 api_error = _coerce_qingflow_error(exc) if api_error is None or not _is_permission_restricted_error(api_error): raise if not result: raise _required_state_read_blocked_error( resource="portal", message=f"portal update requires readable draft state for dash '{dash_key}'", error=api_error, details={"dash_key": dash_key}, ) from exc base_payload = {} store.set_artifact( "portal", "draft_readback_before_update", _post_write_readback_artifact( resource="portal", target={"dash_key": dash_key, "phase": "created_portal_draft_readback"}, error=api_error, ), ) update_payload = self._resolve_portal_payload(compiled.portal_plan["update_payload"], store, base_payload=base_payload) self.portal_tools.portal_update(profile=profile, dash_key=dash_key, payload=update_payload) self._refresh_portal_artifact(profile=profile, store=store, being_draft=True, artifact_key="draft_result") def _build_navigation(self, profile: str, compiled: CompiledSolution, store: RunArtifactStore) -> None: try: created_items = self._create_navigation_items(profile, compiled, store) except Exception as exc: # noqa: BLE001 api_error = _coerce_qingflow_error(exc) if api_error is None or not _is_navigation_plugin_unavailable(api_error): raise self.workspace_tools.workspace_set_plugin_status( profile=profile, plugin_id=NAVIGATION_PLUGIN_ID, being_installed=True, ) created_items = self._create_navigation_items(profile, compiled, store) store.set_artifact( "navigation", "plugin_install", { "plugin_id": NAVIGATION_PLUGIN_ID, "being_installed": True, }, ) store.set_artifact("navigation", "items", created_items) def _refresh_portal_artifact( self, *, profile: str, store: RunArtifactStore, being_draft: bool, artifact_key: str, ) -> None: dash_key = store.get_artifact("portal", "dash_key") if not dash_key: return try: result = self.portal_tools.portal_get(profile=profile, dash_key=dash_key, being_draft=being_draft) except Exception as exc: # noqa: BLE001 api_error = _coerce_qingflow_error(exc) if api_error is None: raise store.set_artifact( "portal", f"{artifact_key}_readback", _post_write_readback_artifact( resource="portal", target={"dash_key": dash_key, "being_draft": being_draft, "artifact_key": artifact_key}, error=api_error, ), ) return store.set_artifact("portal", artifact_key, result) store.set_artifact("portal", "result", result) def _create_navigation_items( self, profile: str, compiled: CompiledSolution, store: RunArtifactStore, ) -> list[dict[str, Any]]: created_items: list[dict[str, Any]] = [] for ordinal, item in enumerate(compiled.navigation_plan, start=1): result = self.navigation_tools.navigation_create( profile=profile, payload=self._resolve_navigation_payload(item["payload"], store, ordinal), ) created = {"item_id": item["item_id"], "result": result} created_items.append(created) for child_ordinal, child in enumerate(item["children"], start=1): child_payload = self._resolve_navigation_payload(child["payload"], store, child_ordinal) child_payload["parentNavigationItemId"] = ((result.get("result") or {}) if isinstance(result.get("result"), dict) else {}).get("navigationItemId") created_items.append( { "item_id": child["item_id"], "result": self.navigation_tools.navigation_create(profile=profile, payload=child_payload), } ) return created_items def _refresh_field_map(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: app_key = self._get_app_key(store, entity.entity_id) schema = self.app_tools.app_get_form_schema( profile=profile, app_key=app_key, form_type=1, being_draft=True, being_apply=None, audit_node_id=None, include_raw=True, ) result = schema.get("result") or {} field_map = extract_field_map(result) label_to_field_id = {label: field_id for field_id, label in entity.field_labels.items()} mapped = {field_id: field_map[label] for label, field_id in label_to_field_id.items() if label in field_map} current_edit_version_no = self._get_edit_version_no(store, entity.entity_id) response_edit_version_no = result.get("editVersionNo") store.set_artifact( "field_maps", entity.entity_id, { "by_field_id": mapped, "bi_by_field_id": {field_id: f"{app_key}:{que_id}" for field_id, que_id in mapped.items()}, "by_label": field_map, "edit_version_no": int(response_edit_version_no or current_edit_version_no or 1), "schema": result, }, ) def _resolve_app_payload(self, payload: dict[str, Any], store: RunArtifactStore) -> dict[str, Any]: data = deepcopy(payload) if "tagIds" in data: tag_id = store.get_artifact("package", "tag_id") data["tagIds"] = [tag_id] if tag_id is not None else [] return data def _resolve_relation_payload(self, entity: CompiledEntity, store: RunArtifactStore) -> dict[str, Any]: payload = deepcopy(entity.form_relation_payload or {}) for line in payload.get("formQues", []): self._resolve_reference_questions(line, entity, store) payload["questionRelations"] = [] return payload def _resolve_reference_questions(self, questions: list[dict[str, Any]], entity: CompiledEntity, store: RunArtifactStore) -> None: for question in questions: if question.get("queType") == 24: for inner_row in question.get("innerQuestions", []): self._resolve_reference_questions(inner_row, entity, store) continue if question.get("queType") != 25: continue target_entity_id = self._target_entity_id_from_label(entity, question["queTitle"]) if not target_entity_id: continue target_field_label = self._target_field_label(entity, question["queTitle"], store) target_field_id = self._target_field_id_from_label(entity, question["queTitle"]) target_meta = store.get_artifact("field_maps", target_entity_id, {}) target_app = store.get_artifact("apps", target_entity_id, {}) target_label_map = target_meta.get("by_label", {}) target_que_id = target_label_map.get(target_field_label, 0) reference_config = deepcopy(question.get("referenceConfig") or {}) reference_config["referAppKey"] = target_app.get("app_key") reference_config["referQueId"] = target_que_id for ordinal, refer_question in enumerate(reference_config.get("referQuestions", []), start=1): refer_field_id = target_field_id if ordinal == 1 else refer_question.get("_field_id") or target_field_id refer_label = self._field_label_from_id(target_entity_id, refer_field_id, store) refer_que_id = target_meta.get("by_field_id", {}).get(refer_field_id) or target_label_map.get(refer_label, 0) refer_field_spec = self._field_spec_from_store(store, target_entity_id, refer_field_id) refer_field_type = refer_field_spec.get("type") refer_question["queId"] = refer_que_id refer_question["queTitle"] = refer_label if refer_field_type: refer_question["queType"] = str(QUESTION_TYPE_MAP[FieldType(refer_field_type)]) refer_question["ordinal"] = ordinal refer_question.pop("_field_id", None) auth_ques = [] for auth_que in reference_config.get("referAuthQues", []): resolved = deepcopy(auth_que) refer_field_id = resolved.pop("_field_id", None) or target_field_id refer_que_id = target_meta.get("by_field_id", {}).get(refer_field_id) if refer_que_id is None: continue resolved["queId"] = refer_que_id resolved["queAuth"] = int(resolved.get("queAuth", 3)) auth_ques.append(resolved) if not auth_ques: fallback_que_id = target_meta.get("by_field_id", {}).get(target_field_id) if fallback_que_id is not None: auth_ques.append({"queId": fallback_que_id, "queAuth": 3}) reference_config["referAuthQues"] = auth_ques reference_config["fieldNameShow"] = bool(reference_config.get("fieldNameShow", True)) fill_rules = [] for fill_rule in reference_config.get("referFillRules", []): resolved = deepcopy(fill_rule) current_field_id = resolved.pop("field_id", None) target_fill_field_id = resolved.pop("target_field_id", None) if not current_field_id or not target_fill_field_id: continue resolved["queId"] = store.get_artifact("field_maps", entity.entity_id, {}).get("by_field_id", {}).get(current_field_id) resolved["relatedQueId"] = target_meta.get("by_field_id", {}).get(target_fill_field_id) resolved["queTitle"] = self._field_label_from_id(entity.entity_id, current_field_id, store) resolved["relatedQueTitle"] = self._field_label_from_id(target_entity_id, target_fill_field_id, store) resolved["currentQuoteQueId"] = question.get("queId") or question.get("queTempId") if resolved["queId"] and resolved["relatedQueId"]: fill_rules.append(resolved) reference_config["referFillRules"] = fill_rules reference_config.pop("_targetFieldId", None) reference_config.pop("_targetEntityId", None) question["referenceConfig"] = reference_config def _resolve_workflow_payload(self, payload: dict[str, Any], node_artifacts: dict[str, int]) -> dict[str, Any]: data = deepcopy(payload) prev_ref = data.pop("prevNodeRef", None) if prev_ref: data["prevId"] = node_artifacts.get(prev_ref, 0) audit_ref = data.pop("auditNodeRef", None) if audit_ref: data["auditNodeId"] = node_artifacts.get(audit_ref, 0) audit_user_infos = data.get("auditUserInfos") if isinstance(audit_user_infos, dict): role_refs = audit_user_infos.pop("role_refs", []) if role_refs: roles = [] for role_ref in role_refs: role_artifact = self._current_store.get_artifact("roles", role_ref, {}) if hasattr(self, "_current_store") else {} role_id = role_artifact.get("role_id") if role_id is None: raise QingflowApiError.config_error(f"workflow role '{role_ref}' has not been created") roles.append( { "roleId": role_id, "roleName": role_artifact.get("role_name") or role_ref, "roleIcon": role_artifact.get("role_icon") or "ex-user-outlined", "beingFrontendConfig": True, } ) audit_user_infos["role"] = roles if data.get("auditUserInfos") is None: data.pop("auditUserInfos", None) return data def _resolve_view_payload( self, payload: dict[str, Any], *, app_key: str, field_map: dict[str, int], schema: dict[str, Any], group_by_field_id: str | None, ) -> dict[str, Any]: data = deepcopy(payload) data["appKey"] = app_key visible_que_ids = [field_map[field_id] for field_id in data.get("viewgraphQueIds", []) if field_id in field_map] data["viewgraphQueIds"] = visible_que_ids data.setdefault("defaultRowHigh", "compact") data.setdefault("asosChartVisible", False) data.setdefault("beingNeedPass", False) data.setdefault("beingAuditRecordVisible", True) data.setdefault("beingQrobotRecordVisible", False) data.setdefault("beingPrintStatus", False) data.setdefault("beingDefaultPrintTplStatus", False) data.setdefault("beingCommentStatus", False) data.setdefault("beingWorkflowNodeFutureListVisible", True) data.setdefault("dataPermissionType", "CUSTOM") data.setdefault("dataScope", "ALL") data.setdefault("needPass", False) data.setdefault("viewgraphPass", "") data.setdefault("beingImageAdaption", False) data.setdefault("clippingMode", "default") data.setdefault("frontCoverQueId", None) data.setdefault("printTpls", []) data.setdefault("usages", []) data.setdefault("asosChartConfig", {"limitType": 1, "asosChartIdList": []}) data.setdefault("viewgraphGanttConfigVO", None) data.setdefault("viewgraphHierarchyConfigVO", None) data.setdefault("viewgraphLimitFormula", "") data.setdefault("buttonConfigDTOList", []) if not data.get("viewgraphQuestions"): data["viewgraphQuestions"] = _build_viewgraph_questions(schema, visible_que_ids) if not data.get("viewgraphSorts"): data["viewgraphSorts"] = [{"queId": 0, "beingSortAscend": True, "queType": 8}] if data.get("viewgraphType") in {"cardView", "boardView", "ganttView", "hierarchyView"}: data["titleQue"] = visible_que_ids[0] if visible_que_ids else None else: data.setdefault("titleQue", None) if data.get("viewgraphType") == "boardView": data["groupQueId"] = field_map.get(group_by_field_id) if group_by_field_id else 1 if data.get("viewgraphType") == "ganttView": gantt_payload = self._resolve_gantt_payload(payload, field_map) if gantt_payload.get("viewgraphGanttConfigVO"): data["viewgraphGanttConfigVO"] = gantt_payload["viewgraphGanttConfigVO"] return data def _resolve_gantt_payload(self, config: dict[str, Any], field_map: dict[str, int]) -> dict[str, Any]: start_field_id = config.get("start_field_id") end_field_id = config.get("end_field_id") title_field_id = config.get("title_field_id") if not any((start_field_id, end_field_id, title_field_id)): return {} gantt_config = { "titleQueId": field_map.get(title_field_id), "startTimeQueId": field_map.get(start_field_id), "endTimeQueId": field_map.get(end_field_id), "defaultTimeDimension": "week", "ganttGroupVOList": [], "ganttDependencyVO": { "dependencyQueId": None, "predecessorTaskQueId": None, "startEndOptionId": None, "startStartOptionId": None, "endEndOptionId": None, "endStartOptionId": None, }, "ganttAutoCalibrationVO": { "autoCalibrationRuleVO": { "startStartBegin": False, "startEndBegin": False, "startEndFinish": False, "endStartBegin": True, "endStartFinish": True, "endEndFinish": False, }, "beingAutoCalibration": False, "userAutoCalibration": False, }, } return { "viewgraphGanttConfigVO": gantt_config, **{key: value for key, value in config.items() if key not in {"start_field_id", "end_field_id", "title_field_id"}}, } def _resolve_qingbi_chart_config_payload( self, payload: dict[str, Any], *, entity: CompiledEntity, field_map: dict[str, int], bi_field_map: dict[str, str], qingbi_fields_by_id: dict[str, dict[str, Any]], app_key: str, ) -> dict[str, Any]: data = { "chartName": payload.get("chartName"), "chartType": payload.get("chartType"), "dataSource": {"dataSourceId": app_key, "dataSourceType": "qingflow"}, "selectedDimensions": self._resolve_qingbi_dimension_fields( payload.get("selectedDimensionFieldIds", []), entity=entity, field_map=field_map, bi_field_map=bi_field_map, qingbi_fields_by_id=qingbi_fields_by_id, ), "selectedMetrics": self._resolve_qingbi_metric_fields( payload.get("selectedMetricFieldIds", []), aggregate=str(payload.get("aggregate") or "count").lower(), entity=entity, field_map=field_map, bi_field_map=bi_field_map, qingbi_fields_by_id=qingbi_fields_by_id, ), "beforeAggregationFilterMatrix": [], "afterAggregationFilterMatrix": [], "chartStyleConfigs": deepcopy(payload.get("chartStyleConfigs", [])), "conditionFormatMatrix": deepcopy(payload.get("conditionFormatMatrix", [])), "displayLimitConfig": deepcopy(payload.get("displayLimitConfig", {"status": 1, "type": "asc", "limit": 20})), "rawDataConfigDTO": deepcopy(payload.get("rawDataConfigDTO", {"beingOpen": False, "authInfo": {"type": "ws", "contactAuth": {"type": "all", "authMembers": {}}, "externalMemberAuth": {"type": "not", "authMembers": {}}}, "fieldInfoList": []})), } for key in ( "selectedTime", "xDimensions", "yDimensions", "xMetrics", "yMetrics", "leftMetrics", "rightMetrics", "pieType", "radarType", "queryConditionFieldIds", "queryConditionStatus", "queryConditionExact", ): if key in payload: data[key] = deepcopy(payload[key]) return data def _resolve_graph_payload(self, payload: dict[str, Any], field_map: dict[str, int]) -> dict[str, Any]: data = deepcopy(payload) if "field_id" in data: field_id = data.pop("field_id") que_id = field_map.get(field_id) if que_id is None: return {} data["queId"] = que_id return data def _resolve_match_rule_groups(self, groups: list[list[dict[str, Any]]], field_map: dict[str, int]) -> list[list[dict[str, Any]]]: resolved_groups: list[list[dict[str, Any]]] = [] for group in groups or []: resolved_group: list[dict[str, Any]] = [] for rule in group or []: resolved_rule = deepcopy(rule) field_id = resolved_rule.pop("field_id", None) if field_id is not None: que_id = field_map.get(field_id) if que_id is None: continue resolved_rule["queId"] = que_id resolved_group.append(resolved_rule) if resolved_group: resolved_groups.append(resolved_group) return resolved_groups def _resolve_portal_payload( self, payload: dict[str, Any], store: RunArtifactStore, *, base_payload: dict[str, Any] | None = None, ) -> dict[str, Any]: data = deepcopy(base_payload) if isinstance(base_payload, dict) else deepcopy(payload) data["dashName"] = payload.get("dashName") or data.get("dashName") data["dashIcon"] = payload.get("dashIcon") or data.get("dashIcon") data["auth"] = deepcopy(payload.get("auth") or data.get("auth")) data["hideCopyright"] = payload.get("hideCopyright", data.get("hideCopyright", False)) tag_id = store.get_artifact("package", "tag_id") data["tags"] = deepcopy(payload.get("tags") or data.get("tags") or []) for tag in data.get("tags", []): if tag.get("tagId") == "__PACKAGE_TAG_ID__": tag["tagId"] = tag_id dash_global_config = deepcopy(payload.get("dashGlobalConfig") or data.get("dashGlobalConfig") or {}) dash_global_config.pop("layout", None) dash_global_config["interval"] = dash_global_config.get("interval") or 60 dash_global_config["beingAutoRefresh"] = bool(dash_global_config.get("beingAutoRefresh", False)) data["dashGlobalConfig"] = dash_global_config if "components" in payload: data["components"] = self._build_portal_components(payload.get("components", []), store) return data def _resolve_navigation_payload(self, payload: dict[str, Any], store: RunArtifactStore, ordinal: int) -> dict[str, Any]: data = deepcopy(payload) if data.get("tagId") == "__PACKAGE_TAG_ID__": data["tagId"] = store.get_artifact("package", "tag_id") if "appRef" in data: ref = data.pop("appRef") data["appKey"] = self._get_app_key(store, ref["entity_id"]) if "chartRef" in data: ref = data.pop("chartRef") chart = store.get_artifact("charts", f"{ref['entity_id']}:{ref['chart_id']}", {}) data["chartKey"] = chart.get("chart_key") if data.get("dashKey") == "__PORTAL_DASH_KEY__": data["dashKey"] = store.get_artifact("portal", "dash_key") if "viewRef" in data: ref = data.pop("viewRef") view = store.get_artifact("views", f"{ref['entity_id']}:{ref['view_id']}", {}) data["viewgraphKey"] = view.get("viewgraph_key") data["ordinal"] = ordinal return data def _resolve_view_reorder_payload(self, profile: str, app_key: str, created_view_keys: list[str]) -> list[dict[str, Any]]: current_view_list = self.view_tools.view_list(profile=profile, app_key=app_key).get("result") or [] if not isinstance(current_view_list, list) or not current_view_list: return [{"ordinalType": "FIXED_VIEW_LIST", "viewKeyList": created_view_keys}] created_key_set = set(created_view_keys) assigned_keys: set[str] = set() payload: list[dict[str, Any]] = [] for group in current_view_list: if not isinstance(group, dict): continue ordinal_type = group.get("ordinalType") view_keys = [ view.get("viewKey") for view in group.get("viewList", []) if isinstance(view, dict) and view.get("viewKey") ] if not ordinal_type or not view_keys: continue prioritized_keys = [key for key in created_view_keys if key in view_keys] if prioritized_keys: assigned_keys.update(prioritized_keys) remaining_keys = [key for key in view_keys if key not in created_key_set] payload.append({"ordinalType": ordinal_type, "viewKeyList": prioritized_keys + remaining_keys}) else: payload.append({"ordinalType": ordinal_type, "viewKeyList": view_keys}) unassigned_keys = [key for key in created_view_keys if key not in assigned_keys] if unassigned_keys: fixed_group = next((group for group in payload if group.get("ordinalType") == "FIXED_VIEW_LIST"), None) if fixed_group is None: payload.insert(0, {"ordinalType": "FIXED_VIEW_LIST", "viewKeyList": unassigned_keys}) else: fixed_group["viewKeyList"] = unassigned_keys + [key for key in fixed_group["viewKeyList"] if key not in set(unassigned_keys)] return payload or [{"ordinalType": "FIXED_VIEW_LIST", "viewKeyList": created_view_keys}] def _build_portal_components(self, components: list[dict[str, Any]], store: RunArtifactStore) -> list[dict[str, Any]]: resolved_components: list[dict[str, Any]] = [] pc_x = 0 pc_y = 0 pc_row_height = 0 mobile_y = 0 reserved_keys = {"sectionId", "title", "sourceType", "ordinal", "chartRef", "viewRef", "text", "url"} for component in components: source_type = component.get("sourceType") extra = {key: deepcopy(value) for key, value in component.items() if key not in reserved_keys} position = deepcopy(extra.pop("position", None)) if position is None: position, pc_x, pc_y, pc_row_height, mobile_y = _portal_component_position( source_type, pc_x=pc_x, pc_y=pc_y, pc_row_height=pc_row_height, mobile_y=mobile_y, ) dash_style = deepcopy(extra.pop("dashStyleConfigBO", None)) if source_type == "chart" and component.get("chartRef"): ref = component["chartRef"] chart = store.get_artifact("charts", f"{ref['entity_id']}:{ref['chart_id']}", {}) chart_config = { "biChartId": chart.get("bi_chart_id") or chart.get("chart_key"), "chartComponentTitle": component.get("title"), "beingShowTitle": True, } chart_config.update(extra.pop("chartConfig", {}) if isinstance(extra.get("chartConfig"), dict) else {}) if not chart_config.get("chartComponentTitle"): chart_config["chartComponentTitle"] = component.get("title") resolved_component: dict[str, Any] = { "type": BI_CHART_COMPONENT_TYPE, "position": position, "chartConfig": _compact_dict(chart_config), } if dash_style is not None: resolved_component["dashStyleConfigBO"] = dash_style resolved_component.update(extra) resolved_components.append( resolved_component ) elif source_type == "view" and component.get("viewRef"): ref = component["viewRef"] view = store.get_artifact("views", f"{ref['entity_id']}:{ref['view_id']}", {}) app_key = store.get_artifact("apps", ref["entity_id"], {}).get("app_key") view_config = { "appKey": app_key, "viewgraphKey": view.get("viewgraph_key"), "viewgraphName": component.get("title"), "formTitle": self._entity_display_name(store, ref["entity_id"]), "componentTitle": component.get("title"), "viewgraphType": _portal_view_type(store, ref["entity_id"], ref["view_id"]), "beingShowTitle": True, "dataManageStatus": True, } view_config.update(extra.pop("viewgraphConfig", {}) if isinstance(extra.get("viewgraphConfig"), dict) else {}) if not view_config.get("componentTitle"): view_config["componentTitle"] = component.get("title") resolved_component = { "type": 10, "position": position, "viewgraphConfig": _compact_dict(view_config), } if dash_style is not None: resolved_component["dashStyleConfigBO"] = dash_style resolved_component.update(extra) resolved_components.append( resolved_component ) elif source_type == "grid": raw_grid_config = extra.pop("gridConfig", {}) if isinstance(extra.get("gridConfig"), dict) else {} grid_config = deepcopy(raw_grid_config) raw_items = grid_config.pop("items", None) if raw_items is None: raw_items = extra.pop("items", []) resolved_component = { "type": 2, "position": position, "gridConfig": _compact_dict( { "gridTitle": grid_config.pop("gridTitle", component.get("title")), "beingShowTitle": bool(grid_config.pop("beingShowTitle", True)), "items": self._resolve_grid_items(raw_items if isinstance(raw_items, list) else [], store), **grid_config, } ), } if dash_style is not None: resolved_component["dashStyleConfigBO"] = dash_style resolved_component.update(extra) resolved_components.append(resolved_component) elif source_type == "filter": filter_groups = extra.pop("filterConfig", []) graph_list = extra.pop("graphList", []) if isinstance(filter_groups, dict): graph_list = filter_groups.get("graphList", graph_list) filter_groups = filter_groups.get("filterConfig", []) resolved_component = { "type": 6, "position": position, "filterConfig": { "filterConfig": self._resolve_dash_filter_groups(filter_groups, store), "graphList": self._resolve_dash_filter_graphs(graph_list, store), }, } if dash_style is not None: resolved_component["dashStyleConfigBO"] = dash_style resolved_component.update(extra) resolved_components.append(resolved_component) elif source_type == "text": text_config = {"text": component.get("text", "")} text_config.update(extra.pop("textConfig", {}) if isinstance(extra.get("textConfig"), dict) else {}) resolved_component = { "type": 5, "position": position, "textConfig": text_config, } if dash_style is not None: resolved_component["dashStyleConfigBO"] = dash_style resolved_component.update(extra) resolved_components.append( resolved_component ) elif source_type == "link": link_config = { "url": component.get("url", ""), "beingLoginAuth": False, } link_config.update(extra.pop("linkConfig", {}) if isinstance(extra.get("linkConfig"), dict) else {}) resolved_component = { "type": 4, "position": position, "linkConfig": link_config, } if dash_style is not None: resolved_component["dashStyleConfigBO"] = dash_style resolved_component.update(extra) resolved_components.append( resolved_component ) return resolved_components def _resolve_grid_items(self, items: list[dict[str, Any]], store: RunArtifactStore) -> list[dict[str, Any]]: resolved_items: list[dict[str, Any]] = [] for ordinal, item in enumerate(items): resolved = self._resolve_grid_item(item, store, ordinal) if resolved is not None: resolved_items.append(resolved) return resolved_items def _resolve_grid_item(self, item: dict[str, Any], store: RunArtifactStore, ordinal: int) -> dict[str, Any] | None: data = deepcopy(item) target_type = str(data.pop("target_type", data.pop("targetType", "app")) or "app").lower() display_title = data.pop("title", None) being_show_title = bool(data.pop("beingShowTitle", True)) jump_mode = int(data.pop("jumpMode", 1)) icon_size = data.pop("iconSize", "40px") icon_url = data.pop("iconUrl", "") if target_type == "app": entity_id = data.pop("entity_id", None) or data.pop("entityId", None) if not entity_id: return None app_key = self._get_app_key(store, str(entity_id)) default_title = self._entity_display_name(store, str(entity_id)) custom_title = display_title or default_title being_custom_title = bool(data.pop("beingCustomTitle", False) or custom_title != default_title) return _compact_dict( { "ordinal": ordinal, "type": 1, "jumpMode": jump_mode, "iconUrl": icon_url, "title": custom_title, "beingShowTitle": being_show_title, "beingCustomTitle": being_custom_title, "customTitle": custom_title, "linkAppKey": app_key, "linkFormType": int(data.pop("linkFormType", 2)), "iconSize": icon_size, **data, } ) if target_type == "portal": dash_key = data.pop("dashKey", None) or store.get_artifact("portal", "dash_key") if not dash_key: return None custom_title = display_title or "门户" return _compact_dict( { "ordinal": ordinal, "type": 2, "jumpMode": jump_mode, "iconUrl": icon_url, "title": custom_title, "beingShowTitle": being_show_title, "beingCustomTitle": bool(data.pop("beingCustomTitle", True)), "customTitle": custom_title, "linkDashKey": dash_key, "iconSize": icon_size, **data, } ) if target_type == "package": tag_id = data.pop("tagId", None) or store.get_artifact("package", "tag_id") if tag_id is None: return None custom_title = display_title or "应用包" return _compact_dict( { "ordinal": ordinal, "type": 3, "jumpMode": jump_mode, "iconUrl": icon_url, "title": custom_title, "beingShowTitle": being_show_title, "beingCustomTitle": bool(data.pop("beingCustomTitle", True)), "customTitle": custom_title, "linkTagId": tag_id, "iconSize": icon_size, **data, } ) if target_type in {"link", "url", "custom_url"}: url = data.pop("url", None) or data.pop("linkUrl", None) if not url: return None custom_title = display_title or "链接" return _compact_dict( { "ordinal": ordinal, "type": 4, "jumpMode": jump_mode, "iconUrl": icon_url, "title": custom_title, "beingShowTitle": being_show_title, "beingCustomTitle": bool(data.pop("beingCustomTitle", True)), "customTitle": custom_title, "linkUrl": url, "iconSize": icon_size, **data, } ) return None def _seed_records(self, profile: str, entity: CompiledEntity, store: RunArtifactStore) -> None: if not entity.sample_records: return app_key = self._get_app_key(store, entity.entity_id) field_meta = store.get_artifact("field_maps", entity.entity_id, {}) or {} field_map = field_meta.get("by_field_id", {}) record_artifacts = store.get_artifact("records", entity.entity_id, {}) or {} for index, sample_record in enumerate(entity.sample_records, start=1): record_key = sample_record.get("record_id") or f"record_{index}" if record_key in record_artifacts: continue answers = self._build_seed_answers(profile, entity, sample_record.get("values", {}), field_map, store) if not answers: continue result = self.record_tools.record_create( profile=profile, app_key=app_key, answers=answers, submit_type=int(sample_record.get("submit_type", 1)), ) apply_result = result.get("result") or {} apply_id = apply_result.get("applyId") or apply_result.get("id") record_artifacts[record_key] = {"apply_id": apply_id, "result": result} store.set_artifact("records", entity.entity_id, record_artifacts) def _build_seed_answers( self, profile: str, entity: CompiledEntity, values: dict[str, Any], field_map: dict[str, int], store: RunArtifactStore, ) -> list[dict[str, Any]]: answers: list[dict[str, Any]] = [] for field_id, raw_value in values.items(): field_spec = entity.field_specs.get(field_id) que_id = field_map.get(field_id) if field_spec is None or que_id is None: continue answer = self._build_answer_detail(profile, field_spec, que_id, raw_value, store) if answer is not None: answers.append(answer) return answers def _build_answer_detail( self, profile: str, field_spec: dict[str, Any], que_id: int, raw_value: Any, store: RunArtifactStore, ) -> dict[str, Any] | None: field_type = FieldType(field_spec["type"]) if raw_value is None: return None if field_type == FieldType.relation: references = raw_value if isinstance(raw_value, list) else [raw_value] values = [] for record_ref in references: apply_id = self._resolve_sample_record_apply_id(field_spec["target_entity_id"], record_ref, store) if apply_id is not None: values.append({"value": str(apply_id)}) if not values: return None return {"queId": que_id, "queType": 25, "values": values, "tableValues": []} if field_type == FieldType.multi_select: items = raw_value if isinstance(raw_value, list) else [raw_value] return {"queId": que_id, "queType": QUESTION_TYPE_MAP[field_type], "values": [{"value": str(item)} for item in items], "tableValues": []} if field_type == FieldType.boolean: return {"queId": que_id, "queType": QUESTION_TYPE_MAP[field_type], "values": [{"value": "是" if bool(raw_value) else "否"}], "tableValues": []} if field_type == FieldType.member: member_value = self._resolve_seed_member_value(profile, raw_value) if member_value is None: return None return {"queId": que_id, "queType": QUESTION_TYPE_MAP[field_type], "values": [member_value], "tableValues": []} if field_type == FieldType.attachment: items = raw_value if isinstance(raw_value, list) else [raw_value] return {"queId": que_id, "queType": QUESTION_TYPE_MAP[field_type], "values": items, "tableValues": []} if field_type in {FieldType.subtable, FieldType.department}: return None return {"queId": que_id, "queType": QUESTION_TYPE_MAP[field_type], "values": [{"value": str(raw_value)}], "tableValues": []} def _resolve_seed_member_value(self, profile: str, raw_value: Any) -> dict[str, Any] | None: if isinstance(raw_value, dict): member_id = raw_value.get("id") or raw_value.get("uid") if not isinstance(member_id, int) or member_id <= 0: return None value = raw_value.get("value") or raw_value.get("name") or str(member_id) member_payload = { "id": member_id, "value": str(value), } if raw_value.get("email"): member_payload["email"] = raw_value.get("email") if raw_value.get("otherInfo") or raw_value.get("other_info"): member_payload["otherInfo"] = raw_value.get("otherInfo") or raw_value.get("other_info") return member_payload if not isinstance(raw_value, int) or raw_value <= 0: return None session_profile = self.record_tools.sessions.get_profile(profile) value = str(raw_value) member_payload: dict[str, Any] = {"id": raw_value, "value": value} if session_profile is not None and raw_value == session_profile.uid: member_payload["value"] = session_profile.nick_name or session_profile.email or value if session_profile.email: member_payload["email"] = session_profile.email return member_payload def _resolve_sample_record_apply_id(self, entity_id: str, record_ref: Any, store: RunArtifactStore) -> int | None: record_key = str(record_ref) entity_records = store.get_artifact("records", entity_id, {}) or {} apply_id = (entity_records.get(record_key) or {}).get("apply_id") return int(apply_id) if apply_id else None def _entity_from_step(self, compiled: CompiledSolution, step_name: str) -> CompiledEntity: entity_id = step_name.split(".")[-1] for entity in compiled.entities: if entity.entity_id == entity_id: return entity raise QingflowApiError.config_error(f"unknown entity '{entity_id}' in execution plan") def _role_from_step(self, compiled: CompiledSolution, step_name: str) -> CompiledRole: role_id = step_name.split(".")[-1] for role in compiled.roles: if role.role_id == role_id: return role raise QingflowApiError.config_error(f"unknown role '{role_id}' in execution plan") def _get_app_key(self, store: RunArtifactStore, entity_id: str) -> str: app_key = store.get_artifact("apps", entity_id, {}).get("app_key") if not app_key: raise QingflowApiError.config_error(f"app_key for entity '{entity_id}' is missing") return app_key def _get_edit_version_no(self, store: RunArtifactStore, entity_id: str) -> int | None: app_artifact = store.get_artifact("apps", entity_id, {}) or {} edit_version_no = app_artifact.get("edit_version_no") if edit_version_no is not None: return int(edit_version_no) field_meta = store.get_artifact("field_maps", entity_id, {}) or {} edit_version_no = field_meta.get("edit_version_no") if edit_version_no is not None: return int(edit_version_no) return None def _set_edit_version_no(self, store: RunArtifactStore, entity_id: str, edit_version_no: int) -> None: app_artifact = store.get_artifact("apps", entity_id, {}) or {} app_artifact["edit_version_no"] = int(edit_version_no) store.set_artifact("apps", entity_id, app_artifact) def _ensure_edit_version( self, profile: str, entity_id: str, store: RunArtifactStore, *, app_key: str | None = None, force_new: bool = False, ) -> int | None: current = None if force_new else self._get_edit_version_no(store, entity_id) if current is not None: return current effective_app_key = app_key or self._get_app_key(store, entity_id) version_result = self.app_tools.app_get_edit_version_no(profile=profile, app_key=effective_app_key).get("result") or {} edit_version_no = version_result.get("editVersionNo") or version_result.get("versionNo") if edit_version_no is None: return None self._set_edit_version_no(store, entity_id, int(edit_version_no)) return int(edit_version_no) def _get_field_map(self, store: RunArtifactStore, entity_id: str) -> dict[str, int]: return store.get_artifact("field_maps", entity_id, {}).get("by_field_id", {}) def _publish_payload(self, store: RunArtifactStore, entity_id: str) -> dict[str, Any]: edit_version_no = self._get_edit_version_no(store, entity_id) or 1 return {"editVersionNo": int(edit_version_no)} def _target_entity_id_from_label(self, entity: CompiledEntity, label: str) -> str | None: for field in entity.field_specs.values(): if field["label"] == label and field.get("target_entity_id"): return field["target_entity_id"] return None def _target_field_label(self, entity: CompiledEntity, label: str, store: RunArtifactStore) -> str: for field in entity.field_specs.values(): if field["label"] == label and field.get("target_field_id"): target_field_id = field["target_field_id"] target_entity_id = field["target_entity_id"] return self._field_label_from_id(target_entity_id, target_field_id, store) return "名称" def _target_field_id_from_label(self, entity: CompiledEntity, label: str) -> str: for field in entity.field_specs.values(): if field["label"] == label and field.get("target_field_id"): return field["target_field_id"] return "title" def _field_label_from_id(self, entity_id: str, field_id: str, store: RunArtifactStore | None) -> str: if store is not None: entity_field_map = store.data.get("normalized_solution_spec", {}).get("entities", []) for entity in entity_field_map: if entity["entity_id"] == entity_id: for field in entity["fields"]: if field["field_id"] == field_id: return field["label"] return "名称" if field_id == "title" else field_id def _entity_display_name(self, store: RunArtifactStore, target_entity_id: str) -> str: for entity in store.data.get("normalized_solution_spec", {}).get("entities", []): if entity["entity_id"] == target_entity_id: return entity["display_name"] return target_entity_id def _entity_data(self, store: RunArtifactStore, entity_id: str) -> dict[str, Any]: for entity in store.data.get("normalized_solution_spec", {}).get("entities", []): if entity["entity_id"] == entity_id: return entity return {} def _field_spec_from_store(self, store: RunArtifactStore, entity_id: str, field_id: str) -> dict[str, Any]: entity = self._entity_data(store, entity_id) for field in entity.get("fields", []): if field.get("field_id") == field_id: return field return {} def _form_stage_error( self, exc: Exception, *, stage_name: str, entity: CompiledEntity, app_key: str, payload: dict[str, Any], ) -> QingflowApiError: nested = _coerce_nested_error_payload(exc) category = str(nested.get("category") or "runtime") message = str(nested.get("message") or f"{stage_name} update failed") return QingflowApiError( category=category, message=message, backend_code=nested.get("backend_code"), request_id=nested.get("request_id"), http_status=nested.get("http_status"), details={ "stage": stage_name, "entity_id": entity.entity_id, "display_name": entity.display_name, "app_key": app_key, "field_count": len(entity.field_specs), "field_catalog": _compiled_field_catalog(entity), "form_payload_summary": _summarize_form_payload(payload), "backend_error": nested or None, }, ) def _resolve_dash_filter_graphs(self, graph_list: list[dict[str, Any]], store: RunArtifactStore) -> list[dict[str, Any]]: resolved_graphs: list[dict[str, Any]] = [] for graph in graph_list: data = deepcopy(graph) graph_ref = data.pop("graphRef", None) if isinstance(data.get("graphRef"), dict) else {} entity_id = data.pop("entity_id", None) or graph_ref.get("entity_id") chart_id = data.pop("chart_id", None) or graph_ref.get("chart_id") view_id = data.pop("view_id", None) or graph_ref.get("view_id") graph_type = str(data.get("graphType") or graph_ref.get("graphType") or ("VIEW" if view_id else "CHART")).upper() if entity_id and chart_id and not data.get("graphKey"): chart = store.get_artifact("charts", f"{entity_id}:{chart_id}", {}) chart_info = chart.get("chart_info", {}) data["graphKey"] = chart.get("bi_chart_id") or chart.get("chart_key") data.setdefault("graphName", chart_info.get("chartName") or chart_id) if entity_id and view_id and not data.get("graphKey"): view = store.get_artifact("views", f"{entity_id}:{view_id}", {}) data["graphKey"] = view.get("viewgraph_key") data.setdefault("graphName", self._entity_display_name(store, entity_id)) data["graphType"] = graph_type if data.get("graphKey"): resolved_graphs.append(_compact_dict(data)) return resolved_graphs def _resolve_dash_filter_groups(self, groups: list[dict[str, Any]], store: RunArtifactStore) -> list[dict[str, Any]]: resolved_groups: list[dict[str, Any]] = [] for group in groups: group_data = deepcopy(group) resolved_items: list[dict[str, Any]] = [] for item in group_data.get("filterGroupConfig", []): item_data = deepcopy(item) conditions = item_data.get("filterCondition", []) or [] item_data["filterCondition"] = [ resolved for condition in conditions if (resolved := self._resolve_dash_filter_condition(condition, store)) is not None ] resolved_items.append(item_data) group_data["filterGroupConfig"] = resolved_items resolved_groups.append(group_data) return resolved_groups def _resolve_dash_filter_condition(self, condition: dict[str, Any], store: RunArtifactStore) -> dict[str, Any] | None: data = deepcopy(condition) graph_ref = data.pop("graphRef", None) if isinstance(data.get("graphRef"), dict) else {} entity_id = data.pop("entity_id", None) or graph_ref.get("entity_id") field_id = data.pop("field_id", None) or graph_ref.get("field_id") chart_id = data.pop("chart_id", None) or graph_ref.get("chart_id") view_id = data.pop("view_id", None) or graph_ref.get("view_id") graph_type = str(data.get("graphType") or graph_ref.get("graphType") or ("VIEW" if view_id else "CHART")).upper() if entity_id and field_id: field_meta = store.get_artifact("field_maps", entity_id, {}) field_spec = self._field_spec_from_store(store, entity_id, field_id) field_label = field_spec.get("label") or self._field_label_from_id(entity_id, field_id, store) qingflow_field_type = field_spec.get("type") if qingflow_field_type: try: data.setdefault("queType", QUESTION_TYPE_MAP[FieldType(qingflow_field_type)]) except (KeyError, ValueError): pass data.setdefault("queTitle", field_label) data.setdefault("queId", field_meta.get("by_field_id", {}).get(field_id)) data.setdefault("queOriginType", data.get("queType")) if qingflow_field_type == "date": data.setdefault("dateType", 0) elif qingflow_field_type == "datetime": data.setdefault("dateType", 1) if entity_id and chart_id and not data.get("chartKey"): chart = store.get_artifact("charts", f"{entity_id}:{chart_id}", {}) chart_info = chart.get("chart_info", {}) data["chartKey"] = chart.get("bi_chart_id") or chart.get("chart_key") bi_field_id = store.get_artifact("field_maps", entity_id, {}).get("bi_by_field_id", {}).get(field_id) if bi_field_id: data["biFieldId"] = bi_field_id data.setdefault("chartName", chart_info.get("chartName") or chart_id) if entity_id and view_id and not data.get("chartKey"): view = store.get_artifact("views", f"{entity_id}:{view_id}", {}) data["chartKey"] = view.get("viewgraph_key") data.setdefault("chartName", self._entity_display_name(store, entity_id)) data["graphType"] = graph_type if not data.get("chartKey") or data.get("queId") is None: return None return _compact_dict(data) def _resolve_chart_headers(self, headers: list[dict[str, Any]], field_map: dict[str, int]) -> list[dict[str, Any]]: resolved_headers: list[dict[str, Any]] = [] for header in headers: field_id = header.get("field_id") que_id = field_map.get(field_id) if field_id else header.get("queId") if que_id is None: continue resolved = deepcopy(header) resolved["queId"] = que_id resolved.pop("field_id", None) resolved_headers.append(resolved) return resolved_headers def _resolve_chart_targets(self, targets: list[dict[str, Any]], field_map: dict[str, int]) -> list[dict[str, Any]]: resolved_targets: list[dict[str, Any]] = [] for target in targets: field_id = target.get("field_id") que_id = field_map.get(field_id) if field_id else target.get("queId") if que_id is None and not target.get("createByFormula"): continue resolved = deepcopy(target) resolved["queId"] = que_id resolved["aggreType"] = self._chart_aggre_type(str(target.get("aggregate") or "count").lower()) resolved.setdefault("createByFormula", False) resolved.setdefault("numberFormat", 1) resolved.setdefault("currencyType", 1) resolved.setdefault("decimalDigit", 0) resolved.setdefault("sort", "none") resolved.pop("field_id", None) resolved.pop("aggregate", None) resolved_targets.append(resolved) return resolved_targets def _chart_aggre_type(self, aggregate: str) -> int: return { "sum": 0, "average": 1, "avg": 1, "max": 2, "min": 3, "first": 4, "last": 5, "count": 6, "count_all": 8, }.get(aggregate, 6) def _resolve_qingbi_dimension_fields( self, field_ids: list[str], *, entity: CompiledEntity, field_map: dict[str, int], bi_field_map: dict[str, str], qingbi_fields_by_id: dict[str, dict[str, Any]], ) -> list[dict[str, Any]]: dimensions: list[dict[str, Any]] = [] for field_id in field_ids: resolved = self._resolve_qingbi_field( entity=entity, field_id=field_id, field_map=field_map, bi_field_map=bi_field_map, qingbi_fields_by_id=qingbi_fields_by_id, field_usage="dimension", aggregate="sum", ) if resolved is not None: dimensions.append(resolved) return dimensions def _resolve_qingbi_metric_fields( self, field_ids: list[str], *, aggregate: str, entity: CompiledEntity, field_map: dict[str, int], bi_field_map: dict[str, str], qingbi_fields_by_id: dict[str, dict[str, Any]], ) -> list[dict[str, Any]]: if aggregate == "count" or not field_ids: return [self._default_qingbi_total_metric()] metrics: list[dict[str, Any]] = [] for field_id in field_ids: resolved = self._resolve_qingbi_field( entity=entity, field_id=field_id, field_map=field_map, bi_field_map=bi_field_map, qingbi_fields_by_id=qingbi_fields_by_id, field_usage="metric", aggregate=aggregate, ) if resolved is not None: metrics.append(resolved) return metrics or [self._default_qingbi_total_metric()] def _resolve_qingbi_field( self, *, entity: CompiledEntity, field_id: str, field_map: dict[str, int], bi_field_map: dict[str, str], qingbi_fields_by_id: dict[str, dict[str, Any]], field_usage: str, aggregate: str, ) -> dict[str, Any] | None: que_id = field_map.get(field_id) bi_field_id = bi_field_map.get(field_id) if que_id is None or not bi_field_id: return None field_spec = entity.field_specs.get(field_id, {}) field_label = entity.field_labels.get(field_id, field_id) qingbi_field = deepcopy(qingbi_fields_by_id.get(bi_field_id, {})) field_type = qingbi_field.get("fieldType") or self._qingbi_field_type_from_spec(field_spec.get("type")) data = { "fieldId": bi_field_id, "fieldName": qingbi_field.get("fieldName") or field_label, "fieldType": field_type, "orderType": "default", "alignType": "left", "dateFormat": "yyyy-MM-dd", "numberFormat": "default", "numberConfig": { "format": "splitter", "unit": "DEFAULT", "prefix": "", "suffix": "", "digit": None, }, "digit": None, "aggreType": self._qingbi_aggregate_type(aggregate if field_usage == "metric" else "sum"), "orderPriority": None, "width": None, "verticalAlign": "middle", "formula": qingbi_field.get("formula"), "fieldSource": qingbi_field.get("fieldSource") or "default", "status": qingbi_field.get("status"), "supId": qingbi_field.get("supId"), "beingTable": bool(qingbi_field.get("beingTable", False)), "returnType": qingbi_field.get("returnType"), } return data def _default_qingbi_total_metric(self) -> dict[str, Any]: return { "fieldId": ":-100", "fieldName": "数据总量", "fieldType": "decimal", "orderType": "default", "alignType": "left", "dateFormat": "yyyy-MM-dd", "numberFormat": "default", "numberConfig": { "format": "splitter", "unit": "DEFAULT", "prefix": "", "suffix": "", "digit": None, }, "digit": None, "aggreType": "sum", "orderPriority": None, "width": None, "verticalAlign": "middle", "beingTable": False, "supId": None, } def _qingbi_aggregate_type(self, aggregate: str) -> str: return { "sum": "sum", "avg": "avg", "average": "avg", "max": "max", "min": "min", "count": "sum", "distinct_count": "sum", }.get(aggregate, "sum") def _qingbi_field_type_from_spec(self, field_type: str | None) -> str: return { "single_select": "singleSelect", "multi_select": "multiSelect", "member": "member", "department": "dept", "date": "datetime", "datetime": "datetime", "number": "decimal", "amount": "decimal", "boolean": "singleSelect", }.get(str(field_type or ""), "string") def extract_field_map(schema: dict[str, Any]) -> dict[str, int]: result: dict[str, int] = {} for line in schema.get("formQues", []): _extract_question_line(line, result, None) return result def _extract_question_line(questions: list[dict[str, Any]], result: dict[str, int], parent_title: str | None) -> None: for question in questions: title = question.get("queTitle") que_id = question.get("queId") que_type = question.get("queType") if title and que_id and que_type != 24: composite_title = title if parent_title is None else f"{parent_title}.{title}" result[composite_title] = que_id result.setdefault(title, que_id) for sub_question in question.get("subQuestions", []) or []: sub_title = sub_question.get("queTitle") sub_que_id = sub_question.get("queId") if title and sub_title and sub_que_id: result[f"{title}.{sub_title}"] = sub_que_id result.setdefault(sub_title, sub_que_id) for inner_row in question.get("innerQuestions", []) or []: _extract_question_line(inner_row, result, title if que_type == 24 else parent_title) def _compact_dict(data: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in data.items() if value is not None} def _extract_workflow_node_id(result: Any, *, expected_type: int | None = None) -> int | None: if expected_type is not None: exact_match = _extract_workflow_node_id_exact(result, expected_type=expected_type) if exact_match is not None: return exact_match return _extract_workflow_node_id_exact(result, expected_type=None) def _extract_workflow_node_id_exact(result: Any, *, expected_type: int | None) -> int | None: if isinstance(result, dict): node_id = result.get("auditNodeId") node_type = result.get("type") if isinstance(node_id, int) and (expected_type is None or node_type == expected_type): return node_id for node in result.values(): extracted = _extract_workflow_node_id_exact(node, expected_type=expected_type) if extracted is not None: return extracted if isinstance(result, list): for node in result: extracted = _extract_workflow_node_id_exact(node, expected_type=expected_type) if extracted is not None: return extracted return None def _extract_workflow_branch_lane_ids(result: Any) -> list[int]: if isinstance(result, dict): branches = result.get("branches") if isinstance(branches, list): lane_ids = [ branch.get("auditNodeId") for branch in branches if isinstance(branch, dict) and isinstance(branch.get("auditNodeId"), int) ] if lane_ids: return lane_ids for node in result.values(): lane_ids = _extract_workflow_branch_lane_ids(node) if lane_ids: return lane_ids if isinstance(result, list): for node in result: lane_ids = _extract_workflow_branch_lane_ids(node) if lane_ids: return lane_ids return [] def _branch_lane_ref(branch_node_id: str, branch_index: int) -> str: return f"__branch_lane__{branch_node_id}__{branch_index}" def _coerce_workflow_nodes(result: Any) -> dict[int, dict[str, Any]]: if not isinstance(result, dict): return {} nodes: dict[int, dict[str, Any]] = {} for node_id, node in result.items(): try: parsed_node_id = int(node_id) except (TypeError, ValueError): continue if isinstance(node, dict): nodes[parsed_node_id] = node return nodes def _workflow_node_is_branch(nodes: dict[int, dict[str, Any]], node_id: int | None) -> bool: if node_id is None: return False node = nodes.get(int(node_id)) return isinstance(node, dict) and node.get("type") == 1 def _find_branch_lane_ids(nodes: dict[int, dict[str, Any]], branch_node_id: int | None) -> list[int]: if branch_node_id is None: return [] lane_ids = [ node_id for node_id, node in nodes.items() if isinstance(node, dict) and node.get("type") == 2 and node.get("prevId") == int(branch_node_id) ] return sorted(lane_ids) def _find_created_branch_node_id( nodes: dict[int, dict[str, Any]], *, before_node_ids: set[int], prev_id: int | None, ) -> int | None: candidates = [ node_id for node_id, node in nodes.items() if node_id not in before_node_ids and isinstance(node, dict) and node.get("type") == 1 ] if prev_id is not None: prev_candidates = [node_id for node_id in candidates if nodes[node_id].get("prevId") == int(prev_id)] if prev_candidates: return prev_candidates[0] return candidates[0] if candidates else None def _find_created_sub_branch_lane_id( nodes: dict[int, dict[str, Any]], *, before_node_ids: set[int], branch_node_id: int | None, ) -> int | None: if branch_node_id is None: return None candidates = [ node_id for node_id, node in nodes.items() if node_id not in before_node_ids and isinstance(node, dict) and node.get("type") == 2 and node.get("prevId") == int(branch_node_id) ] return candidates[0] if candidates else None def _is_navigation_plugin_unavailable(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False backend_code = backend_code_int(error) if backend_code != 50004: return False message = error.message or "" return "导航菜单不可用" in message or "插件未安装" in message def _portal_plan_has_source_type(portal_plan: dict[str, Any], source_type: str) -> bool: components = ((portal_plan.get("update_payload") or {}) if isinstance(portal_plan, dict) else {}).get("components", []) return any(component.get("sourceType") == source_type for component in components if isinstance(component, dict)) def _coerce_qingflow_error(error: Exception) -> QingflowApiError | None: if isinstance(error, QingflowApiError): return error if not isinstance(error, RuntimeError): return None try: payload = json.loads(str(error)) except ValueError: return None if not isinstance(payload, dict): return None message = payload.get("message") category = payload.get("category") if not category or not message: return None return QingflowApiError( category=str(category), message=str(message), backend_code=payload.get("backend_code"), request_id=payload.get("request_id"), http_status=payload.get("http_status"), ) def _is_permission_restricted_error(error: QingflowApiError) -> bool: if is_auth_like_error(error): return False return backend_code_int(error) in {40002, 40027} def _required_state_read_blocked_error( *, resource: str, message: str, error: QingflowApiError, details: dict[str, Any] | None = None, ) -> QingflowApiError: merged_details = deepcopy(details) if isinstance(details, dict) else {} merged_details["state_read_blocked"] = { "resource": resource, "transport_error": { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, "request_id": error.request_id, }, } return QingflowApiError( category=error.category, message=message, backend_code=error.backend_code, request_id=error.request_id, http_status=error.http_status, details=merged_details, ) def _post_write_readback_artifact( *, resource: str, target: dict[str, Any], error: QingflowApiError, ) -> dict[str, Any]: return { "resource": resource, "target": deepcopy(target), "readback_status": "unavailable", "readback_verified": False, "write_executed": True, "safe_to_retry": False, "transport_error": { "http_status": error.http_status, "backend_code": error.backend_code, "category": error.category, "request_id": error.request_id, }, } def _artifacts_have_post_write_readback_pending(value: Any) -> bool: if isinstance(value, dict): if ( value.get("write_executed") is True and value.get("safe_to_retry") is False and value.get("readback_status") == "unavailable" ): return True return any(_artifacts_have_post_write_readback_pending(item) for item in value.values()) if isinstance(value, list): return any(_artifacts_have_post_write_readback_pending(item) for item in value) return False def _portal_component_position( source_type: Any, *, pc_x: int, pc_y: int, pc_row_height: int, mobile_y: int, ) -> tuple[dict[str, Any], int, int, int, int]: source_name = str(source_type or "").lower() if source_name == "filter": cols = 24 rows = 2 elif source_name == "grid": cols = 24 rows = 4 elif source_name == "text": cols = 24 rows = 2 elif source_name == "view": cols = 24 rows = 8 elif source_name == "link": cols = 12 rows = 2 else: cols = 8 rows = 4 if cols == 24: if pc_x != 0: pc_y += pc_row_height pc_x = 0 pc_row_height = 0 position = { "pc": {"cols": 24, "rows": rows, "x": 0, "y": pc_y}, "mobile": {"cols": 6, "rows": rows, "x": 0, "y": mobile_y}, } pc_y += rows mobile_y += rows return position, 0, pc_y, 0, mobile_y if pc_x + cols > 24: pc_y += pc_row_height pc_x = 0 pc_row_height = 0 position = { "pc": {"cols": cols, "rows": rows, "x": pc_x, "y": pc_y}, "mobile": {"cols": 6, "rows": rows, "x": 0, "y": mobile_y}, } pc_x += cols pc_row_height = max(pc_row_height, rows) if pc_x >= 24: pc_y += pc_row_height pc_x = 0 pc_row_height = 0 mobile_y += rows return position, pc_x, pc_y, pc_row_height, mobile_y def _portal_view_type(store: RunArtifactStore, entity_id: str, view_id: str) -> str: view_type_map = { "table": "tableView", "card": "cardView", "board": "boardView", "gantt": "ganttView", "hierarchy": "hierarchyView", } for entity in store.data.get("normalized_solution_spec", {}).get("entities", []): if entity["entity_id"] != entity_id: continue for view in entity.get("views", []): if view.get("view_id") == view_id: return view_type_map.get(view.get("type"), "tableView") return "tableView" def _coerce_nested_error_payload(exc: Exception) -> dict[str, Any]: if isinstance(exc, QingflowApiError): return exc.to_dict() text = str(exc) try: parsed = json.loads(text) except Exception: return {"message": text} if isinstance(parsed, dict): return parsed return {"message": text} def _compiled_field_catalog(entity: CompiledEntity) -> list[dict[str, Any]]: catalog: list[dict[str, Any]] = [] for field_id, field_spec in entity.field_specs.items(): if not isinstance(field_spec, dict): continue catalog.append( { "field_id": field_id, "label": field_spec.get("label"), "type": field_spec.get("type"), "required": field_spec.get("required"), "target_entity_id": field_spec.get("target_entity_id"), } ) return catalog def _summarize_form_payload(payload: dict[str, Any]) -> dict[str, Any]: rows = payload.get("formQues") question_titles: list[str] = [] question_type_counts: dict[str, int] = {} if isinstance(rows, list): for row in rows: for question in _flatten_questions(row): title = question.get("queTitle") if isinstance(title, str) and title: question_titles.append(title) que_type = question.get("queType") if que_type is not None: key = str(que_type) question_type_counts[key] = question_type_counts.get(key, 0) + 1 return { "form_title": payload.get("formTitle"), "edit_version_no": payload.get("editVersionNo"), "question_row_count": len(rows) if isinstance(rows, list) else 0, "question_titles": question_titles, "question_type_counts": question_type_counts, "has_question_relations": bool(payload.get("questionRelations")), } def _flatten_questions(value: Any) -> list[dict[str, Any]]: if isinstance(value, dict): nested = [value] for key in ("innerQuestions", "subQuestions"): children = value.get(key) if isinstance(children, list): for item in children: nested.extend(_flatten_questions(item)) return nested if isinstance(value, list): flattened: list[dict[str, Any]] = [] for item in value: flattened.extend(_flatten_questions(item)) return flattened return [] def _package_item_app_key(item: dict[str, Any]) -> str | None: item_type = item.get("itemType") if item_type is not None and item_type not in {PACKAGE_ITEM_TYPE_FORM, str(PACKAGE_ITEM_TYPE_FORM)}: return None app_key = item.get("appKey") return app_key if isinstance(app_key, str) and app_key else None def _verify_package_attachment(package_tools: PackageTools, *, profile: str, tag_id: int, app_key: str, attempts: int = 2) -> dict[str, Any]: last_detail: dict[str, Any] = {"result": {}} for _ in range(max(attempts, 1)): last_detail = package_tools.package_get(profile=profile, tag_id=tag_id, include_raw=True) result = last_detail.get("result") if isinstance(last_detail.get("result"), dict) else {} verified_items = [deepcopy(existing) for existing in result.get("tagItems", []) if isinstance(existing, dict)] if any(_package_item_app_key(existing) == app_key for existing in verified_items): return last_detail return last_detail def _resolve_package_item_insert_index(tag_items: list[dict[str, Any]], ordinal: Any) -> int: if not isinstance(ordinal, int) or ordinal <= 0: return len(tag_items) return max(0, min(len(tag_items), ordinal - 1)) def _build_viewgraph_questions(schema: dict[str, Any], visible_que_ids: list[int]) -> list[dict[str, Any]]: visible_map = {que_id: index for index, que_id in enumerate(visible_que_ids, start=1)} questions: list[dict[str, Any]] = [] for base_question in schema.get("baseQues", []): question = _build_viewgraph_question(base_question, visible_map, default_auth=3) if question is not None: questions.append(question) for line in schema.get("formQues", []): for question in line: resolved = _build_viewgraph_question(question, visible_map, default_auth=1) if resolved is not None: questions.append(resolved) return questions def _build_viewgraph_question(question: dict[str, Any], visible_map: dict[int, int], *, default_auth: int) -> dict[str, Any] | None: que_id = question.get("queId") if que_id is None: return None sub_questions = [ resolved for sub_question in question.get("subQuestions", []) or [] if (resolved := _build_viewgraph_question(sub_question, visible_map, default_auth=1)) is not None ] inner_questions = [ resolved for inner_row in question.get("innerQuestions", []) or [] for inner_question in inner_row or [] if (resolved := _build_viewgraph_question(inner_question, visible_map, default_auth=1)) is not None ] return { "queId": que_id, "queTitle": question.get("queTitle"), "queType": question.get("queType"), "queAuth": default_auth if que_id <= 5 else 1, "beingListDisplay": que_id in visible_map, "displayOrdinal": visible_map.get(que_id, -1), "subQues": sub_questions, "innerQues": inner_questions, "beingDownload": True, }