TradeMate Dev
c6206787da
项目结构: - backend/ Python FastAPI 后端 - uni-app/ uni-app跨端前端 - docs/ 设计文档 - docker-compose.yml Docker编排 - nginx/scripts/systemd 运维配置 已完成功能: - 用户认证 (JWT) - 智能翻译 + 回复建议 - 营销素材生成 - 客户管理 + 沉默检测 - 报价单管理 - 产品库管理 - 汇率换算 - 推送通知 (uni-push) - WhatsApp Webhook框架 - Celery定时任务
85 lines
3.2 KiB
Python
85 lines
3.2 KiB
Python
from typing import Dict, Any, Optional, List
|
|
from app.ai.router import get_ai_router
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MarketingService:
|
|
def __init__(self):
|
|
self.ai = get_ai_router()
|
|
|
|
async def generate(
|
|
self,
|
|
product_info: Dict[str, Any],
|
|
target: str,
|
|
style: str = "professional",
|
|
language: str = "en",
|
|
count: int = 3,
|
|
) -> List[Dict[str, Any]]:
|
|
results = []
|
|
styles = self._get_style_variants(style, count)
|
|
|
|
for s in styles:
|
|
try:
|
|
result = await self.ai.marketing(product_info, target, s, language)
|
|
results.append({
|
|
"content": result.get("content", ""),
|
|
"style": s,
|
|
"provider": result.get("provider_used", "unknown"),
|
|
})
|
|
except Exception as e:
|
|
logger.warning(f"Marketing generation failed for style '{s}': {e}")
|
|
results.append({"content": "", "style": s, "error": str(e)})
|
|
|
|
return results
|
|
|
|
async def generate_keywords(
|
|
self, product_info: Dict[str, Any], language: str = "en", count: int = 10
|
|
) -> List[str]:
|
|
try:
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"keywords": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
}
|
|
},
|
|
}
|
|
text = f"Product: {product_info.get('name', '')}. {product_info.get('description', '')}"
|
|
result = await self.ai.extract(text, schema)
|
|
keywords = result.get("data", {}).get("keywords", [])
|
|
return keywords[:count]
|
|
except Exception as e:
|
|
logger.warning(f"Keyword generation failed: {e}")
|
|
return []
|
|
|
|
def _get_style_variants(self, base_style: str, count: int) -> List[str]:
|
|
all_styles = ["professional", "friendly", "urgent", "benefit_focused", "storytelling"]
|
|
if base_style in all_styles:
|
|
all_styles.remove(base_style)
|
|
all_styles.insert(0, base_style)
|
|
return all_styles[:count]
|
|
|
|
async def analyze_competitors(
|
|
self, product_info: Dict[str, Any], market: str = "US"
|
|
) -> Dict[str, Any]:
|
|
try:
|
|
text = f"Product: {product_info.get('name', '')} in {market} market. Category: {product_info.get('category', '')}. Description: {product_info.get('description', '')}"
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"price_range": {"type": "string"},
|
|
"key_selling_points": {"type": "array", "items": {"type": "string"}},
|
|
"common_keywords": {"type": "array", "items": {"type": "string"}},
|
|
"market_trends": {"type": "string"},
|
|
"suggestions": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
}
|
|
result = await self.ai.extract(text, schema)
|
|
return result.get("data", {})
|
|
except Exception as e:
|
|
logger.warning(f"Competitor analysis failed: {e}")
|
|
return {}
|