TradeMate Dev
23a31f7c00
- Add silent WeChat login for MP/browser environments - Fix Python 3.6 compatibility (remove typing.Annotated usage) - Marketing page: tab-based content generation with category support - Translate page: add auto-detect language default - Homepage: add TTS playback, announcement ticker, remove redundant quick-actions - Fix FAB button overlap with custom tabbar on customers/quotation pages - Make openai/anthropic imports lazy for Python 3.6 compat
94 lines
4.2 KiB
Python
94 lines
4.2 KiB
Python
from typing import Dict, Any, Optional
|
|
import json
|
|
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"):
|
|
try:
|
|
from anthropic import AsyncAnthropic
|
|
except ImportError:
|
|
raise ImportError(
|
|
"anthropic SDK is required for ClaudeProvider. "
|
|
"Install it with: pip install anthropic"
|
|
)
|
|
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", preference_context: Optional[str] = None) -> Dict[str, Any]:
|
|
system = SYSTEM_PROMPTS["reply"]
|
|
if preference_context:
|
|
system += f"\nUser writing preference: {preference_context}"
|
|
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", preference_context: Optional[str] = None) -> Dict[str, Any]:
|
|
system = SYSTEM_PROMPTS["marketing"]
|
|
if preference_context:
|
|
system += f"\nUser preference: {preference_context}"
|
|
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
|