import inspect
import json
from concurrent.futures import ThreadPoolExecutor
from collections.abc import Callable
from datetime import UTC, datetime

from sqlalchemy import func, select
from sqlalchemy.orm import Session, sessionmaker

from app.config import get_settings
from app.db import SessionLocal
from app.models import Comment, ContentItem, Hotspot, Report, Task, utc_now
from app.platforms.base import PlatformAPIError, TikHubClient
from app.platforms.douyin import DouyinPlatform
from app.platforms.xiaohongshu import XiaohongshuPlatform
from app.schemas import CreateTaskRequest
from app.services.ai_service import OpenAICompatibleAIClient, analyze_comments_with_retry, build_report_summary_prompt, calculate_analysis_status
from app.services.report_service import generate_hotspot_report, generate_item_report


RUNNING_TASK_MESSAGE = "当前有正在运行的任务，请稍后再试"
RESTART_ERROR_MESSAGE = "系统重启，任务被中断"
STALE_PROGRESS_THRESHOLD_SECONDS = 10 * 60
STALE_PROGRESS_ERROR_TYPE = "stale_progress_timeout"
task_executor = ThreadPoolExecutor(max_workers=1)


STAGE_LABELS = {
    "queued": "等待启动",
    "crawl_hotspots": "获取热点中",
    "search_items": "搜索内容中",
    "crawl_comments": "抓取评论中",
    "ai_analysis": "AI 分析中",
    "generate_reports": "生成报告中",
    "success": "已完成",
    "failed": "失败",
}
PLATFORM_LABELS = {
    "xiaohongshu": "小红书",
    "douyin": "抖音",
}


def has_running_task(session: Session) -> bool:
    return session.scalar(select(Task.id).where(Task.status == "running").limit(1)) is not None


def recover_stale_running_tasks(session: Session) -> int:
    now = ensure_utc_datetime(utc_now())
    tasks = list(session.scalars(select(Task).where(Task.status == "running")))
    recovered = 0
    for task in tasks:
        last_progress_at = ensure_utc_datetime(task.last_progress_at or task.started_at or task.created_at)
        if last_progress_at is None or now is None:
            continue
        seconds_since_progress = max(0, int((now - last_progress_at).total_seconds()))
        if seconds_since_progress < STALE_PROGRESS_THRESHOLD_SECONDS:
            continue
        _mark_task_failed(
            task,
            "system",
            STALE_PROGRESS_ERROR_TYPE,
            f"任务超过 {STALE_PROGRESS_THRESHOLD_SECONDS // 60} 分钟没有进度更新，已自动标记失败",
        )
        task.finished_at = task.finished_at or utc_now()
        recovered += 1
    if recovered:
        session.commit()
    return recovered


def recover_running_tasks(session: Session) -> int:
    tasks = list(session.scalars(select(Task).where(Task.status == "running")))
    for task in tasks:
        task.status = "failed"
        task.error_stage = "system"
        task.error_type = "unexpected_restart"
        task.error_message = RESTART_ERROR_MESSAGE
    session.commit()
    return len(tasks)


def build_platform(platform: str):
    settings = get_settings()
    client = TikHubClient(
        base_url=settings.tikhub_base_url,
        api_key=settings.tikhub_api_key,
        timeout_seconds=settings.http_timeout_seconds,
        max_retries=settings.http_max_retries,
    )
    if platform == "xiaohongshu":
        return XiaohongshuPlatform(client)
    if platform == "douyin":
        return DouyinPlatform(client)
    raise ValueError(f"Unsupported platform: {platform}")


def build_ai_client() -> OpenAICompatibleAIClient | None:
    settings = get_settings()
    if settings.ai_base_url and settings.ai_api_key and settings.ai_model:
        return OpenAICompatibleAIClient(
            base_url=settings.ai_base_url,
            api_key=settings.ai_api_key,
            model=settings.ai_model,
            timeout_seconds=settings.ai_timeout_seconds,
        )
    return None


def build_ai_requester(client: OpenAICompatibleAIClient | None = None):
    client = client if client is not None else build_ai_client()
    if client is not None:
        return lambda prompt: client.request(
            prompt,
            system_prompt="只返回 JSON Array，不输出 Markdown 或解释性自然语言。",
        )

    def fallback_requester(_prompt: str) -> str:
        return "[]"

    return fallback_requester


def build_report_summary_provider(client: OpenAICompatibleAIClient | None):
    if client is None:
        return None

    def summarize_report(metrics: dict, typical: dict, *, word_limit: int = 200) -> str:
        return client.request(
            build_report_summary_prompt(metrics, typical, word_limit=word_limit),
            system_prompt="只返回一段中文总结，不使用 Markdown。",
        )

    return summarize_report


def build_ai_dependencies():
    client = build_ai_client()
    requester = build_ai_requester(client)
    return requester, build_report_summary_provider(client)


def create_task(session: Session, request: CreateTaskRequest, *, submit_background: bool = True) -> Task:
    task = Task(
        platform=request.platform,
        status="running",
        hotspot_limit=request.hotspot_limit,
        item_limit_per_hotspot=request.item_limit_per_hotspot,
        comment_limit_per_item=request.comment_limit_per_item,
        total_items_count=0,
        processed_items_count=0,
        successful_items_count=0,
        failed_items_count=0,
        analysis_success_rate=0.0,
        analysis_status="normal",
        current_stage="queued",
        last_progress_at=utc_now(),
    )
    session.add(task)
    session.commit()
    session.refresh(task)
    if submit_background:
        task_executor.submit(run_task, task.id)
    return task


def list_tasks(session: Session) -> list[Task]:
    tasks = list(session.scalars(select(Task).order_by(Task.created_at.desc())))
    for task in tasks:
        hydrate_task_progress(session, task)
    return tasks


def get_task(session: Session, task_id: str) -> Task | None:
    task = session.get(Task, task_id)
    if task is not None:
        hydrate_task_progress(session, task)
    return task


def hydrate_task_progress(session: Session, task: Task) -> Task:
    display_number = calculate_task_display_number(session, task)
    task.display_number = display_number
    task.display_id = str(display_number)
    task.created_at_label = format_datetime_minute(task.created_at)
    task.scale_label = f"{task.hotspot_limit}热点 × {task.item_limit_per_hotspot}内容 × {task.comment_limit_per_item}评论"
    task.error_summary = build_task_error_summary(task)
    task.hotspots_count = session.scalar(select(func.count(Hotspot.id)).where(Hotspot.task_id == task.id)) or 0
    task.comments_count = session.scalar(select(func.count(Comment.id)).where(Comment.task_id == task.id)) or 0
    task.reports_count = session.scalar(select(func.count(Report.id)).where(Report.task_id == task.id)) or 0
    task.is_demo = task.id.startswith("demo-")
    task.last_progress_at = task.last_progress_at or task.created_at
    stage_code = task.current_stage
    if not stage_code and task.status == "success":
        stage_code = "success"
    elif not stage_code and task.status == "failed":
        stage_code = "failed"
    elif not stage_code and task.status == "running":
        stage_code = "queued"
    task.current_stage_label = STAGE_LABELS.get(stage_code or "", stage_code or "等待启动")
    now = ensure_utc_datetime(utc_now())
    running_start = ensure_utc_datetime(task.started_at or task.created_at)
    running_end = ensure_utc_datetime(task.finished_at) or now
    last_progress_at = ensure_utc_datetime(task.last_progress_at)
    running_seconds = max(0, int((running_end - running_start).total_seconds()))
    seconds_since_progress = max(0, int((now - last_progress_at).total_seconds())) if last_progress_at else None
    task.running_seconds = running_seconds
    task.seconds_since_last_progress = seconds_since_progress
    task.running_duration_label = format_duration_zh(running_seconds)
    task.last_progress_ago_label = f"{format_duration_zh(seconds_since_progress)}前" if seconds_since_progress is not None else "暂无记录"
    task.stale_threshold_minutes = STALE_PROGRESS_THRESHOLD_SECONDS // 60
    task.is_progress_stale = (
        task.status == "running"
        and seconds_since_progress is not None
        and seconds_since_progress >= STALE_PROGRESS_THRESHOLD_SECONDS
    )
    return task


def calculate_task_display_number(session: Session, task: Task) -> int:
    task_ids = list(session.scalars(select(Task.id).order_by(Task.created_at, Task.id)))
    try:
        return task_ids.index(task.id) + 1
    except ValueError:
        return len(task_ids) + 1


def format_datetime_minute(value: datetime | None) -> str:
    value = ensure_utc_datetime(value)
    return value.strftime("%Y-%m-%d %H:%M") if value else ""


def build_task_error_summary(task: Task) -> str | None:
    if task.error_message:
        return task.error_message
    if task.error_stage == "system" and task.error_type == "unexpected_restart":
        return RESTART_ERROR_MESSAGE
    if task.error_stage or task.error_type:
        return "任务失败，请查看后端日志"
    return None


def format_duration_zh(total_seconds: int | None) -> str:
    if total_seconds is None:
        return "暂无记录"
    total_seconds = max(0, int(total_seconds))
    minutes = total_seconds // 60
    hours = minutes // 60
    days = hours // 24
    if days > 0:
        remaining_hours = hours % 24
        return f"{days}天{remaining_hours}小时" if remaining_hours else f"{days}天"
    if hours > 0:
        remaining_minutes = minutes % 60
        return f"{hours}小时{remaining_minutes}分钟" if remaining_minutes else f"{hours}小时"
    if minutes > 0:
        return f"{minutes}分钟"
    return "刚刚"


def ensure_utc_datetime(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=UTC)
    return value.astimezone(UTC)


def update_task_stage(task: Task, stage: str) -> None:
    task.current_stage = stage
    task.last_progress_at = utc_now()


def run_task(task_id: str, *, session_factory: Callable[[], Session] | sessionmaker = SessionLocal) -> None:
    session = session_factory()
    close_session = hasattr(session, "close")
    try:
        task = session.get(Task, task_id)
        if task is None:
            return
        task.started_at = task.started_at or utc_now()
        session.commit()
        try:
            platform = build_platform(task.platform)
            ai_requester, report_summary_provider = build_ai_dependencies()

            try:
                update_task_stage(task, "crawl_hotspots")
                session.commit()
                hotspots = platform.fetch_hotspots(limit=task.hotspot_limit)
            except PlatformAPIError as exc:
                _mark_task_failed(task, "crawl_hotspots", exc.error_type, str(exc))
                session.commit()
                return
            except Exception as exc:
                _mark_task_failed(task, "crawl_hotspots", "api_error", str(exc))
                session.commit()
                return

            for hotspot_data in hotspots:
                hotspot = Hotspot(
                    task_id=task.id,
                    platform=task.platform,
                    source_hot_id=hotspot_data.source_hot_id,
                    rank=hotspot_data.rank,
                    title=hotspot_data.title,
                    heat_value=hotspot_data.heat_value,
                    raw_data=json.dumps(hotspot_data.raw_data or {}, ensure_ascii=False),
                )
                session.add(hotspot)
                session.flush()

                try:
                    update_task_stage(task, "search_items")
                    session.commit()
                    items = platform.search_items_by_hotspot(hotspot.title, limit=task.item_limit_per_hotspot)
                except Exception as exc:
                    task.failed_items_count += task.item_limit_per_hotspot
                    task.error_stage = "crawl_items"
                    task.error_type = getattr(exc, "error_type", "api_error")
                    task.error_message = str(exc)
                    session.commit()
                    continue

                task.total_items_count += len(items)
                seen_item_ids: set[str] = set()
                for item_data in items:
                    if item_data.source_item_id in seen_item_ids:
                        continue
                    seen_item_ids.add(item_data.source_item_id)
                    item = ContentItem(
                        task_id=task.id,
                        hotspot_id=hotspot.id,
                        platform=task.platform,
                        source_item_id=item_data.source_item_id,
                        item_type=item_data.item_type,
                        title=item_data.title,
                        summary=item_data.summary,
                        url=item_data.url,
                        status="pending",
                        raw_data=json.dumps(item_data.raw_data or {}, ensure_ascii=False),
                    )
                    session.add(item)
                    session.flush()
                    try:
                        update_task_stage(task, "crawl_comments")
                        session.commit()
                        comments = platform.fetch_comments(item.source_item_id, limit=task.comment_limit_per_item)
                        for comment_data in comments:
                            session.add(
                                Comment(
                                    task_id=task.id,
                                    hotspot_id=hotspot.id,
                                    content_item_id=item.id,
                                    platform=task.platform,
                                    source_comment_id=comment_data.source_comment_id,
                                    content=comment_data.content,
                                    author=comment_data.author,
                                    like_count=comment_data.like_count,
                                    comment_time=comment_data.comment_time,
                                    raw_data=json.dumps(comment_data.raw_data or {}, ensure_ascii=False),
                                )
                            )
                        session.flush()
                        item_comments = list(session.scalars(select(Comment).where(Comment.content_item_id == item.id)))
                        update_task_stage(task, "ai_analysis")
                        session.commit()
                        ai_results = analyze_comments_with_retry(
                            [{"comment_id": comment.id, "content": comment.content} for comment in item_comments],
                            requester=ai_requester,
                            max_retries=get_settings().ai_max_retries,
                            timeout_seconds=get_settings().ai_timeout_seconds,
                        )
                        results_by_comment_id = {result.comment_id: result for result in ai_results}
                        for comment in item_comments:
                            result = results_by_comment_id.get(comment.id)
                            if result is None:
                                comment.sentiment = "unknown"
                                comment.labels = "[]"
                                comment.reason = "ai_missing_result"
                                comment.ai_analysis_status = "failed"
                                continue
                            comment.sentiment = result.sentiment
                            comment.labels = json.dumps(result.labels, ensure_ascii=False)
                            comment.reason = result.reason
                            comment.ai_analysis_status = result.ai_analysis_status
                        item.status = "success"
                        task.successful_items_count += 1
                        update_task_stage(task, "generate_reports")
                        generate_item_report(session, item.id, summary_provider=report_summary_provider)
                    except Exception as exc:
                        item.status = "failed"
                        item.error_stage = "crawl_comments"
                        item.error_type = getattr(exc, "error_type", "api_error")
                        item.error_message = str(exc)
                        task.failed_items_count += 1
                        task.error_stage = item.error_stage
                        task.error_type = item.error_type
                        task.error_message = item.error_message
                    finally:
                        task.processed_items_count += 1
                        session.commit()

            if task.successful_items_count > 0:
                update_task_stage(task, "generate_reports")
                total_comments = session.scalar(select(func.count(Comment.id)).where(Comment.task_id == task.id)) or 0
                success_comments = sum(1 for comment in task.comments if comment.ai_analysis_status == "success")
                task.analysis_success_rate, task.analysis_status = calculate_analysis_status(
                    success_count=success_comments,
                    total_count=total_comments,
                )
                for hotspot in task.hotspots:
                    generate_hotspot_report(session, hotspot.id, summary_provider=report_summary_provider)
                task.status = "success"
                update_task_stage(task, "success")
            else:
                _mark_task_failed(task, task.error_stage or "crawl_items", task.error_type or "no_successful_items", task.error_message or "没有任何内容条目成功")
            task.finished_at = task.finished_at or utc_now()
            session.commit()
        except Exception as exc:
            _mark_task_failed(task, "system", "unexpected_error", str(exc))
            task.finished_at = task.finished_at or utc_now()
            session.commit()
    finally:
        if close_session:
            session.close()


def _mark_task_failed(task: Task, stage: str, error_type: str, message: str) -> None:
    task.status = "failed"
    update_task_stage(task, "failed")
    task.error_stage = stage
    task.error_type = error_type
    task.error_message = message