"""
智能表单收集MCP服务器配置模块
"""
import os
from enum import Enum
from typing import Dict, List, Optional, Any
from pydantic import BaseModel, Field
from datetime import datetime
# 字段类型枚举
class FieldType(str, Enum):
TEXT = "text"
EMAIL = "email"
PHONE = "phone"
DATE = "date"
NUMBER = "number"
TEXTAREA = "textarea"
SELECT = "select"
MULTISELECT = "multiselect"
# 表单字段定义
class FormField(BaseModel):
name: str
label: str
field_type: FieldType
required: bool = True
min_length: Optional[int] = None
max_length: Optional[int] = None
pattern: Optional[str] = None
options: Optional[List[str]] = None
placeholder: Optional[str] = None
help_text: Optional[str] = None
# 表单模板定义
class FormTemplate(BaseModel):
name: str
title: str
description: str
fields: List[FormField]
# DeepSeek大模型配置
class LLMConfig(BaseModel):
provider: str = "deepseek"
model: str = "deepseek-chat"
api_key: str = ""
base_url: str = "https://api.deepseek.com/v1"
max_tokens: int = 4000
temperature: float = 0.7
stream: bool = True # 支持流式输出
# MySQL数据库配置
class DatabaseConfig(BaseModel):
url: str = "mysql+aiomysql://user:password@localhost:3306/database"
pool_size: int = 10
max_overflow: int = 20
pool_timeout: int = 30
pool_recycle: int = 3600
# 系统配置
class SystemConfig(BaseModel):
session_timeout: int = 3600 # 会话超时时间（秒）
max_sessions: int = 1000 # 最大会话数
cleanup_interval: int = 300 # 清理间隔（秒）
log_level: str = "INFO"
enable_cors: bool = True
cors_origins: List[str] = ["*"]
# 矛盾调解表单模板
MEDIATION_FORM_TEMPLATE = FormTemplate(
name="mediation",
title="矛盾调解业务受理表单",
description="用于收集矛盾调解案件的基本信息和事件详情",
fields=[
# 个人基本信息
FormField(
name="applicant_name",
label="申请人姓名",
field_type=FieldType.TEXT,
required=True,
min_length=2,
max_length=20,
placeholder="请输入您的真实姓名",
help_text="请提供申请人的真实姓名"
),
FormField(
name="contact_phone",
label="联系电话",
field_type=FieldType.PHONE,
required=True,
pattern=r"^1[3-9]\d{9}$",
placeholder="请输入11位手机号码",
help_text="请提供有效的手机号码以便联系"
),
FormField(
name="contact_address",
label="联系地址",
field_type=FieldType.TEXTAREA,
required=True,
min_length=10,
max_length=200,
placeholder="请输入详细的联系地址",
help_text="请提供详细的联系地址，包括省市区街道门牌号"
),
# 事件信息
FormField(
name="incident_time",
label="事件发生时间",
field_type=FieldType.TEXT,
required=True,
min_length=5,
max_length=50,
placeholder="如：2024年1月15日下午3点",
help_text="请描述事件发生的具体时间"
),
FormField(
name="incident_location",
label="事件发生地点",
field_type=FieldType.TEXT,
required=True,
min_length=5,
max_length=100,
placeholder="请输入事件发生的具体地点",
help_text="请提供事件发生的详细地点信息"
),
FormField(
name="incident_description",
label="事件详情描述",
field_type=FieldType.TEXTAREA,
required=True,
min_length=20,
max_length=1000,
placeholder="请详细描述事件的经过和争议焦点",
help_text="请详细描述事件的起因、经过、结果和争议焦点"
),
FormField(
name="involved_parties_count",
label="涉及人数",
field_type=FieldType.NUMBER,
required=True,
min_length=1,
placeholder="包括申请人在内的总人数",
help_text="请输入涉及此次矛盾的总人数（包括申请人）"
)
]
)
# 表单模板注册表
FORM_TEMPLATES: Dict[str, FormTemplate] = {
"mediation": MEDIATION_FORM_TEMPLATE
}
# 系统提示词配置
SYSTEM_PROMPTS = {
"welcome": """
欢迎使用智能表单收集系统！
我是您的专属助手，将帮助您完成矛盾调解业务受理表单的填写。
为了更好地为您服务，我需要收集以下信息：
**个人基本信息：**
- 申请人姓名
- 联系电话
- 联系地址
**事件信息：**
- 事件发生时间
- 事件发生地点
- 事件详情描述
- 涉及人数
您可以一次性提供多个信息，我会智能识别并整理。让我们开始吧！
""",
"extraction": """
你是一个专业的信息提取助手。请从用户输入中提取以下字段信息：
字段列表：
- applicant_name: 申请人姓名
- contact_phone: 联系电话
- contact_address: 联系地址
- incident_time: 事件发生时间
- incident_location: 事件发生地点
- incident_description: 事件详情描述
- involved_parties_count: 涉及人数
请仔细分析用户输入，提取所有能识别的字段信息。
对于电话号码，请识别11位手机号格式。
对于人数，请提取数字信息。
对于时间，请识别时间相关描述。
对于地点，请识别地址或位置信息。
返回JSON格式的提取结果。
""",
"response_generation": """
你是一个友好专业的表单收集助手。请根据当前收集状态生成合适的响应：
1. 如果成功提取了信息，请确认收到的信息并询问缺失的字段
2. 如果信息不完整，请友好地引导用户提供缺失信息
3. 如果信息有误，请礼貌地要求用户确认或修正
4. 使用Markdown表格清晰展示收集状态
5. 保持专业友好的语调
请生成合适的响应消息。
"""
}
# 环境变量配置
def get_config() -> Dict[str, Any]:
"""获取系统配置"""
return {
"llm": LLMConfig(
provider=os.getenv("LLM_PROVIDER", "deepseek"),
model=os.getenv("LLM_MODEL", "deepseek-chat"),
api_key=os.getenv("DEEPSEEK_API_KEY", ""),
base_url=os.getenv("LLM_BASE_URL", "https://api.deepseek.com/v1"),
max_tokens=int(os.getenv("LLM_MAX_TOKENS", "4000")),
temperature=float(os.getenv("LLM_TEMPERATURE", "0.7")),
stream=os.getenv("LLM_STREAM", "true").lower() == "true"
),
"database": DatabaseConfig(
url=os.getenv("DATABASE_URL", "mysql+aiomysql://user:password@localhost:3306/database"),
pool_size=int(os.getenv("DB_POOL_SIZE", "10")),
max_overflow=int(os.getenv("DB_MAX_OVERFLOW", "20")),
pool_timeout=int(os.getenv("DB_POOL_TIMEOUT", "30")),
pool_recycle=int(os.getenv("DB_POOL_RECYCLE", "3600"))
),
"system": SystemConfig(
session_timeout=int(os.getenv("SESSION_TIMEOUT", "3600")),
max_sessions=int(os.getenv("MAX_SESSIONS", "1000")),
cleanup_interval=int(os.getenv("CLEANUP_INTERVAL", "300")),
log_level=os.getenv("LOG_LEVEL", "INFO"),
enable_cors=os.getenv("ENABLE_CORS", "true").lower() == "true",
cors_origins=os.getenv("CORS_ORIGINS", "*").split(",")
)
}
