259 lines
8.9 KiB
Python
259 lines
8.9 KiB
Python
#!/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": {
|
|
"api_key": os.getenv("CHINAZ_API_KEY", "YOUR_API_KEY"),
|
|
"base_url": "https://openapi.chinaz.net",
|
|
"endpoints": {
|
|
"copyright_software": {
|
|
"path": "/v1/1036/copyrightsoftware",
|
|
"method": "POST",
|
|
"description": "企业软件著作权查询",
|
|
"required_params": ["entName", "pageNo", "range", "APIKey", "ChinazVer"],
|
|
"optional_params": ["sign"]
|
|
},
|
|
"patent": {
|
|
"path": "/v1/1036/patent",
|
|
"method": "POST",
|
|
"description": "企业专利信息查询",
|
|
"required_params": ["searchKey", "pageNo", "range", "APIKey", "ChinazVer"],
|
|
"optional_params": ["sign", "searchType"]
|
|
},
|
|
"judgement": {
|
|
"path": "/v1/1036/judgementdetailv4",
|
|
"method": "POST",
|
|
"description": "司法综合数据查询",
|
|
"required_params": ["APIKey", "ChinazVer"],
|
|
"optional_params": ["sign", "q", "idCardNo", "datatype", "id", "pageNo"]
|
|
}
|
|
}
|
|
},
|
|
"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"]
|
|
}
|
|
}
|
|
},
|
|
"other_apis": {
|
|
# 可以添加其他第三方API配置
|
|
"example_api": {
|
|
"api_key": os.getenv("EXAMPLE_API_KEY", ""),
|
|
"base_url": "https://api.example.com",
|
|
"endpoints": {}
|
|
}
|
|
},
|
|
"common_settings": {
|
|
"timeout": 60,
|
|
"retries": 3,
|
|
"sign_prefix": "634xz"
|
|
}
|
|
}
|
|
|
|
# 尝试从文件加载配置
|
|
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() |