"""The City Briefing Agent orchestrator.

One run = one HTTP trigger from n8n at 09:00 LA (or a manual POST).
For each show discovered `DAYS_TO_SHOW` days out, the orchestrator runs the
full flowchart:
    hub_lookup → previous_show_lookup → historical_sales_lookup →
    voice_profile_load → youtube_transcripts → perplexity_city → perplexity_news →
    openai_fact_check → openai_generation → claude_ranking →
    html_generation → vps_deployment → url_verification → og_verification →
    mark_complete (status='published')

Multi-show isolation: each show runs inside its own asyncio task. A failure
in one show's stage never cancels sibling shows. Concurrency is bounded by
`SHOW_CONCURRENCY`.
"""

from __future__ import annotations

import asyncio
import logging
import time
from datetime import date, datetime, timedelta, timezone
from typing import Any

from app.config import settings
from app.models import PastShow, Show, VoiceProfile
from app.pipeline.retry import with_retry
from app.pipeline.stages import (
    STAGE_CLAUDE_RANKING,
    STAGE_HUB_LOOKUP,
    STAGE_HISTORICAL_SALES,
    STAGE_HTML_GENERATION,
    STAGE_OG_VERIFICATION,
    STAGE_OPENAI_FACT_CHECK,
    STAGE_OPENAI_GENERATION,
    STAGE_PERPLEXITY_CITY,
    STAGE_PERPLEXITY_NEWS,
    STAGE_PREVIOUS_SHOW,
    STAGE_URL_VERIFICATION,
    STAGE_VPS_DEPLOYMENT,
    STAGE_VOICE_PROFILE,
    STAGE_YOUTUBE_TRANSCRIPTS,
    STAGE_COLUMNS,
    as_float,
    as_int,
    brief_slug,
    redact_for_storage,
    today_in_app_tz,
)
from app.providers.anthropic_client import AnthropicClient
from app.providers.hub import SccClient
from app.providers.openai_client import OpenAIClient
from app.providers.perplexity import PerplexityClient
from app.providers.youtube import YouTubeClient
from app.repositories import brief_repo, run_repo
from app.voice.loader import load_voice_profile

logger = logging.getLogger(__name__)


class OrchestrateBriefsService:
    """Owns a run's orchestration lifecycle and provider lifecycles.

    One instance per run. Not expected to be reused across runs.
    """

    def __init__(self) -> None:
        self.scc = SccClient()
        self.perplexity = PerplexityClient()
        self.youtube = YouTubeClient()
        self.openai = OpenAIClient()
        self.anthropic = AnthropicClient()

    async def aclose(self) -> None:
        for client in (self.scc, self.perplexity, self.youtube, self.openai, self.anthropic):
            try:
                await client.close()
            except Exception:
                logger.debug("provider close failed", exc_info=True)

    async def _record_retry(
        self,
        run_id: str,
        attempt: int,
        error: BaseException | None,
    ) -> None:
        # `with_retry` calls this after every attempt. Count only failed
        # attempts that will actually be retried, not the final failure.
        if error is not None and attempt < 3:
            await run_repo.increment_retry(run_id)

    # ── Public entrypoint ───────────────────────────────────────────────

    async def run(self, *, trigger: str = "schedule",
                  run_date: date | None = None) -> dict[str, Any]:
        """Discover shows DAYS_TO_SHOW out, then orchestrate each.

        Returns a summary dict suitable for the API caller.
        """
        if run_date is None:
            run_date = today_in_app_tz()
        target = run_date + timedelta(days=settings.DAYS_TO_SHOW)
        logger.info(
            "orchestrator start: trigger=%s run_date=%s target_show_date=%s",
            trigger, run_date.isoformat(), target.isoformat(),
        )
        started = time.monotonic()

        # Stage: hub_lookup (run-level). Retries on transient failures.
        try:
            tour_shows_raw = await with_retry(
                lambda: self.scc.find_shows_on_date(target),
                stage_name="hub_lookup",
            )
        except Exception as exc:
            logger.error("hub_lookup run-level failed: %s", exc)
            return {
                "status": "failed",
                "brief_run_id": None,
                "shows": [],
                "started_at": datetime.now(timezone.utc).isoformat(),
                "completed_at": datetime.now(timezone.utc).isoformat(),
                "duration_ms": int((time.monotonic() - started) * 1000),
                "error": "hub_lookup failed \u2014 no shows processed",
            }

        shows: list[Show] = [_to_show(r) for r in (tour_shows_raw or [])]
        concurrency = max(1, settings.SHOW_CONCURRENCY)
        sem = asyncio.Semaphore(concurrency)

        accepted: list[dict[str, Any]] = []
        async def _per_show(show: Show) -> dict[str, Any]:
            async with sem:
                return await self._orchestrate_one_show(show)
        summaries = await asyncio.gather(
            *(_per_show(show) for show in shows),
            return_exceptions=False,
        )
        for s in summaries:
            accepted.append(s)

        duration_ms = int((time.monotonic() - started) * 1000)
        logger.info(
            "orchestrator done: trigger=%s shows=%d duration_ms=%d",
            trigger, len(shows), duration_ms,
        )
        return {
            "status": "ok" if all(s.get("status") == "published" for s in accepted) else "partial" if accepted else "ok",
            "briefs": accepted,
            "started_at": datetime.now(timezone.utc).isoformat(),
            "completed_at": datetime.now(timezone.utc).isoformat(),
            "duration_ms": duration_ms,
        }

    # ── Per-show orchestration ──────────────────────────────────────────

    async def _orchestrate_one_show(self, show: Show) -> dict[str, Any]:
        begin = time.monotonic()
        brief_row = await brief_repo.ensure_pending(
            show_id=show.show_id,
            show_date=show.show_date,
            city=show.city,
            country=show.country,
            venue=show.venue,
            capacity=show.capacity,
        )
        brief_id = str(brief_row["brief_id"])
        run_row = await run_repo.insert(show_id=show.show_id, show_date=show.show_date)
        run_id = str(run_row["brief_run_id"])
        await run_repo.set_stage(run_id, STAGE_HUB_LOOKUP, "ok")

        # Stage 1: check_existing_brief — idempotency gate.
        existing = await brief_repo.fetch_existing_published(show.show_id)
        if existing:
            # Mark every stage skipped because we're not re-running.
            remaining_stages = (
                STAGE_PREVIOUS_SHOW, STAGE_HISTORICAL_SALES, STAGE_VOICE_PROFILE,
                STAGE_YOUTUBE_TRANSCRIPTS,
                STAGE_PERPLEXITY_CITY, STAGE_PERPLEXITY_NEWS,
                STAGE_OPENAI_FACT_CHECK, STAGE_OPENAI_GENERATION,
                STAGE_CLAUDE_RANKING, STAGE_HTML_GENERATION,
                STAGE_VPS_DEPLOYMENT, STAGE_URL_VERIFICATION,
                STAGE_OG_VERIFICATION,
            )
            for stage in remaining_stages:
                await run_repo.set_stage(run_id, stage, "skipped")
            await run_repo.mark_complete(
                run_id, status="ok", duration_ms=int((time.monotonic() - begin) * 1000),
                error="already published",
            )
            return {
                "show_id": show.show_id,
                "city": show.city,
                "show_date": show.show_date.isoformat(),
                "status": "published",
                "public_url": existing.get("public_url"),
                "brief_id": brief_id,
                "brief_run_id": run_id,
                "reused": True,
            }

        # Promote to generating.
        await brief_repo.set_generating(brief_id)

        try:
            context = await self._execute_stages(show, brief_id, run_id)
            duration_ms = int((time.monotonic() - begin) * 1000)
            await brief_repo.mark_published(brief_id, duration_ms)
            await run_repo.set_stage(run_id, STAGE_URL_VERIFICATION, "ok")
            await run_repo.set_stage(run_id, STAGE_OG_VERIFICATION, "ok")
            await run_repo.mark_complete(run_id, status="ok", duration_ms=duration_ms)
            return {
                "show_id": show.show_id,
                "city": show.city,
                "show_date": show.show_date.isoformat(),
                "status": "published",
                "public_url": context.get("public_url"),
                "brief_id": brief_id,
                "brief_run_id": run_id,
            }
        except Exception as exc:
            logger.warning("show %s orchestration failed: %s", show.show_id, exc, exc_info=True)
            err_msg = redact_for_storage(exc)
            await brief_repo.mark_failed(brief_id, err_msg)
            current_run = await run_repo.fetch(run_id)
            if current_run:
                for stage in STAGE_COLUMNS:
                    if current_run.get(stage) == "running":
                        await run_repo.set_stage(
                            run_id,
                            stage,
                            "failed",
                            error=err_msg,
                        )
                        break
            await run_repo.mark_complete(
                run_id, status="failed",
                duration_ms=int((time.monotonic() - begin) * 1000),
                error=err_msg,
            )
            return {
                "show_id": show.show_id,
                "city": show.city,
                "show_date": show.show_date.isoformat(),
                "status": "failed",
                "brief_id": brief_id,
                "brief_run_id": run_id,
                "error": err_msg,
            }

    # ── Stage-by-stage execution ────────────────────────────────────────

    async def _execute_stages(self, show: Show, brief_id: str, run_id: str) -> dict[str, Any]:
        """Run the per-show pipeline until completion. On exception, fails fast.

        Each stage marks itself running→ok/failed in `city_briefs_runs`.
        Each AI stage is wrapped in `with_retry` so transient provider errors
        retry the stage without aborting the whole show.
        """
        ctx: dict[str, Any] = {"show": show, "brief_id": brief_id, "run_id": run_id}

        # Stage 2: previous_show_lookup
        await run_repo.set_stage(run_id, STAGE_PREVIOUS_SHOW, "running")
        prev_show = await self.scc.find_previous_show_in_city(show.city, show.show_date)
        prev: PastShow | None = _to_past_show(prev_show) if prev_show else None
        await run_repo.set_stage(run_id, STAGE_PREVIOUS_SHOW, "ok" if prev else "skipped")
        ctx["previous_show"] = prev

        # Stage 3: historical_sales_lookup
        await run_repo.set_stage(run_id, STAGE_HISTORICAL_SALES, "running")
        sales: dict[str, Any] = {}
        if prev:
            try:
                sales = await with_retry(
                    lambda: self.scc.historical_sales(
                        prev.show_id,
                        show_date=prev.show_date,
                        city=prev.city,
                    ),
                    on_attempt=lambda attempt, error: self._record_retry(
                        run_id, attempt, error
                    ),
                    stage_name=STAGE_HISTORICAL_SALES,
                )
            except Exception as exc:
                logger.warning("historical_sales failed for show %s: %s", prev.show_id, exc)
                sales = {}
        if sales:
            await run_repo.set_stage(run_id, STAGE_HISTORICAL_SALES, "ok")
        else:
            await run_repo.set_stage(
                run_id,
                STAGE_HISTORICAL_SALES,
                "skipped",
                error="no historical sales record matched the previous show",
            )
        ctx["historical_sales"] = sales

        # Stage 4: voice_profile_load
        await run_repo.set_stage(run_id, STAGE_VOICE_PROFILE, "running")
        voice = load_voice_profile()
        await run_repo.set_stage(run_id, STAGE_VOICE_PROFILE, "ok")
        ctx["voice"] = voice

        # Stage 4.5: youtube_transcripts — Max Amini's most recent uploads +
        # their transcripts, used to ground idea-gen so the 20 ideas don't
        # repeat his most recent posted bits. Tolerates missing transcripts
        # per-video (only fails if the Data API listing itself fails).
        await run_repo.set_stage(run_id, STAGE_YOUTUBE_TRANSCRIPTS, "running")
        try:
            recent_transcripts = await with_retry(
                lambda: self.youtube.list_recent_with_transcripts(),
                on_attempt=lambda attempt, error: self._record_retry(
                    run_id, attempt, error
                ),
                stage_name=STAGE_YOUTUBE_TRANSCRIPTS,
            )
            await run_repo.set_stage(run_id, STAGE_YOUTUBE_TRANSCRIPTS, "ok")
        except Exception as exc:
            logger.warning("youtube_transcripts failed for show %s: %s — continuing without transcripts",
                           show.show_id, exc)
            recent_transcripts = []
            await run_repo.set_stage(run_id, STAGE_YOUTUBE_TRANSCRIPTS, "skipped",
                                     error=_truncate_err(str(exc)))
        ctx["recent_transcripts"] = recent_transcripts

        # Stage 5: perplexity_city
        await run_repo.set_stage(run_id, STAGE_PERPLEXITY_CITY, "running")
        city_intel = await with_retry(
            lambda: self.perplexity.city_intelligence(
                city=show.city, country=show.country, show_date=show.show_date,
            ),
            on_attempt=lambda attempt, error: self._record_retry(
                run_id, attempt, error
            ),
            stage_name=STAGE_PERPLEXITY_CITY,
        )
        await run_repo.set_stage(run_id, STAGE_PERPLEXITY_CITY, "ok")
        ctx["city_intelligence"] = city_intel.model_dump()

        # Stage 6: perplexity_news (+ competitor_warnings)
        await run_repo.set_stage(run_id, STAGE_PERPLEXITY_NEWS, "running")
        since = (prev.show_date if prev else None) or (show.show_date - timedelta(days=30))
        news = await with_retry(
            lambda: self.perplexity.local_news(
                city=show.city, country=show.country,
                since_date=since, show_date=show.show_date,
            ),
            on_attempt=lambda attempt, error: self._record_retry(
                run_id, attempt, error
            ),
            stage_name=STAGE_PERPLEXITY_NEWS,
        )
        await run_repo.set_stage(run_id, STAGE_PERPLEXITY_NEWS, "ok")
        ctx["recent_news"] = [it.model_dump() for it in news.recent_news]
        ctx["competitor_warnings"] = [it.model_dump() for it in news.competitor_warnings]

        # Stage 7: openai_fact_check (replaces gemini_fact_check per Phase 12 decision)
        await run_repo.set_stage(run_id, STAGE_OPENAI_FACT_CHECK, "running")
        fact_result = await with_retry(
            lambda: self.openai.fact_check(
                city=show.city, country=show.country, show_date=show.show_date,
                city_intel=ctx["city_intelligence"],
                recent_news=ctx["recent_news"],
            ),
            on_attempt=lambda attempt, error: self._record_retry(
                run_id, attempt, error
            ),
            stage_name=STAGE_OPENAI_FACT_CHECK,
        )
        ctx["verified_facts"] = [f.model_dump() for f in fact_result.verified_facts]
        ctx["removed_facts"] = [f.model_dump() for f in fact_result.removed_facts]
        await run_repo.set_stage(run_id, STAGE_OPENAI_FACT_CHECK, "ok")

        # Stage 8: openai_generation (uses recent_transcripts to avoid repeating bits)
        await run_repo.set_stage(run_id, STAGE_OPENAI_GENERATION, "running")
        ideas = await with_retry(
            lambda: self.openai.generate_twenty_ideas(
                city=show.city, country=show.country, show_date=show.show_date,
                voice=voice,
                verified_facts=ctx["verified_facts"],
                city_intel=ctx["city_intelligence"],
                competitor_warnings=ctx["competitor_warnings"],
                previous_show_summary=_previous_show_summary(prev, sales),
                recent_transcripts=ctx.get("recent_transcripts"),
            ),
            on_attempt=lambda attempt, error: self._record_retry(
                run_id, attempt, error
            ),
            stage_name=STAGE_OPENAI_GENERATION,
        )
        ctx["generated_topics"] = [i.model_dump() for i in ideas]
        await run_repo.set_stage(run_id, STAGE_OPENAI_GENERATION, "ok")

        # Stage 9: claude_ranking (top 10)
        await run_repo.set_stage(run_id, STAGE_CLAUDE_RANKING, "running")
        ranked = await with_retry(
            lambda: self.anthropic.rank_top_ten(
                city=show.city, show_date=show.show_date,
                voice=voice,
                verified_facts=ctx["verified_facts"],
                ideas=ctx["generated_topics"],
            ),
            on_attempt=lambda attempt, error: self._record_retry(
                run_id, attempt, error
            ),
            stage_name=STAGE_CLAUDE_RANKING,
        )
        ctx["ranked_topics"] = [i.model_dump() for i in ranked]
        ctx["crowd_work"] = ctx["ranked_topics"]
        await run_repo.set_stage(run_id, STAGE_CLAUDE_RANKING, "ok")

        # Stage 10: html_generation
        await run_repo.set_stage(run_id, STAGE_HTML_GENERATION, "running")
        slug = brief_slug(show.city, show.show_date)
        public_url = f"{settings.PUBLIC_BASE_URL.rstrip('/')}/{slug}"
        from app.html.generator import generate_brief_html
        html = generate_brief_html(
            show=show,
            voice=voice,
            city_intel=ctx["city_intelligence"],
            recent_news=ctx["recent_news"],
            verified_facts=ctx["verified_facts"],
            removed_facts=ctx["removed_facts"],
            crowd_work=ctx["crowd_work"],
            competitor_warnings=ctx["competitor_warnings"],
            previous_show=prev,
            historical_sales=sales,
            recent_transcripts=ctx.get("recent_transcripts") or [],
            public_url=public_url,
        )
        await run_repo.set_stage(run_id, STAGE_HTML_GENERATION, "ok")
        ctx["html"] = html
        ctx["slug"] = slug
        ctx["public_url"] = public_url

        # Brief state transition generated → deploying
        await brief_repo.set_generated(
            brief_id,
            generated_topics=ctx["generated_topics"],
            ranked_topics=ctx["ranked_topics"],
            verified_facts=ctx["verified_facts"],
            removed_facts=ctx["removed_facts"],
            city_intelligence=ctx["city_intelligence"],
            recent_news=ctx["recent_news"],
            competitor_warnings=ctx["competitor_warnings"],
            crowd_work=ctx["crowd_work"],
            recent_transcripts=ctx.get("recent_transcripts") or [],
            previous_show_date=prev.show_date if prev else None,
            previous_show_venue=prev.venue if prev else None,
            previous_tickets_sold=(
                prev.tickets_sold if prev and prev.tickets_sold is not None
                else _sales_tickets_sold(sales)
            ),
        )

        # Stage 11: vps_deployment
        await run_repo.set_stage(run_id, STAGE_VPS_DEPLOYMENT, "running")
        await brief_repo.set_deploying(brief_id, public_url)
        from app.html.deploy import deploy_brief_html
        await deploy_brief_html(slug=slug, html=html)
        await run_repo.set_stage(run_id, STAGE_VPS_DEPLOYMENT, "ok", public_url=public_url)

        # Stage 12: url_verification
        from app.html.verify import verify_public_url
        await run_repo.set_stage(run_id, STAGE_URL_VERIFICATION, "running")
        verification_base = (
            settings.INTERNAL_PUBLIC_BASE_URL.rstrip("/")
            if settings.INTERNAL_PUBLIC_BASE_URL
            else settings.PUBLIC_BASE_URL.rstrip("/")
        )
        verification_url = f"{verification_base}/{slug}"
        verify_public_url(verification_url)
        # (verify_public_url raises on failure, aborting the brief to "failed")
        await run_repo.set_stage(run_id, STAGE_URL_VERIFICATION, "ok")

        # Stage 13: og_verification
        await run_repo.set_stage(run_id, STAGE_OG_VERIFICATION, "running")
        internal_og_image = settings.INTERNAL_OG_COVER_URL or None
        verify_public_url(
            verification_url,
            require_og=True,
            og_image_url=internal_og_image,
        )
        await run_repo.set_stage(run_id, STAGE_OG_VERIFICATION, "ok")

        return ctx


# ── Helpers ───────────────────────────────────────────────────────────

def _to_show(row: dict[str, Any]) -> Show:
    raw_date = row.get("show_date")
    if isinstance(raw_date, str):
        raw_date = date.fromisoformat(raw_date[:10])
    elif isinstance(raw_date, datetime):
        raw_date = raw_date.date()
    return Show(
        show_id=str(row.get("show_id") or row.get("show_key") or row.get("id") or ""),
        show_date=raw_date,
        city=str(row.get("city") or row.get("city_name") or ""),
        country=str(row.get("country") or row.get("country_name") or ""),
        venue=row.get("venue"),
        capacity=as_int(row.get("capacity")),
        ticket_url=row.get("ticket_url") or row.get("tm_main") or row.get("sale_url"),
    )


def _to_past_show(row: dict[str, Any]) -> PastShow:
    raw_date = row.get("show_date")
    if isinstance(raw_date, str):
        raw_date = date.fromisoformat(raw_date[:10])
    elif isinstance(raw_date, datetime):
        raw_date = raw_date.date()
    return PastShow(
        show_id=str(row.get("show_id") or row.get("show_key") or ""),
        show_date=raw_date,
        city=str(row.get("city") or ""),
        country=str(row.get("country") or row.get("country_name") or ""),
        venue=row.get("venue"),
        capacity=as_int(row.get("capacity")),
        tickets_sold=as_int(row.get("tickets_sold")),
        sold_pct=as_float(row.get("sold_pct")),
    )


def _previous_show_summary(prev: PastShow | None, sales: dict[str, Any]) -> str | None:
    if not prev:
        return None
    parts = [f"{prev.city} {prev.show_date.isoformat()}", f"{prev.venue or '<venue unknown>'}"]
    if prev.tickets_sold is not None:
        parts.append(f"{prev.tickets_sold} sold")
    if prev.sold_pct is not None:
        parts.append(f"{prev.sold_pct:.0%} sold")
    elif (
        sales.get("tickets_sold") is not None
        or sales.get("total_sold") is not None
        or sales.get("total") is not None
    ):
        sold = _sales_tickets_sold(sales)
        parts.append(f"{sold} sold")
    return "; ".join(parts)


def _sales_tickets_sold(sales: dict[str, Any]) -> int | None:
    for key in ("tickets_sold", "total_sold", "total"):
        value = as_int(sales.get(key))
        if value is not None:
            return value
    return None


def _truncate_err(msg: str, limit: int = 1000) -> str:
    """Redact + cap an error string for the city_briefs_runs error column."""
    return redact_for_storage(RuntimeError(msg))[:limit]