from __future__ import annotations from dataclasses import dataclass import re from typing import Any from uuid import uuid4 from .spec_models import EntityKind, FieldType @dataclass class GeneratedAppBuild: app_spec: dict[str, Any] summary: dict[str, Any] class RequirementsBuildError(ValueError): def __init__( self, message: str, *, error_code: str = "REQUIREMENTS_INVALID", recoverable: bool = True, missing_required_fields: list[str] | None = None, invalid_field_types: list[str] | None = None, details: dict[str, Any] | None = None, ) -> None: super().__init__(message) self.error_code = error_code self.recoverable = recoverable self.missing_required_fields = list(missing_required_fields or []) self.invalid_field_types = list(invalid_field_types or []) self.details = dict(details or {}) SAFE_DEFAULT_FIELD_BLUEPRINTS: list[dict[str, Any]] = [ {"field_id": "title", "label": "标题", "type": FieldType.text.value, "required": True}, {"field_id": "description", "label": "说明", "type": FieldType.long_text.value}, {"field_id": "quantity", "label": "数量", "type": FieldType.number.value}, {"field_id": "amount", "label": "金额", "type": FieldType.amount.value}, {"field_id": "event_date", "label": "日期", "type": FieldType.date.value}, {"field_id": "event_time", "label": "日期时间", "type": FieldType.datetime.value}, {"field_id": "owner", "label": "负责人", "type": FieldType.member.value}, {"field_id": "department", "label": "所属部门", "type": FieldType.department.value}, { "field_id": "status", "label": "状态", "type": FieldType.single_select.value, "options": ["待处理", "处理中", "已完成"], }, { "field_id": "tags", "label": "标签", "type": FieldType.multi_select.value, "options": ["标签A", "标签B", "标签C"], }, {"field_id": "phone", "label": "联系电话", "type": FieldType.phone.value}, {"field_id": "email", "label": "邮箱", "type": FieldType.email.value}, {"field_id": "address", "label": "地址", "type": FieldType.address.value}, {"field_id": "attachments", "label": "附件", "type": FieldType.attachment.value}, {"field_id": "enabled", "label": "是否启用", "type": FieldType.boolean.value, "options": ["是", "否"]}, ] ADVANCED_FIELD_BLUEPRINTS: list[dict[str, Any]] = [ { "field_id": "related_record", "label": "关联记录", "type": FieldType.relation.value, "target_field_id": "title", }, { "field_id": "detail_items", "label": "明细", "type": FieldType.subtable.value, "subfields": [ {"field_id": "item_name", "label": "子项名称", "type": FieldType.text.value, "required": True}, {"field_id": "item_amount", "label": "子项金额", "type": FieldType.amount.value}, ], }, ] SUPPORTED_FIELD_BLUEPRINTS: list[dict[str, Any]] = SAFE_DEFAULT_FIELD_BLUEPRINTS + ADVANCED_FIELD_BLUEPRINTS FIELD_BLUEPRINTS_BY_TYPE = {str(item["type"]): item for item in SUPPORTED_FIELD_BLUEPRINTS} SAFE_DEFAULT_FIELD_TYPES = [str(item["type"]) for item in SAFE_DEFAULT_FIELD_BLUEPRINTS] ADVANCED_FIELD_TYPES = [str(item["type"]) for item in ADVANCED_FIELD_BLUEPRINTS] ADVANCED_FIELD_LABELS = { FieldType.relation.value: "relation", FieldType.subtable.value: "subtable", } FIELD_TYPE_ALIASES: dict[str, tuple[str, ...]] = { FieldType.text.value: ("单行文本", "文本", "标题", "名称"), FieldType.long_text.value: ("多行文本", "备注", "描述", "说明"), FieldType.number.value: ("数字", "数量", "计数", "整数"), FieldType.amount.value: ("金额", "金额字段", "预算", "费用"), FieldType.date.value: ("日期", "开始日期", "截止日期"), FieldType.datetime.value: ("日期时间", "时间", "时间戳"), FieldType.member.value: ("成员", "负责人", "审批人"), FieldType.department.value: ("部门", "组织", "团队"), FieldType.single_select.value: ("单选", "状态", "下拉单选"), FieldType.multi_select.value: ("多选", "标签", "下拉多选"), FieldType.phone.value: ("电话", "手机号", "手机"), FieldType.email.value: ("邮箱", "电子邮箱", "email"), FieldType.address.value: ("地址", "所在地", "location"), FieldType.attachment.value: ("附件", "文件", "图片"), FieldType.boolean.value: ("开关", "布尔", "是否"), FieldType.relation.value: ("关联", "引用", "关联记录"), FieldType.subtable.value: ("子表", "明细表", "明细", "table"), } ALL_FIELDS_HINTS = ("所有字段", "全字段", "全部字段", "字段完整", "所有字段类型", "常见字段类型") LAYOUT_HINTS = { "balanced": ("优美", "美观", "清晰", "有层次", "布局优美", "完整布局", "布局完整"), "grouped": ("分组", "section", "grouped"), "compact": ("紧凑", "compact"), "flat": ("平铺", "flat"), } ENTITY_KIND_HINTS = { EntityKind.master.value: ("主数据", "基础资料", "档案", "字典", "客户", "员工"), EntityKind.transaction.value: ("申请", "单据", "审批", "事务", "流程"), EntityKind.activity.value: ("跟进", "活动", "日志", "记录"), } def build_app_spec_from_requirements( *, title: str, requirement_text: str, package_name: str | None, layout_style: str = "auto", ) -> GeneratedAppBuild: cleaned_title = title.strip() cleaned_text = requirement_text.strip() if not cleaned_title: raise ValueError("title is required") if not cleaned_text: raise ValueError("requirement_text is required") resolved_layout = _resolve_layout_style(cleaned_text, layout_style) entity_kind = _resolve_entity_kind(cleaned_text) entity_id = _slugify_identifier(cleaned_title, prefix="entity") package_label = package_name.strip() if isinstance(package_name, str) and package_name.strip() else None selected_types = _resolve_requested_field_types(cleaned_text) advanced_requested = [field_type for field_type in selected_types if field_type in ADVANCED_FIELD_TYPES] all_fields_mode = any(hint in cleaned_text for hint in ALL_FIELDS_HINTS) if all_fields_mode: selected_types = list(SAFE_DEFAULT_FIELD_TYPES) elif not selected_types: selected_types = [ FieldType.text.value, FieldType.long_text.value, FieldType.single_select.value, FieldType.member.value, FieldType.date.value, FieldType.attachment.value, ] resolved_advanced_fields = _resolve_advanced_fields(requirement_text=cleaned_text, entity_id=entity_id) advanced_admitted = [field["type"] for field in resolved_advanced_fields] excluded_advanced_fields = [ ADVANCED_FIELD_LABELS.get(field_type, field_type) for field_type in ADVANCED_FIELD_TYPES if field_type not in advanced_admitted ] fields = _build_fields_for_types(selected_types=selected_types, advanced_fields=resolved_advanced_fields) layout = _build_layout(fields=fields, layout_style=resolved_layout) follow_up_actions = _infer_follow_up_actions(cleaned_text) assumptions = _infer_assumptions( cleaned_text, all_fields_mode=all_fields_mode, layout_style=resolved_layout, excluded_advanced_fields=excluded_advanced_fields, ) app_spec = { "solution_name": cleaned_title, "summary": cleaned_text, "package": {"enabled": True, "name": package_label or cleaned_title}, "publish_policy": {"apps": True, "portal": False, "navigation": False}, "preferences": {"create_package": package_label is None, "create_portal": False, "create_navigation": False}, "assumptions": assumptions, "metadata": { "generator": { "mode": "heuristic_requirements_interpreter", "all_fields_mode": all_fields_mode, "resolved_layout_style": resolved_layout, "field_selection_policy": "safe_default", "excluded_advanced_fields": excluded_advanced_fields, } }, "entities": [ { "entity_id": entity_id, "display_name": cleaned_title, "kind": entity_kind, "title_field_id": "title", "fields": fields, "form_layout": layout, "sample_records": [], "acceptance_criteria": [ "表单可以成功创建并发布", "字段与布局满足需求文本中的显式要求", ], } ], } summary = { "title": cleaned_title, "entity_id": entity_id, "entity_kind": entity_kind, "package_name": package_label, "resolved_layout_style": resolved_layout, "field_count": len(fields), "field_labels": [field["label"] for field in fields], "field_types": [field["type"] for field in fields], "all_fields_mode": all_fields_mode, "field_selection_policy": "safe_default", "excluded_advanced_fields": excluded_advanced_fields, "advanced_fields_requested": [ADVANCED_FIELD_LABELS.get(field_type, field_type) for field_type in advanced_requested], "advanced_fields_admitted": [ADVANCED_FIELD_LABELS.get(field_type, field_type) for field_type in advanced_admitted], "assumptions": assumptions, "follow_up_actions": follow_up_actions, } return GeneratedAppBuild(app_spec=app_spec, summary=summary) def _resolve_requested_field_types(requirement_text: str) -> list[str]: selected: list[str] = [] for field_type, aliases in FIELD_TYPE_ALIASES.items(): if any(alias.lower() in requirement_text.lower() for alias in aliases): selected.append(field_type) if FieldType.text.value not in selected: selected.insert(0, FieldType.text.value) return selected def _build_fields_for_types(*, selected_types: list[str], advanced_fields: list[dict[str, Any]]) -> list[dict[str, Any]]: ordered = [] seen: set[str] = set() for field_type in SAFE_DEFAULT_FIELD_TYPES: if field_type not in selected_types: continue blueprint = FIELD_BLUEPRINTS_BY_TYPE[field_type] field_id = str(blueprint["field_id"]) if field_id in seen: continue seen.add(field_id) ordered.append(_clone_field_blueprint(blueprint)) for field in advanced_fields: field_id = str(field["field_id"]) if field_id in seen: continue seen.add(field_id) ordered.append(_clone_field_blueprint(field)) return ordered def _build_layout(*, fields: list[dict[str, Any]], layout_style: str) -> dict[str, Any]: field_ids = [str(field["field_id"]) for field in fields] if layout_style == "flat": return {"rows": [{"field_ids": [field_id]} for field_id in field_ids]} if layout_style == "compact": pairs = [field_ids[index : index + 2] for index in range(0, len(field_ids), 2)] return {"rows": [{"field_ids": pair} for pair in pairs]} if layout_style == "balanced": return _build_balanced_layout(fields) sections = [] grouped_ids = { "basic": [item for item in field_ids if item in {"title", "description", "quantity", "amount", "event_date", "event_time"}], "collaboration": [item for item in field_ids if item in {"owner", "department", "status", "tags", "enabled"}], "contact": [item for item in field_ids if item in {"phone", "email", "address"}], "extended": [item for item in field_ids if item in {"attachments", "related_record", "detail_items"}], } section_titles = { "basic": "基础信息", "collaboration": "协作信息", "contact": "联系信息", "extended": "扩展信息", } for section_id, ids in grouped_ids.items(): if not ids: continue rows = [{"field_ids": ids[index : index + 2]} for index in range(0, len(ids), 2)] sections.append({"section_id": section_id, "title": section_titles[section_id], "rows": rows}) return {"sections": sections or [{"section_id": "basic", "title": "基础信息", "rows": [{"field_ids": field_ids}]}]} def _resolve_layout_style(requirement_text: str, layout_style: str) -> str: raw = (layout_style or "auto").strip().lower() if raw in {"balanced", "grouped", "compact", "flat"}: return raw for candidate, aliases in LAYOUT_HINTS.items(): if any(alias.lower() in requirement_text.lower() for alias in aliases): return candidate return "grouped" def _resolve_entity_kind(requirement_text: str) -> str: for kind, aliases in ENTITY_KIND_HINTS.items(): if any(alias.lower() in requirement_text.lower() for alias in aliases): return kind return EntityKind.transaction.value def _infer_follow_up_actions(requirement_text: str) -> list[str]: actions: list[str] = [] lower_text = requirement_text.lower() if any(token in lower_text for token in ("流程", "审批", "workflow", "approve")): actions.append("如果需要流程，请在应用创建后继续调用 solution_build_flow。") if any(token in lower_text for token in ("视图", "看板", "view", "board")): actions.append("如果需要视图，请在应用创建后继续调用 solution_build_views。") if any(token in lower_text for token in ("报表", "门户", "chart", "portal")): actions.append("如果需要报表或门户，请在应用创建后继续调用 solution_build_analytics_portal。") return actions def _infer_assumptions( requirement_text: str, *, all_fields_mode: bool, layout_style: str, excluded_advanced_fields: list[str], ) -> list[str]: assumptions = [f"布局风格已解析为 {layout_style}。"] if all_fields_mode: assumptions.append("已将“所有字段”解释为安全默认字段集合，不自动包含 relation/subtable。") if excluded_advanced_fields: assumptions.append(f"以下高级字段未自动纳入：{', '.join(excluded_advanced_fields)}。") return assumptions def _resolve_advanced_fields(*, requirement_text: str, entity_id: str) -> list[dict[str, Any]]: advanced_fields: list[dict[str, Any]] = [] lower_text = requirement_text.lower() if any(alias.lower() in lower_text for alias in FIELD_TYPE_ALIASES[FieldType.relation.value]): relation_field = _build_relation_field(requirement_text=requirement_text, entity_id=entity_id) advanced_fields.append(relation_field) if any(alias.lower() in lower_text for alias in FIELD_TYPE_ALIASES[FieldType.subtable.value]): subtable_field = _build_subtable_field(requirement_text=requirement_text) advanced_fields.append(subtable_field) return advanced_fields def _build_relation_field(*, requirement_text: str, entity_id: str) -> dict[str, Any]: target_phrase = _extract_relation_target(requirement_text) if not target_phrase: raise RequirementsBuildError( "relation field requires an explicit target entity; specify what it should relate to", missing_required_fields=["relation.target_entity_id"], invalid_field_types=["relation"], details={ "requested_field_type": "relation", "suggested_requirement_hint": "例如：关联到客户档案。", "suggested_patch": { "entities": [ { "entity_id": "target_entity", "display_name": "目标应用", "kind": EntityKind.master.value, "title_field_id": "title", "fields": [{"field_id": "title", "label": "标题", "type": FieldType.text.value, "required": True}], "form_layout": {"rows": [{"field_ids": ["title"]}]}, } ] }, }, ) target_entity_id = entity_id if target_phrase in {"当前应用", "当前表单", "自身", "本应用"} else _slugify_identifier(target_phrase, prefix="entity") field = _clone_field_blueprint(FIELD_BLUEPRINTS_BY_TYPE[FieldType.relation.value]) field["target_entity_id"] = target_entity_id return field def _build_subtable_field(*, requirement_text: str) -> dict[str, Any]: subfields = _extract_subtable_subfields(requirement_text) if not subfields: raise RequirementsBuildError( "subtable field requires explicit child field definitions; specify the row fields first", missing_required_fields=["subtable.subfields"], invalid_field_types=["subtable"], details={ "requested_field_type": "subtable", "suggested_requirement_hint": "例如：明细表字段包括 子项名称、数量、金额。", "suggested_patch": { "entities": [ { "fields": [ { "field_id": "detail_items", "label": "明细", "type": FieldType.subtable.value, "subfields": [ {"field_id": "item_name", "label": "子项名称", "type": FieldType.text.value, "required": True}, {"field_id": "quantity", "label": "数量", "type": FieldType.number.value}, {"field_id": "amount", "label": "金额", "type": FieldType.amount.value}, ], } ] } ] }, }, ) field = _clone_field_blueprint(FIELD_BLUEPRINTS_BY_TYPE[FieldType.subtable.value]) field["subfields"] = subfields return field def _extract_relation_target(requirement_text: str) -> str | None: explicit_self = ("当前应用", "当前表单", "自身", "本应用") if any(token in requirement_text for token in explicit_self): return next(token for token in explicit_self if token in requirement_text) patterns = ( r"(?:关联到|关联至|引用到|引用自)\s*([A-Za-z0-9_\-\u4e00-\u9fff]{2,32})", r"(?:关联记录目标|关联目标)\s*[:：]?\s*([A-Za-z0-9_\-\u4e00-\u9fff]{2,32})", ) for pattern in patterns: match = re.search(pattern, requirement_text) if not match: continue candidate = match.group(1).strip() if candidate and candidate not in {"记录", "应用", "表单", "单据"}: return candidate return None def _extract_subtable_subfields(requirement_text: str) -> list[dict[str, Any]]: patterns = ( r"(?:子表|明细表|明细)(?:字段|列)?(?:包括|包含|为|如下|:|：)\s*([^\n。；;]+)", ) for pattern in patterns: match = re.search(pattern, requirement_text) if not match: continue body = match.group(1).strip() parts = [part.strip() for part in re.split(r"[，,、/；;]+", body) if part.strip()] if len(parts) < 2: continue subfields: list[dict[str, Any]] = [] for index, part in enumerate(parts, start=1): resolved_types = _resolve_requested_field_types(part) field_type = resolved_types[0] if resolved_types else FieldType.text.value if field_type in ADVANCED_FIELD_TYPES: field_type = FieldType.text.value label = re.sub(r"(单行文本|多行文本|数字|金额|日期时间|日期|成员|部门|单选|多选|电话|手机号|邮箱|地址|附件|开关)$", "", part).strip("（）() ") if not label: label = f"子项字段{index}" subfield = { "field_id": _slugify_identifier(label, prefix=f"sub_{index}"), "label": label, "type": field_type, "required": index == 1, } if field_type == FieldType.single_select.value: subfield["options"] = ["选项A", "选项B", "选项C"] if field_type == FieldType.multi_select.value: subfield["options"] = ["标签A", "标签B", "标签C"] subfields.append(subfield) return subfields return [] def _build_balanced_layout(fields: list[dict[str, Any]]) -> dict[str, Any]: sections: list[dict[str, Any]] = [] hero_rows: list[dict[str, Any]] = [] detail_fields: list[str] = [] coordination_fields: list[str] = [] contact_fields: list[str] = [] attachment_fields: list[str] = [] advanced_fields: list[str] = [] for field in fields: field_id = str(field["field_id"]) field_type = str(field["type"]) if field_id == "title" or field_type == FieldType.long_text.value: hero_rows.append({"field_ids": [field_id]}) continue if field_type in {FieldType.phone.value, FieldType.email.value, FieldType.address.value}: contact_fields.append(field_id) continue if field_type in {FieldType.attachment.value}: attachment_fields.append(field_id) continue if field_type in {FieldType.relation.value, FieldType.subtable.value}: advanced_fields.append(field_id) continue if field_type in {FieldType.member.value, FieldType.department.value, FieldType.single_select.value, FieldType.multi_select.value, FieldType.boolean.value}: coordination_fields.append(field_id) continue detail_fields.append(field_id) if hero_rows: sections.append({"section_id": "hero", "title": "基础信息", "rows": hero_rows}) if detail_fields: sections.append({"section_id": "details", "title": "业务字段", "rows": _pair_rows(detail_fields)}) if coordination_fields: sections.append({"section_id": "coordination", "title": "协作信息", "rows": _pair_rows(coordination_fields)}) if contact_fields: sections.append({"section_id": "contact", "title": "联系信息", "rows": _pair_rows(contact_fields)}) if attachment_fields: sections.append({"section_id": "attachments", "title": "附件与扩展", "rows": [{"field_ids": [field_id]} for field_id in attachment_fields]}) if advanced_fields: sections.append({"section_id": "advanced", "title": "高级字段", "rows": [{"field_ids": [field_id]} for field_id in advanced_fields]}) if not sections: return {"sections": [{"section_id": "basic", "title": "基础信息", "rows": _pair_rows([str(field["field_id"]) for field in fields])}]} return {"sections": sections} def _pair_rows(field_ids: list[str]) -> list[dict[str, Any]]: return [{"field_ids": field_ids[index : index + 2]} for index in range(0, len(field_ids), 2)] def _clone_field_blueprint(blueprint: dict[str, Any]) -> dict[str, Any]: field = { "field_id": blueprint["field_id"], "label": blueprint["label"], "type": blueprint["type"], "required": bool(blueprint.get("required", False)), } if blueprint.get("options"): field["options"] = list(blueprint["options"]) if blueprint.get("target_field_id"): field["target_field_id"] = blueprint["target_field_id"] if blueprint.get("subfields"): field["subfields"] = [_clone_field_blueprint(subfield) for subfield in blueprint["subfields"]] return field def _slugify_identifier(value: str, *, prefix: str) -> str: normalized = re.sub(r"[^a-zA-Z0-9]+", "_", value).strip("_").lower() if normalized: return normalized[:48] return f"{prefix}_{uuid4().hex[:8]}"