Initial commit: TradeMate 外贸小助手 MVP

项目结构: - backend/ Python FastAPI 后端 - uni-app/ uni-app跨端前端 - docs/ 设计文档 - docker-compose.yml Docker编排 - nginx/scripts/systemd 运维配置已完成功能: - 用户认证 (JWT) - 智能翻译 + 回复建议 - 营销素材生成 - 客户管理 + 沉默检测 - 报价单管理 - 产品库管理 - 汇率换算 - 推送通知 (uni-push) - WhatsApp Webhook框架 - Celery定时任务
2026-05-08 18:17:12 +08:00
commit c6206787da
121 changed files with 11743 additions and 0 deletions
@@ -0,0 +1,83 @@
+from typing import Dict, Any, Optional
+import json
+from anthropic import AsyncAnthropic
+from app.ai.base import AIProvider
+
+
+SYSTEM_PROMPTS = {
+    "marketing": "You are a world-class copywriter for international trade. Write persuasive, "
+                 "culturally-adapted marketing content that converts. You excel at storytelling "
+                 "and emotional appeal in business contexts.",
+    "reply": "You are a senior international sales representative with 20 years of experience. "
+             "Your replies are warm, professional, and strategically move the conversation "
+             "toward closing the deal.",
+    "translate": "You are a professional translator specializing in trade documents. "
+                 "Preserve all numbers, terms, and formatting. Translate meaning, not words.",
+    "extract": "Extract structured data from text. Return ONLY valid JSON.",
+}
+
+
+class ClaudeProvider(AIProvider):
+    def __init__(self, api_key: str, model: str = "claude-sonnet-4-20250514"):
+        self.client = AsyncAnthropic(api_key=api_key)
+        self.model = model
+        self._name = f"claude-sonnet"
+        self._pricing = {"input": 0.003, "output": 0.015}
+
+    async def translate(self, text: str, source_lang: Optional[str], target_lang: str, context: Optional[str] = None) -> Dict[str, Any]:
+        system = SYSTEM_PROMPTS["translate"]
+        if context:
+            system += f"\nContext: {context}"
+        prompt = f"Translate to {target_lang}:\n\n{text}"
+        content = await self._call(system, prompt)
+        return {"translated_text": content, "provider": self.name}
+
+    async def reply(self, inquiry: str, context: Optional[Dict[str, Any]] = None, tone: str = "professional") -> Dict[str, Any]:
+        system = SYSTEM_PROMPTS["reply"]
+        context_str = ""
+        if context:
+            for k, v in context.items():
+                if v:
+                    context_str += f"{k}: {v}\n"
+        prompt = f"{context_str}\nCustomer says:\n{inquiry}\n\nYour reply ({tone} tone):"
+        content = await self._call(system, prompt)
+        return {"reply": content, "provider": self.name}
+
+    async def generate_marketing(self, product_info: Dict[str, Any], target: str, style: str = "professional", language: str = "en") -> Dict[str, Any]:
+        system = SYSTEM_PROMPTS["marketing"]
+        info = json.dumps(product_info, ensure_ascii=False, indent=2)
+        prompt = f"Product:\n{info}\n\nTarget: {target}\nStyle: {style}\nLanguage: {language}\n\nWrite marketing copy:"
+        content = await self._call(system, prompt, max_tokens=1500)
+        return {"content": content, "provider": self.name}
+
+    async def extract_info(self, text: str, schema: Dict[str, Any]) -> Dict[str, Any]:
+        system = SYSTEM_PROMPTS["extract"]
+        prompt = f"Schema:\n{json.dumps(schema, indent=2)}\n\nText:\n{text}\n\nJSON:"
+        content = await self._call(system, prompt, max_tokens=1000)
+        try:
+            data = json.loads(content)
+            return {"data": data, "confidence": 0.9, "provider": self.name}
+        except json.JSONDecodeError:
+            return {"data": {}, "confidence": 0.0, "provider": self.name, "error": "parse_failed"}
+
+    async def _call(self, system: str, prompt: str, max_tokens: int = 1000) -> str:
+        resp = await self.client.messages.create(
+            model=self.model,
+            system=system,
+            messages=[{"role": "user", "content": prompt}],
+            max_tokens=max_tokens,
+            temperature=0.7,
+        )
+        return resp.content[0].text
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def cost_per_1k_tokens(self) -> float:
+        return (self._pricing["input"] + self._pricing["output"]) / 2
+
+    @property
+    def supports_streaming(self) -> bool:
+        return True