297 lines
10 KiB
Python
297 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
RapidAPI库存测试脚本
|
||
自动测试所有订阅的API服务，生成可用性报告
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from datetime import datetime
|
||
from typing import Dict, List, Any
|
||
import os
|
||
|
||
class RapidAPITester:
|
||
"""RapidAPI测试器"""
|
||
|
||
def __init__(self):
|
||
"""初始化测试器"""
|
||
# 从环境变量获取API密钥
|
||
self.api_key = os.getenv('RAPIDAPI_KEY')
|
||
if not self.api_key:
|
||
raise ValueError("RAPIDAPI_KEY环境变量未设置")
|
||
|
||
# API配置 - 基于永动机引擎的配置
|
||
self.api_configs = {
|
||
'alpha_vantage': 'alpha-vantage.p.rapidapi.com',
|
||
'yahoo_finance_1': 'yahoo-finance15.p.rapidapi.com',
|
||
'yh_finance_complete': 'yh-finance.p.rapidapi.com',
|
||
'yahoo_finance_api_data': 'yahoo-finance-api1.p.rapidapi.com',
|
||
'yahoo_finance_realtime': 'yahoo-finance-low-latency.p.rapidapi.com',
|
||
'yh_finance': 'yh-finance-complete.p.rapidapi.com',
|
||
'yahoo_finance_basic': 'yahoo-finance127.p.rapidapi.com',
|
||
'seeking_alpha': 'seeking-alpha.p.rapidapi.com',
|
||
'webull': 'webull.p.rapidapi.com',
|
||
'morning_star': 'morningstar1.p.rapidapi.com',
|
||
'tradingview': 'tradingview-ta.p.rapidapi.com',
|
||
'investing_com': 'investing-cryptocurrency-markets.p.rapidapi.com',
|
||
'finance_api': 'real-time-finance-data.p.rapidapi.com',
|
||
'ms_finance': 'ms-finance.p.rapidapi.com',
|
||
'sec_filings': 'sec-filings.p.rapidapi.com',
|
||
'exchangerate_api': 'exchangerate-api.p.rapidapi.com',
|
||
'crypto_news': 'cryptocurrency-news2.p.rapidapi.com'
|
||
}
|
||
|
||
# 测试端点配置
|
||
self.test_endpoints = {
|
||
'alpha_vantage': '/query?function=GLOBAL_QUOTE&symbol=AAPL',
|
||
'yahoo_finance_1': '/api/yahoo/qu/quote/AAPL',
|
||
'yh_finance_complete': '/stock/v2/get-summary?symbol=AAPL',
|
||
'yahoo_finance_api_data': '/v8/finance/chart/AAPL',
|
||
'yahoo_finance_realtime': '/stock/v2/get-summary?symbol=AAPL',
|
||
'yh_finance': '/stock/v2/get-summary?symbol=AAPL',
|
||
'yahoo_finance_basic': '/api/yahoo/qu/quote/AAPL',
|
||
'seeking_alpha': '/symbols/get-profile?symbols=AAPL',
|
||
'webull': '/stock/search?keyword=AAPL',
|
||
'morning_star': '/market/v2/get-movers?performanceId=0P0000OQN8',
|
||
'tradingview': '/get-analysis?symbol=AAPL&screener=america&exchange=NASDAQ',
|
||
'investing_com': '/coins/get-overview',
|
||
'finance_api': '/stock-price?symbol=AAPL',
|
||
'ms_finance': '/stock/v2/get-summary?symbol=AAPL',
|
||
'sec_filings': '/search?query=AAPL',
|
||
'exchangerate_api': '/latest?base=USD',
|
||
'crypto_news': '/v1/cryptonews'
|
||
}
|
||
|
||
self.results = {}
|
||
|
||
def test_api(self, api_name: str) -> Dict[str, Any]:
|
||
"""
|
||
测试单个API
|
||
|
||
Args:
|
||
api_name: API名称
|
||
|
||
Returns:
|
||
测试结果
|
||
"""
|
||
if api_name not in self.api_configs:
|
||
return {
|
||
'success': False,
|
||
'error': 'API not configured',
|
||
'status_code': None,
|
||
'response_time': 0
|
||
}
|
||
|
||
host = self.api_configs[api_name]
|
||
endpoint = self.test_endpoints.get(api_name, '/')
|
||
|
||
headers = {
|
||
'X-RapidAPI-Key': self.api_key,
|
||
'X-RapidAPI-Host': host,
|
||
'Content-Type': 'application/json'
|
||
}
|
||
|
||
url = f"https://{host}{endpoint}"
|
||
|
||
print(f"🧪 测试 {api_name} ({host})")
|
||
print(f" URL: {url}")
|
||
|
||
start_time = time.time()
|
||
|
||
try:
|
||
response = requests.get(url, headers=headers, timeout=10)
|
||
response_time = time.time() - start_time
|
||
|
||
result = {
|
||
'success': response.status_code == 200,
|
||
'status_code': response.status_code,
|
||
'response_time': round(response_time, 2),
|
||
'response_size': len(response.text),
|
||
'error': None if response.status_code == 200 else response.text[:200]
|
||
}
|
||
|
||
if response.status_code == 200:
|
||
print(f" ✅ 成功 - {response_time:.2f}s - {len(response.text)} bytes")
|
||
# 尝试解析JSON
|
||
try:
|
||
data = response.json()
|
||
result['has_data'] = bool(data)
|
||
result['data_keys'] = list(data.keys()) if isinstance(data, dict) else []
|
||
except:
|
||
result['has_data'] = False
|
||
result['data_keys'] = []
|
||
else:
|
||
print(f" ❌ 失败 - HTTP {response.status_code}")
|
||
print(f" 错误: {response.text[:100]}...")
|
||
|
||
return result
|
||
|
||
except requests.exceptions.Timeout:
|
||
print(f" ⏰ 超时")
|
||
return {
|
||
'success': False,
|
||
'error': 'Request timeout',
|
||
'status_code': None,
|
||
'response_time': 10.0
|
||
}
|
||
except requests.exceptions.RequestException as e:
|
||
print(f" ❌ 请求异常: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'error': f'Request error: {str(e)}',
|
||
'status_code': None,
|
||
'response_time': time.time() - start_time
|
||
}
|
||
except Exception as e:
|
||
print(f" ❌ 未知异常: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'error': f'Unexpected error: {str(e)}',
|
||
'status_code': None,
|
||
'response_time': time.time() - start_time
|
||
}
|
||
|
||
def test_all_apis(self) -> Dict[str, Any]:
|
||
"""测试所有API"""
|
||
print("🚀 开始测试所有RapidAPI服务")
|
||
print("=" * 60)
|
||
|
||
for api_name in self.api_configs.keys():
|
||
result = self.test_api(api_name)
|
||
self.results[api_name] = result
|
||
time.sleep(1) # 避免请求过快
|
||
print()
|
||
|
||
return self.results
|
||
|
||
def generate_report(self) -> str:
|
||
"""生成测试报告"""
|
||
if not self.results:
|
||
return "没有测试结果"
|
||
|
||
# 统计
|
||
total_apis = len(self.results)
|
||
successful_apis = len([r for r in self.results.values() if r['success']])
|
||
failed_apis = total_apis - successful_apis
|
||
|
||
# 按状态分类
|
||
success_list = []
|
||
failed_list = []
|
||
|
||
for api_name, result in self.results.items():
|
||
if result['success']:
|
||
success_list.append({
|
||
'name': api_name,
|
||
'host': self.api_configs[api_name],
|
||
'response_time': result['response_time'],
|
||
'data_keys': result.get('data_keys', [])
|
||
})
|
||
else:
|
||
failed_list.append({
|
||
'name': api_name,
|
||
'host': self.api_configs[api_name],
|
||
'error': result['error'],
|
||
'status_code': result['status_code']
|
||
})
|
||
|
||
# 生成报告
|
||
report = f"""# RapidAPI 测试报告
|
||
|
||
## 📊 测试概览
|
||
|
||
- **测试时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||
- **总API数**: {total_apis}
|
||
- **成功数**: {successful_apis} ({successful_apis/total_apis*100:.1f}%)
|
||
- **失败数**: {failed_apis} ({failed_apis/total_apis*100:.1f}%)
|
||
|
||
## ✅ 可用的API ({len(success_list)}个)
|
||
|
||
"""
|
||
|
||
for api in sorted(success_list, key=lambda x: x['response_time']):
|
||
report += f"### {api['name']}\n"
|
||
report += f"- **主机**: `{api['host']}`\n"
|
||
report += f"- **响应时间**: {api['response_time']}s\n"
|
||
if api['data_keys']:
|
||
report += f"- **数据字段**: {', '.join(api['data_keys'][:5])}\n"
|
||
report += "\n"
|
||
|
||
report += f"## ❌ 失败的API ({len(failed_list)}个)\n\n"
|
||
|
||
for api in failed_list:
|
||
report += f"### {api['name']}\n"
|
||
report += f"- **主机**: `{api['host']}`\n"
|
||
report += f"- **状态码**: {api['status_code']}\n"
|
||
report += f"- **错误**: {api['error'][:100] if api['error'] else 'Unknown'}...\n"
|
||
report += "\n"
|
||
|
||
# 建议
|
||
report += """## 🔧 优化建议
|
||
|
||
### 立即可用的API
|
||
"""
|
||
|
||
fast_apis = [api for api in success_list if api['response_time'] < 2.0]
|
||
if fast_apis:
|
||
report += "以下API响应快速，建议优先使用：\n"
|
||
for api in fast_apis:
|
||
report += f"- **{api['name']}**: {api['response_time']}s\n"
|
||
|
||
report += """
|
||
### 需要修复的API
|
||
"""
|
||
|
||
if failed_list:
|
||
report += "以下API需要检查端点配置或权限：\n"
|
||
for api in failed_list[:5]: # 只显示前5个
|
||
report += f"- **{api['name']}**: {api['error'][:50] if api['error'] else 'Unknown error'}...\n"
|
||
|
||
return report
|
||
|
||
def save_report(self, filename: str = None):
|
||
"""保存报告到文件"""
|
||
if not filename:
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
filename = f"docs/rapidapi/test_report_{timestamp}.md"
|
||
|
||
report = self.generate_report()
|
||
|
||
with open(filename, 'w', encoding='utf-8') as f:
|
||
f.write(report)
|
||
|
||
print(f"📄 报告已保存到: {filename}")
|
||
return filename
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("🧪 RapidAPI库存测试工具")
|
||
print("=" * 40)
|
||
|
||
try:
|
||
tester = RapidAPITester()
|
||
|
||
# 测试所有API
|
||
results = tester.test_all_apis()
|
||
|
||
# 生成并显示报告
|
||
print("\n" + "=" * 60)
|
||
print("📊 测试完成，生成报告...")
|
||
|
||
report = tester.generate_report()
|
||
print(report)
|
||
|
||
# 保存报告
|
||
filename = tester.save_report()
|
||
|
||
# 更新库存文档
|
||
print(f"\n💡 建议更新 docs/rapidapi/api_inventory.md")
|
||
print(f"📁 详细报告: {filename}")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
if __name__ == "__main__":
|
||
main() |