"""admin 跨用户查询(去掉现有 repo 的 user_id 强制过滤)+ 通用游标分页 helper + 用户概览。

现有 app/repositories/ 的 list_* 都强绑单个 user_id(C 端只看自己);admin 要看全量、按条件筛,
所以在这里另起一套。游标约定与现有一致:id 倒序,cursor=上页最后一条 id,返回 (items, next_cursor)。
"""
from __future__ import annotations

from datetime import date, datetime, time, timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal
from zoneinfo import ZoneInfo

from sqlalchemy import Select, and_, asc, case, desc, func, or_, select
from sqlalchemy.orm import Session

from app.core import rewards
from app.core.config import settings
from app.models.ad_ecpm import AdEcpmRecord
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.admin import AdminAuditLog
from app.models.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.coupon_state import CouponPromptEngagement
from app.models.device import DeviceLiveness
from app.models.feedback import Feedback
from app.models.invite import InviteRelation
from app.models.onboarding import OnboardingCompletion
from app.models.price_report import PriceReport
from app.models.savings import SavingsRecord
from app.models.user import User
from app.models.wallet import (
    CashTransaction,
    CoinAccount,
    CoinTransaction,
    WithdrawOrder,
)
from app.repositories import activity, ad_ecpm

# 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
_NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")

# 信息流点位场景 → 金币记录「赚取途径」展示名;NULL/未知 = 历史未分类。
_FEED_SCENE_LABEL = {
    "comparison": "比价信息流",
    "coupon": "领券信息流",
    "welfare": "福利信息流",
}


_DEVICE_MARKETING_NAMES = {
    "23078RKD5C": "Redmi K60 至尊版",
    "M2012K11AC": "Redmi K40",
    "PJA110": "一加 Ace 2 Pro",
    "PPG-AN00": "荣耀 GT Pro",
    "V2166BA": "vivo Y77e",
    "V2309A": "vivo X100",
}


def _device_marketing_name(model: str | None) -> str | None:
    """把线上已知 Build.MODEL 编码转成用户可识别的商品名。"""
    if not model:
        return None
    return _DEVICE_MARKETING_NAMES.get(model.strip().upper())


def _attach_comparison_order_status(db: Session, items: list[ComparisonRecord]) -> None:
    """按 C 端既有口径给比价记录批量补充是否真实下单。"""
    user_ids = {item.user_id for item in items if item.user_id is not None}
    shop_names = {item.store_name for item in items if item.store_name}
    ordered_pairs: set[tuple[int, str]] = set()
    if user_ids and shop_names:
        rows = db.execute(
            select(SavingsRecord.user_id, SavingsRecord.shop_name)
            .where(
                SavingsRecord.user_id.in_(user_ids),
                SavingsRecord.source == "compare",
                SavingsRecord.shop_name.in_(shop_names),
            )
            .distinct()
        ).all()
        ordered_pairs = {
            (row.user_id, row.shop_name)
            for row in rows
            if row.shop_name is not None
        }
    for item in items:
        item.ordered = bool(
            item.user_id is not None
            and item.store_name
            and (item.user_id, item.store_name) in ordered_pairs
        )


def _attach_comparison_device_details(items: list[ComparisonRecord]) -> None:
    """给比价记录补充可读机型名，同时保留原始设备编码。"""
    for item in items:
        item.device_model_name = _device_marketing_name(item.device_model)


def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
    """按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
    candidates = [
        item for item in feedbacks if item.device_model and item.device_model.strip()
    ]
    for item in candidates:
        item.device_model_name = _device_marketing_name(item.device_model)
        item.device_manufacturer = None
        item.rom_version = None
    if not candidates:
        return

    ranked = (
        select(
            Feedback.id.label("feedback_id"),
            ComparisonRecord.device_manufacturer.label("device_manufacturer"),
            ComparisonRecord.rom_version.label("rom_version"),
            func.row_number()
            .over(
                partition_by=Feedback.id,
                order_by=(
                    ComparisonRecord.created_at.desc(),
                    ComparisonRecord.id.desc(),
                ),
            )
            .label("row_num"),
        )
        .join(
            ComparisonRecord,
            and_(
                ComparisonRecord.user_id == Feedback.user_id,
                ComparisonRecord.device_model == Feedback.device_model,
                ComparisonRecord.created_at <= Feedback.created_at,
            ),
        )
        .where(Feedback.id.in_([item.id for item in candidates]))
        .subquery()
    )
    details = {
        row.feedback_id: row
        for row in db.execute(select(ranked).where(ranked.c.row_num == 1)).all()
    }

    for item in candidates:
        detail = details.get(item.id)
        item.device_manufacturer = detail.device_manufacturer if detail else None
        item.rom_version = detail.rom_version if detail else None


def cursor_paginate(
    db: Session, stmt: Select, id_col, *, limit: int, cursor: int | None
) -> tuple[list, int | None]:
    """通用游标分页(id 倒序)。stmt 不要预先带 order_by/limit。

    多取 1 条探测有没有下一页;next_cursor 取本页最后一条的 id(下一页查 id < 它),
    绝不用第 limit+1 条的 id——那条既不在本页也不在下页,会每页边界丢一条(见 audit_log 同款修复)。
    """
    if cursor is not None:
        stmt = stmt.where(id_col < cursor)
    stmt = stmt.order_by(id_col.desc()).limit(limit + 1)
    rows = list(db.execute(stmt).scalars().all())
    has_more = len(rows) > limit
    items = rows[:limit]
    next_cursor = items[-1].id if has_more else None
    return items, next_cursor


def offset_paginate(
    db: Session, stmt: Select, sort_clause: tuple, *, limit: int, cursor: int | None
) -> tuple[list, int | None, int]:
    """offset 分页 + 总数。stmt 只含 where/join,不要预先带 order_by/offset/limit。

    cursor 即 offset(页码分页:offset=(page-1)*pageSize)。返回 (items, next_cursor, total):
    - total:符合筛选条件的总条数(供 antd pagination 渲染页码/共 N 条),count 在 P0 量级开销可忽略;
    - next_cursor:下一页 offset(兼容「加载更多」),末页为 None。
    多取 1 条探测下一页。sort_clause 为 order_by 表达式元组(末位应含 id 保证稳定排序)。"""
    total = int(
        db.execute(select(func.count()).select_from(stmt.subquery())).scalar_one()
    )
    offset = max(cursor or 0, 0)
    rows = list(
        db.execute(stmt.order_by(*sort_clause).offset(offset).limit(limit + 1)).scalars().all()
    )
    has_more = len(rows) > limit
    items = rows[:limit]
    next_cursor = offset + limit if has_more else None
    return items, next_cursor, total


def _attach_last_active(db: Session, users: list[User]) -> None:
    """给本页用户瞬态挂 last_active_at(非 DB 列,供 AdminUserListItem from_attributes 读)。

    口径同 [_last_active_expr];按本页 user_id 批量两次 GROUP BY,防 N+1。
    """
    uids = [u.id for u in users]
    if not uids:
        return
    ev_map = dict(
        db.execute(
            select(AnalyticsEvent.user_id, func.max(AnalyticsEvent.created_at))
            .where(
                AnalyticsEvent.user_id.in_(uids),
                activity.active_event_condition(),
            )
            .group_by(AnalyticsEvent.user_id)
        ).all()
    )
    eng_map = dict(
        db.execute(
            select(CouponPromptEngagement.user_id, func.max(CouponPromptEngagement.created_at))
            .where(
                CouponPromptEngagement.user_id.in_(uids),
                CouponPromptEngagement.engage_type == "claim_started",
            )
            .group_by(CouponPromptEngagement.user_id)
        ).all()
    )
    for u in users:
        candidates = [
            activity.norm_utc(u.created_at),      # baseline 由 last_login_at 改为 created_at(登录不算活跃)
            activity.norm_utc(ev_map.get(u.id)),
            activity.norm_utc(eng_map.get(u.id)),
        ]
        u.last_active_at = max((c for c in candidates if c is not None), default=None)


def list_users(
    db: Session,
    *,
    phone: str | None = None,
    register_channel: str | None = None,
    status: str | None = None,
    nickname: str | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    last_login_from: datetime | None = None,
    last_login_to: datetime | None = None,
    last_active_from: datetime | None = None,
    last_active_to: datetime | None = None,
    sort_by: str = "id",
    sort_order: str = "desc",
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[User], int | None, int]:
    """用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录·最近活跃
    时间范围筛选,按 id·注册时间·最近登录·最近活跃排序;每页附带计算列 last_active_at
    (口径见 [_last_active_expr])。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
    代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
    日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
    # 最近活跃 = max(注册时间, 最近行为事件, 最近领券发起)。baseline 由 last_login_at 改为 created_at
    #(登录不代表在用 App;口径统一到 activity.py,含 home_visible + 比价 + 领券,见 activity.ACTIVE_EVENTS)。
    # 未命中侧 coalesce 到 created_at(恒非空基线)。派生表 1:1,outerjoin 不放大行数。
    ev_agg, eng_agg = activity.last_active_subqueries(db)
    last_active = activity.last_active_expr(
        User.created_at, ev_agg, eng_agg, db.get_bind().dialect.name
    )
    stmt = (
        select(User)
        .outerjoin(ev_agg, ev_agg.c.user_id == User.id)
        .outerjoin(eng_agg, eng_agg.c.user_id == User.id)
    )
    if phone:
        stmt = stmt.where(User.phone.like(f"{phone}%"))  # 前缀匹配
    if register_channel:
        stmt = stmt.where(User.register_channel == register_channel)
    if status:
        stmt = stmt.where(User.status == status)
    if nickname and nickname.strip():
        stmt = stmt.where(User.nickname.ilike(f"%{nickname.strip()}%"))
    if created_from is not None:
        stmt = stmt.where(User.created_at >= _as_utc(created_from))
    if created_to is not None:
        stmt = stmt.where(User.created_at <= _as_utc(created_to))
    if last_login_from is not None:
        stmt = stmt.where(User.last_login_at >= _as_utc(last_login_from))
    if last_login_to is not None:
        stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to))
    if last_active_from is not None:
        stmt = stmt.where(last_active >= _as_utc(last_active_from))
    if last_active_to is not None:
        stmt = stmt.where(last_active <= _as_utc(last_active_to))

    sort_cols = {
        "id": User.id,
        "created_at": User.created_at,
        "last_login_at": User.last_login_at,
        "last_active_at": last_active,
    }
    sort_col = sort_cols.get(sort_by, User.id)
    order_fn = asc if sort_order == "asc" else desc
    id_order = asc(User.id) if sort_order == "asc" else desc(User.id)
    items, next_cursor, total = offset_paginate(
        db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
    )
    _attach_last_active(db, items)
    return items, next_cursor, total


def _attach_user_info(db: Session, records: list[ComparisonRecord | Feedback | PriceReport]) -> None:
    """给每条记录瞬态挂 phone/nickname(非 DB 列,供 admin schema from_attributes 读)。
    按 user_id 鸭子类型,比价记录/反馈/上报通用。"""
    uids = {r.user_id for r in records}
    if not uids:
        return
    rows = db.execute(
        select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
    ).all()
    umap = {uid: (phone, nick) for uid, phone, nick in rows}
    for r in records:
        phone, nick = umap.get(r.user_id, (None, None))
        r.phone = phone
        r.nickname = nick


def _comparison_conditions(
    *,
    user_id: int | None = None,
    phone: str | None = None,
    status: str | None = None,
    business_type: str | None = None,
    store: str | None = None,
    product: str | None = None,
    date_from: date | None = None,
    date_to: date | None = None,
) -> list:
    """比价列表与概览共用筛选条件；日期按北京自然日闭区间解释。"""
    conditions = []
    if user_id is not None:
        conditions.append(ComparisonRecord.user_id == user_id)
    if phone:
        conditions.append(
            ComparisonRecord.user_id.in_(
                select(User.id).where(User.phone.like(f"{phone}%"))
            )
        )
    if status:
        conditions.append(ComparisonRecord.status == status)
    if business_type:
        conditions.append(ComparisonRecord.business_type == business_type)
    if store:
        conditions.append(ComparisonRecord.store_name.like(f"%{store}%"))
    if product:
        conditions.append(ComparisonRecord.product_names.like(f"%{product}%"))
    beijing = ZoneInfo("Asia/Shanghai")
    if date_from is not None:
        start_utc = datetime.combine(date_from, time.min, tzinfo=beijing).astimezone(timezone.utc)
        conditions.append(ComparisonRecord.created_at >= start_utc)
    if date_to is not None:
        end_utc = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=beijing).astimezone(timezone.utc)
        conditions.append(ComparisonRecord.created_at < end_utc)
    return conditions


def list_comparison_records(
    db: Session,
    *,
    user_id: int | None = None,
    phone: str | None = None,
    status: str | None = None,
    business_type: str | None = None,
    store: str | None = None,
    product: str | None = None,
    date_from: date | None = None,
    date_to: date | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[ComparisonRecord], int | None, int]:
    """admin 比价记录列表(debug)。按 user_id 精确 或 phone 前缀定位用户 + status/业务类型筛,
    store(店名)/product(商品名)子串模糊匹配,offset 分页(创建时间倒序、id 兜底)。
    join User 取 phone/nickname 瞬态挂记录上。"""
    conditions = _comparison_conditions(
        user_id=user_id, phone=phone, status=status, business_type=business_type,
        store=store, product=product, date_from=date_from, date_to=date_to,
    )
    stmt = select(ComparisonRecord).where(*conditions)
    items, next_cursor, total = offset_paginate(
        db, stmt,
        (desc(ComparisonRecord.created_at), desc(ComparisonRecord.id)),
        limit=limit, cursor=cursor,
    )
    _attach_user_info(db, items)
    _attach_comparison_order_status(db, items)
    _attach_comparison_device_details(items)
    # 「本次比价看广告的预估收益」:按本页 trace_id 一次性聚合(同 _attach_user_info 逐页范式)。
    # ad_revenue_yuan 非 ORM 列,仅瞬态挂实例上供 AdminComparisonListItem(from_attributes)读出。
    rev = ad_ecpm.revenue_yuan_by_trace(db, [it.trace_id for it in items])
    for it in items:
        it.ad_revenue_yuan = rev.get(it.trace_id, 0.0)
    return items, next_cursor, total


def _comparison_percentile(sorted_values: list[int], q: float) -> int | None:
    """线性插值分位数（非负毫秒值四舍五入；单条数据返回自身）。"""
    if not sorted_values:
        return None
    if len(sorted_values) == 1:
        return sorted_values[0]
    index = (len(sorted_values) - 1) * q
    lower = int(index)
    upper = min(lower + 1, len(sorted_values) - 1)
    value = sorted_values[lower] * (upper - index) + sorted_values[upper] * (index - lower)
    return int(value + 0.5)


def _round_duration_ms(value) -> int | None:
    """将数据库聚合结果按既有口径四舍五入为整数毫秒。"""
    if value is None:
        return None
    return int(Decimal(str(value)).quantize(Decimal("1"), rounding=ROUND_HALF_UP))


def _comparison_duration_aggregate_stmt(conditions: list, status: str, quantiles: tuple[float, ...]):
    """PostgreSQL 耗时聚合语句；每种状态只返回一行。"""
    return select(
        func.avg(ComparisonRecord.total_ms),
        *(
            func.percentile_cont(q).within_group(ComparisonRecord.total_ms)
            for q in quantiles
        ),
    ).where(
        *conditions,
        ComparisonRecord.status == status,
        ComparisonRecord.total_ms.is_not(None),
    )


def _comparison_duration_aggregates(
    db: Session,
    *,
    conditions: list,
    status: str,
    quantiles: tuple[float, ...],
) -> list[int | None]:
    """返回平均值和各分位数；生产 PG 在数据库内聚合，SQLite 仅作测试回退。"""
    if db.bind is not None and db.bind.dialect.name == "postgresql":
        row = db.execute(
            _comparison_duration_aggregate_stmt(conditions, status, quantiles)
        ).one()
        return [_round_duration_ms(value) for value in row]

    # SQLite 没有 percentile_cont；本地/测试只回退读取耗时单列，不加载完整记录。
    values = list(
        db.execute(
            select(ComparisonRecord.total_ms)
            .where(
                *conditions,
                ComparisonRecord.status == status,
                ComparisonRecord.total_ms.is_not(None),
            )
            .order_by(ComparisonRecord.total_ms)
        ).scalars()
    )
    average = _round_duration_ms(sum(values) / len(values)) if values else None
    return [average, *(_comparison_percentile(values, q) for q in quantiles)]


def comparison_records_summary(
    db: Session,
    *,
    user_id: int | None = None,
    phone: str | None = None,
    status: str | None = None,
    business_type: str | None = None,
    store: str | None = None,
    product: str | None = None,
    date_from: date | None = None,
    date_to: date | None = None,
) -> dict:
    """比价记录页概览聚合；主耗时均值及分位数只取成功记录。"""
    conditions = _comparison_conditions(
        user_id=user_id, phone=phone, status=status, business_type=business_type,
        store=store, product=product, date_from=date_from, date_to=date_to,
    )
    row = db.execute(
        select(
            func.count(ComparisonRecord.id),
            func.sum(case((ComparisonRecord.status.in_(("success", "failed")), 1), else_=0)),
            func.sum(case((ComparisonRecord.status == "success", 1), else_=0)),
            func.avg(ComparisonRecord.llm_cost_yuan),
            func.sum(case((
                (ComparisonRecord.status == "success")
                & (ComparisonRecord.saved_amount_cents > 0), 1
            ), else_=0)),
            func.sum(case((ComparisonRecord.status == "cancelled", 1), else_=0)),
        ).where(*conditions)
    ).one()
    started = int(row[0] or 0)
    completed = int(row[1] or 0)
    success = int(row[2] or 0)
    lower_price = int(row[4] or 0)
    cancelled = int(row[5] or 0)
    success_duration_stats = _comparison_duration_aggregates(
        db,
        conditions=conditions,
        status="success",
        quantiles=(0.05, 0.5, 0.95, 0.99),
    )
    cancelled_duration_stats = _comparison_duration_aggregates(
        db,
        conditions=conditions,
        status="cancelled",
        quantiles=(0.05, 0.5, 0.95),
    )
    success_rate_denominator = started - cancelled
    return {
        "started": started,
        "completed": completed,
        "success": success,
        "success_rate": success / success_rate_denominator if success_rate_denominator else None,
        "avg_token_cost": float(row[3]) if row[3] is not None else None,
        "lower_price_rate": lower_price / success if success else None,
        "avg_duration_ms": success_duration_stats[0],
        "p5_duration_ms": success_duration_stats[1],
        "p50_duration_ms": success_duration_stats[2],
        "p95_duration_ms": success_duration_stats[3],
        "p99_duration_ms": success_duration_stats[4],
        "cancelled": cancelled,
        "cancelled_rate": cancelled / started if started else None,
        "cancelled_p5_ms": cancelled_duration_stats[1],
        "cancelled_p50_ms": cancelled_duration_stats[2],
        "cancelled_p95_ms": cancelled_duration_stats[3],
    }


def get_comparison_record(db: Session, record_id: int) -> ComparisonRecord | None:
    """admin 取单条比价记录(任意用户,不限本人;附 phone/nickname 瞬态)。"""
    rec = db.get(ComparisonRecord, record_id)
    if rec is not None:
        _attach_user_info(db, [rec])
        _attach_comparison_order_status(db, [rec])
        _attach_comparison_device_details([rec])
    return rec


def list_onboarding_devices(db: Session, *, limit: int = 500) -> list[dict]:
    """按设备(device_id, ANDROID_ID)聚合 onboarding_completion:每台设备走过引导的账号数 +
    最近完成时间,按最近完成倒序。设备维度新手引导管理用。没走过引导的设备不在表里(本就会
    引导、无需管理)。当前全量返回(上限 limit;调试期设备少,量大再加分页/搜索)。"""
    rows = db.execute(
        select(
            OnboardingCompletion.device_id,
            func.count(OnboardingCompletion.id),
            func.max(OnboardingCompletion.completed_at),
        )
        .group_by(OnboardingCompletion.device_id)
        .order_by(func.max(OnboardingCompletion.completed_at).desc())
        .limit(limit)
    ).all()
    return [
        {"device_id": did, "account_count": int(cnt), "last_completed_at": last}
        for did, cnt, last in rows
    ]


def _heartbeat_seconds_ago(last: datetime | None) -> int | None:
    """距上次心跳的秒数(兼容 SQLite 取回的 naive datetime,按 UTC 处理)。None = 从没心跳。"""
    if last is None:
        return None
    if last.tzinfo is None:
        last = last.replace(tzinfo=timezone.utc)
    return int((datetime.now(timezone.utc) - last).total_seconds())


def _device_model_from_id(device_id: str) -> str:
    """从 device_id(格式 device_<机型>_<hash>)解析机型;不规范则回退原 id。"""
    parts = device_id.split("_")
    if len(parts) >= 3 and parts[0] == "device":
        return "_".join(parts[1:-1]).replace("_", " ")
    return device_id


def _attach_device_derived(devices: list[DeviceLiveness]) -> None:
    """给每台设备瞬态挂 device_model / online / display_state / offline_seconds(供 schema 读)。

    在线判定:开过无障碍(ever_protected)且距上次心跳 ≤ HEARTBEAT_TIMEOUT_MINUTES。
    从没开过无障碍(ever_protected=False)= never(不算掉线,避免把装了没用的设备误报掉线)。
    offline_seconds 仅 offline 时给值(=掉线时长),在线/从未启用为 None。"""
    timeout_sec = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES)) * 60
    for d in devices:
        d.device_model = _device_model_from_id(d.device_id)
        secs = _heartbeat_seconds_ago(d.last_heartbeat_at)
        if not d.ever_protected:
            d.online = False
            d.display_state = "never"
            d.offline_seconds = None
        elif secs is not None and secs <= timeout_sec:
            d.online = True
            d.display_state = "online"
            d.offline_seconds = None
        else:
            d.online = False
            d.display_state = "offline"
            d.offline_seconds = secs


def _attach_device_user_info(db: Session, devices: list[DeviceLiveness]) -> None:
    """给每台设备瞬态挂归属用户 phone/nickname(同 _attach_user_info,供 admin schema 读)。"""
    uids = {d.user_id for d in devices}
    if not uids:
        return
    rows = db.execute(
        select(User.id, User.phone, User.nickname).where(User.id.in_(uids))
    ).all()
    umap = {uid: (phone, nick) for uid, phone, nick in rows}
    for d in devices:
        phone, nick = umap.get(d.user_id, (None, None))
        d.phone = phone
        d.nickname = nick


def _liveness_cutoff() -> datetime:
    """掉线判定分界:此刻 - HEARTBEAT_TIMEOUT_MINUTES。心跳早于它 = 掉线(同 list_overdue 口径)。"""
    timeout_min = max(1, int(settings.HEARTBEAT_TIMEOUT_MINUTES))
    return datetime.now(timezone.utc) - timedelta(minutes=timeout_min)


def list_device_liveness(
    db: Session,
    *,
    status: str | None = None,
    device_id: str | None = None,
    phone: str | None = None,
    user_id: int | None = None,
    sort_by: str = "status",
    sort_order: str = "desc",
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[DeviceLiveness], int | None, int]:
    """设备存活列表(admin 全量)。按 在线情况(online/offline/never)/ 设备id(包含)/
    归属用户(手机号前缀 或 user_id)筛,offset 分页。join user 取 phone/nickname,派生
    device_model/online/display_state/offline_seconds 挂行上。
    默认排序 status=掉线置顶(offline → online → never;掉线组内掉得最久在前)。"""
    cutoff = _liveness_cutoff()
    stmt = select(DeviceLiveness)
    # 在线情况派生筛选(口径同 _attach_device_derived:ever_protected + 心跳是否过阈值)
    if status == "online":
        stmt = stmt.where(
            DeviceLiveness.ever_protected.is_(True),
            DeviceLiveness.last_heartbeat_at.is_not(None),
            DeviceLiveness.last_heartbeat_at >= cutoff,
        )
    elif status == "offline":
        stmt = stmt.where(
            DeviceLiveness.ever_protected.is_(True),
            DeviceLiveness.last_heartbeat_at.is_not(None),
            DeviceLiveness.last_heartbeat_at < cutoff,
        )
    elif status == "never":
        stmt = stmt.where(DeviceLiveness.ever_protected.is_(False))
    if device_id and device_id.strip():
        stmt = stmt.where(DeviceLiveness.device_id.like(f"%{device_id.strip()}%"))
    if user_id is not None:
        stmt = stmt.where(DeviceLiveness.user_id == user_id)
    if phone:
        stmt = stmt.where(
            DeviceLiveness.user_id.in_(select(User.id).where(User.phone.like(f"{phone}%")))
        )

    if sort_by in ("last_heartbeat_at", "created_at"):
        col = (
            DeviceLiveness.last_heartbeat_at
            if sort_by == "last_heartbeat_at"
            else DeviceLiveness.created_at
        )
        order_fn = asc if sort_order == "asc" else desc
        id_order = asc(DeviceLiveness.id) if sort_order == "asc" else desc(DeviceLiveness.id)
        sort_clause: tuple = (order_fn(col), id_order)
    else:
        # 默认「掉线置顶」:offline(0) → online(1) → never(2);掉线组内按心跳最旧(掉得最久)在前
        rank = case(
            (DeviceLiveness.ever_protected.is_(False), 2),
            (DeviceLiveness.last_heartbeat_at < cutoff, 0),
            else_=1,
        )
        sort_clause = (asc(rank), asc(DeviceLiveness.last_heartbeat_at), desc(DeviceLiveness.id))

    items, next_cursor, total = offset_paginate(db, stmt, sort_clause, limit=limit, cursor=cursor)
    _attach_device_user_info(db, items)
    _attach_device_derived(items)
    return items, next_cursor, total


def device_liveness_stats(db: Session) -> dict:
    """顶部卡片:总设备数 + 在线 / 已掉线 / 未启用(按心跳阈值派生,口径同列表)。"""
    cutoff = _liveness_cutoff()

    def _count(*conds) -> int:
        return int(db.execute(select(func.count(DeviceLiveness.id)).where(*conds)).scalar_one())

    total = int(db.execute(select(func.count(DeviceLiveness.id))).scalar_one())
    never = _count(DeviceLiveness.ever_protected.is_(False))
    online = _count(
        DeviceLiveness.ever_protected.is_(True),
        DeviceLiveness.last_heartbeat_at.is_not(None),
        DeviceLiveness.last_heartbeat_at >= cutoff,
    )
    offline = _count(
        DeviceLiveness.ever_protected.is_(True),
        DeviceLiveness.last_heartbeat_at.is_not(None),
        DeviceLiveness.last_heartbeat_at < cutoff,
    )
    return {"total": total, "online": online, "offline": offline, "never": never}


def list_all_coin_transactions(
    db: Session,
    *,
    user_id: int | None = None,
    biz_type: str | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[CoinTransaction], int | None]:
    stmt = select(CoinTransaction)
    if user_id is not None:
        stmt = stmt.where(CoinTransaction.user_id == user_id)
    if biz_type:
        stmt = stmt.where(CoinTransaction.biz_type == biz_type)
    return cursor_paginate(db, stmt, CoinTransaction.id, limit=limit, cursor=cursor)


def list_all_cash_transactions(
    db: Session,
    *,
    user_id: int | None = None,
    biz_type: str | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[CashTransaction], int | None]:
    stmt = select(CashTransaction)
    if user_id is not None:
        stmt = stmt.where(CashTransaction.user_id == user_id)
    if biz_type:
        stmt = stmt.where(CashTransaction.biz_type == biz_type)
    return cursor_paginate(db, stmt, CashTransaction.id, limit=limit, cursor=cursor)


def list_all_withdraw_orders(
    db: Session,
    *,
    user_id: int | None = None,
    status: str | None = None,
    source: str | None = None,
    keyword: str | None = None,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    date_field: str = "created_at",
    sort_by: str = "created_at",
    sort_order: str = "desc",
    quick_filter: str | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[WithdrawOrder], int | None, int]:
    stmt = select(WithdrawOrder)
    needs_user_join = bool(keyword and keyword.strip()) or quick_filter == "high_risk"
    if needs_user_join:
        stmt = stmt.outerjoin(User, User.id == WithdrawOrder.user_id)

    if user_id is not None:
        stmt = stmt.where(WithdrawOrder.user_id == user_id)
    if status:
        stmt = stmt.where(WithdrawOrder.status == status)
    # 提现类型:coin_cash(福利页提现)/ invite_cash(邀请提现);None=全部
    if source:
        stmt = stmt.where(WithdrawOrder.source == source)

    kw = (keyword or "").strip()
    if kw:
        pattern = f"%{kw}%"
        conditions = [
            WithdrawOrder.out_bill_no.ilike(pattern),
            WithdrawOrder.transfer_bill_no.ilike(pattern),
            WithdrawOrder.user_name.ilike(pattern),
            WithdrawOrder.wechat_state.ilike(pattern),
            WithdrawOrder.fail_reason.ilike(pattern),
            User.phone.ilike(pattern),
            User.nickname.ilike(pattern),
            User.wechat_nickname.ilike(pattern),
        ]
        if kw.isdigit():
            conditions.append(WithdrawOrder.user_id == int(kw))
        stmt = stmt.where(or_(*conditions))

    date_col = WithdrawOrder.updated_at if date_field == "updated_at" else WithdrawOrder.created_at
    if date_from is not None:
        stmt = stmt.where(date_col >= _as_utc(date_from))
    if date_to is not None:
        stmt = stmt.where(date_col <= _as_utc(date_to))

    # tz-aware:列为 timestamptz,比较绝对时刻、与 DB 会话时区无关(同 _as_utc / stats.py)
    now = datetime.now(timezone.utc)
    today_start = (
        datetime.now(ZoneInfo("Asia/Shanghai"))
        .replace(hour=0, minute=0, second=0, microsecond=0)
        .astimezone(timezone.utc)
    )
    if quick_filter == "abnormal":
        stmt = stmt.where(
            or_(
                WithdrawOrder.status.in_(("failed", "rejected")),
                WithdrawOrder.fail_reason.is_not(None),
                WithdrawOrder.wechat_state.in_(("FAIL", "CANCELLED", "CLOSED")),
            )
        )
    elif quick_filter == "failed":
        stmt = stmt.where(WithdrawOrder.status == "failed")
    elif quick_filter == "overdue_reviewing":
        stmt = stmt.where(
            WithdrawOrder.status == "reviewing",
            WithdrawOrder.created_at <= now - timedelta(minutes=30),
        )
    elif quick_filter == "pending_overdue":
        stmt = stmt.where(
            WithdrawOrder.status == "pending",
            WithdrawOrder.updated_at <= now - timedelta(minutes=15),
        )
    elif quick_filter == "today":
        stmt = stmt.where(WithdrawOrder.created_at >= today_start)
    elif quick_filter == "high_risk":
        stmt = stmt.where(User.is_high_risk.is_(True))

    sort_cols = {
        "id": WithdrawOrder.id,
        "created_at": WithdrawOrder.created_at,
        "updated_at": WithdrawOrder.updated_at,
        "amount_cents": WithdrawOrder.amount_cents,
    }
    sort_col = sort_cols.get(sort_by, WithdrawOrder.created_at)
    order_fn = asc if sort_order == "asc" else desc
    id_order = asc(WithdrawOrder.id) if sort_order == "asc" else desc(WithdrawOrder.id)
    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)


def _as_utc(value: datetime) -> datetime:
    """前端传 ISO 时间 → 统一成 tz-aware UTC 再比较。

    所有时间列均为 `DateTime(timezone=True)`(Postgres timestamptz);用 tz-aware 绑定参数
    比较的是绝对时刻,与 DB 会话时区无关、恒正确。曾用 naive UTC,正确性依赖会话 TimeZone=UTC,
    生产会话非 UTC 时筛选边界会整体偏移——故统一 tz-aware(与 stats.py / withdraw_summary 一致)。
    无时区入参按 UTC 解释。"""
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value.astimezone(timezone.utc)


def withdraw_list_enrichment(
    db: Session, user_ids: list[int], *, source: str | None = None
) -> dict[int, dict]:
    """批量富化提现单列表:按本页 user_id 取 手机号/昵称 + 各自累计成功提现金额(分)。

    两条聚合查询搞定(避免逐行 N+1)。累计口径与 get_user_overview 的 withdraw_success_cents
    完全一致:SUM(amount_cents) WHERE status='success'。返回 {user_id: {phone, nickname,
    cumulative_success_cents}};某 user_id 查不到用户则不在 dict 里(路由侧回退到默认值)。
    """
    uniq = list(set(user_ids))
    if not uniq:
        return {}
    success_stmt = select(
        WithdrawOrder.user_id,
        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
    ).where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
    if source:
        success_stmt = success_stmt.where(WithdrawOrder.source == source)
    success_rows = db.execute(success_stmt.group_by(WithdrawOrder.user_id)).all()
    success_map = {uid: int(total) for uid, total in success_rows}
    users = db.execute(
        select(
            User.id,
            User.phone,
            User.nickname,
            User.is_high_risk,
            User.high_risk_note,
        ).where(User.id.in_(uniq))
    ).all()
    return {
        uid: {
            "phone": phone,
            "nickname": nickname,
            "is_high_risk": is_high_risk,
            "high_risk_note": high_risk_note,
            "cumulative_success_cents": success_map.get(uid, 0),
        }
        for uid, phone, nickname, is_high_risk, high_risk_note in users
    }


def list_feedbacks(
    db: Session,
    *,
    status: str | None = None,
    source: str | None = None,
    user_id: int | None = None,
    content: str | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    sort_by: str = "id",
    sort_order: str = "desc",
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[Feedback], int | None, int]:
    """反馈工单列表。支持 状态 / 反馈类型(source) / 用户ID / 内容模糊 / 提交时间范围 筛选,
    按 id·提交时间排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一(同 [list_users]),
    代价是翻页期间数据变动可能错位一条——admin 低频场景可接受。返回 (items, next_cursor, total),
    total 供页码分页。created_at 为 timestamptz,日期入参统一转 tz-aware UTC 比较。"""
    stmt = select(Feedback)
    if status:
        stmt = stmt.where(Feedback.status == status)
    if source:
        stmt = stmt.where(Feedback.source == source)
    if user_id is not None:
        stmt = stmt.where(Feedback.user_id == user_id)
    if content and content.strip():
        stmt = stmt.where(Feedback.content.ilike(f"%{content.strip()}%"))
    if created_from is not None:
        stmt = stmt.where(Feedback.created_at >= _as_utc(created_from))
    if created_to is not None:
        stmt = stmt.where(Feedback.created_at <= _as_utc(created_to))

    sort_cols = {"id": Feedback.id, "created_at": Feedback.created_at}
    sort_col = sort_cols.get(sort_by, Feedback.id)
    order_fn = asc if sort_order == "asc" else desc
    id_order = asc(Feedback.id) if sort_order == "asc" else desc(Feedback.id)
    items, next_cursor, total = offset_paginate(
        db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
    )
    _attach_user_info(db, items)  # 列表展示完整手机号(点手机号查该用户全部反馈)
    _attach_feedback_device_details(db, items)
    return items, next_cursor, total


def feedback_summary(db: Session) -> dict:
    """反馈审核台各状态计数(待审核/已采纳/未采纳/合计)。待审核含历史 new 态(与前端 isPending 一致)。"""
    rows = db.execute(
        select(Feedback.status, func.count(Feedback.id)).group_by(Feedback.status)
    ).all()
    by_status = {status: int(count) for status, count in rows}
    return {
        "pending": by_status.get("pending", 0) + by_status.get("new", 0),
        "adopted": by_status.get("adopted", 0),
        "rejected": by_status.get("rejected", 0),
        "total": sum(by_status.values()),
    }


def list_analytics_events(
    db: Session,
    *,
    event: str | None = None,
    device_id: str | None = None,
    user_id: int | None = None,
    session_id: str | None = None,
    created_from: datetime | None = None,
    created_to: datetime | None = None,
    sort_by: str = "id",
    sort_order: str = "desc",
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[AnalyticsEvent], int | None, int]:
    """埋点日志列表(admin 全量)。按 事件名 / 设备ID前缀 / 用户ID / 会话ID / 接收时间范围 筛选,
    按 id·接收时间排序。offset 分页(同 [list_feedbacks])。created_at 为 timestamptz,
    日期入参统一转 tz-aware UTC 比较。"""
    stmt = select(AnalyticsEvent)
    if event:
        stmt = stmt.where(AnalyticsEvent.event == event)
    if device_id and device_id.strip():
        stmt = stmt.where(AnalyticsEvent.device_id.like(f"{device_id.strip()}%"))
    if user_id is not None:
        stmt = stmt.where(AnalyticsEvent.user_id == user_id)
    if session_id and session_id.strip():
        stmt = stmt.where(AnalyticsEvent.session_id == session_id.strip())
    if created_from is not None:
        stmt = stmt.where(AnalyticsEvent.created_at >= _as_utc(created_from))
    if created_to is not None:
        stmt = stmt.where(AnalyticsEvent.created_at <= _as_utc(created_to))

    sort_cols = {"id": AnalyticsEvent.id, "created_at": AnalyticsEvent.created_at}
    sort_col = sort_cols.get(sort_by, AnalyticsEvent.id)
    order_fn = asc if sort_order == "asc" else desc
    id_order = asc(AnalyticsEvent.id) if sort_order == "asc" else desc(AnalyticsEvent.id)
    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)


def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder | None:
    """按商户单号查提现单(admin 重试打款先拿 user_id 用,M3)。"""
    return db.execute(
        select(WithdrawOrder).where(WithdrawOrder.out_bill_no == out_bill_no)
    ).scalar_one_or_none()


def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
    """提现审核台顶部统计。金额单位:分；口径与列表的 source 筛选一致。"""
    summary_stmt = select(
        WithdrawOrder.status,
        func.count(WithdrawOrder.id),
        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
    )
    if source:
        summary_stmt = summary_stmt.where(WithdrawOrder.source == source)
    rows = db.execute(summary_stmt.group_by(WithdrawOrder.status)).all()
    by_status = {
        status: {"count": int(count), "amount_cents": int(amount_cents)}
        for status, count, amount_cents in rows
    }

    today_start = (
        datetime.now(ZoneInfo("Asia/Shanghai"))
        .replace(hour=0, minute=0, second=0, microsecond=0)
        .astimezone(timezone.utc)
    )

    def _today_count(status: str) -> int:
        stmt = select(func.count(WithdrawOrder.id)).where(
            WithdrawOrder.status == status,
            WithdrawOrder.updated_at >= today_start,
        )
        if source:
            stmt = stmt.where(WithdrawOrder.source == source)
        return db.execute(stmt).scalar_one()

    today_success_amount_stmt = select(
        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)
    ).where(
        WithdrawOrder.status == "success",
        WithdrawOrder.updated_at >= today_start,
    )
    if source:
        today_success_amount_stmt = today_success_amount_stmt.where(
            WithdrawOrder.source == source
        )
    today_success_amount = db.execute(today_success_amount_stmt).scalar_one()

    def _status_count(status: str) -> int:
        return by_status.get(status, {}).get("count", 0)

    return {
        "reviewing_count": _status_count("reviewing"),
        "reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
        "pending_count": _status_count("pending"),
        "success_count": _status_count("success"),
        "rejected_count": _status_count("rejected"),
        "failed_count": _status_count("failed"),
        "total_count": sum(item["count"] for item in by_status.values()),
        "today_success_count": _today_count("success"),
        "today_success_amount_cents": int(today_success_amount),
        "today_rejected_count": _today_count("rejected"),
    }


def invite_overview(
    db: Session,
    inviter_user_id: int,
    *,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
) -> dict:
    """邀请提现详情：邀请关系统计及受邀用户首次比价/首单信息。

    首次记录按自增 id 取最早一条，批量查询避免按受邀用户逐行查询。
    邀请成功口径为完成首次比价并已发邀请奖励(compare_reward_granted)。
    """
    relation_stmt = (
        select(InviteRelation, User)
        .join(User, User.id == InviteRelation.invitee_user_id)
        .where(InviteRelation.inviter_user_id == inviter_user_id)
        .order_by(InviteRelation.created_at.asc(), InviteRelation.id.asc())
    )
    if date_from is not None:
        relation_stmt = relation_stmt.where(User.created_at >= _as_utc_naive(date_from))
    if date_to is not None:
        relation_stmt = relation_stmt.where(User.created_at <= _as_utc_naive(date_to))
    relation_rows = db.execute(relation_stmt).all()
    invitee_ids = [relation.invitee_user_id for relation, _ in relation_rows]
    if not invitee_ids:
        return {"invite_total": 0, "invite_success_total": 0, "items": []}

    first_compare_ids = list(
        db.execute(
            select(func.min(ComparisonRecord.id))
            .where(
                ComparisonRecord.user_id.in_(invitee_ids),
                ComparisonRecord.status == "success",
            )
            .group_by(ComparisonRecord.user_id)
        ).scalars()
    )
    comparisons = (
        db.execute(
            select(
                ComparisonRecord.user_id,
                ComparisonRecord.store_name,
                ComparisonRecord.product_names,
            ).where(ComparisonRecord.id.in_(first_compare_ids))
        ).all()
        if first_compare_ids
        else []
    )
    comparison_map = {
        user_id: (store_name, product_names)
        for user_id, store_name, product_names in comparisons
    }

    first_order_ids = list(
        db.execute(
            select(func.min(SavingsRecord.id))
            .where(
                SavingsRecord.user_id.in_(invitee_ids),
                SavingsRecord.source == "compare",
            )
            .group_by(SavingsRecord.user_id)
        ).scalars()
    )
    orders = (
        db.execute(
            select(
                SavingsRecord.user_id,
                SavingsRecord.shop_name,
                SavingsRecord.title,
                SavingsRecord.dishes,
                SavingsRecord.order_amount_cents,
            ).where(SavingsRecord.id.in_(first_order_ids))
        ).all()
        if first_order_ids
        else []
    )
    order_map = {
        user_id: (
            shop_name,
            title or "、".join(dishes or []),
            order_amount_cents,
        )
        for user_id, shop_name, title, dishes, order_amount_cents in orders
    }

    items = []
    for relation, user in relation_rows:
        compare_store, compare_products = comparison_map.get(user.id, (None, None))
        order_store, order_products, order_amount = order_map.get(
            user.id, (None, None, None)
        )
        items.append(
            {
                "user_id": user.id,
                "phone": user.phone,
                "registered_at": user.created_at,
                "invite_success": bool(relation.compare_reward_granted),
                "first_compare_store": compare_store,
                "first_compare_products": compare_products,
                "first_order_store": order_store,
                "first_order_products": order_products,
                "first_order_amount_cents": order_amount,
            }
        )
    return {
        "invite_total": len(items),
        "invite_success_total": sum(1 for item in items if item["invite_success"]),
        "items": items,
    }


def list_withdraw_audit_logs(
    db: Session, out_bill_no: str, *, limit: int = 20
) -> list[AdminAuditLog]:
    return list(
        db.execute(
            select(AdminAuditLog)
            .where(AdminAuditLog.target_type == "withdraw", AdminAuditLog.target_id == out_bill_no)
            .order_by(AdminAuditLog.id.desc())
            .limit(limit)
        ).scalars().all()
    )


def withdraw_risk_flags(
    order: WithdrawOrder,
    user: User | None,
    recent_withdraws: list[WithdrawOrder],
    cash_balance_cents: int,
) -> tuple[list[str], int]:
    flags: list[str] = []
    if user and user.status != "active":
        flags.append(f"账号状态:{user.status}")
    if user and user.created_at:
        created_at = user.created_at.replace(tzinfo=timezone.utc) if user.created_at.tzinfo is None else user.created_at
        if datetime.now(timezone.utc) - created_at < timedelta(hours=24):
            flags.append("新注册用户")
    # 历史异常提现拆「拒绝」「失败」两类(口径不同:拒绝=人工驳回退款,失败=打款失败退款)
    rejected_n = sum(1 for item in recent_withdraws if item.status == "rejected")
    failed_n = sum(1 for item in recent_withdraws if item.status == "failed")
    if rejected_n:
        flags.append(f"提现拒绝{rejected_n}笔")
    if failed_n:
        flags.append(f"提现失败{failed_n}笔")
    recent_reviewing = sum(1 for item in recent_withdraws if item.status == "reviewing")
    if recent_reviewing >= 3:
        flags.append(f"待审核提现偏多:{recent_reviewing}笔")
    if cash_balance_cents < 0:
        flags.append("现金余额为负")
    score = min(100, len(flags) * 20 + (rejected_n + failed_n) * 10)
    return flags, score


def get_user_overview(db: Session, user_id: int) -> dict | None:
    """用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
    user = db.get(User, user_id)
    if user is None:
        return None
    acc = db.get(CoinAccount, user_id)  # 可能为 None(从未发生过金币动作)

    def _count(model, *conds) -> int:
        return db.execute(select(func.count(model.id)).where(*conds)).scalar_one()

    return {
        "user": user,
        "coin_balance": acc.coin_balance if acc else 0,
        "cash_balance_cents": acc.cash_balance_cents if acc else 0,
        "invite_cash_balance_cents": acc.invite_cash_balance_cents if acc else 0,
        "total_coin_earned": acc.total_coin_earned if acc else 0,
        "comparison_total": _count(ComparisonRecord, ComparisonRecord.user_id == user_id),
        "comparison_success": _count(
            ComparisonRecord,
            ComparisonRecord.user_id == user_id,
            ComparisonRecord.status == "success",
        ),
        "withdraw_total": _count(WithdrawOrder, WithdrawOrder.user_id == user_id),
        "withdraw_success_cents": db.execute(
            select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
                WithdrawOrder.user_id == user_id, WithdrawOrder.status == "success"
            )
        ).scalar_one(),
        "feedback_total": _count(Feedback, Feedback.user_id == user_id),
    }


def _as_utc_naive(value: datetime) -> datetime:
    """窗口入参 → UTC naive(= _as_utc 去时区),与库里按 naive UTC 存取的 created_at 同口径比较。
    历史遗留:_window_conds 一直引用本函数却未定义(自定义区间会 NameError),此处补上。"""
    return _as_utc(value).replace(tzinfo=None)


def _window_conds(col, date_from: datetime | None, date_to: datetime | None) -> list:
    """把 [date_from, date_to] 转成对 col(created_at)的过滤条件;都为 None = 全量(注册至今)。"""
    conds = []
    if date_from is not None:
        conds.append(col >= _as_utc_naive(date_from))
    if date_to is not None:
        conds.append(col <= _as_utc_naive(date_to))
    return conds


def _coins_to_cents(coins: int) -> int:
    """累计金币折算成可提现现金(分)。COIN_PER_YUAN=10000 → 100 金币 = 1 分。"""
    return round(coins / (rewards.COIN_PER_YUAN / 100))


def user_reward_stats(
    db: Session,
    user_id: int,
    *,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    withdraw_source: str | None = None,
    app_env: str | None = None,
    revenue_scope: str = "all",
    feed_scene: str | None = None,
) -> dict:
    """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。

    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
    传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
    """
    withdraw_source_conds = (
        [WithdrawOrder.source == withdraw_source] if withdraw_source else []
    )
    wd_success = db.execute(
        select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
            WithdrawOrder.user_id == user_id,
            WithdrawOrder.status == "success",
            *withdraw_source_conds,
            *_window_conds(WithdrawOrder.created_at, date_from, date_to),
        )
    ).scalar_one()
    wd_total = db.execute(
        select(func.count(WithdrawOrder.id)).where(
            WithdrawOrder.user_id == user_id,
            *withdraw_source_conds,
            *_window_conds(WithdrawOrder.created_at, date_from, date_to),
        )
    ).scalar_one()

    acc = db.get(CoinAccount, user_id)  # 现金余额:当前快照,不随窗口
    cash_balance = (
        (
            acc.invite_cash_balance_cents
            if withdraw_source == "invite_cash"
            else acc.cash_balance_cents
        )
        if acc
        else 0
    )

    # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
    # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
    business_ids: set[str] | None = None
    if revenue_scope == "business":
        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
        from app.admin.repositories.ad_revenue import business_code_ids

        business_ids = business_code_ids(db, app_env)

    rv_conds = [
        AdRewardRecord.user_id == user_id,
        AdRewardRecord.reward_scene == "reward_video",
        AdRewardRecord.status == "granted",
        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
    ]
    if app_env is not None:
        rv_conds.append(AdRewardRecord.app_env == app_env)
    if business_ids is not None:
        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
    rv = db.execute(
        select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
            *rv_conds,
        )
    ).all()
    rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
    rv_coins = sum(r.coin for r in rv)

    feed_reward_conds = [
        AdFeedRewardRecord.user_id == user_id,
        AdFeedRewardRecord.status == "granted",
        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
    ]
    if app_env is not None:
        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
    if feed_scene is not None:
        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
    if business_ids is not None:
        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
    feed_rewards = db.execute(
        select(
            AdFeedRewardRecord.unit_count,
            AdFeedRewardRecord.coin,
        ).where(
            *feed_reward_conds,
        )
    ).all()
    feed_coins = sum(f.coin for f in feed_rewards)

    feed_impression_conds = [
        AdEcpmRecord.user_id == user_id,
        AdEcpmRecord.ad_type.in_(("draw", "feed")),
        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
    ]
    if app_env is not None:
        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
    if feed_scene is not None:
        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
    if business_ids is not None:
        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
    feed_impressions = db.execute(
        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
    ).all()
    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]

    trad_coins = db.execute(
        select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
            CoinTransaction.user_id == user_id,
            CoinTransaction.amount > 0,
            CoinTransaction.biz_type.not_in(_NON_TASK_BIZ_TYPES),
            *_window_conds(CoinTransaction.created_at, date_from, date_to),
        )
    ).scalar_one()

    return {
        "withdraw_success_cents": int(wd_success),
        "cash_balance_cents": int(cash_balance),
        "withdraw_total": int(wd_total),
        "traditional_task_cash_cents": _coins_to_cents(int(trad_coins)),
        "reward_video_count": len(rv),
        "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
        "reward_video_cash_cents": _coins_to_cents(rv_coins),
        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
        "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
        "feed_cash_cents": _coins_to_cents(feed_coins),
    }


def _cn_wall_to_utc(dt: datetime) -> datetime:
    """coin_transaction 存的是北京 wall-clock(naive,见 wallet.grant_coins「存北京 wall-clock」),转成 UTC naive,
    与广告表(func.now() UTC)统一 —— 让本函数按同一绝对时刻排序、且前端 apiTime(把无时区时间当 UTC 再 +8 展示)
    口径一致;否则签到会比实际多显示 8 小时(北京时间又被 +8)。"""
    return dt.replace(tzinfo=rewards.CN_TZ).astimezone(timezone.utc).replace(tzinfo=None)


def _coin_record_sort_key(row: dict) -> datetime:
    """金币明细跨数据源排序键：兼容 SQLite naive 与 PostgreSQL aware 时间。"""
    return _as_utc(row["created_at"])


def user_coin_records(
    db: Session,
    user_id: int,
    *,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[dict], int | None, int]:
    """金币发放记录(提现详情底部表),按时间倒序、offset 游标分页 + 总数。

    合并三类来源(Phase1 信息流不分比价/领券):
      - 激励视频 / 签到膨胀 = ad_reward_record(granted)
      - 信息流广告        = ad_feed_reward_record(granted)
      - 签到             = coin_transaction(biz_type='signin')
    每来源各取 offset+limit+1 条(倒序)再归并切片,避免 union 全量拉取。
    """
    offset = max(cursor or 0, 0)
    fetch = offset + limit + 1
    rows: list[dict] = []
    # coin_transaction 存北京 wall-clock(其余表存 UTC);签到窗口边界 +8h 对齐北京,过滤/计数才不偏移 8 小时
    signin_from = date_from + timedelta(hours=8) if date_from is not None else None
    signin_to = date_to + timedelta(hours=8) if date_to is not None else None

    # 三类来源都只取页面需要的列，避免无关 ORM 新列造成旧库查询失败。
    for rec in db.execute(
        select(
            AdRewardRecord.reward_scene,
            AdRewardRecord.created_at,
            AdRewardRecord.ecpm_raw,
            AdRewardRecord.coin,
        )
        .where(
            AdRewardRecord.user_id == user_id,
            AdRewardRecord.status == "granted",
            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
        )
        .order_by(AdRewardRecord.created_at.desc())
        .limit(fetch)
    ).all():
        is_video = rec.reward_scene == "reward_video"
        rows.append({
            "source": rec.reward_scene,
            "source_label": "激励视频" if is_video else "签到膨胀",
            "created_at": rec.created_at,
            "ecpm": rec.ecpm_raw,
            "coin": rec.coin,
        })

    for rec in db.execute(
        select(
            AdFeedRewardRecord.feed_scene,
            AdFeedRewardRecord.created_at,
            AdFeedRewardRecord.ecpm_raw,
            AdFeedRewardRecord.coin,
        )
        .where(
            AdFeedRewardRecord.user_id == user_id,
            AdFeedRewardRecord.status == "granted",
            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
        )
        .order_by(AdFeedRewardRecord.created_at.desc())
        .limit(fetch)
    ).all():
        rows.append({
            "source": "feed",
            "source_label": _FEED_SCENE_LABEL.get(rec.feed_scene, "信息流广告"),
            "created_at": rec.created_at,
            "ecpm": rec.ecpm_raw,
            "coin": rec.coin,
        })

    for rec in db.execute(
        select(CoinTransaction.created_at, CoinTransaction.amount)
        .where(
            CoinTransaction.user_id == user_id,
            CoinTransaction.biz_type == "signin",
            *_window_conds(CoinTransaction.created_at, signin_from, signin_to),
        )
        .order_by(CoinTransaction.created_at.desc())
        .limit(fetch)
    ).all():
        rows.append({
            "source": "signin",
            "source_label": "签到",
            # 北京 wall-clock → UTC,与广告记录统一(前端 apiTime 会 +8 回北京展示,不然签到会多 8 小时)
            "created_at": _cn_wall_to_utc(rec.created_at),
            "ecpm": None,
            "coin": rec.amount,
        })

    # SQLite 常返回 naive datetime，PostgreSQL timestamptz 返回 aware datetime；
    # 统一成 aware UTC 排序，避免线上合并广告记录与签到记录时抛
    # “can't compare offset-naive and offset-aware datetimes”。
    rows.sort(key=_coin_record_sort_key, reverse=True)
    has_more = len(rows) > offset + limit

    # 总数 = 三源在窗口内 granted 计数之和(供前端页码分页渲染页码/共 N 条)
    def _count(model, *conds) -> int:
        return int(db.execute(select(func.count()).select_from(model).where(*conds)).scalar_one())

    total = (
        _count(
            AdRewardRecord, AdRewardRecord.user_id == user_id,
            AdRewardRecord.status == "granted",
            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
        )
        + _count(
            AdFeedRewardRecord, AdFeedRewardRecord.user_id == user_id,
            AdFeedRewardRecord.status == "granted",
            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
        )
        + _count(
            CoinTransaction, CoinTransaction.user_id == user_id,
            CoinTransaction.biz_type == "signin",
            *_window_conds(CoinTransaction.created_at, signin_from, signin_to),
        )
    )
    return rows[offset:offset + limit], (offset + limit if has_more else None), total


def _attach_price_report_comparison(db: Session, records: list[PriceReport]) -> None:
    """给每条上报瞬态挂关联比价记录的 trace + 设备/版本快照:
    trace_id/trace_url(点 Trace 看完整比价过程)、device_model/rom_name/android_version(机型OS版本列)、
    app_version(提交版本号列,= 提交时我们 app 的 versionName)。
    comparison_record_id 为空 / 关联记录查不到 → 全 None。"""
    cids = {r.comparison_record_id for r in records if r.comparison_record_id is not None}
    rows = (
        db.execute(
            select(
                ComparisonRecord.id,
                ComparisonRecord.trace_id,
                ComparisonRecord.trace_url,
                ComparisonRecord.device_model,
                ComparisonRecord.rom_name,
                ComparisonRecord.android_version,
                ComparisonRecord.app_version,
            ).where(ComparisonRecord.id.in_(cids))
        ).all()
        if cids
        else []
    )
    cmap = {row.id: row for row in rows}
    for r in records:
        row = cmap.get(r.comparison_record_id)
        r.trace_id = row.trace_id if row else None
        r.trace_url = row.trace_url if row else None
        r.device_model = row.device_model if row else None
        r.rom_name = row.rom_name if row else None
        r.android_version = row.android_version if row else None
        r.app_version = row.app_version if row else None


def list_price_reports(
    db: Session,
    *,
    status: str | None = None,
    user_id: int | None = None,
    sort_by: str = "id",
    sort_order: str = "desc",
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[PriceReport], int | None, int]:
    """上报更低价列表(admin 全量,可按状态/用户筛)。offset 分页 + total,按 id·提交时间排序。
    join User 取 phone 瞬态挂(列表展示完整手机号);按 comparison_record_id 挂 trace_id/trace_url。"""
    stmt = select(PriceReport)
    if status:
        stmt = stmt.where(PriceReport.status == status)
    if user_id is not None:
        stmt = stmt.where(PriceReport.user_id == user_id)

    sort_cols = {"id": PriceReport.id, "created_at": PriceReport.created_at}
    sort_col = sort_cols.get(sort_by, PriceReport.id)
    order_fn = asc if sort_order == "asc" else desc
    id_order = asc(PriceReport.id) if sort_order == "asc" else desc(PriceReport.id)
    items, next_cursor, total = offset_paginate(
        db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
    )
    _attach_user_info(db, items)
    _attach_price_report_comparison(db, items)
    return items, next_cursor, total


def price_report_summary(db: Session) -> dict:
    """上报审核台各状态计数(待审核/已通过/已拒绝/合计)。"""
    rows = db.execute(
        select(PriceReport.status, func.count(PriceReport.id)).group_by(PriceReport.status)
    ).all()
    by_status = {status: int(count) for status, count in rows}
    return {
        "pending": by_status.get("pending", 0),
        "approved": by_status.get("approved", 0),
        "rejected": by_status.get("rejected", 0),
        "total": sum(by_status.values()),
    }
