邹方成
2322dbad00
refactor(valuation): 重构估值计算参数提取逻辑 fix: 修复行业均值S2计算中的除零错误 feat(api): 新增微信指数计算工具和行业数据查询工具 feat(schema): 在估值模型中添加计算结果字段 refactor: 优化动态质押率计算中的月交易额解析 fix: 处理日期解析异常时返回默认值 docs: 更新API文档中的估值计算请求示例
330 lines
13 KiB
Python
330 lines
13 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": {
|
||
"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", "YOUR_API_KEY"),
|
||
"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": ["token", "videoId"],
|
||
"optional_params": []
|
||
},
|
||
"kuaishou_video_detail": {
|
||
"path": "/api/kuaishou/get-video-detail/v2",
|
||
"method": "GET",
|
||
"description": "该接口用于获取指定快手视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
|
||
"required_params": ["token", "videoId"],
|
||
"optional_params": []
|
||
},
|
||
"bilibili_video_detail": {
|
||
"path": "/api/bilibili/get-video-detail/v2",
|
||
"method": "GET",
|
||
"description": "该接口用于获取指定B站视频的详细信息，包括视频地址、描述文案、作者信息、发布时间、播放量、点赞数、评论数与分享数等",
|
||
"required_params": ["token", "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()
|