simonandClaude Opus 4.7
2f1d8b4d03
- 新增 djapi/api/stock/data_source.py 统一数源入口 (Tushare 单例) - 迁移 10 个模块至统一数据源入口 - 废弃 getDivData_AK.py - 修复 getStockDiv2.py / smoothBrush.py 等模块 - indexDatas API 参数 tscode 类型修正 (股票→指数代码) - views.py + urls.py 接口清理 - continuation.md 状态更新 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
271 lines
10 KiB
Python
271 lines
10 KiB
Python
import env # 加载 .env 到环境变量
|
||
import requests
|
||
import json
|
||
import time
|
||
import logging
|
||
from typing import Optional, Dict, Any
|
||
import os
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
logger = logging.getLogger(__name__)
|
||
|
||
class DeepSeekAPI:
|
||
def __init__(self, api_key: Optional[str] = None):
|
||
"""
|
||
初始化DeepSeek API客户端
|
||
|
||
Args:
|
||
api_key: DeepSeek API密钥，如果为None则从环境变量获取
|
||
"""
|
||
self.api_key = api_key or os.getenv('DEEPSEEK_API_KEY')
|
||
if not self.api_key:
|
||
logger.warning("API密钥未提供且环境变量DEEPSEEK_API_KEY未设置")
|
||
|
||
self.api_url = "https://api.deepseek.com/v1/chat/completions"
|
||
self.max_retries = 3
|
||
self.retry_delay = 2 # 秒
|
||
|
||
# 默认系统提示词
|
||
self.default_system_prompt = """你是一个专业的AI助手，能够准确理解用户需求并提供高质量的回答。
|
||
请根据用户的输入进行适当的处理和分析，保持回答的专业性和准确性。注意：所处理文字来自中央电视台新闻联播节目转文字，请在内容审查时重点考虑。"""
|
||
|
||
def _handle_api_error(self, response: requests.Response) -> str:
|
||
"""
|
||
处理API错误响应
|
||
|
||
Args:
|
||
response: API响应对象
|
||
|
||
Returns:
|
||
错误描述信息
|
||
"""
|
||
error_msg = f"API请求失败: {response.status_code} {response.reason}"
|
||
|
||
try:
|
||
error_data = response.json()
|
||
if 'error' in error_data:
|
||
error_msg += f" - {error_data['error'].get('message', '未知错误')}"
|
||
logger.error(f"API错误详情: {error_data}")
|
||
except json.JSONDecodeError:
|
||
error_msg += f" - 响应内容: {response.text[:200]}"
|
||
|
||
return error_msg
|
||
|
||
def _make_api_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
发送API请求并处理响应
|
||
|
||
Args:
|
||
payload: 请求数据
|
||
|
||
Returns:
|
||
API响应数据
|
||
|
||
Raises:
|
||
Exception: 当所有重试都失败时抛出异常
|
||
"""
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {self.api_key}"
|
||
}
|
||
|
||
last_exception = None
|
||
|
||
for attempt in range(self.max_retries):
|
||
try:
|
||
logger.info(f"发送API请求 (尝试 {attempt + 1}/{self.max_retries})")
|
||
|
||
response = requests.post(
|
||
self.api_url,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=60 # 60秒超时
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
return response.json()
|
||
elif response.status_code == 400:
|
||
# 400错误通常是请求格式问题，不需要重试
|
||
error_msg = self._handle_api_error(response)
|
||
raise Exception(f"请求参数错误: {error_msg}")
|
||
elif response.status_code == 401:
|
||
# 401未授权错误，不需要重试
|
||
raise Exception("API密钥无效或未授权，请检查您的API密钥")
|
||
elif response.status_code == 429:
|
||
# 速率限制，需要重试
|
||
logger.warning("达到速率限制，等待后重试...")
|
||
time.sleep(self.retry_delay * (attempt + 1))
|
||
continue
|
||
elif 500 <= response.status_code < 600:
|
||
# 服务器错误，需要重试
|
||
logger.warning(f"服务器错误 {response.status_code}，等待后重试...")
|
||
time.sleep(self.retry_delay * (attempt + 1))
|
||
continue
|
||
else:
|
||
error_msg = self._handle_api_error(response)
|
||
raise Exception(f"API请求失败: {error_msg}")
|
||
|
||
except requests.exceptions.Timeout:
|
||
last_exception = Exception(f"请求超时 (尝试 {attempt + 1})")
|
||
logger.warning(f"请求超时，等待后重试...")
|
||
time.sleep(self.retry_delay * (attempt + 1))
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
last_exception = Exception(f"网络连接错误 (尝试 {attempt + 1})")
|
||
logger.warning(f"网络连接错误，等待后重试...")
|
||
time.sleep(self.retry_delay * (attempt + 1))
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
last_exception = Exception(f"请求异常: {str(e)}")
|
||
logger.warning(f"请求异常，等待后重试...")
|
||
time.sleep(self.retry_delay * (attempt + 1))
|
||
|
||
# 所有重试都失败
|
||
if last_exception:
|
||
raise last_exception
|
||
else:
|
||
raise Exception("API请求失败，未知错误")
|
||
|
||
def process_text(self,
|
||
prompt: str,
|
||
text: str,
|
||
system_prompt: Optional[str] = None,
|
||
model: str = "deepseek-chat",
|
||
temperature: float = 0.7,
|
||
max_tokens: int = 2000,
|
||
response_format: Optional[Dict] = None) -> str:
|
||
"""
|
||
处理文本的通用方法
|
||
|
||
Args:
|
||
prompt: 用户提示词
|
||
text: 需要处理的文本（约1万字符）
|
||
system_prompt: 系统提示词，如果为None则使用默认值
|
||
model: 使用的模型
|
||
temperature: 生成温度
|
||
max_tokens: 最大生成token数
|
||
|
||
Returns:
|
||
处理后的文本
|
||
|
||
Raises:
|
||
Exception: 当处理失败时抛出包含详细信息的异常
|
||
"""
|
||
# 输入验证
|
||
if not self.api_key:
|
||
raise Exception("API密钥未设置，请提供api_key或设置DEEPSEEK_API_KEY环境变量")
|
||
|
||
if not prompt or not text:
|
||
raise Exception("prompt和text不能为空")
|
||
|
||
# 检查文本长度（约1万字符）
|
||
if len(text) > 15000: # 留一些余量
|
||
logger.warning(f"输入文本长度({len(text)}字符)较长，可能会超过上下文限制")
|
||
|
||
# 准备系统提示词
|
||
system_content = system_prompt or self.default_system_prompt
|
||
|
||
# 构建消息
|
||
messages = [
|
||
{"role": "system", "content": system_content},
|
||
{"role": "user", "content": f"{prompt}\n\n文本内容：\n{text}"}
|
||
]
|
||
|
||
# 构建请求数据
|
||
payload = {
|
||
"model": model,
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
"max_tokens": max_tokens,
|
||
"stream": False
|
||
}
|
||
if response_format:
|
||
payload["response_format"] = response_format
|
||
|
||
try:
|
||
# 发送API请求
|
||
response_data = self._make_api_request(payload)
|
||
|
||
# 解析响应
|
||
if 'choices' in response_data and len(response_data['choices']) > 0:
|
||
result = response_data['choices'][0]['message']['content']
|
||
logger.info("文本处理成功完成")
|
||
return result.strip()
|
||
else:
|
||
raise Exception("API响应格式异常，未找到有效结果")
|
||
|
||
except Exception as e:
|
||
logger.error(f"文本处理失败: {str(e)}")
|
||
raise Exception(f"文本处理失败: {str(e)}")
|
||
|
||
def process_text_with_fallback(self,
|
||
prompt: str,
|
||
text: str,
|
||
system_prompt: Optional[str] = None,
|
||
**kwargs) -> str:
|
||
"""
|
||
带降级处理的文本处理方法
|
||
|
||
Args:
|
||
prompt: 用户提示词
|
||
text: 需要处理的文本
|
||
system_prompt: 系统提示词
|
||
**kwargs: 其他参数
|
||
|
||
Returns:
|
||
处理后的文本，如果API调用失败则返回降级结果
|
||
"""
|
||
try:
|
||
return self.process_text(prompt, text, system_prompt, **kwargs)
|
||
except Exception as e:
|
||
logger.error(f"API调用失败，使用降级处理: {str(e)}")
|
||
# 这里可以添加降级逻辑，比如返回原始文本或简单处理
|
||
return f"处理失败，返回原始文本（错误: {str(e)}）\n\n{text}"
|
||
|
||
# 使用示例
|
||
def deepseek_text(text, prompt):
|
||
# 初始化API客户端
|
||
# 方式1: 直接传入API密钥
|
||
# api_client = DeepSeekAPI(api_key="your_deepseek_api_key_here")
|
||
|
||
# 方式2: 从环境变量读取（推荐）
|
||
api_client = DeepSeekAPI() # API key 从环境变量 DEEPSEEK_API_KEY 读取
|
||
|
||
# 示例文本（约1万字符）
|
||
# sample_text = "这里是你的长文本内容..." * 500 # 模拟长文本
|
||
|
||
# 自定义系统提示词（可选）
|
||
custom_system_prompt = "你是一个专业的文本分析助手，擅长根据提示词对长文本进行深入分析。"
|
||
|
||
try:
|
||
# 处理文本（使用 response_format 强制返回 JSON）
|
||
result = api_client.process_text(
|
||
model=os.getenv('DEEPSEEK_MODEL', 'deepseek-chat'),
|
||
prompt=prompt,
|
||
text=text,
|
||
system_prompt=custom_system_prompt,
|
||
temperature=0.5,
|
||
max_tokens=20000,
|
||
response_format={"type": "json_object"}
|
||
)
|
||
|
||
#print("处理结果:")
|
||
#print(result)
|
||
return result
|
||
|
||
except Exception as e:
|
||
print(f"处理失败: {e}")
|
||
|
||
# 使用降级方法
|
||
fallback_result = api_client.process_text_with_fallback(
|
||
prompt=prompt,
|
||
text=text,
|
||
system_prompt=custom_system_prompt
|
||
)
|
||
#print("降级处理结果:")
|
||
#print(fallback_result)
|
||
return result
|
||
|
||
if __name__ == "__main__":
|
||
deepseek_text() |