from __future__ import annotations from copy import deepcopy from typing import Any from ..spec_models import EntitySpec, FieldType, FormLayoutRowSpec, FormLayoutSectionSpec from .icon_utils import encode_workspace_icon QUESTION_TYPE_MAP = { FieldType.text: 2, FieldType.long_text: 3, FieldType.number: 8, FieldType.amount: 8, FieldType.date: 4, FieldType.datetime: 4, FieldType.member: 5, FieldType.department: 22, FieldType.single_select: 11, FieldType.multi_select: 12, FieldType.phone: 7, FieldType.email: 6, FieldType.address: 21, FieldType.attachment: 13, FieldType.boolean: 10, FieldType.q_linker: 20, FieldType.code_block: 26, FieldType.relation: 25, FieldType.subtable: 18, } OPTION_THEME_PALETTE = [ { "hoverBorderColor": "#D16243", "optionColor": "#FCE5DE", "textColor": "#571C0C", "themeId": 6, "tickColor": "#D16243", }, { "hoverBorderColor": "#63AD0E", "optionColor": "#E6F7D2", "textColor": "#244201", "themeId": 8, "tickColor": "#63AD0E", }, { "hoverBorderColor": "#26ACD1", "optionColor": "#DCF5FC", "textColor": "#054557", "themeId": 10, "tickColor": "#26ACD1", }, ] def compile_entity_form( entity: EntitySpec, *, include_package: bool, ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any] | None, dict[str, dict[str, Any]], dict[str, str]]: field_specs = {field.field_id: field.model_dump(mode="json") for field in entity.fields} field_labels = {field.field_id: field.label for field in entity.fields} app_create_payload = { "appName": entity.display_name, "appIcon": encode_workspace_icon( icon=entity.icon, color=entity.color, title=entity.display_name, fallback_icon_name="template", ), "auth": default_member_auth(), # Reserve the first slot inside a package for the package portal/homepage. "ordinal": (entity.ordinal or 1) + (1 if include_package else 0), } if include_package: app_create_payload["tagIds"] = ["__PACKAGE_TAG_ID__"] questions_by_field_id: dict[str, dict[str, Any]] = {} temp_id = -10000 for field in entity.fields: question, next_temp_id = build_question(field.model_dump(mode="json"), temp_id) questions_by_field_id[field.field_id] = question temp_id = next_temp_id base_questions = build_form_questions(entity, questions_by_field_id, include_reference=False) base_payload = default_form_payload(entity.display_name, base_questions) reference_payload = None if any(field.type == FieldType.relation for field in entity.fields): reference_questions = build_form_questions(entity, questions_by_field_id, include_reference=True) reference_payload = default_form_payload(entity.display_name, reference_questions) return app_create_payload, base_payload, reference_payload, field_specs, field_labels def default_member_auth() -> dict[str, Any]: return { "type": "WORKSPACE", "contactAuth": { "type": "WORKSPACE_ALL", "authMembers": { "member": [], "depart": [], "role": [], "dynamic": [], "includeSubDeparts": None, }, }, "externalMemberAuth": { "type": "NOT", "authMembers": { "member": [], "depart": [], "role": [], "dynamic": [], "includeSubDeparts": None, }, }, } def default_form_payload(form_title: str, form_questions: list[list[dict[str, Any]]]) -> dict[str, Any]: return { "formTitle": form_title, "serialNumType": 1, "formTheme": 0, "editVersionNo": 1, "questionRelations": [], "formQues": form_questions, } def build_form_questions( entity: EntitySpec, questions_by_field_id: dict[str, dict[str, Any]], *, include_reference: bool, ) -> list[list[dict[str, Any]]]: used_field_ids: set[str] = set() lines: list[list[dict[str, Any]]] = [] layout = entity.form_layout if layout is not None: for row in layout.rows: row_questions = _row_questions(row, questions_by_field_id, include_reference=include_reference, used_field_ids=used_field_ids) if row_questions: lines.append(row_questions) for section in layout.sections: section_question = _section_question( section, questions_by_field_id, include_reference=include_reference, used_field_ids=used_field_ids, ) if section_question is not None: lines.append([section_question]) else: for field in entity.fields: if field.type == FieldType.relation and not include_reference: continue lines.append([deepcopy(questions_by_field_id[field.field_id])]) used_field_ids.add(field.field_id) for field in entity.fields: if field.field_id in used_field_ids: continue if field.type == FieldType.relation and not include_reference: continue lines.append([deepcopy(questions_by_field_id[field.field_id])]) return lines def build_question(field: dict[str, Any], temp_id: int) -> tuple[dict[str, Any], int]: que_type = QUESTION_TYPE_MAP[field["type"]] config = field.get("config") or {} question = { "queId": 0, "queTempId": temp_id, "queType": que_type, "queTitle": field["label"], "queWidth": int(config.get("que_width") or config.get("width") or 100), "scanType": 1, "status": 1, "required": field["required"], "queHint": field.get("description") or "", "linkedQuestions": {}, "logicalShow": True, "queDefaultValue": None, "queDefaultType": 1, "subQueWidth": 2, "innerQuestions": [], "beingHide": False, "beingDesensitized": False, } next_temp_id = temp_id - 1 field_type = FieldType(field["type"]) if field_type in (FieldType.number, FieldType.amount, FieldType.text, FieldType.long_text, FieldType.phone, FieldType.email): question.update( { "minWordCount": -1, "maxWordCount": -1, "canDecimal": field_type == FieldType.amount, "numberFormat": 1, "currencyType": 1, } ) if field_type == FieldType.date: question["dateType"] = 0 if field_type == FieldType.datetime: question["dateType"] = 1 if field_type == FieldType.member: question.update({"memberDftValue": None, "memberSelectScopeType": 1, "memberSelectScope": {"member": []}}) if field_type == FieldType.department: question.update({"deptSelectScopeType": 1, "deptSelectScope": {"depart": []}}) if field_type in (FieldType.single_select, FieldType.multi_select, FieldType.boolean): options = field.get("options") or (["是", "否"] if field_type == FieldType.boolean else ["未命名1", "未命名2", "未命名3"]) question.update( { "options": build_options(options, start_temp_id=next_temp_id - len(options) + 1), "unnamedOptions": options, "optDirection": 0, "beingShowColor": True, } ) if field_type == FieldType.multi_select: question["optSelectMode"] = 1 next_temp_id -= len(options) if field_type == FieldType.attachment: question.update( { "fileBeingCaptureOnly": 0, "fileSize": 20, "fileType": [], "pluginStatus": False, "imageCompress": False, "queDefaultValues": {"queId": temp_id, "queTitle": field["label"], "queType": que_type, "values": []}, } ) if field_type == FieldType.address: question.update({"addressPrecision": 1, "queDefaultValues": {"queId": temp_id, "queTitle": field["label"], "queType": que_type, "values": [], "tableValues": []}}) if field_type == FieldType.relation: question.update( { "referenceConfig": build_reference_config(field, temp_id), "queOriginType": 25, "queDefaultType": 2, } ) if field_type == FieldType.code_block: question.update( { "minOpts": -1, "maxOpts": -1, "codeBlockConfig": { "configMode": int(config.get("config_mode") or 1), "codeContent": str(config.get("code_content") or ""), "resultAliasPath": deepcopy(config.get("result_alias_path") or []), "beingHideOnForm": bool(config.get("being_hide_on_form", False)), }, "autoTrigger": bool(config.get("auto_trigger", False)), "customBtnTextStatus": bool(config.get("custom_button_text_enabled", False)), "customBtnText": str(config.get("custom_button_text") or ""), } ) if field_type == FieldType.q_linker: question.update( { "minOpts": -1, "maxOpts": -1, "remoteLookupConfig": deepcopy(config.get("remote_lookup_config") or config), "autoTrigger": bool(config.get("auto_trigger", False)), "customBtnTextStatus": bool(config.get("custom_button_text_enabled", False)), "customBtnText": str(config.get("custom_button_text") or ""), } ) if field_type == FieldType.subtable: sub_questions: list[dict[str, Any]] = [] for subfield in field.get("subfields", []): sub_question, next_temp_id = build_question(subfield, next_temp_id) sub_questions.append(sub_question) question["subQuestions"] = sub_questions question["innerQuestions"] = [deepcopy(sub_questions)] question["queDefaultValues"] = {"queId": temp_id, "queTitle": field["label"], "queType": que_type, "values": [], "tableValues": []} return question, next_temp_id def build_reference_config(field: dict[str, Any], temp_id: int) -> dict[str, Any]: config = field.get("config") or {} display_field_id = field.get("target_field_id") or "title" display_field_que_id = field.get("target_field_que_id") or config.get("target_field_que_id") or 0 display_field_label = config.get("target_field_label") or "__TARGET_FIELD_LABEL__" refer_field_ids = list(config.get("refer_field_ids") or [display_field_id]) refer_field_que_ids = list(config.get("refer_field_que_ids") or []) refer_field_labels = config.get("refer_field_labels") or [display_field_label] refer_field_types = config.get("refer_field_types") refer_questions = [] for ordinal, field_id in enumerate(refer_field_ids, start=1): label = refer_field_labels[ordinal - 1] if ordinal - 1 < len(refer_field_labels) else field_id que_id = refer_field_que_ids[ordinal - 1] if ordinal - 1 < len(refer_field_que_ids) else 0 raw_type = _reference_field_type_value( refer_field_types, field_id=field_id, ordinal=ordinal, default=config.get("target_field_type") if field_id == display_field_id else None, ) refer_questions.append( { "queId": que_id, "queTitle": label, "queType": _normalize_reference_que_type(raw_type) or "2", "queAuth": 3, "ordinal": ordinal, "quoteId": temp_id, "_field_id": field_id, } ) auth_field_ids = list(config.get("auth_field_ids") or refer_field_ids) auth_field_que_ids = list(config.get("auth_field_que_ids") or []) auth_ques = deepcopy(config.get("refer_auth_ques") or []) if not auth_ques: auth_ques = [] for ordinal, field_id in enumerate(auth_field_ids, start=1): que_id = auth_field_que_ids[ordinal - 1] if ordinal - 1 < len(auth_field_que_ids) else 0 auth_ques.append({"queId": que_id, "queAuth": 3, "_field_id": field_id}) return { "referAppKey": "__TARGET_APP_KEY__", "referQueId": display_field_que_id, "customButtonText": config.get("custom_button_text") or "选择数据", "beingTableSource": False, "referQuestions": refer_questions, "referMatchRules": deepcopy(config.get("refer_match_rules") or []), "referFillRules": deepcopy(config.get("refer_fill_rules") or []), "referAuthQues": auth_ques, "canAddData": bool(config.get("can_add_data", False)), "dataAdditionButtonText": config.get("data_addition_button_text") or "新增数据", "canViewProcessLog": bool(config.get("can_view_process_log", True)), "optionalDataNum": int(config.get("optional_data_num", 1)), "beingDataLogVisible": bool(config.get("being_data_log_visible", True)), "beingDefaultFormulaAutoFillEnabled": bool(config.get("being_default_formula_auto_fill_enabled", False)), "defaultValueMatchRules": deepcopy(config.get("default_value_match_rules") or []), "configShowForm": config.get("config_show_form") or "TABLE", "configSortFieldId": None, "configAsc": config.get("config_asc", True), "dataShowForm": config.get("data_show_form") or "CARD", "defaultRow": config.get("default_row"), "fieldNameShow": config.get("field_name_show", True), "dataSortFieldId": None, "dataSortAsc": config.get("data_sort_asc"), "_targetFieldId": display_field_id, "_targetEntityId": field.get("target_entity_id"), } def build_options(options: list[str], *, start_temp_id: int) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] current = start_temp_id for index, option in enumerate(options): items.append( { "otherValue": None, "beingDefault": False, "optId": 0, "optValue": option, "optUrl": "", "beingOtherOpt": False, "beingRequired": False, "applyLimit": -1, "currentApply": 0, "linkQueIds": [], "emptyText": "", "tempId": current, "optionThemeColor": deepcopy(OPTION_THEME_PALETTE[index % len(OPTION_THEME_PALETTE)]), } ) current += 1 return items def _row_questions( row: FormLayoutRowSpec, questions_by_field_id: dict[str, dict[str, Any]], *, include_reference: bool, used_field_ids: set[str], ) -> list[dict[str, Any]]: questions: list[dict[str, Any]] = [] for field_id in row.field_ids: question = deepcopy(questions_by_field_id[field_id]) if question.get("queType") == 25 and not include_reference: continue used_field_ids.add(field_id) questions.append(question) widths = _resolve_row_widths(row, questions) for question, width in zip(questions, widths): question["queWidth"] = width return questions def _section_question( section: FormLayoutSectionSpec, questions_by_field_id: dict[str, dict[str, Any]], *, include_reference: bool, used_field_ids: set[str], ) -> dict[str, Any] | None: inner_rows: list[list[dict[str, Any]]] = [] for row in section.rows: row_questions = _row_questions(row, questions_by_field_id, include_reference=include_reference, used_field_ids=used_field_ids) if row_questions: inner_rows.append(row_questions) if not inner_rows: return None config = section.config or {} return { "queId": 0, "queTempId": -(20000 + sum(ord(ch) for ch in section.section_id)), "queType": 24, "queTitle": section.title, "queWidth": int(config.get("que_width") or 100), "scanType": 1, "status": 1, "required": False, "queHint": config.get("description") or "", "linkedQuestions": {}, "logicalShow": True, "queDefaultValue": None, "queDefaultType": 1, "subQueWidth": 2, "innerQuestions": inner_rows, "beingHide": False, "beingDesensitized": False, "sectionId": config.get("section_id"), } def _resolve_row_widths(row: FormLayoutRowSpec, questions: list[dict[str, Any]]) -> list[int]: if not questions: return [] configured_widths = row.config.get("field_widths") if isinstance(row.config.get("field_widths"), dict) else {} widths: list[int] = [] remaining_slots = 0 assigned_total = 0 for field_id, question in zip(row.field_ids, questions): width = configured_widths.get(field_id) if width is None: widths.append(-1) remaining_slots += 1 continue resolved_width = int(width) widths.append(resolved_width) assigned_total += resolved_width if remaining_slots <= 0: return widths remaining_total = max(0, 100 - assigned_total) auto_widths = _auto_row_widths(remaining_slots, total=remaining_total) auto_index = 0 for index, width in enumerate(widths): if width >= 0: continue widths[index] = auto_widths[auto_index] auto_index += 1 return widths def _auto_row_widths(count: int, *, total: int = 100) -> list[int]: if count <= 0: return [] base = total // count remainder = total % count return [base + (1 if index < remainder else 0) for index in range(count)] def _reference_field_type_value(raw_types: Any, *, field_id: str, ordinal: int, default: Any) -> Any: if isinstance(raw_types, dict): return raw_types.get(field_id, default) if isinstance(raw_types, list): index = ordinal - 1 if 0 <= index < len(raw_types): return raw_types[index] return default def _normalize_reference_que_type(raw_type: Any) -> str | None: if raw_type is None: return None if isinstance(raw_type, int): return str(raw_type) value = str(raw_type).strip() if not value: return None if value.isdigit(): return value try: return str(QUESTION_TYPE_MAP[FieldType(value)]) except ValueError: return None