"""S3 storage backend — mirrors LocalBackend method-for-method, objects instead of files.

All boto3 calls are blocking; each is wrapped in asyncio.to_thread to avoid
stalling the event loop when multiple shards run concurrently.
"""

from __future__ import annotations

import asyncio
import json
from datetime import UTC, datetime
from typing import Any

import boto3
import structlog
from botocore.exceptions import ClientError

from marketing_intelligence.core.config import settings
from marketing_intelligence.core.models import (
    ArtistSnapshotRecord,
    CampaignConfigRecord,
    CampaignDiscoveryRunReasoning,
    CampaignSentiment,
    PostMetrics,
    RunRecord,
    SoundRecord,
)
from marketing_intelligence.storage.backend import StorageBackend

logger = structlog.get_logger("storage.s3")

_NON_VIDEO_SUFFIXES = ("/manifest.json", "/request.json")


def _is_not_found(e: ClientError) -> bool:
    return e.response.get("Error", {}).get("Code") in ("NoSuchKey", "404")


class S3Backend(StorageBackend):
    def __init__(self) -> None:
        if not settings.s3_bucket:
            raise ValueError("STORAGE_BACKEND=s3 requires S3_BUCKET to be set")
        self._bucket = settings.s3_bucket
        self._client = boto3.client("s3", region_name=settings.aws_region)

    async def _get_json(self, key: str) -> dict[str, Any] | None:
        def _get() -> dict[str, Any] | None:
            try:
                obj = self._client.get_object(Bucket=self._bucket, Key=key)
                return dict(json.loads(obj["Body"].read()))
            except ClientError as e:
                if _is_not_found(e):
                    return None
                raise

        return await asyncio.to_thread(_get)

    async def _put_json(self, key: str, data: dict[str, Any]) -> str:
        body = json.dumps(data, indent=2, ensure_ascii=False, default=str).encode(
            "utf-8"
        )
        await asyncio.to_thread(
            self._client.put_object,
            Bucket=self._bucket,
            Key=key,
            Body=body,
            ContentType="application/json",
        )
        return f"s3://{self._bucket}/{key}"

    async def _list_keys(self, prefix: str) -> list[str]:
        def _list() -> list[str]:
            keys: list[str] = []
            paginator = self._client.get_paginator("list_objects_v2")
            for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix):
                keys.extend(obj["Key"] for obj in page.get("Contents", []))
            return keys

        return await asyncio.to_thread(_list)

    async def write_run_record(self, record: RunRecord) -> str | None:
        return await self._put_json(
            f"runs/{record.run_id}.json", record.model_dump(mode="json")
        )

    async def write_campaign_post(
        self,
        config: CampaignConfigRecord,
        sound: SoundRecord | None,
        campaign_id: str,
        run_id: str,
    ) -> str | None:
        data: dict[str, Any] = {"campaign_config": config.model_dump(mode="json")}
        if sound:
            data["sound"] = sound.model_dump(mode="json")
        return await self._put_json(
            f"posts/{campaign_id}/{run_id}/{config.post_id}.json", data
        )

    async def write_post_metrics(
        self,
        items: list[dict[str, Any]],
        campaign_config_key: str,
        run_id: str,
        reasoning: str | None = None,
    ) -> str | None:
        now = datetime.now(UTC).isoformat()
        messages = []
        for item in items:
            post_id = str(item.get("video_id") or item.get("post_id") or "")
            if not post_id:
                continue
            messages.append(
                PostMetrics.from_observation(
                    campaign_config_key=campaign_config_key,
                    post_id=post_id,
                    run_id=run_id,
                    observed_at=item.get("observed_at", now),
                    sound_id=item.get("sound_id"),
                    views=item.get("views"),
                    likes=item.get("likes"),
                    comments=item.get("comments"),
                    shares=item.get("shares"),
                    favorites=item.get("favorites"),
                    prev=item.get("_prev"),
                ).model_dump()
            )

        if not messages:
            return None

        if reasoning:
            result = CampaignDiscoveryRunReasoning(
                campaign_config_key=campaign_config_key,
                run_id=run_id,
                count=len(messages),
                reasoning=reasoning,
                items=items,
            )
            await self._put_json(
                f"{campaign_config_key}/{run_id}/result.json",
                result.model_dump(exclude_none=True),
            )

        return await self._put_json(
            f"metrics/{campaign_config_key}/{run_id}.json",
            {
                "campaign_config_key": campaign_config_key,
                "run_id": run_id,
                "items": messages,
            },
        )

    async def write_artist_snapshot(
        self, record: ArtistSnapshotRecord, campaign_id: str
    ) -> str | None:
        return await self._put_json(
            f"artist_snapshots/{campaign_id}/{record.run_id}.json",
            record.model_dump(mode="json"),
        )

    async def write_sentiment(self, record: CampaignSentiment) -> str | None:
        return await self._put_json(
            f"sentiment/{record.campaign_config_key}/{record.run_id}.json",
            record.model_dump(mode="json"),
        )

    async def read_sentiment(
        self, campaign_config_key: str, run_id: str
    ) -> dict[str, Any] | None:
        return await self._get_json(f"sentiment/{campaign_config_key}/{run_id}.json")

    async def write_pdf_report(
        self, campaign_config_key: str, run_id: str, local_pdf_path: str
    ) -> str | None:
        key = f"sentiment/{campaign_config_key}/{run_id}.pdf"

        def _upload() -> None:
            self._client.upload_file(local_pdf_path, self._bucket, key)

        await asyncio.to_thread(_upload)
        return f"s3://{self._bucket}/{key}"

    async def read_latest_metrics(
        self, campaign_config_key: str
    ) -> dict[str, dict[str, Any]]:
        data = await self._get_json(f"snapshots/{campaign_config_key}/latest.json")
        return data or {}

    async def write_latest_metrics(
        self, campaign_config_key: str, metrics: dict[str, Any]
    ) -> None:
        await self._put_json(f"snapshots/{campaign_config_key}/latest.json", metrics)

    async def write_video_comments(
        self, campaign_id: str, run_id: str, video_id: str, data: dict[str, Any]
    ) -> str | None:
        return await self._put_json(
            f"comments/{campaign_id}/{run_id}/{video_id}.json", data
        )

    async def update_video_comments_sentiment(
        self,
        campaign_id: str,
        run_id: str,
        video_id: str,
        sentiment: str | None,
        confidence: float | None,
        summary: str | None,
        key_themes: list[str] | None,
    ) -> bool:
        key = f"comments/{campaign_id}/{run_id}/{video_id}.json"
        data = await self._get_json(key)
        if data is None:
            return False
        data.update(
            {
                "sentiment": sentiment,
                "confidence": confidence,
                "summary": summary,
                "key_themes": key_themes or [],
            }
        )
        await self._put_json(key, data)
        return True

    async def read_all_video_comments(
        self, campaign_id: str, run_id: str, limit: int | None = None
    ) -> list[dict[str, Any]]:
        keys = sorted(await self._list_keys(f"comments/{campaign_id}/{run_id}/"))
        results = []
        for key in keys:
            if any(key.endswith(s) for s in _NON_VIDEO_SUFFIXES):
                continue
            d = await self._get_json(key)
            if d is None or d.get("sentiment") is not None:
                continue
            results.append(
                {
                    "video_id": d.get("video_id"),
                    "url": d.get("url"),
                    "track_name": d.get("track_name"),
                    "caption": d.get("caption"),
                    "hashtags": d.get("hashtags", []),
                    "sound_id": d.get("sound_id"),
                    "is_original_sound": d.get("is_original_sound", False),
                    "location": d.get("location"),
                    "video_description": d.get("video_description"),
                    "comment_count": d.get("comment_count", 0),
                    "comment_texts": d.get("comment_texts", []),
                }
            )
            if limit is not None and len(results) >= limit:
                break
        return results

    async def read_all_video_sentiments(
        self, campaign_id: str, run_id: str
    ) -> list[dict[str, Any]]:
        keys = sorted(await self._list_keys(f"comments/{campaign_id}/{run_id}/"))
        results = []
        for key in keys:
            if any(key.endswith(s) for s in _NON_VIDEO_SUFFIXES):
                continue
            d = await self._get_json(key)
            if d is None:
                continue
            results.append(
                {
                    "video_id": d.get("video_id"),
                    "url": d.get("url"),
                    "track_name": d.get("track_name"),
                    "comment_count": d.get("comment_count", 0),
                    "sentiment": d.get("sentiment"),
                    "confidence": d.get("confidence"),
                    "summary": d.get("summary"),
                    "key_themes": d.get("key_themes", []),
                    "video_description": d.get("video_description"),
                }
            )
        return results