File size: 3,701 Bytes
d4a4da7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
"""
Current Occasions – AI-driven trending context for ad generation.
Uses the LLM to infer relevant occasions from the current date (holidays,
cultural moments, seasonal themes). No fallbacks; results are cached per day.
"""
from datetime import date
from typing import List, Dict, Any
# Daily cache: date.isoformat() -> list of occasion dicts (title, summary, category, relevance_window)
_OCCASIONS_CACHE: Dict[str, List[Dict]] = {}
def _normalize_ai_occasions(raw: Any) -> List[Dict]:
"""Parse and normalize LLM output to our occasion shape."""
if not raw:
return []
items = raw.get("occasions") if isinstance(raw, dict) else raw
if not isinstance(items, list):
return []
normalized: List[Dict] = []
for o in items:
if not isinstance(o, dict):
continue
title = o.get("title") or o.get("name")
summary = o.get("summary") or o.get("description")
if not title or not summary:
continue
normalized.append({
"title": str(title).strip(),
"summary": str(summary).strip(),
"category": str(o.get("category", "Occasion")).strip() or "Occasion",
"relevance_window": str(o.get("relevance_window", "week")).strip() or "week",
})
return normalized
async def get_current_occasions(today: date | None = None) -> List[Dict]:
"""
Return current occasions for ad trending context.
Results are cached per day.
Returns:
List of dicts: title, summary, category, relevance_window (empty if AI fails).
"""
if today is None:
today = date.today()
cache_key = today.isoformat()
if cache_key in _OCCASIONS_CACHE:
return _OCCASIONS_CACHE[cache_key]
try:
from services.llm import llm_service
except Exception:
_OCCASIONS_CACHE[cache_key] = []
return []
system_prompt = """You are an expert in global holidays, cultural moments, and seasonal themes used for advertising and marketing.
Given a date, list 4–6 occasions that are relevant RIGHT NOW or in the next few days for that date. Include:
- Official holidays (US, India, global where relevant)
- Cultural/seasonal moments (e.g. Valentine's Week with daily themes like Rose Day, Teddy Day; Black Friday; back-to-school)
- Awareness days/weeks/months (e.g. Black History Month, Earth Day)
- Shopping or behavior moments (Singles' Day, Prime Day, etc.)
Be specific to the exact date when it matters (e.g. "Teddy Day" on Feb 10, "Valentine's Day" on Feb 14).
Respond with valid JSON only, in this exact shape (no extra fields):
{"occasions": [{"title": "...", "summary": "One sentence on why it matters for ads.", "category": "Occasion", "relevance_window": "day"|"week"|"month"}]}
Use relevance_window: "day" for single-day events, "week" for a week-long moment, "month" for month-long themes."""
user_prompt = f"""Today's date is {today.isoformat()} ({today.strftime('%A, %B %d, %Y')}).
List 4–6 current or upcoming occasions that are relevant for advertising and marketing on or around this date. Consider global and regional relevance. Output JSON only."""
try:
response = await llm_service.generate_json(
prompt=user_prompt,
system_prompt=system_prompt,
temperature=0.3,
)
occasions = _normalize_ai_occasions(response)
except Exception as e:
print(f"⚠️ AI occasions failed: {e}")
occasions = []
_OCCASIONS_CACHE[cache_key] = occasions
return occasions
def clear_occasions_cache() -> None:
"""Clear the occasions cache (e.g. for tests)."""
global _OCCASIONS_CACHE
_OCCASIONS_CACHE = {}
|