maester_enterprise/automation/intelligence.py

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]
    ]