from __future__ import annotations import json import re from dataclasses import dataclass from datetime import UTC, datetime from typing import cast from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE, DEFAULT_RECORD_LIST_TYPE from ..errors import QingflowApiError, raise_tool_error from ..json_types import JSONObject, JSONScalar, JSONValue from ..list_type_labels import get_record_list_type_label from .base import ToolBase DEFAULT_QUERY_PAGE_SIZE = 50 DEFAULT_SCAN_MAX_PAGES = 10 DEFAULT_ROW_LIMIT = 200 DEFAULT_OUTPUT_PROFILE = "compact" MAX_LIST_COLUMN_LIMIT = 20 MAX_RECORD_COLUMN_LIMIT = 20 MAX_SUMMARY_PREVIEW_COLUMN_LIMIT = 6 BACKEND_LIST_SEARCH_FIELD_LIMIT = 10 SUPPORTED_QUERY_TOOLS = {"record_query", "record_aggregate", "record_get"} MEMBER_QUE_TYPES = {5} DEPARTMENT_QUE_TYPES = {22} DATE_QUE_TYPES = {4} SINGLE_SELECT_QUE_TYPES = {10, 11} MULTI_SELECT_QUE_TYPES = {12, 15} BOOLEAN_QUE_TYPES: set[int] = set() ATTACHMENT_QUE_TYPES = {13} RELATION_QUE_TYPES = {25} SUBTABLE_QUE_TYPES = {18} VERIFY_UNSUPPORTED_WRITE_QUE_TYPES = {14, 34, 35, 36} DEPARTMENT_MEMBER_JUDGE_PREFIX = "deptId_" JUDGE_EQUAL = 0 JUDGE_UNEQUAL = 1 JUDGE_GREATER_OR_EQUAL = 5 JUDGE_LESS_OR_EQUAL = 7 JUDGE_EQUAL_ANY = 9 JUDGE_FUZZY_MATCH = 19 JUDGE_INCLUDE_ANY = 20 MATCH_TYPE_ACCURACY = 1 SCOPE_ALL = 1 SCOPE_NOT_EMPTY = 2 SCOPE_EMPTY = 3 @dataclass(slots=True) class FormField: que_id: int que_title: str que_type: int | None required: bool readonly: bool system: bool options: list[str] aliases: list[str] raw: JSONObject @dataclass(slots=True) class FieldIndex: by_id: dict[str, FormField] by_title: dict[str, list[FormField]] by_alias: dict[str, list[FormField]] @dataclass(slots=True) class ViewFilterCondition: que_id: int | None que_title: str que_type: int | None judge_type: int | None judge_values: list[str] judge_value_details: list[JSONObject] raw: JSONObject @dataclass(slots=True) class ViewSelection: view_key: str view_name: str conditions: list[list[ViewFilterCondition]] @dataclass(slots=True) class RecordInputError(Exception): message: str error_code: str fix_hint: str details: JSONObject | None = None def __post_init__(self) -> None: Exception.__init__(self, self.message) def to_dict(self) -> JSONObject: payload: JSONObject = { "category": "config", "message": self.message, "error_code": self.error_code, "fix_hint": self.fix_hint, } if self.details is not None: payload["details"] = self.details return payload def as_json(self) -> str: return json.dumps(self.to_dict(), ensure_ascii=False) def __str__(self) -> str: return self.as_json() GENERIC_FIELD_PREFIX_ALIASES = ( "当前", "预计", "最近", "最新", ) GENERIC_FIELD_ALIAS_OVERRIDES: dict[str, list[str]] = { "申请时间": ["新增时间", "创建时间"], "更新时间": ["修改时间", "最后更新时间"], } FIELD_LOOKUP_STRIP_RE = re.compile(r"[\s_()（）\[\]【】{}<>·/\\:：-]+") class RecordTools(ToolBase): def __init__(self, sessions, backend) -> None: # type: ignore[no-untyped-def] super().__init__(sessions, backend) self._form_cache: dict[tuple[str, str], JSONObject] = {} self._view_list_cache: dict[tuple[str, str], list[JSONObject]] = {} self._view_config_cache: dict[tuple[str, str], JSONObject] = {} def register(self, mcp: FastMCP) -> None: @mcp.tool() def record_field_resolve( profile: str = DEFAULT_PROFILE, app_key: str = "", query: str | int | None = None, queries: list[str | int] | None = None, top_k: int = 3, fuzzy: bool = True, ) -> JSONObject: return self.record_field_resolve( profile=profile, app_key=app_key, query=query, queries=queries, top_k=top_k, fuzzy=fuzzy, ) @mcp.tool( description=( "Preflight complex read requests before actual execution. Resolves field selectors, validates required arguments, " "and estimates scan scope. Prefer this when the agent is unsure about query shape." ) ) def record_query_plan( profile: str = DEFAULT_PROFILE, tool: str = "record_query", arguments: JSONObject | None = None, resolve_fields: bool = True, ) -> JSONObject: return self.record_query_plan(profile=profile, tool=tool, arguments=arguments or {}, resolve_fields=resolve_fields) @mcp.tool( description=( "Static preflight for record create/update payloads. Supports ergonomic fields{} mapping, resolves field titles, " "and reports blockers before submit." ) ) def record_write_plan( profile: str = DEFAULT_PROFILE, operation: str = "auto", app_key: str = "", apply_id: int | None = None, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, force_refresh_form: bool = False, ) -> JSONObject: return self.record_write_plan( profile=profile, operation=operation, app_key=app_key, apply_id=apply_id, answers=answers or [], fields=fields or {}, force_refresh_form=force_refresh_form, ) @mcp.tool( description=( "Unified read entry for record list / record detail / summary analysis. " "List mode returns flattened wide-table rows only. Use query_mode=auto to route: " "apply_id -> record, amount_column/time_range/stat_policy -> summary, otherwise -> list." ) ) def record_query( profile: str = DEFAULT_PROFILE, query_mode: str = "auto", app_key: str = "", apply_id: int | None = None, page_num: int = 1, page_size: int = DEFAULT_QUERY_PAGE_SIZE, requested_pages: int = 1, scan_max_pages: int = DEFAULT_SCAN_MAX_PAGES, query_key: str | None = None, filters: list[JSONObject] | None = None, sorts: list[JSONObject] | None = None, max_rows: int = DEFAULT_ROW_LIMIT, max_columns: int | None = None, select_columns: list[str | int] | None = None, amount_column: str | int | None = None, time_range: JSONObject | None = None, stat_policy: JSONObject | None = None, strict_full: bool = False, output_profile: str = DEFAULT_OUTPUT_PROFILE, list_type: int = DEFAULT_RECORD_LIST_TYPE, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: return self.record_query( profile=profile, query_mode=query_mode, app_key=app_key, apply_id=apply_id, page_num=page_num, page_size=page_size, requested_pages=requested_pages, scan_max_pages=scan_max_pages, query_key=query_key, filters=filters or [], sorts=sorts or [], max_rows=max_rows, max_columns=max_columns, select_columns=select_columns or [], amount_column=amount_column, time_range=time_range or {}, stat_policy=stat_policy or {}, strict_full=strict_full, output_profile=output_profile, list_type=list_type, view_key=view_key, view_name=view_name, ) @mcp.tool( description=( "Grouped record analysis endpoint. Aggregates record counts and numeric metrics by selected dimensions. " "Use strict_full=true when the result will be used as a final conclusion." ) ) def record_aggregate( profile: str = DEFAULT_PROFILE, app_key: str = "", group_by: list[str | int] | None = None, amount_column: str | int | None = None, metrics: list[str] | None = None, page_num: int = 1, page_size: int = DEFAULT_QUERY_PAGE_SIZE, requested_pages: int = 1, scan_max_pages: int = DEFAULT_SCAN_MAX_PAGES, query_key: str | None = None, filters: list[JSONObject] | None = None, sorts: list[JSONObject] | None = None, time_range: JSONObject | None = None, time_bucket: str | None = None, max_groups: int = 200, strict_full: bool = False, output_profile: str = DEFAULT_OUTPUT_PROFILE, list_type: int = DEFAULT_RECORD_LIST_TYPE, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: return self.record_aggregate( profile=profile, app_key=app_key, group_by=group_by or [], amount_column=amount_column, metrics=metrics or [], page_num=page_num, page_size=page_size, requested_pages=requested_pages, scan_max_pages=scan_max_pages, query_key=query_key, filters=filters or [], sorts=sorts or [], time_range=time_range or {}, time_bucket=time_bucket, max_groups=max_groups, strict_full=strict_full, output_profile=output_profile, list_type=list_type, view_key=view_key, view_name=view_name, ) @mcp.tool( description=( "Create one record. Supports explicit answers[] and ergonomic fields{} mapping by exact field title or queId. " "Use record_write_plan first for complex payloads." ) ) def record_create( profile: str = DEFAULT_PROFILE, app_key: str = "", answers: list[JSONObject] | None = None, fields: JSONObject | None = None, submit_type: int = 1, verify_write: bool = False, force_refresh_form: bool = False, ) -> JSONObject: return self.record_create( profile=profile, app_key=app_key, answers=answers or [], fields=fields or {}, submit_type=submit_type, verify_write=verify_write, force_refresh_form=force_refresh_form, ) @mcp.tool() def record_get( profile: str = DEFAULT_PROFILE, app_key: str = "", apply_id: int = 0, role: int = 1, list_type: int | None = None, audit_node_id: int | None = None, ) -> JSONObject: return self.record_get( profile=profile, app_key=app_key, apply_id=apply_id, role=role, list_type=list_type, audit_node_id=audit_node_id, ) @mcp.tool(description=self._high_risk_tool_description(operation="update", target="record data")) def record_update( profile: str = DEFAULT_PROFILE, app_key: str = "", apply_id: int = 0, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, role: int = 1, verify_write: bool = False, force_refresh_form: bool = False, ) -> JSONObject: return self.record_update( profile=profile, app_key=app_key, apply_id=apply_id, answers=answers or [], fields=fields or {}, role=role, verify_write=verify_write, force_refresh_form=force_refresh_form, ) @mcp.tool(description=self._high_risk_tool_description(operation="delete", target="record data")) def record_delete( profile: str = DEFAULT_PROFILE, app_key: str = "", apply_id: int = 0, list_type: int = DEFAULT_RECORD_LIST_TYPE, ) -> JSONObject: return self.record_delete(profile=profile, app_key=app_key, apply_id=apply_id, list_type=list_type) def record_field_resolve( self, *, profile: str, app_key: str, query: str | int | None, queries: list[str | int] | None, top_k: int, fuzzy: bool, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) requested = [item for item in (queries or []) if item is not None] if query is not None: requested = [query] if not requested: raise_tool_error(QingflowApiError.config_error("query or queries is required")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=False) results = [] for item in requested: text = str(item).strip() if not text: continue results.append({"requested": text, "matches": self._score_field_matches(text, index, fuzzy=fuzzy, top_k=top_k)}) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "app_key": app_key, "query_count": len(results), "results": results, }, } return self._run_record_tool(profile, runner) def record_query_plan(self, *, profile: str, tool: str, arguments: JSONObject, resolve_fields: bool) -> JSONObject: if tool not in SUPPORTED_QUERY_TOOLS: raise_tool_error(QingflowApiError.config_error(f"tool must be one of {sorted(SUPPORTED_QUERY_TOOLS)}")) normalized = _normalize_plan_arguments(tool, arguments) validation = self._validate_plan_arguments(tool, normalized) def runner(session_profile, context): field_mapping: list[JSONObject] = [] view_resolution: JSONObject | None = None if resolve_fields and isinstance(normalized.get("app_key"), str) and normalized.get("app_key"): index = self._get_field_index(profile, context, cast(str, normalized["app_key"]), force_refresh=False) for candidate in _collect_plan_field_candidates(tool, normalized): field_mapping.append(self._resolve_plan_candidate(candidate, index)) view_selection = self._resolve_view_selection( profile, context, cast(str, normalized["app_key"]), view_key=_normalize_optional_text(normalized.get("view_key")), view_name=_normalize_optional_text(normalized.get("view_name")), ) if view_selection is not None: view_resolution = { "resolved": True, "view_key": view_selection.view_key, "view_name": view_selection.view_name, "local_filtering": bool(view_selection.conditions), "condition_group_count": len(view_selection.conditions), } estimate = _build_plan_estimate(tool, normalized) readiness = _assess_plan_readiness(tool, normalized, validation, field_mapping, estimate) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "tool": tool, "normalized_arguments": normalized, "validation": validation, "field_mapping": field_mapping, "view_resolution": view_resolution, "estimate": estimate, "ready_for_final_conclusion": readiness["ready_for_final_conclusion"], "final_conclusion_blockers": readiness["final_conclusion_blockers"], "recommended_next_actions": readiness["recommended_next_actions"], }, } return self._run_record_tool(profile, runner) def record_write_plan( self, *, profile: str, operation: str, app_key: str, apply_id: int | None, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, force_refresh_form: bool = False, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) inferred_operation = operation if operation in {"create", "update"} else ("update" if apply_id else "create") def runner(session_profile, context): schema = self._get_form_schema(profile, context, app_key, force_refresh=force_refresh_form) index = _build_field_index(schema) normalized_fields = fields or {} normalized_answers_input = answers or [] resolved_fields = self._collect_write_plan_field_refs(fields=normalized_fields, answers=normalized_answers_input, index=index) support_matrix = _summarize_write_support(resolved_fields) invalid_fields: list[JSONObject] = [] normalized_answers: list[JSONObject] = [] try: normalized_answers = self._resolve_answers( profile, context, app_key, answers=normalized_answers_input, fields=normalized_fields, force_refresh_form=force_refresh_form, ) except RecordInputError as error: invalid_fields.append( { "location": _stringify_json(error.details.get("location") if error.details else None), "message": error.message, "error_code": error.error_code, "field": error.details.get("field") if error.details and isinstance(error.details.get("field"), dict) else None, "expected_format": error.details.get("expected_format") if error.details and isinstance(error.details.get("expected_format"), dict) else None, "received_value": error.details.get("received_value") if error.details else None, } ) readonly_or_system_fields = [ { "que_id": entry.get("que_id"), "que_title": entry.get("que_title"), "que_type": entry.get("que_type"), "readonly": entry.get("readonly"), "system": entry.get("system"), "source": entry.get("source"), "requested": entry.get("requested"), } for entry in resolved_fields if bool(entry.get("resolved")) and (bool(entry.get("readonly")) or bool(entry.get("system"))) ] provided_field_ids = { str(answer.get("queId")) for answer in normalized_answers if isinstance(answer.get("queId"), int) and int(answer["queId"]) > 0 } missing_required_fields = [] for field in index.by_id.values(): if not field.required or str(field.que_id) in provided_field_ids: continue missing_required_fields.append( { "que_id": field.que_id, "que_title": field.que_title, "que_type": field.que_type, "reason": "required field not provided", } ) question_relations = _collect_question_relations(schema) option_links = _collect_option_links(resolved_fields) validation = { "valid": not invalid_fields and not missing_required_fields and not readonly_or_system_fields, "missing_required_fields": missing_required_fields, "likely_hidden_required_fields": [], "readonly_or_system_fields": readonly_or_system_fields, "invalid_fields": invalid_fields, "warnings": [ "record_write_plan is a static preflight built from form metadata; runtime visibility and dynamic linkage can still reject writes." ], } blockers = [] if invalid_fields: blockers.append("payload contains invalid field values") if missing_required_fields: blockers.append("required fields are missing") if readonly_or_system_fields: blockers.append("payload writes readonly or system-managed fields") if question_relations: validation["warnings"].append("form contains questionRelations; linked visibility and runtime required rules may differ at submit time.") actions = ["Use record_field_resolve when field titles are ambiguous."] if support_matrix["restricted"]: actions.append("Review write_format.required_presteps for restricted fields before submit.") if invalid_fields: actions.append("Fix invalid_fields before calling record_create or record_update.") if missing_required_fields: actions.append("Fill missing required fields before submit.") if readonly_or_system_fields: actions.append("Remove readonly/system fields from payload before submit.") if question_relations: actions.append("Treat ready_to_submit as static-only because linked fields can still appear at runtime.") return { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "operation": inferred_operation, "app_key": app_key, "apply_id": apply_id, "normalized_answers": normalized_answers, "resolved_fields": resolved_fields, "support_matrix": support_matrix, "validation": validation, "dependencies": { "question_relations_present": bool(question_relations), "question_relations": question_relations, "option_links": option_links, }, "ready_to_submit": validation["valid"], "blockers": blockers, "recommended_next_actions": actions, }, } return self._run_record_tool(profile, runner) def record_query( self, *, profile: str, query_mode: str, app_key: str, apply_id: int | None, page_num: int, page_size: int, requested_pages: int, scan_max_pages: int, query_key: str | None, filters: list[JSONObject], sorts: list[JSONObject], max_rows: int, max_columns: int | None, select_columns: list[str | int], amount_column: str | int | None, time_range: JSONObject, stat_policy: JSONObject, strict_full: bool, output_profile: str, list_type: int, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: resolved_mode = _resolve_query_mode(query_mode, apply_id=apply_id, amount_column=amount_column, time_range=time_range, stat_policy=stat_policy) if resolved_mode == "record": return self._record_query_record( profile=profile, app_key=app_key, apply_id=apply_id, select_columns=select_columns, max_columns=max_columns, output_profile=output_profile, list_type=list_type, ) if resolved_mode == "summary": return self._record_query_summary( profile=profile, app_key=app_key, page_num=page_num, page_size=page_size, requested_pages=requested_pages, scan_max_pages=scan_max_pages, query_key=query_key, filters=filters, sorts=sorts, max_rows=max_rows, max_columns=max_columns, select_columns=select_columns, amount_column=amount_column, time_range=time_range, stat_policy=stat_policy, strict_full=strict_full, output_profile=output_profile, list_type=list_type, view_key=view_key, view_name=view_name, ) return self._record_query_list( profile=profile, app_key=app_key, page_num=page_num, page_size=page_size, requested_pages=requested_pages, scan_max_pages=scan_max_pages, query_key=query_key, filters=filters, sorts=sorts, max_rows=max_rows, max_columns=max_columns, select_columns=select_columns, time_range=time_range, output_profile=output_profile, list_type=list_type, view_key=view_key, view_name=view_name, ) def record_aggregate( self, *, profile: str, app_key: str, group_by: list[str | int], amount_column: str | int | None, metrics: list[str], page_num: int, page_size: int, requested_pages: int, scan_max_pages: int, query_key: str | None, filters: list[JSONObject], sorts: list[JSONObject], time_range: JSONObject, time_bucket: str | None, max_groups: int, strict_full: bool, output_profile: str, list_type: int, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if max_groups <= 0: raise_tool_error(QingflowApiError.config_error("max_groups must be positive")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=False) view_selection = self._resolve_view_selection(profile, context, app_key, view_key=view_key, view_name=view_name) dept_member_cache: dict[int, set[int]] = {} group_fields = [self._resolve_field_selector(item, index, location="group_by") for item in group_by] amount_field = self._resolve_field_selector(amount_column, index, location="amount_column") if amount_column is not None else None time_field = self._resolve_time_range_column(time_range, index) match_rules = self._resolve_match_rules(context, filters, index) sort_rules = self._resolve_sorts(sorts, index) match_rules = self._append_time_range_filter(match_rules, time_range, time_field) metric_names = _normalize_metrics(metrics, include_sum=amount_field is not None) query_id = _query_id() pages_to_scan = min(max(requested_pages, 1), max(scan_max_pages, 1)) current_page = max(page_num, 1) scanned_pages = 0 scanned_records = 0 source_pages: list[int] = [] result_amount: int | None = None has_more = False group_stats: dict[str, JSONObject] = {} total_amount = 0.0 while scanned_pages < pages_to_scan: page = self._search_page( context, app_key=app_key, page_num=current_page, page_size=page_size, query_key=query_key, match_rules=match_rules, sorts=sort_rules, search_que_ids=None, list_type=list_type, ) scanned_pages += 1 source_pages.append(current_page) rows = page.get("list") items = rows if isinstance(rows, list) else [] if result_amount is None: result_amount = _effective_total(page, page_size) has_more = _page_has_more(page, current_page, page_size, len(items)) for item in items: if not isinstance(item, dict): continue answers = item.get("answers") answer_list = answers if isinstance(answers, list) else [] if not self._matches_view_selection( context, answer_list, view_selection=view_selection, dept_member_cache=dept_member_cache, ): continue scanned_records += 1 group_payload = { field.que_title: _extract_field_value(answer_list, field) for field in group_fields } if time_bucket and time_field is not None: group_payload["time_bucket"] = _to_time_bucket(_extract_field_value(answer_list, time_field), time_bucket) group_key = json.dumps(group_payload, ensure_ascii=False, sort_keys=True) bucket = group_stats.get(group_key) if bucket is None: bucket = {"group": group_payload, "count": 0, "amount_total": None, "metrics": {}} group_stats[group_key] = bucket bucket["count"] = int(bucket["count"]) + 1 amount_value = _coerce_amount(_extract_field_value(answer_list, amount_field)) if amount_field is not None else None if amount_value is not None: total_amount += amount_value bucket["amount_total"] = float(bucket.get("amount_total") or 0.0) + amount_value metrics_payload = bucket["metrics"] if isinstance(bucket.get("metrics"), dict) else {} for metric in metric_names: current_metric = metrics_payload.get(metric) metric_state = current_metric if isinstance(current_metric, dict) else {"count": 0, "sum": 0.0, "min": None, "max": None} metric_state["count"] = int(metric_state["count"]) + 1 if amount_value is not None: metric_state["sum"] = float(metric_state["sum"]) + amount_value metric_state["min"] = amount_value if metric_state["min"] is None else min(float(metric_state["min"]), amount_value) metric_state["max"] = amount_value if metric_state["max"] is None else max(float(metric_state["max"]), amount_value) metrics_payload[metric] = metric_state bucket["metrics"] = metrics_payload if len(group_stats) >= max_groups: break if len(group_stats) >= max_groups or not has_more: break current_page += 1 groups = [] for bucket in group_stats.values(): metrics_payload = bucket["metrics"] if isinstance(bucket.get("metrics"), dict) else {} rendered_metrics: JSONObject = {} for metric_name, metric_state in metrics_payload.items(): if not isinstance(metric_state, dict): continue count = int(metric_state.get("count", 0) or 0) amount_sum = float(metric_state.get("sum", 0.0) or 0.0) metric_result: JSONObject = {} if metric_name == "count": metric_result["value"] = count else: if metric_name == "sum": metric_result["value"] = amount_sum elif metric_name == "avg": metric_result["value"] = (amount_sum / count) if count else None elif metric_name == "min": metric_result["value"] = metric_state.get("min") elif metric_name == "max": metric_result["value"] = metric_state.get("max") rendered_metrics[metric_name] = metric_result groups.append( { "group": bucket["group"], "count": bucket["count"], "count_ratio": (int(bucket["count"]) / scanned_records) if scanned_records else 0, "amount_total": None if amount_field is None else _coerce_amount(bucket.get("amount_total")), "amount_ratio": None, "metrics": rendered_metrics, } ) groups.sort(key=lambda item: int(item["count"]), reverse=True) if amount_field is not None and total_amount > 0: for item in groups: amount_total = _coerce_amount(item.get("amount_total")) item["amount_ratio"] = (amount_total / total_amount) if amount_total is not None else None effective_result_amount = scanned_records if view_selection is not None else (result_amount or scanned_records) completeness = _build_completeness( result_amount=effective_result_amount, returned_items=len(groups), fetched_pages=scanned_pages, requested_pages=pages_to_scan, has_more=has_more, next_page_token=None, is_complete=not has_more and len(groups) < max_groups, omitted_items=max(0, effective_result_amount - len(groups)), extra={ "raw_scan_complete": not has_more, "scan_limit_hit": has_more, "scanned_pages": scanned_pages, "scan_limit": pages_to_scan, "output_page_complete": len(groups) < max_groups, "raw_next_page_token": None, "output_next_page_token": None, "stop_reason": "source_exhausted" if not has_more else "scan_limit", }, ) evidence = { "query_id": query_id, "app_key": app_key, "filters": _echo_filters(match_rules), "selected_columns": [field.que_title for field in group_fields], "time_range": time_range or None, "source_pages": source_pages, "view": _view_selection_payload(view_selection), } if strict_full and not bool(completeness.get("raw_scan_complete")): self._raise_need_more_data(completeness, evidence, "Aggregate result is incomplete; increase requested_pages or scan_max_pages.") response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "app_key": app_key, "view": _view_selection_payload(view_selection), "summary": { "total_count": scanned_records, "total_amount": total_amount if amount_field is not None else None, }, "groups": groups, "completeness": completeness, }, } if output_profile == "verbose": response["completeness"] = completeness response["evidence"] = evidence response["resolved_mappings"] = { "group_by": [_field_mapping_entry("group_by", field, requested=field.que_title) for field in group_fields], "amount_column": _field_mapping_entry("amount", amount_field, requested=amount_field.que_title) if amount_field is not None else None, } return response return self._run_record_tool(profile, runner) def record_create( self, *, profile: str, app_key: str, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, submit_type: int = 1, verify_write: bool = False, force_refresh_form: bool = False, ) -> JSONObject: if submit_type not in (0, 1): raise_tool_error(QingflowApiError.config_error("submit_type must be 0 or 1")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=force_refresh_form) if verify_write else None normalized_answers = self._resolve_answers( profile, context, app_key, answers=answers or [], fields=fields or {}, force_refresh_form=force_refresh_form, ) self._validate_record_write(app_key, normalized_answers) result = self.backend.request("POST", context, f"/app/{app_key}/apply", json_body={"type": submit_type, "answers": normalized_answers}) apply_id = _coerce_count(result.get("applyId")) if isinstance(result, dict) else None verification = self._verify_record_write_result( context, app_key=app_key, apply_id=apply_id, normalized_answers=normalized_answers, index=cast(FieldIndex, index), ) if verify_write and index is not None else None verified = True if verification is None else bool(verification.get("verified")) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "result": result, "normalized_answers": normalized_answers, "status": "completed" if verified else "verification_failed", "ok": verified, "apply_id": apply_id, "verify_write": verify_write, "write_verified": verified if verify_write else None, "verification": verification, "resource": {"type": "record", "apply_id": apply_id} if apply_id is not None else None, } return self._run_record_tool(profile, runner) def record_get( self, *, profile: str, app_key: str, apply_id: int, role: int, list_type: int | None, audit_node_id: int | None, ) -> JSONObject: normalized_apply_id = self._validate_app_and_record(app_key, apply_id) def runner(session_profile, context): params: JSONObject = {"role": role} if list_type is not None: params["listType"] = list_type if audit_node_id is not None: params["auditNodeId"] = audit_node_id result = self.backend.request("GET", context, f"/app/{app_key}/apply/{normalized_apply_id}", params=params) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "apply_id": normalized_apply_id, "result": result, } return self._run_record_tool(profile, runner) def record_search( self, *, profile: str, app_key: str, page_num: int, page_size: int, query_key: str | None, match_rules: list[JSONObject], sorts: list[JSONObject], search_que_ids: list[int] | None, list_type: int, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=False) view_selection = self._resolve_view_selection(profile, context, app_key, view_key=view_key, view_name=view_name) dept_member_cache: dict[int, set[int]] = {} result = self._search_page( context, app_key=app_key, page_num=page_num, page_size=page_size, query_key=query_key, match_rules=match_rules, sorts=sorts, search_que_ids=search_que_ids, list_type=list_type, ) rows = result.get("list") raw_rows = rows if isinstance(rows, list) else [] filtered_rows = [ item for item in raw_rows if isinstance(item, dict) and self._matches_view_selection( context, item.get("answers") if isinstance(item.get("answers"), list) else [], view_selection=view_selection, dept_member_cache=dept_member_cache, ) ] if isinstance(rows, list): result = dict(result) result["list"] = filtered_rows returned_rows = len(filtered_rows) reported_total = _coerce_count(result.get("total")) if reported_total is None: reported_total = _coerce_count(result.get("count")) effective_count = returned_rows if view_selection is not None else (max(reported_total or 0, returned_rows) if reported_total is not None else returned_rows) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "page": result, "list_type": list_type, "list_type_label": get_record_list_type_label(list_type), "view": _view_selection_payload(view_selection), "page_num": page_num, "page_size": page_size, "reported_total": reported_total, "returned_rows": returned_rows, "effective_count": effective_count, } return self._run_record_tool(profile, runner) def record_update( self, *, profile: str, app_key: str, apply_id: int, answers: list[JSONObject] | None = None, fields: JSONObject | None = None, role: int = 1, verify_write: bool = False, force_refresh_form: bool = False, ) -> JSONObject: normalized_apply_id = self._validate_app_and_record(app_key, apply_id) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=force_refresh_form) if verify_write else None normalized_answers = self._resolve_answers( profile, context, app_key, answers=answers or [], fields=fields or {}, force_refresh_form=force_refresh_form, ) self._validate_record_write(app_key, normalized_answers, apply_id=normalized_apply_id) result = self.backend.request( "POST", context, f"/app/{app_key}/apply/{normalized_apply_id}", json_body={"role": role, "answers": normalized_answers}, ) verification = self._verify_record_write_result( context, app_key=app_key, apply_id=normalized_apply_id, normalized_answers=normalized_answers, index=cast(FieldIndex, index), ) if verify_write and index is not None else None verified = True if verification is None else bool(verification.get("verified")) return self._attach_human_review_notice( { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "apply_id": normalized_apply_id, "result": result, "normalized_answers": normalized_answers, "status": "completed" if verified else "verification_failed", "ok": verified, "verify_write": verify_write, "write_verified": verified if verify_write else None, "verification": verification, "resource": {"type": "record", "apply_id": normalized_apply_id}, }, operation="update", target="record data", ) return self._run_record_tool(profile, runner) def record_delete(self, *, profile: str, app_key: str, apply_id: int, list_type: int) -> JSONObject: normalized_apply_id = self._validate_app_and_record(app_key, apply_id) def runner(session_profile, context): result = self.backend.request( "DELETE", context, f"/app/{app_key}/apply", json_body={"type": list_type, "applyIds": [normalized_apply_id]}, ) return self._attach_human_review_notice( { "profile": profile, "ws_id": session_profile.selected_ws_id, "request_route": self._request_route_payload(context), "app_key": app_key, "apply_id": normalized_apply_id, "result": result, }, operation="delete", target="record data", ) return self._run_record_tool(profile, runner) def _record_query_record( self, *, profile: str, app_key: str, apply_id: int | None, select_columns: list[str | int], max_columns: int | None, output_profile: str, list_type: int, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if apply_id is None or apply_id <= 0: raise_tool_error(QingflowApiError.config_error("apply_id is required")) if not select_columns: raise_tool_error(QingflowApiError.config_error("select_columns is required in record mode")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=False) resolved_column_cap = _bounded_column_limit( max_columns, default_limit=MAX_RECORD_COLUMN_LIMIT, hard_limit=MAX_RECORD_COLUMN_LIMIT, ) selected_fields = self._resolve_select_columns( select_columns, index, max_columns=max_columns, default_limit=MAX_RECORD_COLUMN_LIMIT, ) result = self.backend.request( "GET", context, f"/app/{app_key}/apply/{apply_id}", params={"role": 1, "listType": list_type}, ) answers = result.get("answers") if isinstance(result, dict) else None answer_list = answers if isinstance(answers, list) else [] row = _build_flat_row(answer_list, selected_fields, apply_id=apply_id) completeness = _build_completeness( result_amount=1, returned_items=1, fetched_pages=1, requested_pages=1, has_more=False, next_page_token=None, is_complete=True, omitted_items=0, extra={}, ) evidence = { "query_id": _query_id(), "app_key": app_key, "filters": [], "selected_columns": [field.que_title for field in selected_fields], "time_range": None, "source_pages": [1], } response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "mode": "record", "source_tool": "record_get", "record": { "apply_id": apply_id, "row": row, "applied_limits": { "column_cap": resolved_column_cap, "selected_columns": [field.que_title for field in selected_fields], }, }, }, "output_profile": output_profile, "next_page_token": None, } if output_profile == "verbose": response["completeness"] = completeness response["evidence"] = evidence response["resolved_mappings"] = { "select_columns": [_field_mapping_entry("row", field, requested=field.que_title) for field in selected_fields] } return response return self._run_record_tool(profile, runner) def _record_query_list( self, *, profile: str, app_key: str, page_num: int, page_size: int, requested_pages: int, scan_max_pages: int, query_key: str | None, filters: list[JSONObject], sorts: list[JSONObject], max_rows: int, max_columns: int | None, select_columns: list[str | int], time_range: JSONObject, output_profile: str, list_type: int, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if not select_columns: raise_tool_error(QingflowApiError.config_error("select_columns is required in list mode")) if max_rows <= 0: raise_tool_error(QingflowApiError.config_error("max_rows must be positive")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=False) view_selection = self._resolve_view_selection(profile, context, app_key, view_key=view_key, view_name=view_name) dept_member_cache: dict[int, set[int]] = {} resolved_column_cap = _bounded_column_limit( max_columns, default_limit=MAX_LIST_COLUMN_LIMIT, hard_limit=MAX_LIST_COLUMN_LIMIT, ) selected_fields = self._resolve_select_columns( select_columns, index, max_columns=max_columns, default_limit=MAX_LIST_COLUMN_LIMIT, ) selected_field_batches = _chunk_fields(selected_fields, BACKEND_LIST_SEARCH_FIELD_LIMIT) primary_search_que_ids: list[int] | None = [field.que_id for field in selected_field_batches[0]] if view_selection is not None and not _view_selection_supported_by_search_ids(view_selection, primary_search_que_ids): primary_search_que_ids = None remaining_field_batches: list[list[FormField]] = [] selected_fields_from_primary = selected_fields else: remaining_field_batches = selected_field_batches[1:] primary_search_que_ids = primary_search_que_ids or None primary_que_ids = set(primary_search_que_ids or []) selected_fields_from_primary = [field for field in selected_fields if field.que_id in primary_que_ids] time_field = self._resolve_time_range_column(time_range, index) match_rules = self._resolve_match_rules(context, filters, index) sort_rules = self._resolve_sorts(sorts, index) match_rules = self._append_time_range_filter(match_rules, time_range, time_field) scan_limit = min(max(requested_pages, 1), max(scan_max_pages, 1)) current_page = max(page_num, 1) scanned_pages = 0 rows: list[JSONObject] = [] matched_records = 0 result_amount: int | None = None reported_total: int | None = None has_more = False source_pages: list[int] = [] while scanned_pages < scan_limit and len(rows) < max_rows: page = self._search_page( context, app_key=app_key, page_num=current_page, page_size=page_size, query_key=query_key, match_rules=match_rules, sorts=sort_rules, search_que_ids=primary_search_que_ids, list_type=list_type, ) scanned_pages += 1 source_pages.append(current_page) page_rows = page.get("list") items = page_rows if isinstance(page_rows, list) else [] if result_amount is None: reported_total = _coerce_count(page.get("total")) if reported_total is None: reported_total = _coerce_count(page.get("count")) result_amount = _effective_total(page, page_size) has_more = _page_has_more(page, current_page, page_size, len(items)) page_output_rows: list[JSONObject] = [] for item in items: if not isinstance(item, dict): continue answers = item.get("answers") answer_list = answers if isinstance(answers, list) else [] if not self._matches_view_selection( context, answer_list, view_selection=view_selection, dept_member_cache=dept_member_cache, ): continue matched_records += 1 apply_id = _coerce_count(item.get("applyId")) or _coerce_count(item.get("id")) row = _build_flat_row(answer_list, selected_fields_from_primary, apply_id=apply_id) rows.append(row) page_output_rows.append(row) if len(rows) >= max_rows: break if page_output_rows and remaining_field_batches: page_row_map = { _coerce_count(row.get("apply_id")): row for row in page_output_rows if isinstance(row, dict) and _coerce_count(row.get("apply_id")) is not None } for batch in remaining_field_batches: extra_page = self._search_page( context, app_key=app_key, page_num=current_page, page_size=page_size, query_key=query_key, match_rules=match_rules, sorts=sort_rules, search_que_ids=[field.que_id for field in batch], list_type=list_type, ) extra_rows = extra_page.get("list") extra_items = extra_rows if isinstance(extra_rows, list) else [] for extra_item in extra_items: if not isinstance(extra_item, dict): continue apply_id = _coerce_count(extra_item.get("applyId")) or _coerce_count(extra_item.get("id")) if apply_id is None or apply_id not in page_row_map: continue extra_answers = extra_item.get("answers") extra_answer_list = extra_answers if isinstance(extra_answers, list) else [] partial_row = _build_flat_row(extra_answer_list, batch, apply_id=apply_id) partial_row.pop("apply_id", None) page_row_map[apply_id].update(partial_row) if not has_more: break current_page += 1 effective_result_amount = matched_records if view_selection is not None else (result_amount or len(rows)) completeness = _build_completeness( result_amount=effective_result_amount, returned_items=len(rows), fetched_pages=scanned_pages, requested_pages=scan_limit, has_more=has_more, next_page_token=None, is_complete=not has_more and len(rows) < max_rows, omitted_items=max(0, effective_result_amount - len(rows)), extra={}, ) evidence = { "query_id": _query_id(), "app_key": app_key, "filters": _echo_filters(match_rules), "selected_columns": [field.que_title for field in selected_fields], "time_range": time_range or None, "source_pages": source_pages, "view": _view_selection_payload(view_selection), } response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "mode": "list", "source_tool": "record_search", "view": _view_selection_payload(view_selection), "list": { "rows": rows, "pagination": { "page_num": page_num, "page_size": page_size, "requested_pages": scan_limit, "result_amount": effective_result_amount, "returned_items": len(rows), }, "applied_limits": { "row_cap": max_rows, "column_cap": resolved_column_cap, "selected_columns": [field.que_title for field in selected_fields], }, }, }, "output_profile": output_profile, "next_page_token": None, } if output_profile == "verbose": response["completeness"] = completeness evidence["backend_reported_total"] = reported_total response["evidence"] = evidence response["resolved_mappings"] = { "select_columns": [_field_mapping_entry("row", field, requested=field.que_title) for field in selected_fields], "filters": [_field_mapping_entry("filter", entry["field"], requested=entry["requested"]) for entry in self._resolve_filter_field_entries(filters, index)], "time_range": _field_mapping_entry("time", time_field, requested=time_field.que_title) if time_field is not None else None, } return response return self._run_record_tool(profile, runner) def _record_query_summary( self, *, profile: str, app_key: str, page_num: int, page_size: int, requested_pages: int, scan_max_pages: int, query_key: str | None, filters: list[JSONObject], sorts: list[JSONObject], max_rows: int, max_columns: int | None, select_columns: list[str | int], amount_column: str | int | None, time_range: JSONObject, stat_policy: JSONObject, strict_full: bool, output_profile: str, list_type: int, view_key: str | None = None, view_name: str | None = None, ) -> JSONObject: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) def runner(session_profile, context): index = self._get_field_index(profile, context, app_key, force_refresh=False) view_selection = self._resolve_view_selection(profile, context, app_key, view_key=view_key, view_name=view_name) dept_member_cache: dict[int, set[int]] = {} amount_field = self._resolve_field_selector(amount_column, index, location="amount_column") if amount_column is not None else None time_field = self._resolve_time_range_column(time_range, index) resolved_column_cap = _bounded_column_limit( max_columns, default_limit=MAX_SUMMARY_PREVIEW_COLUMN_LIMIT, hard_limit=MAX_SUMMARY_PREVIEW_COLUMN_LIMIT, ) preview_fields = self._resolve_summary_preview_fields(select_columns, index, amount_field, time_field, max_columns=max_columns) match_rules = self._resolve_match_rules(context, filters, index) sort_rules = self._resolve_sorts(sorts, index) match_rules = self._append_time_range_filter(match_rules, time_range, time_field) include_negative = bool(stat_policy.get("include_negative", True)) include_null = bool(stat_policy.get("include_null", False)) scan_limit = min(max(requested_pages, 1), max(scan_max_pages, 1)) current_page = max(page_num, 1) scanned_pages = 0 scanned_records = 0 result_amount: int | None = None has_more = False source_pages: list[int] = [] preview_rows: list[JSONObject] = [] total_amount = 0.0 missing_count = 0 by_day: dict[str, JSONObject] = {} while scanned_pages < scan_limit: page = self._search_page( context, app_key=app_key, page_num=current_page, page_size=page_size, query_key=query_key, match_rules=match_rules, sorts=sort_rules, search_que_ids=None, list_type=list_type, ) scanned_pages += 1 source_pages.append(current_page) page_rows = page.get("list") items = page_rows if isinstance(page_rows, list) else [] if result_amount is None: result_amount = _effective_total(page, page_size) has_more = _page_has_more(page, current_page, page_size, len(items)) for item in items: if not isinstance(item, dict): continue answers = item.get("answers") answer_list = answers if isinstance(answers, list) else [] if not self._matches_view_selection( context, answer_list, view_selection=view_selection, dept_member_cache=dept_member_cache, ): continue scanned_records += 1 if len(preview_rows) < max_rows: apply_id = _coerce_count(item.get("applyId")) or _coerce_count(item.get("id")) preview_rows.append(_build_flat_row(answer_list, preview_fields, apply_id=apply_id)) amount_value = _coerce_amount(_extract_field_value(answer_list, amount_field)) if amount_field is not None else None if amount_field is not None: if amount_value is None: if not include_null: missing_count += 1 elif include_negative or amount_value >= 0: total_amount += amount_value day_key = _to_time_bucket(_extract_field_value(answer_list, time_field), "day") if time_field is not None else "all" bucket = by_day.get(day_key) if bucket is None: bucket = {"day": day_key, "count": 0, "amount_total": 0.0 if amount_field is not None else None} by_day[day_key] = bucket bucket["count"] = int(bucket["count"]) + 1 if amount_field is not None and amount_value is not None and (include_negative or amount_value >= 0): bucket["amount_total"] = float(bucket.get("amount_total") or 0.0) + amount_value if not has_more: break current_page += 1 raw_scan_complete = not has_more effective_result_amount = scanned_records if view_selection is not None else max(result_amount or 0, scanned_records) completeness = _build_completeness( result_amount=effective_result_amount, returned_items=len(preview_rows), fetched_pages=scanned_pages, requested_pages=scan_limit, has_more=has_more, next_page_token=None, is_complete=raw_scan_complete and len(preview_rows) < max_rows, omitted_items=max(0, effective_result_amount - len(preview_rows)), extra={ "raw_scan_complete": raw_scan_complete, "scan_limit_hit": has_more, "scanned_pages": scanned_pages, "scan_limit": scan_limit, "output_page_complete": len(preview_rows) < max_rows, "raw_next_page_token": None, "output_next_page_token": None, "stop_reason": "source_exhausted" if raw_scan_complete else "scan_limit", }, ) evidence = { "query_id": _query_id(), "app_key": app_key, "filters": _echo_filters(match_rules), "selected_columns": [field.que_title for field in preview_fields], "time_range": time_range or None, "source_pages": source_pages, "view": _view_selection_payload(view_selection), } if strict_full and not raw_scan_complete: self._raise_need_more_data(completeness, evidence, "Summary is incomplete; increase requested_pages or scan_max_pages.") response: JSONObject = { "profile": profile, "ws_id": session_profile.selected_ws_id, "ok": True, "request_route": self._request_route_payload(context), "data": { "mode": "summary", "source_tool": "record_search", "view": _view_selection_payload(view_selection), "summary": { "summary": { "total_count": scanned_records, "total_amount": total_amount if amount_field is not None else None, "by_day": sorted(by_day.values(), key=lambda item: str(item.get("day"))), "missing_count": missing_count, }, "rows": preview_rows, "completeness": completeness, "applied_limits": { "row_cap": max_rows, "column_cap": resolved_column_cap, "selected_columns": [field.que_title for field in preview_fields], }, }, }, "output_profile": output_profile, "next_page_token": None, } if output_profile == "verbose": response["completeness"] = completeness response["evidence"] = evidence response["resolved_mappings"] = { "select_columns": [_field_mapping_entry("row", field, requested=field.que_title) for field in preview_fields], "amount_column": _field_mapping_entry("amount", amount_field, requested=amount_field.que_title) if amount_field is not None else None, "time_range": _field_mapping_entry("time", time_field, requested=time_field.que_title) if time_field is not None else None, } return response return self._run_record_tool(profile, runner) def _get_form_schema(self, profile: str, context, app_key: str, *, force_refresh: bool) -> JSONObject: # type: ignore[no-untyped-def] cache_key = (profile, app_key) 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 = _normalize_form_schema(schema) self._form_cache[cache_key] = normalized return normalized def _get_field_index(self, profile: str, context, app_key: str, *, force_refresh: bool) -> FieldIndex: # type: ignore[no-untyped-def] return _build_field_index(self._get_form_schema(profile, context, app_key, force_refresh=force_refresh)) def _get_view_list(self, profile: str, context, app_key: str) -> list[JSONObject]: # type: ignore[no-untyped-def] cache_key = (profile, app_key) if cache_key in self._view_list_cache: return self._view_list_cache[cache_key] payload = self.backend.request("GET", context, f"/app/{app_key}/view/viewList") normalized = _normalize_view_list(payload) self._view_list_cache[cache_key] = normalized return normalized def _get_view_config(self, profile: str, context, view_key: str) -> JSONObject: # type: ignore[no-untyped-def] cache_key = (profile, view_key) if cache_key in self._view_config_cache: return self._view_config_cache[cache_key] payload = self.backend.request("GET", context, f"/view/{view_key}/viewConfig") normalized = payload if isinstance(payload, dict) else {} self._view_config_cache[cache_key] = normalized return normalized def _resolve_view_selection( self, profile: str, context, # type: ignore[no-untyped-def] app_key: str, *, view_key: str | None, view_name: str | None, ) -> ViewSelection | None: requested_key = _normalize_optional_text(view_key) requested_name = _normalize_optional_text(view_name) if requested_key is None and requested_name is None: return None views = self._get_view_list(profile, context, app_key) selected: JSONObject | None = None if requested_key is not None: selected = next((item for item in views if _normalize_optional_text(item.get("viewKey")) == requested_key), None) if selected is None and requested_name is not None: exact_matches = [item for item in views if _normalize_optional_text(item.get("viewName")) == requested_name] if len(exact_matches) > 1: raise_tool_error(QingflowApiError.config_error(f"view_name '{requested_name}' is ambiguous; pass view_key instead")) selected = exact_matches[0] if exact_matches else None resolved_view_key = requested_key or _normalize_optional_text(selected.get("viewKey") if isinstance(selected, dict) else None) if not resolved_view_key: raise_tool_error(QingflowApiError.config_error(f"cannot resolve view '{requested_name or requested_key}' for app {app_key}")) config = self._get_view_config(profile, context, resolved_view_key) resolved_view_name = ( _normalize_optional_text(config.get("viewgraphName")) or _normalize_optional_text(selected.get("viewName") if isinstance(selected, dict) else None) or requested_name or resolved_view_key ) return ViewSelection( view_key=resolved_view_key, view_name=resolved_view_name, conditions=_compile_view_conditions(config), ) def _get_department_member_ids(self, context, dept_id: int) -> set[int]: # type: ignore[no-untyped-def] page_num = 1 page_size = 200 member_ids: set[int] = set() while True: payload = self.backend.request( "GET", context, "/contact", params={"pageNum": page_num, "pageSize": page_size, "deptId": dept_id, "containDisable": False}, ) result = payload.get("result") if isinstance(payload, dict) else None rows = result if isinstance(result, list) else [] for item in rows: if not isinstance(item, dict): continue member_id = _coerce_count(item.get("uid", item.get("id"))) if member_id is not None: member_ids.add(member_id) page_amount = _coerce_count(payload.get("pageAmount")) if isinstance(payload, dict) else None if page_amount is not None: if page_num >= page_amount: break elif len(rows) < page_size: break page_num += 1 return member_ids def _matches_view_selection( self, context, # type: ignore[no-untyped-def] answer_list: list[JSONValue], *, view_selection: ViewSelection | None, dept_member_cache: dict[int, set[int]], ) -> bool: if view_selection is None or not view_selection.conditions: return True for group in view_selection.conditions: if all( _match_view_condition( answer_list, condition, dept_member_cache=dept_member_cache, dept_member_resolver=lambda dept_id: self._get_department_member_ids(context, dept_id), ) for condition in group ): return True return False def _search_page( self, context, # type: ignore[no-untyped-def] *, app_key: str, page_num: int, page_size: int, query_key: str | None, match_rules: list[JSONObject], sorts: list[JSONObject], search_que_ids: list[int] | None, list_type: int, ) -> JSONObject: body: JSONObject = {"type": list_type, "pageNum": page_num, "pageSize": page_size} if query_key: body["queryKey"] = query_key if match_rules: queries = [] normalized_match_rules = [] for item in match_rules: if not isinstance(item, dict): continue queries.extend(_normalize_list_query_rules(item)) normalized_match_rules.extend(_normalize_list_match_rules(item)) if queries: body["queries"] = queries if normalized_match_rules: body["matchRules"] = normalized_match_rules if sorts: normalized_sorts = [rule for rule in (_normalize_list_sort_rule(item) for item in sorts if isinstance(item, dict)) if rule] if normalized_sorts: body["sorts"] = normalized_sorts if search_que_ids: body["searchQueIds"] = search_que_ids result = self.backend.request("POST", context, f"/app/{app_key}/apply/filter", json_body=body) return result if isinstance(result, dict) else {} def _resolve_answers( self, profile: str, context, # type: ignore[no-untyped-def] app_key: str, *, answers: list[JSONObject], fields: JSONObject, force_refresh_form: bool, ) -> list[JSONObject]: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if not answers and not fields: raise_tool_error(QingflowApiError.config_error("either answers or fields is required")) if not answers and fields: index = self._get_field_index(profile, context, app_key, force_refresh=force_refresh_form) return [self._build_field_answer(profile, context, index, key, value) for key, value in fields.items() if value is not None] if answers and not _answers_need_resolution(answers) and not fields: return answers index = self._get_field_index(profile, context, app_key, force_refresh=force_refresh_form) normalized = [self._normalize_answer_item(profile, context, index, item) for item in answers] if fields: normalized.extend(self._build_field_answer(profile, context, index, key, value) for key, value in fields.items() if value is not None) return normalized def _normalize_answer_item(self, profile: str, context, index: FieldIndex, item: JSONObject) -> JSONObject: # type: ignore[no-untyped-def] field = self._resolve_field_from_answer_item(item, index) if field.que_type in SUBTABLE_QUE_TYPES: table_values_input: JSONValue = item.get("tableValues") if table_values_input is None: table_values_input = item.get("rows", item.get("value", item.get("values"))) normalized_rows = self._normalize_subtable_rows(profile, context, field, table_values_input, location=field.que_title) payload: JSONObject = { "queId": field.que_id, "queType": field.que_type or 18, "values": [], "tableValues": normalized_rows, } previous_row_ordinals = item.get("previousTableRowOrdinalList") if isinstance(previous_row_ordinals, list): payload["previousTableRowOrdinalList"] = previous_row_ordinals return payload self._raise_if_verify_unsupported_write_field(field, item, location=field.que_title) if "values" in item and isinstance(item["values"], list): values = item["values"] elif "value" in item: values = [item["value"]] else: raise RecordInputError( message=f"answer for field '{field.que_title}' requires value or values", error_code="MISSING_VALUE", fix_hint="Pass value for scalar fields, or values for multi-value fields.", details={"location": field.que_title, "field": _field_ref_payload(field), "expected_format": _write_format_for_field(field)}, ) return { "queId": field.que_id, "queType": field.que_type or 2, "values": self._normalize_field_values(profile, context, field, values), "tableValues": item.get("tableValues") if isinstance(item.get("tableValues"), list) else [], } def _build_field_answer(self, profile: str, context, index: FieldIndex, field_selector: str, raw_value: JSONValue) -> JSONObject: # type: ignore[no-untyped-def] field = self._resolve_field_selector(field_selector, index, location="fields") if field.que_type in SUBTABLE_QUE_TYPES: return { "queId": field.que_id, "queType": field.que_type or 18, "values": [], "tableValues": self._normalize_subtable_rows(profile, context, field, raw_value, location=str(field_selector)), } self._raise_if_verify_unsupported_write_field(field, raw_value, location=str(field_selector)) values = raw_value if isinstance(raw_value, list) and field.que_type in MULTI_SELECT_QUE_TYPES else [raw_value] return { "queId": field.que_id, "queType": field.que_type or 2, "values": self._normalize_field_values(profile, context, field, values), "tableValues": [], } def _raise_if_verify_unsupported_write_field(self, field: FormField, raw_value: JSONValue, *, location: str) -> None: if field.que_type not in VERIFY_UNSUPPORTED_WRITE_QUE_TYPES: return raise RecordInputError( message=f"field '{field.que_title}' uses unsupported direct writes", error_code="UNSUPPORTED_WRITE_FORMAT", fix_hint=_unsupported_write_fix_hint(field.que_type), details={ "location": location, "field": _field_ref_payload(field), "expected_format": _write_format_for_field(field), "received_value": raw_value, }, ) def _normalize_field_values( self, profile: str, context, # type: ignore[no-untyped-def] field: FormField, raw_values: list[JSONValue], ) -> list[JSONObject]: if field.que_type in SINGLE_SELECT_QUE_TYPES: return [_option_value(raw_values[0], field)] if field.que_type in MULTI_SELECT_QUE_TYPES: return [_option_value(value, field) for value in raw_values] if field.que_type in BOOLEAN_QUE_TYPES: return [{"value": _boolean_display(raw_values[0])}] if field.que_type in MEMBER_QUE_TYPES: return [_member_value(profile, value) for value in _expand_values(raw_values)] if field.que_type in DEPARTMENT_QUE_TYPES: return [self._department_value_from_selector(context, value) for value in _expand_values(raw_values)] if field.que_type in ATTACHMENT_QUE_TYPES: return [_attachment_value(value) for value in _expand_values(raw_values)] if field.que_type in RELATION_QUE_TYPES: return [_relation_value(value) for value in _expand_values(raw_values)] return [{"value": _stringify_json(raw_values[0])}] def _normalize_subtable_rows( self, profile: str, context, # type: ignore[no-untyped-def] table_field: FormField, raw_rows: JSONValue, *, location: str, ) -> list[list[JSONObject]]: row_values = raw_rows if isinstance(raw_rows, dict): if "rows" in raw_rows: row_values = raw_rows.get("rows") elif "tableValues" in raw_rows: row_values = raw_rows.get("tableValues") if row_values is None: return [] if not isinstance(row_values, list): raise RecordInputError( message=f"field '{table_field.que_title}' requires subtable rows", error_code="INVALID_SUBTABLE_VALUE", fix_hint="Pass subtables as a list of row objects or a native tableValues array.", details={"location": location, "field": _field_ref_payload(table_field), "expected_format": _write_format_for_field(table_field), "received_value": raw_rows}, ) subtable_index = self._subtable_field_index(table_field) normalized_rows: list[list[JSONObject]] = [] for row_ordinal, row in enumerate(row_values): normalized_rows.append( self._normalize_subtable_row( profile, context, table_field, subtable_index, row, row_ordinal=row_ordinal, location=location, ) ) return normalized_rows def _normalize_subtable_row( self, profile: str, context, # type: ignore[no-untyped-def] table_field: FormField, subtable_index: FieldIndex, row: JSONValue, *, row_ordinal: int, location: str, ) -> list[JSONObject]: row_id: int | None = None normalized_cells: list[JSONObject] = [] if isinstance(row, dict): row_id = _coerce_count(row.get("__row_id__", row.get("row_id", row.get("rowId")))) if isinstance(row.get("answers"), list): normalized_cells = [ self._normalize_subtable_answer_item(profile, context, subtable_index, item, location=f"{location}[{row_ordinal}]") for item in row["answers"] if isinstance(item, dict) ] elif isinstance(row.get("fields"), dict): normalized_cells = [ self._build_subtable_field_answer(profile, context, subtable_index, key, value, location=f"{location}[{row_ordinal}]") for key, value in row["fields"].items() if value is not None ] else: row_fields = { key: value for key, value in row.items() if key not in {"__row_id__", "row_id", "rowId", "answers", "fields", "rows", "tableValues"} } normalized_cells = [ self._build_subtable_field_answer(profile, context, subtable_index, key, value, location=f"{location}[{row_ordinal}]") for key, value in row_fields.items() if value is not None ] elif isinstance(row, list): normalized_cells = [ self._normalize_subtable_answer_item(profile, context, subtable_index, item, location=f"{location}[{row_ordinal}]") for item in row if isinstance(item, dict) ] else: raise RecordInputError( message=f"field '{table_field.que_title}' row {row_ordinal + 1} has unsupported shape", error_code="INVALID_SUBTABLE_ROW", fix_hint="Pass each subtable row as an object keyed by subfield title, or a list of native answer objects.", details={"location": location, "field": _field_ref_payload(table_field), "row_ordinal": row_ordinal, "received_value": row}, ) if not normalized_cells: raise RecordInputError( message=f"field '{table_field.que_title}' row {row_ordinal + 1} is empty", error_code="EMPTY_SUBTABLE_ROW", fix_hint="Provide at least one subfield value in each subtable row, or omit the row entirely.", details={"location": location, "field": _field_ref_payload(table_field), "row_ordinal": row_ordinal, "received_value": row}, ) if row_id is not None: for cell in normalized_cells: cell.setdefault("rowId", row_id) return normalized_cells def _normalize_subtable_answer_item( self, profile: str, context, # type: ignore[no-untyped-def] subtable_index: FieldIndex, item: JSONObject, *, location: str, ) -> JSONObject: field = self._resolve_field_from_answer_item(item, subtable_index) if field.que_type in SUBTABLE_QUE_TYPES: raise RecordInputError( message=f"field '{field.que_title}' uses unsupported nested subtable writes", error_code="UNSUPPORTED_WRITE_FORMAT", fix_hint="Nested subtable writes are not supported in app-user tools.", details={"location": location, "field": _field_ref_payload(field), "expected_format": _write_format_for_field(field), "received_value": item}, ) self._raise_if_verify_unsupported_write_field(field, item, location=location) if "values" in item and isinstance(item["values"], list): values = item["values"] elif "value" in item: values = [item["value"]] else: raise RecordInputError( message=f"subtable field '{field.que_title}' requires value or values", error_code="MISSING_VALUE", fix_hint="Pass value for scalar subtable fields, or values for multi-value subtable fields.", details={"location": location, "field": _field_ref_payload(field), "expected_format": _write_format_for_field(field)}, ) payload: JSONObject = { "queId": field.que_id, "queType": field.que_type or 2, "values": self._normalize_field_values(profile, context, field, values), "tableValues": [], } row_id = _coerce_count(item.get("rowId", item.get("row_id"))) if row_id is not None: payload["rowId"] = row_id return payload def _build_subtable_field_answer( self, profile: str, context, # type: ignore[no-untyped-def] subtable_index: FieldIndex, field_selector: str, raw_value: JSONValue, *, location: str, ) -> JSONObject: field = self._resolve_field_selector(field_selector, subtable_index, location=location) if field.que_type in SUBTABLE_QUE_TYPES: raise RecordInputError( message=f"field '{field.que_title}' uses unsupported nested subtable writes", error_code="UNSUPPORTED_WRITE_FORMAT", fix_hint="Nested subtable writes are not supported in app-user tools.", details={"location": location, "field": _field_ref_payload(field), "expected_format": _write_format_for_field(field), "received_value": raw_value}, ) self._raise_if_verify_unsupported_write_field(field, raw_value, location=location) values = raw_value if isinstance(raw_value, list) and field.que_type in MULTI_SELECT_QUE_TYPES else [raw_value] return { "queId": field.que_id, "queType": field.que_type or 2, "values": self._normalize_field_values(profile, context, field, values), "tableValues": [], } def _subtable_field_index(self, table_field: FormField) -> FieldIndex: raw = table_field.raw if isinstance(table_field.raw, dict) else {} schema: JSONObject = {} if isinstance(raw.get("subQuestions"), list): schema["formQues"] = [raw["subQuestions"]] elif isinstance(raw.get("innerQuestions"), list): schema["formQues"] = raw["innerQuestions"] index = _build_field_index(schema) if not index.by_id: raise RecordInputError( message=f"field '{table_field.que_title}' does not expose subtable column metadata", error_code="SUBTABLE_SCHEMA_UNAVAILABLE", fix_hint="Refresh the form schema and ensure the subtable columns are visible before writing.", details={"field": _field_ref_payload(table_field), "raw": raw}, ) return index def _subtable_field_index_optional(self, table_field: FormField | None) -> FieldIndex | None: if table_field is None: return None try: return self._subtable_field_index(table_field) except RecordInputError: return None def _run_record_tool(self, profile: str, func, *, require_workspace: bool = True): # type: ignore[no-untyped-def] try: return self._run(profile, func, require_workspace=require_workspace) except RecordInputError as error: raise_tool_error( QingflowApiError( category="config", message=error.message, backend_code=error.error_code, details={ "error_code": error.error_code, "fix_hint": error.fix_hint, "record_input": error.to_dict(), }, ) ) def _request_route_payload(self, context) -> JSONObject: # type: ignore[no-untyped-def] describe_route = getattr(self.backend, "describe_route", None) if callable(describe_route): payload = describe_route(context) if isinstance(payload, dict): return payload return { "base_url": getattr(context, "base_url", None), "qf_version": getattr(context, "qf_version", None), "qf_version_source": getattr(context, "qf_version_source", None) or ("context" if getattr(context, "qf_version", None) else "unknown"), } def _resolve_field_selector(self, selector: str | int | None, index: FieldIndex, *, location: str) -> FormField: if selector is None: raise RecordInputError( message=f"{location} requires a field selector", error_code="MISSING_FIELD_SELECTOR", fix_hint="Pass an exact field title or queId.", ) requested = str(selector).strip() requested_key = _normalize_field_lookup_key(requested) if not requested: raise RecordInputError( message=f"{location} contains an empty field selector", error_code="EMPTY_FIELD_SELECTOR", fix_hint="Pass an exact field title or queId.", ) if requested.isdigit(): field = index.by_id.get(str(int(requested))) if field is not None: return field raise RecordInputError( message=f"{location} references unknown queId '{requested}'", error_code="FIELD_NOT_FOUND", fix_hint="Use record_field_resolve to confirm the exact field id.", details={"location": location, "requested": requested, "requested_key": requested_key}, ) matches = index.by_title.get(requested_key, []) if len(matches) == 1: return matches[0] if len(matches) > 1: raise RecordInputError( message=f"{location} field '{requested}' is ambiguous", error_code="AMBIGUOUS_FIELD", fix_hint="Use numeric queId, or resolve the field first with record_field_resolve.", details={ "location": location, "requested": requested, "requested_key": requested_key, "matched_via": "title", "candidates": [_field_ref_payload(item) for item in matches], }, ) alias_matches = index.by_alias.get(requested_key, []) if len(alias_matches) == 1: return alias_matches[0] if len(alias_matches) > 1: raise RecordInputError( message=f"{location} field '{requested}' is ambiguous", error_code="AMBIGUOUS_FIELD", fix_hint="Use a more specific field title, or run record_field_resolve to inspect the alias candidates.", details={ "location": location, "requested": requested, "requested_key": requested_key, "matched_via": "alias", "candidates": [_field_ref_payload(item) for item in alias_matches], }, ) raise RecordInputError( message=f"{location} cannot resolve field '{requested}'", error_code="FIELD_NOT_FOUND", fix_hint="Use record_field_resolve to confirm the exact field title.", details={ "location": location, "requested": requested, "requested_key": requested_key, "suggestions": self._score_field_matches(requested, index, fuzzy=True, top_k=5), }, ) def _resolve_field_from_answer_item(self, item: JSONObject, index: FieldIndex) -> FormField: for key in ("queId", "que_id", "queTitle", "que_title"): if key in item: return self._resolve_field_selector(cast(str | int, item[key]), index, location="answers") raise RecordInputError( message="answer item requires queId/que_id or queTitle/que_title", error_code="MISSING_FIELD_SELECTOR", fix_hint="Provide a field selector in each answer item.", ) def _resolve_select_columns( self, selectors: list[str | int], index: FieldIndex, *, max_columns: int | None, default_limit: int, ) -> list[FormField]: if not selectors: raise_tool_error(QingflowApiError.config_error("select_columns is required")) limit = _bounded_column_limit(max_columns, default_limit=default_limit, hard_limit=default_limit) fields: list[FormField] = [] seen: set[int] = set() for selector in selectors: field = self._resolve_field_selector(selector, index, location="select_columns") if field.que_id in seen: continue fields.append(field) seen.add(field.que_id) if len(fields) >= limit: break return fields def _resolve_summary_preview_fields( self, selectors: list[str | int], index: FieldIndex, amount_field: FormField | None, time_field: FormField | None, *, max_columns: int | None, ) -> list[FormField]: if selectors: return self._resolve_select_columns( selectors, index, max_columns=max_columns, default_limit=MAX_SUMMARY_PREVIEW_COLUMN_LIMIT, ) limit = _bounded_column_limit( max_columns, default_limit=MAX_SUMMARY_PREVIEW_COLUMN_LIMIT, hard_limit=MAX_SUMMARY_PREVIEW_COLUMN_LIMIT, ) candidates: list[FormField] = [] title_candidate = _pick_title_field(index) for field in (title_candidate, amount_field, time_field): if field is None or any(existing.que_id == field.que_id for existing in candidates): continue candidates.append(field) if not candidates: candidates = list(index.by_id.values())[:limit] return candidates[:limit] def _resolve_match_rules(self, context, filters: list[JSONObject], index: FieldIndex) -> list[JSONObject]: # type: ignore[no-untyped-def] rules: list[JSONObject] = [] for item in filters: if not isinstance(item, dict): continue selector = _extract_filter_selector(item) field = self._resolve_field_selector(cast(str | int | None, selector), index, location="filters") if selector is not None else None if field is not None and field.que_type in DEPARTMENT_QUE_TYPES: department_rule = self._build_department_filter_rule(context, field, item) if department_rule: rules.append(department_rule) continue rule: JSONObject = {} if field is not None: rule["queId"] = field.que_id if field.que_type is not None: rule["queType"] = field.que_type for source, target in ( ("search_key", "searchKey"), ("searchKey", "searchKey"), ("search_keys", "searchKeys"), ("searchKeys", "searchKeys"), ("min_value", "minValue"), ("minValue", "minValue"), ("max_value", "maxValue"), ("maxValue", "maxValue"), ("scope", "scope"), ("search_options", "searchOptions"), ("searchOptions", "searchOptions"), ("search_user_ids", "searchUserIds"), ("searchUserIds", "searchUserIds"), ): if source in item: rule[target] = item[source] operator = _stringify_json(item.get("operator", item.get("op"))).strip().lower() if "searchKey" not in rule and "searchKeys" not in rule and "searchOptions" not in rule and "searchUserIds" not in rule and "minValue" not in rule and "maxValue" not in rule: value = item.get("value", item.get("values")) if operator in {"gte", "gt"} and value is not None: rule["minValue"] = _stringify_json(value) elif operator in {"lte", "lt"} and value is not None: rule["maxValue"] = _stringify_json(value) elif operator == "between": lower, upper = _coerce_filter_range(value) if lower is not None: rule["minValue"] = lower if upper is not None: rule["maxValue"] = upper elif value is not None: if field is not None and field.que_type in SINGLE_SELECT_QUE_TYPES | MULTI_SELECT_QUE_TYPES: option_ids = _normalize_option_filter_ids(value) if option_ids: rule["searchOptions"] = option_ids elif isinstance(value, list): rule["searchKeys"] = [_stringify_json(entry) for entry in value] else: rule["searchKey"] = _stringify_json(value) elif field is not None and field.que_type in MEMBER_QUE_TYPES: member_ids = _normalize_member_filter_ids(value) if member_ids: rule["searchUids"] = member_ids elif isinstance(value, list): rule["searchKeys"] = [_stringify_json(entry) for entry in value] else: rule["searchKey"] = _stringify_json(value) if rule: rules.append(rule) return rules def _build_department_filter_rule(self, context, field: FormField, item: JSONObject) -> JSONObject | None: # type: ignore[no-untyped-def] raw_value: JSONValue | None = None for key in ("search_options", "searchOptions", "search_keys", "searchKeys", "search_key", "searchKey", "value", "values"): if key in item: raw_value = cast(JSONValue, item[key]) break if raw_value is None: return None details = self._resolve_department_filter_details(context, raw_value) if not details: return None judge_type = JUDGE_EQUAL if len(details) == 1 else JUDGE_EQUAL_ANY return _department_filter_rule(field.que_id, field.que_type, details, judge_type=judge_type) def _resolve_department_filter_details(self, context, raw_value: JSONValue) -> list[JSONObject]: # type: ignore[no-untyped-def] values = raw_value if isinstance(raw_value, list) else [raw_value] details: list[JSONObject] = [] seen: set[int] = set() for item in _expand_values(values): detail = self._resolve_department_filter_detail(context, item) dept_id = _coerce_count(detail.get("id")) if dept_id is None or dept_id in seen: continue seen.add(dept_id) details.append(detail) return details def _resolve_department_filter_detail(self, context, raw_value: JSONValue) -> JSONObject: # type: ignore[no-untyped-def] if isinstance(raw_value, dict): dept_id = _coerce_count(raw_value.get("id", raw_value.get("deptId"))) if dept_id is not None: return { "id": dept_id, "value": _stringify_json(raw_value.get("value", raw_value.get("name", raw_value.get("deptName", dept_id)))), } keyword = _normalize_optional_text(raw_value.get("value", raw_value.get("name", raw_value.get("deptName")))) if keyword: return self._lookup_department_filter_detail(context, keyword) raise RecordInputError( message="department filters require id/deptId or a department name", error_code="INVALID_DEPARTMENT_FILTER", fix_hint="Pass department filters like {'field': '所在部门', 'value': {'id': 11, 'value': '示例部门'}} or use an exact department name.", details={"received_value": raw_value}, ) dept_id = _coerce_count(raw_value) if dept_id is not None: return {"id": dept_id, "value": str(dept_id)} keyword = _normalize_optional_text(raw_value) if keyword is None: raise RecordInputError( message="department filters require a non-empty department selector", error_code="INVALID_DEPARTMENT_FILTER", fix_hint="Pass a numeric dept id, a department object, or an exact department name.", details={"received_value": raw_value}, ) return self._lookup_department_filter_detail(context, keyword) def _lookup_department_filter_detail(self, context, keyword: str) -> JSONObject: # type: ignore[no-untyped-def] return self._lookup_department_detail(context, keyword, purpose="filter") def _lookup_department_detail(self, context, keyword: str, *, purpose: str) -> JSONObject: # type: ignore[no-untyped-def] payload = self.backend.request( "GET", context, "/contact/deptByPage", params={"keyword": keyword, "pageNum": 1, "pageSize": 20}, ) rows = payload.get("list") if isinstance(payload, dict) else None items = [item for item in rows if isinstance(item, dict)] if isinstance(rows, list) else [] normalized_keyword = keyword.strip() exact = [ item for item in items if _normalize_optional_text(item.get("deptName", item.get("value", item.get("name")))) == normalized_keyword ] matches = exact or items if len(matches) != 1: raise RecordInputError( message=f"department {purpose} '{keyword}' is ambiguous or not found", error_code="DEPARTMENT_NOT_RESOLVED", fix_hint="Pass an exact department name or a numeric dept id.", details={"keyword": keyword, "matches": [{"id": item.get('deptId', item.get('id')), "value": item.get('deptName', item.get('value', item.get('name')))} for item in items[:10]]}, ) matched = matches[0] dept_id = _coerce_count(matched.get("deptId", matched.get("id"))) if dept_id is None: raise RecordInputError( message=f"department {purpose} '{keyword}' resolved to an invalid department payload", error_code="DEPARTMENT_NOT_RESOLVED", fix_hint="Pass a numeric dept id directly.", details={"keyword": keyword, "matched": matched}, ) return {"id": dept_id, "value": _stringify_json(matched.get("deptName", matched.get("value", matched.get("name", dept_id))))} def _department_value_from_selector(self, context, value: JSONValue) -> JSONObject: # type: ignore[no-untyped-def] if isinstance(value, dict): dept_id = _coerce_count(value.get("id", value.get("deptId"))) if dept_id is not None: return { "id": dept_id, "value": _stringify_json(value.get("value", value.get("name", value.get("deptName", dept_id)))), } keyword = _normalize_optional_text(value.get("value", value.get("name", value.get("deptName")))) if keyword: return self._lookup_department_detail(context, keyword, purpose="value") raise RecordInputError( message="department values require id/deptId or a department name", error_code="INVALID_DEPARTMENT_VALUE", fix_hint="Pass department values like {'id': 11, 'value': '示例部门'}, {'deptId': 11, 'deptName': '示例部门'}, or an exact department name.", details={"received_value": value}, ) dept_id = _coerce_count(value) if dept_id is not None: return {"id": dept_id, "value": str(dept_id)} keyword = _normalize_optional_text(value) if keyword is None: raise RecordInputError( message="department values require a numeric id, department object, or department name", error_code="INVALID_DEPARTMENT_VALUE", fix_hint="Pass department ids like 11, exact department names like '示例部门', or objects like {'id': 11, 'value': '示例部门'}.", details={"received_value": value}, ) return self._lookup_department_detail(context, keyword, purpose="value") def _resolve_filter_field_entries(self, filters: list[JSONObject], index: FieldIndex) -> list[JSONObject]: entries: list[JSONObject] = [] for item in filters: if not isinstance(item, dict): continue selector = _extract_filter_selector(item) if selector is None: continue field = self._resolve_field_selector(cast(str | int | None, selector), index, location="filters") entries.append({"requested": str(selector), "field": field}) return entries def _resolve_sorts(self, sorts: list[JSONObject], index: FieldIndex) -> list[JSONObject]: resolved: list[JSONObject] = [] for item in sorts: if not isinstance(item, dict): continue selector = _extract_sort_selector(item) if selector is None: continue field = self._resolve_field_selector(cast(str | int | None, selector), index, location="sorts") ascend = _resolve_sort_ascend(item) resolved.append({"queId": field.que_id, "isAscend": ascend}) return resolved def _resolve_time_range_column(self, time_range: JSONObject, index: FieldIndex) -> FormField | None: if not time_range or not isinstance(time_range, dict): return None column = time_range.get("column") if column is None: return None return self._resolve_field_selector(cast(str | int, column), index, location="time_range.column") def _build_time_range_filter(self, time_range: JSONObject, field: FormField) -> JSONObject | None: if not time_range: return None value_from = time_range.get("from") value_to = time_range.get("to") if value_from is None and value_to is None: return None rule: JSONObject = {"queId": field.que_id} if field.que_type is not None: rule["queType"] = field.que_type if value_from is not None: rule["minValue"] = _stringify_json(value_from) if value_to is not None: rule["maxValue"] = _stringify_json(value_to) return rule def _append_time_range_filter( self, match_rules: list[JSONObject], time_range: JSONObject, field: FormField | None, ) -> list[JSONObject]: if field is None: return match_rules time_rule = self._build_time_range_filter(time_range, field) if time_rule is None: return match_rules for item in match_rules: if not isinstance(item, dict): continue if _coerce_count(item.get("queId")) != field.que_id: continue if item.get("minValue") is not None or item.get("maxValue") is not None: return match_rules return [*match_rules, time_rule] def _collect_write_plan_field_refs(self, *, fields: JSONObject, answers: list[JSONObject], index: FieldIndex) -> list[JSONObject]: refs: list[JSONObject] = [] for field_key in fields.keys(): refs.append(self._resolve_write_plan_field_ref("fields", field_key, index)) for item in answers: if not isinstance(item, dict): continue selector = item.get("queId", item.get("que_id", item.get("queTitle", item.get("que_title")))) if selector is None: continue refs.append(self._resolve_write_plan_field_ref("answers", str(selector), index)) return refs def _resolve_write_plan_field_ref(self, source: str, requested: str, index: FieldIndex) -> JSONObject: try: field = self._resolve_field_selector(requested, index, location=source) return { "source": source, "requested": requested, "resolved": True, "que_id": field.que_id, "que_title": field.que_title, "que_type": field.que_type, "required": field.required, "readonly": field.readonly, "system": field.system, "write_format": _write_format_for_field(field), "reason": None, } except RecordInputError as error: return { "source": source, "requested": requested, "resolved": False, "que_id": None, "que_title": None, "que_type": None, "required": None, "readonly": None, "system": None, "write_format": {"kind": "unknown"}, "reason": error.message, } def _resolve_plan_candidate(self, candidate: JSONObject, index: FieldIndex) -> JSONObject: requested = str(candidate.get("requested", "")).strip() role = str(candidate.get("role", "field")) try: field = self._resolve_field_selector(requested, index, location=role) return { "role": role, "requested": requested, "resolved": True, "que_id": field.que_id, "que_title": field.que_title, "que_type": field.que_type, "reason": None, } except RecordInputError as error: return { "role": role, "requested": requested, "resolved": False, "que_id": None, "que_title": None, "que_type": None, "reason": error.message, } def _validate_plan_arguments(self, tool: str, arguments: JSONObject) -> JSONObject: missing_required: list[str] = [] warnings: list[str] = [] if tool in {"record_query", "record_aggregate"} and not arguments.get("app_key"): missing_required.append("app_key") if tool == "record_get": if not arguments.get("app_key"): missing_required.append("app_key") if not arguments.get("apply_id"): missing_required.append("apply_id") if not arguments.get("select_columns"): missing_required.append("select_columns") query_mode = _resolve_query_mode( str(arguments.get("query_mode", "auto")), apply_id=_coerce_count(arguments.get("apply_id")), amount_column=arguments.get("amount_column"), time_range=cast(JSONObject, arguments.get("time_range") if isinstance(arguments.get("time_range"), dict) else {}), stat_policy=cast(JSONObject, arguments.get("stat_policy") if isinstance(arguments.get("stat_policy"), dict) else {}), ) if tool == "record_query" else None if tool == "record_query" and query_mode in {"list", "record"} and not arguments.get("select_columns"): missing_required.append("select_columns") if tool == "record_query" and query_mode == "summary" and not arguments.get("amount_column") and not arguments.get("time_range"): warnings.append("summary mode without amount_column or time_range only returns row counts") return {"valid": not missing_required, "missing_required": missing_required, "warnings": warnings} def _validate_app_and_record(self, app_key: str, apply_id: int | str) -> int: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) normalized_apply_id = _coerce_count(apply_id) if normalized_apply_id is None or normalized_apply_id <= 0: raise_tool_error(QingflowApiError.config_error("apply_id must be positive")) return normalized_apply_id def _validate_record_write(self, app_key: str, answers: list[JSONObject], apply_id: int | None = None) -> None: if not app_key: raise_tool_error(QingflowApiError.config_error("app_key is required")) if apply_id is not None and apply_id <= 0: raise_tool_error(QingflowApiError.config_error("apply_id must be positive")) if not isinstance(answers, list) or not answers: raise_tool_error(QingflowApiError.config_error("answers must be a non-empty array")) def _verify_record_write_result( self, context, # type: ignore[no-untyped-def] *, app_key: str, apply_id: int | None, normalized_answers: list[JSONObject], index: FieldIndex, ) -> JSONObject: if apply_id is None: return { "verified": False, "error": "missing_apply_id", "missing_fields": [], "empty_fields": [], "count_mismatches": [], } record = self.backend.request( "GET", context, f"/app/{app_key}/apply/{apply_id}", params={"role": 1, "listType": DEFAULT_RECORD_LIST_TYPE}, ) answers = record.get("answers") if isinstance(record, dict) else None answer_list = answers if isinstance(answers, list) else [] actual_by_id = { que_id: item for item in answer_list 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 normalized_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 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, "missing_fields": missing_fields, "empty_fields": empty_fields, "count_mismatches": count_mismatches, } def _verify_subtable_write_result( self, *, field: FormField | None, expected_rows: list[JSONValue], actual_rows: list[JSONValue], missing_fields: list[JSONObject], empty_fields: list[JSONObject], count_mismatches: list[JSONObject], ) -> None: field_payload = _field_ref_payload(field) if field is not None else {"que_id": None} if not actual_rows: empty_fields.append(field_payload) return if len(actual_rows) < len(expected_rows): count_mismatches.append( { **field_payload, "expected_count": len(expected_rows), "actual_count": len(actual_rows), "unit": "rows", } ) subtable_index = self._subtable_field_index_optional(field) actual_rows_by_row_id = _index_subtable_rows_by_row_id(actual_rows) for row_ordinal, raw_expected_row in enumerate(expected_rows): expected_row = [item for item in raw_expected_row if isinstance(item, dict)] if isinstance(raw_expected_row, list) else [] expected_row_id = _subtable_row_id(expected_row) actual_row = None if expected_row_id is not None: actual_row = actual_rows_by_row_id.get(expected_row_id) if actual_row is None and row_ordinal < len(actual_rows): candidate = actual_rows[row_ordinal] actual_row = candidate if isinstance(candidate, list) else None if actual_row is None: missing_fields.append({**field_payload, "row_ordinal": row_ordinal}) continue actual_cells_by_id = { que_id: item for item in actual_row if isinstance(item, dict) and (que_id := _coerce_count(item.get("queId"))) is not None } for expected_cell in expected_row: que_id = _coerce_count(expected_cell.get("queId")) if que_id is None or que_id <= 0: continue subfield = subtable_index.by_id.get(str(que_id)) if subtable_index is not None else None cell_payload = _subtable_cell_ref_payload(field, subfield, que_id=que_id, row_ordinal=row_ordinal, row_id=expected_row_id) actual_cell = actual_cells_by_id.get(que_id) if actual_cell is None: missing_fields.append(cell_payload) continue actual_values = actual_cell.get("values") if isinstance(actual_cell.get("values"), list) else [] if not actual_values: empty_fields.append(cell_payload) continue expected_values = expected_cell.get("values") if isinstance(expected_cell.get("values"), list) else [] if expected_values and len(actual_values) < len(expected_values): count_mismatches.append( { **cell_payload, "expected_count": len(expected_values), "actual_count": len(actual_values), "unit": "values", } ) def _score_field_matches(self, requested: str, index: FieldIndex, *, fuzzy: bool, top_k: int) -> list[JSONObject]: requested_text = requested.strip() requested_key = _normalize_field_lookup_key(requested_text) if not requested_key: return [] matches: list[JSONObject] = [] for field in index.by_id.values(): best_score = 0.0 match_type = "none" matched_alias: str | None = None if requested_text.isdigit() and field.que_id == int(requested_text): best_score = 1.0 match_type = "id_exact" else: best_score, match_type = _score_candidate_text(requested_key, field.que_title, fuzzy=fuzzy, label="title") for alias in field.aliases: alias_score, alias_match_type = _score_candidate_text(requested_key, alias, fuzzy=fuzzy, label="alias") if alias_score > best_score: best_score = alias_score match_type = alias_match_type matched_alias = alias if best_score <= 0: continue payload = { "que_id": field.que_id, "que_title": field.que_title, "que_type": field.que_type, "score": round(best_score, 4), "match_type": match_type, } if matched_alias is not None: payload["matched_alias"] = matched_alias matches.append(payload) matches.sort(key=lambda item: float(item["score"]), reverse=True) return matches[: max(top_k, 1)] def _raise_need_more_data(self, completeness: JSONObject, evidence: JSONObject, message: str) -> None: raise RuntimeError( json.dumps( { "ok": False, "code": "NEED_MORE_DATA", "status": "need_more_data", "message": message, "details": {"completeness": completeness, "evidence": evidence}, }, ensure_ascii=False, ) ) def _normalize_form_schema(payload: JSONValue) -> JSONObject: if isinstance(payload, dict) and isinstance(payload.get("formQues"), list): return payload if isinstance(payload, list): for item in payload: if isinstance(item, dict) and isinstance(item.get("formQues"), list): return item return {} def _normalize_view_list(payload: JSONValue) -> list[JSONObject]: if not isinstance(payload, list): return [] flattened: list[JSONObject] = [] for group in payload: if not isinstance(group, dict): continue view_list = group.get("viewList") if not isinstance(view_list, list): continue for item in view_list: if isinstance(item, dict) and item.get("viewKey"): flattened.append(item) return flattened def _compile_view_conditions(config: JSONObject) -> list[list[ViewFilterCondition]]: raw_limit = config.get("viewgraphLimit") if not isinstance(raw_limit, list): return [] compiled: list[list[ViewFilterCondition]] = [] for raw_group in raw_limit: group_items = raw_group if isinstance(raw_group, list) else [raw_group] group: list[ViewFilterCondition] = [] for raw_condition in group_items: if not isinstance(raw_condition, dict): continue group.append( ViewFilterCondition( que_id=_coerce_count(raw_condition.get("queId")), que_title=_stringify_json(raw_condition.get("queTitle")).strip(), que_type=_coerce_count(raw_condition.get("queType")), judge_type=_coerce_count(raw_condition.get("judgeType")), judge_values=[_stringify_json(item).strip() for item in cast(list[JSONValue], raw_condition.get("judgeValues") or []) if _stringify_json(item).strip()], judge_value_details=[item for item in cast(list[JSONValue], raw_condition.get("judgeValueDetails") or []) if isinstance(item, dict)], raw=raw_condition, ) ) if group: compiled.append(group) return compiled def _build_field_index(schema: JSONObject) -> FieldIndex: by_id: dict[str, FormField] = {} by_title: dict[str, list[FormField]] = {} by_alias: dict[str, list[FormField]] = {} all_questions = [ *[(question, True) for question in _flatten_questions(schema.get("baseQues"))], *[(question, False) for question in _flatten_questions(schema.get("formQues"))], ] for question, is_base_question in all_questions: que_id = _coerce_count(question.get("queId")) title = _stringify_json(question.get("queTitle")).strip() if que_id is None or que_id < 0 or not title: continue field = FormField( que_id=que_id, que_title=title, que_type=_coerce_count(question.get("queType")), required=bool(question.get("required") or question.get("beingRequired")), readonly=bool(question.get("readonly") or question.get("beingReadonly") or is_base_question), system=bool(question.get("system") or question.get("beingSystem") or is_base_question), options=_extract_question_options(question), aliases=[], raw=question, ) if str(que_id) in by_id: continue field.aliases = sorted(_field_alias_candidates(field)) by_id[str(que_id)] = field by_title.setdefault(_normalize_field_lookup_key(title), []).append(field) for alias in field.aliases: by_alias.setdefault(_normalize_field_lookup_key(alias), []).append(field) return FieldIndex(by_id=by_id, by_title=by_title, by_alias=by_alias) def _flatten_questions(payload: JSONValue) -> list[JSONObject]: flattened: list[JSONObject] = [] if isinstance(payload, dict): if "queId" in payload or "queTitle" in payload: flattened.append(payload) for value in payload.values(): flattened.extend(_flatten_questions(value)) elif isinstance(payload, list): for item in payload: flattened.extend(_flatten_questions(item)) return flattened def _extract_question_options(question: JSONObject) -> list[str]: options = question.get("options") if not isinstance(options, list): return [] values: list[str] = [] for item in options: if isinstance(item, dict): value = item.get("optValue", item.get("value")) if value is not None: values.append(_stringify_json(value)) elif item is not None: values.append(_stringify_json(item)) return values def _normalize_plan_arguments(tool: str, arguments: JSONObject) -> JSONObject: normalized = cast(JSONObject, _parse_json_like(arguments)) alias_map = { "appKey": "app_key", "applyId": "apply_id", "queryMode": "query_mode", "pageNum": "page_num", "pageSize": "page_size", "requestedPages": "requested_pages", "scanMaxPages": "scan_max_pages", "queryKey": "query_key", "maxRows": "max_rows", "maxColumns": "max_columns", "selectColumns": "select_columns", "amountColumn": "amount_column", "timeRange": "time_range", "strictFull": "strict_full", "outputProfile": "output_profile", "listType": "list_type", "viewKey": "view_key", "viewName": "view_name", "groupBy": "group_by", "timeBucket": "time_bucket", "maxGroups": "max_groups", "forceRefreshForm": "force_refresh_form", } result = dict(normalized) for alias, canonical in alias_map.items(): if alias in result and canonical not in result: result[canonical] = result[alias] if tool == "record_get" and "query_mode" not in result: result["query_mode"] = "record" return result def _collect_plan_field_candidates(tool: str, arguments: JSONObject) -> list[JSONObject]: candidates: list[JSONObject] = [] if tool in {"record_query", "record_get"}: for item in _as_selector_list(arguments.get("select_columns")): candidates.append({"role": "select_columns", "requested": str(item)}) amount = arguments.get("amount_column") if amount is not None: candidates.append({"role": "amount_column", "requested": str(amount)}) time_range = arguments.get("time_range") if isinstance(time_range, dict) and time_range.get("column") is not None: candidates.append({"role": "time_range.column", "requested": str(time_range["column"])}) if tool == "record_aggregate": for item in _as_selector_list(arguments.get("group_by")): candidates.append({"role": "group_by", "requested": str(item)}) amount = arguments.get("amount_column") if amount is not None: candidates.append({"role": "amount_column", "requested": str(amount)}) time_range = arguments.get("time_range") if isinstance(time_range, dict) and time_range.get("column") is not None: candidates.append({"role": "time_range.column", "requested": str(time_range["column"])}) for item in _as_object_list(arguments.get("filters")): selector = _extract_filter_selector(item) if selector is not None: candidates.append({"role": "filter", "requested": str(selector)}) for item in _as_object_list(arguments.get("sorts", arguments.get("sort"))): selector = _extract_sort_selector(item) if selector is not None: candidates.append({"role": "sort", "requested": str(selector)}) return candidates def _build_plan_estimate(tool: str, arguments: JSONObject) -> JSONObject: page_size = _coerce_count(arguments.get("page_size")) or DEFAULT_QUERY_PAGE_SIZE requested_pages = _coerce_count(arguments.get("requested_pages")) or 1 scan_max_pages = _coerce_count(arguments.get("scan_max_pages")) or requested_pages estimated_scan_pages = min(requested_pages, scan_max_pages) may_hit_limits = estimated_scan_pages > DEFAULT_SCAN_MAX_PAGES reasons = [] if may_hit_limits: reasons.append("requested scan pages exceed the default analysis budget") if tool == "record_query": routed_mode = _resolve_query_mode( str(arguments.get("query_mode", "auto")), apply_id=_coerce_count(arguments.get("apply_id")), amount_column=arguments.get("amount_column"), time_range=cast(JSONObject, arguments.get("time_range") if isinstance(arguments.get("time_range"), dict) else {}), stat_policy=cast(JSONObject, arguments.get("stat_policy") if isinstance(arguments.get("stat_policy"), dict) else {}), ) if routed_mode == "list": reasons.append("list mode is not a safe final-analysis endpoint") return { "page_size": page_size, "requested_pages": requested_pages, "scan_max_pages": scan_max_pages, "estimated_scan_pages": estimated_scan_pages, "estimated_items_upper_bound": page_size * estimated_scan_pages, "may_hit_limits": may_hit_limits, "reasons": reasons, "probe": None, } def _assess_plan_readiness( tool: str, arguments: JSONObject, validation: JSONObject, field_mapping: list[JSONObject], estimate: JSONObject, ) -> JSONObject: blockers: list[str] = [] actions: list[str] = [] if not bool(validation.get("valid")): blockers.append("arguments are not valid") actions.append("Fix missing_required before execution.") unresolved = [item for item in field_mapping if not bool(item.get("resolved"))] if unresolved: blockers.append("one or more fields are unresolved") actions.append("Use record_field_resolve to resolve field ids before execution.") if tool == "record_query": routed_mode = _resolve_query_mode( str(arguments.get("query_mode", "auto")), apply_id=_coerce_count(arguments.get("apply_id")), amount_column=arguments.get("amount_column"), time_range=cast(JSONObject, arguments.get("time_range") if isinstance(arguments.get("time_range"), dict) else {}), stat_policy=cast(JSONObject, arguments.get("stat_policy") if isinstance(arguments.get("stat_policy"), dict) else {}), ) if routed_mode == "list": blockers.append("list mode is not a safe final-analysis endpoint") actions.append("Use record_query(summary) or record_aggregate for final statistics.") if routed_mode == "record": blockers.append("record mode is a detail endpoint, not a final-analysis endpoint") if tool in {"record_query", "record_aggregate"} and not bool(arguments.get("strict_full")): blockers.append("strict_full should be true for final conclusions") actions.append("Set strict_full=true so incomplete scans block final conclusions.") actions.append("After execution, verify completeness before using the result as a final conclusion.") return { "ready_for_final_conclusion": not blockers, "final_conclusion_blockers": blockers, "recommended_next_actions": actions, } def _resolve_query_mode( query_mode: str, *, apply_id: int | None, amount_column: JSONValue, time_range: JSONObject, stat_policy: JSONObject, ) -> str: if query_mode in {"list", "record", "summary"}: return query_mode if apply_id is not None and apply_id > 0: return "record" if amount_column is not None or time_range or stat_policy: return "summary" return "list" def _bounded_column_limit(max_columns: int | None, *, default_limit: int, hard_limit: int) -> int: if max_columns is None: return default_limit return max(1, min(max_columns, hard_limit)) def _chunk_fields(fields: list[FormField], chunk_size: int) -> list[list[FormField]]: if chunk_size <= 0: raise ValueError("chunk_size must be positive") if not fields: return [] return [fields[index : index + chunk_size] for index in range(0, len(fields), chunk_size)] def _view_selection_supported_by_search_ids(view_selection: ViewSelection, search_que_ids: list[int] | None) -> bool: if not view_selection.conditions: return True if not search_que_ids: return False allowed_ids = set(search_que_ids) for group in view_selection.conditions: for condition in group: if condition.que_id is None or condition.que_id not in allowed_ids: return False return True def _build_flat_row(answer_list: list[JSONValue], fields: list[FormField], *, apply_id: int | None) -> JSONObject: row: JSONObject = {"apply_id": apply_id} for field in fields: row[field.que_title] = _extract_field_value(answer_list, field) return row def _extract_field_value(answer_list: list[JSONValue], field: FormField | None) -> JSONValue: if field is None: return None for answer in answer_list: if not isinstance(answer, dict): continue answer_que_id = _coerce_count(answer.get("queId")) answer_title = _stringify_json(answer.get("queTitle")).strip() if answer_que_id == field.que_id or (answer_title and answer_title == field.que_title): values = answer.get("values") if not isinstance(values, list) or not values: return None extracted = [_extract_value_item(item) for item in values] return extracted[0] if len(extracted) == 1 else extracted return None def _extract_value_item(value: JSONValue) -> JSONValue: if isinstance(value, dict): if "value" in value: return value["value"] if "name" in value: return value["name"] if "email" in value: return value["email"] if "id" in value: return value["id"] return value def _field_mapping_entry(role: str, field: FormField | None, *, requested: str) -> JSONObject: return { "role": role, "requested": requested, "resolved": field is not None, "que_id": field.que_id if field is not None else None, "que_title": field.que_title if field is not None else None, "que_type": field.que_type if field is not None else None, } def _query_id() -> str: return datetime.now(UTC).isoformat() def _view_selection_payload(view_selection: ViewSelection | None) -> JSONObject | None: if view_selection is None: return None return { "view_key": view_selection.view_key, "view_name": view_selection.view_name, "local_filtering": bool(view_selection.conditions), "condition_group_count": len(view_selection.conditions), } def _build_completeness( *, result_amount: int, returned_items: int, fetched_pages: int, requested_pages: int, has_more: bool, next_page_token: str | None, is_complete: bool, omitted_items: int, extra: JSONObject, ) -> JSONObject: payload: JSONObject = { "result_amount": result_amount, "returned_items": returned_items, "fetched_pages": fetched_pages, "requested_pages": requested_pages, "actual_scanned_pages": fetched_pages, "has_more": has_more, "next_page_token": next_page_token, "is_complete": is_complete, "partial": not is_complete, "omitted_items": omitted_items, "omitted_chars": 0, } payload.update(extra) return payload def _page_has_more(page: JSONObject, current_page: int, page_size: int, returned_rows: int) -> bool: page_amount = _coerce_count(page.get("pageAmount")) if page_amount is not None: return current_page < page_amount return returned_rows >= page_size def _effective_total(page: JSONObject, page_size: int) -> int: rows = page.get("list") returned_rows = len(rows) if isinstance(rows, list) else 0 reported = _coerce_count(page.get("total")) if reported is None: reported = _coerce_count(page.get("count")) if reported is not None: return max(reported, returned_rows) page_amount = _coerce_count(page.get("pageAmount")) if page_amount is not None: return page_amount * page_size return returned_rows def _pick_title_field(index: FieldIndex) -> FormField | None: priority_keywords = ("名称", "标题", "主题", "name", "title", "subject") for field in index.by_id.values(): lowered = field.que_title.lower() if any(keyword in lowered for keyword in priority_keywords): return field return next(iter(index.by_id.values()), None) def _normalize_metrics(metrics: list[str], *, include_sum: bool) -> list[str]: normalized = [item for item in metrics if item in {"count", "sum", "avg", "min", "max"}] if not normalized: return ["count", "sum"] if include_sum else ["count"] if not include_sum: return [item for item in normalized if item == "count"] or ["count"] return normalized def _coerce_count(value: JSONValue) -> int | None: if isinstance(value, bool) or value is None: return None if isinstance(value, int): return value if isinstance(value, float): return int(value) if isinstance(value, str): text = value.strip() if not text: return None try: return int(text) except ValueError: return None return None def _coerce_amount(value: JSONValue) -> float | None: if isinstance(value, bool) or value is None: return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): text = value.strip() if not text: return None negative = False if text.startswith("(") and text.endswith(")"): negative = True text = text[1:-1].strip() for symbol in ("¥", "￥", "$"): text = text.replace(symbol, "") text = text.replace(",", "").replace(" ", "") if negative and text and not text.startswith("-"): text = f"-{text}" try: return float(text) except ValueError: return None return None def _to_time_bucket(value: JSONValue, bucket: str) -> str: text = _stringify_json(value).strip() if not text: return "unknown" parsed = _parse_datetime_like(text) if parsed is None: return text[:10] if bucket == "day" and len(text) >= 10 else text if bucket == "month": return parsed.strftime("%Y-%m") if bucket == "week": iso_year, iso_week, _ = parsed.isocalendar() return f"{iso_year}-W{iso_week:02d}" return parsed.strftime("%Y-%m-%d") def _parse_datetime_like(text: str) -> datetime | None: cleaned = text.strip().replace("/", "-") for candidate in (cleaned, cleaned.replace(" ", "T")): try: return datetime.fromisoformat(candidate) except ValueError: continue return None def _echo_filters(filters: list[JSONObject]) -> list[JSONObject]: return [dict(item) for item in filters] def _extract_filter_selector(item: JSONObject) -> JSONValue: for key in ("que_id", "queId", "field", "field_id", "fieldId", "column", "queTitle", "que_title"): if key in item: return item[key] return None def _extract_sort_selector(item: JSONObject) -> JSONValue: for key in ("que_id", "queId", "field", "field_id", "fieldId", "column", "queTitle", "que_title"): if key in item: return item[key] return None def _resolve_sort_ascend(item: JSONObject) -> bool: if "isAscend" in item: return bool(item["isAscend"]) if "ascend" in item: return bool(item["ascend"]) direction = _stringify_json(item.get("direction", item.get("order"))).strip().lower() return direction not in {"desc", "descending", "-1"} def _coerce_filter_range(value: JSONValue) -> tuple[str | None, str | None]: if isinstance(value, dict): lower = value.get("from", value.get("min", value.get("start"))) upper = value.get("to", value.get("max", value.get("end"))) return _stringify_json(lower) if lower is not None else None, _stringify_json(upper) if upper is not None else None if isinstance(value, list) and value: lower = value[0] if len(value) >= 1 else None upper = value[1] if len(value) >= 2 else None return _stringify_json(lower) if lower is not None else None, _stringify_json(upper) if upper is not None else None return None, None def _normalize_filter_values(value: JSONValue) -> list[str]: if isinstance(value, list): return [_stringify_json(item) for item in value] return [_stringify_json(value)] def _normalize_optional_text(value: JSONValue) -> str | None: if isinstance(value, str): text = value.strip() return text or None if isinstance(value, (int, float)): text = str(value).strip() return text or None return None def _match_view_condition( answer_list: list[JSONValue], condition: ViewFilterCondition, *, dept_member_cache: dict[int, set[int]], dept_member_resolver, ) -> bool: # type: ignore[no-untyped-def] answer = _find_answer_for_condition(answer_list, condition) if answer is None: return False values = answer.get("values") answer_values = values if isinstance(values, list) else [] if not answer_values: return False if condition.que_type in MEMBER_QUE_TYPES: dept_ids = _extract_condition_dept_ids(condition) if dept_ids: allowed_ids: set[int] = set() for dept_id in dept_ids: if dept_id not in dept_member_cache: dept_member_cache[dept_id] = set(dept_member_resolver(dept_id)) allowed_ids.update(dept_member_cache[dept_id]) return any( member_id in allowed_ids for member_id in (_coerce_count(item.get("id")) if isinstance(item, dict) else None for item in answer_values) if member_id is not None ) condition_ids = _extract_condition_ids(condition) if condition_ids: answer_ids = [ member_id for member_id in (_coerce_count(item.get("id")) if isinstance(item, dict) else None for item in answer_values) if member_id is not None ] if any(answer_id in condition_ids for answer_id in answer_ids): return True condition_texts = _extract_condition_texts(condition) if not condition_texts: return False answer_texts = [_normalize_answer_value_text(item) for item in answer_values] return any(answer_text in condition_texts for answer_text in answer_texts if answer_text) def _find_answer_for_condition(answer_list: list[JSONValue], condition: ViewFilterCondition) -> JSONObject | None: target_title = condition.que_title.strip() for answer in answer_list: if not isinstance(answer, dict): continue answer_que_id = _coerce_count(answer.get("queId")) if condition.que_id is not None and answer_que_id == condition.que_id: return answer answer_title = _stringify_json(answer.get("queTitle")).strip() if target_title and answer_title == target_title: return answer return None def _extract_condition_dept_ids(condition: ViewFilterCondition) -> list[int]: dept_ids: list[int] = [] for raw in condition.judge_values: if raw.startswith(DEPARTMENT_MEMBER_JUDGE_PREFIX): dept_id = _coerce_count(raw[len(DEPARTMENT_MEMBER_JUDGE_PREFIX):]) if dept_id is not None: dept_ids.append(dept_id) return dept_ids def _extract_condition_ids(condition: ViewFilterCondition) -> set[int]: ids: set[int] = set() for raw in condition.judge_values: if raw.startswith(DEPARTMENT_MEMBER_JUDGE_PREFIX): continue value = _coerce_count(raw) if value is not None: ids.add(value) for item in condition.judge_value_details: value = _coerce_count(item.get("id", item.get("uid"))) if value is not None: ids.add(value) return ids def _extract_condition_texts(condition: ViewFilterCondition) -> set[str]: values: set[str] = set() for raw in condition.judge_values: if raw.startswith(DEPARTMENT_MEMBER_JUDGE_PREFIX): continue normalized = raw.strip() if normalized: values.add(normalized) for item in condition.judge_value_details: for key in ("value", "dataValue", "name"): normalized = _normalize_optional_text(item.get(key)) if normalized: values.add(normalized) return values def _normalize_answer_value_text(value: JSONValue) -> str | None: if isinstance(value, dict): for key in ("value", "dataValue", "name", "email"): normalized = _normalize_optional_text(value.get(key)) if normalized: return normalized value_id = _coerce_count(value.get("id", value.get("uid"))) return str(value_id) if value_id is not None else None return _normalize_optional_text(value) def _normalize_list_query_rules(item: JSONObject) -> list[JSONObject]: if _looks_like_backend_match_rule(item): rule = _match_rule_to_query_rule(item) return [rule] if rule else [] que_id = item.get("queId", item.get("que_id")) que_type = _coerce_count(item.get("queType", item.get("que_type"))) if que_type in DEPARTMENT_QUE_TYPES and _department_filter_details_from_item(item): return [] rule: JSONObject = {} if que_id is not None: rule["queId"] = que_id if que_type is not None: rule["queType"] = que_type for source, target in ( ("searchKey", "searchKey"), ("search_key", "searchKey"), ("searchKeys", "searchKeys"), ("search_keys", "searchKeys"), ("minValue", "minValue"), ("min_value", "minValue"), ("maxValue", "maxValue"), ("max_value", "maxValue"), ("scope", "scope"), ("searchOptions", "searchOptions"), ("search_options", "searchOptions"), ("searchUserIds", "searchUserIds"), ("search_user_ids", "searchUserIds"), ("searchUids", "searchUids"), ("search_uids", "searchUids"), ): if source in item: rule[target] = item[source] return [rule] if rule else [] def _normalize_list_match_rules(item: JSONObject) -> list[JSONObject]: if _looks_like_backend_match_rule(item): normalized = _normalize_backend_match_rule(item) return [normalized] if normalized else [] que_id = _coerce_count(item.get("queId", item.get("que_id"))) que_type = _coerce_count(item.get("queType", item.get("que_type"))) if que_id is None: return [] if que_type in DEPARTMENT_QUE_TYPES: details = _department_filter_details_from_item(item) if details: judge_type = JUDGE_EQUAL if len(details) == 1 else JUDGE_EQUAL_ANY return [_department_filter_rule(que_id, que_type, details, judge_type=judge_type)] min_value = item.get("minValue", item.get("min_value")) max_value = item.get("maxValue", item.get("max_value")) if que_type in DATE_QUE_TYPES and (min_value is not None or max_value is not None): return [] match_rules: list[JSONObject] = [] scope = _coerce_count(item.get("scope")) if scope is not None and scope != SCOPE_ALL: match_rule: JSONObject = { "queId": que_id, "judgeType": JUDGE_UNEQUAL if scope == SCOPE_NOT_EMPTY else JUDGE_EQUAL, "judgeValues": [], "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) search_key = _normalize_optional_text(item.get("searchKey", item.get("search_key"))) if search_key is not None: match_rule = { "queId": que_id, "judgeType": JUDGE_FUZZY_MATCH, "judgeValues": [search_key], "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) search_keys = item.get("searchKeys", item.get("search_keys")) normalized_search_keys = [_stringify_json(entry) for entry in search_keys] if isinstance(search_keys, list) else [] if normalized_search_keys: judge_type = JUDGE_INCLUDE_ANY if que_type in MULTI_SELECT_QUE_TYPES | MEMBER_QUE_TYPES else JUDGE_EQUAL_ANY match_rule = { "queId": que_id, "judgeType": judge_type, "judgeValues": normalized_search_keys, "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) if min_value is not None: match_rule = { "queId": que_id, "judgeType": JUDGE_GREATER_OR_EQUAL, "judgeValues": [_stringify_json(min_value)], "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) if max_value is not None: match_rule = { "queId": que_id, "judgeType": JUDGE_LESS_OR_EQUAL, "judgeValues": [_stringify_json(max_value)], "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) search_options = item.get("searchOptions", item.get("search_options")) normalized_search_options = [_stringify_json(entry) for entry in search_options] if isinstance(search_options, list) else [] if normalized_search_options: judge_type = JUDGE_INCLUDE_ANY if que_type in MULTI_SELECT_QUE_TYPES else JUDGE_EQUAL_ANY match_rule = { "queId": que_id, "judgeType": judge_type, "judgeValues": normalized_search_options, "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) search_uids = item.get("searchUids", item.get("search_uids", item.get("searchUserIds", item.get("search_user_ids")))) normalized_search_uids = [str(member_id) for member_id in _normalize_member_filter_ids(search_uids)] if search_uids is not None else [] if normalized_search_uids: match_rule = { "queId": que_id, "judgeType": JUDGE_INCLUDE_ANY, "judgeValues": normalized_search_uids, "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: match_rule["queType"] = que_type match_rules.append(match_rule) return match_rules def _looks_like_backend_match_rule(item: JSONObject) -> bool: return any(key in item for key in ("judgeType", "judge_type", "judgeValues", "judge_values", "judgeValueDetails", "judge_value_details")) def _normalize_backend_match_rule(item: JSONObject) -> JSONObject: que_id = _coerce_count(item.get("queId", item.get("que_id"))) if que_id is None: return {} rule: JSONObject = {"queId": que_id} que_title = _normalize_optional_text(item.get("queTitle", item.get("que_title"))) if que_title is not None: rule["queTitle"] = que_title que_type = _coerce_count(item.get("queType", item.get("que_type"))) if que_type is not None: rule["queType"] = que_type judge_type = _coerce_count(item.get("judgeType", item.get("judge_type"))) if judge_type is not None: rule["judgeType"] = judge_type judge_values = item.get("judgeValues", item.get("judge_values")) if isinstance(judge_values, list): rule["judgeValues"] = [_stringify_json(value) for value in judge_values] judge_value_details = item.get("judgeValueDetails", item.get("judge_value_details")) if isinstance(judge_value_details, list): rule["judgeValueDetails"] = [value for value in judge_value_details if isinstance(value, dict)] control_value = _coerce_count(item.get("controlValue", item.get("control_value"))) if control_value is not None: rule["controlValue"] = control_value control_time_start_value = item.get("controlTimeStartValue", item.get("control_time_start_value")) if control_time_start_value is not None: rule["controlTimeStartValue"] = _stringify_json(control_time_start_value) control_time_end_value = item.get("controlTimeEndValue", item.get("control_time_end_value")) if control_time_end_value is not None: rule["controlTimeEndValue"] = _stringify_json(control_time_end_value) match_type = _coerce_count(item.get("matchType", item.get("match_type"))) rule["matchType"] = match_type if match_type is not None else MATCH_TYPE_ACCURACY return rule def _department_filter_rule( que_id: int, que_type: int | None, details: list[JSONObject], *, judge_type: int, ) -> JSONObject: rule: JSONObject = { "queId": que_id, "judgeType": judge_type, "judgeValues": [str(detail["id"]) for detail in details if detail.get("id") is not None], "judgeValueDetails": [ {"id": detail["id"], "value": _stringify_json(detail.get("value", detail["id"]))} for detail in details if detail.get("id") is not None ], "matchType": MATCH_TYPE_ACCURACY, } if que_type is not None: rule["queType"] = que_type return rule def _department_filter_details_from_item(item: JSONObject) -> list[JSONObject]: values: list[JSONValue] = [] for key in ("searchOptions", "search_options", "searchKeys", "search_keys", "searchKey", "search_key", "value", "values"): if key in item: raw = item[key] values = raw if isinstance(raw, list) else [raw] break details: list[JSONObject] = [] seen: set[int] = set() for value in _expand_values(values): detail = _department_filter_detail_from_value(value) if detail is None: continue dept_id = _coerce_count(detail.get("id")) if dept_id is None or dept_id in seen: continue seen.add(dept_id) details.append(detail) return details def _department_filter_detail_from_value(value: JSONValue) -> JSONObject | None: if isinstance(value, dict): dept_id = _coerce_count(value.get("id", value.get("deptId"))) if dept_id is None: return None return {"id": dept_id, "value": _stringify_json(value.get("value", value.get("name", value.get("deptName", dept_id))))} dept_id = _coerce_count(value) if dept_id is None: return None return {"id": dept_id, "value": str(dept_id)} def _match_rule_to_query_rule(item: JSONObject) -> JSONObject: que_id = _coerce_count(item.get("queId", item.get("que_id"))) if que_id is None: return {} que_type = _coerce_count(item.get("queType", item.get("que_type"))) judge_type = _coerce_count(item.get("judgeType", item.get("judge_type"))) judge_values = item.get("judgeValues", item.get("judge_values")) values = [_stringify_json(value) for value in judge_values] if isinstance(judge_values, list) else [] rule: JSONObject = {"queId": que_id} if que_type is not None: rule["queType"] = que_type if judge_type == JUDGE_FUZZY_MATCH and values: rule["searchKey"] = values[0] elif judge_type == JUDGE_GREATER_OR_EQUAL and values: rule["minValue"] = values[0] elif judge_type == JUDGE_LESS_OR_EQUAL and values: rule["maxValue"] = values[0] elif judge_type == JUDGE_EQUAL_ANY and values: if que_type in SINGLE_SELECT_QUE_TYPES | MULTI_SELECT_QUE_TYPES | DEPARTMENT_QUE_TYPES: rule["searchOptions"] = values else: rule["searchKeys"] = values elif judge_type == JUDGE_INCLUDE_ANY and values: if que_type in MEMBER_QUE_TYPES: numeric_ids = [member_id for member_id in (_coerce_count(value) for value in values) if member_id is not None] if numeric_ids: rule["searchUids"] = numeric_ids else: rule["searchUserIds"] = values elif que_type in MULTI_SELECT_QUE_TYPES: rule["searchOptions"] = values else: rule["searchKeys"] = values elif judge_type == JUDGE_UNEQUAL and not values: rule["scope"] = SCOPE_NOT_EMPTY elif judge_type == JUDGE_EQUAL and not values: rule["scope"] = SCOPE_EMPTY query_keys = {"searchKey", "searchKeys", "minValue", "maxValue", "searchOptions", "searchUids", "searchUserIds", "scope"} return rule if any(key in rule for key in query_keys) else {} def _normalize_list_sort_rule(item: JSONObject) -> JSONObject: rule: JSONObject = {} que_id = item.get("queId", item.get("que_id")) if que_id is not None: rule["queId"] = que_id if "isAscend" in item: rule["isAscend"] = bool(item["isAscend"]) elif "ascend" in item or "direction" in item or "order" in item: rule["isAscend"] = _resolve_sort_ascend(item) return rule def _normalize_member_filter_ids(value: JSONValue) -> list[int]: values = value if isinstance(value, list) else [value] member_ids: list[int] = [] for item in values: if isinstance(item, dict): candidate = item.get("id", item.get("uid", item.get("userId"))) else: candidate = item member_id = _coerce_count(candidate) if member_id is not None: member_ids.append(member_id) return member_ids def _normalize_option_filter_ids(value: JSONValue) -> list[int]: values = value if isinstance(value, list) else [value] option_ids: list[int] = [] for item in values: candidate: JSONValue if isinstance(item, dict): candidate = item.get("optionId", item.get("optId", item.get("id"))) else: candidate = item option_id = _coerce_count(candidate) if option_id is not None: option_ids.append(option_id) return option_ids def _collect_question_relations(schema: JSONObject) -> list[JSONObject]: relations = schema.get("questionRelations") if not isinstance(relations, list): return [] return [item for item in relations if isinstance(item, dict)] def _collect_option_links(resolved_fields: list[JSONObject]) -> list[JSONObject]: links: list[JSONObject] = [] for entry in resolved_fields: write_format = entry.get("write_format") if not isinstance(write_format, dict): continue option_links = write_format.get("option_links") if not isinstance(option_links, list): continue for item in option_links: if isinstance(item, dict): links.append(item) return links def _answers_need_resolution(answers: list[JSONObject]) -> bool: for item in answers: if not isinstance(item, dict): return True if "queId" not in item or "queType" not in item: return True if "values" not in item and "value" in item: return True if "queTitle" in item or "que_title" in item or "que_id" in item: return True if _subtable_answer_needs_resolution(item): return True return False def _subtable_answer_needs_resolution(item: JSONObject) -> bool: if any(key in item for key in ("rows", "fields")): return True que_type = _coerce_count(item.get("queType")) table_values = item.get("tableValues") if que_type not in SUBTABLE_QUE_TYPES: return False if table_values is None: return True if not isinstance(table_values, list): return True for row in table_values: if isinstance(row, dict): return True if not isinstance(row, list): return True for cell in row: if not isinstance(cell, dict): return True if "queId" not in cell or "queType" not in cell: return True if "values" not in cell and "value" in cell: return True if any(key in cell for key in ("queTitle", "que_title", "que_id", "fields", "answers", "rows")): return True return False def _expand_values(values: list[JSONValue]) -> list[JSONValue]: if len(values) == 1 and isinstance(values[0], list): nested = values[0] return [item for item in nested] return values def _option_value(raw_value: JSONValue, field: FormField) -> JSONObject: if isinstance(raw_value, dict): value = raw_value.get("value", raw_value.get("optValue")) payload: JSONObject = {"value": _stringify_json(value)} if "optionId" in raw_value: payload["optionId"] = raw_value["optionId"] return payload text = _stringify_json(raw_value) if field.options and text not in field.options: raise RecordInputError( message=f"field '{field.que_title}' uses unknown option '{text}'", error_code="OPTION_NOT_FOUND", fix_hint="Use record_field_resolve or inspect the form to confirm allowed option values.", details={"field": _field_ref_payload(field), "expected_format": _write_format_for_field(field), "received_value": raw_value}, ) return {"value": text} def _boolean_display(value: JSONValue) -> str: if isinstance(value, bool): return "是" if value else "否" text = _stringify_json(value).strip().lower() if text in {"true", "1", "yes", "y", "是"}: return "是" if text in {"false", "0", "no", "n", "否"}: return "否" return _stringify_json(value) def _member_value(profile: str, value: JSONValue) -> JSONObject: if isinstance(value, dict): member_id = _coerce_count(value.get("id", value.get("uid", value.get("userId")))) user_id = value.get("userId") if member_id is None and not (isinstance(user_id, str) and user_id.strip()): raise RecordInputError( message="member values require id, uid, or userId", error_code="INVALID_MEMBER_VALUE", fix_hint="Pass member values like {'id': 2, 'value': '张三'} or {'userId': 'u_123', 'userName': '张三'}.", details={"received_value": value, "location": profile}, ) payload: JSONObject = { "value": _stringify_json(value.get("value", value.get("name", value.get("userName", member_id if member_id is not None else user_id)))) } if member_id is not None: payload["id"] = member_id elif isinstance(user_id, str) and user_id.strip(): payload["userId"] = user_id.strip() if value.get("email") is not None: payload["email"] = value["email"] if value.get("otherInfo") is not None: payload["otherInfo"] = value["otherInfo"] return payload member_id = _coerce_count(value) if member_id is None and not (isinstance(value, str) and value.strip()): raise RecordInputError( message="member values require numeric ids, userIds, or member objects", error_code="INVALID_MEMBER_VALUE", fix_hint="Pass member ids like 2, userIds like 'u_123', or objects like {'id': 2, 'value': '张三'}.", details={"received_value": value, "location": profile}, ) if member_id is not None: return {"id": member_id, "value": str(member_id)} return {"userId": str(value).strip(), "value": str(value).strip()} def _department_value(value: JSONValue) -> JSONObject: if isinstance(value, dict): dept_id = _coerce_count(value.get("id", value.get("deptId"))) if dept_id is None: raise RecordInputError( message="department values require id or deptId", error_code="INVALID_DEPARTMENT_VALUE", fix_hint="Pass department values like {'id': 11, 'value': '示例部门'} or {'deptId': 11, 'deptName': '示例部门'}.", details={"received_value": value}, ) return {"id": dept_id, "value": _stringify_json(value.get("value", value.get("name", value.get("deptName", dept_id))))} dept_id = _coerce_count(value) if dept_id is None: raise RecordInputError( message="department values require numeric ids or department objects", error_code="INVALID_DEPARTMENT_VALUE", fix_hint="Pass department ids like 11 or objects like {'id': 11, 'value': '示例部门'}.", details={"received_value": value}, ) return {"id": dept_id, "value": str(dept_id)} def _attachment_value(value: JSONValue) -> JSONObject: if isinstance(value, dict): if "value" not in value and "url" not in value: raise RecordInputError( message="attachment values require value/url", error_code="INVALID_ATTACHMENT_VALUE", fix_hint="Pass attachments like {'value': 'https://.../a.pdf', 'name': 'a.pdf'}.", details={"received_value": value}, ) payload: JSONObject = {"value": value.get("value", value.get("url"))} if value.get("name") is not None: payload["name"] = value["name"] return payload return {"value": _stringify_json(value)} def _relation_value(value: JSONValue) -> JSONObject: if isinstance(value, dict): apply_id = value.get("apply_id", value.get("applyId", value.get("value", value.get("id")))) if apply_id is None: raise RecordInputError( message="relation values require apply_id/applyId/value/id", error_code="INVALID_RELATION_VALUE", fix_hint="Pass relation values like {'apply_id': 5001} or numeric apply ids.", details={"received_value": value}, ) return {"value": _stringify_json(apply_id)} return {"value": _stringify_json(value)} def _field_ref_payload(field: FormField) -> JSONObject: payload = {"que_id": field.que_id, "que_title": field.que_title, "que_type": field.que_type} if field.aliases: payload["aliases"] = field.aliases[:8] return payload def _normalize_field_lookup_key(value: str | int | None) -> str: text = _stringify_json(value).strip().lower() if not text: return "" return FIELD_LOOKUP_STRIP_RE.sub("", text) def _field_alias_candidates(field: FormField) -> set[str]: title = field.que_title.strip() aliases = set(GENERIC_FIELD_ALIAS_OVERRIDES.get(title, [])) for prefix in GENERIC_FIELD_PREFIX_ALIASES: if title.startswith(prefix) and len(title) > len(prefix) + 1: aliases.add(title[len(prefix):]) if title.endswith("所在部门"): aliases.add("部门") aliases.add(f"{title.removesuffix('所在部门')}部门") if title.endswith("阶段"): aliases.add("阶段") if title.endswith("来源"): aliases.add("来源") if title.endswith("类型"): aliases.add("类型") if title.endswith("等级"): aliases.add("等级") if title.endswith("名称"): aliases.add("名称") if title.endswith("编号"): aliases.add("编号") if title.endswith("金额"): aliases.add("金额") if field.que_type in MEMBER_QUE_TYPES: if "负责人" in title: aliases.add("负责人") if field.que_type in DEPARTMENT_QUE_TYPES: aliases.add("部门") if field.que_type in DATE_QUE_TYPES: if title.endswith("时间"): aliases.add("时间") if title.endswith("日期"): aliases.add("日期") normalized_title = _normalize_field_lookup_key(title) return { alias.strip() for alias in aliases if alias and _normalize_field_lookup_key(alias) and _normalize_field_lookup_key(alias) != normalized_title } def _subtable_columns_for_field(field: FormField) -> list[JSONObject]: raw = field.raw if isinstance(field.raw, dict) else {} columns_source = raw.get("subQuestions") if isinstance(columns_source, list): flattened = _flatten_questions(columns_source) else: inner_questions = raw.get("innerQuestions") flattened = _flatten_questions(inner_questions) if isinstance(inner_questions, list) else [] columns: list[JSONObject] = [] for question in flattened: que_id = _coerce_count(question.get("queId")) que_title = _normalize_optional_text(question.get("queTitle")) if que_id is None or que_title is None: continue columns.append( { "que_id": que_id, "que_title": que_title, "que_type": _coerce_count(question.get("queType")), } ) return columns def _write_support_payload( *, support_level: str, kind: str, examples: JSONValue | None = None, accepted_aliases: list[str] | None = None, options: list[str] | None = None, reason: str | None = None, fix_hint: str | None = None, required_presteps: list[str] | None = None, subfields: list[JSONObject] | None = None, row_shape: str | None = None, ) -> JSONObject: payload: JSONObject = { "support_level": support_level, "kind": kind, } if examples is not None: payload["examples"] = examples if accepted_aliases: payload["accepted_aliases"] = accepted_aliases if options is not None: payload["options"] = options if reason is not None: payload["reason"] = reason if fix_hint is not None: payload["fix_hint"] = fix_hint if required_presteps: payload["required_presteps"] = required_presteps if subfields: payload["subfields"] = subfields if row_shape is not None: payload["row_shape"] = row_shape return payload def _score_candidate_text(requested_key: str, candidate_text: str, *, fuzzy: bool, label: str) -> tuple[float, str]: candidate_key = _normalize_field_lookup_key(candidate_text) if not candidate_key: return 0.0, "none" if candidate_key == requested_key: return (0.98 if label == "title" else 0.95), f"{label}_exact" if candidate_key.startswith(requested_key): return (0.9 if label == "title" else 0.87), f"{label}_prefix" if requested_key in candidate_key: return (0.84 if label == "title" else 0.8), f"{label}_contains" if fuzzy: similarity = _normalized_text_similarity(requested_key, candidate_key) if similarity >= 0.3: return similarity, f"{label}_fuzzy" return 0.0, "none" def _write_format_for_field(field: FormField) -> JSONObject: if field.que_type in VERIFY_UNSUPPORTED_WRITE_QUE_TYPES: return _write_support_payload( support_level="unsupported", kind="unsupported_direct_write", reason=_unsupported_write_reason(field.que_type), fix_hint=_unsupported_write_fix_hint(field.que_type), ) if field.que_type in SUBTABLE_QUE_TYPES: return _write_support_payload( support_level="restricted", kind="subtable_rows", accepted_aliases=["rows", "tableValues", "rowId", "row_id", "__row_id__"], row_shape="list of row objects keyed by subfield title, or native tableValues rows", required_presteps=[ "Use the current subfield titles from the form schema.", "When updating existing rows, include rowId/row_id only if the source record already exposes it.", ], subfields=_subtable_columns_for_field(field), examples=[{"rows": [{"子字段1": "值", "子字段2": 1}]}], ) if field.que_type in MEMBER_QUE_TYPES: return _write_support_payload( support_level="restricted", kind="member_list", examples=[{"id": 2, "value": "张三"}], accepted_aliases=["uid", "userId", "userName"], required_presteps=["Pass member ids/userIds or member objects; display names alone are not resolved automatically."], ) if field.que_type in DEPARTMENT_QUE_TYPES: return _write_support_payload( support_level="restricted", kind="department_list", examples=[{"id": 11, "value": "示例部门"}], accepted_aliases=["deptId", "deptName"], required_presteps=["Pass a dept id/object, or an exact department name that resolves uniquely."], ) if field.que_type in ATTACHMENT_QUE_TYPES: return _write_support_payload( support_level="restricted", kind="attachment_list", examples=[{"value": "https://files.example.com/a.pdf", "name": "a.pdf"}], required_presteps=["Upload the file first with file_upload_local, then write the returned attachment value/url."], ) if field.que_type in RELATION_QUE_TYPES: return _write_support_payload( support_level="restricted", kind="relation_record", examples=[{"value": "5001"}], required_presteps=["Query the target app first and use the referenced apply_id."], ) if field.que_type in SINGLE_SELECT_QUE_TYPES: return _write_support_payload(support_level="full", kind="single_select", options=field.options) if field.que_type in MULTI_SELECT_QUE_TYPES: return _write_support_payload(support_level="full", kind="multi_select", options=field.options) if field.que_type in BOOLEAN_QUE_TYPES: return _write_support_payload(support_level="full", kind="boolean_label", examples=["是", "否"]) if field.que_type in DATE_QUE_TYPES: return _write_support_payload(support_level="full", kind="date_string", examples=["2026-03-13 10:00:00"]) return _write_support_payload(support_level="full", kind="scalar_text") def _summarize_write_support(resolved_fields: list[JSONObject]) -> JSONObject: summary: JSONObject = { "full": [], "restricted": [], "unsupported": [], } for entry in resolved_fields: if not bool(entry.get("resolved")): continue write_format = entry.get("write_format") if not isinstance(write_format, dict): continue support_level = str(write_format.get("support_level", "full")) bucket = summary.get(support_level) if not isinstance(bucket, list): continue bucket.append( { "source": entry.get("source"), "requested": entry.get("requested"), "que_id": entry.get("que_id"), "que_title": entry.get("que_title"), "que_type": entry.get("que_type"), "kind": write_format.get("kind"), } ) return summary def _subtable_row_id(row: list[JSONObject]) -> int | None: for cell in row: if not isinstance(cell, dict): continue row_id = _coerce_count(cell.get("rowId", cell.get("row_id"))) if row_id is not None: return row_id return None def _index_subtable_rows_by_row_id(rows: list[JSONValue]) -> dict[int, list[JSONObject]]: indexed: dict[int, list[JSONObject]] = {} for row in rows: if not isinstance(row, list): continue normalized_row = [item for item in row if isinstance(item, dict)] row_id = _subtable_row_id(normalized_row) if row_id is not None: indexed[row_id] = normalized_row return indexed def _subtable_cell_ref_payload( table_field: FormField | None, subfield: FormField | None, *, que_id: int, row_ordinal: int, row_id: int | None, ) -> JSONObject: payload: JSONObject = { "parent_que_id": table_field.que_id if table_field is not None else None, "parent_que_title": table_field.que_title if table_field is not None else None, "row_ordinal": row_ordinal, "que_id": que_id, "que_title": subfield.que_title if subfield is not None else None, "que_type": subfield.que_type if subfield is not None else None, } if row_id is not None: payload["row_id"] = row_id return payload def _unsupported_write_reason(que_type: int | None) -> str: if que_type == 14: return "time range fields require a backend-specific native payload shape that app-user tools do not build yet" if que_type == 34: return "image recognition fields are runtime AI fields and direct writes are not reliably persisted" if que_type == 35: return "image generation fields are runtime AI fields and direct writes are not reliably persisted" if que_type == 36: return "document parsing fields are runtime AI fields and direct writes are not reliably persisted" return "direct writes are not supported for this field type" def _unsupported_write_fix_hint(que_type: int | None) -> str: if que_type == 14: return "Avoid direct writes for time range fields until native payload support is added." if que_type in {34, 35, 36}: return "Avoid direct writes for AI/runtime fields, or submit through the product UI/workflow path that populates them." return "Avoid direct writes for this field type." def _parse_json_like(value: JSONValue) -> JSONValue: if isinstance(value, dict): return {key: _parse_json_like(item) for key, item in value.items()} if isinstance(value, list): return [_parse_json_like(item) for item in value] if isinstance(value, str): text = value.strip() if not text: return value if (text.startswith("{") and text.endswith("}")) or (text.startswith("[") and text.endswith("]")): try: return _parse_json_like(cast(JSONValue, json.loads(text))) except json.JSONDecodeError: return value if text.lower() in {"true", "false"}: return text.lower() == "true" if text.isdigit(): return int(text) return value def _as_selector_list(value: JSONValue) -> list[str | int]: if isinstance(value, list): return [cast(str | int, item) for item in value if isinstance(item, (str, int))] return [] def _as_object_list(value: JSONValue) -> list[JSONObject]: if isinstance(value, list): return [cast(JSONObject, item) for item in value if isinstance(item, dict)] return [] def _stringify_json(value: JSONValue) -> str: if value is None: return "" if isinstance(value, str): return value if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): return str(value) return json.dumps(value, ensure_ascii=False) def _normalized_text_similarity(left: str, right: str) -> float: if not left or not right: return 0.0 if left == right: return 1.0 left_bigrams = _bigrams(left) right_bigrams = _bigrams(right) if not left_bigrams or not right_bigrams: return 0.0 intersection = len(left_bigrams & right_bigrams) union = len(left_bigrams | right_bigrams) return (intersection / union) if union else 0.0 def _bigrams(text: str) -> set[str]: normalized = text.replace(" ", "") if len(normalized) <= 1: return {normalized} return {normalized[index : index + 2] for index in range(len(normalized) - 1)}