pipeline/agents.py

"""Claude agents.

- break_script:        the MVP donor's script-chunking skill — split one free-form
                       Swedish script into hook / body / cta chunks (SPEC_NEW §7.5).
                       Each chunk is sized to fit a 4–8 s clip (≤ 16 words).
- rephrase_spoken_line: SPEC_NEW §12.5 — used by the RAI policy in veo.py.
- strategist / write_script: SPEC_NEW §7.5 — Phase 4 (stubs below).

Both LLM helpers return JSON-or-text only; break_script parses defensively and
retries once with the parse error appended (the contract).
"""

from __future__ import annotations

import base64
import json
import re

from psycopg.types.json import Json

from .config import get_settings
from .duration import MAX_SPOKEN_WORDS, count_spoken_words

# ── break_script: free script → chunks (the MVP donor's clip-production skill) ──

BREAK_SCRIPT_SYSTEM = f"""You break a short Swedish UGC video script into spoken \
chunks for short vertical video clips. Each chunk becomes ONE 4–8 second clip in \
which the avatar says exactly that line to camera.

Rules:
- Output one hook (the opening line), zero to four body chunks, and one cta (the \
closing line).
- EVERY chunk must be at most {MAX_SPOKEN_WORDS} spoken words — a hard limit. Aim \
for 6–14 words per body chunk; split any longer sentence into several body chunks \
rather than exceeding {MAX_SPOKEN_WORDS} words.
- The number of body chunks follows the script length (aim 1–4). If the script is \
very short, return an empty body list (just hook + cta).
- Keep natural, spoken Swedish (talspråk — contractions, "asså", "typ" where it \
fits). Do not invent claims or add words the script does not imply.
- The hook must front-load the main point in the first ~5 words.
- Spell out every number or price in Swedish words ("trehundra kronor", never \
"300 kr"). No digits anywhere.
- Do NOT use any quotation marks inside a chunk.
- Output ONLY this JSON object — no prose, no markdown fences:
{{"hook": "string", "body_segments": ["string", ...], "cta": "string"}}"""


def _split_long(text: str, max_words: int = MAX_SPOKEN_WORDS) -> list[str]:
    """Safety net: greedily split a chunk that exceeds the word cap so a slightly
    disobedient model never blocks video creation."""
    words = text.split()
    if len(words) <= max_words:
        return [text.strip()]
    return [" ".join(words[i:i + max_words]) for i in range(0, len(words), max_words)]


def _parse_chunks(raw: str) -> dict:
    s = raw.strip()
    a, b = s.find("{"), s.rfind("}")
    if a == -1 or b == -1:
        raise ValueError("no JSON object in break_script output")
    obj = json.loads(s[a:b + 1])
    hook = str(obj.get("hook") or "").strip()
    cta = str(obj.get("cta") or "").strip()
    body_raw = [str(x).strip() for x in (obj.get("body_segments") or []) if str(x).strip()]
    if not hook:
        raise ValueError("break_script returned no hook")
    # Guarantee every chunk fits a clip (hook/cta are short by nature; bodies may
    # need the safety split).
    body: list[str] = []
    for chunk in body_raw:
        body.extend(_split_long(chunk))
    if count_spoken_words(hook) > MAX_SPOKEN_WORDS:
        pieces = _split_long(hook)
        hook = pieces[0]
        body = pieces[1:] + body
    return {"hook": hook, "body_segments": body, "cta": cta}


def _fallback_chunks(script_text: str) -> dict:
    """Key-free chunking: split into sentences, cap each at MAX_SPOKEN_WORDS, map to
    hook / body[] / cta. Keeps the manus→clips flow working with no ANTHROPIC_API_KEY
    (e.g. a free public demo). Mirrors the deterministic fallbacks of the other agents."""
    cleaned = re.sub(r"[\"“”']", "", script_text).strip()
    sentences = [p.strip() for p in re.split(r"(?<=[.!?])\s+|\n+", cleaned) if p.strip()]
    pieces: list[str] = []
    for sent in sentences:
        pieces.extend(_split_long(sent))
    if not pieces:
        raise ValueError("empty script")
    hook = pieces[0]
    if len(pieces) == 1:
        return {"hook": hook, "body_segments": [], "cta": ""}
    return {"hook": hook, "body_segments": pieces[1:-1], "cta": pieces[-1]}


def break_script(script_text: str, avatar: dict | None = None) -> dict:
    """Split a free-form Swedish script into {hook, body_segments[], cta} sized to
    4–8 s clips. Uses Claude when ANTHROPIC_API_KEY is set; otherwise a deterministic
    rule-based fallback (so the flow works key-free, e.g. a public demo)."""
    script_text = (script_text or "").strip()
    if not script_text:
        raise ValueError("empty script")
    s = get_settings()
    if not s.anthropic_api_key:
        return _fallback_chunks(script_text)

    import anthropic

    client = anthropic.Anthropic(api_key=s.anthropic_api_key)
    last_err: str | None = None
    for attempt in range(2):
        user = script_text if attempt == 0 else (
            f"{script_text}\n\nDitt förra svar gick inte att tolka som JSON "
            f"({last_err}). Svara ENBART med giltig JSON enligt formatet."
        )
        resp = client.messages.create(
            model=s.anthropic_model, max_tokens=1024,
            system=BREAK_SCRIPT_SYSTEM, messages=[{"role": "user", "content": user}],
        )
        text = "".join(b.text for b in resp.content if getattr(b, "type", None) == "text")
        try:
            return _parse_chunks(text)
        except Exception as exc:  # noqa: BLE001 — retry once, then fail loud
            last_err = str(exc)
    raise RuntimeError(f"break_script could not produce valid JSON: {last_err}")


# ── RAI rephrase (SPEC_NEW §12.5) ────────────────────────────────────────────

REPHRASE_SYSTEM = """You receive ONE spoken sentence in Swedish that an avatar says \
straight to camera in a short UGC ad. A video model's safety filter blocked it — \
almost always because the line sounds too imperative, makes a health claim, or reads \
as overt sales/marketing copy.

Rewrite ONLY this one sentence. Make it softer, more conversational, more \
descriptive, less directly imperative — while preserving the original intent and the \
avatar's natural voice. Hard constraints:
- Keep it in colloquial spoken Swedish (talspråk).
- At most 16 words.
- No digits — spell any number or price out in Swedish words ("trehundra kronor").
- No quotation marks of any kind.

Return ONLY the rewritten sentence as plain text. No prefix, no quotes, no commentary."""


def rephrase_spoken_line(spoken_text: str) -> str:
    """Minimally rewrite a single Swedish line so it is likelier to pass Veo's RAI
    filter on retry (SPEC_NEW §12.5). Returns the bare sentence."""
    s = get_settings()
    s.require("ANTHROPIC_API_KEY")

    import anthropic

    client = anthropic.Anthropic(api_key=s.anthropic_api_key)
    response = client.messages.create(
        model=s.anthropic_model, max_tokens=512,
        system=REPHRASE_SYSTEM, messages=[{"role": "user", "content": spoken_text}],
    )
    text = "".join(b.text for b in response.content if getattr(b, "type", None) == "text").strip()
    text = text.strip().strip('"').strip("”").strip("“").strip()
    if not text:
        raise RuntimeError("rephrase returned empty text")
    return text


# ── Phase 4: strategist + scriptwriter (the learning brain, SPEC_NEW §7.5) ────
#
# strategize():  aggregate the last N days of metrics per CREATIVE attribute
#   (angle + avatar), explore/exploit 70/30, pick a product (§15.5), insert briefs.
#   It NEVER sees, reasons about, or emits a render seed (§12.4): its SQL reads
#   videos.tags only, so a seed cannot enter its feature space.
# write_script(): turn one brief into a validated {hook, body_segments[], cta} and
#   wire it through flow.create_video, then stamp brief/product/mode/hypothesis onto
#   the row — with NO seed key in videos.tags (§12.4).

# EXPLOIT-basis data guards (mirror prompts.strategist; the SQL pre-filters, the
# prompt restates). A creative needs real signal before we double down on it.
MIN_EXPLOIT_IMPRESSIONS = 2000
MIN_EXPLOIT_SPEND_CPA_MULTIPLE = 1.5


def _aggregate_creative_stats(days: int, target_cpa: float) -> list[dict]:
    """Aggregate metrics_daily (last `days`) per CREATIVE attribute — angle
    (videos.tags->>'angle') + avatar_id — joining publishes → videos (SPEC_NEW §7.5).

    The strategist's feature space is creative levers ONLY: this SQL reads
    videos.tags, never videos.seed (§12.4). Each row carries the EXPLOIT guard
    verdict (`exploitable`) so the fallback and the prompt agree on what is proven.
    """
    from . import db

    rows = db.fetch_all(
        """
        select
            v.tags->>'angle'            as angle,
            v.avatar_id                 as avatar_id,
            coalesce(sum(m.impressions), 0) as impressions,
            coalesce(sum(m.spend), 0)       as spend,
            coalesce(sum(m.video_3s), 0)    as video_3s,
            coalesce(sum(m.purchases), 0)   as purchases,
            coalesce(sum(m.revenue), 0)     as revenue
          from metrics_daily m
          join publishes p on p.ad_id = m.ad_id
          join videos    v on v.id = p.video_id
         where m.date >= (current_date - %s::int)
         group by v.tags->>'angle', v.avatar_id
         order by purchases desc, revenue desc
        """,
        (days,),
    )
    spend_floor = MIN_EXPLOIT_SPEND_CPA_MULTIPLE * float(target_cpa or 0.0)
    stats: list[dict] = []
    for r in rows:
        impressions = int(r["impressions"] or 0)
        spend = float(r["spend"] or 0.0)
        video_3s = int(r["video_3s"] or 0)
        purchases = int(r["purchases"] or 0)
        revenue = float(r["revenue"] or 0.0)
        cpa = round(spend / purchases, 2) if purchases else None
        three_s_rate = round(video_3s / impressions, 4) if impressions else None
        exploitable = impressions >= MIN_EXPLOIT_IMPRESSIONS and spend >= spend_floor
        stats.append({
            "angle": r["angle"],
            "avatar_id": str(r["avatar_id"]) if r["avatar_id"] else None,
            "impressions": impressions,
            "spend": round(spend, 2),
            "video_3s": video_3s,
            "purchases": purchases,
            "revenue": round(revenue, 2),
            "cpa": cpa,
            "three_s_rate": three_s_rate,
            "exploitable": exploitable,
        })
    return stats


# A small rotation of fresh angles for EXPLORE / cold start, so the strategist
# always has something new to try even with zero data.
_EXPLORE_ANGLES = [
    "problem-agitate", "social-proof", "before-after", "founder-story",
    "myth-busting", "day-in-the-life", "unboxing-reaction", "comparison",
]


def _round_robin(items: list, i: int):
    return items[i % len(items)] if items else None


def _fallback_briefs(
    stats: list[dict], avatars: list[dict], products: list[dict], n: int, target_cpa: float,
) -> list[dict]:
    """Deterministic strategist for no-key / dry-run (SPEC_NEW §7.5).

    Exploit the best PROVEN angle+avatar (most purchases, then revenue); explore a
    fresh angle on an existing avatar; round-robin avatars/products across the rest.
    Never emits a seed (§12.4). Cold start (no proven stats) yields explore briefs.
    """
    avatar_ids = [str(a["id"]) for a in avatars]
    product_ids = [str(p["id"]) for p in products]
    used_angles = {(s.get("angle") or "").strip() for s in stats if s.get("angle")}
    proven = [s for s in stats if s.get("exploitable") and s.get("avatar_id") in set(avatar_ids)]
    proven.sort(key=lambda s: (s.get("purchases", 0), s.get("revenue", 0.0)), reverse=True)

    # Target split: 70% exploit, 30% explore, but only as many exploits as we have
    # proven creatives to base them on (everything else becomes an explore).
    want_exploit = min(len(proven), round(n * 0.7)) if proven else 0
    # Guarantee at least one explore when n >= 2 (the acceptance test wants a mix).
    if n >= 2 and want_exploit >= n:
        want_exploit = n - 1

    briefs: list[dict] = []
    pi = 0

    # EXPLOIT briefs — double down on the best proven angle+avatar.
    for k in range(want_exploit):
        src = proven[k % len(proven)]
        product_id = _round_robin(product_ids, pi) if product_ids else None
        pi += 1
        cpa = src.get("cpa")
        cpa_txt = f"{cpa} kr CPA" if cpa is not None else "an unknown CPA"
        briefs.append({
            "mode": "exploit",
            "angle": src.get("angle") or "social-proof",
            "avatar_id": src["avatar_id"],
            "product_id": product_id,
            "hypothesis": (
                f"Doubling down on the proven {src.get('angle') or 'social-proof'} angle "
                f"with this avatar should keep CPA at or below {cpa_txt} while scaling spend."
            ),
            "rationale": (
                f"Proven creative: {src.get('purchases', 0)} purchases, "
                f"{src.get('revenue', 0.0)} revenue over {src.get('impressions', 0)} impressions "
                f"({cpa_txt}); clears the {MIN_EXPLOIT_IMPRESSIONS}-impression / "
                f"{MIN_EXPLOIT_SPEND_CPA_MULTIPLE}x-CPA-spend guard, so it is safe to exploit."
            ),
            "basis": src,
        })

    # EXPLORE briefs — fresh angles on existing avatars, round-robined.
    fresh = [a for a in _EXPLORE_ANGLES if a not in used_angles] or list(_EXPLORE_ANGLES)
    ai, fa = 0, 0
    while len(briefs) < n:
        avatar_id = _round_robin(avatar_ids, ai)
        ai += 1
        angle = _round_robin(fresh, fa)
        fa += 1
        product_id = _round_robin(product_ids, pi) if product_ids else None
        pi += 1
        briefs.append({
            "mode": "explore",
            "angle": angle,
            "avatar_id": avatar_id,
            "product_id": product_id,
            "hypothesis": (
                f"A fresh {angle} angle on this avatar may open a new winning creative "
                f"lane that the current data does not yet cover."
            ),
            "rationale": (
                "Explore lane: this angle/avatar pairing has little or no proven data "
                f"(under the {MIN_EXPLOIT_IMPRESSIONS}-impression guard), so it is a new "
                "bet rather than an exploit — worth a probe to widen the learning."
                if stats else
                "Cold start: no performance data yet, so every brief is an explore probe "
                "across fresh angles and the available avatars."
            ),
            "basis": {"cold_start": not stats, "explored_angle": angle},
        })

    return briefs[:n]


def _parse_strategist_json(raw: str) -> list[dict]:
    """Defensive parse of the strategist's JSON (first '{' to last '}'). Returns the
    briefs list; raises ValueError on anything malformed."""
    s = raw.strip()
    a, b = s.find("{"), s.rfind("}")
    if a == -1 or b == -1:
        raise ValueError("no JSON object in strategist output")
    obj = json.loads(s[a:b + 1])
    briefs = obj.get("briefs")
    if not isinstance(briefs, list) or not briefs:
        raise ValueError("strategist JSON has no non-empty 'briefs' array")
    out: list[dict] = []
    for item in briefs:
        if not isinstance(item, dict):
            continue
        mode = str(item.get("mode") or "").strip().lower()
        angle = str(item.get("angle") or "").strip()
        avatar_id = item.get("avatar_id")
        if mode not in ("exploit", "explore") or not angle or not avatar_id:
            raise ValueError(f"malformed brief: {item!r}")
        out.append({
            "mode": mode,
            "angle": angle,
            "avatar_id": str(avatar_id),
            "product_id": str(item["product_id"]) if item.get("product_id") else None,
            "hypothesis": str(item.get("hypothesis") or "").strip(),
            "rationale": str(item.get("rationale") or "").strip(),
        })
    if not out:
        raise ValueError("strategist returned no usable briefs")
    return out


def strategize(days: int = 14, n: int | None = None) -> list[dict]:
    """Decide the next N creatives and insert them as pending briefs (SPEC_NEW §7.5).

    Aggregates the last `days` of metrics_daily per creative attribute (angle +
    avatar), honors a 70/30 exploit/explore split under the EXPLOIT data guard
    (< 2000 impressions or < 1.5× target-CPA spend is unproven), picks a product
    per brief (§15.5), and NEVER touches a render seed (§12.4). Uses Claude
    (STRATEGIST_SYSTEM) when ANTHROPIC_API_KEY is set; otherwise a deterministic
    fallback runs (so dry-run/tests work). Cold start still yields explore briefs.

    Returns the inserted brief dicts (uuids stringified).
    """
    from . import db
    from .prompts import strategist as sp

    n = 5 if n is None else int(n)
    if n < 1:
        raise ValueError("strategize: n must be >= 1")

    s = get_settings()
    avatars = db.fetch_all(
        "select id, name, persona, prompt_template from avatars "
        "where coalesce(status,'active') = 'active' order by created_at"
    )
    if not avatars:
        raise RuntimeError("strategize: no avatars exist — create at least one first")
    products = db.fetch_all(
        "select id, name, knowledge from products where active = true order by created_at"
    )
    stats = _aggregate_creative_stats(days, s.target_cpa)

    briefs: list[dict] = []
    if s.anthropic_api_key:
        import anthropic

        # The model never sees under-powered creatives as exploitable, and never a
        # seed: stats carry only creative attributes (§12.4).
        exploit_pool = [dict(st) for st in stats if st.get("exploitable")]
        prompt_stats = exploit_pool + [st for st in stats if not st.get("exploitable")]
        avatar_ids = {str(a["id"]) for a in avatars}
        product_ids = {str(p["id"]) for p in products}
        by_key = {(st.get("angle"), st.get("avatar_id")): st for st in stats}

        client = anthropic.Anthropic(api_key=s.anthropic_api_key)
        user = sp.build_user_prompt(prompt_stats, avatars, products, n)
        last_err: str | None = None
        for attempt in range(2):
            content = user if attempt == 0 else (
                f"{user}\n\nYour previous reply could not be used ({last_err}). "
                f"Reply with ONLY valid JSON in the required format."
            )
            resp = client.messages.create(
                model=s.anthropic_model, max_tokens=2048,
                system=sp.STRATEGIST_SYSTEM, messages=[{"role": "user", "content": content}],
            )
            text = "".join(b.text for b in resp.content if getattr(b, "type", None) == "text")
            try:
                parsed = _parse_strategist_json(text)
            except Exception as exc:  # noqa: BLE001 — retry once, then fall back
                last_err = str(exc)
                continue
            # Keep only briefs on a real avatar; attach the metric snapshot as basis.
            for bf in parsed:
                if bf["avatar_id"] not in avatar_ids:
                    continue
                if bf["product_id"] not in product_ids:
                    bf["product_id"] = None
                bf["basis"] = by_key.get((bf["angle"], bf["avatar_id"])) or {
                    "angle": bf["angle"], "avatar_id": bf["avatar_id"], "novel": True,
                }
                briefs.append(bf)
            if briefs:
                break
            last_err = "no brief landed on a real avatar"
        if not briefs:  # Claude failed twice → deterministic fallback (still works)
            briefs = _fallback_briefs(stats, avatars, products, n, s.target_cpa)
    else:
        briefs = _fallback_briefs(stats, avatars, products, n, s.target_cpa)

    # Persist each brief (source=strategist, status=pending, basis = the snapshot).
    inserted: list[dict] = []
    with db.connect() as conn:
        for bf in briefs:
            basis = bf.get("basis") or {}
            row = conn.execute(
                """
                insert into briefs
                    (source, mode, angle, avatar_id, product_id, hypothesis, rationale, basis, status)
                values ('strategist', %s, %s, %s, %s, %s, %s, %s, 'pending')
                returning id, source, mode, angle, avatar_id, product_id, hypothesis,
                          rationale, basis, status, created_at
                """,
                (bf["mode"], bf["angle"], bf["avatar_id"], bf.get("product_id"),
                 bf.get("hypothesis") or "", bf.get("rationale") or "",
                 Json(basis)),
            ).fetchone()
            d = dict(row)
            d["id"] = str(d["id"])
            d["avatar_id"] = str(d["avatar_id"]) if d.get("avatar_id") else None
            d["product_id"] = str(d["product_id"]) if d.get("product_id") else None
            inserted.append(d)
    return inserted


def _winning_hook_examples(limit: int = 3) -> list[str]:
    """Recent hooks from high-purchase videos (SPEC_NEW §7.5 winning-hook context).
    Best-effort — returns [] when there is no performance data yet."""
    from . import db

    rows = db.fetch_all(
        """
        select sc.hook, coalesce(sum(m.purchases), 0) as purchases
          from publishes p
          join videos v   on v.id = p.video_id
          join scripts sc on sc.id = v.script_id
          join metrics_daily m on m.ad_id = p.ad_id
         where sc.hook is not null and sc.hook <> ''
         group by sc.hook
         order by purchases desc, max(p.published_at) desc nulls last
         limit %s
        """,
        (limit,),
    )
    return [r["hook"] for r in rows if (r.get("purchases") or 0) > 0]


def _validate_segments(hook: str, body_segments: list[str], cta: str) -> None:
    """Every segment must pass the Veo line validator AND fit a duration bucket
    (SPEC_NEW §7.5). Raises on the first offender."""
    from .duration import decide_duration
    from .prompts.veo_segment import validate_spoken_text

    for text in [hook, *body_segments, cta]:
        if not text:
            continue
        validate_spoken_text(text)
        decide_duration(text)


def _safety_split_script(hook: str, body_segments: list[str], cta: str) -> tuple[str, list[str], str]:
    """Final safety net (mirrors break_script): greedily split any over-cap line so a
    slightly disobedient model can never block video creation. Hook/CTA overflow
    spills into the body so they stay single, front-loaded lines."""
    body: list[str] = []
    for seg in body_segments:
        body.extend(p for p in _split_long(seg) if p)
    hook_pieces = _split_long(hook)
    hook, body = hook_pieces[0], hook_pieces[1:] + body
    if cta:
        cta_pieces = _split_long(cta)
        cta, body = cta_pieces[-1], body + cta_pieces[:-1]
    return hook, body, cta


def _fallback_script(brief: dict, avatar: dict) -> dict:
    """Deterministic scriptwriter for no-key / dry-run (SPEC_NEW §7.5).

    Hand-built talspråk lines that ALL pass validate_spoken_text + decide_duration
    (≤ 18 words, no digits, no quotes, numbers spelled out) — 10/10, every run.
    Front-loaded hook, concrete CTA. Personalized lightly from the brief angle.
    """
    angle = (brief.get("angle") or "social-proof").strip()
    name = (avatar.get("name") or "").strip()
    intro = f"Asså, {name}, " if name else "Asså, "
    # Each line is short, colloquial, digit-free, quote-free, ≤ 18 words.
    hook = "Asså det här bytte faktiskt allt för mig"
    body = [
        f"{intro}jag var helt skeptisk i början, typ att det aldrig skulle funka.",
        "Men efter bara nån vecka kände jag verkligen skillnaden, helt ärligt.",
        "Nu kan jag liksom inte tänka mig att vara utan den längre.",
    ]
    if angle in ("before-after", "comparison"):
        body[0] = "Innan var det jobbigt varje dag, asså det tog typ all energi."
        body[1] = "Nu efteråt är det liksom en helt annan känsla, mycket lugnare."
    elif angle in ("myth-busting",):
        body[0] = "Alla sa att sånt här aldrig funkar, men det stämmer faktiskt inte."
    cta = "Testa den själv idag, du kommer inte ångra dig."
    hook, body, cta = _safety_split_script(hook, body, cta)
    return {"hook": hook, "body_segments": body, "cta": cta}


def _parse_script_json(raw: str) -> dict:
    """Defensive parse of the scriptwriter's JSON (first '{' to last '}')."""
    s = raw.strip()
    a, b = s.find("{"), s.rfind("}")
    if a == -1 or b == -1:
        raise ValueError("no JSON object in scriptwriter output")
    obj = json.loads(s[a:b + 1])
    hook = str(obj.get("hook") or "").strip()
    cta = str(obj.get("cta") or "").strip()
    body = [str(x).strip() for x in (obj.get("body_segments") or []) if str(x).strip()]
    if not hook:
        raise ValueError("scriptwriter returned no hook")
    return {"hook": hook, "body_segments": body, "cta": cta}


def write_script(brief_id: str) -> dict:
    """Turn one brief into a validated Swedish UGC script and wire it into the
    render flow (SPEC_NEW §7.5).

    Loads the brief + its avatar + recent winning-hook examples, produces
    {hook, body_segments[], cta} where EVERY segment passes
    veo_segment.validate_spoken_text AND duration.decide_duration (≤ 18 words, no
    digits, no quotes, numbers spelled out). Claude (SCRIPTWRITER_SYSTEM) when keyed
    — parse + retry once with the validator error appended, final safety split;
    deterministic fallback otherwise (itself 10/10 on the validator).

    Persists via flow.create_video (NO seed key in videos.tags — §12.4), links the
    script.brief_id, stamps videos.product_id + tags {brief_id, mode, hypothesis},
    and marks the brief 'scripted'. Returns
    {script_id, video_id, hook, body_segments, cta}.
    """
    from . import db, flow
    from .prompts import scriptwriter as swp

    brief = db.fetch_one("select * from briefs where id = %s", (brief_id,))
    if brief is None:
        raise ValueError(f"no brief {brief_id}")
    avatar = db.fetch_one("select * from avatars where id = %s", (brief["avatar_id"],))
    if avatar is None:
        raise RuntimeError(f"brief {brief_id} has no avatar")

    s = get_settings()
    examples = _winning_hook_examples()

    script: dict | None = None
    if s.anthropic_api_key:
        import anthropic

        client = anthropic.Anthropic(api_key=s.anthropic_api_key)
        user = swp.build_user_prompt(brief, avatar, examples)
        last_err: str | None = None
        for attempt in range(2):
            content = user if attempt == 0 else (
                f"{user}\n\nDitt förra svar dög inte ({last_err}). Svara ENBART med "
                f"giltig JSON enligt formatet, och håll varje rad kort och korrekt."
            )
            resp = client.messages.create(
                model=s.anthropic_model, max_tokens=1024,
                system=swp.SCRIPTWRITER_SYSTEM, messages=[{"role": "user", "content": content}],
            )
            text = "".join(b.text for b in resp.content if getattr(b, "type", None) == "text")
            try:
                cand = _parse_script_json(text)
                # Final safety split, then assert every line is renderable.
                cand["hook"], cand["body_segments"], cand["cta"] = _safety_split_script(
                    cand["hook"], cand["body_segments"], cand["cta"]
                )
                _validate_segments(cand["hook"], cand["body_segments"], cand["cta"])
                script = cand
                break
            except Exception as exc:  # noqa: BLE001 — retry once with the error appended
                last_err = str(exc)
        if script is None:
            script = _fallback_script(brief, avatar)
    else:
        script = _fallback_script(brief, avatar)

    # Belt-and-suspenders: the fallback (and any accepted Claude output) must pass.
    _validate_segments(script["hook"], script["body_segments"], script["cta"])

    # Wire into the existing render flow — create_video validates again + builds the
    # hook takes. It writes tags {source, angle} only; NO seed (§12.4).
    video_id = flow.create_video(
        brief["avatar_id"], script["hook"], script["body_segments"], script["cta"],
        angle=brief["angle"],
    )

    # Link the script to its brief, stamp the creative attributes onto the video
    # (merging tags — still no seed key — §12.4), and advance the brief.
    extra_tags = {
        "brief_id": str(brief_id),
        "mode": brief["mode"],
        "hypothesis": brief.get("hypothesis") or "",
    }
    with db.connect() as conn:
        video = conn.execute("select script_id, tags from videos where id = %s", (video_id,)).fetchone()
        script_id = str(video["script_id"])
        merged_tags = dict(video["tags"] or {})
        merged_tags.update(extra_tags)
        merged_tags.pop("seed", None)  # invariant: a seed can never live in tags (§12.4)
        conn.execute("update scripts set brief_id = %s where id = %s", (brief_id, video["script_id"]))
        conn.execute(
            "update videos set product_id = %s, tags = %s where id = %s",
            (brief.get("product_id"), Json(merged_tags), video_id),
        )
        conn.execute("update briefs set status = 'scripted' where id = %s", (brief_id,))

    return {
        "script_id": script_id,
        "video_id": str(video_id),
        "hook": script["hook"],
        "body_segments": script["body_segments"],
        "cta": script["cta"],
    }


# ── Phase 6 (B-roll, SPEC_NEW §15.4) ─────────────────────────────────────────

BROLL_PLAN_SYSTEM = """You plan B-roll cutaways for a Swedish UGC video ad. You see a \
PRODUCT IMAGE and the avatar's spoken BODY lines — each line is one short clip where \
the avatar talks to camera. For some lines we cut away to a B-roll shot of the product \
(image-to-video from the product image) while the avatar keeps talking underneath.

For each body line decide whether to overlay B-roll, and if so write a short \
image-to-video prompt for a product shot that VISUALLY matches what the line is about.
Rules:
- Overlay B-roll on roughly HALF the lines — the ones where showing the product helps. \
Never on every line (the avatar's face must return between cutaways).
- The prompt describes THE PRODUCT FROM THE ATTACHED IMAGE in a natural real-world \
Swedish setting, vertical 9:16, soft daylight, authentic UGC, a slow gentle camera \
move, no people speaking, no on-screen text. Keep the product identical to the image.
- Respect any forbidden claims — never depict them.
- Output ONLY a JSON array, one object per body line in order:
  [{"segment_idx": <int>, "overlay": <bool>, "seedance_prompt": "<string, empty if overlay is false>"}]"""


def _fallback_broll_prompt(product: dict, line: str) -> str:
    name = product.get("name") or "the product"
    return (f"A clean, authentic UGC product shot of {name} (the product from the attached image), "
            f"in a natural Swedish home setting, vertical 9:16, soft daylight, a slow gentle camera "
            f"move, no people speaking, no on-screen text. Keep the product identical to the image.")


def _product_image_block(image: str | None) -> dict | None:
    if not image:
        return None
    try:
        if image.startswith(("http://", "https://")):
            import httpx
            r = httpx.get(image, timeout=60)
            r.raise_for_status()
            data, mime = r.content, (r.headers.get("content-type") or "image/jpeg").split(";")[0]
        else:
            from pathlib import Path
            p = Path(image)
            if not p.is_file():
                return None
            data = p.read_bytes()
            mime = {".png": "image/png", ".webp": "image/webp"}.get(p.suffix.lower(), "image/jpeg")
        return {"type": "image", "source": {"type": "base64", "media_type": mime,
                                            "data": base64.standard_b64encode(data).decode()}}
    except Exception:  # noqa: BLE001 — image unavailable → planner falls back
        return None


def plan_broll(product: dict, segments: list[dict]) -> list[dict]:
    """Decide which body segments get a B-roll overlay + a Seedance prompt for each
    (SPEC_NEW §15.4). Claude content-match when an image + ANTHROPIC_API_KEY are
    available; otherwise a deterministic fallback (every other body line) so the
    pipeline stays demoable. Returns [{segment_idx, overlay, seedance_prompt}]."""
    body = [s for s in segments if s["kind"] == "body"]
    if not body:
        return []
    s = get_settings()
    image_block = _product_image_block(product.get("image_path"))

    if not s.anthropic_api_key or image_block is None:
        return [
            {"segment_idx": seg["idx"], "overlay": True,
             "seedance_prompt": _fallback_broll_prompt(product, seg["spoken_text"])}
            for i, seg in enumerate(body) if i % 2 == 0
        ]

    import anthropic

    knowledge = product.get("knowledge") or {}
    lines = "\n".join(f'{seg["idx"]}: "{seg["spoken_text"]}"' for seg in body)
    user_text = (
        f"Product: {product.get('name', '')}\n"
        f"Knowledge: {json.dumps(knowledge, ensure_ascii=False)}\n\n"
        f"Body lines (segment_idx: line):\n{lines}\n\n"
        f"Return the JSON array (one object per body line, in order)."
    )
    client = anthropic.Anthropic(api_key=s.anthropic_api_key)
    resp = client.messages.create(
        model=s.anthropic_model, max_tokens=1500, system=BROLL_PLAN_SYSTEM,
        messages=[{"role": "user", "content": [image_block, {"type": "text", "text": user_text}]}],
    )
    text = "".join(b.text for b in resp.content if getattr(b, "type", None) == "text").strip()
    a, b = text.find("["), text.rfind("]")
    if a == -1 or b == -1:
        raise RuntimeError(f"plan_broll: no JSON array in response: {text[:200]}")
    plan = json.loads(text[a:b + 1])
    body_idxs = {seg["idx"] for seg in body}
    return [
        {"segment_idx": int(p["segment_idx"]), "overlay": bool(p.get("overlay")),
         "seedance_prompt": str(p.get("seedance_prompt") or "")}
        for p in plan if int(p.get("segment_idx", -1)) in body_idxs
    ]