import logging
from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import (
    AgentCustomer,
    AgentEvaluationLog,
    AgentRiskEvent,
    AgentSettlement,
    AgentStatusLog,
    AgentTier,
    Order,
    OrderRefund,
    OrderStatus,
    UserAgent,
)
from app.models.user_agent import AgentStatus
from app.services import membership_service, notification_service


log = logging.getLogger("qycw.agent.lifecycle")

EVALUATION_WINDOW_DAYS = 30
AGENT_SETTLEMENT_STATUSES = {"pending", "frozen", "settled"}
MAX_OBSERVE_DAYS = 3650


def _status_value(status: Any) -> str | None:
    if status is None:
        return None
    return status.value if hasattr(status, "value") else str(status)


def _utc_datetime(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


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


def _positive_days(value: Any, *, default: int) -> int:
    try:
        days = int(value or default)
    except (TypeError, ValueError):
        days = default
    return max(1, min(MAX_OBSERVE_DAYS, days))


def agent_settlement_enabled(status: Any) -> bool:
    return _status_value(status) in {AgentStatus.ACTIVE.value, AgentStatus.OBSERVING.value}


def _paid_amount(order: Order) -> int:
    return max(0, int(order.total_amount or 0) - int(order.discount_amount or 0))


def _tier_rank(tier: AgentTier) -> tuple[int, int, int, int, int, int]:
    return (
        int(tier.monthly_sales_required or 0),
        int(tier.monthly_active_customers_required or 0),
        int(tier.min_total_spend or 0),
        int(tier.min_member_level or 0),
        -int(tier.cost_rate or 0),
        int(tier.sort_order or 0),
    )


def _metrics_pass_tier(metrics: dict[str, int], tier: AgentTier) -> bool:
    return (
        int(metrics["net_sales"]) >= int(tier.monthly_sales_required or 0)
        and int(metrics["active_customers"]) >= int(tier.monthly_active_customers_required or 0)
    )


def _underperform_reasons(metrics: dict[str, int], tier: AgentTier) -> list[str]:
    reasons: list[str] = []
    if int(metrics["net_sales"]) < int(tier.monthly_sales_required or 0):
        reasons.append(
            f"rolling 30-day net sales {metrics['net_sales']} < required {int(tier.monthly_sales_required or 0)}"
        )
    if int(metrics["active_customers"]) < int(tier.monthly_active_customers_required or 0):
        reasons.append(
            "rolling 30-day active customers "
            f"{metrics['active_customers']} < required {int(tier.monthly_active_customers_required or 0)}"
        )
    return reasons


def _risk_reasons(metrics: dict[str, int], tier: AgentTier, agent: UserAgent) -> list[str]:
    reasons: list[str] = []
    max_refund_rate = int(tier.max_refund_rate_bps or 0)
    if max_refund_rate > 0 and int(metrics["refund_rate_bps"]) >= max_refund_rate and int(metrics["refund_amount"]) > 0:
        reasons.append(f"refund rate {metrics['refund_rate_bps']}bps >= {max_refund_rate}bps")

    max_subsidy = int(tier.max_monthly_subsidy_cents or 0)
    min_balance = int(tier.min_balance_for_subsidy_cents or 0)
    if (
        max_subsidy > 0
        and int(metrics["subsidy_loss"]) >= max_subsidy
        and int(agent.agent_balance or 0) < min_balance
    ):
        reasons.append(
            f"subsidy loss {metrics['subsidy_loss']} >= {max_subsidy} "
            f"and balance {int(agent.agent_balance or 0)} < {min_balance}"
        )
    return reasons


async def calculate_agent_metrics(
    db: AsyncSession,
    agent_user_id: int,
    *,
    now: datetime | None = None,
) -> dict[str, int | str]:
    now = _utc_datetime(now) or _now_utc()
    since = now - timedelta(days=EVALUATION_WINDOW_DAYS)

    customer_ids = select(AgentCustomer.customer_user_id).where(
            AgentCustomer.agent_user_id == agent_user_id,
            AgentCustomer.status == "active",
        )
    participant_filter = or_(Order.user_id == int(agent_user_id), Order.user_id.in_(customer_ids))

    order_rows = (
        await db.execute(
            select(Order).where(
                participant_filter,
                Order.created_at >= since,
                Order.status.in_([OrderStatus.PAID, OrderStatus.REFUNDED]),
            )
        )
    ).scalars().all()

    refund_by_order: dict[int, int] = {}
    if order_rows:
        refund_rows = await db.execute(
            select(OrderRefund.order_id, func.coalesce(func.sum(OrderRefund.refund_amount), 0))
            .join(Order, Order.id == OrderRefund.order_id)
            .where(
                participant_filter,
                Order.created_at >= since,
                Order.status.in_([OrderStatus.PAID, OrderStatus.REFUNDED]),
                OrderRefund.status == "success",
            )
            .group_by(OrderRefund.order_id)
        )
        refund_by_order = {int(order_id): int(amount or 0) for order_id, amount in refund_rows.all()}

    gross_sales = 0
    refund_amount = 0
    net_by_customer: dict[int, int] = {}
    for order in order_rows:
        paid = _paid_amount(order)
        refunded = min(paid, int(refund_by_order.get(int(order.id), 0)))
        net = max(0, paid - refunded)
        gross_sales += paid
        refund_amount += refunded
        net_by_customer[int(order.user_id)] = net_by_customer.get(int(order.user_id), 0) + net

    subsidy_loss = (
        await db.scalar(
            select(func.coalesce(func.sum(func.abs(AgentSettlement.delta_amount)), 0)).where(
                AgentSettlement.user_id == agent_user_id,
                AgentSettlement.settlement_type == "agent_subsidy",
                AgentSettlement.status.in_(list(AGENT_SETTLEMENT_STATUSES)),
                AgentSettlement.created_at >= since,
                AgentSettlement.delta_amount < 0,
            )
        )
        or 0
    )

    refund_rate_bps = int(refund_amount * 10000 / gross_sales) if gross_sales > 0 else 0
    return {
        "period_start": since.isoformat(),
        "period_end": now.isoformat(),
        "gross_sales": int(gross_sales),
        "refund_amount": int(refund_amount),
        "net_sales": max(0, int(gross_sales) - int(refund_amount)),
        "refund_rate_bps": refund_rate_bps,
        "active_customers": sum(1 for amount in net_by_customer.values() if amount > 0),
        "subsidy_loss": int(subsidy_loss),
    }


def _snapshot(metrics: dict[str, Any], tier: AgentTier | None = None) -> dict[str, Any]:
    data = dict(metrics)
    if tier:
        data["tier"] = {
            "id": tier.id,
            "code": tier.code,
            "monthly_sales_required": int(tier.monthly_sales_required or 0),
            "monthly_active_customers_required": int(tier.monthly_active_customers_required or 0),
            "max_refund_rate_bps": int(tier.max_refund_rate_bps or 0),
            "max_monthly_subsidy_cents": int(tier.max_monthly_subsidy_cents or 0),
            "min_balance_for_subsidy_cents": int(tier.min_balance_for_subsidy_cents or 0),
        }
    return data


async def _write_status_log(
    db: AsyncSession,
    agent: UserAgent,
    *,
    source: str,
    action: str,
    before_status: str | None,
    before_tier_id: int | None,
    reason: str,
    operator_id: int | None = None,
    snapshot: dict[str, Any] | None = None,
) -> None:
    db.add(
        AgentStatusLog(
            agent_id=agent.id,
            user_id=agent.user_id,
            operator_id=operator_id,
            source=source,
            action=action,
            before_status=before_status,
            after_status=_status_value(agent.status),
            before_tier_id=before_tier_id,
            after_tier_id=agent.agent_tier_id,
            reason=reason,
            snapshot=snapshot,
        )
    )


async def _notify(db: AsyncSession, user_id: int, title: str, content: str) -> None:
    try:
        async with db.begin_nested():
            await notification_service.notify(
                db,
                user_id=user_id,
                category="agent",
                title=title,
                content=content,
                link="/agent",
            )
    except Exception as exc:  # noqa: BLE001
        log.warning("agent lifecycle notification failed user=%s: %s", user_id, exc)


async def _sync_membership_tier(db: AsyncSession, agent: UserAgent, tier: AgentTier | None) -> None:
    membership = await membership_service.get_or_create_membership(db, agent.user_id)
    membership.agent_tier = tier.code if tier else None


async def _resolve_agent_tier(db: AsyncSession, agent: UserAgent, tier: AgentTier | None = None) -> AgentTier | None:
    if tier:
        return tier
    if agent.agent_tier:
        return agent.agent_tier
    if agent.agent_tier_id:
        return await db.get(AgentTier, agent.agent_tier_id)
    return None


async def _find_downgrade_target(
    db: AsyncSession,
    current_tier: AgentTier,
    metrics: dict[str, int],
) -> AgentTier | None:
    result = await db.execute(select(AgentTier).where(AgentTier.enabled == True))  # noqa: E712
    current_rank = _tier_rank(current_tier)
    candidates = [
        tier
        for tier in result.scalars().all()
        if tier.id != current_tier.id and _tier_rank(tier) < current_rank and _metrics_pass_tier(metrics, tier)
    ]
    if not candidates:
        return None
    candidates.sort(key=_tier_rank, reverse=True)
    return candidates[0]


async def _write_evaluation_log(
    db: AsyncSession,
    agent: UserAgent,
    tier: AgentTier | None,
    metrics: dict[str, Any],
    *,
    passed: bool,
    risk_locked: bool,
    action: str,
    reason: str | None,
) -> None:
    period_start = datetime.fromisoformat(str(metrics["period_start"]))
    period_end = datetime.fromisoformat(str(metrics["period_end"]))
    db.add(
        AgentEvaluationLog(
            agent_id=agent.id,
            user_id=agent.user_id,
            tier_id=tier.id if tier else None,
            period_start=period_start,
            period_end=period_end,
            net_sales=int(metrics["net_sales"]),
            gross_sales=int(metrics["gross_sales"]),
            refund_amount=int(metrics["refund_amount"]),
            refund_rate_bps=int(metrics["refund_rate_bps"]),
            active_customers=int(metrics["active_customers"]),
            subsidy_loss=int(metrics["subsidy_loss"]),
            agent_balance=int(agent.agent_balance or 0),
            passed=1 if passed else 0,
            risk_locked=1 if risk_locked else 0,
            action=action,
            reason=reason,
            snapshot=_snapshot(metrics, tier),
        )
    )


async def evaluate_agent(
    db: AsyncSession,
    agent: UserAgent,
    *,
    now: datetime | None = None,
    force: bool = False,
) -> dict[str, Any]:
    now = _utc_datetime(now) or _now_utc()
    if not agent.id:
        await db.flush()

    if _status_value(agent.status) not in {
        AgentStatus.ACTIVE.value,
        AgentStatus.OBSERVING.value,
    }:
        return {"action": "skipped", "reason": "agent status is not evaluable"}
    if not agent.agent_tier_id or not agent.agent_tier:
        return {"action": "skipped", "reason": "agent has no tier"}

    tier = agent.agent_tier
    metrics = await calculate_agent_metrics(db, agent.user_id, now=now)
    metrics_int = {k: v for k, v in metrics.items() if isinstance(v, int)}
    passed = _metrics_pass_tier(metrics_int, tier)
    risk_reasons = _risk_reasons(metrics_int, tier, agent)

    before_status = _status_value(agent.status)
    before_tier_id = agent.agent_tier_id
    action = "none"
    reason: str | None = None

    if risk_reasons:
        reason = "; ".join(risk_reasons)
        if before_status != AgentStatus.RISK_LOCKED.value:
            agent.status = AgentStatus.RISK_LOCKED
            agent.risk_locked_at = now
            agent.risk_reason = reason
            action = "risk_locked"
            db.add(
                AgentRiskEvent(
                    agent_id=agent.id,
                    user_id=agent.user_id,
                    event_type="auto_risk_lock",
                    severity="critical",
                    status="open",
                    reason=reason,
                    snapshot=_snapshot(metrics, tier),
                )
            )
            await _write_status_log(
                db,
                agent,
                source="system",
                action=action,
                before_status=before_status,
                before_tier_id=before_tier_id,
                reason=reason,
                snapshot=_snapshot(metrics, tier),
            )
            await _notify(db, agent.user_id, "代理权益已风控冻结", f"系统检测到风险：{reason}。请联系客服处理。")
        passed = False
    elif passed:
        if before_status == AgentStatus.OBSERVING.value:
            agent.status = AgentStatus.ACTIVE
            agent.observe_started_at = None
            agent.observe_until = None
            action = "restore_active"
            reason = "rolling 30-day metrics recovered"
            await _write_status_log(
                db,
                agent,
                source="system",
                action=action,
                before_status=before_status,
                before_tier_id=before_tier_id,
                reason=reason,
                snapshot=_snapshot(metrics, tier),
            )
            await _notify(db, agent.user_id, "代理观察期已解除", "您的代理考核指标已恢复达标，代理状态已恢复正常。")
    else:
        reasons = _underperform_reasons(metrics_int, tier)
        reason = "; ".join(reasons) or "rolling 30-day metrics below tier requirements"
        if not bool(tier.auto_downgrade_enabled) and not force:
            action = "not_pass_manual_only"
        elif before_status == AgentStatus.ACTIVE.value:
            agent.status = AgentStatus.OBSERVING
            agent.observe_started_at = now
            agent.observe_until = now + timedelta(days=_positive_days(tier.observe_days, default=7))
            action = "observe"
            await _write_status_log(
                db,
                agent,
                source="system",
                action=action,
                before_status=before_status,
                before_tier_id=before_tier_id,
                reason=reason,
                snapshot=_snapshot(metrics, tier),
            )
            await _notify(
                db,
                agent.user_id,
                "代理进入降级观察期",
                f"您的代理考核暂未达标：{reason}。观察期至 {agent.observe_until.strftime('%Y-%m-%d %H:%M')}。",
            )
        elif before_status == AgentStatus.OBSERVING.value and (
            force or (agent.observe_until is not None and now >= (_utc_datetime(agent.observe_until) or now))
        ):
            target = await _find_downgrade_target(db, tier, metrics_int)
            if target:
                agent.status = AgentStatus.ACTIVE
                agent.agent_tier_id = target.id
                agent.agent_tier = target
                agent.observe_started_at = None
                agent.observe_until = None
                agent.risk_reason = None
                await _sync_membership_tier(db, agent, target)
                action = "downgrade"
                await _write_status_log(
                    db,
                    agent,
                    source="system",
                    action=action,
                    before_status=before_status,
                    before_tier_id=before_tier_id,
                    reason=reason,
                    snapshot=_snapshot(metrics, tier),
                )
                await _notify(db, agent.user_id, "代理档位已自动降级", f"观察期结束后仍未达标，代理档位已调整为 {target.name}。")
            else:
                agent.status = AgentStatus.SUSPENDED
                agent.agent_tier_id = None
                agent.observe_started_at = None
                agent.observe_until = None
                await _sync_membership_tier(db, agent, None)
                action = "return_member"
                await _write_status_log(
                    db,
                    agent,
                    source="system",
                    action=action,
                    before_status=before_status,
                    before_tier_id=before_tier_id,
                    reason=reason,
                    snapshot=_snapshot(metrics, tier),
                )
                await _notify(db, agent.user_id, "代理身份已退回会员", "观察期结束后仍未满足任一代理档位要求，已退回普通会员推广身份。")

    agent.last_evaluated_at = now
    agent.last_evaluation_snapshot = _snapshot(metrics, tier)
    await _write_evaluation_log(
        db,
        agent,
        tier,
        metrics,
        passed=passed,
        risk_locked=bool(risk_reasons),
        action=action,
        reason=reason,
    )
    await db.flush()
    return {"action": action, "passed": passed, "risk_locked": bool(risk_reasons), "metrics": metrics, "reason": reason}


async def evaluate_agents(db: AsyncSession, *, limit: int = 100) -> dict[str, int]:
    result = await db.execute(
        select(UserAgent)
        .where(UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]))
        .order_by(UserAgent.last_evaluated_at.asc(), UserAgent.id.asc())
        .limit(limit)
    )
    agents = result.scalars().all()
    summary = {"checked": 0, "observe": 0, "restore_active": 0, "downgrade": 0, "return_member": 0, "risk_locked": 0, "failed": 0}
    for agent in agents:
        try:
            async with db.begin_nested():
                outcome = await evaluate_agent(db, agent)
            summary["checked"] += 1
            action = str(outcome.get("action") or "none")
            if action in summary:
                summary[action] += 1
        except Exception:
            summary["failed"] += 1
            log.exception("agent lifecycle evaluation failed: agent=%s", agent.id)
    return summary


async def manual_agent_action(
    db: AsyncSession,
    agent: UserAgent,
    *,
    action: str,
    reason: str,
    operator_id: int | None,
    tier: AgentTier | None = None,
) -> UserAgent:
    text = str(reason or "").strip()
    if not text:
        raise ValueError("reason is required")
    before_status = _status_value(agent.status)
    before_tier_id = agent.agent_tier_id

    if action == "restore":
        if tier:
            agent.agent_tier_id = tier.id
            agent.agent_tier = tier
        if not agent.agent_tier_id:
            raise ValueError("tier is required to restore agent")
        restore_tier = await _resolve_agent_tier(db, agent, tier)
        if not restore_tier:
            raise ValueError("tier is required to restore agent")
        agent.status = AgentStatus.ACTIVE
        agent.observe_started_at = None
        agent.observe_until = None
        agent.risk_locked_at = None
        agent.risk_reason = None
        await _sync_membership_tier(db, agent, restore_tier)
    elif action == "observe":
        observe_tier = await _resolve_agent_tier(db, agent, tier)
        if not observe_tier:
            raise ValueError("tier is required to observe agent")
        if tier:
            agent.agent_tier_id = tier.id
            agent.agent_tier = tier
            await _sync_membership_tier(db, agent, tier)
        agent.status = AgentStatus.OBSERVING
        agent.observe_started_at = _now_utc()
        agent.observe_until = agent.observe_started_at + timedelta(
            days=_positive_days(observe_tier.observe_days, default=7)
        )
    elif action == "risk_lock":
        agent.status = AgentStatus.RISK_LOCKED
        agent.risk_locked_at = _now_utc()
        agent.risk_reason = text
        db.add(
            AgentRiskEvent(
                agent_id=agent.id,
                user_id=agent.user_id,
                event_type="manual_risk_lock",
                severity="critical",
                status="open",
                reason=text,
                snapshot=agent.last_evaluation_snapshot,
            )
        )
    elif action == "suspend":
        agent.status = AgentStatus.SUSPENDED
        agent.agent_tier_id = None
        agent.observe_started_at = None
        agent.observe_until = None
        await _sync_membership_tier(db, agent, None)
    elif action == "terminate":
        agent.status = AgentStatus.TERMINATED
        agent.agent_tier_id = None
        agent.observe_started_at = None
        agent.observe_until = None
        await _sync_membership_tier(db, agent, None)
    else:
        raise ValueError("unsupported action")

    await _write_status_log(
        db,
        agent,
        source="admin",
        action=action,
        before_status=before_status,
        before_tier_id=before_tier_id,
        reason=text,
        operator_id=operator_id,
        snapshot=agent.last_evaluation_snapshot,
    )
    await _notify(db, agent.user_id, "代理状态已调整", f"管理员已调整您的代理状态：{text}")
    await db.flush()
    return agent
