"""用户服务：资料 / 钱包 / 积分明细 / 注销 / 公开主页（对应《03-接口文档》第 4 章）"""
from datetime import datetime, timedelta

import redis.asyncio as aioredis
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import biz_config
from app.core import errors
from app.core.response import BizException
from app.core.security import verify_password
from app.models.billing import PointTransaction
from app.models.user import User
from app.models.video import Video
from app.services import membership_service, sensitive_service, video_service, wallet_service

DEACTIVATE_FREEZE_DAYS = 7


def _mask_phone(phone: str | None) -> str | None:
    """手机号脱敏：138****8000"""
    if not phone or len(phone) < 7:
        return phone
    return f"{phone[:3]}****{phone[-4:]}"


def _fmt(dt: datetime | None) -> str | None:
    return dt.strftime("%Y-%m-%d %H:%M:%S") if dt else None


async def _get_user(db: AsyncSession, user_id: int) -> User:
    user = await db.get(User, user_id)
    if user is None:
        raise BizException(errors.UNAUTHORIZED, http_status=401)
    return user


async def _me_payload(db: AsyncSession, user: User) -> dict:
    """个人信息完整结构（4.1/4.2 共用）"""
    ms = await membership_service.get_membership(db, user.id)
    active = membership_service.membership_active(ms)
    points = await wallet_service.get_balance(db, user.id)
    return {
        "user_id": user.id,
        "phone": _mask_phone(user.phone),
        "email": user.email,
        "nickname": user.nickname,
        "avatar_url": user.avatar_url,
        "bio": user.bio,
        "gender": user.gender,
        "is_member": active,
        "member_plan": ms.current_plan if active else None,
        "member_expire_at": _fmt(ms.expire_at) if active else None,
        "points": points,
    }


async def get_me(db: AsyncSession, user_id: int) -> dict:
    return await _me_payload(db, await _get_user(db, user_id))


async def update_me(db: AsyncSession, redis: aioredis.Redis, user_id: int, req) -> dict:
    """修改资料：昵称/简介过敏感词"""
    user = await _get_user(db, user_id)
    if req.nickname is not None:
        if await sensitive_service.find_sensitive(db, redis, req.nickname):
            raise BizException(errors.SENSITIVE_WORD)
        user.nickname = req.nickname
    if req.bio is not None:
        if await sensitive_service.find_sensitive(db, redis, req.bio):
            raise BizException(errors.SENSITIVE_WORD)
        user.bio = req.bio
    if req.gender is not None:
        user.gender = req.gender
    await db.commit()
    return await _me_payload(db, user)


async def set_avatar(db: AsyncSession, user_id: int, avatar_url: str) -> dict:
    user = await _get_user(db, user_id)
    user.avatar_url = avatar_url
    await db.commit()
    return {"avatar_url": avatar_url}


# ---------- 钱包 / 积分明细 ----------

async def get_wallet(db: AsyncSession, redis: aioredis.Redis, user_id: int) -> dict:
    wallet = await wallet_service.get_or_create_wallet(db, user_id)
    # 评论奖励进度（Redis Hash：count 进度 / daily 当日已发 / date 当日日期）
    prog = await redis.hgetall(f"comment:prog:{user_id}")  # type: ignore[misc]
    reward_cfg = (biz_config.get("points") or {}).get("comment_reward") or {}
    today = f"{datetime.now():%Y%m%d}"
    return {
        "points": wallet.points,
        "total_recharge": wallet.total_recharge,
        "total_earned": wallet.total_earned,
        "total_consumed": wallet.total_consumed,
        "comment_reward_progress": {
            "current": int(prog.get("count", 0)),
            "target": int(reward_cfg.get("per_count", 10)),
            "today_rewarded": int(prog.get("daily", 0)) if prog.get("date") == today else 0,
            "daily_limit": int(reward_cfg.get("daily_limit", 5)),
        },
    }


async def get_points_records(
    db: AsyncSession, user_id: int, page: int, page_size: int, biz_type: int | None
) -> dict:
    stmt = select(PointTransaction).where(PointTransaction.user_id == user_id)
    if biz_type is not None:
        stmt = stmt.where(PointTransaction.biz_type == biz_type)
    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
    rows = (
        await db.execute(
            stmt.order_by(PointTransaction.id.desc()).offset((page - 1) * page_size).limit(page_size)
        )
    ).scalars().all()
    return {
        "total": total,
        "list": [
            {
                "id": r.id,
                "change_amount": r.change_amount,
                "balance_after": r.balance_after,
                "biz_type": r.biz_type,
                "remark": r.remark,
                "created_at": _fmt(r.created_at),
            }
            for r in rows
        ],
    }


# ---------- 注销 ----------

async def deactivate(db: AsyncSession, redis: aioredis.Redis, user_id: int, password: str) -> dict:
    """申请注销：密码二次确认 → 进入 7 天冻结期 → 全端 Token 立即失效。

    冻结期内登录返回 1009 + can_revoke（见 auth_service.login）；到期匿名化由定时任务执行。
    """
    user = await _get_user(db, user_id)
    if not verify_password(password, user.password_hash):
        raise BizException(errors.PARAM_ERROR, "密码错误")
    if user.deactivate_at is not None:
        raise BizException(errors.PARAM_ERROR, "账号已在注销冻结期内")

    now = datetime.now()
    user.deactivate_at = now
    # 全端 Token 失效：递增令牌纪元 + 清 refresh 白名单
    await redis.incr(f"auth:epoch:{user_id}")
    await redis.delete(f"auth:refresh:{user_id}")
    await db.commit()

    revoke_before = now + timedelta(days=DEACTIVATE_FREEZE_DAYS)
    return {"deactivate_at": _fmt(now), "revoke_before": _fmt(revoke_before)}


# ---------- 公开主页 ----------

async def get_public_profile(db: AsyncSession, user_id: int) -> dict:
    user = await db.get(User, user_id)
    if user is None or user.status != 1:
        raise BizException(errors.NOT_FOUND, "用户不存在")
    base = select(Video).where(Video.user_id == user_id, Video.status == video_service.STATUS_PUBLISHED)
    video_count = (await db.execute(select(func.count()).select_from(base.subquery()))).scalar() or 0
    total_likes = (await db.execute(select(func.coalesce(func.sum(Video.like_count), 0)).where(
        Video.user_id == user_id, Video.status == video_service.STATUS_PUBLISHED
    ))).scalar() or 0
    return {
        "user_id": user.id,
        "nickname": user.nickname,
        "avatar_url": user.avatar_url,
        "bio": user.bio,
        "video_count": video_count,
        "total_likes": int(total_likes),
    }


async def get_user_videos(
    db: AsyncSession, user_id: int, page: int, page_size: int, viewer_user_id: int | None
) -> dict:
    """他人作品列表：仅已发布"""
    stmt = select(Video).where(Video.user_id == user_id, Video.status == video_service.STATUS_PUBLISHED)
    total = (await db.execute(select(func.count()).select_from(stmt.subquery()))).scalar() or 0
    videos = (
        await db.execute(
            stmt.order_by(Video.published_at.desc()).offset((page - 1) * page_size).limit(page_size)
        )
    ).scalars().all()
    return {"total": total, "list": await video_service.serialize_videos(db, list(videos), viewer_user_id)}