import re import html import requests import json from typing import List, Dict, Any, Optional, Callable from pathlib import Path def html_to_text(html_content: str) -> str: """HTML转文本""" if not html_content: return "" clean = re.compile(r"<[^>]+>") text = clean.sub("", html_content) text = html.unescape(text) return re.sub(r"\s+", " ", text).strip() def get_unique_match_count(search_text: str, filter_words: List[str]) -> int: """获取唯一匹配计数""" sorted_keywords = sorted(filter_words, key=len, reverse=True) match_count = 0 remaining_text = search_text.lower() for keyword in sorted_keywords: kw_lower = keyword.lower() if kw_lower in remaining_text: match_count += 1 remaining_text = remaining_text.replace(kw_lower, "", 1) return match_count def call_csharp_api( backend_url: str, token: str, uoName: str, functionName: str, SParms: dict ) -> str: """调用C# API的通用方法""" print(f"🔧 API调用调试信息:") print(f" - 后端地址: {backend_url}") print(f" - Token: {'已配置' if token else '未配置'}") print(f" - 功能: {functionName}") print(f" - 参数: {SParms}") if not backend_url or not token: error_msg = f"错误:未配置后端地址或认证令牌。后端: {backend_url or '未配置'}, Token: {'已配置' if token else '未配置'}" print(f"❌ {error_msg}") return error_msg headers = { "Accept": "application/json, text/plain, */*", "Content-Type": "application/json", "X-TOKEN": token, "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", } payload = { "token": token, "CList": [ { "uoName": uoName, "functionName": functionName, "SParms": SParms, "ifcommit": True, "returnStrList": [], } ], "language": "zh-cn", } try: print(f"🌐 发送API请求到: {backend_url}") response = requests.post(backend_url, headers=headers, json=payload, timeout=30) print(f"📡 响应状态码: {response.status_code}") if response.status_code == 200: data = response.json() return process_api_response(data) else: error_msg = f"API请求失败,状态码: {response.status_code}" print(f"❌ {error_msg}") return error_msg except Exception as e: error_msg = f"API调用异常: {str(e)}" print(f"❌ {error_msg}") return error_msg def process_api_response(data: Dict[str, Any]) -> str: """处理API响应""" try: inner_json_str = data.get("reJob", {}).get("0", "{}") inner_data = json.loads(inner_json_str) if "err_msg" in inner_data: return f"API返回错误: {inner_data['err_msg']}" if "data" in inner_data: data_list = inner_data["data"] if not data_list: return "NO_DATA" if isinstance(data_list[0], dict): headers = list(data_list[0].keys()) result = [",".join(headers)] for row in data_list: result.append(",".join([str(row.get(h, "")) for h in headers])) return "\n".join(result) return json.dumps(data, ensure_ascii=False) except Exception as e: return f"响应处理错误: {str(e)}" # 工具配置管理函数 def load_tool_config( config_path: Path, get_default_config: Optional[Callable] = None ) -> Dict[str, Any]: """ 加载工具配置的通用函数 Args: config_path: 配置文件路径 get_default_config: 获取默认配置的回调函数,如果不提供则返回空字典 """ if not config_path.exists(): print(f"警告: 配置文件不存在: {config_path}") if get_default_config: return get_default_config() return {} try: with open(config_path, "r", encoding="utf-8") as f: return json.load(f) except json.JSONDecodeError as e: print(f"错误: 配置文件格式不正确: {e}") if get_default_config: return get_default_config() return {} except Exception as e: print(f"错误: 读取配置文件失败: {e}") if get_default_config: return get_default_config() return {} def assemble_tool_description(tool_config: Dict[str, Any]) -> str: """组装工具描述,将所有键值组合成一个完整的字符串""" if not tool_config: return "" description_parts = [] # 基础描述 if "基础描述" in tool_config: description_parts.append(tool_config["基础描述"]) # 功能说明 if "功能说明" in tool_config: description_parts.append(f"\n功能: {tool_config['功能说明']}") # 入参说明 if "入参说明" in tool_config: if isinstance(tool_config["入参说明"], dict): description_parts.append("\n参数:") for param, desc in tool_config["入参说明"].items(): description_parts.append(f" {param}: {desc}") else: description_parts.append(f"\n参数说明: {tool_config['入参说明']}") # 返回值说明 if "返回值说明" in tool_config: if isinstance(tool_config["返回值说明"], dict): description_parts.append("\n返回:") for key, value in tool_config["返回值说明"].items(): if isinstance(value, list): description_parts.append(f" {key}:") for item in value: description_parts.append(f" - {item}") else: description_parts.append(f" {key}: {value}") else: description_parts.append(f"\n返回结果: {tool_config['返回值说明']}") # 输出格式要求 if "输出格式要求" in tool_config: if isinstance(tool_config["输出格式要求"], list): description_parts.append("\n输出要求:") for requirement in tool_config["输出格式要求"]: description_parts.append(f" - {requirement}") else: description_parts.append(f"\n注意: {tool_config['输出格式要求']}") # 使用示例 if "使用示例" in tool_config: description_parts.append(f"\n示例: {tool_config['使用示例']}") return "\n".join(description_parts) def get_tool_prompt( tool_name: str, default_config_func: Optional[Callable] = None ) -> str: """ 获取工具的完整提示词 Args: tool_name: 工具名称 default_config_func: 获取默认配置的函数 """ # 计算配置文件路径 current_file = Path(__file__) config_path = current_file.parent.parent / "config" / "tool_config.json" # 加载配置 config = load_tool_config(config_path, default_config_func) # 获取工具配置 tool_config = config.get(tool_name, {}) # 如果配置为空且提供了默认配置函数,使用默认配置 if not tool_config and default_config_func: default_config = default_config_func() if isinstance(default_config, dict) and tool_name in default_config: tool_config = default_config[tool_name] elif isinstance(default_config, dict) and not default_config: # 如果返回的是整个配置字典 tool_config = default_config else: tool_config = {} # 组装描述 if tool_config: return assemble_tool_description(tool_config) else: return f"执行 {tool_name} 功能"