"""
从全部 S/A 级需求及其拓展点位触发 find_agent 视频发现；有效视频满 450 提前结束。

任务层负责查库与组装上下文；Agent 负责搜索、画像与分池落库。

调度模型：
1. 当天全部待执行需求先入内存任务队列（不预建 DB 记录）；
2. 维护最多 N 个 worker 线程循环取任务；
3. 执行前检查当日通过视频数，达标后停止领取新任务（允许略超）；
4. 单任务超时（默认 600s）后该 worker 退出，监督线程自动补齐到 N。
"""
from __future__ import annotations

import logging
import queue
import threading
import uuid
from concurrent.futures import (
    FIRST_COMPLETED,
    ThreadPoolExecutor,
    TimeoutError as FuturesTimeoutError,
    wait,
)
from datetime import datetime
from time import monotonic
from typing import Any
from zoneinfo import ZoneInfo

from agents.find_agent.demand_run import (
    FindDemandContext,
    discover_videos_for_demand,
    filter_pending_contexts,
    list_find_demand_contexts,
    serialize_find_demand_context,
)
from agents.find_agent.runtime import find_agent_timeout_seconds
from supply_infra.config import get_infra_settings
from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
from supply_infra.db.session import ensure_mysql_pool_capacity, get_session
from supply_infra.services.video_discovery_service import get_video_discovery_service

logger = logging.getLogger(__name__)

_DEFAULT_WORKERS = 5
_DAILY_PASSED_VIDEO_LIMIT = 450
_SUPERVISOR_POLL_SECONDS = 0.5
_QUEUE_GET_TIMEOUT_SECONDS = 1.0


def _resolve_biz_dt(biz_dt: str | None) -> str:
    if biz_dt:
        text = str(biz_dt).strip()
        if len(text) == 8 and text.isdigit():
            return text
        raise ValueError(f"biz_dt 格式无效，应为 YYYYMMDD: {biz_dt!r}")

    with get_session() as session:
        latest = DemandGradeRepository(session).get_latest_biz_dt()
    if latest:
        return str(latest)

    timezone = ZoneInfo(get_infra_settings().scheduler_timezone)
    return datetime.now(timezone).strftime("%Y%m%d")


def process_single_discover(
    ctx: FindDemandContext,
    *,
    force: bool = False,
) -> dict[str, Any]:
    """并发 worker：对单条需求记录执行 find_agent。

    video_discovery_run 记录在 discover_videos_for_demand 开始执行时创建。
    """
    summary = serialize_find_demand_context(ctx)
    logger.info(
        "discover videos started: grade_id=%s demand=%s videos=%d points=%d",
        ctx.demand_grade_id,
        ctx.demand_name,
        ctx.video_count,
        ctx.point_count,
    )
    try:
        execution = discover_videos_for_demand(ctx, force=force)
        if execution.skipped:
            return {
                "success": True,
                "skipped": True,
                **summary,
                "skip_reason": execution.skip_reason,
            }

        if not execution.succeeded:
            return {
                "success": False,
                "skipped": False,
                "run_id": execution.run_id,
                **summary,
                "error": execution.failure_reason or "find_agent_incomplete",
                "iterations": (
                    execution.agent_result.iterations
                    if execution.agent_result is not None
                    else None
                ),
            }

        agent_result = execution.agent_result
        assert agent_result is not None
        logger.info(
            "discover videos done: run_id=%s grade_id=%s demand=%s videos=%d points=%d iterations=%d",
            execution.run_id,
            ctx.demand_grade_id,
            ctx.demand_name,
            ctx.video_count,
            ctx.point_count,
            agent_result.iterations,
        )
        return {
            "success": True,
            "skipped": False,
            "run_id": execution.run_id,
            **summary,
            "business_outcome": execution.business_outcome,
            "goal_met": execution.goal_met,
            "valid_primary_count": execution.valid_primary_count,
            "iterations": agent_result.iterations,
            "content_chars": len(agent_result.content or ""),
        }
    except Exception as exc:
        error_text = str(exc)
        logger.exception(
            "discover videos failed: grade_id=%s demand=%s videos=%d",
            ctx.demand_grade_id,
            ctx.demand_name,
            ctx.video_count,
        )
        return {
            "success": False,
            "skipped": False,
            **summary,
            "error": error_text,
        }


def _count_passed_videos(biz_dt: str) -> int:
    return get_video_discovery_service().count_passed_videos(biz_dt)


def _apply_item_result(result: dict[str, Any], item_result: dict[str, Any]) -> None:
    result["processed"] += 1
    if item_result.get("skipped"):
        result["skipped"] += 1
        return
    if item_result.get("success"):
        result["succeeded"] += 1
        business_outcome = item_result.get("business_outcome")
        if business_outcome in {"goal_met", "partial", "no_match"}:
            result[str(business_outcome)] += 1
        return

    result["failed"] += 1
    result["errors"].append(
        {
            "demand_grade_id": item_result.get("demand_grade_id"),
            "demand_name": item_result.get("demand_name"),
            "video_count": item_result.get("video_count"),
            "run_id": item_result.get("run_id"),
            "error": item_result.get("error"),
        }
    )


def _run_task_with_timeout(
    ctx: FindDemandContext,
    *,
    force: bool,
    timeout_seconds: float,
) -> dict[str, Any]:
    """在独立线程中执行单任务，超时后向上抛出 FuturesTimeoutError。

    超时后不阻塞等待底层线程（shutdown wait=False），以便 worker 能退出并由监督方补齐。
    """
    executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="find-task")
    try:
        future = executor.submit(process_single_discover, ctx, force=force)
        return future.result(timeout=timeout_seconds)
    finally:
        executor.shutdown(wait=False, cancel_futures=True)


def _drain_task_queue(task_queue: queue.Queue[FindDemandContext]) -> int:
    discarded = 0
    while True:
        try:
            task_queue.get_nowait()
        except queue.Empty:
            break
        discarded += 1
        task_queue.task_done()
    return discarded


def _run_worker_pool(
    *,
    contexts: list[FindDemandContext],
    biz_dt: str,
    worker_count: int,
    force: bool,
    result: dict[str, Any],
    task_timeout_seconds: float,
) -> None:
    """任务队列 + 固定规模 worker 池；超时退出后自动补齐线程。"""
    task_queue: queue.Queue[FindDemandContext] = queue.Queue()
    for ctx in contexts:
        task_queue.put(ctx)

    stop_event = threading.Event()
    stats_lock = threading.Lock()
    worker_seq = 0
    started_at = monotonic()

    def worker_loop(worker_id: int) -> str:
        """循环取任务；单任务超时则退出，由监督方补线程。"""
        logger.info(
            "discover worker started: worker_id=%s biz_dt=%s timeout_seconds=%.0f",
            worker_id,
            biz_dt,
            task_timeout_seconds,
        )
        exit_reason = "queue_empty"
        while not stop_event.is_set():
            try:
                ctx = task_queue.get(timeout=_QUEUE_GET_TIMEOUT_SECONDS)
            except queue.Empty:
                if stop_event.is_set() or task_queue.empty():
                    exit_reason = "queue_empty"
                    break
                continue

            try:
                if stop_event.is_set():
                    exit_reason = "stopped"
                    break

                try:
                    passed_videos = _count_passed_videos(biz_dt)
                except Exception:
                    logger.exception(
                        "discover worker count_passed_videos failed: worker_id=%s",
                        worker_id,
                    )
                    with stats_lock:
                        result["failed"] += 1
                        result["processed"] += 1
                        result["errors"].append(
                            {
                                "demand_grade_id": ctx.demand_grade_id,
                                "demand_name": ctx.demand_name,
                                "error": "count_passed_videos_failed",
                            }
                        )
                    continue

                with stats_lock:
                    result["passed_videos"] = passed_videos

                if passed_videos >= _DAILY_PASSED_VIDEO_LIMIT:
                    with stats_lock:
                        result["stopped_by_passed_video_limit"] = True
                    stop_event.set()
                    exit_reason = "passed_video_limit"
                    logger.info(
                        "discover worker stop by daily limit before execute: "
                        "worker_id=%s grade_id=%s demand=%s passed_videos=%s",
                        worker_id,
                        ctx.demand_grade_id,
                        ctx.demand_name,
                        passed_videos,
                    )
                    break

                logger.info(
                    "discover worker pickup: worker_id=%s grade_id=%s demand=%s "
                    "queue_remaining~=%s passed_videos=%s",
                    worker_id,
                    ctx.demand_grade_id,
                    ctx.demand_name,
                    task_queue.qsize(),
                    passed_videos,
                )

                try:
                    item_result = _run_task_with_timeout(
                        ctx,
                        force=force,
                        timeout_seconds=task_timeout_seconds,
                    )
                except FuturesTimeoutError:
                    error_text = (
                        f"find_agent worker timed out after {task_timeout_seconds:.0f}s"
                    )
                    logger.error(
                        "discover worker task timeout, worker will exit: "
                        "worker_id=%s grade_id=%s demand=%s timeout_seconds=%.0f",
                        worker_id,
                        ctx.demand_grade_id,
                        ctx.demand_name,
                        task_timeout_seconds,
                    )
                    with stats_lock:
                        result["failed"] += 1
                        result["processed"] += 1
                        result["errors"].append(
                            {
                                "demand_grade_id": ctx.demand_grade_id,
                                "demand_name": ctx.demand_name,
                                "video_count": ctx.video_count,
                                "error": error_text,
                            }
                        )
                    exit_reason = "task_timeout"
                    break
                except Exception as exc:
                    logger.exception(
                        "discover worker unexpected error: worker_id=%s grade_id=%s demand=%s",
                        worker_id,
                        ctx.demand_grade_id,
                        ctx.demand_name,
                    )
                    with stats_lock:
                        result["failed"] += 1
                        result["processed"] += 1
                        result["errors"].append(
                            {
                                "demand_grade_id": ctx.demand_grade_id,
                                "demand_name": ctx.demand_name,
                                "error": str(exc),
                            }
                        )
                    continue

                with stats_lock:
                    _apply_item_result(result, item_result)
                    logger.info(
                        "discover worker item finished: worker_id=%s grade_id=%s demand=%s "
                        "success=%s skipped=%s total_processed=%d",
                        worker_id,
                        ctx.demand_grade_id,
                        ctx.demand_name,
                        item_result.get("success"),
                        item_result.get("skipped"),
                        result["processed"],
                    )

                try:
                    passed_videos = _count_passed_videos(biz_dt)
                except Exception:
                    logger.exception(
                        "discover worker post-count failed: worker_id=%s",
                        worker_id,
                    )
                    continue

                with stats_lock:
                    result["passed_videos"] = passed_videos
                    if passed_videos >= _DAILY_PASSED_VIDEO_LIMIT:
                        result["stopped_by_passed_video_limit"] = True
                        stop_event.set()
                        exit_reason = "passed_video_limit"
                        break
            finally:
                task_queue.task_done()

        logger.info(
            "discover worker exited: worker_id=%s reason=%s elapsed_seconds=%.1f",
            worker_id,
            exit_reason,
            monotonic() - started_at,
        )
        return exit_reason

    with ThreadPoolExecutor(
        max_workers=worker_count,
        thread_name_prefix="find-agent",
    ) as executor:
        futures: set[Any] = set()

        def _spawn_worker() -> None:
            nonlocal worker_seq
            worker_seq += 1
            futures.add(executor.submit(worker_loop, worker_seq))

        for _ in range(worker_count):
            _spawn_worker()

        while futures:
            done, futures = wait(
                futures,
                timeout=_SUPERVISOR_POLL_SECONDS,
                return_when=FIRST_COMPLETED,
            )
            if not done:
                alive = len(futures)
                logger.info(
                    "discover workers heartbeat: biz_dt=%s alive=%d/%d "
                    "processed=%d queue_remaining~=%s passed_videos=%s "
                    "elapsed_seconds=%d",
                    biz_dt,
                    alive,
                    worker_count,
                    result["processed"],
                    task_queue.qsize(),
                    result.get("passed_videos"),
                    int(monotonic() - started_at),
                )
                # 超时/异常退出后补齐到目标线程数
                if (
                    not stop_event.is_set()
                    and not task_queue.empty()
                    and alive < worker_count
                ):
                    for _ in range(worker_count - alive):
                        if task_queue.empty():
                            break
                        logger.info(
                            "discover worker replenish: biz_dt=%s alive=%d target=%d",
                            biz_dt,
                            alive,
                            worker_count,
                        )
                        _spawn_worker()
                        alive += 1
                continue

            for future in done:
                try:
                    exit_reason = future.result()
                except Exception:
                    logger.exception(
                        "discover worker future crashed: biz_dt=%s",
                        biz_dt,
                    )
                    exit_reason = "crashed"

                if (
                    not stop_event.is_set()
                    and not task_queue.empty()
                    and len(futures) < worker_count
                ):
                    logger.info(
                        "discover worker replenish after exit: biz_dt=%s "
                        "exit_reason=%s alive=%d target=%d",
                        biz_dt,
                        exit_reason,
                        len(futures),
                        worker_count,
                    )
                    _spawn_worker()

        if stop_event.is_set():
            discarded = _drain_task_queue(task_queue)
            if discarded:
                logger.info(
                    "discover discarded pending tasks after stop: biz_dt=%s discarded=%d",
                    biz_dt,
                    discarded,
                )


def discover_videos_from_demands(
    biz_dt: str | None = None,
    *,
    workers: int = _DEFAULT_WORKERS,
    offset: int = 0,
    limit: int | None = None,
    top_limit: int | None = None,
    skip_finished: bool = True,
    force: bool = False,
) -> dict[str, Any]:
    """
    对指定业务日全部 S/A 级需求（有拓展点位）以任务队列方式并发调用 find_agent。

    - 先把当天待执行需求全部放入内存队列（此时不写 video_discovery_run）
    - 固定 worker 线程循环取任务；单任务开始执行时再创建 run 记录
    - 执行前检查当日 primary 通过数，达标后不再领取新任务（允许略超）
    - 单任务默认 600s 超时后该 worker 退出，监督方自动补齐线程
    """
    started_at = datetime.now()
    batch_run_id = uuid.uuid4().hex
    task_timeout_seconds = find_agent_timeout_seconds()

    try:
        resolved_biz_dt, contexts = list_find_demand_contexts(
            biz_dt,
            top_limit=top_limit,
        )
    except Exception as exc:
        logger.exception("discover_videos_from_demands preflight failed")
        return {
            "success": False,
            "error": str(exc),
            "started_at": started_at.isoformat(),
            "finished_at": datetime.now().isoformat(),
        }

    preload_stats = {"total_loaded": len(contexts), "skipped_already_done": 0}
    if not force:
        contexts, preload_stats = filter_pending_contexts(
            contexts,
            resolved_biz_dt,
            skip_finished=skip_finished,
        )

    if offset > 0:
        contexts = contexts[int(offset) :]

    if limit is not None and limit >= 0:
        contexts = contexts[: int(limit)]

    passed_videos = _count_passed_videos(resolved_biz_dt)

    logger.info(
        "discover_videos_from_demands start: biz_dt=%s batch_run_id=%s workers=%s "
        "top_limit=%s offset=%s pending=%s skipped=%s passed_videos=%s "
        "passed_video_limit=%s task_timeout_seconds=%.0f",
        resolved_biz_dt,
        batch_run_id,
        workers,
        top_limit,
        offset,
        len(contexts),
        preload_stats.get("skipped_already_done", 0),
        passed_videos,
        _DAILY_PASSED_VIDEO_LIMIT,
        task_timeout_seconds,
    )

    result: dict[str, Any] = {
        "success": True,
        "batch_run_id": batch_run_id,
        "biz_dt": resolved_biz_dt,
        "started_at": started_at.isoformat(),
        "offset": int(offset),
        "top_limit": top_limit,
        **preload_stats,
        "total_records": len(contexts),
        "workers": 0,
        "task_timeout_seconds": task_timeout_seconds,
        "processed": 0,
        "succeeded": 0,
        "goal_met": 0,
        "partial": 0,
        "no_match": 0,
        "skipped": 0,
        "failed": 0,
        "errors": [],
        "passed_videos": passed_videos,
        "passed_video_limit": _DAILY_PASSED_VIDEO_LIMIT,
        "stopped_by_passed_video_limit": passed_videos
        >= _DAILY_PASSED_VIDEO_LIMIT,
    }

    if not contexts or passed_videos >= _DAILY_PASSED_VIDEO_LIMIT:
        finished_at = datetime.now()
        result["finished_at"] = finished_at.isoformat()
        result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
        logger.info("discover_videos_from_demands finished: %s", result)
        return result

    worker_count = max(1, min(int(workers), len(contexts)))
    ensure_mysql_pool_capacity(worker_count)
    result["workers"] = worker_count

    _run_worker_pool(
        contexts=contexts,
        biz_dt=resolved_biz_dt,
        worker_count=worker_count,
        force=force,
        result=result,
        task_timeout_seconds=task_timeout_seconds,
    )

    # 最终再读一次，避免结果里的 passed_videos 过旧
    try:
        result["passed_videos"] = _count_passed_videos(resolved_biz_dt)
        if result["passed_videos"] >= _DAILY_PASSED_VIDEO_LIMIT:
            result["stopped_by_passed_video_limit"] = True
    except Exception:
        logger.exception("discover_videos_from_demands final count_passed_videos failed")

    finished_at = datetime.now()
    result["finished_at"] = finished_at.isoformat()
    result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
    result["success"] = result["failed"] == 0

    logger.info("discover_videos_from_demands finished: %s", result)
    return result


if __name__ == "__main__":
    import sys

    from supply_infra.scheduler.cli_result import run_cli

    def _read_flag_value(args: list[str], flag: str) -> int | None:
        if flag not in args:
            return None
        idx = args.index(flag)
        if idx + 1 >= len(args):
            return None
        return int(args[idx + 1])

    _args = sys.argv[1:]
    _biz_dt = _args[0] if _args and not _args[0].startswith("-") else None

    run_cli(
        lambda: discover_videos_from_demands(
            _biz_dt,
            workers=_read_flag_value(_args, "--workers") or _DEFAULT_WORKERS,
            offset=_read_flag_value(_args, "--offset") or 0,
            limit=_read_flag_value(_args, "--limit"),
            top_limit=_read_flag_value(_args, "--top-limit"),
            skip_finished="--force" not in _args,
            force="--force" in _args,
        ),
        label="discover_videos_from_demands",
    )
