api_extract.py

import requests
import json
import os
import time
from pathlib import Path

class LLMClient:
    """
    LLM API客户端类，封装了多种API提供商的调用功能
    """
    # 预定义的API提供商配置
    API_PROVIDERS = {
        "siliconflow": {
            "name": "硅基流动",
            "url": "https://api.siliconflow.cn/v1/chat/completions",
            "header_format": "Bearer {api_key}",
            "default_model": "tencent/Hunyuan-MT-7B"
        },
        "gemini": {
            "name": "Google Gemini",
            "url": "https://generativelanguage.googleapis.com/v1/models/gemini-pro:generateContent",
            "header_format": "Bearer {api_key}",
            "default_model": "gemini-2.5-flash"
        },
        "anthropic": {
            "name": "Anthropic",
            "url": "https://api.anthropic.com/v1/messages",
            "header_format": "Bearer {api_key}",
            "default_model": "claude-3-opus-20240229"
        }
    }

    def __init__(self):
        """
        初始化LLM客户端
        """
        self.provider = None
        self.api_key = None
        self.model = None
        self.custom_url = None
        self.custom_headers = None
        self.custom_payload = None
        self.conversation_history = []

    def call_api(self, prompt):
        """
        调用API并处理响应
        
        Args:
            prompt: 用户提问
        
        Returns:
            结构化的响应结果
        """
        try:
            # 根据provider准备请求参数
            if self.provider == "custom":
                if not self.custom_url:
                    print("错误: 自定义API必须提供URL")
                    return None
                
                url = self.custom_url
                
                # 构建请求头
                headers = self.custom_headers.copy() if self.custom_headers else {}
                if not headers.get("Content-Type"):
                    headers["Content-Type"] = "application/json"
                
                # 如果自定义请求体存在则使用，否则构建标准格式
                if self.custom_payload:
                    data = self.custom_payload
                else:
                    # 默认构建兼容大多数LLM API的请求体
                    data = {
                        "model": self.model or "default-model",
                        "messages": [
                            {"role": "user", "content": prompt}
                        ],
                        "temperature": 0.7,
                        "stream": False
                    }
            else:
                # 使用预定义的提供商配置
                if self.provider not in self.API_PROVIDERS:
                    print(f"错误: 未知的API提供商: {self.provider}")
                    return None
                
                provider_config = self.API_PROVIDERS[self.provider]
                url = provider_config["url"]
                
                # 构建请求头
                headers = {
                    "Content-Type": "application/json",
                    "Authorization": provider_config["header_format"].format(api_key=self.api_key)
                }
                
                # 特殊处理不同API的请求格式
                if self.provider == "anthropic":
                    # Anthropic API使用不同的请求格式
                    data = {
                        "model": self.model or provider_config["default_model"],
                        "messages": [
                            {"role": "user", "content": prompt}
                        ],
                        "max_tokens": 1000,
                        "temperature": 0.7
                    }
                elif self.provider == "gemini":
                    # Google Gemini API使用不同的请求格式
                    data = {
                        "contents": [
                            {
                                "parts": [
                                    {"text": prompt}
                                ]
                            }
                        ],
                        "generationConfig": {
                            "temperature": 0.7
                        }
                    }
                else:
                    # SiliconFlow格式 (原OpenAI兼容格式)
                    data = {
                        "model": self.model or provider_config["default_model"],
                        "messages": [
                            {"role": "user", "content": prompt}
                        ],
                        "temperature": 0.7,
                        "stream": False
                    }
            
            # 发送请求
            print(f"正在发送{self.provider} API请求到 {url}...")
            response = requests.post(url, headers=headers, data=json.dumps(data))
            
            # 检查响应状态
            response.raise_for_status()
            
            # 获取完整响应内容
            full_response = response.json()
            
            # 详细解析和打印返回信息
            print("\n=== 原始API返回信息 ===")
            print(json.dumps(full_response, indent=2, ensure_ascii=False))
            
            print("\n=== 回答 ===")
            
            # 提取响应内容（适配不同API的响应格式）
            content = ""
            
            # Anthropic格式
            if self.provider == "anthropic" and "content" in full_response:
                for item in full_response["content"]:
                    if item.get("type") == "text":
                        content = item.get("text", "")
                        break
                print(f" {content}")
            
            # Google Gemini格式
            elif self.provider == "gemini" and "candidates" in full_response:
                for candidate in full_response["candidates"]:
                    if "content" in candidate:
                        for part in candidate["content"].get("parts", []):
                            if "text" in part:
                                content = part["text"]
                                print(f"{content}")
                                break
                        if content:
                            break
            
            # SiliconFlow格式 (原OpenAI兼容格式)
            elif "choices" in full_response and len(full_response["choices"]) > 0:
                choice = full_response["choices"][0]
                
                if "message" in choice:
                    message = choice["message"]
                    content = message.get('content', '无内容')
                    print(f"{content}")
                elif "text" in choice:
                    content = choice.get('text', '无内容')
                    print(f"{content}")
            
            # 更新对话历史
            self.conversation_history.append({
                "role": "user",
                "content": prompt
            })
            self.conversation_history.append({
                "role": "assistant",
                "content": content
            })
            
            # 返回结构化的结果
            return {
                "provider": self.provider,
                "url": url,
                "model": self.model or (self.provider in self.API_PROVIDERS and self.API_PROVIDERS[self.provider]["default_model"]),
                "content": content,
                "full_response": full_response
            }
            
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {e}")
            if hasattr(e, 'response') and e.response is not None:
                try:
                    error_info = e.response.json()
                    print(f"错误详情: {json.dumps(error_info, indent=2, ensure_ascii=False)}")
                except:
                    print(f"原始错误响应: {e.response.text}")
            return None
        except json.JSONDecodeError:
            print("解析响应失败，可能不是有效的JSON格式")
            print(f"原始响应: {response.text}")
            return None
        except Exception as e:
            print(f"发生未预期的错误: {e}")
            return None

    def load_api_key_from_config(self, provider="siliconflow"):
        """
        从配置文件加载特定提供商的API密钥
        
        Args:
            provider: API提供商标识符
        
        Returns:
            对应的API密钥，如果不存在则返回None
        """
        # 定义可能的配置文件路径
        paths_to_try = [
            Path(__file__).parent / "config.json",
            Path.home() / ".llm_api_client" / "config.json"
        ]
        
        # 尝试从所有可能的路径加载配置
        for config_path in paths_to_try:
            print(f"尝试从 {config_path} 加载配置")
            
            if config_path.exists():
                try:
                    with open(config_path, "r", encoding="utf-8") as f:
                        config = json.load(f)
                        # 首先尝试获取特定提供商的API密钥
                        if "api_keys" in config and isinstance(config["api_keys"], dict):
                            if provider in config["api_keys"] and config["api_keys"][provider]:
                                print(f"从 {config_path} 成功加载 {provider} 的API密钥")
                                return config["api_keys"][provider]
                        # 兼容旧版本配置格式
                        elif provider == "siliconflow" and "api_key" in config and config["api_key"]:
                            print(f"从 {config_path} 成功加载硅基流动的API密钥")
                            return config["api_key"]
                except json.JSONDecodeError:
                    print(f"配置文件 {config_path} 格式错误")
                except PermissionError:
                    print(f"没有权限读取配置文件: {config_path}")
                except Exception as e:
                    print(f"读取配置文件 {config_path} 时出错: {e}")
        
        print(f"未找到 {provider} 的API密钥配置")
        return None

    def save_api_key_to_config(self, api_key, provider="siliconflow"):
        """
        将特定提供商的API密钥保存到配置文件
        
        Args:
            api_key: 要保存的API密钥
            provider: API提供商标识符
            
        Returns:
            保存成功返回True，失败返回False
        """
        # 初始化新的配置字典
        config = {
            "api_keys": {},
            "note": "请不要分享此文件，包含敏感信息"
        }
        
        # 定义可能的配置文件路径，优先使用用户目录下的路径
        paths_to_try = [
            Path.home() / ".llm_api_client" / "config.json",  # 用户目录下的路径优先
            Path(__file__).parent / "config.json"  # 项目目录下的路径作为备选
        ]
        
        # 尝试所有可能的路径进行保存
        for config_path in paths_to_try:
            print(f"尝试保存配置到 {config_path}")
            
            # 尝试先读取现有配置（如果存在）
            temp_config = config.copy()
            try:
                if config_path.exists():
                    with open(config_path, "r", encoding="utf-8") as f:
                        existing_config = json.load(f)
                        # 保留现有配置信息，但确保api_keys字典存在
                        if isinstance(existing_config, dict):
                            # 合并api_keys（如果存在）
                            if "api_keys" in existing_config and isinstance(existing_config["api_keys"], dict):
                                temp_config["api_keys"] = existing_config["api_keys"].copy()
                            # 兼容旧格式，将api_key转移到api_keys[siliconflow]
                            elif "api_key" in existing_config and existing_config["api_key"]:
                                temp_config["api_keys"]["siliconflow"] = existing_config["api_key"]
                            # 保留其他字段
                            for key, value in existing_config.items():
                                if key not in ["api_key", "api_keys", "note", "created_at", "updated_at"]:
                                    temp_config[key] = value
            except Exception as e:
                print(f"读取现有配置文件 {config_path} 时出错: {e}")
                # 继续尝试写入，不影响保存新配置
            
            # 添加或更新特定提供商的API密钥
            temp_config["api_keys"][provider] = api_key
            
            # 更新时间戳
            temp_config["updated_at"] = str(time.time())
            
            # 如果是新文件，添加创建时间
            if not config_path.exists():
                temp_config["created_at"] = temp_config["updated_at"]
            elif "created_at" not in temp_config:
                # 保留现有创建时间或设置为当前时间
                temp_config["created_at"] = str(time.time())
            
            # 确保目录存在
            try:
                config_path.parent.mkdir(parents=True, exist_ok=True)
            except Exception as e:
                print(f"创建配置目录时出错: {e}")
                continue  # 尝试下一个路径
            
            print(f"准备写入配置到: {config_path}")
            
            # 尝试写入配置文件
            try:
                with open(config_path, "w", encoding="utf-8") as f:
                    json.dump(temp_config, f, indent=2, ensure_ascii=False)
                print(f"{provider} API密钥已保存到 {config_path}")
                print("⚠️  安全提示: 请确保不要将config.json文件提交到版本控制系统!")
                return True
            except PermissionError:
                print(f"权限错误: 无法写入配置文件 {config_path}")
                # 继续尝试下一个路径
                continue
            except Exception as e:
                print(f"写入配置文件时出错: {e}")
                # 继续尝试下一个路径
                continue
        
        # 如果所有路径都失败
        print("所有可能的配置路径都无法保存API密钥")
        print("请检查您的系统权限，或尝试以管理员身份运行程序")
        return False

    def select_provider(self):
        """
        选择API提供商
        """
        print("请选择API提供商:")
        options = list(self.API_PROVIDERS.keys()) + ["custom"]
        for i, provider in enumerate(options, 1):
            if provider == "custom":
                print(f"{i}. 自定义API")
            else:
                provider_info = self.API_PROVIDERS[provider]
                print(f"{i}. {provider_info['name']} (默认模型: {provider_info['default_model']})")
        
        provider_choice = input(f"请输入选项 (1-{len(options)}): ")
        
        # 验证选择
        try:
            idx = int(provider_choice) - 1
            if 0 <= idx < len(options):
                self.provider = options[idx]
            else:
                print("无效选项，使用默认提供商: 硅基流动")
                self.provider = "siliconflow"
        except ValueError:
            print("无效选项，使用默认提供商: 硅基流动")
            self.provider = "siliconflow"

    def configure_custom_api(self):
        """
        配置自定义API
        """
        self.custom_url = input("请输入自定义API地址: ")
        if not self.custom_url:
            print("API地址不能为空")
            return False
        
        # 询问是否需要自定义请求头
        if input("是否需要自定义请求头？(y/n，默认n): ").lower() == "y":
            self.custom_headers = {}
            print("请输入请求头信息（每行一个，格式为'key:value'，空行结束）:")
            while True:
                line = input().strip()
                if not line:
                    break
                if ":" in line:
                    key, value = line.split(":", 1)
                    self.custom_headers[key.strip()] = value.strip()
                else:
                    print(f"无效的请求头格式: {line}")
        
        # 询问是否需要自定义请求体
        if input("是否需要完全自定义请求体？(y/n，默认n): ").lower() == "y":
            print("请输入JSON格式的请求体: ")
            try:
                self.custom_payload = json.loads(input())
            except json.JSONDecodeError:
                print("无效的JSON格式，将使用默认请求体格式")
                self.custom_payload = None
        
        return True

    def setup_api_key(self):
        """
        设置API密钥
        """
        # 获取API密钥（除非是完全自定义且不需要密钥）
        if self.provider == "custom" and self.custom_headers and "Authorization" in self.custom_headers:
            self.api_key = None
            print("已使用自定义请求头中的授权信息")
            return True
        else:
            # 尝试从配置文件加载API密钥
            self.api_key = self.load_api_key_from_config(self.provider)
            
            # 如果没有配置文件或配置文件中没有对应的API密钥，提示用户输入
            if not self.api_key:
                provider_name = self.provider if self.provider == "custom" else self.API_PROVIDERS[self.provider]["name"]
                print(f"未找到{provider_name}的API密钥，需要进行配置")
                self.api_key = input(f"请输入你的{provider_name} API密钥: ")
                
                if not self.api_key:
                    print("API密钥不能为空")
                    return False
                
                # 询问是否保存API密钥
                save_choice = input("是否保存API密钥到配置文件？(y/n): ").lower()
                if save_choice == "y":
                    self.save_api_key_to_config(self.api_key, self.provider)
            else:
                provider_name = self.provider if self.provider == "custom" else self.API_PROVIDERS[self.provider]["name"]
                print(f"已从配置文件加载{provider_name} API密钥")
        
        return True

    def select_model(self):
        """
        选择模型
        """
        if self.provider != "custom" or not self.custom_payload:
            if self.provider != "custom":
                default_model = self.API_PROVIDERS[self.provider]["default_model"]
                print(f"\n默认模型: {default_model}")
                
                # 询问是否使用自定义模型
                if input("是否使用自定义模型？(y/n，默认n): ").lower() == "y":
                    self.model = input("请输入自定义模型名称: ")
            else:
                # 自定义API也可以指定模型
                model_input = input("请输入模型名称（可选，留空跳过）: ")
                if model_input.strip():
                    self.model = model_input

    def start_conversation(self):
        """
        开始对话循环
        """
        while True:
            # 设置提示词
            prompt = input("\n请输入你的问题 (输入'退出'或'q'结束对话): ")
            if prompt.lower() in ['退出', 'q']:
                print("感谢使用，再见！")
                break
            if not prompt:
                prompt = "请简要介绍你自己"
            
            # 调用API
            self.call_api(prompt)
            
            # 询问是否继续对话
            continue_chat = input("\n是否继续对话？(y/n): ").lower()
            if continue_chat != 'y':
                print("感谢使用，再见！")
                break

    def print_usage_tips(self):
        """
        打印使用提示
        """
        print("\n=== 使用提示 ===")
        print("1. 支持的API提供商：" + ", ".join([self.API_PROVIDERS[p]["name"] for p in self.API_PROVIDERS]))
        print("2. 配置文件已设置为隐藏属性（在Windows系统上）")
        print("3. 请务必保护好你的API密钥，避免泄露")
        print("4. 可以在custom模式下灵活配置各种API接口")
        print("5. 对话过程中可以随时输入'退出'或'q'结束对话")

    def run(self):
        """
        运行客户端主流程
        """
        print("=== 多提供商LLM API调用工具 ===\n")
        print("这个脚本支持调用多种大语言模型API并查看详细返回信息")
        
        # 选择提供商
        self.select_provider()
        
        # 如果是自定义API，进行配置
        if self.provider == "custom":
            if not self.configure_custom_api():
                return
        
        # 设置API密钥
        if not self.setup_api_key():
            return
        
        # 选择模型
        self.select_model()
        
        # 开始对话
        self.start_conversation()
        
        # 打印使用提示
        self.print_usage_tips()

# 保持兼容性，保留原有的硅基流动API调用函数
def call_siliconflow_api(api_key, prompt, model="tencent/Hunyuan-MT-7B"):
    """
    调用硅基流动API并详细解析返回结果（兼容旧版本）
    
    Args:
        api_key: 硅基流动API密钥
        prompt: 用户提问
        model: 模型名称，默认为免费的腾讯混元模型
    """
    client = LLMClient()
    client.provider = "siliconflow"
    client.api_key = api_key
    client.model = model
    result = client.call_api(prompt)
    return result.get("full_response") if result else None

# 保持兼容性，提供原有的call_api函数
def call_api(provider, api_key, prompt, model=None, custom_url=None, custom_headers=None, custom_payload=None):
    """
    通用API调用函数，支持多种API提供商（兼容旧版本）
    
    Args:
        provider: API提供商标识符或"custom"表示自定义
        api_key: API密钥
        prompt: 用户提问
        model: 模型名称，如果为None则使用提供商默认模型
        custom_url: 自定义API地址（当provider为"custom"时必需）
        custom_headers: 自定义请求头（当provider为"custom"时可选）
        custom_payload: 自定义请求体（当provider为"custom"时可选）
    """
    client = LLMClient()
    client.provider = provider
    client.api_key = api_key
    client.model = model
    client.custom_url = custom_url
    client.custom_headers = custom_headers
    client.custom_payload = custom_payload
    return client.call_api(prompt)

def main():
    """
    主函数，使用面向对象方式实现
    """
    # 创建并运行LLM客户端
    client = LLMClient()
    client.run()

if __name__ == "__main__":
    main()