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 date, datetime, time, timedelta, timezone
from decimal import Decimal, InvalidOperation

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.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.coupon_state import CouponPromptEngagement
from app.models.cps_order import CpsOrder
from app.models.feedback import Feedback
from app.models.savings import SavingsRecord
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))
COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
    *COUPON_REWARD_BIZ_TYPES,
    *COMPARISON_REWARD_BIZ_TYPES,
    *EXCLUDED_REWARD_BIZ_TYPES,
    *UNCLASSIFIED_FEED_BIZ_TYPES,
)
MEITUAN_CPS_INVALID_STATUSES = ("4", "5")
MEITUAN_CPS_SETTLED_STATUS = "6"
COMPARE_START_EVENT = "real_compare_start"
COUPON_START_EVENT = "real_coupon_start"
JD_CPS_INVALID_CODES = {
    "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
    "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
}
JD_CPS_UNPAID_CODES = {"15"}


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):登录 + 开始比价 + 开始领券,按用户去重。

    广告收益报表复用这个函数;历史窗口 DAU 由 dashboard_overview 的 period 口径另算。
    """
    today_bj = datetime.now(_BEIJING).date()
    today_start = _beijing_today_start_utc()
    tomorrow_start = today_start + timedelta(days=1)
    login_user_ids = _id_set(
        db,
        select(User.id).where(User.last_login_at >= today_start, User.last_login_at < tomorrow_start),
    )
    compare_start_user_ids = _event_user_ids(
        db, (COMPARE_START_EVENT,), today_start, tomorrow_start
    )
    coupon_event_user_ids = _event_user_ids(
        db, (COUPON_START_EVENT,), today_start, tomorrow_start
    )
    coupon_claim_user_ids = _id_set(
        db,
        select(CouponPromptEngagement.user_id).where(
            CouponPromptEngagement.engage_date == today_bj,
            CouponPromptEngagement.engage_type == "claim_started",
        ),
    )
    return len(login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids)


def _default_period_end() -> date:
    """新版大盘不含今日,默认窗口结束日=北京时间昨天。"""
    return datetime.now(_BEIJING).date() - timedelta(days=1)


def _normalize_period(date_from: date | None, date_to: date | None) -> tuple[date, date]:
    end = date_to or _default_period_end()
    start = date_from or end
    if start > end:
        start, end = end, start
    return start, end


def _period_bounds(date_from: date, date_to: date) -> tuple[datetime, datetime, datetime, datetime]:
    """返回同一北京自然日窗口的 UTC aware 边界和北京 naive 边界。

    user.created_at / last_login_at 是 UTC aware 口径;比较/金币等历史上有北京 naive
    写入,所以两套边界同时保留。
    """
    start_bj = datetime.combine(date_from, time.min, tzinfo=_BEIJING)
    end_bj = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=_BEIJING)
    start_utc = start_bj.astimezone(timezone.utc)
    end_utc = end_bj.astimezone(timezone.utc)
    return (
        start_utc,
        end_utc,
        start_bj.replace(tzinfo=None),
        end_bj.replace(tzinfo=None),
    )


def _date_range(date_from: date, date_to: date) -> list[date]:
    days = (date_to - date_from).days
    return [date_from + timedelta(days=i) for i in range(days + 1)]


def _id_set(db: Session, stmt) -> set[int]:
    return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}


def _event_user_ids(
    db: Session, event_names: tuple[str, ...], start_utc: datetime, end_utc: datetime
) -> set[int]:
    return _id_set(
        db,
        select(AnalyticsEvent.user_id).where(
            AnalyticsEvent.user_id.is_not(None),
            AnalyticsEvent.event.in_(event_names),
            AnalyticsEvent.created_at >= start_utc,
            AnalyticsEvent.created_at < end_utc,
        ),
    )


def _commission_rate_percent(raw: str | None) -> Decimal | None:
    """美团 commissionRate 原值: "300"=3%, "10"=0.1%;也兼容 "3%"。"""
    if raw is None:
        return None
    s = str(raw).strip()
    if not s:
        return None
    try:
        if s.endswith("%"):
            return Decimal(s[:-1])
        val = Decimal(s)
    except (InvalidOperation, ValueError):
        return None
    return val / Decimal("100")


def _jd_valid_order(order: CpsOrder) -> bool:
    code = str(order.jd_valid_code).strip() if order.jd_valid_code is not None else ""
    return bool(code and code not in JD_CPS_INVALID_CODES and code not in JD_CPS_UNPAID_CODES)


def dashboard_overview(
    db: Session, *, date_from: date | None = None, date_to: date | None = None
) -> dict:
    today_start = _beijing_today_start_utc()
    period_from, period_to = _normalize_period(date_from, date_to)
    start_utc, end_utc, start_local, end_local = _period_bounds(period_from, period_to)

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

    def _user_id_set(stmt) -> set[int]:
        return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}

    # ===== 用户 =====
    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
    period_comparison_conds = (
        ComparisonRecord.created_at >= start_local,
        ComparisonRecord.created_at < end_local,
    )
    period_comparison_total = _count(ComparisonRecord, *period_comparison_conds)
    period_comparison_success = _count(
        ComparisonRecord,
        *period_comparison_conds,
        ComparisonRecord.status == "success",
    )
    period_comparison_success_rate = (
        round(period_comparison_success / period_comparison_total, 4)
        if period_comparison_total
        else 0.0
    )
    period_saved_positive_count = _count(
        ComparisonRecord,
        *period_comparison_conds,
        ComparisonRecord.status == "success",
        ComparisonRecord.saved_amount_cents > 0,
    )
    period_saved_positive_sum = _sum(
        ComparisonRecord.saved_amount_cents,
        *period_comparison_conds,
        ComparisonRecord.status == "success",
        ComparisonRecord.saved_amount_cents > 0,
    )
    period_avg_saved_cents = (
        round(period_saved_positive_sum / period_saved_positive_count)
        if period_saved_positive_count
        else None
    )
    period_avg_duration_ms = db.execute(
        select(func.avg(ComparisonRecord.total_ms)).where(
            *period_comparison_conds,
            ComparisonRecord.total_ms.is_not(None),
            ComparisonRecord.total_ms > 0,
        )
    ).scalar_one()
    period_avg_duration_ms = (
        round(float(period_avg_duration_ms))
        if period_avg_duration_ms is not None
        else None
    )

    ordered_exists = (
        select(SavingsRecord.id)
        .where(
            SavingsRecord.user_id == ComparisonRecord.user_id,
            SavingsRecord.source == "compare",
            SavingsRecord.shop_name.is_not(None),
            SavingsRecord.shop_name == ComparisonRecord.store_name,
        )
        .exists()
    )
    period_ordered_count = _count(
        ComparisonRecord,
        *period_comparison_conds,
        ComparisonRecord.store_name.is_not(None),
        ordered_exists,
    )

    # ===== 日期窗口用户 =====
    period_new_user_ids = _user_id_set(
        select(User.id).where(User.created_at >= start_utc, User.created_at < end_utc)
    )
    login_user_ids = _user_id_set(
        select(User.id).where(User.last_login_at >= start_utc, User.last_login_at < end_utc)
    )
    compare_start_user_ids = _event_user_ids(
        db, (COMPARE_START_EVENT,), start_utc, end_utc
    )
    coupon_event_user_ids = _event_user_ids(
        db, (COUPON_START_EVENT,), start_utc, end_utc
    )
    coupon_claim_user_ids = _user_id_set(
        select(CouponPromptEngagement.user_id).where(
            CouponPromptEngagement.engage_date >= period_from,
            CouponPromptEngagement.engage_date <= period_to,
            CouponPromptEngagement.engage_type == "claim_started",
        )
    )
    period_active_user_ids = (
        login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids
    )
    period_retained_new_user_ids = period_new_user_ids & period_active_user_ids
    period_retention_rate = (
        round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4)
        if period_new_user_ids
        else None
    )
    trend_points: list[dict] = []
    for cur_date in _date_range(period_from, period_to):
        day_start_utc, day_end_utc, day_start_local, day_end_local = _period_bounds(
            cur_date, cur_date
        )
        daily_comparison_conds = (
            ComparisonRecord.created_at >= day_start_local,
            ComparisonRecord.created_at < day_end_local,
        )
        daily_login_user_ids = _user_id_set(
            select(User.id).where(
                User.last_login_at >= day_start_utc,
                User.last_login_at < day_end_utc,
            )
        )
        daily_compare_start_user_ids = _event_user_ids(
            db, (COMPARE_START_EVENT,), day_start_utc, day_end_utc
        )
        daily_coupon_event_user_ids = _event_user_ids(
            db, (COUPON_START_EVENT,), day_start_utc, day_end_utc
        )
        daily_coupon_claim_user_ids = _user_id_set(
            select(CouponPromptEngagement.user_id).where(
                CouponPromptEngagement.engage_date == cur_date,
                CouponPromptEngagement.engage_type == "claim_started",
            )
        )
        trend_points.append(
            {
                "date": cur_date,
                "active_users": len(
                    daily_login_user_ids
                    | daily_compare_start_user_ids
                    | daily_coupon_event_user_ids
                    | daily_coupon_claim_user_ids
                ),
                "new_users": _count(
                    User,
                    User.created_at >= day_start_utc,
                    User.created_at < day_end_utc,
                ),
                "comparisons": _count(ComparisonRecord, *daily_comparison_conds),
            }
        )

    period_coin_conds = (
        CoinTransaction.created_at >= start_local,
        CoinTransaction.created_at < end_local,
        CoinTransaction.amount > 0,
    )
    period_reward_video_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
    )
    period_feed_ad_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type == "feed_ad_reward",
    )
    period_signin_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type == "signin",
    )
    period_signin_boost_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type == "signin_boost",
    )
    period_task_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.like("task_%"),
    )
    period_coupon_reward_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
    )
    period_comparison_reward_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
    )
    period_regular_task_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES),
    )
    period_cps_orders = list(
        db.execute(
            select(CpsOrder).where(
                CpsOrder.pay_time >= start_utc,
                CpsOrder.pay_time < end_utc,
            )
        ).scalars()
    )
    period_meituan_orders = [
        o for o in period_cps_orders if (o.platform or "meituan") == "meituan"
    ]
    period_meituan_valid_orders = [
        o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES
    ]
    period_jd_orders = [o for o in period_cps_orders if o.platform == "jd"]
    period_jd_valid_orders = [o for o in period_jd_orders if _jd_valid_order(o)]
    period_jd_invalid_orders = [
        o for o in period_jd_orders if o.jd_valid_code and not _jd_valid_order(o)
    ]
    period_meituan_hit_count = 0
    period_meituan_miss_count = 0
    period_meituan_unknown_rate_count = 0
    for order in period_meituan_valid_orders:
        rate = _commission_rate_percent(order.commission_rate)
        if rate is None:
            period_meituan_unknown_rate_count += 1
        elif rate < Decimal("1"):
            period_meituan_miss_count += 1
        else:
            period_meituan_hit_count += 1
    period_meituan_hit_denominator = period_meituan_hit_count + period_meituan_miss_count
    period_meituan_hit_rate = (
        round(period_meituan_hit_count / period_meituan_hit_denominator, 4)
        if period_meituan_hit_denominator
        else None
    )

    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,
        },
        "period": {
            "date_from": period_from,
            "date_to": period_to,
            "users": {
                "new": len(period_new_user_ids),
                "active": len(period_active_user_ids),
                "retained_new_users": len(period_retained_new_user_ids),
                "retention_rate": period_retention_rate,
                "retention_note": (
                    "口径:登录(last_login_at)+开始比价(real_compare_start)+"
                    "开始领券(real_coupon_start/claim_started),按用户去重"
                ),
            },
            "comparison": {
                "total": period_comparison_total,
                "success": period_comparison_success,
                "success_rate": period_comparison_success_rate,
                "ordered": period_ordered_count,
                "average_duration_ms": period_avg_duration_ms,
                "average_saved_cents": period_avg_saved_cents,
            },
            "coins": {
                "granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
                "reward_video_coin_total": period_reward_video_coin_total,
                "feed_ad_coin_total": period_feed_ad_coin_total,
                "signin_coin_total": period_signin_coin_total,
                "signin_boost_coin_total": period_signin_boost_coin_total,
                "task_coin_total": period_task_coin_total,
                "coupon_reward_coin_total": period_coupon_reward_coin_total,
                "comparison_reward_coin_total": period_comparison_reward_coin_total,
                "regular_task_coin_total": period_regular_task_coin_total,
            },
            "cash": {
                "withdraw_success_cents": _sum(
                    WithdrawOrder.amount_cents,
                    WithdrawOrder.status == "success",
                    WithdrawOrder.created_at >= start_local,
                    WithdrawOrder.created_at < end_local,
                ),
            },
            "trend": trend_points,
        },
        "feedback": {
            "new": _count(Feedback, Feedback.status.in_(("pending", "new"))),
        },
        "cps": {
            "available": True,
            "note": "美团/JD CPS 读 cps_order 对账订单;淘宝佣金暂空",
            "meituan_order_count": len(period_meituan_valid_orders),
            "meituan_commission_cents": sum(
                o.commission_cents or 0 for o in period_meituan_valid_orders
            ),
            "meituan_hit_count": period_meituan_hit_count,
            "meituan_miss_count": period_meituan_miss_count,
            "meituan_unknown_rate_count": period_meituan_unknown_rate_count,
            "meituan_hit_rate": period_meituan_hit_rate,
            "jd_order_count": len(period_jd_valid_orders),
            # 数据大盘京东 CPS 只看实际佣金,不再用预估佣金兜底。
            "jd_commission_cents": sum(
                o.actual_commission_cents or 0 for o in period_jd_valid_orders
            ),
            "jd_actual_commission_cents": sum(
                o.actual_commission_cents or 0 for o in period_jd_valid_orders
            ),
            "jd_estimated_commission_cents": sum(
                o.estimated_commission_cents or 0 for o in period_jd_valid_orders
            ),
            "jd_invalid_count": len(period_jd_invalid_orders),
        },
    }