"""短剧采集服务：从内容源拉取短剧列表/详情，导入为 series + videos（幂等）"""
import asyncio
import logging
from datetime import datetime
from urllib.parse import quote

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.video import CollectionRecord, CollectionSource, Series, Video

logger = logging.getLogger(__name__)

# 内容源配置（域名轮换时只需改这里）
SOURCE_KEY = "mdcmai4"
SOURCE_CODE = "madou"     # sources 表中的 code，用于给导入内容打「来源」标
UPSTREAM = "https://mdcmai4.xyz"
PRODUCT_ID = 1
CATEGORY_ID = 12          # 本站分类：AI短剧
AUTHOR_USER_ID = 1        # 导入内容归属用户
SELF_API_BASE = "https://vapi.256kb.top"  # 本站 API 对外地址

HEADERS = {"User-Agent": "Mozilla/5.0 (Linux; Android 13) AppleWebKit/537.36"}


async def _fetch_json(client: httpx.AsyncClient, url: str, params: dict | None = None) -> dict:
    resp = await client.get(url, params=params, headers=HEADERS, timeout=20)
    resp.raise_for_status()
    body = resp.json()
    if body.get("code") != 200 or not isinstance(body.get("data"), dict):
        raise RuntimeError(f"上游响应异常: {url} -> {str(body)[:200]}")
    return body["data"]


def _proxy_url(path: str) -> str:
    """把源站相对 m3u8 路径转为本站代理地址（域名轮换只改 SELF_API_BASE/上游常量）"""
    return f"{SELF_API_BASE}/api/v1/m3u8/proxy?path={quote(path, safe='')}"


async def default_source_id(db: AsyncSession) -> int:
    """麻豆等内容源的来源 id（sources.code 匹配；缺失时取第一个启用来源）"""
    from app.models.video import Source

    sid = (
        await db.execute(select(Source.id).where(Source.code == SOURCE_CODE))
    ).scalar_one_or_none()
    if sid:
        return sid
    sid = (
        await db.execute(select(Source.id).where(Source.status == 1).order_by(Source.sort, Source.id))
    ).scalars().first()
    return sid or 0


async def _import_drama(client: httpx.AsyncClient, db: AsyncSession, drama_id: int) -> tuple[int, int]:
    """导入单部短剧，返回 (新增集数, 跳过集数)"""
    detail = await _fetch_json(client, f"{UPSTREAM}/api/v1/short-dramas/{drama_id}", {"productId": PRODUCT_ID})

    record = (
        await db.execute(
            select(CollectionRecord).where(
                CollectionRecord.source_key == SOURCE_KEY, CollectionRecord.source_id == drama_id
            )
        )
    ).scalar_one_or_none()

    published_at = None
    try:
        published_at = datetime.fromisoformat(detail.get("publishedAt", "")) or None
    except ValueError:
        pass

    series = await db.get(Series, record.series_id) if record else None
    if series is None:
        series = Series(
            title=(detail.get("title") or f"短剧{drama_id}")[:100],
            cover_url=f"{UPSTREAM}{detail.get('coverUrl', '')}",
            description=(detail.get("description") or "")[:500],
            author_id=AUTHOR_USER_ID,
            category_id=CATEGORY_ID,
            source_id=await default_source_id(db),
            total_episodes=0,
            view_count=int(detail.get("heatCount") or 0),
            status=1,
        )
        db.add(series)
        await db.flush()
        if record:  # 系列曾被删除：重挂映射
            record.series_id = series.id
        else:
            db.add(CollectionRecord(source_key=SOURCE_KEY, source_id=drama_id, series_id=series.id))
        await db.flush()

    # 已有集数去重
    existing = set(
        (
            await db.execute(
                select(Video.episode_no).where(Video.series_id == series.id, Video.episode_no.is_not(None))
            )
        ).scalars().all()
    )

    created = skipped = 0
    for ep in detail.get("episodes", []):
        ep_no = int(ep.get("episodeNo") or 0)
        if not ep_no or ep_no in existing:
            skipped += 1
            continue
        db.add(
            Video(
                user_id=AUTHOR_USER_ID,
                category_id=CATEGORY_ID,
                source_id=series.source_id,
                series_id=series.id,
                episode_no=ep_no,
                title=(ep.get("titleOverride") or ep.get("title") or f"第{ep_no}集")[:100],
                description=(detail.get("description") or "")[:500],
                cover_url=f"{UPSTREAM}{detail.get('coverUrl', '')}",
                video_url=_proxy_url(ep.get("videoUrl", "")),
                duration=int(ep.get("durationSec") or 0),
                status=2,
                published_at=published_at,
            )
        )
        existing.add(ep_no)
        created += 1

    series.total_episodes = len(existing)
    return created, skipped


async def sync_drama_by_id(db: AsyncSession, drama_id: int) -> dict:
    """手动同步单部短剧（新增或补充缺失集数）；返回统计"""
    async with httpx.AsyncClient(follow_redirects=True) as client:
        created, skipped = await _import_drama(client, db, drama_id)
        await db.commit()
    return {"drama_id": drama_id, "episodes_created": created, "episodes_skipped": skipped}


async def sync_short_dramas(
    db: AsyncSession,
    max_pages: int = 0,
    page_size: int = 20,
    sort_by: str = "heat",
    delay: float = 0.25,
) -> dict:
    """按列表页增量同步短剧；返回统计"""
    stat = {"dramas_new": 0, "dramas_exist": 0, "episodes_created": 0, "episodes_skipped": 0, "pages": 0}
    async with httpx.AsyncClient(follow_redirects=True) as client:
        page = 1
        while True:
            data = await _fetch_json(
                client,
                f"{UPSTREAM}/api/v1/short-dramas",
                {"productId": PRODUCT_ID, "sortBy": sort_by, "page": page, "size": page_size},
            )
            items = data.get("items", [])
            stat["pages"] = page
            logger.info("列表第 %s/%s 页，%s 部", page, data.get("totalPages"), len(items))

            for item in items:
                drama_id = int(item.get("id") or 0)
                if not drama_id:
                    continue
                exists = (
                    await db.execute(
                        select(CollectionRecord.id).where(
                            CollectionRecord.source_key == SOURCE_KEY, CollectionRecord.source_id == drama_id
                        )
                    )
                ).scalar_one_or_none()
                if exists:
                    stat["dramas_exist"] += 1
                    continue
                try:
                    created, skipped = await _import_drama(client, db, drama_id)
                    await db.commit()
                    stat["dramas_new"] += 1
                    stat["episodes_created"] += created
                    stat["episodes_skipped"] += skipped
                    logger.info("导入短剧 %s《%s》：新增 %s 集", drama_id, item.get("title"), created)
                except Exception:
                    await db.rollback()
                    logger.exception("导入短剧 %s 失败，跳过", drama_id)
                await asyncio.sleep(delay)

            total_pages = int(data.get("totalPages") or 1)
            if page >= total_pages or (max_pages and page >= max_pages):
                break
            page += 1
            await asyncio.sleep(delay)
    return stat


# ---------- 单视频采集（通用 URL 模板，后台自助配置的采集源）----------

def _abs_cover(item: dict) -> str:
    """封面转绝对地址：源站 /api/... 代理路径或 /uploads/... 静态路径"""
    u = item.get("coverUrl") or ""
    if u.startswith("http"):
        return u
    return f"{UPSTREAM}{u}" if u else ""


async def _import_video_item(db: AsyncSession, source: CollectionSource, item: dict) -> bool:
    """导入单条视频，已存在返回 False"""
    vid = int(item.get("id") or 0)
    if not vid or not (item.get("videoUrl") or "").endswith(".m3u8"):
        return False
    exists = (
        await db.execute(
            select(CollectionRecord.id).where(
                CollectionRecord.source_key == SOURCE_KEY, CollectionRecord.source_id == vid
            )
        )
    ).scalar_one_or_none()
    if exists:
        return False

    published_at = None
    try:
        published_at = datetime.fromisoformat(item.get("publishedAt", "")) or None
    except ValueError:
        pass

    v = Video(
        user_id=AUTHOR_USER_ID,
        category_id=source.category_id,
        source_id=source.source_id or await default_source_id(db),
        title=(item.get("title") or f"视频{vid}")[:100],
        description=(item.get("description") or "")[:500],
        cover_url=_abs_cover(item)[:500],
        video_url=_proxy_url(item["videoUrl"]),
        duration=int(item.get("durationSec") or 0),
        status=2,
        view_count=int(item.get("viewCount") or 0),
        like_count=int(item.get("likeCount") or 0),
        favorite_count=int(item.get("favoriteCount") or 0),
        published_at=published_at,
    )
    db.add(v)
    await db.flush()
    db.add(CollectionRecord(source_key=SOURCE_KEY, source_id=vid, video_id=v.id))
    return True


async def sync_source(
    db: AsyncSession,
    source: CollectionSource,
    max_pages: int = 0,
    stop_after_existing: int = 30,
    delay: float = 0.25,
    should_cancel=None,
) -> dict:
    """按源的 URL 模板翻页采集；连续 stop_after_existing 条已存在则提前停止（增量）。
    should_cancel 返回 True 时每页/每条边界退出（用于手动取消）"""
    stat = {"created": 0, "skipped": 0, "pages": 0, "stopped_early": False, "cancelled": False}
    consecutive_existing = 0

    def _cancelled() -> bool:
        return bool(should_cancel and should_cancel())

    async with httpx.AsyncClient(follow_redirects=True) as client:
        page = 1
        while True:
            if _cancelled():
                stat["cancelled"] = True
                logger.info("通用采集源 %s 同步被取消（第 %s 页前）", source.id, page)
                return stat
            url = source.list_url.replace("{{page}}", str(page))
            data = await _fetch_json(client, url)
            items = data.get("items", [])
            if not items:
                break
            stat["pages"] = page
            for item in items:
                if _cancelled():
                    stat["cancelled"] = True
                    logger.info("通用采集源 %s 同步被取消（处理 item 中）", source.id)
                    return stat
                if await _import_video_item(db, source, item):
                    stat["created"] += 1
                    consecutive_existing = 0
                else:
                    stat["skipped"] += 1
                    consecutive_existing += 1
                if consecutive_existing >= stop_after_existing:
                    stat["stopped_early"] = True
                    break
            await db.commit()
            if stat["stopped_early"]:
                break
            total_pages = int(data.get("totalPages") or page)
            if page >= total_pages or (max_pages and page >= max_pages):
                break
            page += 1
            await asyncio.sleep(delay)

    source.last_sync_at = datetime.now()
    await db.commit()
    return stat