""" 模板AI辅助工具 - 使用AI智能分析文档内容并识别占位符替换位置 """ import os import json import requests from typing import Dict, List, Optional, Tuple from dotenv import load_dotenv # 加载环境变量 load_dotenv() class TemplateAIHelper: """模板AI辅助类，用于智能分析文档内容""" def __init__(self): self.api_key = os.getenv('SILICONFLOW_API_KEY') self.model = os.getenv('SILICONFLOW_MODEL', 'deepseek-ai/DeepSeek-V3.2-Exp') self.api_url = "https://api.siliconflow.cn/v1/chat/completions" if not self.api_key: raise Exception("未配置 SILICONFLOW_API_KEY，请在 .env 文件中设置") def test_api_connection(self) -> bool: """ 测试API连接是否正常 Returns: 是否连接成功 """ try: print(" [测试] 正在测试硅基流动API连接...") test_payload = { "model": self.model, "messages": [ { "role": "user", "content": "测试" } ], "max_tokens": 10 } headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } response = requests.post( self.api_url, json=test_payload, headers=headers, timeout=10 ) if response.status_code == 200: print(" [测试] ✓ API连接正常") return True else: print(f" [测试] ✗ API连接失败: {response.status_code} - {response.text[:200]}") return False except Exception as e: print(f" [测试] ✗ API连接测试失败: {e}") return False def analyze_document_content( self, document_text: str, available_fields: List[Dict], document_type: str = "未知" ) -> List[Dict]: """ 分析文档内容，识别需要替换为占位符的位置 Args: document_text: 文档文本内容 available_fields: 可用字段列表，格式: [{"field_code": "xxx", "field_name": "xxx", "description": "xxx"}] document_type: 文档类型 Returns: 替换建议列表，格式: [ { "original_text": "原始文本", "replacement": "{{field_code}}", "field_code": "field_code", "field_name": "字段名称", "confidence": 0.9, "position": "段落/表格" } ] """ try: print(f" [AI] 正在分析文本内容（长度: {len(document_text)} 字符）...") # 构建字段信息字符串 fields_info = "\n".join([ f"- {field['field_name']} ({{{{field_code}}}}): {field.get('description', '')}" for field in available_fields ]) # 构建提示词 prompt = f"""你是一个专业的文档模板分析助手。请分析以下文档内容，识别所有可以替换为占位符的位置。 文档类型：{document_type} 可用字段列表： {fields_info} 文档内容： {document_text} 请仔细分析文档内容，识别以下类型的可替换内容： 1. 明确的字段值（如姓名、单位、职务等） 2. 示例值（如"XXX"、"张三"、"某公司"等） 3. 组合字段（如"山西XXXX集团有限公司(职务+姓名）"应替换为对应的占位符组合） 4. 日期、时间等格式化的值 5. 任何看起来是示例或占位符的内容 对于组合字段，如果包含多个字段信息，请使用多个占位符的组合，例如： - "山西XXXX集团有限公司(职务+姓名）" → "{{{{target_organization_and_position}}}}({{{{target_name}}}})" - "张三，男，1980年5月" → "{{{{target_name}}}}，{{{{target_gender}}}}，{{{{target_date_of_birth}}}}" 请以JSON格式返回分析结果，格式如下： {{ "replacements": [ {{ "original_text": "原始文本内容", "replacement": "{{{{field_code}}}}或组合占位符", "field_code": "字段编码（如果是组合，用逗号分隔）", "field_name": "字段名称", "confidence": 0.9, "position": "位置描述（如：第X段、表格第X行第X列）", "reason": "替换原因" }} ] }} 只返回JSON，不要其他说明文字。""" # 调用AI API payload = { "model": self.model, "messages": [ { "role": "system", "content": "你是一个专业的文档模板分析助手，能够准确识别文档中需要替换为占位符的内容。请严格按照JSON格式返回结果，不要添加任何解释性文字。" }, { "role": "user", "content": prompt } ], "temperature": 0.2, "max_tokens": 4000 } headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } print(f" [AI] 正在调用API...") response = requests.post( self.api_url, json=payload, headers=headers, timeout=60 ) print(f" [AI] API响应状态: {response.status_code}") if response.status_code != 200: error_msg = response.text[:500] if len(response.text) > 500 else response.text raise Exception(f"API调用失败: {response.status_code} - {error_msg}") result = response.json() print(f" [AI] API调用成功，正在解析响应...") # 提取AI返回的内容 if 'choices' in result and len(result['choices']) > 0: content = result['choices'][0]['message']['content'] # 尝试解析JSON try: # 如果返回的是代码块，提取JSON部分 if '```json' in content: json_start = content.find('```json') + 7 json_end = content.find('```', json_start) content = content[json_start:json_end].strip() elif '```' in content: json_start = content.find('```') + 3 json_end = content.find('```', json_start) content = content[json_start:json_end].strip() parsed_result = json.loads(content) replacements = parsed_result.get('replacements', []) print(f" [AI] ✓ 分析完成，识别到 {len(replacements)} 个替换建议") return replacements except json.JSONDecodeError as e: print(f" [AI] ⚠ JSON解析失败: {e}") print(f" [AI] 原始响应内容: {content[:200]}...") # 如果JSON解析失败，返回空列表 return [] else: raise Exception("API返回格式异常") except requests.exceptions.Timeout: print(f" [AI] ✗ 请求超时（60秒），跳过AI分析") return [] except Exception as e: print(f" [AI] ✗ 分析失败: {e}") import traceback traceback.print_exc() return [] def analyze_paragraph( self, paragraph_text: str, available_fields: List[Dict], document_type: str = "未知" ) -> List[Dict]: """ 分析单个段落，识别需要替换的位置 Args: paragraph_text: 段落文本 available_fields: 可用字段列表 document_type: 文档类型 Returns: 替换建议列表 """ # 跳过空内容或太短的内容（少于10个字符） if not paragraph_text or len(paragraph_text.strip()) < 10: return [] # 如果文本已经包含占位符，跳过 if '{{' in paragraph_text and '}}' in paragraph_text: return [] return self.analyze_document_content(paragraph_text, available_fields, document_type) def analyze_table_cell( self, cell_text: str, available_fields: List[Dict], document_type: str = "未知", row: int = 0, col: int = 0 ) -> List[Dict]: """ 分析表格单元格，识别需要替换的位置 Args: cell_text: 单元格文本 available_fields: 可用字段列表 document_type: 文档类型 row: 行号 col: 列号 Returns: 替换建议列表 """ if not cell_text or len(cell_text.strip()) < 3: return [] replacements = self.analyze_document_content(cell_text, available_fields, document_type) # 添加位置信息 for replacement in replacements: replacement['position'] = f"表格第{row+1}行第{col+1}列" return replacements def get_available_fields_for_document(doc_config: Dict, field_name_to_code: Dict) -> List[Dict]: """ 根据文档配置获取可用字段列表 Args: doc_config: 文档配置 field_name_to_code: 字段名称到编码的映射 Returns: 可用字段列表 """ available_fields = [] for field_code in doc_config.get('fields', []): # 查找字段名称 field_name = None for name, code in field_name_to_code.items(): if code == field_code: field_name = name break if field_name: available_fields.append({ 'field_code': field_code, 'field_name': field_name, 'description': f"{field_name}字段" }) return available_fields