"""15 天不活跃清零业务逻辑(纯同步,可单测)。worker 只是它的 asyncio 外壳。

活跃口径复用 app.repositories.activity;清零走 wallet.grant_*(负数出账、写流水、不 commit)。
逐用户独立事务,一个失败不影响其余。
"""
from __future__ import annotations

import logging
from datetime import date, datetime

from sqlalchemy import or_, select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session

from app.core.rewards import CN_TZ
from app.integrations.notifier import InactivityNotifier
from app.models.inactivity import InactivityNotificationLog, InactivityResetLog
from app.models.user import User
from app.models.wallet import CoinAccount
from app.repositories import activity
from app.repositories import wallet as wallet_repo

logger = logging.getLogger("shagua.inactivity")

RESET_BIZ_TYPE = "inactivity_reset"
RESET_REMARK = "15天不活跃清零"

# 清零候选口径:金币或折算现金有余额即入选。**邀请现金不算**——它是产品红线、不清零
# (见 wallet.CoinAccount 注释),只有邀请现金余额的用户没有可清项,故不入选。
_ANY_BALANCE = or_(
    CoinAccount.coin_balance > 0,
    CoinAccount.cash_balance_cents > 0,
)


def _base_query(db: Session):
    """select(user_id, last_active, 三桶余额),join CoinAccount + 两活跃子查询。"""
    ev_sub, eng_sub = activity.last_active_subqueries(db)
    dialect = db.get_bind().dialect.name
    last_active = activity.last_active_expr(User.created_at, ev_sub, eng_sub, dialect)
    stmt = (
        select(
            User.id.label("user_id"),
            last_active.label("last_active"),
            CoinAccount.coin_balance,
            CoinAccount.cash_balance_cents,
            CoinAccount.invite_cash_balance_cents,
        )
        .join(CoinAccount, CoinAccount.user_id == User.id)
        .outerjoin(ev_sub, ev_sub.c.user_id == User.id)
        .outerjoin(eng_sub, eng_sub.c.user_id == User.id)
    )
    return stmt, last_active


def _cn_date(dt: datetime) -> date:
    """datetime → 北京自然日(naive 视为 UTC)。"""
    return activity.norm_utc(dt).astimezone(CN_TZ).date()


def _inactive_days(last_active: datetime, today: date) -> int:
    return (today - _cn_date(last_active)).days


def select_inactive_users(db: Session, *, cutoff: datetime):
    """应清零用户:last_active < cutoff 且金币/折算现金有余额(邀请现金不清、不计)。
    返回 Row 列表(值已快照,可跨 commit)。"""
    stmt, last_active = _base_query(db)
    stmt = stmt.where(_ANY_BALANCE, last_active < activity.as_utc(cutoff))
    return db.execute(stmt).all()


def clear_user(db: Session, *, user_id: int, last_active: datetime, inactive_days: int,
               reason: str, dry_run: bool = False) -> bool:
    """单用户清零(独立事务、行锁)。金币 + 折算现金归零 + 写审计 + 2 条流水;**邀请现金不清**
    (产品红线,见 wallet.CoinAccount 注释),仅作快照记入审计。返回是否真处理了(有可清余额)。

    dry_run=True:**只写审计名单、不动钱不写流水**(灰度看名单)。按 streak 去重——本 streak
    已记过(reset_at > last_active)就跳,避免 worker 每日重复记。"""
    acc = wallet_repo.get_or_create_account(db, user_id, commit=False, lock=True)
    coin, cash, invite = acc.coin_balance, acc.cash_balance_cents, acc.invite_cash_balance_cents
    if coin == 0 and cash == 0:  # 邀请现金不清,故不算"有可清余额"
        return False
    if dry_run and db.execute(
        select(InactivityResetLog.id).where(
            InactivityResetLog.user_id == user_id,
            InactivityResetLog.reset_at > activity.as_utc(last_active),
        ).limit(1)
    ).first():
        return False  # dry-run:本 streak 已记过审计,不重复记
    log = InactivityResetLog(
        user_id=user_id, coin_balance_before=coin, cash_balance_cents_before=cash,
        invite_cash_balance_cents_before=invite, last_active_at=activity.norm_utc(last_active),
        inactive_days=inactive_days, reason=reason,
    )
    db.add(log)
    db.flush()  # 拿 log.id 作 ref_id 交叉链接审计↔流水
    if not dry_run:  # dry-run 只记审计名单,不真出账
        ref = str(log.id)
        if coin:
            wallet_repo.grant_coins(db, user_id, -coin, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
        if cash:
            wallet_repo.grant_cash(db, user_id, -cash, biz_type=RESET_BIZ_TYPE, ref_id=ref, remark=RESET_REMARK)
        # 邀请现金(invite_cash_balance_cents)刻意不动:两本账物理隔离、邀请金是产品红线。
    db.commit()
    return True


def run_reset_once(db: Session, *, reset_days: int, today: date, dry_run: bool = False) -> dict:
    """扫一轮清零。逐用户独立 commit,失败隔离。dry_run=True 只记审计名单、不动钱(见 clear_user)。"""
    stats = {"scanned": 0, "cleared": 0, "failed": 0}
    cutoff = activity.reset_cutoff(reset_days, today)
    reason = f"inactive_{reset_days}d" + ("_dryrun" if dry_run else "")
    rows = select_inactive_users(db, cutoff=cutoff)  # 先物化,避免边遍历边 commit
    for row in rows:
        stats["scanned"] += 1
        idays = _inactive_days(row.last_active, today)
        try:
            if clear_user(db, user_id=row.user_id, last_active=row.last_active,
                          inactive_days=idays, reason=reason, dry_run=dry_run):
                stats["cleared"] += 1
        except SQLAlchemyError:
            db.rollback()
            stats["failed"] += 1
    return stats


def select_warn_candidates(db: Session, *, clear_cutoff: datetime, warn_hi: datetime):
    """预警候选:clear_cutoff <= last_active < warn_hi 且有可清余额(即已进预警窗、尚未到清零)。"""
    stmt, last_active = _base_query(db)
    stmt = stmt.where(
        _ANY_BALANCE,
        last_active >= activity.as_utc(clear_cutoff),
        last_active < activity.as_utc(warn_hi),
    )
    return db.execute(stmt).all()


def run_warn_once(db: Session, notifier: InactivityNotifier, *,
                  reset_days: int, warn_stages: list[int], today: date) -> dict:
    """扫一轮预警。每人取"最紧急的已到达档",按 streak 去重(notification_log.created_at > last_active)。
    预警只涉及会被清的金币 + 折算现金;邀请现金不清、不预警(仅在 notification_log 记快照)。
    逐用户 try/except 隔离:单用户通知器抛错 / DB 错不阻断其余,也绝不能拖累后续清零。"""
    stats = {"warned": 0, "warn_skipped": 0, "warn_failed": 0}
    if not warn_stages:
        return stats
    clear_cutoff = activity.reset_cutoff(reset_days, today)                # 到此即清零,不再预警
    warn_hi = activity.reset_cutoff(reset_days - max(warn_stages), today)  # 最早预警档边界
    ascending = sorted(warn_stages)                                       # 最紧急(最小 k)在前
    for row in select_warn_candidates(db, clear_cutoff=clear_cutoff, warn_hi=warn_hi):
        idays = _inactive_days(row.last_active, today)
        stage = next((k for k in ascending if idays >= reset_days - k), None)
        if stage is None:  # 防御:候选已在预警窗内、stage 必命中,此分支实际不可达
            continue
        try:
            already = db.execute(
                select(InactivityNotificationLog.id).where(
                    InactivityNotificationLog.user_id == row.user_id,
                    InactivityNotificationLog.stage == stage,
                    InactivityNotificationLog.created_at > activity.as_utc(row.last_active),
                ).limit(1)
            ).first()
            if already:
                stats["warn_skipped"] += 1
                continue
            status = notifier.warn(
                user_id=row.user_id, coin=row.coin_balance, cash_cents=row.cash_balance_cents,
                stage=stage, days_until_reset=reset_days - idays,
            )
            db.add(InactivityNotificationLog(
                user_id=row.user_id, stage=stage, inactive_days=idays,
                coin_balance=row.coin_balance, cash_balance_cents=row.cash_balance_cents,
                invite_cash_balance_cents=row.invite_cash_balance_cents,  # 快照,不参与"将清"额度
                channel=notifier.channel, status=status,
            ))
            db.commit()
            stats["warned"] += 1
        except Exception:  # noqa: BLE001 - 单用户预警失败(通知器抛错/DB 错)隔离,不阻断其余、不拖累清零
            db.rollback()
            stats["warn_failed"] += 1
    return stats


def run_once(db: Session, *, notifier: InactivityNotifier, reset_days: int,
             warn_stages: list[int], today: date, dry_run: bool = False) -> dict:
    """一轮完整任务:先预警(阶段 A)再清零(阶段 B)。返回合并统计。
    预警整段异常也**绝不阻塞清零**——清零是核心、不可逆资金操作,不能被通知故障拖住。
    dry_run=True(灰度默认):只记审计名单、不清、**也不预警**(不通知一个不会发生的清零)。"""
    warn = {"warned": 0, "warn_skipped": 0, "warn_failed": 0}
    if not dry_run:
        try:
            warn = run_warn_once(db, notifier, reset_days=reset_days, warn_stages=warn_stages, today=today)
        except Exception:  # noqa: BLE001 - 预警阶段整体失败(如候选查询失败)也要继续清零
            logger.exception("inactivity warn phase failed; proceeding to reset")
            db.rollback()
            warn = {"warned": 0, "warn_skipped": 0, "warn_failed": 0, "warn_phase_error": 1}
    reset = run_reset_once(db, reset_days=reset_days, today=today, dry_run=dry_run)
    return {**warn, **reset}