TradeMate Dev
137 lines
5.6 KiB
Python
137 lines
5.6 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):
|
|
ai_router = get_ai_router()
|
|
self.ai = ai_router
|
|
self._ai_available = len(ai_router.providers) > 0
|
|
|
|
async def generate(
|
|
self,
|
|
product_info: Dict[str, Any],
|
|
target: str,
|
|
style: str = "professional",
|
|
language: str = "en",
|
|
count: int = 3,
|
|
preference_context: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
name = product_info.get("name", "")
|
|
desc = product_info.get("description", "")
|
|
|
|
if not self._ai_available:
|
|
return self._template_fallback(name, desc, target, style, count, language)
|
|
|
|
results = []
|
|
styles = self._get_style_variants(style, count)
|
|
|
|
for s in styles:
|
|
try:
|
|
result = await self.ai.marketing(product_info, target, s, language, preference_context)
|
|
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
|
|
|
|
def _template_fallback(self, name: str, desc: str, target: str, style: str, count: int, language: str) -> List[Dict[str, Any]]:
|
|
styles = self._get_style_variants(style, count)
|
|
results = []
|
|
for s in styles:
|
|
if language == "zh":
|
|
results.append({
|
|
"content": f"【{name}】产品介绍\n\n{desc}\n\n我们诚挚向您推荐{name}，产品品质优良，价格具有竞争力，欢迎联系我们获取更多信息。",
|
|
"style": s,
|
|
"provider": "template",
|
|
})
|
|
else:
|
|
results.append({
|
|
"content": f"Subject: Introduction of {name}\n\nDear Customer,\n\nWe are pleased to introduce our {name}. {desc}\n\n{self._get_closing(s)}",
|
|
"style": s,
|
|
"provider": "template",
|
|
})
|
|
return results
|
|
|
|
def _get_closing(self, style: str) -> str:
|
|
closings = {
|
|
"professional": "Looking forward to your favorable reply. Best regards.",
|
|
"friendly": "Hope to hear from you soon! Warm regards.",
|
|
"urgent": "Please contact us at your earliest convenience. Best regards.",
|
|
"benefit_focused": "Don't miss this opportunity to boost your business. Contact us today!",
|
|
"storytelling": "Let us help you tell your brand story. Get in touch!",
|
|
}
|
|
return closings.get(style, closings["professional"])
|
|
|
|
async def generate_keywords(
|
|
self, product_info: Dict[str, Any], language: str = "en", count: int = 10
|
|
) -> List[str]:
|
|
name = product_info.get("name", "")
|
|
desc = product_info.get("description", "")
|
|
|
|
if not self._ai_available:
|
|
words = name.split() + [w for w in desc.split() if len(w) > 4]
|
|
return list(dict.fromkeys(words))[:count]
|
|
|
|
try:
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"keywords": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
}
|
|
},
|
|
}
|
|
text = f"Product: {name}. {desc}"
|
|
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]:
|
|
if not self._ai_available:
|
|
return {
|
|
"price_range": "Contact us for pricing",
|
|
"key_selling_points": [product_info.get("name", "")],
|
|
"common_keywords": [],
|
|
"market_trends": "AI analysis unavailable. Please configure an AI provider in settings.",
|
|
"suggestions": ["Set up an AI provider for competitor insights"],
|
|
}
|
|
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 {}
|