# -*- coding: utf-8 -*- """文档对话模块数据模型。""" from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field class SelectedSection(BaseModel): index: str = Field(default="", description="章节编号,例如 2.1") title: str = Field(default="", description="章节标题") content: str = Field(default="", description="当前编辑器中的章节内容") code: str = Field(default="", description="章节代码标识") chapter_level_1: str = Field(default="", description="可选的一级章节分类,用于检索范围限定") chapter_level_2: str = Field(default="", description="可选的二级章节分类,用于检索范围限定") class DocumentContext(BaseModel): before: str = Field(default="", description="前文上下文片段") after: str = Field(default="", description="后文上下文片段") siblings: List[Dict[str, Any]] = Field(default_factory=list) references: List[Dict[str, Any]] = Field(default_factory=list) retrieval_filters: Dict[str, Any] = Field(default_factory=dict) class DocumentChatRequest(BaseModel): user_id: str message: str = Field(..., min_length=1, description="用户消息内容") selected_section: Optional[SelectedSection] = Field(default=None, description="选中的章节;为空时表示通用问题") conversation_id: Optional[str] = None task_id: Optional[str] = None project_info: Dict[str, Any] = Field(default_factory=dict) document_context: DocumentContext = Field(default_factory=DocumentContext) conversation_history: List[Dict[str, Any]] = Field(default_factory=list) response_mode: Literal["json", "sse"] = "json" class Config: extra = "forbid" class IntentResult(BaseModel): intent: Literal["document_modify", "document_answer", "clarify", "unsupported"] confidence: float = Field(default=0.0, ge=0.0, le=1.0) skill_name: Optional[str] = None operation: str = "" target_scope: str = "selected_section" normalized_instruction: str = "" needs_clarification: bool = False clarification_question: str = "" reason: str = "" warnings: List[str] = Field(default_factory=list) class DocumentChatSkillInput(BaseModel): user_id: str user_message: str selected_section: Optional[SelectedSection] = None intent_result: IntentResult conversation_id: Optional[str] = None task_id: Optional[str] = None project_info: Dict[str, Any] = Field(default_factory=dict) document_context: DocumentContext = Field(default_factory=DocumentContext) conversation_history: List[Dict[str, Any]] = Field(default_factory=list) class DocumentChatSkillOutput(BaseModel): skill_name: str response_type: Literal["answer", "proposal", "clarify", "unsupported", "general_answer", "error"] answer: Optional[str] = None old_content: Optional[str] = None proposed_content: Optional[str] = None change_summary: List[str] = Field(default_factory=list) references: List[Dict[str, Any]] = Field(default_factory=list) warnings: List[str] = Field(default_factory=list) class DiffItem(BaseModel): type: Literal["equal", "insert", "delete", "replace", "full_content"] old_text: str = "" new_text: str = "" class DiffResult(BaseModel): old_content_hash: str new_content_hash: str diff: List[DiffItem] = Field(default_factory=list) diff_granularity: Literal["line", "full_content"] = "line" class DocumentChatData(BaseModel): callback_task_id: str response_type: Literal["answer", "proposal", "clarify", "unsupported", "general_answer", "error"] intent_result: Optional[Dict[str, Any]] = None answer: Optional[str] = None proposed_content: Optional[str] = None old_content_hash: Optional[str] = None new_content_hash: Optional[str] = None diff: List[Dict[str, Any]] = Field(default_factory=list) diff_granularity: Optional[str] = None change_summary: List[str] = Field(default_factory=list) references: List[Dict[str, Any]] = Field(default_factory=list) retrieval_status: Optional[str] = None retrieval_metrics: Dict[str, Any] = Field(default_factory=dict) warnings: List[str] = Field(default_factory=list) selected_section: Dict[str, Any] = Field(default_factory=dict) error_message: Optional[str] = None class DocumentChatResponse(BaseModel): code: int message: str data: Optional[DocumentChatData] = None def model_to_dict(value: Any) -> Dict[str, Any]: """将 Pydantic v1/v2 模型转为字典。""" if value is None: return {} if isinstance(value, dict): return value if hasattr(value, "model_dump"): return value.model_dump() if hasattr(value, "dict"): return value.dict() return dict(value)