shaguabijia-app-server/app/repositories/savings.py

"""省钱 CRUD:累计/战绩聚合 + 明细分页(纯真实比价上报数据)。

设计:聚合逻辑都是"生产级"真实计算(SUM、按日分组、连续天数),数据只来自真实比价
下单上报(`source='compare'`,见 create_from_report)。

⚠️ 2026-06 起**停用 demo 兜底**:比价下单上报链路已通(Android `PriceBotService` →
`POST /api/v1/order/report`),新用户无真实记录时直接显示空/0(配合「我的」页省钱战绩
卡的「未比价锁定态」)。历史遗留的 `source='demo'` 行不再被任何统计口径计入。

聚合在 Python 里算(每用户记录量很小),避免 SQLite 跨时区 date 比较的坑——
created_at 带时区,统一转成北京时间的 date 再分组。
"""
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta

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

from app.core.rewards import CN_TZ, cn_today
from app.models.savings import SavingsRecord
from app.schemas.order import OrderReportRequest


@dataclass
class SavingsSummary:
    total_saved_cents: int
    order_count: int
    avg_saved_cents: int  # 平均每单省(分)


@dataclass
class SavingsBattle:
    week_saved_cents: int  # 本周(周一起)已省
    beat_percent: int  # 超过百分之多少用户
    streak_days: int  # 连续省钱天数
    compare_count: int  # 累计完成比价次数(= 真实 compare 上报记录数)


def _local_date(dt: datetime):
    """把(可能带时区的)created_at 转成北京时间的 date。"""
    if dt.tzinfo is None:
        return dt.date()
    return dt.astimezone(CN_TZ).date()


def _compare_records(db: Session, user_id: int) -> list[SavingsRecord]:
    """统计/明细口径:只用真实比价上报(`source='compare'`)记录。

    2026-06 起停用 demo 兜底:无真实记录直接返回空(新用户显示 0/锁定态)。
    历史 `source='demo'` 行不计入。
    """
    stmt = select(SavingsRecord).where(
        SavingsRecord.user_id == user_id, SavingsRecord.source == "compare"
    )
    return list(db.execute(stmt).scalars().all())


def get_summary(db: Session, user_id: int) -> SavingsSummary:
    records = _compare_records(db, user_id)
    total = sum(r.saved_amount_cents for r in records)
    count = len(records)
    avg = total // count if count else 0
    return SavingsSummary(total_saved_cents=total, order_count=count, avg_saved_cents=avg)


def _streak_days(dates: set) -> int:
    """从最近活跃日往前数连续有省钱记录的天数。"""
    if not dates:
        return 0
    cur = max(dates)
    streak = 0
    while cur in dates:
        streak += 1
        cur = cur - timedelta(days=1)
    return streak


def get_battle(db: Session, user_id: int) -> SavingsBattle:
    records = _compare_records(db, user_id)

    today = cn_today()
    week_start = today - timedelta(days=today.weekday())  # 本周一
    week_saved = sum(
        r.saved_amount_cents for r in records if _local_date(r.created_at) >= week_start
    )

    dates = {_local_date(r.created_at) for r in records}
    streak = _streak_days(dates)

    beat_percent = _compute_beat_percent(db, user_id)

    return SavingsBattle(
        week_saved_cents=week_saved,
        beat_percent=beat_percent,
        streak_days=streak,
        compare_count=len(records),
    )


def _compute_beat_percent(db: Session, user_id: int) -> int:
    """超过百分之多少用户:按"累计省下金额"在所有有真实(compare)记录的用户中做真实分位。

    = 严格少于我的其他用户数 / 其他用户总数 * 100。
    全部省得比我少 → 100;只有自己一个用户 → 0(无可比)。
    """
    rows = db.execute(
        select(
            SavingsRecord.user_id,
            func.sum(SavingsRecord.saved_amount_cents),
        )
        .where(SavingsRecord.source == "compare")
        .group_by(SavingsRecord.user_id)
    ).all()
    totals = {uid: (total or 0) for uid, total in rows}
    others = {uid: t for uid, t in totals.items() if uid != user_id}
    if not others:
        return 0
    my_total = totals.get(user_id, 0)
    beaten = sum(1 for t in others.values() if t < my_total)
    return round(100 * beaten / len(others))


def list_records(
    db: Session,
    user_id: int,
    *,
    limit: int = 20,
    cursor: int | None = None,
) -> tuple[list[SavingsRecord], int | None]:
    """省钱明细分页(按 id 倒序,游标式)。只列真实比价上报(`source='compare'`)记录。"""
    stmt = select(SavingsRecord).where(
        SavingsRecord.user_id == user_id, SavingsRecord.source == "compare"
    )
    if cursor is not None:
        stmt = stmt.where(SavingsRecord.id < cursor)
    stmt = stmt.order_by(SavingsRecord.id.desc()).limit(limit)

    items = list(db.execute(stmt).scalars().all())
    next_cursor = items[-1].id if len(items) == limit else None
    return items, next_cursor


def create_from_report(
    db: Session, user_id: int, req: OrderReportRequest
) -> tuple[SavingsRecord, bool]:
    """真实比价上报 → 写入一条 savings_record(source='compare')。返回 (记录, 是否重复上报)。

    省额 = 源平台原价 − 实付(下限 0);原价缺失则记 0。
    幂等:同 (user_id, client_event_id) 已存在则直接返回旧记录,duplicated=True,不新增。
    """
    existing = db.execute(
        select(SavingsRecord).where(
            SavingsRecord.user_id == user_id,
            SavingsRecord.client_event_id == req.client_event_id,
        )
    ).scalar_one_or_none()
    if existing is not None:
        return existing, True

    paid = req.paid_amount_cents
    original = req.original_price_cents
    saved = max(0, original - paid) if original is not None else 0
    rec = SavingsRecord(
        user_id=user_id,
        order_amount_cents=paid,
        saved_amount_cents=saved,
        original_price_cents=original,
        compared_price_cents=req.compared_price_cents,
        platform=req.platform,
        title=req.shop_name,  # 明细卡标题用门店名
        shop_name=req.shop_name,
        dishes=req.dishes or [],
        pay_channel=req.pay_channel,
        platform_package=req.platform_package,
        source_platform_name=req.source_platform_name,
        source_deeplink=req.source_deeplink,
        client_event_id=req.client_event_id,
        device_id=req.device_id,
        source="compare",
        # created_at 显式存 naive 北京 wall-clock(与 demo 行、聚合 _local_date 的 naive 分支一致)。
        # 不用列默认 server_default=func.now()——SQLite 下它返回 UTC,会让明细页时间早 8 小时。
        # ⚠️ 生产 PG(timestamptz)对 naive 的解释与 SQLite 不同,迁前端到 PG 时需确认时区一致。
        created_at=datetime.now(CN_TZ).replace(tzinfo=None),
    )
    db.add(rec)
    db.commit()
    db.refresh(rec)
    return rec, False