c6206787da
项目结构: - backend/ Python FastAPI 后端 - uni-app/ uni-app跨端前端 - docs/ 设计文档 - docker-compose.yml Docker编排 - nginx/scripts/systemd 运维配置 已完成功能: - 用户认证 (JWT) - 智能翻译 + 回复建议 - 营销素材生成 - 客户管理 + 沉默检测 - 报价单管理 - 产品库管理 - 汇率换算 - 推送通知 (uni-push) - WhatsApp Webhook框架 - Celery定时任务
52 lines
2.0 KiB
Python
52 lines
2.0 KiB
Python
from typing import Dict, Any, Optional
|
|
import httpx
|
|
from app.ai.base import AIProvider
|
|
|
|
|
|
class DeepLProvider(AIProvider):
|
|
def __init__(self, api_key: str, endpoint: str = "https://api.deepl.com/v2"):
|
|
self.api_key = api_key
|
|
self.endpoint = endpoint
|
|
self._name = "deepl"
|
|
self._cost_per_char = 0.000006
|
|
|
|
async def translate(self, text: str, source_lang: Optional[str], target_lang: str, context: Optional[str] = None) -> Dict[str, Any]:
|
|
params = {
|
|
"auth_key": self.api_key,
|
|
"text": text,
|
|
"target_lang": target_lang.upper()[:2],
|
|
}
|
|
if source_lang and source_lang != "auto":
|
|
params["source_lang"] = source_lang.upper()[:2]
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(f"{self.endpoint}/translate", data=params, timeout=15)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
t = data["translations"][0]
|
|
return {
|
|
"translated_text": t["text"],
|
|
"provider": self.name,
|
|
"detected_source_lang": t.get("detected_source_language", source_lang),
|
|
"char_count": len(text),
|
|
"cost": len(text) * self._cost_per_char,
|
|
}
|
|
|
|
async def reply(self, inquiry: str, context: Optional[Dict[str, Any]] = None, tone: str = "professional") -> Dict[str, Any]:
|
|
raise NotImplementedError("DeepL does not support reply generation")
|
|
|
|
async def generate_marketing(self, product_info: Dict[str, Any], target: str, style: str = "professional", language: str = "en") -> Dict[str, Any]:
|
|
raise NotImplementedError("DeepL does not support marketing generation")
|
|
|
|
async def extract_info(self, text: str, schema: Dict[str, Any]) -> Dict[str, Any]:
|
|
raise NotImplementedError("DeepL does not support info extraction")
|
|
|
|
@property
|
|
def name(self) -> str:
|
|
return self._name
|
|
|
|
@property
|
|
def cost_per_1k_tokens(self) -> float:
|
|
return self._cost_per_char * 1000
|