perf(pricebot): 透传复用共享 httpx 单例,免每请求重建 SSL 上下文 + 绕过进程代理

coupon/compare 透传原先每请求 async with httpx.AsyncClient(...) 新建: 每次重建 SSL 上下文(加载 certifi CA)实测 ~1s+,而 pricebot 是纯 HTTP 透传根本用不到 TLS;且 trust_env 默认 True 会把 http://localhost:8000 经进程代理(Clash)再绕几秒。抽 app/core/pricebot_client.py 共享单例:trust_env=False 直连,lifespan 启动预热(SSL 一次性成本付在启动)、关停 aclose;timeout 下放到 .post() 保留 coupon 30s / compare 60s 差异。keep-alive 复用 TCP,每帧降到个位数 ms。
2026-06-28 09:28:46 +08:00
17 changed files with 32 additions and 8549 deletions
@@ -5,8 +5,7 @@ user.last_login_at / comparison_record.status / withdraw_order.status)要加索
 """
 from __future__ import annotations
-from datetime import date, datetime, time, timedelta, timezone
+from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -14,27 +13,12 @@ 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.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"
 def _beijing_today_start_utc() -> datetime:
@@ -47,87 +31,30 @@ def _beijing_today_start_utc() -> datetime:
 def today_dau(db: Session) -> int:
    """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。
-    广告收益报表复用这个函数;历史窗口 DAU 由 dashboard_overview 的 period 口径另算。
+    大盘与广告收益报表共用此口径,单一来源避免漂移。
    ⚠️ last_login_at 是单值字段(只存最后一次登录时刻),故只能算「今日」,
    无法回溯历史某天的 DAU——调用方按此约束决定历史区间是否展示。
    """
    today_start = _beijing_today_start_utc()
-    return int(
+    return db.execute(
-        db.execute(select(func.count(User.id)).where(User.last_login_at >= today_start)).scalar_one()
+        select(func.count(User.id)).where(User.last_login_at >= today_start)
-    )
+    ).scalar_one()
-def _default_period_end() -> date:
+def dashboard_overview(db: Session) -> dict:
    """新版大盘不含今日,默认窗口结束日=北京时间昨天。"""
    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 _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 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())
+        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 int(db.execute(stmt).scalar_one())
+        return 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(
@@ -147,204 +74,6 @@ def dashboard_overview(
    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_user_ids = _user_id_set(
        select(ComparisonRecord.user_id).where(*period_comparison_conds)
    )
    coupon_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_user_ids | coupon_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_user_ids = _user_id_set(
            select(ComparisonRecord.user_id).where(*daily_comparison_conds)
        )
        daily_coupon_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_user_ids | daily_coupon_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_meituan_orders = list(
        db.execute(
            select(CpsOrder).where(
                CpsOrder.pay_time >= start_utc,
                CpsOrder.pay_time < end_utc,
            )
        ).scalars()
    )
    period_meituan_valid_orders = [
        o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES
    ]
    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": {
@@ -403,61 +132,7 @@ def dashboard_overview(
            "success": comparison_success,
            "success_rate": success_rate,
        },
-        "period": {
+        "feedback": {"new": _count(Feedback, Feedback.status.in_(("pending", "new")))},
-            "date_from": period_from,
+        # CPS 收入数据源未接(referral-link 只换链接,转化/佣金未回收)→ 前端显示"待接入"。
-            "date_to": period_to,
+        "cps": {"available": False, "note": "CPS 转化数据未接入(P2)"},
            "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)+已上报比价记录+领券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": "美团 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,
        },
    }
@@ -7,7 +7,7 @@
 from __future__ import annotations
 import time
-from datetime import date as _date, datetime, time as _dt_time, timedelta, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Annotated
 from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
@@ -340,77 +340,24 @@ def generate_referral_links(
 # ───────────── 订单对账 ─────────────
 _BEIJING = timezone(timedelta(hours=8))
 def _parse_day(value: str | None, *, field: str) -> _date | None:
    if value is None:
        return None
    try:
        return _date.fromisoformat(value)
    except ValueError as e:
        raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e
 def _reconcile_range_to_ts(
    date_from: _date | None, date_to: _date | None, days: int
 ) -> tuple[int, int]:
    if date_from is None and date_to is None:
        now = int(time.time())
        return now - days * 86400, now
    start_day = date_from or date_to
    end_day = date_to or date_from
    if start_day is None or end_day is None:
        raise HTTPException(status_code=422, detail="日期参数不完整")
    if start_day > end_day:
        start_day, end_day = end_day, start_day
    if (end_day - start_day).days + 1 > 90:
        raise HTTPException(status_code=422, detail="美团订单查询最长 90 天")
    start_dt = datetime.combine(start_day, _dt_time.min, tzinfo=_BEIJING)
    end_dt = datetime.combine(end_day + timedelta(days=1), _dt_time.min, tzinfo=_BEIJING)
    return int(start_dt.timestamp()), int(end_dt.timestamp())
@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账")
 def reconcile_orders(
    request: Request,
    admin: Annotated[AdminUser, Depends(require_role("finance"))],
    db: AdminDb,
-    date_from: Annotated[str | None, Query(description="起始日 YYYY-MM-DD")] = None,
+    days: Annotated[int, Query(ge=1, le=90)] = 7,
    date_to: Annotated[str | None, Query(description="结束日 YYYY-MM-DD")] = None,
    days: Annotated[int, Query(ge=1, le=90, description="未传日期时默认拉近 N 天")] = 7,
    sid: Annotated[str | None, Query(max_length=64)] = None,
    query_time_type: Annotated[int, Query(ge=1, le=2)] = 1,
 ) -> CpsReconcileResult:
-    start_ts, end_ts = _reconcile_range_to_ts(
+    now = int(time.time())
        _parse_day(date_from, field="date_from"),
        _parse_day(date_to, field="date_to"),
        days,
    )
    try:
        result = cps_repo.reconcile_orders(
-            db,
+            db, start_time=now - days * 86400, end_time=now, sid=sid,
            start_time=start_ts,
            end_time=end_ts,
            query_time_type=query_time_type,
            sid=sid,
        )
    except MeituanCpsError as e:
        raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e
    write_audit(
        db, admin, action="cps.order.reconcile", target_type="cps_order", target_id=None,
-        detail={
+        detail={"days": days, "sid": sid, **result}, ip=get_client_ip(request), commit=True,
            "date_from": date_from,
            "date_to": date_to,
            "days": days,
            "sid": sid,
            "query_time_type": query_time_type,
            **result,
        },
        ip=get_client_ip(request),
        commit=True,
    )
    return CpsReconcileResult(**result)
@@ -1,9 +1,7 @@
 """admin 数据大盘(只读聚合)。"""
 from __future__ import annotations
-from datetime import date
+from fastapi import APIRouter, Depends
 from fastapi import APIRouter, Depends, Query
 from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import stats
@@ -17,11 +15,5 @@ router = APIRouter(
@router.get("/overview", response_model=DashboardOverview, summary="大盘核心指标")
-def overview(
+def overview(db: AdminDb) -> DashboardOverview:
-    db: AdminDb,
+    return DashboardOverview.model_validate(stats.dashboard_overview(db))
    date_from: date | None = Query(None, description="北京时间自然日起始日 YYYY-MM-DD"),
    date_to: date | None = Query(None, description="北京时间自然日结束日 YYYY-MM-DD"),
 ) -> DashboardOverview:
    return DashboardOverview.model_validate(
        stats.dashboard_overview(db, date_from=date_from, date_to=date_to)
    )
@@ -1,8 +1,6 @@
 """admin 大盘 schemas(对应 stats.dashboard_overview 的嵌套结构)。"""
 from __future__ import annotations
 from datetime import date
 from pydantic import BaseModel
@@ -40,56 +38,6 @@ class DashboardComparison(BaseModel):
    success_rate: float
 class DashboardPeriodUsers(BaseModel):
    new: int
    active: int
    retained_new_users: int
    retention_rate: float | None = None
    retention_note: str
 class DashboardPeriodComparison(BaseModel):
    total: int
    success: int
    success_rate: float
    ordered: int
    average_duration_ms: int | None = None
    average_saved_cents: int | None = None
 class DashboardPeriodCoins(BaseModel):
    granted_total: int
    reward_video_coin_total: int = 0
    feed_ad_coin_total: int = 0
    signin_coin_total: int = 0
    signin_boost_coin_total: int = 0
    task_coin_total: int = 0
    coupon_reward_coin_total: int = 0
    comparison_reward_coin_total: int = 0
    regular_task_coin_total: int = 0
 class DashboardPeriodCash(BaseModel):
    withdraw_success_cents: int
 class DashboardTrendPoint(BaseModel):
    date: date
    active_users: int
    new_users: int
    comparisons: int
 class DashboardPeriod(BaseModel):
    date_from: date
    date_to: date
    users: DashboardPeriodUsers
    comparison: DashboardPeriodComparison
    coins: DashboardPeriodCoins
    cash: DashboardPeriodCash
    trend: list[DashboardTrendPoint] = []
 class DashboardFeedback(BaseModel):
    new: int
@@ -97,12 +45,6 @@ class DashboardFeedback(BaseModel):
 class DashboardCps(BaseModel):
    available: bool
    note: str
    meituan_order_count: int = 0
    meituan_commission_cents: int = 0
    meituan_hit_count: int = 0
    meituan_miss_count: int = 0
    meituan_unknown_rate_count: int = 0
    meituan_hit_rate: float | None = None
 class DashboardOverview(BaseModel):
@@ -110,6 +52,5 @@ class DashboardOverview(BaseModel):
    coins: DashboardCoins
    cash: DashboardCash
    comparison: DashboardComparison
    period: DashboardPeriod
    feedback: DashboardFeedback
    cps: DashboardCps
@@ -11,7 +11,6 @@ from __future__ import annotations
 import hmac
 import logging
 from datetime import datetime, timedelta, timezone
 from typing import Annotated
 from fastapi import APIRouter, Header, HTTPException, status
@@ -22,7 +21,6 @@ from app.repositories import launch_confirm_sample as repo
 from app.schemas.launch_confirm_sample import (
    LaunchConfirmSampleIn,
    LaunchConfirmSampleOut,
    LaunchConfirmSampleRow,
 )
 logger = logging.getLogger("shagua.internal.launch_confirm")
@@ -63,34 +61,3 @@ def report_launch_confirm_sample(
        payload.exec_success, payload.trace_id,
    )
    return LaunchConfirmSampleOut(id=sid)
@router.get(
    "/launch-confirm-samples",
    response_model=list[LaunchConfirmSampleRow],
    summary="启动确认窗兜底样本列表(沉淀脚本 distill 读; server→server, 不走 JWT)",
 )
 def list_launch_confirm_samples(
    db: DbSession,
    x_internal_secret: Annotated[str | None, Header()] = None,
    exec_success: bool | None = None,
    host_package: str | None = None,
    since_days: int | None = None,
    limit: int = 1000,
 ) -> list[LaunchConfirmSampleRow]:
    """读样本供 pricebot 的 distill_launch_confirm.py 聚合沉淀回 PROFILES。
    与上报端点同一把共享密钥;exec_success/host_package/since_days 均可选,limit 默认 1000。
    """
    _check_secret(x_internal_secret)
    since = None
    if since_days and since_days > 0:
        since = datetime.now(timezone.utc) - timedelta(days=since_days)
    rows = repo.list_samples(
        db,
        exec_success=exec_success,
        host_package=host_package,
        since=since,
        limit=limit,
    )
    return [LaunchConfirmSampleRow.model_validate(r) for r in rows]
@@ -19,7 +19,6 @@ from app.models.coupon_state import (  # noqa: F401
    CouponDailyCompletion,
    CouponPromptEngagement,
 )
 from app.models.cps_order import CpsOrder  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401
 from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401
@@ -2,10 +2,8 @@
 from __future__ import annotations
 import logging
 from datetime import datetime
 from typing import Optional
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 from app.models.launch_confirm_sample import LaunchConfirmSample
@@ -35,27 +33,3 @@ def insert_sample(db: Session, payload: LaunchConfirmSampleIn) -> int:
    db.commit()
    db.refresh(row)
    return row.id
 def list_samples(
    db: Session,
    *,
    exec_success: Optional[bool] = None,
    host_package: Optional[str] = None,
    since: Optional[datetime] = None,
    limit: int = 1000,
 ) -> list[LaunchConfirmSample]:
    """按条件查样本(pricebot 的 distill_launch_confirm.py 沉淀工具读)。created_at 升序。
    过滤项都可空:exec_success(只看放行成功的)/ host_package(只看某宿主包)/
    since(>= created_at)。limit 兜底防一次拉爆全表。
    """
    stmt = select(LaunchConfirmSample)
    if exec_success is not None:
        stmt = stmt.where(LaunchConfirmSample.exec_success == exec_success)
    if host_package:
        stmt = stmt.where(LaunchConfirmSample.host_package == host_package)
    if since is not None:
        stmt = stmt.where(LaunchConfirmSample.created_at >= since)
    stmt = stmt.order_by(LaunchConfirmSample.created_at.asc()).limit(limit)
    return list(db.execute(stmt).scalars().all())
@@ -123,7 +123,6 @@ class ComparisonRecordIn(BaseModel):
    # pricebot done.params.trace_url 原样上报,落库供记录页「复制调试链接」(dir 名含落盘
    # 时分秒前端拼不出,必须由后端透传)。
    trace_url: str | None = Field(None, description="本次比价公网调试链接")
    total_ms: int | None = Field(None, description="整场比价墙钟耗时(ms)")
    # ===== debug 维度(客户端采集上报;旧客户端不带 → None。仅 admin 比价记录页用)=====
    # 必须显式声明,否则 model_dump() 落 raw_payload 时被 pydantic 静默丢弃(同上面 coupon_saved 的坑)。
@@ -173,7 +172,6 @@ class ComparisonRecordOut(BaseModel):
    items: list = []
    comparison_results: list = []
    skipped_dish_names: list = []
    total_ms: int | None = None
    # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
    # 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。
    ordered: bool = False
@@ -1,9 +1,7 @@
 """启动确认窗兜底样本的内部上报模型(pricebot → app-server)。"""
 from __future__ import annotations
-from datetime import datetime
+from pydantic import BaseModel
 from pydantic import BaseModel, ConfigDict
 class LaunchConfirmSampleIn(BaseModel):
@@ -26,20 +24,3 @@ class LaunchConfirmSampleOut(BaseModel):
    """落库结果。"""
    id: int
 class LaunchConfirmSampleRow(BaseModel):
    """单条样本(列表读出;沉淀脚本 distill 用,payload 原样带出)。"""
    model_config = ConfigDict(from_attributes=True)
    id: int
    created_at: datetime
    trace_id: str | None = None
    device_id: str | None = None
    host_package: str | None = None
    target_app: str | None = None
    system_locale: str | None = None
    exec_success: bool = False
    dialog_title: str | None = None
    payload: dict | None = None
@@ -37,7 +37,7 @@ B 安装并首启 App
        └─ POST /api/v1/invite/bind { invite_code, channel="clipboard" }
        ↓
 后端 repositories/invite.py bind()
-  └─ 过四道防线 → 建 invite_relation(邀请金币已下线,不写金币流水)
+  └─ 过四道防线 → 建 invite_relation + 给 A、B 各发金币(同事务原子提交)
 ```
 手动填码这条:B 在邀请页输码 → `InviteRepository.bindManual()` → `POST /bind { channel="manual" }` → 同一个 `bind()`。
@@ -52,20 +52,20 @@ B 安装并首启 App
 |---|---|
 | 端点 | `app/api/v1/invite.py`:`GET /api/v1/invite/me`(返回 `invite_code` + `share_url` + 战绩)、`POST /api/v1/invite/bind`(绑定,`channel` = `clipboard` / `manual`)。**均需 Bearer 鉴权**。 |
 | share_url 构造 | `invite.py` 的 `my_invite`:`settings.INVITE_LANDING_URL + "?ref=" + code`。`INVITE_LANDING_URL` 在 `app/core/config.py`(默认 `https://app-api.shaguabijia.com/media/dl.html`)。 |
-| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 兼容累计金币字段,当前恒为 0)。 |
+| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 累计金币)。 |
 | 数据模型 | `app/models/invite.py` 的 `InviteRelation`(`inviter_user_id` / `invitee_user_id` / `channel` / `status` / `inviter_coin` / `invitee_coin` / `created_at`)+ `app/models/user.py` 的 `User.invite_code` 列。 |
 | 迁移 | `alembic/versions/invite_code_and_relation.py`:给 `user` 加 `invite_code`(唯一索引)+ 建 `invite_relation` 表。`down_revision = 11a1d08c6f55`。 |
 | 收发模型 | `app/schemas/invite.py`:`InviteInfoOut` / `BindInviteIn` / `BindInviteOut`。 |
-| 新人窗口 | `app/core/rewards.py`:`INVITE_NEW_USER_WINDOW_HOURS`(72)。邀请金币已下线,不再配置邀请金币常量。 |
+| 奖励常量 | `app/core/rewards.py`:`INVITE_INVITER_COINS` / `INVITE_INVITEE_COINS`(各 10000 = 1 元)、`INVITE_NEW_USER_WINDOW_HOURS`(72)。 |
 **`bind()` 的四道防线(防重复 / 防刷,看 `repositories/invite.py`):**
-1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复绑定)。
+1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复发奖)。
 2. **自邀屏蔽**:`inviter == invitee` → `self_invite`。
-3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才生效,挡存量老用户互相填码刷关系 → 否则 `not_eligible`。
+3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才发奖,挡存量老用户互相填码薅羊毛 → 否则 `not_eligible`。
 4. **手机号唯一**(天然限量):每个 B = 一个真实手机号账号。
-邀请金币已下线:`bind()` 只记录绑定关系,不再写 `coin_transaction`;响应里的金币字段保留兼容旧客户端,当前恒为 0。
+发金币复用 `repositories/wallet.py` 的 `grant_coins`,与建关系记录在**同一事务**提交,保证"建关系 + 双方加金币"原子。
 ### 3.2 前端(shaguabijia-app-android)
@@ -120,7 +120,7 @@ B 安装并首启 App
 ### 4.4 测试硬约束 / 坑(都是机制,不是 bug)
 - **B 必须用新手机号**:`invitee_user_id` 唯一,一个 B 只能绑一次;反复测要换号(或手删 `invite_relation` 那行 + 回滚金币)。
- **72h 新人闸**:B 注册后 72 小时内绑定才生效(刚注册肯定满足)。
+- **72h 新人闸**:B 注册后 72 小时内绑才发奖(刚注册肯定满足)。
 - **A ≠ B**:自邀被屏蔽。
 - **B 从点下载到首启 App 之间别复制别的东西**:剪贴板会被覆盖 → 归因丢(剪贴板 deferred deeplink 的固有脆弱性)。
 - **笔记本 IP 别变**:debug 包把 `BASE_URL` 的 IP 烧死在编译期,DHCP 一换就连不上 → 给笔记本固定个 LAN IP。
@@ -1,52 +0,0 @@
 /**
 * SGApi —— H5 调 app-server 后端的薄封装。
 *
 * 同源：H5 由 app-server 的 /media/h5/ 托管，后端在 /api/v1，同 host:port → 用相对路径、无 CORS。
 * 鉴权：JWT Bearer。token 经 SGBridge.getToken() 从原生取(原生持登录态)；浏览器调试走 bridge 的 mock token。
 * 401：交原生拉登录(requestLogin)兜底，本次请求按失败 reject；正式的 refresh 重试策略阶段2 再补。
 *
 * 依赖 shared/bridge.js 先加载(取 token)。
 */
 (function (global) {
  'use strict';
  var BASE = '/api/v1';
  function authHeaders() {
    var t = (global.SGBridge && global.SGBridge.getToken()) || '';
    var h = { 'Content-Type': 'application/json' };
    if (t) h['Authorization'] = 'Bearer ' + t;
    return h;
  }
  function handle(res) {
    if (res.status === 401) {
      // 未授权：拉原生登录(异步)，本次请求按失败处理，调用方自行决定是否重试
      if (global.SGBridge) global.SGBridge.requestLogin();
      return Promise.reject(new Error('unauthorized'));
    }
    if (!res.ok) {
      return res.text().then(function (t) {
        return Promise.reject(new Error('http ' + res.status + ' ' + t));
      });
    }
    // 204 / 空体兜底
    return res.text().then(function (t) { return t ? JSON.parse(t) : null; });
  }
  /** GET /api/v1<path>。path 以 / 开头，如 '/savings/battle'。 */
  function apiGet(path) {
    return fetch(BASE + path, { method: 'GET', headers: authHeaders() }).then(handle);
  }
  /** POST /api/v1<path>，body 自动 JSON 序列化。 */
  function apiPost(path, body) {
    return fetch(BASE + path, {
      method: 'POST',
      headers: authHeaders(),
      body: JSON.stringify(body || {}),
    }).then(handle);
  }
  global.SGApi = { base: BASE, get: apiGet, post: apiPost };
 })(window);
@@ -1,162 +0,0 @@
 /**
 * SGBridge —— 傻瓜比价 H5 ↔ Android 原生 的桥。
 *
 * 背景：四个主 tab(首页/福利/记录/我的)由原生 Compose 改造为 WebView 加载本工程 H5。
 * H5 只负责"画 + 取后端数据"；凡需要原生能力(登录态 / 跳转 / 跳外卖 App / 比价领券 /
 * 权限 / 定位 / Toast / 激励视频)，一律经本桥调用原生。
 *
 * 协议两个方向：
 *  ① H5 → 原生：Android 端 WebView.addJavascriptInterface(obj, "SGBridgeNative")。
 *     obj 方法都是【同步】：查询类返回 String(JSON 或纯串)，动作类无返回。
 *  ② 原生 → H5：原生执行 evaluateJavascript("window.SGBridge._emit('<event>', '<json>')")。
 *     事件：onAuthChange(登录态变) / onBalanceChange(余额变) / onSigninChange(签到态变) / onResume(回前台刷新)。
 *
 * 离线兜底：浏览器里(无 SGBridgeNative)走 MOCK，便于不装 App 直接在本地 server 调样式 / 渲染。
 * 与原生实现对应：见 shaguabijia-app-android 的 SGBridge.kt(方法名逐个对齐本文件)。
 */
 (function (global) {
  'use strict';
  var native = global.SGBridgeNative || null;
  var hasNative = !!native;
  // ---- 离线 MOCK(仅无原生时生效，便于浏览器调试渲染；真机一律走 native) ----
  var MOCK = {
    authState: { loggedIn: true, userId: 1, nickname: '冰', avatarUrl: '', phone: '188****8888' },
    token: 'mock-token-for-browser-debug',
    deviceId: 'browser-debug-device',
    appVersion: '0.0.0-debug',
  };
  function safeParse(s, fallback) {
    try { return s ? JSON.parse(s) : fallback; } catch (e) { return fallback; }
  }
  // ====== 查询类(同步返回) ======
  /** 当前登录态 + 用户基本信息 → {loggedIn, userId, nickname, avatarUrl, phone}。 */
  function getAuthState() {
    if (hasNative && native.getAuthState) return safeParse(native.getAuthState(), { loggedIn: false });
    return MOCK.authState;
  }
  /** 后端鉴权用的 access token(空串=未登录)。原生持登录态，H5 调后端前取它拼 Bearer。 */
  function getToken() {
    if (hasNative && native.getToken) return native.getToken() || '';
    return MOCK.token;
  }
  /** 设备唯一标识(心跳 / 领券状态查询等用)。 */
  function getDeviceId() {
    if (hasNative && native.getDeviceId) return native.getDeviceId() || '';
    return MOCK.deviceId;
  }
  /** App 版本号。 */
  function getAppVersion() {
    if (hasNative && native.getAppVersion) return native.getAppVersion() || '';
    return MOCK.appVersion;
  }
  /** 已安装的目标电商/外卖 App 包名数组(原生 InstalledApps 探测)。H5 选平台弹窗据此判真实装机态。 */
  function getInstalledApps() {
    if (hasNative && native.getInstalledApps) return safeParse(native.getInstalledApps(), []);
    // MOCK(浏览器无原生):给主流已装,便于本地预览选平台弹窗正常显示"有"。
    return ['com.sankuai.meituan', 'com.taobao.taobao', 'com.jingdong.app.mall', 'me.ele'];
  }
  /** 今日是否已领券(置灰「去领取」→「去查看」)。原生读 CompareButtonState(SP 按天);无桥默认 false。 */
  function getCouponClaimedToday() {
    if (hasNative && native.getCouponClaimedToday) return !!native.getCouponClaimedToday();
    return false;
  }
  // ====== 动作类(无返回；异步结果走事件) ======
  /** 跳原生页。route 取值对齐安卓 Routes(invite / settings / feedback / withdrawal / compareRecords / reportFlow / guideVideo / compareResult / coinHistory / cashHistory / welfareRules ...)。 */
  function navigate(route) {
    if (hasNative && native.navigate) native.navigate(route);
    else console.log('[SGBridge mock] navigate →', route);
  }
  /** 拉起极光一键登录。结果异步经 onAuthChange 事件回来(不在此函数返回)。 */
  function requestLogin() {
    if (hasNative && native.requestLogin) native.requestLogin();
    else console.log('[SGBridge mock] requestLogin');
  }
  /** 原生居中 Toast。 */
  function toast(msg) {
    if (hasNative && native.toast) native.toast(String(msg));
    else console.log('[SGBridge mock] toast →', msg);
  }
  /** 跳美团/外卖 App(deeplink 优先；空则原生按包名启动，未装可跳应用商店)。 */
  function openMeituan(deeplink) {
    if (hasNative && native.openMeituan) native.openMeituan(deeplink || '');
    else console.log('[SGBridge mock] openMeituan →', deeplink);
  }
  /** 触发 agent 比价流程(原生起无障碍引擎)。 */
  function startCompare() {
    if (hasNative && native.startCompare) native.startCompare();
    else console.log('[SGBridge mock] startCompare');
  }
  /** 触发一键领券(原生先校验悬浮窗/无障碍权限，再起前台服务)。platforms: string[]。 */
  function startCouponClaim(platforms) {
    var json = JSON.stringify(platforms || []);
    if (hasNative && native.startCouponClaim) native.startCouponClaim(json);
    else console.log('[SGBridge mock] startCouponClaim →', json);
  }
  /** 按候选包名拉起对应平台 App。照原生 HomePicker 点击跳转:原生在该平台一组候选包里取首个可
   *  getLaunchIntentForPackage 的拉起(NEW_TASK|CLEAR_TASK 冷启到平台首页)。packages: string[](一个平台一组候选包,任一可拉即拉)。 */
  function launchApp(packages) {
    var json = JSON.stringify(packages || []);
    if (hasNative && native.launchApp) native.launchApp(json);
    else console.log('[SGBridge mock] launchApp →', json);
  }
  // ====== 原生 → H5 事件总线 ======
  var listeners = {}; // event → [fn]
  /** 订阅原生事件。返回取消订阅函数。 */
  function on(event, fn) {
    (listeners[event] || (listeners[event] = [])).push(fn);
    return function off() {
      listeners[event] = (listeners[event] || []).filter(function (f) { return f !== fn; });
    };
  }
  /** 供原生回调：window.SGBridge._emit('onAuthChange', '{...}')。payload 可为 JSON 串或对象。 */
  function _emit(event, payload) {
    var data = typeof payload === 'string' ? safeParse(payload, payload) : payload;
    (listeners[event] || []).forEach(function (fn) {
      try { fn(data); } catch (e) { console.error('[SGBridge] listener error', event, e); }
    });
  }
  global.SGBridge = {
    hasNative: hasNative,
    // 查询
    getAuthState: getAuthState,
    getToken: getToken,
    getDeviceId: getDeviceId,
    getAppVersion: getAppVersion,
    getInstalledApps: getInstalledApps,
    getCouponClaimedToday: getCouponClaimedToday,
    // 动作
    navigate: navigate,
    requestLogin: requestLogin,
    toast: toast,
    openMeituan: openMeituan,
    startCompare: startCompare,
    startCouponClaim: startCouponClaim,
    launchApp: launchApp,
    // 事件
    on: on,
    _emit: _emit,
  };
 })(window);
@@ -305,12 +305,11 @@ def test_feed_reward_grants_by_10_second_units(client) -> None:
        "duration_seconds": 30,
        "adn": "pangle",
        "slot_id": "slot_feed",
        "display_coin": 4,
    }
    r = client.post("/api/v1/ad/feed-reward", json=payload, headers=_auth(token))
    assert r.status_code == 200, r.text
    body = r.json()
-    expected = 4
+    expected = sum(calculate_ad_reward_coin("200", i) for i in range(1, 4))
    assert body["granted"] is True
    assert body["status"] == "granted"
    assert body["unit_count"] == 3
@@ -64,8 +64,7 @@ def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
    assert data["users"]["total"] >= 1
    assert data["coins"]["granted_total"] >= 5000
    assert "success_rate" in data["comparison"]
-    assert data["cps"]["available"] is True
+    assert data["cps"]["available"] is False
    assert "meituan_order_count" in data["cps"]
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
@@ -44,7 +44,6 @@ def _food_payload(trace_id: str) -> dict:
        "skipped_dish_names": ["黑牛肉卷"],
        "total_dish_count": 3,
        "information": "在美团找到同店,到手价 ¥123.50",
        "total_ms": 12345,
    }
@@ -72,8 +71,6 @@ def test_report_and_derive(client) -> None:
    assert d["information"] == "在美团找到同店,到手价 ¥123.50"
    assert d["store_name"] == "海底捞(朝阳店)"
    assert d["total_dish_count"] == 3
    assert d["total_ms"] == 12345
    assert d["raw_payload"]["total_ms"] == 12345
    assert d["skipped_dish_count"] == 1
    assert d["skipped_dish_names"] == ["黑牛肉卷"]
    assert len(d["comparison_results"]) == 3
@@ -52,8 +52,7 @@ def call_raw(path: str, body_obj: dict) -> dict:
    }
    url = f"{settings.MT_CPS_HOST}{path}"
    t0 = time.time()
-    # trust_env=False: 美团是国内域名,强制直连绕开本机代理(代理会掐断 TLS 握手,报 SSL EOF)
+    resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
    resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC, trust_env=False)
    ms = int((time.time() - t0) * 1000)
    try:
        j = resp.json()