#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 第三方API配置管理 支持从配置文件或环境变量加载API配置信息 """ import os import json from typing import Dict, Any, Optional from pathlib import Path class APIConfig: """API配置管理器""" def __init__(self, config_file: str = None): """ 初始化配置管理器 Args: config_file: 配置文件路径 """ self.config_file = config_file or os.path.join(os.path.dirname(__file__), 'api_config.json') self.config = self._load_config() def _load_config(self) -> Dict[str, Any]: """加载配置""" # 默认配置 default_config = { "chinaz": { "base_url": "https://openapi.chinaz.net", "endpoints": { "copyright_software": { "APIKey": os.getenv("CHINAZ_COPYRIGHT_API_KEY", "YOUR_API_KEY"), "path": "/v1/1036/copyrightsoftware", "method": "POST", "description": "企业软件著作权查询", "required_params": ["entName", "pageNo", "range", "APIKey", "ChinazVer"], "optional_params": ["sign"] }, "patent": { "APIKey": os.getenv("CHINAZ_PATENT_API_KEY", "apiuser_quantity_045e9c832d253a1fdfd41edd1a85a254_4e9f3a38e384414e97fdbd33cca13124"), "path": "/v1/1036/patent", "method": "POST", "description": "企业专利信息查询", "required_params": ["searchKey", "pageNo", "range", "APIKey", "ChinazVer"], "optional_params": ["sign", "searchType"] }, "judgement": { "APIKey": os.getenv("CHINAZ_JUDGEMENT_API_KEY", "apiuser_quantity_11970d0811cda56260af2e16f1803a8f_b3966d7a0c6e46f2a57fea19f56945db"), "path": "/v1/1036/judgementdetailv4", "method": "POST", "description": "司法综合数据查询", "required_params": ["APIKey", "ChinazVer"], "optional_params": ["sign", "q", "idCardNo", "datatype", "id", "pageNo"] }, "recognition_ocr": { "APIKey": os.getenv("CHINAZ_OCR_API_KEY", "apiuser_quantity_d0848e65f7b7ae50ff6f1ae37e413586_31466987e2ef402ba3142b72680a92fc"), "path": "/v1/1024/recognition_ocr", "method": "POST", "description": "图片OCR识别", "required_params": ["url", "APIKey", "ChinazVer"], "optional_params": ["sign"] }, } }, "justoneapi": { "api_key": os.getenv("JUSTONEAPI_TOKEN", "YNSbIjdU"), "base_url": "https://api.justoneapi.com", "timeout": 30, "retries": 3, "endpoints": { "xiaohongshu_note_detail": { "path": "/api/xiaohongshu/get-note-detail/v7", "method": "GET", "description": "小红书笔记详情查询", "required_params": ["noteId"], "optional_params": ["token"] }, "xiaohongshu_user_info": { "path": "/api/xiaohongshu/get-user-info/v7", "method": "GET", "description": "小红书用户信息查询", "required_params": ["userId"], "optional_params": ["token"] }, "xiaohongshu_search_notes": { "path": "/api/xiaohongshu/search-notes/v7", "method": "GET", "description": "小红书笔记搜索", "required_params": ["keyword"], "optional_params": ["page", "size", "token"] }, "weibo_user_detail": { "path": "/api/weibo/get-user-detail/v1", "method": "GET", "description": "获取用户信息 包括昵称、头像、用户ID、粉丝数、关注数、简介、认证状态等公开信息", "required_params": ["token", "uid"], "optional_params": [] }, "weixin_user_detail": { "path": "/api/weixin/get-user-post/v1", "method": "GET", "description": "公众号发布的文章内容，包括标题、作者、发布时间、内容摘要以及阅读数、点赞数与转发数等互动数据", "required_params": ["token", "wxid"], "optional_params": [] }, "douyin_video_detail": { "path": "/api/douyin/get-video-detail/v2", "method": "GET", "description": "该接口用于获取指定抖音视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等", "required_params": ["videoId"], "optional_params": [] }, "kuaishou_video_detail": { "path": "/api/kuaishou/get-video-detail/v2", "method": "GET", "description": "该接口用于获取指定快手视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等", "required_params": ["videoId"], "optional_params": [] }, "bilibili_video_detail": { "path": "/api/bilibili/get-video-detail/v2", "method": "GET", "description": "该接口用于获取指定B站视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等", "required_params": [ "bvid"], "optional_params": [] } } }, # 微信指数 "dajiala": { "base_url": "https://www.dajiala.com", "timeout": 30, "retries": 3, "endpoints": { "web_search": { "path": "/fbmain/monitor/v3/web_search", "method": "POST", "description": "极致聊指数搜索", "api_key": os.getenv("JIZHILIAO_API_KEY", "JZL089ef0b7d0315d96"), "verifycode": os.getenv("JIZHILIAO_VERIFYCODE", ""), "required_params": ["keyword", "mode", "BusinessType", "sub_search_type"], "optional_params": ["key", "verifycode"] } } }, # dify "dify": { "base_url": "http://106.14.211.94", "api_key": "app-FJXEWdKv63oq1F4rHb4I8kvE", "timeout": 30, "retries": 3, "endpoints": { "workflows_run": { "path": "/v1/workflows/run", "method": "POST", "description": "运行工作流", "required_params": ["inputs", "response_mode", "user"], "optional_params": [], "headers": { "Authorization": "Bearer {api_key}", "Content-Type": "application/json" } } } } } # 尝试从文件加载配置 if os.path.exists(self.config_file): try: with open(self.config_file, 'r', encoding='utf-8') as f: file_config = json.load(f) # 合并配置 self._merge_config(default_config, file_config) except Exception as e: print(f"加载配置文件失败: {e}") return default_config def _merge_config(self, default: Dict[str, Any], custom: Dict[str, Any]) -> None: """递归合并配置""" for key, value in custom.items(): if key in default and isinstance(default[key], dict) and isinstance(value, dict): self._merge_config(default[key], value) else: default[key] = value def get_api_config(self, provider: str) -> Optional[Dict[str, Any]]: """ 获取指定提供商的API配置 Args: provider: API提供商名称 Returns: Optional[Dict[str, Any]]: API配置信息 """ return self.config.get(provider) def get_endpoint_config(self, provider: str, endpoint: str) -> Optional[Dict[str, Any]]: """ 获取指定端点的配置 Args: provider: API提供商名称 endpoint: 端点名称 Returns: Optional[Dict[str, Any]]: 端点配置信息 """ api_config = self.get_api_config(provider) if api_config and 'endpoints' in api_config: return api_config['endpoints'].get(endpoint) return None def get_api_key(self, provider: str) -> Optional[str]: """ 获取API密钥 Args: provider: API提供商名称 Returns: Optional[str]: API密钥 """ api_config = self.get_api_config(provider) return api_config.get('api_key') if api_config else None def get_base_url(self, provider: str) -> Optional[str]: """ 获取基础URL Args: provider: API提供商名称 Returns: Optional[str]: 基础URL """ api_config = self.get_api_config(provider) return api_config.get('base_url') if api_config else None def set_api_key(self, provider: str, api_key: str) -> None: """ 设置API密钥 Args: provider: API提供商名称 api_key: API密钥 """ if provider not in self.config: self.config[provider] = {} self.config[provider]['api_key'] = api_key def add_endpoint(self, provider: str, endpoint_name: str, path: str, method: str = 'POST', description: str = '', required_params: list = None, optional_params: list = None) -> None: """ 添加新的API端点 Args: provider: API提供商名称 endpoint_name: 端点名称 path: API路径 method: HTTP方法 description: 描述信息 required_params: 必需参数列表 optional_params: 可选参数列表 """ if provider not in self.config: self.config[provider] = {'endpoints': {}} if 'endpoints' not in self.config[provider]: self.config[provider]['endpoints'] = {} self.config[provider]['endpoints'][endpoint_name] = { 'path': path, 'method': method.upper(), 'description': description, 'required_params': required_params or [], 'optional_params': optional_params or [] } def save_config(self) -> None: """保存配置到文件""" try: # 创建目录（如果不存在） os.makedirs(os.path.dirname(self.config_file), exist_ok=True) with open(self.config_file, 'w', encoding='utf-8') as f: json.dump(self.config, f, indent=2, ensure_ascii=False) print(f"配置已保存到: {self.config_file}") except Exception as e: print(f"保存配置失败: {e}") def get_common_settings(self) -> Dict[str, Any]: """获取通用设置""" return self.config.get('common_settings', {}) def list_providers(self) -> list: """列出所有API提供商""" return [key for key in self.config.keys() if key != 'common_settings'] def list_endpoints(self, provider: str) -> list: """ 列出指定提供商的所有端点 Args: provider: API提供商名称 Returns: list: 端点名称列表 """ api_config = self.get_api_config(provider) if api_config and 'endpoints' in api_config: return list(api_config['endpoints'].keys()) return [] # 创建全局配置实例 api_config = APIConfig()