import logging
import traceback
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import func, inspect, select

from app.database import async_session
from app.models import TaskRunLog
from app.services.agent_lifecycle_service import evaluate_agents
from app.services.agent_service import settle_due_rewards
from app.services.instance_lifecycle_service import process_expired_instances, send_expiration_reminders
from app.services.instance_renewal_service import process_auto_renewals
from app.services.promotion_coupon_service import notify_expiring_coupons, refund_expired_member_coupons
from app.services.upstream_action_service import sync_pending_actions
from app.services.zjmf_stock_sync_service import sync_zjmf_product_stocks


log = logging.getLogger("qycw.task.center")

TaskHandler = Callable[[], Awaitable[dict[str, Any]]]


@dataclass(frozen=True)
class TaskDefinition:
    key: str
    name: str
    group: str
    description: str
    manual_enabled: bool
    interval_label: str
    handler: TaskHandler


_running: set[str] = set()


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _duration_ms(started_at: datetime, finished_at: datetime) -> int:
    return max(0, int((finished_at - started_at).total_seconds() * 1000))


def _error_text(exc: BaseException) -> str:
    text = "".join(traceback.format_exception_only(type(exc), exc)).strip()
    return text[:4000]


def _utc_iso(value: datetime | None) -> str | None:
    if value is None:
        return None
    if value.tzinfo is None:
        value = value.replace(tzinfo=timezone.utc)
    else:
        value = value.astimezone(timezone.utc)
    return value.isoformat().replace("+00:00", "Z")


async def _task_agent_lifecycle() -> dict[str, Any]:
    async with async_session() as session:
        try:
            summary = await evaluate_agents(session, limit=200)
            await session.commit()
            return summary
        except Exception:
            await session.rollback()
            raise


async def _task_instance_lifecycle() -> dict[str, Any]:
    async with async_session() as session:
        try:
            auto_renew = await process_auto_renewals(session)
            reminders = await send_expiration_reminders(session)
            expired = await process_expired_instances(session)
            await session.commit()
            return {"auto_renew": auto_renew, "reminders": reminders, "expired": expired}
        except Exception:
            await session.rollback()
            raise


async def _task_agent_settlement_release() -> dict[str, Any]:
    async with async_session() as session:
        try:
            summary = await settle_due_rewards(session, limit=200)
            await session.commit()
            return summary
        except Exception:
            await session.rollback()
            raise


async def _task_upstream_action_sync() -> dict[str, Any]:
    async with async_session() as session:
        try:
            summary = await sync_pending_actions(session)
            await session.commit()
            return summary
        except Exception:
            await session.rollback()
            raise


async def _task_zjmf_stock_sync() -> dict[str, Any]:
    async with async_session() as session:
        try:
            summary = await sync_zjmf_product_stocks(session)
            await session.commit()
            return summary
        except Exception:
            await session.rollback()
            raise


async def _task_promotion_coupon_refund_expired() -> dict[str, Any]:
    async with async_session() as session:
        try:
            summary = await refund_expired_member_coupons(session, limit=200)
            await session.commit()
            return summary
        except Exception:
            await session.rollback()
            raise


async def _task_promotion_coupon_notify_expiring() -> dict[str, Any]:
    async with async_session() as session:
        try:
            summary = await notify_expiring_coupons(session, limit=200)
            await session.commit()
            return summary
        except Exception:
            await session.rollback()
            raise


TASKS: dict[str, TaskDefinition] = {
    "instance_lifecycle": TaskDefinition(
        key="instance_lifecycle",
        name="实例生命周期",
        group="资源",
        description="处理自动续费、到期提醒、到期暂停和删除。",
        manual_enabled=True,
        interval_label="每 30 分钟",
        handler=_task_instance_lifecycle,
    ),
    "agent_lifecycle_evaluate": TaskDefinition(
        key="agent_lifecycle_evaluate",
        name="代理自动考核",
        group="代理",
        description="滚动 30 天考核代理，处理观察期、自动降档、退回会员和风控锁定。",
        manual_enabled=True,
        interval_label="每 60 分钟",
        handler=_task_agent_lifecycle,
    ),
    "agent_settlement_release": TaskDefinition(
        key="agent_settlement_release",
        name="代理结算释放",
        group="代理",
        description="扫描到期冻结奖励并释放代理/会员推广结算。",
        manual_enabled=True,
        interval_label="按系统设置",
        handler=_task_agent_settlement_release,
    ),
    "upstream_action_sync": TaskDefinition(
        key="upstream_action_sync",
        name="上游任务同步",
        group="资源",
        description="同步退订、暂停、删除、磁盘回退等上游动作结果。",
        manual_enabled=True,
        interval_label="按系统设置",
        handler=_task_upstream_action_sync,
    ),
    "zjmf_stock_sync": TaskDefinition(
        key="zjmf_stock_sync",
        name="魔方财务库存同步",
        group="资源",
        description="定时读取已绑定魔方财务产品的上游库存，并同步本地 stock_policy/stock。",
        manual_enabled=True,
        interval_label="按系统设置",
        handler=_task_zjmf_stock_sync,
    ),
    "promotion_coupon_refund_expired": TaskDefinition(
        key="promotion_coupon_refund_expired",
        name="过期会员券退积分",
        group="优惠券",
        description="将过期未使用的会员券积分成本退还给用户。",
        manual_enabled=True,
        interval_label="每 60 分钟",
        handler=_task_promotion_coupon_refund_expired,
    ),
    "promotion_coupon_notify_expiring": TaskDefinition(
        key="promotion_coupon_notify_expiring",
        name="优惠券到期提醒",
        group="优惠券",
        description="给即将过期的优惠券发送提醒。",
        manual_enabled=True,
        interval_label="每 60 分钟",
        handler=_task_promotion_coupon_notify_expiring,
    ),
}


def get_task_definition(task_key: str) -> TaskDefinition:
    task = TASKS.get(task_key)
    if not task:
        raise ValueError("unknown task")
    return task


async def _create_run_log(task: TaskDefinition, *, trigger_type: str, operator_id: int | None) -> int:
    async with async_session() as session:
        if not await _task_log_table_exists(session):
            return 0
        item = TaskRunLog(
            task_key=task.key,
            task_name=task.name,
            trigger_type=trigger_type,
            status="running",
            operator_id=operator_id,
            started_at=_now(),
        )
        session.add(item)
        await session.commit()
        return int(item.id)


async def _task_log_table_exists(session) -> bool:
    try:
        return bool(await session.run_sync(lambda sync_session: inspect(sync_session.get_bind()).has_table("task_run_logs")))
    except Exception as exc:  # noqa: BLE001
        log.warning("task run log table check failed: %s", exc)
        return False


async def _finish_run_log(
    run_id: int,
    *,
    status: str,
    started_at: datetime,
    summary: dict[str, Any] | None = None,
    error_message: str | None = None,
) -> TaskRunLog | None:
    if run_id <= 0:
        return None
    finished_at = _now()
    async with async_session() as session:
        item = await session.get(TaskRunLog, run_id)
        if not item:
            return None
        item.status = status
        item.finished_at = finished_at
        item.duration_ms = _duration_ms(started_at, finished_at)
        item.summary = summary
        item.error_message = error_message
        await session.commit()
        await session.refresh(item)
        return item


async def run_task(task_key: str, *, trigger_type: str, operator_id: int | None = None) -> dict[str, Any]:
    task = get_task_definition(task_key)
    if trigger_type == "manual" and not task.manual_enabled:
        raise ValueError("task does not support manual trigger")
    if task.key in _running:
        return {"accepted": False, "status": "skipped", "reason": "task is already running", "task_key": task.key}

    _running.add(task.key)
    started_at = _now()
    run_id = await _create_run_log(task, trigger_type=trigger_type, operator_id=operator_id)
    try:
        summary = await task.handler()
        item = await _finish_run_log(run_id, status="success", started_at=started_at, summary=summary)
        return {"accepted": True, "status": "success", "task_key": task.key, "run": task_run_to_dict(item), "summary": summary}
    except Exception as exc:
        log.exception("task center run failed: %s", task.key)
        item = await _finish_run_log(
            run_id,
            status="failed",
            started_at=started_at,
            error_message=_error_text(exc),
        )
        if trigger_type == "manual":
            raise
        return {"accepted": True, "status": "failed", "task_key": task.key, "run": task_run_to_dict(item)}
    finally:
        _running.discard(task.key)


def task_run_to_dict(item: TaskRunLog | None) -> dict[str, Any] | None:
    if item is None:
        return None
    return {
        "id": item.id,
        "task_key": item.task_key,
        "task_name": item.task_name,
        "trigger_type": item.trigger_type,
        "status": item.status,
        "operator_id": item.operator_id,
        "started_at": _utc_iso(item.started_at),
        "finished_at": _utc_iso(item.finished_at),
        "duration_ms": item.duration_ms,
        "summary": item.summary,
        "error_message": item.error_message,
        "created_at": _utc_iso(item.created_at),
    }


async def list_task_runs(
    *,
    task_key: str | None = None,
    status: str | None = None,
    trigger_type: str | None = None,
    offset: int = 0,
    limit: int = 50,
) -> dict[str, Any]:
    async with async_session() as session:
        if not await _task_log_table_exists(session):
            return {"items": [], "total": 0}
        stmt = select(TaskRunLog)
        count_stmt = select(func.count()).select_from(TaskRunLog)
        conditions = []
        if task_key:
            conditions.append(TaskRunLog.task_key == task_key)
        if status:
            conditions.append(TaskRunLog.status == status)
        if trigger_type:
            conditions.append(TaskRunLog.trigger_type == trigger_type)
        for condition in conditions:
            stmt = stmt.where(condition)
            count_stmt = count_stmt.where(condition)
        total = int(await session.scalar(count_stmt) or 0)
        rows = (
            await session.execute(
                stmt.order_by(TaskRunLog.id.desc()).offset(offset).limit(limit)
            )
        ).scalars().all()
        return {"items": [task_run_to_dict(item) for item in rows], "total": total}


async def list_tasks() -> list[dict[str, Any]]:
    async with async_session() as session:
        latest_by_key: dict[str, TaskRunLog] = {}
        failed_by_key: dict[str, int] = {}
        if await _task_log_table_exists(session):
            latest_rows = (
                await session.execute(
                    select(TaskRunLog)
                    .where(TaskRunLog.id.in_(select(func.max(TaskRunLog.id)).group_by(TaskRunLog.task_key)))
                )
            ).scalars().all()
            latest_by_key = {item.task_key: item for item in latest_rows}

            failure_rows = (
                await session.execute(
                    select(TaskRunLog.task_key, func.count(TaskRunLog.id))
                    .where(TaskRunLog.status == "failed")
                    .group_by(TaskRunLog.task_key)
                )
            ).all()
            failed_by_key = {task_key: int(count or 0) for task_key, count in failure_rows}

    items = []
    for task in TASKS.values():
        latest = latest_by_key.get(task.key)
        items.append(
            {
                "key": task.key,
                "name": task.name,
                "group": task.group,
                "description": task.description,
                "manual_enabled": task.manual_enabled,
                "interval_label": task.interval_label,
                "running": task.key in _running,
                "latest_run": task_run_to_dict(latest),
                "failed_runs": failed_by_key.get(task.key, 0),
            }
        )
    return items
