File size: 4,842 Bytes
76d412c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """
历史数据平台API客户端
最小化实现，只包含必要的功能
"""
import json
import requests
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime, timedelta
class HistoricalDataPlatformClient:
"""
历史数据平台API客户端
"""
def __init__(
self,
base_url: str = "",
api_key: Optional[str] = None,
timeout: int = 30,
retry_times: int = 3
):
"""
初始化API客户端
参数:
base_url: API基础URL
api_key: API密钥（可选）
timeout: 超时时间（秒）
retry_times: 重试次数
"""
self.base_url = base_url.rstrip('/')
self.api_key = api_key
self.timeout = timeout
self.retry_times = retry_times
# 从配置文件加载（如果base_url为空）
if not self.base_url:
self._load_config()
def _load_config(self):
"""从配置文件加载API配置"""
try:
config_path = Path(__file__).parent.parent / "configs" / "api_config.json"
if config_path.exists():
with open(config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
api_config = config.get('historical_data_platform', {})
self.base_url = api_config.get('base_url', '')
self.api_key = api_config.get('api_key') or self.api_key
self.timeout = api_config.get('timeout', 30)
self.retry_times = api_config.get('retry_times', 3)
except Exception as e:
print(f"⚠️ 加载API配置失败: {e}")
def _request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
data: Optional[Dict] = None
) -> Optional[Dict]:
"""发送HTTP请求"""
if not self.base_url:
print("⚠️ API base_url未配置")
return None
url = f"{self.base_url}{endpoint}"
headers = {'Content-Type': 'application/json'}
if self.api_key:
headers['Authorization'] = f'Bearer {self.api_key}'
for attempt in range(self.retry_times):
try:
if method.upper() == 'GET':
response = requests.get(url, params=params, headers=headers, timeout=self.timeout)
else:
response = requests.post(url, json=data, params=params, headers=headers, timeout=self.timeout)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == self.retry_times - 1:
print(f"⚠️ API请求失败: {e}")
return None
continue
return None
def get_raw_data(
self,
device_id: str,
days: int = 7,
start_date: Optional[str] = None,
end_date: Optional[str] = None
) -> Optional[Dict]:
"""
获取用户原始数据
参数:
device_id: 用户ID
days: 过去N天（如果start_date和end_date未提供）
start_date: 开始日期（YYYY-MM-DD格式）
end_date: 结束日期（YYYY-MM-DD格式）
返回:
{
"deviceId": "user123",
"data_points": [...],
"total_count": 100
}
"""
endpoint = f"/api/raw-data/{device_id}"
params = {}
if start_date and end_date:
params['start_date'] = start_date
params['end_date'] = end_date
else:
params['days'] = days
return self._request('GET', endpoint, params=params)
def get_user_profile(self, device_id: str) -> Optional[Dict]:
"""
获取用户个性化信息
返回:
{
"deviceId": "user123",
"age_group": "30-35岁",
"sex": "男性",
...
}
"""
endpoint = f"/api/user-profile/{device_id}"
return self._request('GET', endpoint)
def get_historical_results(
self,
device_id: str,
days: int = 7
) -> Optional[Dict]:
"""
获取历史检测结果
返回:
{
"deviceId": "user123",
"daily_results": [...]
}
"""
endpoint = f"/api/historical-results/{device_id}"
params = {'days': days}
return self._request('GET', endpoint, params=params)
|