shaguabijia-app-server/app/admin/repositories/queries.py

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

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

from sqlalchemy import Select, func, select
from sqlalchemy.orm import Session

from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.user import User
from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder


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 list_users(
    db: Session,
    *,
    phone: str | None = None,
    register_channel: str | None = None,
    status: str | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[User], int | None]:
    stmt = select(User)
    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)
    return cursor_paginate(db, stmt, User.id, limit=limit, cursor=cursor)


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,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[WithdrawOrder], int | None]:
    stmt = select(WithdrawOrder)
    if user_id is not None:
        stmt = stmt.where(WithdrawOrder.user_id == user_id)
    if status:
        stmt = stmt.where(WithdrawOrder.status == status)
    return cursor_paginate(db, stmt, WithdrawOrder.id, limit=limit, cursor=cursor)


def list_feedbacks(
    db: Session,
    *,
    status: str | None = None,
    user_id: int | None = None,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[Feedback], int | None]:
    stmt = select(Feedback)
    if status:
        stmt = stmt.where(Feedback.status == status)
    if user_id is not None:
        stmt = stmt.where(Feedback.user_id == user_id)
    return cursor_paginate(db, stmt, Feedback.id, 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 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,
        "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),
    }