from __future__ import annotations import re from datetime import datetime, timedelta, timezone from typing import Any from .models import BrandKit, Campaign, ComplianceCheckRequest, ContentCalendarPlanRequest, Platform, TrendSignal, Variant from .store import new_id, now_iso PLATFORM_RULES: dict[Platform, dict[str, Any]] = { "tiktok": {"duration": 28, "hashtags": ["#tiktok", "#fyp"], "style": "fast_hook"}, "instagram_reels": {"duration": 30, "hashtags": ["#reels", "#explore"], "style": "polished"}, "facebook_shorts": {"duration": 35, "hashtags": ["#shorts", "#facebookreels"], "style": "direct"}, "youtube_shorts": {"duration": 40, "hashtags": ["#shorts", "#youtube"], "style": "searchable"}, } CTA_POOL = [ "Follow for the next part.", "Save this before you forget it.", "Comment your take below.", "Share this with someone who needs it.", ] def _keywords(text: str, limit: int = 6) -> list[str]: words = re.findall(r"[a-zA-Z0-9]+", text.lower()) blocked = {"the", "and", "for", "with", "this", "that", "from", "your", "you", "are", "into", "about"} unique = [] for word in words: if len(word) < 4 or word in blocked or word in unique: continue unique.append(word) if len(unique) >= limit: break return unique or ["video", "creator", "growth"] def generate_hooks(topic: str, niche: str, count: int) -> list[str]: base = topic.strip().rstrip(".") niche_text = f" in {niche}" if niche else "" patterns = [ "Most people miss this about {base}{niche}.", "Here is the fastest way to understand {base}.", "Stop scrolling if you care about {base}.", "This one detail changes how {base} works.", "Before you try {base}, watch this.", "The simple version of {base} nobody explains.", ] return [patterns[index % len(patterns)].format(base=base, niche=niche_text) for index in range(count)] def generate_hashtags(topic: str, niche: str, platform: Platform) -> list[str]: tags = [f"#{word}" for word in _keywords(f"{topic} {niche}", 8)] tags.extend(PLATFORM_RULES[platform]["hashtags"]) deduped = [] for tag in tags: normalized = re.sub(r"[^#a-zA-Z0-9_]", "", tag) if normalized and normalized not in deduped: deduped.append(normalized) return deduped[:12] def build_script(topic: str, hook: str, tone: str, index: int) -> str: return ( f"{hook}\n" f"Point one: define the problem around {topic} in plain language.\n" f"Point two: show the practical mistake or opportunity.\n" f"Point three: give one action the viewer can use today.\n" f"Keep the tone {tone}. This is variant {index + 1}." ) def build_render_payload(campaign: Campaign, variant: Variant, brand: BrandKit | None) -> dict[str, Any]: scenes = [ { "start": 0, "duration": max(3, round(variant.duration_seconds / 3, 2)), "media": campaign.source_asset or campaign.source_url or "upload://source", "caption": variant.hook, "layout": "fill", "background": "blur", "transition": "fade", }, { "start": round(variant.duration_seconds / 3, 2), "duration": max(3, round(variant.duration_seconds / 3, 2)), "media": campaign.source_asset or campaign.source_url or "upload://source", "caption": variant.caption, "layout": "fill", "background": "blur", "transition": "smooth", }, { "start": round((variant.duration_seconds / 3) * 2, 2), "duration": max(3, round(variant.duration_seconds / 3, 2)), "media": campaign.source_asset or campaign.source_url or "upload://source", "caption": variant.cta, "layout": "fill", "background": "blur", "transition": "fade", }, ] payload = { "template": variant.template, "creative_style": variant.creative_style, "platform": variant.platform, "output_name": f"{variant.id}.mp4", "auto_subtitles": True, "subtitle_format": "ass", "normalize": True, "metadata": { "campaign_id": campaign.id, "variant_id": variant.id, "title": variant.title, "hashtags": variant.hashtags, }, "scenes": scenes, } if brand: payload["watermark"] = brand.logo_url payload["metadata"]["brand"] = brand.model_dump() return payload def generate_variants(campaign: Campaign, brand: BrandKit | None = None) -> list[Variant]: hooks = generate_hooks(campaign.topic, campaign.niche, campaign.quantity) created = now_iso() variants = [] for index in range(campaign.quantity): platform = campaign.platforms[index % len(campaign.platforms)] rules = PLATFORM_RULES[platform] title = f"{campaign.topic}: Part {index + 1}" hook = hooks[index] script = build_script(campaign.topic, hook, campaign.tone, index) variant = Variant( id=new_id("var"), campaign_id=campaign.id, platform=platform, title=title, hook=hook, script=script, caption=f"{hook} {CTA_POOL[index % len(CTA_POOL)]}", hashtags=generate_hashtags(campaign.topic, campaign.niche, platform), cta=CTA_POOL[index % len(CTA_POOL)], template="tiktok_classic", creative_style=rules["style"], duration_seconds=rules["duration"], safe_zone=(brand.safe_zone if brand else {"top": 160, "bottom": 280, "left": 64, "right": 64}), publish_targets=[platform], created_at=created, updated_at=created, ) variant.render_payload = build_render_payload(campaign, variant, brand) variants.append(variant) return variants def recommend_schedule(count: int, start: str | None = None) -> list[str]: if start: try: current = datetime.fromisoformat(start.replace("Z", "+00:00")) except ValueError: current = datetime.now(timezone.utc) else: current = datetime.now(timezone.utc) + timedelta(hours=2) slots = [] for index in range(count): slot = current + timedelta(hours=index * 6) if slot.hour < 8: slot = slot.replace(hour=8, minute=0) if slot.hour > 21: slot = (slot + timedelta(days=1)).replace(hour=9, minute=0) slots.append(slot.replace(microsecond=0).isoformat()) return slots def score_analytics(metrics: dict[str, Any]) -> dict[str, Any]: views = max(int(metrics.get("views", 0)), 1) engagement = int(metrics.get("likes", 0)) + int(metrics.get("comments", 0)) * 2 + int(metrics.get("shares", 0)) * 3 + int(metrics.get("saves", 0)) * 3 engagement_rate = round(engagement / views, 4) completion_rate = float(metrics.get("completion_rate", 0)) viral_score = round((engagement_rate * 60) + (completion_rate * 40), 2) return { "engagement_rate": engagement_rate, "completion_rate": completion_rate, "viral_score": viral_score, "recommendation": "scale" if viral_score >= 20 else "iterate", } def score_hook(text: str) -> dict[str, Any]: lowered = text.lower() signals = { "curiosity": any(word in lowered for word in ("why", "most people", "nobody", "secret", "miss")), "urgency": any(word in lowered for word in ("stop", "before", "today", "now", "fastest")), "clarity": 35 <= len(text) <= 120, "specificity": bool(re.search(r"\d|one|two|three|simple|fastest", lowered)), } score = sum(25 for enabled in signals.values() if enabled) return {"score": score, "signals": signals, "recommendation": "use" if score >= 75 else "rewrite"} def score_script(script: str) -> dict[str, Any]: words = script.split() has_structure = all(marker in script.lower() for marker in ("point one", "point two", "point three")) estimated_seconds = round(len(words) / 2.6, 1) score = 40 if has_structure: score += 25 if 18 <= estimated_seconds <= 45: score += 20 if len(set(_keywords(script, 10))) >= 5: score += 15 return {"score": min(score, 100), "word_count": len(words), "estimated_seconds": estimated_seconds, "has_structure": has_structure} def score_caption(caption: str) -> dict[str, Any]: words = caption.split() readable = len(words) <= 35 has_cta = any(word in caption.lower() for word in ("follow", "save", "comment", "share", "watch")) score = 50 + (25 if readable else 0) + (25 if has_cta else 0) return {"score": score, "word_count": len(words), "readable": readable, "has_cta": has_cta} def retention_prediction(variant: Variant) -> dict[str, Any]: hook_score = score_hook(variant.hook)["score"] script_score = score_script(variant.script)["score"] caption_score = score_caption(variant.caption)["score"] base = round((hook_score * 0.45 + script_score * 0.35 + caption_score * 0.2) / 100, 3) timeline = [] for second in range(0, variant.duration_seconds + 1, max(1, variant.duration_seconds // 5)): decay = second / max(variant.duration_seconds, 1) * 0.35 timeline.append({"second": second, "predicted_retention": round(max(0.2, base - decay), 3)}) return {"predicted_completion_rate": round(max(0.2, base - 0.18), 3), "timeline": timeline} def compliance_check(payload: ComplianceCheckRequest) -> dict[str, Any]: issues = [] warnings = [] max_duration = { "tiktok": 180, "instagram_reels": 90, "facebook_shorts": 90, "youtube_shorts": 60, }[payload.platform] if payload.duration_seconds > max_duration: issues.append(f"Duration exceeds {payload.platform} recommended short-form limit of {max_duration}s.") if len(payload.hashtags) > 15: warnings.append("Hashtag count is high; consider using 5-12 focused tags.") restricted_terms = {"guaranteed", "miracle", "cure", "risk-free", "get rich quick"} text = f"{payload.title} {payload.caption}".lower() found = sorted(term for term in restricted_terms if term in text) if found: issues.append(f"Potential compliance terms found: {', '.join(found)}.") if payload.metadata.get("sponsored") and not payload.has_disclosure: issues.append("Sponsored content should include a disclosure.") return {"passed": not issues, "issues": issues, "warnings": warnings} def rewrite_script(script: str) -> str: lines = [line.strip() for line in script.splitlines() if line.strip()] if not lines: return script if not lines[0].lower().startswith(("stop", "most", "here", "before", "this")): lines.insert(0, "Stop scrolling. This is the part that matters.") if not any("follow" in line.lower() or "save" in line.lower() for line in lines): lines.append("Save this and follow for the next practical example.") return "\n".join(lines) def generate_content_calendar(payload: ContentCalendarPlanRequest) -> list[dict[str, Any]]: total = payload.days * payload.posts_per_day slots = recommend_schedule(total, payload.start_at) topics = [ f"{payload.niche} mistake to avoid", f"{payload.niche} quick win", f"{payload.niche} beginner lesson", f"{payload.niche} case study", f"{payload.niche} myth vs fact", ] entries = [] for index, slot in enumerate(slots): platform = payload.platforms[index % len(payload.platforms)] topic = topics[index % len(topics)] entries.append( { "workspace_id": payload.workspace_id, "platform": platform, "scheduled_at": slot, "topic": topic, "format": ["talking_head", "broll_caption", "quote_card", "storytime"][index % 4], "hook": generate_hooks(topic, payload.niche, 1)[0], } ) return entries def trend_recommendations(signals: list[TrendSignal], niche: str = "") -> list[dict[str, Any]]: filtered = [signal for signal in signals if not niche or signal.niche == niche or not signal.niche] ranked = sorted(filtered, key=lambda item: (item.score + item.velocity), reverse=True) return [ { "keyword": signal.keyword, "platform": signal.platform, "score": signal.score, "velocity": signal.velocity, "campaign_topic": f"{signal.keyword} for {niche or signal.niche or 'your audience'}", "hook": generate_hooks(signal.keyword, niche or signal.niche, 1)[0], } for signal in ranked[:20] ]