mirror of
https://github.com/xinnan-tech/xiaozhi-esp32-server.git
synced 2026-07-21 22:53:56 +08:00
65 lines
2.7 KiB
Python
65 lines
2.7 KiB
Python
import httpx
|
||
from typing import Dict, Any, List
|
||
from config.logger import setup_logging
|
||
|
||
TAG = __name__
|
||
|
||
class ContextDataProvider:
|
||
"""数据上下文填充，负责从配置的API获取数据"""
|
||
|
||
def __init__(self, config: Dict[str, Any], logger=None):
|
||
self.config = config
|
||
self.logger = logger or setup_logging()
|
||
self.context_data = ""
|
||
|
||
def fetch_all(self, device_id: str) -> str:
|
||
"""获取所有配置的上下文数据"""
|
||
context_providers = self.config.get("context_providers", [])
|
||
if not context_providers:
|
||
return ""
|
||
|
||
formatted_lines = []
|
||
for provider in context_providers:
|
||
url = provider.get("url")
|
||
headers = provider.get("headers", {})
|
||
|
||
if not url:
|
||
continue
|
||
|
||
try:
|
||
headers = headers.copy() if isinstance(headers, dict) else {}
|
||
# 将 device_id 添加到请求头
|
||
headers["device_id"] = device_id
|
||
|
||
# 发送请求
|
||
response = httpx.get(url, headers=headers, timeout=3)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if isinstance(result, dict):
|
||
if result.get("code") == 0:
|
||
data = result.get("data")
|
||
# 格式化数据
|
||
if isinstance(data, dict):
|
||
for k, v in data.items():
|
||
formatted_lines.append(f"- **{k}：** {v}")
|
||
elif isinstance(data, list):
|
||
for item in data:
|
||
formatted_lines.append(f"- {item}")
|
||
else:
|
||
formatted_lines.append(f"- {data}")
|
||
else:
|
||
self.logger.bind(tag=TAG).warning(f"API {url} 返回错误码: {result.get('msg')}")
|
||
else:
|
||
self.logger.bind(tag=TAG).warning(f"API {url} 返回的不是JSON字典")
|
||
else:
|
||
self.logger.bind(tag=TAG).warning(f"API {url} 请求失败: {response.status_code}")
|
||
except Exception as e:
|
||
self.logger.bind(tag=TAG).error(f"获取上下文数据 {url} 失败: {e}")
|
||
|
||
# 将所有格式化后的行拼接成一个字符串
|
||
self.context_data = "\n".join(formatted_lines)
|
||
if self.context_data:
|
||
self.logger.bind(tag=TAG).debug(f"已注入动态上下文数据:\n{self.context_data}")
|
||
return self.context_data
|