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

"""admin 大盘聚合查询(全局 count/sum/DAU/成功率)。全部只读、不改任何数据。

⚠️ 性能:这些是全表 count/sum,P0 数据量小够用;用户量上来后热点字段(user.created_at /
user.last_login_at / comparison_record.status / withdraw_order.status)要加索引,或改增量统计表。
"""
from __future__ import annotations

from datetime import datetime, timedelta, timezone

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

from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.comparison import ComparisonRecord
from app.models.feedback import Feedback
from app.models.signin import SigninBoostRecord, SigninRecord
from app.models.user import User
from app.models.wallet import CoinTransaction, WithdrawOrder

_BEIJING = timezone(timedelta(hours=8))


def _beijing_today_start_utc() -> datetime:
    """北京时间今天 0 点对应的 UTC 时刻(DAU / 今日新增按北京时区切天)。"""
    now_bj = datetime.now(_BEIJING)
    start_bj = now_bj.replace(hour=0, minute=0, second=0, microsecond=0)
    return start_bj.astimezone(timezone.utc)


def today_dau(db: Session) -> int:
    """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。

    大盘与广告收益报表共用此口径,单一来源避免漂移。
    ⚠️ last_login_at 是单值字段(只存最后一次登录时刻),故只能算「今日」,
    无法回溯历史某天的 DAU——调用方按此约束决定历史区间是否展示。
    """
    today_start = _beijing_today_start_utc()
    return db.execute(
        select(func.count(User.id)).where(User.last_login_at >= today_start)
    ).scalar_one()


def dashboard_overview(db: Session) -> dict:
    today_start = _beijing_today_start_utc()

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

    def _sum(col, *conds) -> int:
        stmt = select(func.coalesce(func.sum(col), 0))
        if conds:
            stmt = stmt.where(*conds)
        return db.execute(stmt).scalar_one()

    # ===== 用户 =====
    by_status = dict(
        db.execute(select(User.status, func.count(User.id)).group_by(User.status)).all()
    )

    # ===== 提现状态分布 =====
    wd_by_status = dict(
        db.execute(
            select(WithdrawOrder.status, func.count(WithdrawOrder.id)).group_by(
                WithdrawOrder.status
            )
        ).all()
    )

    # ===== 比价 =====
    comparison_total = _count(ComparisonRecord)
    comparison_success = _count(ComparisonRecord, ComparisonRecord.status == "success")
    success_rate = round(comparison_success / comparison_total, 4) if comparison_total else 0.0

    return {
        "users": {
            "total": _count(User),
            "active": by_status.get("active", 0),
            "disabled": by_status.get("disabled", 0),
            "deleted": by_status.get("deleted", 0),
            "new_today": _count(User, User.created_at >= today_start),
            "dau": today_dau(db),
        },
        "coins": {
            # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
            "granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
            "reward_video_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
            ),
            "reward_video_watch_count": _count(
                AdRewardRecord,
                AdRewardRecord.reward_scene == "reward_video",
                AdRewardRecord.status == "granted",
            ),
            "feed_ad_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type == "feed_ad_reward",
            ),
            "feed_ad_watch_count": _count(
                AdFeedRewardRecord,
                AdFeedRewardRecord.status == "granted",
            ),
            "signin_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type == "signin",
            ),
            "signin_count": _count(SigninRecord),
            "signin_boost_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type == "signin_boost",
            ),
            "signin_boost_watch_count": _count(SigninBoostRecord),
        },
        "cash": {
            "withdraw_success_cents": _sum(
                WithdrawOrder.amount_cents, WithdrawOrder.status == "success"
            ),
            "withdraw_pending_count": wd_by_status.get("pending", 0),
            "withdraw_success_count": wd_by_status.get("success", 0),
            "withdraw_failed_count": wd_by_status.get("failed", 0),
        },
        "comparison": {
            "total": comparison_total,
            "success": comparison_success,
            "success_rate": success_rate,
        },
        "feedback": {"new": _count(Feedback, Feedback.status.in_(("pending", "new")))},
        # CPS 收入数据源未接(referral-link 只换链接,转化/佣金未回收)→ 前端显示"待接入"。
        "cps": {"available": False, "note": "CPS 转化数据未接入(P2)"},
    }