Merge branch 'main' of https://gitea.shaguabijia.com/WonderableAI/shaguabijia-app-server into feat/client-applog-ingest

# Conflicts:
#	.env.example
#	docs/api/README.md

.env.example

@@ -27,55 +27,7 @@ JG_PRIVATE_KEY_PATH=./secrets/jverify_rsa_private.pem
 JG_VERIFY_ENDPOINT=https://api.verification.jpush.cn/v1/web/loginTokenVerify
 JG_REQUEST_TIMEOUT_SEC=15
 
-# ===== 厂商直推(无障碍保护存活告警 + 消息中心 13 类通知)=====
-# 敏感密钥只放 .env / 服务器环境变量,不要提交到 git。
-# 各厂商配置状态可随时 GET /api/v1/push/vendors 查看(缺哪些键一目了然)。
-ANDROID_PACKAGE_NAME=com.jishisongfu.shaguabijia
-PUSH_REQUEST_TIMEOUT_SEC=15
-PUSH_TIME_TO_LIVE_SEC=86400
-
-HONOR_PUSH_APP_ID=
-HONOR_PUSH_CLIENT_ID=
-HONOR_PUSH_CLIENT_SECRET=
-HONOR_PUSH_TOKEN_ENDPOINT=https://iam.developer.honor.com/auth/token
-HONOR_PUSH_SEND_ENDPOINT_TEMPLATE=https://push-api.cloud.honor.com/api/v1/{app_id}/sendMessage
-
-# 华为 Push Kit:AGC 控制台 → 项目设置 → 常规 → 应用,AppId + AppSecret
-HUAWEI_PUSH_APP_ID=
-HUAWEI_PUSH_APP_SECRET=
-HUAWEI_PUSH_TOKEN_ENDPOINT=https://oauth-login.cloud.huawei.com/oauth2/v3/token
-HUAWEI_PUSH_SEND_ENDPOINT_TEMPLATE=https://push-api.cloud.huawei.com/v1/{app_id}/messages:send
-
-VIVO_PUSH_APP_ID=
-VIVO_PUSH_APP_KEY=
-VIVO_PUSH_APP_SECRET=
-VIVO_PUSH_AUTH_ENDPOINT=https://api-push.vivo.com.cn/message/auth
-VIVO_PUSH_SEND_ENDPOINT=https://api-push.vivo.com.cn/message/send
-# vivo 未上架测试时可用 push_mode=1; 上架正式推送改为 0。
-VIVO_PUSH_MODE=1
-VIVO_PUSH_NOTIFY_TYPE=4
-VIVO_PUSH_CATEGORY=DEVICE_REMINDER
-
-XIAOMI_PUSH_APP_SECRET=
-XIAOMI_PUSH_SEND_ENDPOINT=https://api.xmpush.xiaomi.com/v3/message/regid
-XIAOMI_PUSH_CHANNEL_ID=
-XIAOMI_PUSH_TEMPLATE_ID=
-XIAOMI_PUSH_TEMPLATE_TITLE=
-XIAOMI_PUSH_TEMPLATE_DESCRIPTION=
-# 可选: JSON 字符串，支持 {title}/{alert} 占位符，例如 {"title":"{title}","content":"{alert}"}
-XIAOMI_PUSH_TEMPLATE_PARAM_JSON=
-
-OPPO_PUSH_APP_KEY=
-OPPO_PUSH_MASTER_SECRET=
-OPPO_PUSH_AUTH_ENDPOINT=https://api.push.oppomobile.com/server/v1/auth
-OPPO_PUSH_SEND_ENDPOINT=https://api.push.oppomobile.com/server/v1/message/notification/unicast
-# OPPO 新消息分类(2024-11-20 后创建的应用必须携带 category;channel_id 为后台「通道ID」;
-# notify_level 0=不传走默认,内容营销类仅支持 1/2)
-OPPO_PUSH_CHANNEL_ID=
-OPPO_PUSH_CATEGORY=
-OPPO_PUSH_NOTIFY_LEVEL=0
-
-# ===== 无障碍保护存活监控(推送 + pull 后置兜底)=====
+# ===== 无障碍保护存活监控(pull 后置检测;本期不接推送)=====
 HEARTBEAT_MONITOR_ENABLED=true
 HEARTBEAT_TIMEOUT_MINUTES=60
 HEARTBEAT_SCAN_INTERVAL_SEC=60
@@ -113,13 +65,6 @@ JD_UNION_APP_SECRET=
 JD_UNION_SITE_ID=
 JD_UNION_AUTH_KEY=
 
-# 美团 + 京东订单每天北京时间 05:00 自动对账；按更新时间回拉近 3 天，重叠防漏单并刷新状态。
-# 手动对账按钮不受该开关影响。通常保持开启；临时停自动任务时设为 false。
-CPS_AUTO_RECONCILE_ENABLED=true
-CPS_AUTO_RECONCILE_RUN_HOUR=5
-CPS_AUTO_RECONCILE_LOOKBACK_DAYS=3
-CPS_AUTO_RECONCILE_CHECK_INTERVAL_SEC=60
-
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。
@@ -193,18 +138,14 @@ PANGLE_REPORT_SECURITY_KEY=
 PANGLE_REPORT_SITE_ID_PROD=5830519
 PANGLE_REPORT_SITE_ID_TEST=5832303
 
-# ===== 可观测(OpenObserve 接口指标)=====
-# 采集每个接口 QPS + 耗时 + 错误率,批量直采到 OpenObserve(本地 Docker,见 deploy/openobserve/)。
-# 默认关;开启需 ENABLED=true 且填 USER/PASSWORD(与 docker-compose 里 root 账号一致)。
-# 未开/缺凭证 → 中间件透传、worker 不启动,整套 no-op,不影响业务。
-OBSERVE_ENABLED=false
-OBSERVE_ENDPOINT=http://localhost:5080
-OBSERVE_ORG=default
-OBSERVE_STREAM=app_requests
-OBSERVE_USER=admin@shaguabijia.local
-OBSERVE_PASSWORD=Complexpass#123
-# 进阶(一般不用改):攒批间隔秒 / 单批最大条数 / 有界队列上限(满则丢) / 上报超时秒
-OBSERVE_FLUSH_INTERVAL_SEC=5
-OBSERVE_BATCH_MAX=200
-OBSERVE_QUEUE_MAX=10000
-OBSERVE_TIMEOUT_SEC=5
+# ===== 客户端运行日志上报(POST /api/v1/applog/batch)=====
+# 客户端批量上报的 App 运行日志逐条落到独立滚动文件 logs/app-client.log,供阿里云 Logtail
+# 采进【独立 SLS logstore】(与服务日志 app-server.log 分开;滚动机制相同,trace_id 可跨层检索)。
+# 全部有默认值,不填即用默认(定义见 app/core/client_log.py 与 app/api/v1/applog.py)。
+# CLIENT_LOG_FILE=logs/app-client.log      # 落盘路径
+# CLIENT_LOG_MAX_BYTES=20971520            # 单文件 20MB 滚动
+# CLIENT_LOG_BACKUP_COUNT=10               # 保留 10 个 → ~200MB 缓冲(给 Logtail 断线留余量)
+# CLIENT_LOG_SERVICE_NAME=app-client       # 输出行 service 字段
+# APPLOG_MAX_BATCH=500                      # 单批最大条数(超 → 422;导入期常量,改需重启)
+# APPLOG_MAX_BODY_BYTES=2097152            # 请求体上限 2MB(超 → 413;运行期可调)
+# APPLOG_MAX_MSG_BYTES=8192                # 单条 msg 超此字节数截断

alembic/versions/user_manual_risk_fields.py

@@ -0,0 +1,76 @@
+"""add manual high-risk flag and note to user
+
+Revision ID: user_manual_risk_fields
+Revises: risk_monitor_generic
+"""
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "user_manual_risk_fields"
+down_revision: str | None = "risk_monitor_generic"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
+_INVITE_WITHDRAW_PAGE = "invite-withdraws"
+
+
+def _finance_role() -> sa.TableClause:
+    return sa.table(
+        "admin_role",
+        sa.column("name", sa.String),
+        sa.column("pages", _JSON),
+    )
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("user") as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "is_high_risk",
+                sa.Boolean(),
+                server_default=sa.false(),
+                nullable=False,
+            )
+        )
+        batch_op.add_column(sa.Column("high_risk_note", sa.Text(), nullable=True))
+        batch_op.create_index("ix_user_is_high_risk", ["is_high_risk"], unique=False)
+    role = _finance_role()
+    conn = op.get_bind()
+    row = conn.execute(
+        sa.select(role.c.pages).where(role.c.name == "finance")
+    ).scalar_one_or_none()
+    if row is not None and _INVITE_WITHDRAW_PAGE not in (row or []):
+        conn.execute(
+            role.update()
+            .where(role.c.name == "finance")
+            .values(pages=[*(row or []), _INVITE_WITHDRAW_PAGE])
+        )
+
+
+def downgrade() -> None:
+    role = _finance_role()
+    conn = op.get_bind()
+    row = conn.execute(
+        sa.select(role.c.pages).where(role.c.name == "finance")
+    ).scalar_one_or_none()
+    if row is not None:
+        conn.execute(
+            role.update()
+            .where(role.c.name == "finance")
+            .values(
+                pages=[
+                    page for page in (row or []) if page != _INVITE_WITHDRAW_PAGE
+                ]
+            )
+        )
+    with op.batch_alter_table("user") as batch_op:
+        batch_op.drop_index("ix_user_is_high_risk")
+        batch_op.drop_column("high_risk_note")
+        batch_op.drop_column("is_high_risk")

app/admin/permissions.py

@@ -23,7 +23,8 @@ PERMISSION_CATALOG: list[dict] = [
         {"key": "cps", "label": "CPS收益"},
     ]},
     {"group": "奖励审核", "pages": [
-        {"key": "withdraws", "label": "提现审核"},
+        {"key": "invite-withdraws", "label": "邀请提现审核"},
+        {"key": "withdraws", "label": "其他提现审核"},
         {"key": "price-reports", "label": "低价审核"},
         {"key": "feedbacks", "label": "用户反馈"},
     ]},
@@ -59,7 +60,7 @@ BUILTIN_ROLES: list[dict] = [
         "cps", "risk-monitor", "device-liveness", "price-reports", "feedbacks", "huawei-review",
     ]},
     {"name": "finance", "label": "财务", "pages": [
-        "dashboard", "ad-revenue-report", "cps", "withdraws",
+        "dashboard", "ad-revenue-report", "cps", "invite-withdraws", "withdraws",
     ]},
     {"name": "tech", "label": "技术", "pages": [
         "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",

app/admin/repositories/ad_revenue.py

@@ -431,6 +431,38 @@ def ad_revenue_report(
         for k, v in type_map.items()
     }
 
+    # 经营看板的两类 eCPM 必须按真实展示的 SDK eCPM 加权，不能用发奖状态修正后的
+    # 收益反推。Draw 包含新 draw 与历史 feed；看视频包含福利与提现视频。
+    category_map: dict[str, dict] = {}
+    for event in events:
+        category = (
+            "draw" if event["ad_type"] in {"draw", "feed"}
+            else "video" if event["ad_type"] in {"reward_video", "withdrawal_video"}
+            else None
+        )
+        if category is None:
+            continue
+        stat = category_map.setdefault(category, {
+            "impressions": 0,
+            "revenue_yuan": 0.0,
+            "ecpm_fen_sum": 0.0,
+        })
+        impressions = int(event["impressions"])
+        stat["impressions"] += impressions
+        stat["revenue_yuan"] += event["revenue_yuan"]
+        stat["ecpm_fen_sum"] += rewards.parse_ecpm_fen(event["ecpm"]) * impressions
+    category_stats = {
+        key: {
+            "impressions": value["impressions"],
+            "revenue_yuan": round(value["revenue_yuan"], 6),
+            "ecpm_yuan": round(
+                value["ecpm_fen_sum"] / value["impressions"] / 100.0,
+                6,
+            ) if value["impressions"] else 0.0,
+        }
+        for key, value in category_map.items()
+    }
+
     # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
     # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
     # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
@@ -485,6 +517,7 @@ def ad_revenue_report(
         "daily": daily,
         "hourly": hourly,
         "type_stats": type_stats,
+        "category_stats": category_stats,
         "scene_stats": scene_stats,
         "dau": dau,
         "items": main_rows[offset:offset + limit],

app/admin/repositories/mutations.py

@@ -44,6 +44,25 @@ def set_user_debug_trace(
     return user
 
 
+def set_user_risk(
+    db: Session,
+    user: User,
+    *,
+    is_high_risk: bool,
+    note: str | None,
+    commit: bool = True,
+) -> User:
+    """设置人工风险结论和备注，支持与审计日志共用同一事务。"""
+    user.is_high_risk = is_high_risk
+    user.high_risk_note = note.strip() if is_high_risk and note else None
+    if commit:
+        db.commit()
+        db.refresh(user)
+    else:
+        db.flush()
+    return user
+
+
 def update_feedback_status(
     db: Session, feedback: Feedback, *, status: str, commit: bool = True
 ) -> Feedback:

app/admin/repositories/queries.py

@@ -22,14 +22,15 @@ from app.models.comparison import ComparisonRecord
 from app.models.coupon_state import CouponPromptEngagement
 from app.models.device import DeviceLiveness
 from app.models.feedback import Feedback
+from app.models.invite import InviteRelation
 from app.models.onboarding import OnboardingCompletion
 from app.models.price_report import PriceReport
+from app.models.savings import SavingsRecord
 from app.models.user import User
 from app.models.wallet import (
     CashTransaction,
     CoinAccount,
     CoinTransaction,
-    InviteCashTransaction,
     WithdrawOrder,
 )
 from app.repositories import activity, ad_ecpm
@@ -62,6 +63,40 @@ def _device_marketing_name(model: str | None) -> str | None:
     return _DEVICE_MARKETING_NAMES.get(model.strip().upper())
 
 
+def _attach_comparison_order_status(db: Session, items: list[ComparisonRecord]) -> None:
+    """按 C 端既有口径给比价记录批量补充是否真实下单。"""
+    user_ids = {item.user_id for item in items if item.user_id is not None}
+    shop_names = {item.store_name for item in items if item.store_name}
+    ordered_pairs: set[tuple[int, str]] = set()
+    if user_ids and shop_names:
+        rows = db.execute(
+            select(SavingsRecord.user_id, SavingsRecord.shop_name)
+            .where(
+                SavingsRecord.user_id.in_(user_ids),
+                SavingsRecord.source == "compare",
+                SavingsRecord.shop_name.in_(shop_names),
+            )
+            .distinct()
+        ).all()
+        ordered_pairs = {
+            (row.user_id, row.shop_name)
+            for row in rows
+            if row.shop_name is not None
+        }
+    for item in items:
+        item.ordered = bool(
+            item.user_id is not None
+            and item.store_name
+            and (item.user_id, item.store_name) in ordered_pairs
+        )
+
+
+def _attach_comparison_device_details(items: list[ComparisonRecord]) -> None:
+    """给比价记录补充可读机型名，同时保留原始设备编码。"""
+    for item in items:
+        item.device_model_name = _device_marketing_name(item.device_model)
+
+
 def _attach_feedback_device_details(db: Session, feedbacks: list[Feedback]) -> None:
     """按同一用户、同一设备编码及提交时间补齐厂商和 ROM 大版本。"""
     candidates = [
@@ -343,6 +378,8 @@ def list_comparison_records(
         limit=limit, cursor=cursor,
     )
     _attach_user_info(db, items)
+    _attach_comparison_order_status(db, items)
+    _attach_comparison_device_details(items)
     # 「本次比价看广告的预估收益」:按本页 trace_id 一次性聚合(同 _attach_user_info 逐页范式)。
     # ad_revenue_yuan 非 ORM 列,仅瞬态挂实例上供 AdminComparisonListItem(from_attributes)读出。
     rev = ad_ecpm.revenue_yuan_by_trace(db, [it.trace_id for it in items])
@@ -489,6 +526,8 @@ def get_comparison_record(db: Session, record_id: int) -> ComparisonRecord | Non
     rec = db.get(ComparisonRecord, record_id)
     if rec is not None:
         _attach_user_info(db, [rec])
+        _attach_comparison_order_status(db, [rec])
+        _attach_comparison_device_details([rec])
     return rec
 
 
@@ -776,16 +815,7 @@ def list_all_withdraw_orders(
     elif quick_filter == "today":
         stmt = stmt.where(WithdrawOrder.created_at >= today_start)
     elif quick_filter == "high_risk":
-        stmt = stmt.where(
-            or_(
-                WithdrawOrder.user_name.is_(None),
-                WithdrawOrder.user_name == "",
-                User.status != "active",
-                User.created_at >= now - timedelta(hours=24),
-                WithdrawOrder.status.in_(("failed", "rejected")),
-                WithdrawOrder.fail_reason.is_not(None),
-            )
-        )
+        stmt = stmt.where(User.is_high_risk.is_(True))
 
     sort_cols = {
         "id": WithdrawOrder.id,
@@ -811,7 +841,9 @@ def _as_utc(value: datetime) -> datetime:
     return value.astimezone(timezone.utc)
 
 
-def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict]:
+def withdraw_list_enrichment(
+    db: Session, user_ids: list[int], *, source: str | None = None
+) -> dict[int, dict]:
     """批量富化提现单列表:按本页 user_id 取 手机号/昵称 + 各自累计成功提现金额(分)。
 
     两条聚合查询搞定(避免逐行 N+1)。累计口径与 get_user_overview 的 withdraw_success_cents
@@ -821,25 +853,32 @@ def withdraw_list_enrichment(db: Session, user_ids: list[int]) -> dict[int, dict
     uniq = list(set(user_ids))
     if not uniq:
         return {}
-    success_rows = db.execute(
-        select(
-            WithdrawOrder.user_id,
-            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
-        )
-        .where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
-        .group_by(WithdrawOrder.user_id)
-    ).all()
+    success_stmt = select(
+        WithdrawOrder.user_id,
+        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+    ).where(WithdrawOrder.user_id.in_(uniq), WithdrawOrder.status == "success")
+    if source:
+        success_stmt = success_stmt.where(WithdrawOrder.source == source)
+    success_rows = db.execute(success_stmt.group_by(WithdrawOrder.user_id)).all()
     success_map = {uid: int(total) for uid, total in success_rows}
     users = db.execute(
-        select(User.id, User.phone, User.nickname).where(User.id.in_(uniq))
+        select(
+            User.id,
+            User.phone,
+            User.nickname,
+            User.is_high_risk,
+            User.high_risk_note,
+        ).where(User.id.in_(uniq))
     ).all()
     return {
         uid: {
             "phone": phone,
             "nickname": nickname,
+            "is_high_risk": is_high_risk,
+            "high_risk_note": high_risk_note,
             "cumulative_success_cents": success_map.get(uid, 0),
         }
-        for uid, phone, nickname in users
+        for uid, phone, nickname, is_high_risk, high_risk_note in users
     }
 
 
@@ -946,15 +985,16 @@ def get_withdraw_by_out_bill_no(db: Session, out_bill_no: str) -> WithdrawOrder
     ).scalar_one_or_none()
 
 
-def withdraw_summary(db: Session) -> dict:
-    """提现审核台顶部统计。金额单位:分。"""
-    rows = db.execute(
-        select(
-            WithdrawOrder.status,
-            func.count(WithdrawOrder.id),
-            func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
-        ).group_by(WithdrawOrder.status)
-    ).all()
+def withdraw_summary(db: Session, *, source: str | None = None) -> dict:
+    """提现审核台顶部统计。金额单位:分；口径与列表的 source 筛选一致。"""
+    summary_stmt = select(
+        WithdrawOrder.status,
+        func.count(WithdrawOrder.id),
+        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0),
+    )
+    if source:
+        summary_stmt = summary_stmt.where(WithdrawOrder.source == source)
+    rows = db.execute(summary_stmt.group_by(WithdrawOrder.status)).all()
     by_status = {
         status: {"count": int(count), "amount_cents": int(amount_cents)}
         for status, count, amount_cents in rows
@@ -967,31 +1007,154 @@ def withdraw_summary(db: Session) -> dict:
     )
 
     def _today_count(status: str) -> int:
-        return db.execute(
-            select(func.count(WithdrawOrder.id)).where(
-                WithdrawOrder.status == status,
-                WithdrawOrder.updated_at >= today_start,
-            )
-        ).scalar_one()
-
-    today_success_amount = db.execute(
-        select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
-            WithdrawOrder.status == "success",
+        stmt = select(func.count(WithdrawOrder.id)).where(
+            WithdrawOrder.status == status,
             WithdrawOrder.updated_at >= today_start,
         )
-    ).scalar_one()
+        if source:
+            stmt = stmt.where(WithdrawOrder.source == source)
+        return db.execute(stmt).scalar_one()
+
+    today_success_amount_stmt = select(
+        func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)
+    ).where(
+        WithdrawOrder.status == "success",
+        WithdrawOrder.updated_at >= today_start,
+    )
+    if source:
+        today_success_amount_stmt = today_success_amount_stmt.where(
+            WithdrawOrder.source == source
+        )
+    today_success_amount = db.execute(today_success_amount_stmt).scalar_one()
+
+    def _status_count(status: str) -> int:
+        return by_status.get(status, {}).get("count", 0)
 
     return {
-        "reviewing_count": by_status.get("reviewing", {}).get("count", 0),
+        "reviewing_count": _status_count("reviewing"),
         "reviewing_amount_cents": by_status.get("reviewing", {}).get("amount_cents", 0),
-        "pending_count": by_status.get("pending", {}).get("count", 0),
-        "failed_count": by_status.get("failed", {}).get("count", 0),
+        "pending_count": _status_count("pending"),
+        "success_count": _status_count("success"),
+        "rejected_count": _status_count("rejected"),
+        "failed_count": _status_count("failed"),
+        "total_count": sum(item["count"] for item in by_status.values()),
         "today_success_count": _today_count("success"),
         "today_success_amount_cents": int(today_success_amount),
         "today_rejected_count": _today_count("rejected"),
     }
 
 
+def invite_overview(
+    db: Session,
+    inviter_user_id: int,
+    *,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+) -> dict:
+    """邀请提现详情：邀请关系统计及受邀用户首次比价/首单信息。
+
+    首次记录按自增 id 取最早一条，批量查询避免按受邀用户逐行查询。
+    邀请成功口径为完成首次比价并已发邀请奖励(compare_reward_granted)。
+    """
+    relation_stmt = (
+        select(InviteRelation, User)
+        .join(User, User.id == InviteRelation.invitee_user_id)
+        .where(InviteRelation.inviter_user_id == inviter_user_id)
+        .order_by(InviteRelation.created_at.asc(), InviteRelation.id.asc())
+    )
+    if date_from is not None:
+        relation_stmt = relation_stmt.where(User.created_at >= _as_utc_naive(date_from))
+    if date_to is not None:
+        relation_stmt = relation_stmt.where(User.created_at <= _as_utc_naive(date_to))
+    relation_rows = db.execute(relation_stmt).all()
+    invitee_ids = [relation.invitee_user_id for relation, _ in relation_rows]
+    if not invitee_ids:
+        return {"invite_total": 0, "invite_success_total": 0, "items": []}
+
+    first_compare_ids = list(
+        db.execute(
+            select(func.min(ComparisonRecord.id))
+            .where(
+                ComparisonRecord.user_id.in_(invitee_ids),
+                ComparisonRecord.status == "success",
+            )
+            .group_by(ComparisonRecord.user_id)
+        ).scalars()
+    )
+    comparisons = (
+        db.execute(
+            select(
+                ComparisonRecord.user_id,
+                ComparisonRecord.store_name,
+                ComparisonRecord.product_names,
+            ).where(ComparisonRecord.id.in_(first_compare_ids))
+        ).all()
+        if first_compare_ids
+        else []
+    )
+    comparison_map = {
+        user_id: (store_name, product_names)
+        for user_id, store_name, product_names in comparisons
+    }
+
+    first_order_ids = list(
+        db.execute(
+            select(func.min(SavingsRecord.id))
+            .where(
+                SavingsRecord.user_id.in_(invitee_ids),
+                SavingsRecord.source == "compare",
+            )
+            .group_by(SavingsRecord.user_id)
+        ).scalars()
+    )
+    orders = (
+        db.execute(
+            select(
+                SavingsRecord.user_id,
+                SavingsRecord.shop_name,
+                SavingsRecord.title,
+                SavingsRecord.dishes,
+                SavingsRecord.order_amount_cents,
+            ).where(SavingsRecord.id.in_(first_order_ids))
+        ).all()
+        if first_order_ids
+        else []
+    )
+    order_map = {
+        user_id: (
+            shop_name,
+            title or "、".join(dishes or []),
+            order_amount_cents,
+        )
+        for user_id, shop_name, title, dishes, order_amount_cents in orders
+    }
+
+    items = []
+    for relation, user in relation_rows:
+        compare_store, compare_products = comparison_map.get(user.id, (None, None))
+        order_store, order_products, order_amount = order_map.get(
+            user.id, (None, None, None)
+        )
+        items.append(
+            {
+                "user_id": user.id,
+                "phone": user.phone,
+                "registered_at": user.created_at,
+                "invite_success": bool(relation.compare_reward_granted),
+                "first_compare_store": compare_store,
+                "first_compare_products": compare_products,
+                "first_order_store": order_store,
+                "first_order_products": order_products,
+                "first_order_amount_cents": order_amount,
+            }
+        )
+    return {
+        "invite_total": len(items),
+        "invite_success_total": sum(1 for item in items if item["invite_success"]),
+        "items": items,
+    }
+
+
 def list_withdraw_audit_logs(
     db: Session, out_bill_no: str, *, limit: int = 20
 ) -> list[AdminAuditLog]:
@@ -1012,8 +1175,6 @@ def withdraw_risk_flags(
     cash_balance_cents: int,
 ) -> tuple[list[str], int]:
     flags: list[str] = []
-    if not order.user_name:
-        flags.append("缺少提现实名")
     if user and user.status != "active":
         flags.append(f"账号状态:{user.status}")
     if user and user.created_at:
@@ -1036,125 +1197,6 @@ def withdraw_risk_flags(
     return flags, score
 
 
-def _check_withdraw_ledger_side(
-    orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str
-) -> dict:
-    """对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。
-
-    orders 已按 source 过滤到本账;txns 是本账流水表里 withdraw_biz/refund_biz 两类流水。
-    规则:每单发起应有一条扣款流水(ref_id=out_bill_no);失败/拒绝单应有且仅一条退款流水;
-    非退款终态不应出现退款流水。四个计数全为 0 即本账自洽。
-    """
-    withdraw_refs = {txn.ref_id for txn in txns if txn.biz_type == withdraw_biz}
-    refund_counts: dict[str, int] = {}
-    for txn in txns:
-        if txn.biz_type == refund_biz and txn.ref_id:
-            refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
-
-    missing_withdraw = 0
-    missing_refund = 0
-    refund_on_non_terminal = 0
-    for order in orders:
-        if order.out_bill_no not in withdraw_refs:
-            missing_withdraw += 1
-        has_refund = refund_counts.get(order.out_bill_no, 0) > 0
-        if order.status in {"failed", "rejected"} and not has_refund:
-            missing_refund += 1
-        if has_refund and order.status not in {"failed", "rejected"}:
-            refund_on_non_terminal += 1
-
-    return {
-        "missing_withdraw": missing_withdraw,
-        "missing_refund": missing_refund,
-        "duplicate_refund": sum(1 for count in refund_counts.values() if count > 1),
-        "refund_on_non_terminal": refund_on_non_terminal,
-    }
-
-
-def withdraw_ledger_check(db: Session) -> dict:
-    """现金账本校验:两本物理隔离的账各自对账(产品红线:coin_cash / invite_cash 不串)。
-
-    普通现金:CoinAccount.cash_balance_cents ↔ cash_transaction(withdraw/withdraw_refund);
-    邀请奖励金:CoinAccount.invite_cash_balance_cents ↔ invite_cash_transaction
-              (invite_withdraw/invite_withdraw_refund)。
-    提现单按 source 分流到对应账核对——邀请提现的流水写在 invite_cash_transaction 表,
-    绝不能拿去和普通现金流水比(否则每笔邀请提现单都会被误报「缺扣款/缺退款流水」)。
-    分流口径与 create_withdraw 一致:仅 source==invite_cash 走邀请账,其余(含历史空值)归普通现金。
-    """
-    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
-    coin_orders = [o for o in orders if o.source != "invite_cash"]
-    invite_orders = [o for o in orders if o.source == "invite_cash"]
-
-    # —— 普通现金账(coin_cash) ——
-    cash_balance_total = int(
-        db.execute(select(func.coalesce(func.sum(CoinAccount.cash_balance_cents), 0))).scalar_one()
-    )
-    cash_txn_total = int(
-        db.execute(select(func.coalesce(func.sum(CashTransaction.amount_cents), 0))).scalar_one()
-    )
-    cash_txns = list(
-        db.execute(
-            select(CashTransaction).where(
-                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
-            )
-        ).scalars().all()
-    )
-    coin = _check_withdraw_ledger_side(
-        coin_orders, cash_txns, withdraw_biz="withdraw", refund_biz="withdraw_refund"
-    )
-    cash_diff = cash_balance_total - cash_txn_total
-
-    # —— 邀请奖励金账(invite_cash,独立账户 + 独立流水表) ——
-    invite_balance_total = int(
-        db.execute(
-            select(func.coalesce(func.sum(CoinAccount.invite_cash_balance_cents), 0))
-        ).scalar_one()
-    )
-    invite_txn_total = int(
-        db.execute(
-            select(func.coalesce(func.sum(InviteCashTransaction.amount_cents), 0))
-        ).scalar_one()
-    )
-    invite_txns = list(
-        db.execute(
-            select(InviteCashTransaction).where(
-                InviteCashTransaction.biz_type.in_(("invite_withdraw", "invite_withdraw_refund"))
-            )
-        ).scalars().all()
-    )
-    invite = _check_withdraw_ledger_side(
-        invite_orders, invite_txns,
-        withdraw_biz="invite_withdraw", refund_biz="invite_withdraw_refund",
-    )
-    invite_diff = invite_balance_total - invite_txn_total
-
-    ok = (
-        cash_diff == 0
-        and invite_diff == 0
-        and all(v == 0 for v in coin.values())
-        and all(v == 0 for v in invite.values())
-    )
-    return {
-        "ok": ok,
-        # 普通现金账(coin_cash:金币兑换的现金)
-        "cash_balance_total_cents": cash_balance_total,
-        "cash_transaction_total_cents": cash_txn_total,
-        "balance_diff_cents": cash_diff,
-        "missing_withdraw_txn_count": coin["missing_withdraw"],
-        "missing_refund_txn_count": coin["missing_refund"],
-        "duplicate_refund_txn_count": coin["duplicate_refund"],
-        "refund_txn_on_non_terminal_count": coin["refund_on_non_terminal"],
-        # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)
-        "invite_cash_balance_total_cents": invite_balance_total,
-        "invite_cash_transaction_total_cents": invite_txn_total,
-        "invite_balance_diff_cents": invite_diff,
-        "invite_missing_withdraw_txn_count": invite["missing_withdraw"],
-        "invite_missing_refund_txn_count": invite["missing_refund"],
-        "invite_duplicate_refund_txn_count": invite["duplicate_refund"],
-        "invite_refund_txn_on_non_terminal_count": invite["refund_on_non_terminal"],
-    }
-
-
 def get_user_overview(db: Session, user_id: int) -> dict | None:
     """用户 360 概览:基础资料 + 钱包余额 + 各项 count。历史明细走各自分页接口(带 user_id 过滤)。"""
     user = db.get(User, user_id)
@@ -1214,6 +1256,7 @@ def user_reward_stats(
     *,
     date_from: datetime | None = None,
     date_to: datetime | None = None,
+    withdraw_source: str | None = None,
 ) -> dict:
     """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
 
@@ -1221,22 +1264,35 @@ def user_reward_stats(
     平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
     传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
     """
+    withdraw_source_conds = (
+        [WithdrawOrder.source == withdraw_source] if withdraw_source else []
+    )
     wd_success = db.execute(
         select(func.coalesce(func.sum(WithdrawOrder.amount_cents), 0)).where(
             WithdrawOrder.user_id == user_id,
             WithdrawOrder.status == "success",
+            *withdraw_source_conds,
             *_window_conds(WithdrawOrder.created_at, date_from, date_to),
         )
     ).scalar_one()
     wd_total = db.execute(
         select(func.count(WithdrawOrder.id)).where(
             WithdrawOrder.user_id == user_id,
+            *withdraw_source_conds,
             *_window_conds(WithdrawOrder.created_at, date_from, date_to),
         )
     ).scalar_one()
 
     acc = db.get(CoinAccount, user_id)  # 现金余额:当前快照,不随窗口
-    cash_balance = acc.cash_balance_cents if acc else 0
+    cash_balance = (
+        (
+            acc.invite_cash_balance_cents
+            if withdraw_source == "invite_cash"
+            else acc.cash_balance_cents
+        )
+        if acc
+        else 0
+    )
 
     # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
     # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。

app/admin/routers/users.py

@@ -16,6 +16,7 @@ from app.admin.schemas.user import (
     GrantCashRequest,
     GrantCoinsRequest,
     SetDebugTraceRequest,
+    SetUserRiskRequest,
     SetUserStatusRequest,
     UserCoinRecord,
     UserRewardStats,
@@ -84,12 +85,21 @@ def get_user_reward_stats(
     db: AdminDb,
     date_from: Annotated[datetime | None, Query()] = None,
     date_to: Annotated[datetime | None, Query()] = None,
+    withdraw_source: Annotated[
+        str | None, Query(pattern="^(coin_cash|invite_cash)$")
+    ] = None,
 ) -> UserRewardStats:
     """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
     if not user_repo.user_exists(db, user_id):
         raise HTTPException(status_code=404, detail="用户不存在")
     return UserRewardStats(
-        **queries.user_reward_stats(db, user_id, date_from=date_from, date_to=date_to)
+        **queries.user_reward_stats(
+            db,
+            user_id,
+            date_from=date_from,
+            date_to=date_to,
+            withdraw_source=withdraw_source,
+        )
     )
 
 
@@ -138,6 +148,50 @@ def set_user_status(
     return OkResponse()
 
 
+@router.post("/{user_id}/risk", response_model=OkResponse, summary="设置人工高风险标记与备注")
+def set_user_risk(
+    user_id: int,
+    body: SetUserRiskRequest,
+    request: Request,
+    admin: Annotated[
+        AdminUser, Depends(require_role("operator", "finance"))
+    ],
+    db: AdminDb,
+) -> OkResponse:
+    user = user_repo.get_user_by_id(db, user_id)
+    if user is None:
+        raise HTTPException(status_code=404, detail="用户不存在")
+    before = {
+        "is_high_risk": user.is_high_risk,
+        "high_risk_note": user.high_risk_note,
+    }
+    mutations.set_user_risk(
+        db,
+        user,
+        is_high_risk=body.is_high_risk,
+        note=body.note,
+        commit=False,
+    )
+    write_audit(
+        db,
+        admin,
+        action="user.risk.set",
+        target_type="user",
+        target_id=user_id,
+        detail={
+            "before": before,
+            "after": {
+                "is_high_risk": user.is_high_risk,
+                "high_risk_note": user.high_risk_note,
+            },
+        },
+        ip=get_client_ip(request),
+        commit=False,
+    )
+    db.commit()
+    return OkResponse()
+
+
 @router.post("/{user_id}/debug-trace", response_model=OkResponse, summary="开关调试链接权限")
 def set_user_debug_trace(
     user_id: int,

app/admin/routers/withdraw.py

@@ -1,4 +1,4 @@
-"""admin 提现:列表(读)+ 单笔重试查单 + 批量对账(写,带审计)。
+"""admin 提现:列表(读)+ 单笔/批量查单与审核操作(写,带审计)。
 
 提现的钱逻辑(查微信/退款/撤单/幂等)全部复用 app.repositories.wallet,admin 只触发 + 记审计。
 这些 wallet 函数内部各自 commit(涉及微信调用),审计在其后单独 commit:操作本身幂等,
@@ -14,17 +14,16 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, get_client_ip, get_current_admin, require_role
 from app.admin.repositories import queries
-from app.admin.schemas.common import CursorPage
 from app.admin.schemas.admin import AdminAuditLogOut
+from app.admin.schemas.common import CursorPage
 from app.admin.schemas.wallet import (
     CashTxnOut,
-    ReconcileResult,
+    InviteOverviewOut,
+    WithdrawBulkItemResult,
     WithdrawBulkRejectRequest,
     WithdrawBulkRequest,
     WithdrawBulkResult,
-    WithdrawBulkItemResult,
     WithdrawDetailOut,
-    WithdrawLedgerCheckOut,
     WithdrawListItemOut,
     WithdrawOrderOut,
     WithdrawRejectRequest,
@@ -83,8 +82,16 @@ def list_withdraws(
         cursor=cursor,
     )
     # 联表带出手机号/昵称 + 累计成功提现(本页 user_id 批量富化,2 条聚合查询,无 N+1)。
-    enrichment = queries.withdraw_list_enrichment(db, [o.user_id for o in items])
-    _empty = {"phone": None, "nickname": None, "cumulative_success_cents": 0}
+    enrichment = queries.withdraw_list_enrichment(
+        db, [o.user_id for o in items], source=source
+    )
+    _empty = {
+        "phone": None,
+        "nickname": None,
+        "is_high_risk": False,
+        "high_risk_note": None,
+        "cumulative_success_cents": 0,
+    }
     out_items = [
         WithdrawListItemOut.model_validate(o).model_copy(
             update=enrichment.get(o.user_id, _empty)
@@ -95,8 +102,13 @@ def list_withdraws(
 
 
 @router.get("/summary", response_model=WithdrawSummaryOut, summary="提现审核台统计")
-def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
-    return WithdrawSummaryOut(**queries.withdraw_summary(db))
+def withdraws_summary(
+    db: AdminDb,
+    source: Annotated[
+        str | None, Query(pattern="^(coin_cash|invite_cash)$")
+    ] = None,
+) -> WithdrawSummaryOut:
+    return WithdrawSummaryOut(**queries.withdraw_summary(db, source=source))
 
 
 @router.get(
@@ -153,13 +165,13 @@ def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
     )
 
 
-@router.get("/ledger-check", response_model=WithdrawLedgerCheckOut, summary="提现资金账本校验")
-def withdraw_ledger_check(db: AdminDb) -> WithdrawLedgerCheckOut:
-    return WithdrawLedgerCheckOut(**queries.withdraw_ledger_check(db))
-
-
 @router.get("/{out_bill_no}", response_model=WithdrawDetailOut, summary="提现单详情")
-def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
+def withdraw_detail(
+    out_bill_no: str,
+    db: AdminDb,
+    date_from: Annotated[datetime | None, Query()] = None,
+    date_to: Annotated[datetime | None, Query()] = None,
+) -> WithdrawDetailOut:
     order = queries.get_withdraw_by_out_bill_no(db, out_bill_no)
     if order is None:
         raise HTTPException(status_code=404, detail="提现单不存在")
@@ -172,6 +184,8 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
             id=user.id,
             phone=user.phone,
             nickname=user.nickname,
+            is_high_risk=user.is_high_risk,
+            high_risk_note=user.high_risk_note,
             status=user.status,
             wechat_nickname=user.wechat_nickname,
             wechat_avatar_url=user.wechat_avatar_url,
@@ -195,37 +209,36 @@ def withdraw_detail(out_bill_no: str, db: AdminDb) -> WithdrawDetailOut:
         recent_withdraws,
         overview["cash_balance_cents"] if overview else 0,
     )
+    detail_enrichment = queries.withdraw_list_enrichment(
+        db, [order.user_id], source=order.source
+    ).get(order.user_id, {})
 
     return WithdrawDetailOut(
         order=WithdrawOrderOut.model_validate(order),
         user=user_snapshot,
+        cumulative_success_cents=int(
+            detail_enrichment.get("cumulative_success_cents", 0)
+        ),
         risk_flags=risk_flags,
         risk_score=risk_score,
         recent_withdraws=[WithdrawOrderOut.model_validate(o) for o in recent_withdraws],
         recent_cash_transactions=[CashTxnOut.model_validate(t) for t in recent_cash_transactions],
         audit_logs=[AdminAuditLogOut.model_validate(log) for log in audit_logs],
+        invite_overview=(
+            InviteOverviewOut(
+                **queries.invite_overview(
+                    db,
+                    order.user_id,
+                    date_from=date_from,
+                    date_to=date_to,
+                )
+            )
+            if order.source == "invite_cash"
+            else None
+        ),
     )
 
 
-# 注意:/reconcile 必须在 /{out_bill_no}/refresh 之前声明(静态路径优先于路径参数)
-@router.post("/reconcile", response_model=ReconcileResult, summary="批量对账(扫超时 pending 单)")
-def reconcile(
-    request: Request,
-    admin: Annotated[AdminUser, Depends(require_role("finance"))],
-    db: AdminDb,
-    older_than_minutes: Annotated[int, Query(ge=0)] = 15,
-) -> ReconcileResult:
-    try:
-        result = wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
-    except wxpay.WxPayNotConfiguredError as e:
-        raise HTTPException(status_code=503, detail="微信支付未配置") from e
-    write_audit(
-        db, admin, action="withdraw.reconcile", target_type="withdraw", target_id=None,
-        detail=result, ip=get_client_ip(request), commit=True,
-    )
-    return ReconcileResult(**result)
-
-
 def _bulk_result(items: list[WithdrawBulkItemResult]) -> WithdrawBulkResult:
     success = sum(1 for item in items if item.ok)
     return WithdrawBulkResult(

app/admin/schemas/ad_revenue.py

@@ -75,6 +75,7 @@ class AdRevenueTypeStat(BaseModel):
 
     impressions: int = Field(..., description="该类型展示条数合计")
     revenue_yuan: float = Field(..., description="该类型预估收益合计(元)")
+    ecpm_yuan: float | None = Field(None, description="按真实展示次数加权的 SDK eCPM(元/千次)")
 
 
 class AdRevenueRow(BaseModel):
@@ -144,6 +145,10 @@ class AdRevenueReportOut(BaseModel):
         default_factory=dict,
         description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
     )
+    category_stats: dict[str, AdRevenueTypeStat] = Field(
+        default_factory=dict,
+        description="按经营分类小计：draw=draw+历史 feed，video=reward_video+withdrawal_video",
+    )
     scene_stats: dict[str, AdRevenueTypeStat] = Field(
         default_factory=dict,
         description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"

app/admin/schemas/comparison.py

@@ -29,6 +29,7 @@ class AdminComparisonListItem(BaseModel):
     source_price_cents: int | None = None
     best_price_cents: int | None = None
     saved_amount_cents: int | None = None
+    ordered: bool = False
     # debug 概览
     total_ms: int | None = None
     step_count: int | None = None
@@ -39,8 +40,10 @@ class AdminComparisonListItem(BaseModel):
     # 本次比价 LLM 总成本(元,按当时价冻结);旧记录/未回填为 None → 前端「成本」列回退估算。见 services/llm_cost.py。
     llm_cost_yuan: float | None = None
     device_model: str | None = None
+    device_model_name: str | None = None
     rom_vendor: str | None = None
     rom_name: str | None = None
+    rom_version: int | None = None
     android_version: str | None = None
     app_version: str | None = None
     ad_revenue_yuan: float = 0.0   # 本次比价看的信息流广告预估收益(元),queries 瞬态挂 ORM 实例上
@@ -84,7 +87,6 @@ class AdminComparisonDetail(AdminComparisonListItem):
     skipped_dish_names: list = []
     # 全量环境
     device_manufacturer: str | None = None
-    rom_version: int | None = None
     android_sdk: int | None = None
     app_version_code: int | None = None
     source_app_version: str | None = None

app/admin/schemas/user.py

@@ -16,6 +16,8 @@ class AdminUserListItem(BaseModel):
     register_channel: str
     status: str
     debug_trace_enabled: bool = False
+    is_high_risk: bool = False
+    high_risk_note: str | None = None
     wechat_openid: str | None = None
     wechat_nickname: str | None = None
     created_at: datetime
@@ -118,3 +120,19 @@ class SetUserStatusRequest(BaseModel):
 
 class SetDebugTraceRequest(BaseModel):
     enabled: bool = Field(..., description="是否给该用户开「复制调试链接」权限")
+
+
+class SetUserRiskRequest(BaseModel):
+    is_high_risk: bool = Field(..., description="是否标记为高风险用户")
+    note: str | None = Field(
+        None,
+        max_length=500,
+        description="高风险备注；标记高风险时必填，解除后清空",
+    )
+
+    @field_validator("note")
+    @classmethod
+    def validate_note(cls, value: str | None, info):
+        if info.data.get("is_high_risk") and not (value or "").strip():
+            raise ValueError("标记高风险时必须填写原因")
+        return value.strip() if value and value.strip() else None

app/admin/schemas/wallet.py

@@ -60,6 +60,8 @@ class WithdrawListItemOut(WithdrawOrderOut):
 
     phone: str | None = None
     nickname: str | None = None
+    is_high_risk: bool = False
+    high_risk_note: str | None = None
     cumulative_success_cents: int = 0  # 累计成功提现 = SUM(amount_cents) WHERE status='success'
 
 
@@ -67,7 +69,10 @@ class WithdrawSummaryOut(BaseModel):
     reviewing_count: int
     reviewing_amount_cents: int
     pending_count: int
+    success_count: int
+    rejected_count: int
     failed_count: int
+    total_count: int
     today_success_count: int
     today_success_amount_cents: int
     today_rejected_count: int
@@ -77,6 +82,8 @@ class WithdrawUserSnapshot(BaseModel):
     id: int
     phone: str
     nickname: str | None = None
+    is_high_risk: bool = False
+    high_risk_note: str | None = None
     status: str
     wechat_nickname: str | None = None
     wechat_avatar_url: str | None = None
@@ -87,19 +94,34 @@ class WithdrawUserSnapshot(BaseModel):
     withdraw_success_cents: int
 
 
+class InviteeDetailOut(BaseModel):
+    user_id: int
+    phone: str
+    registered_at: datetime
+    invite_success: bool
+    first_compare_store: str | None = None
+    first_compare_products: str | None = None
+    first_order_store: str | None = None
+    first_order_products: str | None = None
+    first_order_amount_cents: int | None = None
+
+
+class InviteOverviewOut(BaseModel):
+    invite_total: int
+    invite_success_total: int
+    items: list[InviteeDetailOut]
+
+
 class WithdrawDetailOut(BaseModel):
     order: WithdrawOrderOut
     user: WithdrawUserSnapshot | None = None
+    cumulative_success_cents: int = 0
     risk_flags: list[str]
     risk_score: int
     recent_withdraws: list[WithdrawOrderOut]
     recent_cash_transactions: list[CashTxnOut]
     audit_logs: list[AdminAuditLogOut]
-
-
-class ReconcileResult(BaseModel):
-    checked: int
-    resolved: int
+    invite_overview: InviteOverviewOut | None = None
 
 
 class WithdrawBulkRequest(BaseModel):
@@ -128,26 +150,6 @@ class WithdrawBulkResult(BaseModel):
     items: list[WithdrawBulkItemResult]
 
 
-class WithdrawLedgerCheckOut(BaseModel):
-    ok: bool
-    # 普通现金账(coin_cash:金币兑换的现金)
-    cash_balance_total_cents: int
-    cash_transaction_total_cents: int
-    balance_diff_cents: int
-    missing_withdraw_txn_count: int
-    missing_refund_txn_count: int
-    duplicate_refund_txn_count: int
-    refund_txn_on_non_terminal_count: int
-    # 邀请奖励金账(invite_cash:与普通现金物理隔离,各自对账)。默认 0 向后兼容。
-    invite_cash_balance_total_cents: int = 0
-    invite_cash_transaction_total_cents: int = 0
-    invite_balance_diff_cents: int = 0
-    invite_missing_withdraw_txn_count: int = 0
-    invite_missing_refund_txn_count: int = 0
-    invite_duplicate_refund_txn_count: int = 0
-    invite_refund_txn_on_non_terminal_count: int = 0
-
-
 class WxpayHealthCheckOut(BaseModel):
     ok: bool
     wxpay_configured: bool

app/api/deps.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 import logging
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, status
+from fastapi import Depends, HTTPException, Request, status
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
 
@@ -77,6 +77,18 @@ def get_current_user_optional(
     return user
 
 
+def get_client_ip(request: Request) -> str:
+    """客户端真实 IP:生产经 nginx 反代优先 X-Forwarded-For 首段,否则直连 IP。
+
+    路由层取 IP 的规范实现(analytics / applog 等共用);core 层(ratelimit)因不能
+    反向依赖 app.api,自留一份私有副本。
+    """
+    xff = request.headers.get("x-forwarded-for")
+    if xff:
+        return xff.split(",")[0].strip()
+    return request.client.host if request.client else ""
+
+
 CurrentUser = Annotated[User, Depends(get_current_user)]
 OptionalUser = Annotated[User | None, Depends(get_current_user_optional)]
 DbSession = Annotated[Session, Depends(get_db)]

app/api/v1/analytics.py

@@ -10,7 +10,7 @@ import logging
 
 from fastapi import APIRouter, HTTPException, Request
 
-from app.api.deps import DbSession
+from app.api.deps import DbSession, get_client_ip
 from app.repositories import analytics as analytics_repo
 from app.repositories import analytics_selfstat as selfstat_repo
 from app.schemas.analytics import AnalyticsBatchIn, AnalyticsIngestOut
@@ -20,19 +20,11 @@ router = APIRouter(prefix="/api/v1/analytics", tags=["analytics"])
 logger = logging.getLogger("shagua.analytics")
 
 
-def _client_ip(request: Request) -> str:
-    """取客户端 IP:生产经 nginx 反代优先 X-Forwarded-For 第一段,否则直连 IP(同 admin get_client_ip)。"""
-    xff = request.headers.get("x-forwarded-for")
-    if xff:
-        return xff.split(",")[0].strip()
-    return request.client.host if request.client else ""
-
-
 @router.post("/events", response_model=AnalyticsIngestOut, summary="批量上报埋点事件")
 def ingest_events(
     batch: AnalyticsBatchIn, request: Request, db: DbSession
 ) -> AnalyticsIngestOut:
-    n = analytics_repo.record_batch(db, batch, client_ip=_client_ip(request))
+    n = analytics_repo.record_batch(db, batch, client_ip=get_client_ip(request))
     return AnalyticsIngestOut(received=n)
 
 

app/api/v1/applog.py

@@ -0,0 +1,50 @@
+"""客户端运行日志批量上报接口。
+
+POST /api/v1/applog/batch — 批量接收客户端运行日志,逐条写专用滚动文件 logs/app-client.log
+(供 Logtail 采进独立 SLS logstore)。鉴权同 analytics(不强制登录,user_id 可选在 body)。
+fire-and-forget:写失败也不 500(避免客户端重试风暴);超批 422、超体积 413、msg 超限截断。
+"""
+from __future__ import annotations
+
+import os
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+
+from app.api.deps import get_client_ip
+from app.core.client_log import write_records
+from app.core.ratelimit import rate_limit
+from app.schemas.applog import AppLogBatchIn, AppLogIngestOut
+
+router = APIRouter(prefix="/api/v1/applog", tags=["applog"])
+
+
+def _enforce_body_limit(request: Request) -> None:
+    """依赖:body 声明过大直接 413(在 body 校验前拦截)。缺 Content-Length 由 nginx 兜底。"""
+    max_bytes = int(os.getenv("APPLOG_MAX_BODY_BYTES", str(1024 * 1024 * 2)))
+    cl = request.headers.get("content-length")
+    if cl is not None and cl.isdigit() and int(cl) > max_bytes:
+        raise HTTPException(status_code=413, detail="日志批量过大")
+
+
+@router.post(
+    "/batch",
+    response_model=AppLogIngestOut,
+    summary="批量上报客户端运行日志",
+    dependencies=[
+        Depends(rate_limit(120, 60, "applog-batch")),
+        Depends(_enforce_body_limit),
+    ],
+)
+def ingest_logs(batch: AppLogBatchIn, request: Request) -> AppLogIngestOut:
+    received, dropped = write_records(
+        batch.logs,
+        meta={
+            "device_id": batch.device_id,
+            "user_id": batch.user_id,
+            "app_ver": batch.app_ver,
+            "platform": batch.platform,
+            "sent_at": batch.sent_at,
+        },
+        client_ip=get_client_ip(request),
+    )
+    return AppLogIngestOut(received=received, dropped=dropped)

app/api/v1/compare.py

@@ -25,7 +25,7 @@ import uuid
 from typing import Any
 
 import httpx
-from fastapi import APIRouter, HTTPException, Request, status
+from fastapi import APIRouter, BackgroundTasks, HTTPException, Request, status
 from fastapi.concurrency import run_in_threadpool
 
 from app.api.deps import DbSession, OptionalUser
@@ -36,6 +36,7 @@ from app.core.pricebot_router import pick_pricebot
 from app.db.session import SessionLocal
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare")
 
@@ -80,7 +81,7 @@ def _harvest_running_blocking(
 def _harvest_done_blocking(
     trace_id: str, user_id: int | None, done_params: dict, business_type: str,
     device_id: str | None, device_info: dict | None, trace_url: str | None,
-) -> None:
+) -> int:
     with SessionLocal() as db:
         rec, newly_success = crud_compare.harvest_done(
             db, trace_id=trace_id, user_id=user_id, done_params=done_params,
@@ -102,6 +103,7 @@ def _harvest_done_blocking(
         # 不在此处发邀请奖:#113 已把发奖口径从「比价」移到「实际下单」(order.py),harvest
         # 只记录比价、不发奖。否则比价先于下单 + try_reward 幂等闸会让奖落在「比价」这步,
         # 架空 #113 的「下单才发奖」防刷意图(newly_success 仅留作日志观测)。
+        return rec.id
 
 
 def _harvest_abort_blocking(
@@ -243,7 +245,12 @@ async def intent_precoupon_step(
 
 
 @router.post("/price/step", summary="外卖比价 Phase 2 步进 (透传 + done 落库)")
-async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dict[str, Any]:
+async def price_step(
+    request: Request,
+    background_tasks: BackgroundTasks,
+    user: OptionalUser,
+    db: DbSession,
+) -> dict[str, Any]:
     _ensure_compare_allowed(user, db)
     resp, trace_id, meta = await _forward(request, "/api/price/step", user)
     # 最终 done 帧(command=done 且 continue=false)→ harvest 更新成终态。
@@ -252,12 +259,15 @@ async def price_step(request: Request, user: OptionalUser, db: DbSession) -> dic
     if action.get("command") == "done" and not resp.get("continue", True):
         done_params = action.get("params") or {}
         try:
-            await run_in_threadpool(
+            record_id = await run_in_threadpool(
                 _harvest_done_blocking, trace_id, (user.id if user else None),
                 done_params, "food",
                 meta.get("device_id"), meta.get("device_info"),
                 resp.get("trace_url") or done_params.get("trace_url"),
             )
+            background_tasks.add_task(
+                backfill_comparison_llm_cost, record_id, trace_id
+            )
         except Exception as e:  # noqa: BLE001
             logger.warning("harvest_done failed trace=%s: %s", trace_id, e)
     return resp

app/api/v1/compare_record.py

@@ -16,8 +16,6 @@ import logging
 from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, status
 
 from app.api.deps import CurrentUser, DbSession
-from app.db.session import SessionLocal
-from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud_compare
 from app.repositories import risk as risk_repo
 from app.schemas.compare_record import (
@@ -30,8 +28,7 @@ from app.schemas.compare_record import (
     ComparisonRecordOut,
     ComparisonRecordPage,
 )
-from app.services.llm_cost import compute_llm_cost, get_llm_prices
-from app.services.pricebot_llm_calls import fetch_llm_calls
+from app.services.comparison_llm_backfill import backfill_comparison_llm_cost
 
 logger = logging.getLogger("shagua.compare_record")
 
@@ -121,32 +118,7 @@ def report_record(
 def _backfill_llm_calls(record_id: int, trace_id: str) -> None:
     """后台回填本次比价的 LLM 调用明细 + 派生 llm_call_count/retry_count。
     独立 DB session(请求 session 此时已关);拉取/写库失败只 log,绝不影响已落库的上报。"""
-    calls = fetch_llm_calls(trace_id)
-    if not calls:
-        return
-    db = SessionLocal()
-    try:
-        rec = db.get(ComparisonRecord, record_id)
-        if rec is None:
-            return
-        rec.llm_calls = calls
-        rec.llm_call_count = len(calls)
-        rec.retry_count = sum(1 for c in calls if c.get("error"))
-        # token 累加(usage 已被 pricebot llm_client 归一为 prompt/completion_tokens;
-        # error 的调用 usage 可能为 None,or {} 兜底)
-        rec.input_tokens = sum((c.get("usage") or {}).get("prompt_tokens") or 0 for c in calls)
-        rec.output_tokens = sum((c.get("usage") or {}).get("completion_tokens") or 0 for c in calls)
-        # 本次比价 LLM 成本(元)+ 当时单价快照:按 app_config 现价逐模型算好冻结(services/llm_cost.py)。
-        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(calls, get_llm_prices(db))
-        db.commit()
-        logger.info(
-            "backfill llm_calls trace=%s n=%d in_tok=%d out_tok=%d",
-            trace_id, len(calls), rec.input_tokens, rec.output_tokens,
-        )
-    except Exception as e:  # noqa: BLE001 best-effort
-        logger.warning("backfill llm_calls failed trace=%s: %s", trace_id, e)
-    finally:
-        db.close()
+    backfill_comparison_llm_cost(record_id, trace_id)
 
 
 @router.get(

app/core/client_log.py

@@ -0,0 +1,157 @@
+"""客户端运行日志专用落盘 writer(独立于服务端 app-server.log)。
+
+- 独占 logger "shagua.client_log" + 自己的 RotatingFileHandler,propagate=False → 不污染 app-server.log。
+- 每条按「白名单键(client_ts/level/trace_id/tag/msg)提顶层 + 其余并入 data」封装,再
+  json.dumps 成一行写出(钉死 SLS 索引列;见 spec §5)。formatter 用 %(message)s——行本身
+  已是 JSON,不能再过 JsonFormatter 二次编码。
+- 滚动 20MB×10(env 可调),与服务日志同机制。
+  ⚠️ 依赖 --workers 1:RotatingFileHandler 多进程并发 doRollover 会损坏/丢日志;扩 worker
+  前换 QueueHandler→单写入者 / 外部 logrotate(copytruncate) / 写 stdout 交 journald。
+
+服务端补的字段(time/source/service/client_ip/device_id/...)是「事实」,与客户端自述分开。
+`time` 用服务端接收时间作 SLS 主时间(客户端时钟不可信),client_ts 另存为可查字段。
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+# 仅这些客户端键提到输出行顶层;其余(含客户端自带 data)一律并入 data,防 SLS 索引列爆炸
+_TOP_LEVEL_KEYS = ("client_ts", "level", "trace_id", "tag", "msg")
+
+# trace_id/tag/level 是 SLS 索引字段(spec §8):给长度上限,防客户端塞超大值撑爆索引/抬升成本。
+# (msg 另有字节截断;data 内的值不限,留待后续「服务端脱敏」knob。)
+_MAX_LEVEL_LEN = 16
+_MAX_TRACE_ID_LEN = 256
+_MAX_TAG_LEN = 128
+
+_logger: logging.Logger | None = None
+
+
+def _max_msg_bytes() -> int:
+    # 每次调用现读 env(不设模块级常量):便于运行期调整 / 测试 monkeypatch,开销可忽略。
+    return int(os.getenv("APPLOG_MAX_MSG_BYTES", "8192"))
+
+
+def _build_logger() -> logging.Logger:
+    lg = logging.getLogger("shagua.client_log")  # 与仓库 shagua.* 业务 logger 命名一致
+    lg.setLevel(logging.INFO)
+    lg.propagate = False  # 不冒泡到 root → 不写进 app-server.log
+    log_file = os.getenv("CLIENT_LOG_FILE") or str(
+        Path(os.getenv("LOG_DIR", "logs")) / "app-client.log"
+    )
+    Path(log_file).parent.mkdir(parents=True, exist_ok=True)
+    handler = RotatingFileHandler(
+        log_file,
+        maxBytes=int(os.getenv("CLIENT_LOG_MAX_BYTES", str(20 * 1024 * 1024))),
+        backupCount=int(os.getenv("CLIENT_LOG_BACKUP_COUNT", "10")),
+        encoding="utf-8",
+    )
+    handler.setFormatter(logging.Formatter("%(message)s"))  # 行已是 JSON,不再包装
+    lg.handlers = [handler]
+    return lg
+
+
+def get_logger() -> logging.Logger:
+    global _logger
+    if _logger is None:
+        _logger = _build_logger()
+    return _logger
+
+
+def reset_client_logger() -> None:
+    """测试用:关闭并丢弃当前 logger,使下次 get_logger 按当时 env 重建(切临时文件)。"""
+    global _logger
+    if _logger is not None:
+        for h in list(_logger.handlers):
+            h.close()
+        _logger.handlers = []
+    _logger = None
+
+
+def _truncate_msg(msg: str) -> tuple[str, bool]:
+    raw = msg.encode("utf-8")
+    limit = _max_msg_bytes()
+    if len(raw) <= limit:
+        return msg, False
+    # 按字节截断后解码,忽略截断处半个多字节字符
+    return raw[:limit].decode("utf-8", "ignore") + "…[truncated]", True
+
+
+def _build_line(
+    record: dict, *, meta: dict, client_ip: str, service: str, now_iso: str
+) -> str:
+    out: dict = {
+        "time": now_iso,
+        "source": "client",
+        "service": service,
+        "client_ip": client_ip,
+    }
+    # 批级公共字段(非空才带)
+    for k in ("device_id", "user_id", "app_ver", "platform", "sent_at"):
+        v = meta.get(k)
+        if v is not None:
+            out[k] = v
+    # 白名单键提顶层(索引字段做长度上限 + 统一转 str,保证 SLS 里类型/大小可控)
+    if record.get("level") is not None:
+        out["level"] = str(record["level"])[:_MAX_LEVEL_LEN].upper()
+    if record.get("trace_id"):
+        out["trace_id"] = str(record["trace_id"])[:_MAX_TRACE_ID_LEN]
+    if record.get("tag"):
+        out["tag"] = str(record["tag"])[:_MAX_TAG_LEN]
+    if record.get("client_ts") is not None:
+        out["client_ts"] = record["client_ts"]
+    if record.get("msg") is not None:
+        msg, truncated = _truncate_msg(str(record["msg"]))
+        out["msg"] = msg
+        if truncated:
+            out["msg_truncated"] = True
+    # 其余键并入 data:先收白名单外的散键(兜底),再让客户端显式的 data 覆盖同名散键
+    # —— 显式 data 为准,不静默丢客户端明确给的值(保真)。
+    data: dict = {}
+    for k, v in record.items():
+        if k in _TOP_LEVEL_KEYS or k == "data":
+            continue
+        data[k] = v
+    client_data = record.get("data")
+    if isinstance(client_data, dict):
+        data.update(client_data)
+    if data:
+        out["data"] = data
+    return json.dumps(out, ensure_ascii=False, default=str)
+
+
+def write_records(
+    records: list[dict], *, meta: dict, client_ip: str
+) -> tuple[int, int]:
+    """把一批客户端日志逐条写入专用文件。返回 (received, dropped)。
+
+    尽力而为(fire-and-forget):logger 初始化或单条写入失败只跳过并计 dropped,
+    不抛给上层——端点因此永不因写日志而 500。
+    """
+    try:
+        lg = get_logger()
+    except Exception:  # noqa: BLE001 — 初始化失败也不能让端点 500
+        logging.getLogger("shagua.applog").exception("client log writer init failed")
+        return 0, len(records)
+    service = os.getenv("CLIENT_LOG_SERVICE_NAME", "app-client")
+    # 一批共用同一「服务端接收时间」:降开销,且语义上是服务端「收到」而非逐条「处理」时间。
+    now_iso = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
+    received = dropped = 0
+    for rec in records:
+        try:
+            line = _build_line(
+                rec, meta=meta, client_ip=client_ip, service=service, now_iso=now_iso
+            )
+            lg.info(line)
+            received += 1
+        except Exception:  # noqa: BLE001 — 坏条跳过,不影响其余
+            logging.getLogger("shagua.applog").exception(
+                "client log record dropped client_ip=%s", client_ip
+            )
+            dropped += 1
+    return received, dropped

app/core/config.py

@@ -377,6 +377,12 @@ class Settings(BaseSettings):
     # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
     # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
     INTERNAL_API_SECRET: str = ""
+    # Current compare clients are persisted by server-side harvest. Repair any
+    # recent terminal rows left without LLM usage by transient upstream/auth failures.
+    LLM_COST_BACKFILL_ENABLED: bool = True
+    LLM_COST_BACKFILL_INTERVAL_SEC: int = 300
+    LLM_COST_BACKFILL_BATCH_SIZE: int = 100
+    LLM_COST_BACKFILL_LOOKBACK_DAYS: int = 30
 
     # ===== 媒体文件(用户头像上传)=====
     # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。

app/core/llm_cost_backfill_worker.py

@@ -0,0 +1,63 @@
+"""Periodic repair worker for comparison records with missing LLM token cost."""
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+from app.core.config import settings
+from app.services.comparison_llm_backfill import repair_missing_comparison_llm_costs
+from app.services.pricebot_llm_calls import pricebot_llm_auth_ready
+
+logger = logging.getLogger("shagua.llm_cost_backfill_worker")
+
+
+async def _run_loop() -> None:
+    interval = max(60, int(settings.LLM_COST_BACKFILL_INTERVAL_SEC))
+    logger.info(
+        "LLM cost backfill worker started interval=%ss batch=%s lookback_days=%s",
+        interval,
+        settings.LLM_COST_BACKFILL_BATCH_SIZE,
+        settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
+    )
+    try:
+        while True:
+            try:
+                auth_ready = await asyncio.to_thread(pricebot_llm_auth_ready)
+                if auth_ready:
+                    result = await asyncio.to_thread(
+                        repair_missing_comparison_llm_costs,
+                        limit=settings.LLM_COST_BACKFILL_BATCH_SIZE,
+                        lookback_days=settings.LLM_COST_BACKFILL_LOOKBACK_DAYS,
+                    )
+                    logger.info("LLM cost backfill batch result=%s", result)
+                else:
+                    logger.error(
+                        "LLM cost backfill skipped: PriceBot internal auth is not ready"
+                    )
+            except Exception:  # noqa: BLE001
+                logger.exception("LLM cost backfill batch failed")
+            await asyncio.sleep(interval)
+    except asyncio.CancelledError:
+        logger.info("LLM cost backfill worker stopped")
+        raise
+
+
+def start_llm_cost_backfill_worker() -> asyncio.Task | None:
+    if not settings.LLM_COST_BACKFILL_ENABLED:
+        logger.info("LLM cost backfill worker disabled")
+        return None
+    if not settings.INTERNAL_API_SECRET:
+        logger.warning(
+            "LLM cost backfill worker not started: INTERNAL_API_SECRET is empty"
+        )
+        return None
+    return asyncio.create_task(_run_loop(), name="llm-cost-backfill")
+
+
+async def stop_llm_cost_backfill_worker(task: asyncio.Task | None) -> None:
+    if task is None:
+        return
+    task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await task

app/main.py

@@ -21,6 +21,7 @@ from app.api.internal.price import router as internal_price_router
 from app.api.internal.store import router as internal_store_router
 from app.api.v1.ad import router as ad_router
 from app.api.v1.analytics import router as analytics_router
+from app.api.v1.applog import router as applog_router
 from app.api.v1.auth import router as auth_router
 from app.api.v1.compare import router as compare_router
 from app.api.v1.compare_milestone import router as compare_milestone_router
@@ -60,6 +61,10 @@ from app.core.inactivity_reset_worker import (
     start_inactivity_reset_worker,
     stop_inactivity_reset_worker,
 )
+from app.core.llm_cost_backfill_worker import (
+    start_llm_cost_backfill_worker,
+    stop_llm_cost_backfill_worker,
+)
 from app.core.logging import setup_logging
 from app.core.observe import RequestMetricsMiddleware
 from app.core.observe_worker import (
@@ -103,6 +108,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
     daily_exchange_task = start_daily_exchange_worker()
     observe_task = start_observe_worker()
     inactivity_task = start_inactivity_reset_worker()
+    llm_cost_backfill_task = start_llm_cost_backfill_worker()
     try:
         yield
     finally:
@@ -112,6 +118,7 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         await stop_daily_exchange_worker(daily_exchange_task)
         await stop_observe_worker(observe_task)
         await stop_inactivity_reset_worker(inactivity_task)
+        await stop_llm_cost_backfill_worker(llm_cost_backfill_task)
         await aclose_pricebot_client()
         mt_meituan.close_client()
         logger.info("shutting down")
@@ -147,6 +154,7 @@ app.include_router(auth_router)
 app.include_router(user_router)
 app.include_router(feedback_router)
 app.include_router(analytics_router)
+app.include_router(applog_router)
 app.include_router(invite_router)
 app.include_router(coupon_router)
 app.include_router(device_router)

app/models/user.py

@@ -11,7 +11,7 @@ from __future__ import annotations
 
 from datetime import datetime, timezone
 
-from sqlalchemy import Boolean, DateTime, Integer, String, false, func
+from sqlalchemy import Boolean, DateTime, Integer, String, Text, false, func
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.db.base import Base
@@ -62,6 +62,13 @@ class User(Base):
         Boolean, nullable=False, default=False, server_default=false()
     )
 
+    # 运营人工风险标记。与自动风控分值分开：这里表达人工复核结论，并保留可编辑备注，
+    # 供邀请提现、其他提现和用户管理三个页面统一展示。
+    is_high_risk: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False, server_default=false(), index=True
+    )
+    high_risk_note: Mapped[str | None] = mapped_column(Text, nullable=True)
+
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False
     )

app/repositories/wallet.py

@@ -6,6 +6,7 @@
 """
 from __future__ import annotations
 
+import logging
 import re
 import unicodedata
 import uuid
@@ -20,7 +21,6 @@ from app.core.config import settings
 from app.core.rewards import COIN_PER_CENT, coins_to_cents
 from app.integrations import wxpay
 from app.models.user import User
-from app.repositories.user import apply_wechat_display_identity
 from app.models.wallet import (
     CashTransaction,
     CoinAccount,
@@ -29,12 +29,55 @@ from app.models.wallet import (
     WechatTransferAuthorization,
     WithdrawOrder,
 )
+from app.repositories.user import apply_wechat_display_identity
 from app.services import notification_events
 
+logger = logging.getLogger(__name__)
+
 # 微信转账终态:成功 / 失败(失败/取消/关闭都退款)
 _WX_STATE_SUCCESS = "SUCCESS"
 _WX_STATE_FAILED = {"FAIL", "CANCELLED", "CLOSED"}
 _WX_STATE_WAIT_CONFIRM = "WAIT_USER_CONFIRM"  # 用户还没在微信确认页确认
+_WX_FAIL_REASON_LABELS = {
+    "ACCOUNT_FROZEN": "用户微信账户被冻结",
+    "ACCOUNT_NOT_EXIST": "用户微信账户不存在",
+    "BANK_CARD_ACCOUNT_ABNORMAL": "用户银行卡已销户、冻结、作废或挂失",
+    "BANK_CARD_BANK_INFO_WRONG": "用户登记的银行或分支行信息有误",
+    "BANK_CARD_CARD_INFO_WRONG": "用户银行卡户名或卡号有误",
+    "BANK_CARD_COLLECTIONS_ABOVE_QUOTA": "用户银行卡收款达到限额",
+    "BANK_CARD_PARAM_ERROR": "用户收款银行卡信息错误",
+    "BANK_CARD_STATUS_ABNORMAL": "用户银行卡状态异常",
+    "BLOCK_B2C_USERLIMITAMOUNT_BSRULE_MONTH": "用户本月转账收款已达限额",
+    "BLOCK_B2C_USERLIMITAMOUNT_MONTH": "用户账户存在风险，本月收款受限",
+    "DAY_RECEIVED_COUNT_EXCEED": "用户当日收款次数已达上限",
+    "DAY_RECEIVED_QUOTA_EXCEED": "用户当日收款额度已达上限",
+    "EXCEEDED_ESTIMATED_AMOUNT": "转账金额超过预约金额范围",
+    "ID_CARD_NOT_CORRECT": "收款人身份证校验不通过",
+    "MCH_CANCEL": "商户已撤销付款",
+    "MERCHANT_REJECT": "商户转账验密人已驳回",
+    "MERCHANT_NOT_CONFIRM": "商户转账验密人超时未确认",
+    "NAME_NOT_CORRECT": "收款人姓名校验不通过",
+    "OPENID_INVALID": "用户 OpenID 无效或不属于当前 AppID",
+    "OTHER_FAIL_REASON_TYPE": "微信返回其他失败原因",
+    "OVERDUE_CLOSE": "超过微信系统重试期，订单自动关闭",
+    "PAYEE_ACCOUNT_ABNORMAL": "用户微信账户收款异常",
+    "PAYER_ACCOUNT_ABNORMAL": "商户账户付款受限",
+    "PRODUCT_AUTH_CHECK_FAIL": "商户未开通转账权限或权限已冻结",
+    "REALNAME_ACCOUNT_RECEIVED_QUOTA_EXCEED": "用户微信实名账户收款受限",
+    "REAL_NAME_CHECK_FAIL": "用户未完成微信实名认证",
+    "RECEIVE_ACCOUNT_NOT_CONFIGURE": "商户未配置收款用户列表",
+    "RESERVATION_INFO_NOT_MATCH": "转账信息与预约信息不一致",
+    "RESERVATION_SCENE_NOT_MATCH": "转账场景与预约场景不一致",
+    "RESERVATION_STATE_INVALID": "预约转账单状态异常",
+    "TRANSFER_QUOTA_EXCEED": "用户单笔收款额度已达上限",
+    "TRANSFER_REMARK_SET_FAIL": "微信转账备注设置失败",
+    "TRANSFER_RISK": "该笔转账存在风险，已被微信拦截",
+    "TRANSFER_SCENE_INVALID": "商户未获取当前转账场景",
+    "TRANSFER_SCENE_UNAVAILABLE": "当前转账场景暂不可用",
+    "RELATED_ORDER_TRANSFER_AMOUNT_EXCEED": "关联订单累计付款金额超过上限",
+    "RELATED_ORDER_TRANSFER_COUNT_EXCEED": "关联订单累计付款次数超过上限",
+    "BUDGET_NOT_ENOUGH": "商户预算资金不足",
+}
 # 占用新人档「一次性」资格的提现状态:进行中(reviewing/pending)或成功打款(success)。
 # 被拒/转账失败/解绑退回(rejected/failed,均已退款、钱没到手)不在此列 → 新人档恢复可提
 # (2026-07-16 修正:此前判定不看状态,解绑微信退回后 0.1 被误判已用、资格永久锁死)。
@@ -479,6 +522,19 @@ def _try_deduct_cash(db: Session, user_id: int, amount_cents: int, source: str =
 
 def _add_cash(db: Session, user_id: int, amount_cents: int, source: str = "coin_cash") -> int:
     """原子增加指定账户余额(退款用),返回加后余额。source 决定退回哪个账户(两账户隔离)。"""
+    # Normal withdrawals always create the account before deducting funds. This
+    # fallback covers legacy rows, hand-written fixtures, and broken migrations:
+    # a missing balance snapshot must not make a legitimate refund fail with 500.
+    if db.get(CoinAccount, user_id) is None:
+        logger.error(
+            "withdraw refund found missing coin_account; recreating empty account: "
+            "user_id=%s source=%s amount_cents=%s",
+            user_id,
+            source,
+            amount_cents,
+        )
+        get_or_create_account(db, user_id, commit=False)
+
     col = _balance_col(source)
     db.execute(
         update(CoinAccount)
@@ -582,6 +638,32 @@ def _wx_not_found(result: dict) -> bool:
     return "NOT_FOUND" in str(code)
 
 
+def _wechat_api_error_reason(data: object) -> str:
+    """Format a non-200 WeChat API response for operator display."""
+    if not isinstance(data, dict):
+        return f"微信发起转账失败：{data}"
+    code = str(data.get("code") or "").strip()
+    message = str(data.get("message") or "").strip()
+    if code and message:
+        return f"微信发起转账失败：{message}（{code}）"
+    return f"微信发起转账失败：{message or code or '未知错误'}"
+
+
+def _wechat_terminal_failure_reason(data: dict, state: str) -> str:
+    """Translate WeChat query ``fail_reason`` while preserving unknown codes."""
+    code = str(data.get("fail_reason") or "").strip()
+    if code:
+        label = _WX_FAIL_REASON_LABELS.get(code)
+        if label:
+            return f"微信转账失败：{label}（{code}）"
+        return f"微信转账失败：{code}"
+    if state == "CANCELLED":
+        return "微信转账已撤销（CANCELLED）"
+    if state == "CLOSED":
+        return "微信转账已关闭（CLOSED）"
+    return f"微信转账失败（状态：{state or 'FAIL'}）"
+
+
 def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> None:
     """转账调用结果不明(超时/异常/非200)时,**先查单再决定**,绝不盲目退款(防退款后又到账)。
       - 微信查到 SUCCESS        → 钱已出,置 success,不退款
@@ -607,13 +689,17 @@ def _settle_after_ambiguous(db: Session, order: WithdrawOrder, reason: str) -> N
 
     state = q["data"].get("state", "")
     order.wechat_state = state
+    order.transfer_bill_no = q["data"].get("transfer_bill_no") or order.transfer_bill_no
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
-        order.transfer_bill_no = q["data"].get("transfer_bill_no")
         db.commit()
         notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
-        _refund_withdraw(db, order, reason=reason)
+        _refund_withdraw(
+            db,
+            order,
+            reason=_wechat_terminal_failure_reason(q["data"], state),
+        )
     else:
         order.package_info = q["data"].get("package_info") or order.package_info
         db.commit()
@@ -967,6 +1053,14 @@ def _apply_transfer_result(db: Session, order: WithdrawOrder, data: dict) -> Wit
     order.package_info = data.get("package_info")  # 免确认转账无此字段(None);确认模式带它供拉确认页
     if data.get("state") == _WX_STATE_SUCCESS:
         order.status = "success"
+    elif data.get("state") in _WX_STATE_FAILED:
+        _refund_withdraw(
+            db,
+            order,
+            reason=_wechat_terminal_failure_reason(data, str(data.get("state") or "")),
+        )
+        db.refresh(order)
+        return order
     db.commit()
     db.refresh(order)
     if order.status == "success":  # 免确认转账直接到账 → PRD #3 提现到账
@@ -1011,7 +1105,11 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
             return order
         if result["status_code"] != 200:
             # 金额安全:查转账单后定夺,绝不盲退(未创建→退款,已创建→按真实状态)
-            _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
+            _settle_after_ambiguous(
+                db,
+                order,
+                reason=_wechat_api_error_reason(result["data"]),
+            )
             # 授权有效性:回查授权单,微信侧已失效(用户关闭/风控)→标 closed,下次提现自动回退方式一重新授权
             _refresh_active_auth(db, order.user_id)
             db.refresh(order)
@@ -1039,7 +1137,11 @@ def execute_withdraw_transfer(db: Session, order: WithdrawOrder) -> WithdrawOrde
         return order
 
     if result["status_code"] != 200:
-        _settle_after_ambiguous(db, order, reason=str(result["data"].get("message") or result["data"]))
+        _settle_after_ambiguous(
+            db,
+            order,
+            reason=_wechat_api_error_reason(result["data"]),
+        )
         db.refresh(order)
         return order
 
@@ -1096,19 +1198,24 @@ def refresh_withdraw_status(
     ).scalar_one_or_none()
     if order is None:
         raise WithdrawOrderNotFound
-    if order.status != "pending":
-        return order  # 已终态,不再查
+    if order.status not in {"pending", "failed"}:
+        return order
+    enrich_failed_order = order.status == "failed"
 
     try:
         result = wxpay.query_transfer(out_bill_no)
     except wxpay.WxPayNotConfiguredError:
         raise
     except Exception as exc:  # noqa: BLE001 - 查单失败不能把运营后台打成 500
+        if enrich_failed_order:
+            return order
         order.fail_reason = f"微信查单异常,保持pending: {exc}"[:256]
         db.commit()
         db.refresh(order)
         return order
     if result["status_code"] != 200:
+        if enrich_failed_order:
+            return order
         if _wx_not_found(result):
             # 微信明确无此单 → 转账从未创建(如崩溃在扣款后/调用前),退款安全
             _refund_withdraw(db, order, reason="微信无此单,已退回")
@@ -1117,12 +1224,32 @@ def refresh_withdraw_status(
 
     state = result["data"].get("state", "")
     order.wechat_state = state
+    order.transfer_bill_no = (
+        result["data"].get("transfer_bill_no") or order.transfer_bill_no
+    )
+    if enrich_failed_order:
+        if state in _WX_STATE_FAILED:
+            order.fail_reason = _wechat_terminal_failure_reason(
+                result["data"], state
+            )[:256]
+        elif state == _WX_STATE_SUCCESS:
+            order.fail_reason = (
+                "资金状态异常：本地已退款，但微信查单显示已到账，请人工核查"
+            )
+        db.commit()
+        db.refresh(order)
+        return order
+
     if state == _WX_STATE_SUCCESS:
         order.status = "success"
         db.commit()
         notification_events.notify_withdraw_success(db, order)  # PRD #3 提现到账
     elif state in _WX_STATE_FAILED:
-        _refund_withdraw(db, order, reason=f"微信转账状态 {state}")
+        _refund_withdraw(
+            db,
+            order,
+            reason=_wechat_terminal_failure_reason(result["data"], state),
+        )
     elif state == _WX_STATE_WAIT_CONFIRM and cancel_if_unconfirmed:
         # 用户从确认页回来了却仍未确认 → 视为放弃:撤销微信单(防事后确认导致重复打款)后退款。
         # 撤单失败(可能已被确认进 ACCEPTED 的竞态)则保持 pending,等下次查询。

app/schemas/applog.py

@@ -0,0 +1,31 @@
+"""客户端运行日志批量上报 schema。
+
+批级公共字段(device_id/user_id/app_ver/platform/sent_at)发一次;logs 为原始 dict 列表,
+**不强类型**——尽力而为的日志链路,单条内容异常不该让整批 422。每条的「白名单键 + data
+兜底」拆分在写入层 [app.core.client_log] 做(见 spec §4/§5)。
+"""
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+# 单批条数上限。注意:这是**导入期**常量(Pydantic Field(max_length=) 在类定义时求值),
+# 改它需重启进程;要运行期可调的上限用 APPLOG_MAX_BODY_BYTES(端点里 call-time 读)。
+_MAX_BATCH = int(os.getenv("APPLOG_MAX_BATCH", "500"))
+
+
+class AppLogBatchIn(BaseModel):
+    device_id: str = Field(max_length=64)
+    user_id: int | None = None
+    app_ver: str | None = Field(default=None, max_length=32)
+    platform: str | None = Field(default=None, max_length=16)
+    sent_at: int | None = None
+    logs: list[dict[str, Any]] = Field(min_length=1, max_length=_MAX_BATCH)
+
+
+class AppLogIngestOut(BaseModel):
+    ok: bool = True
+    received: int
+    dropped: int = 0

app/schemas/compare_record.py

@@ -85,6 +85,10 @@ class ComparisonResultIn(BaseModel):
     status: str | None = None
     # 门店打烊原因(price 为 None 时带): 与 status="store_closed" 等价的更早信号, 一并落库供前端兜底判打烊。
     store_closed: str | None = None
+    # 该平台缺菜(漏菜)数量(pricebot 冗余进 comparison_results 行): 有价但少买了菜时 >0。
+    # 记录页三平台网格据此逐格标"缺少 X 个菜品"(网格只拿 comparison_results, 拿不到 platform_results)。
+    # 必须显式声明: 落库走 model_dump(), 不声明会被 pydantic 静默丢弃 → 记录页拿不到数量。
+    skipped_dish_count: int | None = None
 
 
 class ComparisonRecordIn(BaseModel):

app/services/comparison_llm_backfill.py

@@ -0,0 +1,156 @@
+"""Persist and repair comparison-record LLM token costs."""
+from __future__ import annotations
+
+import logging
+import time
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services.llm_cost import compute_llm_cost, get_llm_prices
+from app.services.pricebot_llm_calls import fetch_llm_calls
+
+logger = logging.getLogger("shagua.comparison_llm_backfill")
+
+
+def _utc_to_beijing_naive(value: datetime) -> datetime:
+    """Convert a DB UTC timestamp to comparison_record's Beijing wall-clock."""
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=UTC)
+    return value.astimezone(CN_TZ).replace(tzinfo=None)
+
+
+def _store_calls(record_id: int, trace_id: str, calls: list[dict]) -> bool:
+    """Store calls and all derived fields atomically."""
+    with SessionLocal() as db:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is None or rec.trace_id != trace_id:
+            logger.warning(
+                "LLM cost backfill record mismatch record_id=%s trace=%s",
+                record_id,
+                trace_id,
+            )
+            return False
+
+        # Never recalculate a frozen historical cost with a newer price config.
+        if rec.llm_cost_yuan is not None and rec.llm_calls:
+            return False
+
+        rec.llm_calls = calls
+        rec.llm_call_count = len(calls)
+        rec.retry_count = sum(1 for call in calls if call.get("error"))
+        rec.input_tokens = sum(
+            (call.get("usage") or {}).get("prompt_tokens") or 0 for call in calls
+        )
+        rec.output_tokens = sum(
+            (call.get("usage") or {}).get("completion_tokens") or 0 for call in calls
+        )
+        rec.llm_cost_yuan, rec.llm_price_snapshot = compute_llm_cost(
+            calls, get_llm_prices(db)
+        )
+        db.commit()
+        logger.info(
+            "LLM cost backfilled trace=%s calls=%d input_tokens=%d "
+            "output_tokens=%d cost=%s",
+            trace_id,
+            len(calls),
+            rec.input_tokens,
+            rec.output_tokens,
+            rec.llm_cost_yuan,
+        )
+        return True
+
+
+def backfill_comparison_llm_cost(
+    record_id: int,
+    trace_id: str,
+    *,
+    attempts: int = 3,
+    retry_delays: tuple[float, ...] = (1.0, 3.0),
+) -> bool:
+    """Fetch and persist one record, retrying short-lived upstream races."""
+    if not settings.INTERNAL_API_SECRET or not trace_id:
+        logger.warning(
+            "LLM cost backfill skipped trace=%s: INTERNAL_API_SECRET is not configured",
+            trace_id,
+        )
+        return False
+    total_attempts = max(1, attempts)
+    for attempt in range(total_attempts):
+        calls = fetch_llm_calls(trace_id)
+        if calls:
+            try:
+                return _store_calls(record_id, trace_id, calls)
+            except Exception:  # noqa: BLE001 - background repair must stay alive
+                logger.exception(
+                    "LLM cost store failed trace=%s record_id=%s",
+                    trace_id,
+                    record_id,
+                )
+                return False
+
+        if attempt + 1 < total_attempts:
+            delay = retry_delays[min(attempt, len(retry_delays) - 1)] if retry_delays else 0
+            if delay > 0:
+                time.sleep(delay)
+
+    logger.warning(
+        "LLM cost backfill has no calls trace=%s record_id=%s attempts=%d",
+        trace_id,
+        record_id,
+        total_attempts,
+    )
+    return False
+
+
+def repair_missing_comparison_llm_costs(
+    *,
+    limit: int = 100,
+    lookback_days: int = 30,
+) -> dict[str, int]:
+    """Repair a bounded batch of recent terminal records with missing cost."""
+    cutoff = datetime.now(CN_TZ).replace(tzinfo=None) - timedelta(
+        days=max(1, lookback_days)
+    )
+    with SessionLocal() as db:
+        # app_config has no price history. Repricing a record from before the
+        # current config became effective would fabricate a historical cost, so
+        # only repair records at/after that timestamp.
+        price_config_updated_at = db.execute(
+            select(AppConfig.updated_at).where(AppConfig.key == "llm_token_price")
+        ).scalar_one_or_none()
+        date_conditions = [ComparisonRecord.created_at >= cutoff]
+        if price_config_updated_at is not None:
+            date_conditions.append(
+                ComparisonRecord.created_at
+                >= _utc_to_beijing_naive(price_config_updated_at)
+            )
+        candidates = list(
+            db.execute(
+                select(ComparisonRecord.id, ComparisonRecord.trace_id)
+                .where(
+                    *date_conditions,
+                    ComparisonRecord.status.in_(("success", "failed")),
+                    ComparisonRecord.llm_cost_yuan.is_(None),
+                )
+                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
+                .limit(max(1, limit))
+            ).all()
+        )
+
+    repaired = 0
+    for record_id, trace_id in candidates:
+        if backfill_comparison_llm_cost(
+            record_id, trace_id, attempts=1, retry_delays=()
+        ):
+            repaired += 1
+    return {
+        "candidates": len(candidates),
+        "repaired": repaired,
+        "unresolved": len(candidates) - repaired,
+    }

app/services/pricebot_llm_calls.py

@@ -20,19 +20,67 @@ from app.core.pricebot_router import pick_pricebot
 logger = logging.getLogger("shagua.pricebot_llm")
 
 
+def pricebot_llm_auth_ready() -> bool:
+    """Verify every configured PriceBot instance accepts the shared secret."""
+    secret = settings.INTERNAL_API_SECRET
+    if not secret:
+        logger.error("PriceBot LLM auth check failed: INTERNAL_API_SECRET is empty")
+        return False
+    for base in settings.pricebot_instances:
+        url = f"{base.rstrip('/')}/api/internal/llm_calls/__auth_probe__"
+        try:
+            resp = httpx.get(
+                url, headers={"X-Internal-Secret": secret}, timeout=3.0
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.error("PriceBot LLM auth check unavailable base=%s: %s", base, exc)
+            return False
+        if resp.status_code != 200:
+            logger.error(
+                "PriceBot LLM auth check rejected base=%s status=%s; "
+                "verify both services use the same INTERNAL_API_SECRET",
+                base,
+                resp.status_code,
+            )
+            return False
+    return True
+
+
 def fetch_llm_calls(trace_id: str) -> list[dict]:
     """返回该次比价的 LLM 调用明细列表(每条 {scene,model,input_messages,output,usage,latency_ms,error});
     未配密钥 / 无 trace_id / 拉取失败 → []。"""
     secret = settings.INTERNAL_API_SECRET
     if not secret or not trace_id:
         return []
-    base = pick_pricebot(trace_id).rstrip("/")
-    url = f"{base}/api/internal/llm_calls/{trace_id}"
-    try:
-        resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
-        if resp.status_code == 200:
-            return resp.json().get("calls", []) or []
-        logger.warning("fetch_llm_calls trace=%s status=%s", trace_id, resp.status_code)
-    except Exception as e:  # noqa: BLE001 — best-effort,任何异常都不该影响上报
-        logger.warning("fetch_llm_calls trace=%s failed: %s", trace_id, e)
+    preferred = pick_pricebot(trace_id)
+    # LLM JSONL is instance-local. If the cluster topology changed after a
+    # historical trace was created, consistent hashing may now point elsewhere;
+    # probe the remaining configured instances only when the preferred one is empty.
+    bases = [preferred, *(base for base in settings.pricebot_instances if base != preferred)]
+    for base in bases:
+        url = f"{base.rstrip('/')}/api/internal/llm_calls/{trace_id}"
+        try:
+            resp = httpx.get(url, headers={"X-Internal-Secret": secret}, timeout=5.0)
+            if resp.status_code == 200:
+                calls = resp.json().get("calls", []) or []
+                if calls:
+                    return calls
+                continue
+            if resp.status_code in (401, 403):
+                logger.error(
+                    "fetch_llm_calls rejected trace=%s base=%s status=%s; "
+                    "INTERNAL_API_SECRET differs between app-server and PriceBot",
+                    trace_id,
+                    base,
+                    resp.status_code,
+                )
+            else:
+                logger.warning(
+                    "fetch_llm_calls trace=%s base=%s status=%s",
+                    trace_id,
+                    base,
+                    resp.status_code,
+                )
+        except Exception as e:  # noqa: BLE001 — best-effort
+            logger.warning("fetch_llm_calls trace=%s base=%s failed: %s", trace_id, base, e)
     return []

deploy/comparison-llm-cost.md

@@ -0,0 +1,40 @@
+# 比价 TOKEN 成本采集与补偿
+
+## 部署前置
+
+App Server 与 PriceBot 使用各自独立的 `.env`，但下面的值必须完全一致：
+
+- `/opt/shaguabijia-app-server/.env`
+- `/opt/pricebot-backend/.env`
+- 配置项：`INTERNAL_API_SECRET`
+
+不要把密钥原文写入日志、命令历史或 Git。修改后同时重启两个服务。
+
+App Server 启动后会逐个探测 `PRICEBOT_INSTANCES` 的内部读取接口。鉴权不一致时会记录
+`PriceBot LLM auth check rejected`，并跳过本轮补偿，避免对所有缺失记录重复发送失败请求。
+
+## 数据链路
+
+1. 当前客户端由 App Server 在 PriceBot 最终 `done` 帧到达时 harvest 比价记录。
+2. harvest 成功后立即异步读取同一 `trace_id` 的 LLM 调用，冻结 Token、成本和单价快照。
+3. 周期 worker 扫描近期 `success/failed` 且 `llm_cost_yuan IS NULL` 的记录进行补偿；
+   为避免用现价伪造历史成本，只处理当前单价配置生效时间之后的记录。
+4. 管理后台顶部“平均 TOKEN 成本”使用筛选范围内已冻结成本的数据库平均值。
+
+## 上线验收（只读 SQL）
+
+```sql
+SELECT
+  (created_at AT TIME ZONE 'Asia/Shanghai')::date AS day,
+  count(*) AS records,
+  count(llm_cost_yuan) AS cost_records,
+  round(avg(llm_cost_yuan)::numeric, 6) AS avg_token_cost
+FROM comparison_record
+WHERE created_at >= now() - interval '3 days'
+GROUP BY 1
+ORDER BY 1 DESC;
+```
+
+新产生的正常终态比价记录应在短时间内写入 `input_tokens`、`output_tokens` 和
+`llm_cost_yuan`。历史记录只有在 PriceBot 的对应 trace JSONL 仍保留时才能准确回填；
+原始调用已经清理的记录不能用估算值冒充真实成本。

docs/api/README.md

@@ -1,13 +1,9 @@
 # 傻瓜比价 App 后端 — API 接口文档（索引）
 
 > Base URL:生产 `https://app-api.shaguabijia.com`;本地联调 `http://<开发机>:8770`
-> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**(⚠️ 例外:消息通知中心 `notifications` 族与厂商推送 `push` 族按 PRD 前端契约用 **camelCase**,见各自文档)
+> 协议:HTTP / JSON,请求与响应体均 `application/json`,字段统一 **snake_case**
 > 鉴权:需鉴权的接口在请求头带 `Authorization: Bearer <access_token>`
-<<<<<<< HEAD
-> 最后更新:2026-07-14（新增 **消息通知中心** 3 端点(M1-M3,虚拟数据阶段)与 **厂商推送测试** 3 端点(P1-P3,荣耀/华为/小米/OPPO/vivo);上一次 2026-06-23 补全 device/internal/CPS 短链等整族端点）
-=======
 > 最后更新:2026-07-09（① 比价透传改「软鉴权 + trace_id 签发 + harvest 落库」(#112 尾声帧 `trace/epilogue` 一并补录);② 新端点:`user/onboarding/reset`(#114)、`GET /internal/launch-confirm-samples`(#91);③ 参数更新:提现族 `source` 分账(#82/#121)、`wallet/account` 邀请奖励金余额、美团 feed/top-sales 按城市过滤(#116)、admin 调现金 `account` 目标账户(#95);④ **Admin 索引补全到当前全量**:新家族 roles(#117/#126)/coupon-data(#99)/device-liveness(#80)/event-logs(#83)/price-reports(#94)/CPS 运营台/提现审核族,及 feedbacks 采纳拒绝(#94/#105)、marquee 模式与真实条浏览(#122/#123)等。上一次 2026-07-03）
->>>>>>> origin/main
 > 架构:`app/api/v1/` 只放很轻的接口层;穿山甲/微信支付/极光/短信/美团等 SDK 集成的重逻辑在 `app/integrations/`,实现细节见 [docs/integrations/](../integrations/README.md)。
 
 ---
@@ -82,6 +78,7 @@
 | **签到**（前缀 `/api/v1/signin`） |||
 | 25 | `GET /api/v1/signin/status` | Bearer | [详情](./signin/signin-status.md) |
 | 26 | `POST /api/v1/signin` | Bearer | [详情](./signin/signin-do.md) |
+| 26a | `POST /api/v1/signin/boost` | Bearer | [详情](./signin/signin-boost.md) |
 | **任务**（前缀 `/api/v1/tasks`） |||
 | 27 | `GET /api/v1/tasks` | Bearer | [详情](./tasks/tasks-list.md) |
 | 28 | `POST /api/v1/tasks/{task_key}/claim` | Bearer | [详情](./tasks/tasks-claim.md) |
@@ -92,7 +89,6 @@
 | **看广告发奖**（前缀 `/api/v1/ad`） |||
 | 32 | `GET /api/v1/ad/pangle-callback` | 验签 | [详情](./ad/ad-pangle-callback.md) |
 | 33 | `GET /api/v1/ad/reward-status` | Bearer | [详情](./ad/ad-reward-status.md) |
-| 33a | `GET /api/v1/ad/reward-result/{ad_session_id}` | Bearer | [详情](./ad/ad-reward-result.md)（本次实发金币 + 本轮膨胀累计 `round_coin`,弹窗数字用它） |
 | 34 | `POST /api/v1/ad/test-grant` | Bearer | [详情](./ad/ad-test-grant.md) |
 | 35 | `POST /api/v1/ad/ecpm-report` | Bearer | [详情](./ad/ad-ecpm-report.md) |
 | 35a | `POST /api/v1/ad/feed-reward` | Bearer | [详情](./ad/ad-feed-reward.md) |
@@ -107,33 +103,19 @@
 | 36c | `POST /api/v1/user/onboarding/reset` | Bearer | [详情](./user/user-onboarding.md)（重置本设备引导标记,下次登录重走,#114） |
 | 37 | `DELETE /api/v1/user` | Bearer | [详情](./user/user-delete.md) |
 | **帮助与反馈**（前缀 `/api/v1/feedback`） |||
-<<<<<<< HEAD
-| 38 | `POST /api/v1/feedback` | Bearer | [详情](./feedback.md) |
-| 38a | `GET /api/v1/feedback/config` | Bearer | 反馈页「加群二维码」卡配置（开关 + 二维码图 + 三行文案）(无单独文档) |
-| 38b | `GET /api/v1/feedback/records` | Bearer | 我的反馈历史（pending/adopted/rejected）(无单独文档) |
-| **消息通知中心**（前缀 `/api/v1/notifications`;⚠️ 本族对外 **camelCase**;虚拟数据阶段:内存 mock,重启复位） |||
-| M1 | `GET /api/v1/notifications` | Bearer | [详情](./notifications.md)（消息列表,分页;13 类型卡片字段 + sentAt/isRead;服务端已按时间倒序排好,不分组） |
-| M2 | `GET /api/v1/notifications/unread-count` | Bearer | [详情](./notifications.md)（未读总数,首页铃铛角标;>99 → "99+",0 → null 隐藏） |
-| M3 | `POST /api/v1/notifications/read` | Bearer | [详情](./notifications.md)（标记已读:`{ids:[...]}` 单条/多条 或 `{all:true}` 进通知中心全量清零;幂等） |
-| **厂商推送测试**（前缀 `/api/v1/push`;荣耀/华为/小米/OPPO/vivo 五通道联调三件套,同为 camelCase） |||
-| P1 | `GET /api/v1/push/vendors` | Bearer | [详情](./push-vendor-test.md)（5 厂商服务端凭据配置状态,缺哪些 .env 键一目了然） |
-| P2 | `GET /api/v1/push/templates` | Bearer | [详情](./push-vendor-test.md)（13 类通知的 push 标题/正文模板 + PRD 示例渲染效果） |
-| P3 | `POST /api/v1/push/test` | Bearer | [详情](./push-vendor-test.md)（测试发送:默认 mock 不真发;mock=false 真发;可联动插一条站内 mock 通知闭环验证已读） |
-=======
 | 38 | `POST /api/v1/feedback` | Bearer | [详情](./other/feedback.md) |
 | 38a | `GET /api/v1/feedback/config` | Bearer | [详情](./other/feedback-config.md)（反馈页「加群二维码」卡配置:开关+二维码图+三行文案） |
 | 38b | `GET /api/v1/feedback/records` | Bearer | [详情](./other/feedback-records.md)（我的反馈历史,pending/adopted/rejected） |
-| **埋点 & 订单上报**（前缀分散;全部 Bearer 除 analytics/events 不强制登录） |||
+| **埋点 & 订单上报 / 客户端日志**（前缀分散;全部 Bearer 除 analytics/events、applog/batch 不强制登录） |||
 | E1 | `POST /api/v1/analytics/events` | 无 | [详情](./other/analytics-events.md)（批量上报埋点事件,不强制登录,每批最多200条） |
 | E2 | `POST /api/v1/order/report` | Bearer | [详情](./other/order-report.md)（上报归因订单,比价后5分钟内点链接+支付金额与比价价相差≤1元） |
->>>>>>> origin/main
+| E3 | `POST /api/v1/applog/batch` | 无 | [详情](./other/applog-batch.md)（批量上报客户端运行日志,逐条落独立文件供 Logtail 采进 SLS,不强制登录,每批≤500条） |
 | **首页门面数据 / 客户端配置**（前缀 `/api/v1/platform`；全平台展示数字 + 运营开关,**全部不鉴权**,登录前可读） |||
 | 39 | `GET /api/v1/platform/stats` | 无 | [详情](./platform/platform-stats.md) |
 | 40 | `GET /api/v1/platform/savings-feed` | 无 | [详情](./savings/platform-savings-feed.md) |
 | 40a | `GET /api/v1/platform/flags` | 无 | [详情](./platform/platform-flags.md)（客户端运营 feature flag,比价/领券期广告开关等,拉取后缓存） |
 | 40b | `GET /api/v1/platform/ad-config` | 无 | [详情](./platform/platform-ad-config.md)（客户端拉广告配置:穿山甲 app_id+各位ID+各场景开关;不含验签密钥） |
 | 40c | `GET /api/v1/platform/app-version` | 无 | [详情](./platform/platform-app-version.md)（最新 App 版本,OTA 检查更新;与本机 versionCode 比） |
-| 40d | `GET /api/v1/platform/huawei-review` | 无 | [详情](./platform/platform-huawei-review.md)（华为审核开关:快速设置权限步能否被用户关闭;仅华为 ROM 客户端拉） |
 | **微信支付回调**（前缀 `/api/v1/wxpay`） |||
 | W1 | `POST /api/v1/wxpay/transfer-auth-notify` | 无 | 免确认收款授权结果通知（一期 stub:仅应答 200 不验签不改账,授权状态靠主动查询兜底)(无单独文档) |
 | **CPS 群发短链落地**（**无前缀**,挂域名根;公网不鉴权） |||
@@ -172,7 +154,6 @@
 | A12 | `GET /admin/api/ad-revenue-report` | admin | [详情](./admin/ad/admin-ad-revenue-report.md)(广告收益报表:分页/场景/`app_env` 筛 + **DAU/ARPU** #120;真实收益侧接穿山甲日表 #92) |
 | A13 | `GET / PATCH /admin/api/ad-config` | operator/finance | 广告配置(穿山甲 ID/验签密钥/各场景开关;C 端只读版见 40b)(无单独文档,见 `app/admin/routers/ad_config.py`) |
 | A14 | `GET /admin/api/config`、`PATCH /config/{key}` | operator/finance | 运营可配置项([app_config](../database/app_config.md):奖励常量/提现地板价等;#117 修系统配置下发)(无单独文档,见 `app/admin/routers/config.py`) |
-| A16 | `GET / PATCH /admin/api/huawei-review` | operator/tech | 华为审核开关(快速设置权限步能否被用户关闭,落 `app_config.huawei_review`;C 端只读版见 40d)(无单独文档,见 `app/admin/routers/huawei_review.py`) |
 | **A·管理员与角色**(super_admin):`GET`/`POST` `/admins`、`PATCH`/`DELETE` `/admins/{id}`(#126 删除+`pages_override`)、`GET`/`POST` `/roles`、`GET /roles/catalog`、`PATCH`/`DELETE` `/roles/{id}`(#117/#126 自定义角色) ||| [列表](./admin/admins/admin-admins-list.md) / [建](./admin/admins/admin-admin-create.md) / [改+删](./admin/admins/admin-admin-update.md) / [角色](./admin/admin-roles.md) |
 | A15 | `GET /admin/api/audit-logs` | admin | [详情](./admin/admin-audit-logs.md) |
 | **A·CPS 运营台**:群/活动 CRUD、`POST /referral-links`、`POST /orders/reconcile`(美团+京东 #90)、`GET /orders`、`/stats`、群 `timeseries`/`daily`/`wx-users`/`day-users`(#79) ||| [详情](./admin/admin-cps.md) |

docs/api/other/applog-batch.md

@@ -0,0 +1,87 @@
+# POST /api/v1/applog/batch — 批量上报客户端运行日志
+
+> 所属:客户端日志组（前缀 `/api/v1/applog`） | 鉴权:无（不强制登录，`user_id` 可选带上） | [← 返回 API 索引](../README.md)
+
+批量接收客户端 App 运行日志（自动化步骤 / 网络 / 崩溃 / 调试等），**逐条**封装成单行 JSON 写入独立滚动文件 `logs/app-client.log`（与服务日志 `app-server.log` 分开），由阿里云 Logtail 采进**独立 SLS logstore**，**不落库**。服务端补 `client_ip`（X-Forwarded-For）与 `time`（接收时间，SLS 主时间）。`trace_id` 提到输出行顶层，便于在 SLS 里跨「客户端 / 服务端」两个 logstore 按 trace 拼出端到端链路。**fire-and-forget**:写失败也返回 2xx，不 500。
+
+## 入参
+
+批级公共字段发一次;`logs` 里每条只带日志本身。**每条只有 `client_ts/level/trace_id/tag/msg` 会提到输出顶层，其余自定义字段一律并入输出的 `data`**（防 SLS 索引列爆炸）。
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---|---|
+| `device_id` | string | ✅（≤64） | 设备 ID（限流 / 分组键） |
+| `user_id` | int \| null | ❌ | 登录用户 ID（未登录可空） |
+| `app_ver` | string \| null | ❌（≤32） | App 版本 |
+| `platform` | string \| null | ❌（≤16） | 平台（`android` / `ios` / `harmony`） |
+| `sent_at` | int \| null | ❌ | 批次发送时间（epoch ms） |
+| `logs` | list[object] | ✅（1-500 条） | 日志数组（每条为对象，内部字段**不强类型**） |
+| `logs[].client_ts` | int | ❌ | 端侧日志时间（epoch ms） |
+| `logs[].level` | string | ❌ | 级别（服务端归一化大写，截断 ≤16） |
+| `logs[].trace_id` | string \| null | ❌ | 关联服务端比价链路的 trace（有服务端交互的日志带上，截断 ≤256） |
+| `logs[].tag` | string \| null | ❌ | 模块 / 分类（截断 ≤128） |
+| `logs[].msg` | string | ❌ | 消息主体（超 8KB **字节**截断，加 `msg_truncated` 标记） |
+| `logs[].*` | any | ❌ | 其余任意自定义字段 → 一律并入输出的 `data`（客户端自带的 `data` 对象会被合并进来） |
+
+Mock 入参:
+```json
+{
+  "device_id": "android_abc123def456",
+  "user_id": 42,
+  "app_ver": "0.1.5",
+  "platform": "android",
+  "sent_at": 1719993700000,
+  "logs": [
+    {
+      "client_ts": 1719993600000,
+      "level": "info",
+      "tag": "automation",
+      "msg": "compare flow start",
+      "trace_id": "t_ab12cd34"
+    },
+    {
+      "client_ts": 1719993615000,
+      "level": "error",
+      "tag": "network",
+      "msg": "timeout calling /price/step",
+      "trace_id": "t_ab12cd34",
+      "http_status": 504,
+      "retry": 2
+    }
+  ]
+}
+```
+> 上例第二条的 `http_status` / `retry` 不在白名单 → 会被并入落盘行的 `data`:`{"http_status":504,"retry":2}`。
+
+## 出参
+
+响应 `200`:`AppLogIngestOut`
+
+| 字段 | 类型 | 说明 |
+|---|---|---|
+| `ok` | bool | 固定 `true` |
+| `received` | int | 成功写入文件的条数 |
+| `dropped` | int | 跳过的条数（服务端处理 / 序列化失败;正常为 `0`） |
+
+Mock 出参:
+```json
+{
+  "ok": true,
+  "received": 2,
+  "dropped": 0
+}
+```
+
+## 错误码
+- `413` 请求体超过上限（默认 2MB;服务端查 `Content-Length`，在 body 校验前拦截。缺该头时由 nginx `client_max_body_size` 兜底）
+- `422` `logs` 为空或超过 500 条 / `device_id` 缺失 / 字段类型不符
+- `429` 触发限流（同 IP 每分钟 > 120 次）
+
+## 说明
+- **落盘 → SLS**:逐条写独立文件 `logs/app-client.log`（单行 JSON，`propagate=False` 不污染 `app-server.log`），由 Logtail JSON 模式采进**独立 logstore**;不进数据库。落盘行除白名单字段外，服务端另补 `time`（接收时间）、`source="client"`、`service`、`client_ip` 及批级 `device_id/user_id/app_ver/platform/sent_at`。
+- **`trace_id` 跨层检索**:字段名与服务端日志一致。客户端应给**有服务端交互**的日志带上当初 API 返回的 `trace_id`（如 `/api/v1/price/step` 等透传族返回的 trace），即可在 SLS 里 `trace_id: "xxx"` 一查拼出「客户端视角 + 服务端比价链路」;纯客户端日志不带即可。
+- **白名单 + `data` 兜底**:只有 `client_ts/level/trace_id/tag/msg` 上顶层，其余键并入 `data`，把 SLS 索引列钉死在固定集合，防客户端任意 key 撑爆索引。索引字段另有长度上限（`level`≤16 / `trace_id`≤256 / `tag`≤128 / `msg`≤8KB）。
+- **fire-and-forget**:写文件失败也返回 2xx（避免客户端重试风暴）;网络重试可能在 SLS 造成重复条目，可接受。
+- **量级建议**:客户端做等级过滤 / 采样，攒到一定量再批量上报;单批 ≤500 条、body ≤2MB。
+- `client_ts` 是端侧时间（客户端时钟不可信），`time` 由服务端补（可靠时间轴）。
+- `user_id` 不靠 JWT:未登录态也要采日志;带上便于按用户排查。

docs/superpowers/plans/2026-07-19-client-applog-ingest.md

@@ -0,0 +1,561 @@
+# 客户端运行日志批量上报 Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 新增 `POST /api/v1/applog/batch`，把客户端批量上报的运行日志逐条写进独立滚动文件 `logs/app-client.log`，供 Aliyun Logtail 采进独立 SLS logstore。
+
+**Architecture:** 复用两个现成范式——analytics 的「批量入口 + 服务端补 IP」+ `app-server.log` 的「单行 JSON + RotatingFileHandler + Logtail」。核心逻辑在写入层 `app/core/client_log.py`：专用 `client` logger（`propagate=False`，不污染服务日志），每条按「白名单键提顶层 + 其余并入 `data`」封装成一行 JSON，`trace_id` 提到顶层以便跨层检索。端点瘦、fire-and-forget（写失败不 500）。
+
+**Tech Stack:** FastAPI、Pydantic v2、Python `logging.handlers.RotatingFileHandler`、pytest + `TestClient`。
+
+**Spec:** [docs/superpowers/specs/2026-07-19-client-applog-ingest-design.md](../specs/2026-07-19-client-applog-ingest-design.md)
+
+---
+
+## File Structure
+
+| 文件 | 职责 | 动作 |
+|---|---|---|
+| `app/core/client_log.py` | 专用 logger/handler、滚动配置、白名单+`data` 兜底与信封拼装、`write_records()` | Create |
+| `app/schemas/applog.py` | `AppLogBatchIn`（`logs: list[dict]`）/ `AppLogIngestOut` | Create |
+| `app/api/v1/applog.py` | 瘦路由：body 上限依赖 → 调 writer → 返回计数 | Create |
+| `app/main.py` | 注册 `applog_router` | Modify |
+| `tests/test_applog.py` | writer 单测 + 端点集成测试 | Create |
+
+前置约束（写进代码注释）：`RotatingFileHandler` 多进程并发 `doRollover()` 会损坏/丢日志；本方案依赖生产 `--workers 1`（与限流器/SMS 码/SQLite 同一既有假设）。扩 worker 前必须换 `QueueHandler`→单写入者 / 外部 logrotate / 写 stdout 交 journald。
+
+---
+
+## Task 1: 专用落盘 writer `app/core/client_log.py`
+
+**Files:**
+- Create: `app/core/client_log.py`
+- Test: `tests/test_applog.py`
+
+- [ ] **Step 1: 写失败测试（writer 层）**
+
+创建 `tests/test_applog.py`：
+
+```python
+"""客户端运行日志上报:writer 单测 + 端点集成测试。"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from app.core import client_log
+
+
+@pytest.fixture()
+def client_log_file(tmp_path, monkeypatch):
+    """把客户端日志切到临时文件,并重置 writer 单例使其按当时 env 重建。"""
+    p = tmp_path / "app-client.log"
+    monkeypatch.setenv("CLIENT_LOG_FILE", str(p))
+    client_log.reset_client_logger()
+    yield p
+    client_log.reset_client_logger()
+
+
+def _read_lines(p: Path) -> list[dict]:
+    text = p.read_text(encoding="utf-8").strip()
+    return [json.loads(ln) for ln in text.splitlines() if ln]
+
+
+# ---------------- writer 层 ----------------
+
+def _meta(**kw) -> dict:
+    base = {"device_id": "d-1", "user_id": None, "app_ver": None,
+            "platform": None, "sent_at": None}
+    base.update(kw)
+    return base
+
+
+def test_writer_writes_one_line_per_record(client_log_file):
+    recs = [
+        {"client_ts": 1737000000000, "level": "info", "msg": "hello"},
+        {"client_ts": 1737000000001, "level": "error", "msg": "boom", "tag": "net"},
+    ]
+    received, dropped = client_log.write_records(
+        recs, meta=_meta(user_id=42, app_ver="1.2.3", platform="android"),
+        client_ip="1.2.3.4",
+    )
+    assert (received, dropped) == (2, 0)
+    lines = _read_lines(client_log_file)
+    assert len(lines) == 2
+    assert lines[0]["source"] == "client"
+    assert lines[0]["service"] == "app-client"
+    assert lines[0]["device_id"] == "d-1"
+    assert lines[0]["user_id"] == 42
+    assert lines[0]["app_ver"] == "1.2.3"
+    assert lines[0]["client_ip"] == "1.2.3.4"
+    assert lines[0]["level"] == "INFO"  # 归一化大写
+    assert lines[0]["msg"] == "hello"
+    assert lines[0]["client_ts"] == 1737000000000
+    assert lines[1]["tag"] == "net"
+
+
+def test_writer_hoists_trace_id_to_top_level(client_log_file):
+    client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x", "trace_id": "abc123"}],
+        meta=_meta(), client_ip="",
+    )
+    assert _read_lines(client_log_file)[0]["trace_id"] == "abc123"
+
+
+def test_writer_sweeps_unknown_keys_into_data(client_log_file):
+    client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x",
+          "foo": 123, "data": {"bar": "baz"}}],
+        meta=_meta(), client_ip="",
+    )
+    line = _read_lines(client_log_file)[0]
+    assert "foo" not in line               # 白名单外不进顶层
+    assert line["data"]["foo"] == 123      # 兜底进 data
+    assert line["data"]["bar"] == "baz"    # 客户端自带 data 合并进来
+
+
+def test_writer_truncates_oversize_msg(client_log_file, monkeypatch):
+    monkeypatch.setenv("APPLOG_MAX_MSG_BYTES", "10")
+    received, dropped = client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x" * 100}],
+        meta=_meta(), client_ip="",
+    )
+    assert (received, dropped) == (1, 0)   # 截断而非丢弃
+    line = _read_lines(client_log_file)[0]
+    assert line["msg_truncated"] is True
+    assert line["msg"].endswith("…[truncated]")
+
+
+def test_writer_logger_does_not_propagate(client_log_file):
+    lg = client_log.get_logger()
+    assert lg.name == "client"
+    assert lg.propagate is False           # 不冒泡到 root → 不写 app-server.log
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_applog.py -q`
+Expected: FAIL —— `AttributeError: module 'app.core.client_log' has no attribute 'reset_client_logger'`（模块尚不存在）。
+
+- [ ] **Step 3: 实现 `app/core/client_log.py`**
+
+```python
+"""客户端运行日志专用落盘 writer(独立于服务端 app-server.log)。
+
+- 独占 logger "client" + 自己的 RotatingFileHandler,propagate=False → 不污染 app-server.log。
+- 每条按「白名单键(client_ts/level/trace_id/tag/msg)提顶层 + 其余并入 data」封装,再
+  json.dumps 成一行写出(钉死 SLS 索引列;见 spec §5)。formatter 用 %(message)s——行本身
+  已是 JSON,不能再过 JsonFormatter 二次编码。
+- 滚动 20MB×10(env 可调),与服务日志同机制。
+  ⚠️ 依赖 --workers 1:RotatingFileHandler 多进程并发 doRollover 会损坏/丢日志;扩 worker
+  前换 QueueHandler→单写入者 / 外部 logrotate(copytruncate) / 写 stdout 交 journald。
+
+服务端补的字段(time/source/service/client_ip/device_id/...)是「事实」,与客户端自述分开。
+`time` 用服务端接收时间作 SLS 主时间(客户端时钟不可信),client_ts 另存为可查字段。
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+from datetime import datetime
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+# 仅这些客户端键提到输出行顶层;其余(含客户端自带 data)一律并入 data,防 SLS 索引列爆炸
+_TOP_LEVEL_KEYS = ("client_ts", "level", "trace_id", "tag", "msg")
+
+_logger: logging.Logger | None = None
+
+
+def _max_msg_bytes() -> int:
+    return int(os.getenv("APPLOG_MAX_MSG_BYTES", "8192"))
+
+
+def _build_logger() -> logging.Logger:
+    lg = logging.getLogger("client")
+    lg.setLevel(logging.INFO)
+    lg.propagate = False  # 不冒泡到 root → 不写进 app-server.log
+    log_file = os.getenv("CLIENT_LOG_FILE") or str(
+        Path(os.getenv("LOG_DIR", "logs")) / "app-client.log"
+    )
+    Path(log_file).parent.mkdir(parents=True, exist_ok=True)
+    handler = RotatingFileHandler(
+        log_file,
+        maxBytes=int(os.getenv("CLIENT_LOG_MAX_BYTES", str(20 * 1024 * 1024))),
+        backupCount=int(os.getenv("CLIENT_LOG_BACKUP_COUNT", "10")),
+        encoding="utf-8",
+    )
+    handler.setFormatter(logging.Formatter("%(message)s"))  # 行已是 JSON,不再包装
+    lg.handlers = [handler]
+    return lg
+
+
+def get_logger() -> logging.Logger:
+    global _logger
+    if _logger is None:
+        _logger = _build_logger()
+    return _logger
+
+
+def reset_client_logger() -> None:
+    """测试用:关闭并丢弃当前 logger,使下次 get_logger 按当时 env 重建(切临时文件)。"""
+    global _logger
+    if _logger is not None:
+        for h in list(_logger.handlers):
+            h.close()
+        _logger.handlers = []
+    _logger = None
+
+
+def _truncate_msg(msg: str) -> tuple[str, bool]:
+    raw = msg.encode("utf-8")
+    limit = _max_msg_bytes()
+    if len(raw) <= limit:
+        return msg, False
+    # 按字节截断后解码,忽略截断处半个多字节字符
+    return raw[:limit].decode("utf-8", "ignore") + "…[truncated]", True
+
+
+def _build_line(
+    record: dict, *, meta: dict, client_ip: str, service: str, now_iso: str
+) -> str:
+    out: dict = {
+        "time": now_iso,
+        "source": "client",
+        "service": service,
+        "client_ip": client_ip,
+    }
+    # 批级公共字段(非空才带)
+    for k in ("device_id", "user_id", "app_ver", "platform", "sent_at"):
+        v = meta.get(k)
+        if v is not None:
+            out[k] = v
+    # 白名单键提顶层
+    if record.get("level") is not None:
+        out["level"] = str(record["level"]).upper()
+    if record.get("trace_id"):
+        out["trace_id"] = record["trace_id"]
+    if record.get("tag"):
+        out["tag"] = record["tag"]
+    if record.get("client_ts") is not None:
+        out["client_ts"] = record["client_ts"]
+    if record.get("msg") is not None:
+        msg, truncated = _truncate_msg(str(record["msg"]))
+        out["msg"] = msg
+        if truncated:
+            out["msg_truncated"] = True
+    # 其余键(含客户端自带 data)并入 data
+    data: dict = {}
+    client_data = record.get("data")
+    if isinstance(client_data, dict):
+        data.update(client_data)
+    for k, v in record.items():
+        if k in _TOP_LEVEL_KEYS or k == "data":
+            continue
+        data[k] = v
+    if data:
+        out["data"] = data
+    return json.dumps(out, ensure_ascii=False, default=str)
+
+
+def write_records(
+    records: list[dict], *, meta: dict, client_ip: str
+) -> tuple[int, int]:
+    """把一批客户端日志逐条写入专用文件。返回 (received, dropped)。
+
+    尽力而为(fire-and-forget):logger 初始化或单条写入失败只跳过并计 dropped,
+    不抛给上层——端点因此永不因写日志而 500。
+    """
+    try:
+        lg = get_logger()
+    except Exception:  # noqa: BLE001 — 初始化失败也不能让端点 500
+        logging.getLogger("shagua.applog").exception("client log writer init failed")
+        return 0, len(records)
+    service = os.getenv("CLIENT_LOG_SERVICE_NAME", "app-client")
+    now_iso = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
+    received = dropped = 0
+    for rec in records:
+        try:
+            line = _build_line(
+                rec, meta=meta, client_ip=client_ip, service=service, now_iso=now_iso
+            )
+            lg.info(line)
+            received += 1
+        except Exception:  # noqa: BLE001 — 坏条跳过,不影响其余
+            dropped += 1
+    return received, dropped
+```
+
+- [ ] **Step 4: 跑测试确认通过**
+
+Run: `pytest tests/test_applog.py -q`
+Expected: PASS（5 个 writer 测试全绿）。
+
+- [ ] **Step 5: ruff**
+
+Run: `ruff check app/core/client_log.py tests/test_applog.py`
+Expected: 无错误（如有 import 排序等自动可修问题：`ruff check --fix`）。
+
+- [ ] **Step 6: 提交**
+
+```bash
+git add app/core/client_log.py tests/test_applog.py
+git commit -m "feat(applog): 客户端日志专用落盘 writer(白名单+data 兜底, propagate=False)"
+```
+
+---
+
+## Task 2: 端点 + schema + 注册
+
+**Files:**
+- Create: `app/schemas/applog.py`
+- Create: `app/api/v1/applog.py`
+- Modify: `app/main.py`（import + `include_router`）
+- Test: `tests/test_applog.py`（追加端点用例）
+
+- [ ] **Step 1: 追加失败测试（端点层）**
+
+在 `tests/test_applog.py` 末尾追加：
+
+```python
+# ---------------- 端点层 ----------------
+
+def _post(client, body):
+    return client.post("/api/v1/applog/batch", json=body)
+
+
+def test_endpoint_happy_path(client, client_log_file):
+    body = {
+        "device_id": "d-1", "user_id": 42, "app_ver": "1.2.3", "platform": "android",
+        "logs": [
+            {"client_ts": 1, "level": "info", "msg": "a", "trace_id": "t1"},
+            {"client_ts": 2, "level": "warn", "msg": "b"},
+        ],
+    }
+    resp = _post(client, body)
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "received": 2, "dropped": 0}
+    lines = _read_lines(client_log_file)
+    assert len(lines) == 2
+    assert lines[0]["trace_id"] == "t1"       # 端到端:trace_id 落到文件顶层
+    assert lines[0]["client_ip"]              # 服务端补了 IP
+
+
+def test_endpoint_rejects_batch_over_max(client, client_log_file):
+    body = {"device_id": "d-1",
+            "logs": [{"client_ts": i, "level": "info", "msg": str(i)} for i in range(501)]}
+    resp = _post(client, body)
+    assert resp.status_code == 422           # Pydantic max_length=500
+
+
+def test_endpoint_rejects_body_over_limit(client, client_log_file, monkeypatch):
+    monkeypatch.setenv("APPLOG_MAX_BODY_BYTES", "50")
+    body = {"device_id": "d-1", "logs": [{"client_ts": 1, "level": "info", "msg": "x" * 500}]}
+    resp = _post(client, body)
+    assert resp.status_code == 413           # 依赖查 Content-Length,body 校验前拦截
+
+
+def test_endpoint_returns_200_on_write_failure(client, client_log_file, monkeypatch):
+    class _BoomLogger:
+        name = "client"
+        propagate = False
+        handlers: list = []
+
+        def info(self, *a, **k):
+            raise RuntimeError("disk full")
+
+    monkeypatch.setattr(client_log, "get_logger", lambda: _BoomLogger())
+    body = {"device_id": "d-1", "logs": [
+        {"client_ts": 1, "level": "info", "msg": "a"},
+        {"client_ts": 2, "level": "info", "msg": "b"}]}
+    resp = _post(client, body)
+    assert resp.status_code == 200           # fire-and-forget:写失败不 500
+    assert resp.json() == {"ok": True, "received": 0, "dropped": 2}
+
+
+def test_endpoint_missing_device_id_is_422(client, client_log_file):
+    resp = _post(client, {"logs": [{"client_ts": 1, "level": "info", "msg": "a"}]})
+    assert resp.status_code == 422           # device_id 必填
+```
+
+- [ ] **Step 2: 跑测试确认失败**
+
+Run: `pytest tests/test_applog.py -q`
+Expected: FAIL —— 端点未注册，`POST /api/v1/applog/batch` 返回 404（happy-path 断言 200 失败）。
+
+- [ ] **Step 3: 实现 schema `app/schemas/applog.py`**
+
+```python
+"""客户端运行日志批量上报 schema。
+
+批级公共字段(device_id/user_id/app_ver/platform/sent_at)发一次;logs 为原始 dict 列表,
+**不强类型**——尽力而为的日志链路,单条内容异常不该让整批 422。每条的「白名单键 + data
+兜底」拆分在写入层 [app.core.client_log] 做(见 spec §4/§5)。
+"""
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+_MAX_BATCH = int(os.getenv("APPLOG_MAX_BATCH", "500"))
+
+
+class AppLogBatchIn(BaseModel):
+    device_id: str = Field(max_length=64)
+    user_id: int | None = None
+    app_ver: str | None = Field(default=None, max_length=32)
+    platform: str | None = Field(default=None, max_length=16)
+    sent_at: int | None = None
+    logs: list[dict[str, Any]] = Field(min_length=1, max_length=_MAX_BATCH)
+
+
+class AppLogIngestOut(BaseModel):
+    ok: bool = True
+    received: int
+    dropped: int = 0
+```
+
+- [ ] **Step 4: 实现端点 `app/api/v1/applog.py`**
+
+```python
+"""客户端运行日志批量上报接口。
+
+POST /api/v1/applog/batch — 批量接收客户端运行日志,逐条写专用滚动文件 logs/app-client.log
+(供 Logtail 采进独立 SLS logstore)。鉴权同 analytics(不强制登录,user_id 可选在 body)。
+fire-and-forget:写失败也不 500(避免客户端重试风暴);超批 422、超体积 413、msg 超限截断。
+"""
+from __future__ import annotations
+
+import os
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+
+from app.core.client_log import write_records
+from app.core.ratelimit import rate_limit
+from app.schemas.applog import AppLogBatchIn, AppLogIngestOut
+
+router = APIRouter(prefix="/api/v1/applog", tags=["applog"])
+
+
+def _client_ip(request: Request) -> str:
+    """取客户端 IP:生产经 nginx 反代优先 X-Forwarded-For 首段,否则直连 IP(同 analytics)。"""
+    xff = request.headers.get("x-forwarded-for")
+    if xff:
+        return xff.split(",")[0].strip()
+    return request.client.host if request.client else ""
+
+
+def _enforce_body_limit(request: Request) -> None:
+    """依赖:body 声明过大直接 413(在 body 校验前拦截)。缺 Content-Length 由 nginx 兜底。"""
+    max_bytes = int(os.getenv("APPLOG_MAX_BODY_BYTES", str(1024 * 1024 * 2)))
+    cl = request.headers.get("content-length")
+    if cl is not None and cl.isdigit() and int(cl) > max_bytes:
+        raise HTTPException(status_code=413, detail="日志批量过大")
+
+
+@router.post(
+    "/batch",
+    response_model=AppLogIngestOut,
+    summary="批量上报客户端运行日志",
+    dependencies=[
+        Depends(rate_limit(120, 60, "applog-batch")),
+        Depends(_enforce_body_limit),
+    ],
+)
+def ingest_logs(batch: AppLogBatchIn, request: Request) -> AppLogIngestOut:
+    received, dropped = write_records(
+        batch.logs,
+        meta={
+            "device_id": batch.device_id,
+            "user_id": batch.user_id,
+            "app_ver": batch.app_ver,
+            "platform": batch.platform,
+            "sent_at": batch.sent_at,
+        },
+        client_ip=_client_ip(request),
+    )
+    return AppLogIngestOut(received=received, dropped=dropped)
+```
+
+- [ ] **Step 5: 注册路由 `app/main.py`**
+
+在 import 区（analytics_router 之后，约 [app/main.py:23](../../../app/main.py#L23)）加：
+
+```python
+from app.api.v1.applog import router as applog_router
+```
+
+在 `include_router` 区（`app.include_router(analytics_router)` 之后，约 [app/main.py:125](../../../app/main.py#L125)）加：
+
+```python
+app.include_router(applog_router)
+```
+
+- [ ] **Step 6: 跑测试确认通过**
+
+Run: `pytest tests/test_applog.py -q`
+Expected: PASS（writer 5 + 端点 5，共 10 个）。
+
+- [ ] **Step 7: 全量测试 + ruff**
+
+Run: `pytest -q && ruff check app/api/v1/applog.py app/schemas/applog.py app/main.py`
+Expected: 全绿、无 lint 错误。
+
+- [ ] **Step 8: 提交**
+
+```bash
+git add app/api/v1/applog.py app/schemas/applog.py app/main.py tests/test_applog.py
+git commit -m "feat(applog): POST /api/v1/applog/batch 批量上报端点(限流+体积上限+fire-and-forget)"
+```
+
+---
+
+## Task 3: `.env.example` 文档化新 env（可选但推荐）
+
+**Files:**
+- Modify: `.env.example`
+
+- [ ] **Step 1: 追加 env 说明**
+
+在 `.env.example` 末尾（或日志相关区块）追加，让运维知道可调项：
+
+```bash
+# 客户端运行日志上报(POST /api/v1/applog/batch)→ 落 logs/app-client.log 供 Logtail 采集
+# 独立于服务日志 app-server.log;滚动机制同服务日志。默认值见 app/core/client_log.py。
+# CLIENT_LOG_FILE=logs/app-client.log
+# CLIENT_LOG_MAX_BYTES=20971520      # 单文件 20MB 滚动
+# CLIENT_LOG_BACKUP_COUNT=10         # 保留 10 个 → ~200MB 缓冲
+# CLIENT_LOG_SERVICE_NAME=app-client # 输出行 service 字段
+# APPLOG_MAX_BATCH=500               # 单批最大条数(超 → 422)
+# APPLOG_MAX_BODY_BYTES=2097152      # 请求体上限 2MB(超 → 413)
+# APPLOG_MAX_MSG_BYTES=8192          # 单条 msg 超限截断
+```
+
+- [ ] **Step 2: 提交**
+
+```bash
+git add .env.example
+git commit -m "docs(applog): .env.example 补充客户端日志上报可调 env"
+```
+
+> 若仓库无 `.env.example`（以 `git ls-files .env.example` 确认），跳过本任务。
+
+---
+
+## Definition of Done
+
+- [ ] `pytest -q` 全绿（含新增 `tests/test_applog.py` 10 用例）。
+- [ ] `ruff check .` 无错误。
+- [ ] `POST /api/v1/applog/batch` 手动冒烟：发一批含 `trace_id` 的日志，确认 `logs/app-client.log` 出现对应单行 JSON、`trace_id` 在顶层、未知键落在 `data`，且 `logs/app-server.log` **未**被写入客户端记录。
+- [ ] Spec 的运维项（独立 logstore、Logtail JSON 模式采 `app-client.log`、给 `trace_id` 建索引、nginx `client_max_body_size` 对齐）已同步给运维（不在本仓代码内，见 spec §8）。
+
+## Self-Review 结论（作者已核对）
+
+- **Spec 覆盖**：§4 端点/限额 → Task 2；§5 白名单+data 兜底 → Task 1 `_build_line`；§6 落盘/滚动/单 worker 注释 → Task 1；§7 trace_id 顶层 → Task 1 + 端点测试；§9 fire-and-forget → Task 1 `write_records` + 端点测试；§13 env → Task 1/2 读取 + Task 3 文档化。§8 为纯运维配置，列入 Definition of Done。
+- **无占位符**：所有步骤含完整代码/命令/预期。
+- **类型/命名一致**：`write_records(records, *, meta, client_ip) -> (received, dropped)`、`get_logger()`、`reset_client_logger()`、`_TOP_LEVEL_KEYS`、`AppLogBatchIn/AppLogIngestOut` 在 Task 1/2 间一致引用。

docs/superpowers/specs/2026-07-19-client-applog-ingest-design.md

@@ -0,0 +1,200 @@
+# 客户端运行日志批量上报 → 落专用文件 → SLS 采集 设计
+
+- 日期：2026-07-19
+- 状态：设计已评审，待写实现计划
+- 相关代码：[app/core/logging.py](../../../app/core/logging.py)（服务端日志落盘范式）、[app/api/v1/analytics.py](../../../app/api/v1/analytics.py)（批量上报入口范式）、[deploy/shaguabijia-app-server.service](../../../deploy/shaguabijia-app-server.service)（`--workers 1`）
+
+## 1. 背景与目标
+
+Android 客户端会在本地攒一批 App 运行日志（自动化步骤 / 网络 / 崩溃 / 调试等），批量以 JSON 数组发给后端。后端把这些日志**逐条**写进一个**专用滚动日志文件**，文件滚动规则与服务日志一致；再由阿里云 SLS 的 Logtail 采集该文件、进独立 logstore。
+
+目标：
+
+1. 新增 `POST /api/v1/applog/batch` 接收批量客户端日志。
+2. 逐条落进 `logs/app-client.log`（**独立于** `app-server.log`），单行 JSON、大小滚动、供 Logtail JSON 模式零正则采集。
+3. 客户端日志进**独立 logstore**，但 **可按 `trace_id` 检索**，以便和服务端比价链路跨层对齐。
+4. 入口安全：鉴权同 analytics（不强制 Bearer），加体积/条数上限、限流，防撑盘与日志注入。
+
+本方案是两个**已验证范式的组合**：analytics 的「批量入口 + 服务端补 IP/接收时间」+ `app-server.log` 的「单行 JSON + RotatingFileHandler + Logtail」。不是新架构。
+
+## 2. 非目标（本期不做，YAGNI）
+
+- **异步队列 / 后台 writer**：`<10 万条/天` ≈ 均值 1.2 条/秒，10× 突发 ~12/秒，请求路径同步写文件足矣。
+- **去重**：网络重试会在 SLS 产生重复条目。本期**接受重复**并在文档写明；后续如需，加客户端 `batch_id` 短窗口去重。
+- **服务端脱敏**：本期靠**客户端侧**控制等级/采样与不打 PII；服务端只做体积截断。脱敏留作后续 knob。
+- **多 worker 支持**：见 §5 的单 worker 约束。
+- **入库 / admin 查询界面**：日志的归宿是 SLS，不落 DB。
+
+## 3. 方案总览与备选取舍
+
+**采用 A**：客户端 → 后端接口 → 专用滚动文件 → Logtail → SLS。
+
+| 方案 | 说明 | 为何不选 |
+|---|---|---|
+| **A（选定）** | 后端中转落文件，Logtail 采集 | — 代码最少、完全复用现成文件→Logtail 管线；AK/SK 不进 APK；Logtail 天然提供落盘缓冲+断点续传 |
+| B | 客户端直连 SLS（Producer SDK / Web-Tracking） | 要么把凭证埋进 APK，要么额外跑 STS 换 token 服务；服务端难做鉴权/富化/脱敏 |
+| C | 后端调 SLS PutLogs API（不落文件） | 请求路径硬依赖 SLS 可用性，需自建缓冲/背压/重试——等于重造 Logtail |
+
+## 4. 端点契约
+
+`POST /api/v1/applog/batch`
+
+- 新路由 `app/api/v1/applog.py`，在 [app/main.py](../../../app/main.py) `import ... as applog_router` 并 `app.include_router(applog_router)`（紧挨 analytics）。
+- 鉴权同 analytics：**不强制 Bearer**，`user_id` 可选放 body；服务端补 `client_ip`（复用 analytics 里的 `_client_ip` 取 `X-Forwarded-For` 首段逻辑）与接收时间。
+
+### 请求体 `AppLogBatchIn`（批级公共字段发一次，省带宽）
+
+| 字段 | 类型 | 必填 | 约束/说明 |
+|---|---|---|---|
+| `device_id` | str | 是 | `max_length=64`；限流/分组键 |
+| `user_id` | int? | 否 | 可选，未登录态也采集 |
+| `app_ver` | str? | 否 | `max_length=32` |
+| `platform` | str? | 否 | `max_length=16`（android/ios/harmony） |
+| `sent_at` | int? | 否 | 本批上报时刻 epoch ms |
+| `logs` | list[dict] | 是 | `min_length=1, max_length=500`；**每条为对象**，逐条按 §5 契约处理 |
+
+`logs` 用 `list[dict[str, Any]]` 而非强类型列表：这是**尽力而为**的日志链路，单条内容异常不应让整批 422 失败。超过 500 条由 Pydantic `max_length` 触发 422（客户端应更小批）。
+
+### 响应 `AppLogIngestOut`
+
+```jsonc
+{ "ok": true, "received": 128, "dropped": 2 }
+```
+- `received`：成功写入文件的条数。
+- `dropped`：服务端处理/序列化失败被跳过的条数（正常为 0，属异常兜底计数）。**oversize 的 `msg` 是截断而非丢弃**，不计入 dropped。
+
+### 限额（防撑盘 / 注入 / DoS）
+
+| 限额 | 默认 | 超限行为 | env |
+|---|---|---|---|
+| 单批条数 | 500 | 422（Pydantic） | `APPLOG_MAX_BATCH` |
+| body 字节 | 2 MB | 413（依赖查 `Content-Length`，在 body 校验前拦截；缺该头由 nginx `client_max_body_size` 兜底） | `APPLOG_MAX_BODY_BYTES` |
+| 单条 `msg` 字节 | 8192 | 截断 + 标记，不丢 | `APPLOG_MAX_MSG_BYTES` |
+
+同时**对齐 nginx `client_max_body_size`**（见 [deploy/nginx](../../../deploy/nginx/)）避免反代先于应用截断。限流复用 `app/core/ratelimit.py`（项目现有 IP 固定窗口）；具体挂法参照现有已限流写端点，测试环境 `RATE_LIMIT_ENABLED=false` 关闭。
+
+## 5. 每条记录契约 + 白名单键 + `data` 兜底
+
+**客户端每条日志的识别键（仅这些提到输出行顶层）：**
+
+| 键 | 类型 | 说明 |
+|---|---|---|
+| `client_ts` | int | 端事件时间 epoch ms（与 analytics 命名一致） |
+| `level` | str | 服务端归一化为大写；不做硬枚举拒绝（fire-and-forget） |
+| `trace_id` | str? | **§6 的核心**：有服务端交互的日志带上当初 API 返回的 trace |
+| `tag` | str? | 模块/分类，便于 SLS 过滤 |
+| `msg` | str | 消息主体，超 `APPLOG_MAX_MSG_BYTES` 截断并加标记 |
+
+**其余任意自定义字段 → 服务端一律收进单个 `data` 对象。**
+
+**为什么这么设计（防 SLS 索引列爆炸）：** Logtail JSON 模式下每个**顶层 key** 都会成为 logstore 一个可索引字段/列。若放任客户端往顶层写任意 key（甚至把动态 id 拼进 key），logstore 会长出成千上万个不同顶层列 → 索引成本膨胀、可能撞字段数上限、schema 混乱到无法建稳定 dashboard/告警，一个客户端 bug 就能把 logstore 搞脏。
+
+**「兜底」= 服务端强制、不信任客户端。** `logs` 以 `list[dict]` 原样收下（不强类型，单条异常不该让整批 422），在**写入层 `client_log.py`** 逐条拆分：白名单键提顶层，**其余键（含客户端自带的 `data`）统一并进 `data`**（而非透传到顶层，也非静默丢弃——日志要保真）。于是**输出行顶层列恒定**，无论客户端怎么发。
+
+这与现有 `JsonFormatter` 平铺 `phase/step/command/cost_ms` 同思路，区别：那些 extra 是**服务端可信有限**的键；客户端不可信无界，故只给一个 `data` 沙盒。
+
+## 6. 落盘 writer
+
+新模块 `app/core/client_log.py`：
+
+- 惰性单例 `logging.getLogger("client")`，**`propagate=False`**（否则冒泡进 root 被 `app-server.log` 二次写入并污染）。
+- 独占一个 `RotatingFileHandler`，formatter 为 `%(message)s`——**不复用 root 的 `JsonFormatter`**（那会把已是 JSON 的行二次编码成字符串）。writer 自己拼信封 dict 后 `json.dumps(..., ensure_ascii=False, default=str)` 得到**一行**，`logger.info(line)` 写出。
+- 复用 `logging` 模块的 handler 锁保证多线程（uvicorn threadpool）并发写安全。
+- 幂等 setup（仿 `setup_logging` 的 `_CONFIGURED` 守卫）。
+
+**滚动规则（与服务日志同机制 `RotatingFileHandler`，尺寸给客户端量级）：**
+
+| 参数 | 默认 | env |
+|---|---|---|
+| 文件路径 | `logs/app-client.log` | `CLIENT_LOG_FILE` / `LOG_DIR` |
+| `maxBytes` | 20 MB | `CLIENT_LOG_MAX_BYTES` |
+| `backupCount` | 10 | `CLIENT_LOG_BACKUP_COUNT` |
+| `service` 字段 | `app-client` | `CLIENT_LOG_SERVICE_NAME` |
+
+≈ 200 MB / ~2 天缓冲，给 Logtail 断线留余量（按 <10 万条/天、条均值估算）。**滚动文件不 gzip**——Logtail 读不了压缩包会丢数据。
+
+> ⚠️ **单 worker 约束**：`RotatingFileHandler` 多进程并发 `doRollover()` 会损坏/丢日志。当前生产 `--workers 1`（与限流器/SMS 码/SQLite 写锁同一既有假设，见 [deploy/shaguabijia-app-server.service](../../../deploy/shaguabijia-app-server.service)）故安全。**代码注释显式标注**：扩 worker 前必须换 `QueueHandler`→单写入者 或外部 logrotate（copytruncate）或写 stdout 交 journald 采集。
+
+**每条输出行（写入 `app-client.log`）：**
+
+```jsonc
+{ "time": "2026-07-19T12:00:00.123",  // 服务端接收时间 = SLS 主时间(客户端时钟不可信)
+  "source": "client", "service": "app-client",
+  "level": "ERROR", "trace_id": "abc123", "tag": "automation", "msg": "...",
+  "device_id": "d-xxx", "user_id": 123, "app_ver": "1.2.3", "platform": "android",
+  "client_ip": "1.2.3.4", "client_ts": 1737000000123, "sent_at": 1737000005000,
+  "data": { /* 客户端其余任意字段 */ } }
+```
+- `time` 用服务端接收时间作 SLS 主时间；`client_ts` 保留为可查字段（时钟漂移不影响检索基准）。
+- 值为空的可选字段省略，保持行精简（仿服务端 formatter 省略空 `trace_id`）。
+
+## 7. trace_id 检索（满足目标 #3）
+
+关键：**per-record 把 `trace_id` 提到输出行顶层，字段名与服务端日志完全一致（`trace_id`）**。来源是客户端在**它发起过服务端调用的那些日志**里带上当初 API 返回的 trace（如 `/api/v1/compare/*` 由 [pricebot_router](../../../app/core/pricebot_router.py) 透传的 trace）。没有服务端交互的纯客户端日志不带 `trace_id`，正常。
+
+于是 SLS 里对客户端 logstore 与服务端 logstore 各查 `trace_id: "xxx"`，即可拼出「客户端自动化视角 + 服务端比价链路」的端到端故事。
+
+> **需客户端配合**：给有服务端 trace 的日志记录打上 `trace_id`（Android 侧改动，见 §10）。
+
+## 8. Logtail / SLS 侧配置（运维，非本仓代码）
+
+- 新建**独立 logstore**（独立保留期，客户端日志建议**比服务端短**以控成本）。
+- 新 Logtail 配置采集 `logs/app-client.log`，**JSON 模式**（每行一条 JSON，零正则，与 `app-server.log` 同套路）。
+- **把 `trace_id` 配成索引字段**（否则 #3 查不了）；`level` / `tag` / `device_id` 亦建议建索引。
+- 时间字段用输出行 `time`（服务端接收时间）。
+
+## 9. 失败语义 / 安全
+
+- **fire-and-forget**：写文件失败 → 服务端记一笔（`shagua.applog` logger）+ 仍返回 2xx，**绝不 500**（对比 selfstat 的 503 是 DB 关键链路，日志不是；500 会招致客户端重试风暴+重复上报）。
+- **日志行注入防护**：一律 `json.dumps` 重新序列化，内嵌 `\n` 被转义，客户端伪造不出假日志行；绝不把客户端原始字符串直接写文件。
+- **撑盘/DoS**：§4 的条数/体积/msg 上限 + IP 限流；`device_id` 便于后续拉黑滥用设备。
+
+## 10. 客户端契约（Android 侧需配合，属另一仓）
+
+1. 每条日志结构：顶层放 `client_ts / level / trace_id? / tag? / msg`，其余自定义字段放 `data`（否则会被服务端兜底挪进 `data`）。
+2. 有服务端交互的日志带上对应 `trace_id`。
+3. 等级/采样与 PII 控制在客户端侧做（省流量、免服务端脱敏）。
+4. 单批 ≤500 条、body ≤2MB；失败可重试（服务端接受重复）。
+
+## 11. 代码落点
+
+| 文件 | 职责 |
+|---|---|
+| `app/api/v1/applog.py` | 瘦路由：body 上限依赖 → 解析 `AppLogBatchIn` → 调 writer → 返回 `received/dropped`；`_client_ip` 复用 analytics 逻辑、IP 限流 |
+| `app/schemas/applog.py` | `AppLogBatchIn`（`logs: list[dict]` 不强类型）/ `AppLogIngestOut` |
+| `app/core/client_log.py` | 专用 logger/handler、滚动配置、白名单+`data` 兜底与信封拼装 `write_records(...)` |
+| `app/main.py` | 注册 `applog_router` |
+
+## 12. 测试计划（`tests/test_applog.py`）
+
+用临时目录做 `CLIENT_LOG_FILE`（测试前置 env + 重置 writer 单例）。
+
+1. **落盘逐行**：POST N 条 → 200，`received=N`，文件恰 N 行、每行合法 JSON、关键字段齐。
+2. **trace_id 顶层**：带 `trace_id` 的记录 → 输出行顶层出现 `trace_id`。
+3. **未知键兜底**：记录带 `foo` → 输出行顶层无 `foo`，`data.foo` 存在。
+4. **msg 截断**：`msg` > 8KB → 截断+标记，仍 `received`（不进 dropped）。
+5. **超批拒绝**：>500 条 → 422。
+6. **超体积拒绝**：`Content-Length` > 2MB → 413。
+7. **写失败不 500**：monkeypatch writer 抛错 → 仍 2xx。
+8. **不污染服务日志**：`client` logger `propagate=False`，写客户端日志不落 `app-server.log`。
+
+## 13. env 变量汇总
+
+| env | 默认 | 用途 |
+|---|---|---|
+| `CLIENT_LOG_FILE` | `logs/app-client.log` | 客户端日志文件路径 |
+| `CLIENT_LOG_MAX_BYTES` | `20971520`(20MB) | 单文件滚动阈值 |
+| `CLIENT_LOG_BACKUP_COUNT` | `10` | 保留滚动文件数 |
+| `CLIENT_LOG_SERVICE_NAME` | `app-client` | 输出行 `service` 字段 |
+| `APPLOG_MAX_BATCH` | `500` | 单批最大条数 |
+| `APPLOG_MAX_BODY_BYTES` | `2097152`(2MB) | 请求体上限 |
+| `APPLOG_MAX_MSG_BYTES` | `8192` | 单条 msg 截断阈值 |
+
+滚动/文件类 env 在 `client_log.py` 用 `os.getenv` 读取（与 [logging.py](../../../app/core/logging.py) 风格一致）；请求限额类同样以 `os.getenv` 兜默认。
+
+## 14. 已知取舍 / 未来工作
+
+- **重复**：本期接受 SLS 重复条目；需要时加 `batch_id` 去重。
+- **多 worker**：见 §6 约束；扩容前迁移写入模型。
+- **服务端脱敏**：留作后续 knob。
+- **更多设备维度**（os/model/rom）：需要时加到批级字段或让客户端放 `data`；可经 `device_id` 与 analytics/device 表关联。

tests/test_admin_ad_revenue_scope.py

@@ -211,6 +211,59 @@ def test_reward_video_incomplete_playback_has_zero_revenue() -> None:
         db.close()
 
 
+def test_business_category_ecpm_merges_draw_feed_and_both_video_types() -> None:
+    db = SessionLocal()
+    phone = "18800009993"
+    category_date = "2040-02-05"
+    try:
+        user = User(phone=phone, username="29999999993", register_channel="sms")
+        db.add(user)
+        db.flush()
+        db.add_all([
+            AdEcpmRecord(
+                user_id=user.id, ad_type="draw", ad_session_id="category-draw",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="10000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 1, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="feed", ad_session_id="category-feed",
+                app_env="prod", our_code_id="prod-draw", ecpm_raw="20000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 2, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="reward_video", ad_session_id="category-reward",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="30000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 3, tzinfo=UTC),
+            ),
+            AdEcpmRecord(
+                user_id=user.id, ad_type="withdrawal_video", ad_session_id="category-withdraw",
+                app_env="prod", our_code_id="prod-reward", ecpm_raw="50000",
+                report_date=category_date, created_at=datetime(2040, 2, 5, 4, tzinfo=UTC),
+            ),
+        ])
+        db.commit()
+
+        result = ad_revenue.ad_revenue_report(
+            db,
+            date_from=category_date,
+            date_to=category_date,
+            user_id=user.id,
+            app_env="prod",
+            revenue_scope="all",
+        )
+
+        assert result["category_stats"] == {
+            "draw": {"impressions": 2, "revenue_yuan": 0.3, "ecpm_yuan": 150.0},
+            "video": {"impressions": 2, "revenue_yuan": 0.8, "ecpm_yuan": 400.0},
+        }
+    finally:
+        db.rollback()
+        db.execute(delete(AdEcpmRecord).where(AdEcpmRecord.report_date == category_date))
+        db.execute(delete(User).where(User.phone == phone))
+        db.commit()
+        db.close()
+
+
 def test_feed_reward_details_keep_each_record_adn() -> None:
     db = SessionLocal()
     phone = "18800009994"

tests/test_admin_read.py

@@ -13,6 +13,9 @@ from app.admin.repositories import queries
 from app.db.session import SessionLocal, engine
 from app.models.comparison import ComparisonRecord
 from app.models.feedback import Feedback
+from app.models.invite import InviteRelation
+from app.models.savings import SavingsRecord
+from app.models.user import User
 from app.models.wallet import CashTransaction, WithdrawOrder
 from app.repositories import user as user_repo
 from app.repositories import wallet as wallet_repo
@@ -165,6 +168,57 @@ def test_user_reward_detail_does_not_select_unrelated_new_ad_columns(
     assert records.status_code == 200, records.text
 
 
+def test_user_reward_stats_can_scope_withdrawals_by_account(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    uid = _seed_user_with_data("13800000023")
+    db = SessionLocal()
+    try:
+        account = wallet_repo.get_or_create_account(db, uid)
+        account.cash_balance_cents = 123
+        account.invite_cash_balance_cents = 456
+        db.add_all(
+            [
+                WithdrawOrder(
+                    user_id=uid,
+                    out_bill_no="rewardstatsinvite00000001",
+                    amount_cents=250,
+                    source="invite_cash",
+                    status="success",
+                ),
+                WithdrawOrder(
+                    user_id=uid,
+                    out_bill_no="rewardstatsinvite00000002",
+                    amount_cents=50,
+                    source="invite_cash",
+                    status="reviewing",
+                ),
+            ]
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    coin = admin_client.get(
+        f"/admin/api/users/{uid}/reward-stats",
+        params={"withdraw_source": "coin_cash"},
+        headers=_auth(admin_token),
+    )
+    invite = admin_client.get(
+        f"/admin/api/users/{uid}/reward-stats",
+        params={"withdraw_source": "invite_cash"},
+        headers=_auth(admin_token),
+    )
+    assert coin.status_code == 200, coin.text
+    assert invite.status_code == 200, invite.text
+    assert coin.json()["withdraw_success_cents"] == 100
+    assert coin.json()["withdraw_total"] == 1
+    assert coin.json()["cash_balance_cents"] == 123
+    assert invite.json()["withdraw_success_cents"] == 250
+    assert invite.json()["withdraw_total"] == 2
+    assert invite.json()["cash_balance_cents"] == 456
+
+
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
     """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
     naive = datetime(2038, 1, 1, 8, 0)
@@ -321,10 +375,19 @@ def test_withdraw_list_exposes_source_and_filters(
     db = SessionLocal()
     try:
         uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id
+        # 直接从 ORM 查询，验证列表富化返回人工风险结论。
+        user = db.get(User, uid)
+        assert user is not None
+        user.is_high_risk = True
+        user.high_risk_note = "邀请行为异常"
         db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001",
                              amount_cents=100, status="success", source="coin_cash"))
         db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001",
                              amount_cents=200, status="success", source="invite_cash"))
+        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview",
+                             amount_cents=50, status="reviewing", source="coin_cash"))
+        db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview",
+                             amount_cents=80, status="reviewing", source="invite_cash"))
         db.commit()
     finally:
         db.close()
@@ -339,6 +402,307 @@ def test_withdraw_list_exposes_source_and_filters(
     )
     items = r.json()["items"]
     assert items and all(it["source"] == "invite_cash" for it in items)
+    assert all(it["is_high_risk"] is True for it in items)
+    assert all(it["high_risk_note"] == "邀请行为异常" for it in items)
+    # 累计提现按当前审核页账户隔离，不把普通现金的 100 分串进邀请金。
+    assert all(it["cumulative_success_cents"] == 200 for it in items)
+
+    invite_summary = admin_client.get(
+        "/admin/api/withdraws/summary",
+        params={"source": "invite_cash"},
+        headers=_auth(admin_token),
+    ).json()
+    other_summary = admin_client.get(
+        "/admin/api/withdraws/summary",
+        params={"source": "coin_cash"},
+        headers=_auth(admin_token),
+    ).json()
+    assert invite_summary["reviewing_count"] >= 1
+    assert other_summary["reviewing_count"] >= 1
+    assert invite_summary["reviewing_amount_cents"] >= 80
+    assert other_summary["reviewing_amount_cents"] >= 50
+
+
+def test_invite_withdraw_detail_contains_invitee_first_actions(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    db = SessionLocal()
+    try:
+        inviter = user_repo.upsert_user_for_login(
+            db, phone="13800009031", register_channel="sms"
+        )
+        invitee = user_repo.upsert_user_for_login(
+            db, phone="13800009032", register_channel="sms"
+        )
+        db.add(
+            InviteRelation(
+                inviter_user_id=inviter.id,
+                invitee_user_id=invitee.id,
+                status="effective",
+                compare_reward_granted=True,
+                compare_reward_cents=50,
+            )
+        )
+        db.add(
+            ComparisonRecord(
+                user_id=invitee.id,
+                trace_id="invite-detail-first-compare",
+                status="success",
+                store_name="首次比价商家",
+                product_names="商品甲、商品乙",
+            )
+        )
+        db.add(
+            SavingsRecord(
+                user_id=invitee.id,
+                order_amount_cents=1888,
+                saved_amount_cents=300,
+                source="compare",
+                shop_name="首次下单商家",
+                title="首次下单商品",
+                dishes=[],
+            )
+        )
+        bill = "inviteoverviewbill0001"
+        db.add(
+            WithdrawOrder(
+                user_id=inviter.id,
+                out_bill_no=bill,
+                amount_cents=100,
+                status="reviewing",
+                source="invite_cash",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        f"/admin/api/withdraws/{bill}", headers=_auth(admin_token)
+    )
+    assert response.status_code == 200, response.text
+    overview = response.json()["invite_overview"]
+    assert overview["invite_total"] == 1
+    assert overview["invite_success_total"] == 1
+    item = overview["items"][0]
+    assert item["phone"] == "13800009032"
+    assert item["first_compare_store"] == "首次比价商家"
+    assert item["first_compare_products"] == "商品甲、商品乙"
+    assert item["first_order_store"] == "首次下单商家"
+    assert item["first_order_products"] == "首次下单商品"
+    assert item["first_order_amount_cents"] == 1888
+
+    filtered = admin_client.get(
+        f"/admin/api/withdraws/{bill}",
+        params={
+            "date_from": "2099-01-01T00:00:00Z",
+            "date_to": "2099-01-02T00:00:00Z",
+        },
+        headers=_auth(admin_token),
+    )
+    assert filtered.status_code == 200, filtered.text
+    assert filtered.json()["invite_overview"]["invite_total"] == 0
+
+
+def test_withdraw_summary_counts_match_status_lists_by_source(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    """每个审核页签的汇总数必须等于相同 source、status 的列表总数。"""
+    db = SessionLocal()
+    try:
+        uid = user_repo.upsert_user_for_login(
+            db, phone="13800009003", register_channel="sms"
+        ).id
+        for index, status in enumerate(
+            ("reviewing", "pending", "success", "rejected", "failed"), start=1
+        ):
+            db.add(
+                WithdrawOrder(
+                    user_id=uid,
+                    out_bill_no=f"summary{uid}invite{index}",
+                    amount_cents=index * 10,
+                    status=status,
+                    source="invite_cash",
+                )
+            )
+        db.add(
+            WithdrawOrder(
+                user_id=uid,
+                out_bill_no=f"summary{uid}coinfailed",
+                amount_cents=99,
+                status="failed",
+                source="coin_cash",
+            )
+        )
+        db.commit()
+    finally:
+        db.close()
+
+    status_fields = {
+        "reviewing": "reviewing_count",
+        "pending": "pending_count",
+        "success": "success_count",
+        "rejected": "rejected_count",
+        "failed": "failed_count",
+    }
+    for source in ("coin_cash", "invite_cash"):
+        summary_response = admin_client.get(
+            "/admin/api/withdraws/summary",
+            params={"source": source},
+            headers=_auth(admin_token),
+        )
+        assert summary_response.status_code == 200, summary_response.text
+        summary = summary_response.json()
+
+        for status, field in status_fields.items():
+            list_response = admin_client.get(
+                "/admin/api/withdraws",
+                params={"source": source, "status": status, "limit": 1},
+                headers=_auth(admin_token),
+            )
+            assert list_response.status_code == 200, list_response.text
+            assert summary[field] == list_response.json()["total"]
+
+        reviewing_response = admin_client.get(
+            "/admin/api/withdraws",
+            params={"source": source, "status": "reviewing", "limit": 100},
+            headers=_auth(admin_token),
+        )
+        assert reviewing_response.status_code == 200, reviewing_response.text
+        reviewing_body = reviewing_response.json()
+        assert reviewing_body["total"] == len(reviewing_body["items"])
+        assert summary["reviewing_amount_cents"] == sum(
+            item["amount_cents"] for item in reviewing_body["items"]
+        )
+
+        all_response = admin_client.get(
+            "/admin/api/withdraws",
+            params={"source": source, "limit": 1},
+            headers=_auth(admin_token),
+        )
+        assert all_response.status_code == 200, all_response.text
+        assert summary["total_count"] == all_response.json()["total"]
+        assert summary["total_count"] == sum(
+            summary[field] for field in status_fields.values()
+        )
+
+    assert admin_client.get(
+        "/admin/api/withdraws/summary",
+        params={"source": "unknown"},
+        headers=_auth(admin_token),
+    ).status_code == 422
+
+
+def test_comparison_records_show_readable_device_and_rom_version(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    db = SessionLocal()
+    try:
+        user = user_repo.upsert_user_for_login(
+            db, phone="13800009041", register_channel="sms"
+        )
+        record = ComparisonRecord(
+            user_id=user.id,
+            trace_id="comparison-readable-device",
+            status="success",
+            device_model="V2166BA",
+            device_manufacturer="vivo",
+            rom_vendor="vivo",
+            rom_name="OriginOS",
+            rom_version=4,
+            android_version="14",
+        )
+        db.add(record)
+        db.commit()
+        record_id = record.id
+        user_id = user.id
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/comparison-records",
+        params={"user_id": user_id},
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    item = response.json()["items"][0]
+    assert item["device_model"] == "V2166BA"
+    assert item["device_model_name"] == "vivo Y77e"
+    assert item["rom_name"] == "OriginOS"
+    assert item["rom_version"] == 4
+
+    detail = admin_client.get(
+        f"/admin/api/comparison-records/{record_id}",
+        headers=_auth(admin_token),
+    )
+    assert detail.status_code == 200, detail.text
+    assert detail.json()["device_model_name"] == "vivo Y77e"
+    assert detail.json()["rom_version"] == 4
+
+
+def test_comparison_records_show_real_order_status(
+    admin_client: TestClient, admin_token: str
+) -> None:
+    db = SessionLocal()
+    try:
+        user = user_repo.upsert_user_for_login(
+            db, phone="13800009042", register_channel="sms"
+        )
+        ordered_record = ComparisonRecord(
+            user_id=user.id,
+            trace_id="comparison-ordered-shop",
+            status="success",
+            store_name="真实下单店",
+        )
+        unordered_record = ComparisonRecord(
+            user_id=user.id,
+            trace_id="comparison-demo-order-shop",
+            status="success",
+            store_name="演示下单店",
+        )
+        db.add_all([ordered_record, unordered_record])
+        db.flush()
+        db.add_all(
+            [
+                SavingsRecord(
+                    user_id=user.id,
+                    order_amount_cents=1800,
+                    saved_amount_cents=300,
+                    shop_name="真实下单店",
+                    source="compare",
+                    client_event_id="admin-comparison-real-order",
+                ),
+                SavingsRecord(
+                    user_id=user.id,
+                    order_amount_cents=1500,
+                    saved_amount_cents=200,
+                    shop_name="演示下单店",
+                    source="demo",
+                ),
+            ]
+        )
+        db.commit()
+        ordered_record_id = ordered_record.id
+        user_id = user.id
+    finally:
+        db.close()
+
+    response = admin_client.get(
+        "/admin/api/comparison-records",
+        params={"user_id": user_id},
+        headers=_auth(admin_token),
+    )
+    assert response.status_code == 200, response.text
+    by_trace = {item["trace_id"]: item for item in response.json()["items"]}
+    assert by_trace["comparison-ordered-shop"]["ordered"] is True
+    assert by_trace["comparison-demo-order-shop"]["ordered"] is False
+
+    detail = admin_client.get(
+        f"/admin/api/comparison-records/{ordered_record_id}",
+        headers=_auth(admin_token),
+    )
+    assert detail.status_code == 200, detail.text
+    assert detail.json()["ordered"] is True
 
 
 def test_ad_coin_audit_full_count_truncate_and_only_mismatch(

tests/test_admin_roles.py

@@ -190,7 +190,13 @@ def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
     assert roles["finance"]["label"] == "财务"
     assert roles["tech"]["label"] == "技术"
     # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
-    assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
+    assert set(roles["finance"]["pages"]) == {
+        "dashboard",
+        "ad-revenue-report",
+        "cps",
+        "invite-withdraws",
+        "withdraws",
+    }
     assert set(roles["tech"]["pages"]) == {
         "dashboard", "risk-monitor", "device-liveness", "analytics-health", "config",
         "ad-revenue", "huawei-review", "event-logs", "audit-logs",

tests/test_admin_write.py

@@ -103,6 +103,61 @@ def _seed_price_report(phone: str) -> int:
         db.close()
 
 
+def test_set_user_risk_requires_note_and_writes_audit(
+    admin_client: TestClient, operator_token: str
+) -> None:
+    uid = _seed_user("13900000991")
+
+    missing_note = admin_client.post(
+        f"/admin/api/users/{uid}/risk",
+        json={"is_high_risk": True, "note": "   "},
+        headers=_auth(operator_token),
+    )
+    assert missing_note.status_code == 422
+
+    marked = admin_client.post(
+        f"/admin/api/users/{uid}/risk",
+        json={"is_high_risk": True, "note": "邀请记录存在批量异常"},
+        headers=_auth(operator_token),
+    )
+    assert marked.status_code == 200, marked.text
+
+    listed = admin_client.get(
+        "/admin/api/users",
+        params={"phone": "13900000991"},
+        headers=_auth(operator_token),
+    ).json()["items"][0]
+    assert listed["is_high_risk"] is True
+    assert listed["high_risk_note"] == "邀请记录存在批量异常"
+
+    db = SessionLocal()
+    try:
+        audit = db.execute(
+            select(AdminAuditLog)
+            .where(
+                AdminAuditLog.action == "user.risk.set",
+                AdminAuditLog.target_id == str(uid),
+            )
+            .order_by(AdminAuditLog.id.desc())
+        ).scalars().first()
+        assert audit is not None
+        assert audit.detail["after"]["is_high_risk"] is True
+    finally:
+        db.close()
+
+    cleared = admin_client.post(
+        f"/admin/api/users/{uid}/risk",
+        json={"is_high_risk": False, "note": None},
+        headers=_auth(operator_token),
+    )
+    assert cleared.status_code == 200
+    overview = admin_client.get(
+        f"/admin/api/users/{uid}", headers=_auth(operator_token)
+    ).json()
+    assert overview["user"]["is_high_risk"] is False
+    assert overview["user"]["high_risk_note"] is None
+
+
 # ===== 调金币 =====
 
 def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
@@ -579,15 +634,47 @@ def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> N
     ).status_code == 404
 
 
-def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
+def test_withdraw_refresh_enriches_existing_failed_reason(
+    admin_client: TestClient, finance_token: str, monkeypatch
+) -> None:
+    uid = _seed_user("13900000017")
+    with SessionLocal() as db:
+        db.add(
+            WithdrawOrder(
+                user_id=uid,
+                out_bill_no="adminfailedreason01",
+                amount_cents=100,
+                status="failed",
+                wechat_state="FAIL",
+                fail_reason="微信转账状态 FAIL",
+            )
+        )
+        db.commit()
+
     from app.repositories import wallet as wr
+
     monkeypatch.setattr(
-        wr.wxpay, "query_transfer",
-        lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
+        wr.wxpay,
+        "query_transfer",
+        lambda out_bill_no: {
+            "status_code": 200,
+            "data": {
+                "state": "FAIL",
+                "fail_reason": "TRANSFER_RISK",
+                "transfer_bill_no": "wx_failed_reason_01",
+            },
+        },
     )
-    r = admin_client.post(
-        "/admin/api/withdraws/reconcile", params={"older_than_minutes": 0},
+
+    response = admin_client.post(
+        "/admin/api/withdraws/adminfailedreason01/refresh",
         headers=_auth(finance_token),
     )
-    assert r.status_code == 200, r.text
-    assert "checked" in r.json() and "resolved" in r.json()
+
+    assert response.status_code == 200, response.text
+    assert response.json()["status"] == "failed"
+    assert response.json()["wechat_state"] == "FAIL"
+    assert response.json()["transfer_bill_no"] == "wx_failed_reason_01"
+    assert response.json()["fail_reason"] == (
+        "微信转账失败：该笔转账存在风险，已被微信拦截（TRANSFER_RISK）"
+    )

tests/test_applog.py

@@ -0,0 +1,182 @@
+"""客户端运行日志上报:writer 单测 + 端点集成测试。"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from app.core import client_log
+
+
+@pytest.fixture()
+def client_log_file(tmp_path, monkeypatch):
+    """把客户端日志切到临时文件,并重置 writer 单例使其按当时 env 重建。"""
+    p = tmp_path / "app-client.log"
+    monkeypatch.setenv("CLIENT_LOG_FILE", str(p))
+    client_log.reset_client_logger()
+    yield p
+    client_log.reset_client_logger()
+
+
+def _read_lines(p: Path) -> list[dict]:
+    text = p.read_text(encoding="utf-8").strip()
+    return [json.loads(ln) for ln in text.splitlines() if ln]
+
+
+# ---------------- writer 层 ----------------
+
+def _meta(**kw) -> dict:
+    base = {"device_id": "d-1", "user_id": None, "app_ver": None,
+            "platform": None, "sent_at": None}
+    base.update(kw)
+    return base
+
+
+def test_writer_writes_one_line_per_record(client_log_file):
+    recs = [
+        {"client_ts": 1737000000000, "level": "info", "msg": "hello"},
+        {"client_ts": 1737000000001, "level": "error", "msg": "boom", "tag": "net"},
+    ]
+    received, dropped = client_log.write_records(
+        recs, meta=_meta(user_id=42, app_ver="1.2.3", platform="android"),
+        client_ip="1.2.3.4",
+    )
+    assert (received, dropped) == (2, 0)
+    lines = _read_lines(client_log_file)
+    assert len(lines) == 2
+    assert lines[0]["source"] == "client"
+    assert lines[0]["service"] == "app-client"
+    assert lines[0]["device_id"] == "d-1"
+    assert lines[0]["user_id"] == 42
+    assert lines[0]["app_ver"] == "1.2.3"
+    assert lines[0]["client_ip"] == "1.2.3.4"
+    assert lines[0]["level"] == "INFO"  # 归一化大写
+    assert lines[0]["msg"] == "hello"
+    assert lines[0]["client_ts"] == 1737000000000
+    assert lines[1]["tag"] == "net"
+
+
+def test_writer_hoists_trace_id_to_top_level(client_log_file):
+    client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x", "trace_id": "abc123"}],
+        meta=_meta(), client_ip="",
+    )
+    assert _read_lines(client_log_file)[0]["trace_id"] == "abc123"
+
+
+def test_writer_sweeps_unknown_keys_into_data(client_log_file):
+    client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x",
+          "foo": 123, "data": {"bar": "baz"}}],
+        meta=_meta(), client_ip="",
+    )
+    line = _read_lines(client_log_file)[0]
+    assert "foo" not in line               # 白名单外不进顶层
+    assert line["data"]["foo"] == 123      # 兜底进 data
+    assert line["data"]["bar"] == "baz"    # 客户端自带 data 合并进来
+
+
+def test_writer_truncates_oversize_msg(client_log_file, monkeypatch):
+    monkeypatch.setenv("APPLOG_MAX_MSG_BYTES", "10")
+    received, dropped = client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x" * 100}],
+        meta=_meta(), client_ip="",
+    )
+    assert (received, dropped) == (1, 0)   # 截断而非丢弃
+    line = _read_lines(client_log_file)[0]
+    assert line["msg_truncated"] is True
+    assert line["msg"].endswith("…[truncated]")
+    body = line["msg"].removesuffix("…[truncated]")
+    assert len(body.encode("utf-8")) <= 10   # 截断体不超过字节上限
+
+
+def test_writer_caps_indexed_field_lengths(client_log_file):
+    client_log.write_records(
+        [{"client_ts": 1, "level": "info" * 20, "msg": "x",
+          "trace_id": "t" * 1000, "tag": "g" * 1000}],
+        meta=_meta(), client_ip="",
+    )
+    line = _read_lines(client_log_file)[0]
+    assert len(line["level"]) <= 16       # 索引字段(spec §8)做长度上限
+    assert len(line["trace_id"]) <= 256
+    assert len(line["tag"]) <= 128
+
+
+def test_writer_logger_does_not_propagate(client_log_file):
+    lg = client_log.get_logger()
+    assert lg.name == "shagua.client_log"
+    assert lg.propagate is False           # 不冒泡到 root → 不写 app-server.log
+
+
+def test_writer_drops_bad_record_without_raising(client_log_file, monkeypatch):
+    def _boom(*a, **kw):
+        raise RuntimeError("boom")
+    monkeypatch.setattr(client_log, "_build_line", _boom)
+    received, dropped = client_log.write_records(
+        [{"client_ts": 1, "level": "info", "msg": "x"}],
+        meta=_meta(), client_ip="",
+    )
+    assert (received, dropped) == (0, 1)   # 坏条计 dropped,且不抛给上层
+
+
+# ---------------- 端点层 ----------------
+
+def _post(client, body):
+    return client.post("/api/v1/applog/batch", json=body)
+
+
+def test_endpoint_happy_path(client, client_log_file):
+    body = {
+        "device_id": "d-1", "user_id": 42, "app_ver": "1.2.3", "platform": "android",
+        "logs": [
+            {"client_ts": 1, "level": "info", "msg": "a", "trace_id": "t1"},
+            {"client_ts": 2, "level": "warn", "msg": "b"},
+        ],
+    }
+    resp = _post(client, body)
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "received": 2, "dropped": 0}
+    lines = _read_lines(client_log_file)
+    assert len(lines) == 2
+    assert lines[0]["trace_id"] == "t1"       # 端到端:trace_id 落到文件顶层
+    assert lines[0]["client_ip"]              # 服务端补了 IP
+
+
+def test_endpoint_rejects_batch_over_max(client, client_log_file):
+    body = {"device_id": "d-1",
+            "logs": [{"client_ts": i, "level": "info", "msg": str(i)} for i in range(501)]}
+    resp = _post(client, body)
+    assert resp.status_code == 422           # Pydantic max_length=500
+
+
+def test_endpoint_rejects_body_over_limit(client, client_log_file, monkeypatch):
+    # 只测有 Content-Length 的常规情形(TestClient/requests 恒发该头)。缺该头(chunked)
+    # 时依赖 nginx client_max_body_size 兜底,不在应用层测。
+    monkeypatch.setenv("APPLOG_MAX_BODY_BYTES", "50")
+    body = {"device_id": "d-1", "logs": [{"client_ts": 1, "level": "info", "msg": "x" * 500}]}
+    resp = _post(client, body)
+    assert resp.status_code == 413           # 依赖查 Content-Length,body 校验前拦截
+
+
+def test_endpoint_returns_200_on_write_failure(client, client_log_file, monkeypatch):
+    class _BoomLogger:
+        name = "shagua.client_log"
+        propagate = False
+        handlers: list = []
+
+        def info(self, *a, **k):
+            raise RuntimeError("disk full")
+
+    monkeypatch.setattr(client_log, "get_logger", lambda: _BoomLogger())
+    body = {"device_id": "d-1", "logs": [
+        {"client_ts": 1, "level": "info", "msg": "a"},
+        {"client_ts": 2, "level": "info", "msg": "b"}]}
+    resp = _post(client, body)
+    assert resp.status_code == 200           # fire-and-forget:写失败不 500
+    assert resp.json() == {"ok": True, "received": 0, "dropped": 2}
+
+
+def test_endpoint_missing_device_id_is_422(client, client_log_file):
+    resp = _post(client, {"logs": [{"client_ts": 1, "level": "info", "msg": "a"}]})
+    assert resp.status_code == 422           # device_id 必填

tests/test_compare_harvest.py

@@ -9,17 +9,16 @@ pricebot 用 httpx mock,不真连(同 test_compare_proxy)。
 """
 from __future__ import annotations
 
-import json
 import uuid
 from unittest.mock import MagicMock, patch
 
 import httpx
+from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
 from app.schemas.compare_record import ComparisonRecordIn
-from sqlalchemy import select
 
 
 def _tid() -> str:
@@ -216,7 +215,9 @@ def test_price_step_done_harvests_success(client) -> None:
                   "action": {"command": "done", "params": _done_params()},
                   "trace_url": "https://price.shaguabijia.com/traces/done2/"}
     p, _cap = _mock_pricebot(done_frame)
-    with p:
+    with p, patch(
+        "app.api.v1.compare.backfill_comparison_llm_cost"
+    ) as backfill:
         r = client.post("/api/v1/price/step", json=_stub_body(trace_id=tid, step=8))
     assert r.status_code == 200
     with SessionLocal() as db:
@@ -224,6 +225,7 @@ def test_price_step_done_harvests_success(client) -> None:
         assert rec is not None and rec.status == "success"
         assert rec.best_platform_id == "meituan"
         assert rec.saved_amount_cents == 500
+        backfill.assert_called_once_with(rec.id, tid)
 
 
 def test_trace_finalize_harvests_abort(client) -> None:

tests/test_comparison_llm_backfill.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from app.core.rewards import CN_TZ
+from app.db.session import SessionLocal
+from app.models.app_config import AppConfig
+from app.models.comparison import ComparisonRecord
+from app.services import comparison_llm_backfill
+
+
+def _record(
+    trace_id: str,
+    *,
+    status: str = "success",
+    created_at: datetime | None = None,
+) -> int:
+    with SessionLocal() as db:
+        rec = ComparisonRecord(
+            trace_id=trace_id,
+            status=status,
+            created_at=created_at or datetime.now(),
+        )
+        db.add(rec)
+        db.commit()
+        return rec.id
+
+
+def _delete(record_id: int) -> None:
+    with SessionLocal() as db:
+        rec = db.get(ComparisonRecord, record_id)
+        if rec is not None:
+            db.delete(rec)
+            db.commit()
+
+
+def test_backfill_retries_then_persists_cost(monkeypatch):
+    record_id = _record("llm-retry-1")
+    calls = [
+        {
+            "model": "unknown-model",
+            "error": None,
+            "usage": {"prompt_tokens": 1000, "completion_tokens": 500},
+        }
+    ]
+    responses = iter([[], calls])
+    monkeypatch.setattr(
+        comparison_llm_backfill,
+        "fetch_llm_calls",
+        lambda trace_id: next(responses),
+    )
+    sleeps: list[float] = []
+    monkeypatch.setattr(comparison_llm_backfill.time, "sleep", sleeps.append)
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
+
+    try:
+        assert comparison_llm_backfill.backfill_comparison_llm_cost(
+            record_id, "llm-retry-1", attempts=2, retry_delays=(0.25,)
+        )
+        assert sleeps == [0.25]
+        with SessionLocal() as db:
+            rec = db.get(ComparisonRecord, record_id)
+            assert rec.input_tokens == 1000
+            assert rec.output_tokens == 500
+            assert rec.llm_cost_yuan is not None
+            assert rec.llm_calls == calls
+    finally:
+        _delete(record_id)
+
+
+def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
+    missing_id = _record("llm-repair-missing")
+    running_id = _record("llm-repair-running", status="running")
+    calls = [
+        {
+            "model": "unknown-model",
+            "error": None,
+            "usage": {"prompt_tokens": 100, "completion_tokens": 20},
+        }
+    ]
+    seen: list[str] = []
+
+    def fetch(trace_id: str) -> list[dict]:
+        seen.append(trace_id)
+        return calls
+
+    monkeypatch.setattr(comparison_llm_backfill, "fetch_llm_calls", fetch)
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
+    try:
+        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
+            limit=10, lookback_days=1
+        )
+        assert result["repaired"] >= 1
+        assert "llm-repair-missing" in seen
+        assert "llm-repair-running" not in seen
+        with SessionLocal() as db:
+            assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
+            assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
+    finally:
+        _delete(missing_id)
+        _delete(running_id)
+
+
+def test_repair_excludes_records_before_current_price_config(monkeypatch):
+    price_changed_at = datetime.now(UTC) - timedelta(hours=1)
+    before_change = (price_changed_at - timedelta(minutes=30)).astimezone(CN_TZ)
+    after_change = (price_changed_at + timedelta(minutes=30)).astimezone(CN_TZ)
+    before_id = _record(
+        "llm-before-price-change",
+        created_at=before_change.replace(tzinfo=None),
+    )
+    after_id = _record(
+        "llm-after-price-change",
+        created_at=after_change.replace(tzinfo=None),
+    )
+    with SessionLocal() as db:
+        existing = db.get(AppConfig, "llm_token_price")
+        if existing is not None:
+            db.delete(existing)
+            db.flush()
+        db.add(
+            AppConfig(
+                key="llm_token_price",
+                value={"default": {"input_per_1m": 1, "output_per_1m": 1}},
+                updated_at=price_changed_at,
+            )
+        )
+        db.commit()
+
+    seen: list[str] = []
+
+    def backfill(record_id: int, trace_id: str, **kwargs) -> bool:
+        seen.append(trace_id)
+        return True
+
+    monkeypatch.setattr(
+        comparison_llm_backfill,
+        "backfill_comparison_llm_cost",
+        backfill,
+    )
+    try:
+        result = comparison_llm_backfill.repair_missing_comparison_llm_costs(
+            limit=10_000,
+            lookback_days=1,
+        )
+        assert "llm-after-price-change" in seen
+        assert "llm-before-price-change" not in seen
+        assert result["repaired"] == len(seen)
+        assert result["unresolved"] == 0
+    finally:
+        _delete(before_id)
+        _delete(after_id)
+        with SessionLocal() as db:
+            config = db.get(AppConfig, "llm_token_price")
+            if config is not None:
+                db.delete(config)
+                db.commit()

tests/test_comparison_result_in_skipped.py

@@ -0,0 +1,17 @@
+"""ComparisonResultIn 保留逐平台 skipped_dish_count(落库/回读用)。
+
+记录页三平台网格要按逐平台缺菜数标"缺少 X 个菜品"; pricebot 现把 skipped_dish_count 冗余进
+comparison_results 行。上报边界必须显式声明该字段, 否则 model_dump() 会静默丢弃(POST 路径),
+记录页拿不到。纯 schema, 不碰 DB。
+"""
+from app.schemas.compare_record import ComparisonResultIn
+
+
+def test_comparison_result_in_preserves_skipped_dish_count():
+    r = ComparisonResultIn(platform_id="jd_waimai", price=25.0, skipped_dish_count=2)
+    assert r.model_dump().get("skipped_dish_count") == 2
+
+
+def test_comparison_result_in_skipped_defaults_none():
+    r = ComparisonResultIn(platform_id="meituan", price=42.0, is_source=True)
+    assert r.model_dump().get("skipped_dish_count") is None

tests/test_invite_cash_withdraw.py

@@ -127,6 +127,51 @@ def test_invite_cash_reject_refunds_invite_account(client, monkeypatch) -> None:
         db.close()
 
 
+def test_invite_cash_reject_recreates_missing_account(client, monkeypatch) -> None:
+    """A legacy or hand-written order without CoinAccount can still be refunded."""
+    _patch_userinfo(monkeypatch, "openid_ic_missing_account")
+    token = _login(client, "13800004012")
+    _seed_balances(client, token, "13800004012", cash=0, invite_cash=200)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    response = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 200, "source": "invite_cash"},
+        headers=_auth(token),
+    )
+    assert response.status_code == 200, response.text
+    bill = response.json()["out_bill_no"]
+
+    with SessionLocal() as db:
+        user = db.execute(
+            select(User).where(User.phone == "13800004012")
+        ).scalar_one()
+        db.delete(db.get(CoinAccount, user.id))
+        db.commit()
+
+    _reject(bill, "missing account fixture")
+
+    with SessionLocal() as db:
+        user = db.execute(
+            select(User).where(User.phone == "13800004012")
+        ).scalar_one()
+        account = db.get(CoinAccount, user.id)
+        order = crud_wallet._get_withdraw_or_raise(db, bill)
+        refunds = db.execute(
+            select(InviteCashTransaction).where(
+                InviteCashTransaction.user_id == user.id,
+                InviteCashTransaction.biz_type == "invite_withdraw_refund",
+                InviteCashTransaction.ref_id == bill,
+            )
+        ).scalars().all()
+
+        assert account is not None
+        assert account.invite_cash_balance_cents == 200
+        assert order.status == "rejected"
+        assert order.fail_reason == "missing account fixture"
+        assert len(refunds) == 1 and refunds[0].amount_cents == 200
+
+
 def test_two_accounts_withdraw_independent(client, monkeypatch) -> None:
     """两账户各提各的不串:提 invite_cash(拒绝→验证退款回原账户),再提 cash,各扣各账户互不串。"""
     _patch_userinfo(monkeypatch, "openid_ic_3")

tests/test_llm_cost.py

@@ -132,6 +132,7 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
     from app.models.app_config import AppConfig
     from app.models.comparison import ComparisonRecord
     from app.repositories import app_config
+    from app.services import comparison_llm_backfill
 
     sample = [
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1512, "completion_tokens": 22}},
@@ -139,7 +140,12 @@ def test_backfill_llm_calls_stores_cost_and_snapshot(monkeypatch):
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1940, "completion_tokens": 142}},
         {"model": "qwen3.5-flash", "error": None, "usage": {"prompt_tokens": 1325, "completion_tokens": 13}},
     ]
-    monkeypatch.setattr(compare_record, "fetch_llm_calls", lambda trace_id: sample)
+    monkeypatch.setattr(
+        comparison_llm_backfill, "fetch_llm_calls", lambda trace_id: sample
+    )
+    monkeypatch.setattr(
+        comparison_llm_backfill.settings, "INTERNAL_API_SECRET", "test-secret"
+    )
 
     db = SessionLocal()
     try:

tests/test_pricebot_llm_calls.py

@@ -0,0 +1,55 @@
+from unittest.mock import MagicMock
+
+from app.core.config import settings
+from app.services import pricebot_llm_calls
+
+
+def test_auth_probe_accepts_matching_secret(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
+    response = MagicMock(status_code=200)
+    get = MagicMock(return_value=response)
+    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
+
+    assert pricebot_llm_calls.pricebot_llm_auth_ready() is True
+    assert get.call_count == len(settings.pricebot_instances)
+    assert all(
+        call.kwargs["headers"]["X-Internal-Secret"] == "same-secret"
+        for call in get.call_args_list
+    )
+
+
+def test_auth_probe_rejects_mismatched_secret(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "app-server-secret")
+    monkeypatch.setattr(
+        pricebot_llm_calls.httpx,
+        "get",
+        MagicMock(return_value=MagicMock(status_code=403)),
+    )
+
+    assert pricebot_llm_calls.pricebot_llm_auth_ready() is False
+
+
+def test_fetch_falls_back_to_other_instance_when_hash_target_is_empty(monkeypatch):
+    monkeypatch.setattr(settings, "INTERNAL_API_SECRET", "same-secret")
+    monkeypatch.setattr(
+        settings,
+        "PRICEBOT_INSTANCES",
+        "http://pricebot-1:8000,http://pricebot-2:8000",
+    )
+    monkeypatch.setattr(
+        pricebot_llm_calls,
+        "pick_pricebot",
+        lambda trace_id: "http://pricebot-1:8000",
+    )
+    calls = [{"model": "qwen", "usage": {"prompt_tokens": 1}}]
+
+    def get(url, **kwargs):
+        response = MagicMock(status_code=200)
+        response.json.return_value = {
+            "calls": [] if "pricebot-1" in url else calls
+        }
+        return response
+
+    monkeypatch.setattr(pricebot_llm_calls.httpx, "get", get)
+
+    assert pricebot_llm_calls.fetch_llm_calls("trace-after-rescale") == calls

tests/test_withdraw.py

@@ -9,7 +9,7 @@ from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.user import User
-from app.models.wallet import CoinAccount, WithdrawOrder, WechatTransferAuthorization
+from app.models.wallet import CoinAccount, WechatTransferAuthorization, WithdrawOrder
 from app.repositories import wallet as crud_wallet
 
 
@@ -232,6 +232,78 @@ def test_withdraw_approve_transfer_fail_refunds(client, monkeypatch) -> None:
     assert r.json()["items"][0]["status"] == "failed"
 
 
+def test_approved_withdraw_query_fail_exposes_wechat_reason(client, monkeypatch) -> None:
+    """审核通过后微信终态失败时，保存官方失败码对应原因并退回余额。"""
+    monkeypatch.setattr(
+        "app.integrations.wxpay.code_to_userinfo",
+        lambda code: {
+            "openid": "openid_query_fail",
+            "nickname": None,
+            "avatar_url": None,
+            "raw": {},
+        },
+    )
+    monkeypatch.setattr(
+        "app.integrations.wxpay.create_transfer",
+        lambda openid, amount_fen, out_bill_no, user_name=None: {
+            "status_code": 200,
+            "data": {
+                "state": "WAIT_USER_CONFIRM",
+                "package_info": "pkg_query_fail",
+                "transfer_bill_no": "tb_query_fail",
+            },
+        },
+    )
+    token = _login(client, "13800002025")
+    _seed_cash(client, token, "13800002025", 100)
+    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
+
+    response = client.post(
+        "/api/v1/wallet/withdraw",
+        json={"amount_cents": 50},
+        headers=_auth(token),
+    )
+    bill = response.json()["out_bill_no"]
+    _approve(bill)
+
+    monkeypatch.setattr(
+        "app.integrations.wxpay.query_transfer",
+        lambda out_bill_no: {
+            "status_code": 200,
+            "data": {
+                "state": "FAIL",
+                "fail_reason": "PAYEE_ACCOUNT_ABNORMAL",
+                "transfer_bill_no": "tb_query_fail",
+            },
+        },
+    )
+    response = client.get(
+        "/api/v1/wallet/withdraw/status",
+        params={"out_bill_no": bill},
+        headers=_auth(token),
+    )
+
+    assert response.status_code == 200, response.text
+    assert response.json()["status"] == "failed"
+    assert response.json()["wechat_state"] == "FAIL"
+    assert response.json()["fail_reason"] == (
+        "微信转账失败：用户微信账户收款异常（PAYEE_ACCOUNT_ABNORMAL）"
+    )
+    with SessionLocal() as db:
+        order = db.execute(
+            select(WithdrawOrder).where(WithdrawOrder.out_bill_no == bill)
+        ).scalar_one()
+        assert order.transfer_bill_no == "tb_query_fail"
+    account = client.get("/api/v1/wallet/account", headers=_auth(token))
+    assert account.json()["cash_balance_cents"] == 100
+
+
+def test_unknown_wechat_fail_reason_code_is_preserved() -> None:
+    assert crud_wallet._wechat_terminal_failure_reason(
+        {"fail_reason": "NEW_WECHAT_REASON"}, "FAIL"
+    ) == "微信转账失败：NEW_WECHAT_REASON"
+
+
 def test_withdraw_idempotent_same_bill_no(client, monkeypatch) -> None:
     """#2 同 out_bill_no 重试:只转一次,第二次返回同一单,余额只扣一次。"""
     monkeypatch.setattr("app.integrations.wxpay.code_to_userinfo", lambda code: {"openid": "openid_idem", "nickname": None, "avatar_url": None, "raw": {}})

tests/test_withdraw_ledger_check.py

@@ -1,151 +0,0 @@
-"""提现现金账本校验(admin ledger-check)测试:两本物理隔离的账各自对账。
-
-历史盲区:`withdraw_ledger_check` 曾拿全部提现单去和**普通现金流水**(cash_transaction)比对,
-而 source=invite_cash 的提现单流水其实在 invite_cash_transaction 表,导致每笔邀请提现单都被
-误报「缺扣款/缺退款流水」。这里用真实提现 API 造单 + before/after 差值断言锁定修复:
-  1) 邀请提现单不再污染普通现金账的缺流水计数;
-  2) 邀请账户已被纳入对账(能抓到它自己的缺流水);
-  3) 普通现金账的原有对账未被改坏。
-
-conftest 的库是 session 级共享、测试间不清,故一律用 before/after 差值,只反映本用例造的数据。
-"""
-from __future__ import annotations
-
-from sqlalchemy import delete, select
-
-from app.admin.repositories.queries import withdraw_ledger_check
-from app.db.session import SessionLocal
-from app.models.user import User
-from app.models.wallet import CoinAccount, InviteCashTransaction
-from app.repositories import wallet as crud_wallet
-
-
-def _login(client, phone: str) -> str:
-    client.post("/api/v1/auth/sms/send", json={"phone": phone})
-    r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
-    assert r.status_code == 200, r.text
-    return r.json()["access_token"]
-
-
-def _auth(token: str) -> dict[str, str]:
-    return {"Authorization": f"Bearer {token}"}
-
-
-def _patch_userinfo(monkeypatch, openid: str) -> None:
-    monkeypatch.setattr(
-        "app.integrations.wxpay.code_to_userinfo",
-        lambda code: {"openid": openid, "nickname": None, "avatar_url": None, "raw": {}},
-    )
-
-
-def _seed_balances(client, token: str, phone: str, *, cash: int = 0, invite_cash: int = 0) -> None:
-    client.get("/api/v1/wallet/account", headers=_auth(token))
-    db = SessionLocal()
-    try:
-        user = db.execute(select(User).where(User.phone == phone)).scalar_one()
-        acc = db.get(CoinAccount, user.id)
-        acc.cash_balance_cents = cash
-        acc.invite_cash_balance_cents = invite_cash
-        db.commit()
-    finally:
-        db.close()
-
-
-def _reject(bill: str, reason: str = "测试拒绝") -> None:
-    db = SessionLocal()
-    try:
-        crud_wallet.reject_withdraw(db, bill, reason)
-    finally:
-        db.close()
-
-
-def _ledger() -> dict:
-    db = SessionLocal()
-    try:
-        return withdraw_ledger_check(db)
-    finally:
-        db.close()
-
-
-def test_rejected_invite_withdraw_not_flagged_missing(client, monkeypatch) -> None:
-    """核心回归:一笔被拒绝的 invite_cash 提现单,扣款/退款流水都在 invite_cash_transaction,
-    不应让普通现金账的缺扣款/缺退款计数增加(修复前每笔会各 +1)。"""
-    before = _ledger()
-
-    _patch_userinfo(monkeypatch, "openid_lc_1")
-    token = _login(client, "13800005001")
-    _seed_balances(client, token, "13800005001", cash=0, invite_cash=500)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-    _reject(r.json()["out_bill_no"])  # rejected + 退款流水落 invite_cash_transaction
-
-    after = _ledger()
-
-    # 普通现金账不该因这笔 invite 单产生缺流水(修复前会各 +1 → 就是页面上看到的误报)
-    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
-    assert after["missing_refund_txn_count"] == before["missing_refund_txn_count"]
-    # 邀请账扣款 + 退款流水齐全,邀请账自身也不该缺
-    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]
-    assert after["invite_missing_refund_txn_count"] == before["invite_missing_refund_txn_count"]
-
-
-def test_invite_ledger_detects_missing_withdraw_txn(client, monkeypatch) -> None:
-    """删掉一笔 invite 提现单的扣款流水 → 邀请账缺扣款计数 +1、ok=False,
-    证明邀请账户已真正纳入对账(修复前邀请账完全不校验、永远报不出问题)。"""
-    _patch_userinfo(monkeypatch, "openid_lc_2")
-    token = _login(client, "13800005002")
-    _seed_balances(client, token, "13800005002", cash=0, invite_cash=500)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        json={"amount_cents": 200, "source": "invite_cash"},
-        headers=_auth(token),
-    )
-    bill = r.json()["out_bill_no"]
-
-    before = _ledger()
-    db = SessionLocal()
-    try:
-        db.execute(
-            delete(InviteCashTransaction).where(
-                InviteCashTransaction.ref_id == bill,
-                InviteCashTransaction.biz_type == "invite_withdraw",
-            )
-        )
-        db.commit()
-    finally:
-        db.close()
-    after = _ledger()
-
-    assert (
-        after["invite_missing_withdraw_txn_count"]
-        == before["invite_missing_withdraw_txn_count"] + 1
-    )
-    assert after["ok"] is False
-
-
-def test_coin_cash_withdraw_still_reconciled(client, monkeypatch) -> None:
-    """普通现金 coin_cash 提现单齐全时不新增缺流水(确保分账改造没弄坏原有普通现金对账)。"""
-    _patch_userinfo(monkeypatch, "openid_lc_3")
-    token = _login(client, "13800005003")
-    _seed_balances(client, token, "13800005003", cash=500, invite_cash=0)
-    client.post("/api/v1/wallet/bind-wechat", json={"code": "c"}, headers=_auth(token))
-
-    before = _ledger()
-    r = client.post(
-        "/api/v1/wallet/withdraw",
-        # 50 分 = 0.5 元档(7-9 起 coin_cash 只能提预设档位)
-        json={"amount_cents": 50, "source": "coin_cash"},
-        headers=_auth(token),
-    )
-    assert r.status_code == 200, r.text
-    after = _ledger()
-
-    # 普通现金提现扣款流水随单写入 cash_transaction,缺扣款计数不变;邀请账更不受影响
-    assert after["missing_withdraw_txn_count"] == before["missing_withdraw_txn_count"]
-    assert after["invite_missing_withdraw_txn_count"] == before["invite_missing_withdraw_txn_count"]