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} 功能"