LongjoeAgent/tools/base_tool.py

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