feat(admin/ad): 提现自动对账后台开关 + 客户端广告 flags 端点 + 穿山甲多 m-key 验签 (#59)

- 自动对账接入 app_config 运行时配置(新增 bool 配置类型):env 部署总闸 + DB 运营日常开关
  双层;worker 每轮读 DB 即时生效,健康检查改报「env AND DB」实际生效态
- 新增 GET /api/v1/platform/flags(不鉴权)下发 comparing_ad_enabled 远程 kill-switch,
  客户端拉取后缓存;空库回退默认 True
- 穿山甲发奖回调支持多激励位 m-key(三命名项 PANGLE_REWARD_SECRET_TEST/_DEDICATED/_PROD
  + 旧逗号分隔合并去重),verify_callback_sign_any 逐个验签任一通过即受理;向后兼容旧单 key
- 补 test_platform / bool config 用例;app_config 文档同步

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: OuYingJun1024 <1034284404@qq.com>
Reviewed-on: #59
Co-authored-by: ouzhou <ouzhou@wonderable.ai>
Co-committed-by: ouzhou <ouzhou@wonderable.ai>

.env.example

@@ -88,9 +88,17 @@ AUTO_EXCHANGE_ENABLED=true
 
 # ===== 穿山甲激励视频(服务端发奖回调)=====
 # 看完激励视频后穿山甲服务器 S2S 回调本服务发金币(客户端不参与发奖)。
-# PANGLE_REWARD_SECRET 是穿山甲后台配置的"奖励校验密钥",用于验签,从后台取到后填这里;
+# 穿山甲"奖励校验密钥"(m-key),验签用,从 GroMore 后台各广告位取到后填这里。
-# 配齐并把 ENABLED=true 后,/api/v1/ad/pangle-callback 才受理回调(否则 503)。
+# 配齐(任一非空)并把 ENABLED=true 后,/api/v1/ad/pangle-callback 才受理回调(否则 503)。
+# 每个激励位 m-key 不同但共用同一回调 URL → 各位分开一行配(留空的忽略);验签逐个试、任一通过即接受。
 PANGLE_CALLBACK_ENABLED=false
+# 测试应用 激励位 104099649
+PANGLE_REWARD_SECRET_TEST=
+# 测试应用 专属激励位 104127529
+PANGLE_REWARD_SECRET_TEST_DEDICATED=
+# 正式应用 激励位 104099389
+PANGLE_REWARD_SECRET_PROD=
+# (旧用法,仍兼容:单个或逗号分隔的多个 m-key,会与上面三个合并去重)
 PANGLE_REWARD_SECRET=
 # ⚠️ 仅本地联调:true 时开放 POST /api/v1/ad/test-grant,让 debug 客户端看完广告直接发奖,
 # 验证"看广告→金币到账"全链路(未部署公网、穿山甲 S2S 打不到本地时用)。生产必须 false(绕过反作弊)。

app/admin/routers/config.py

@@ -43,6 +43,9 @@ def _validate(key: str, value: Any) -> None:
             for k, v in value.items()
         ):
             raise ValueError("需为 {字符串: 整数} 映射")
+    elif t == "bool":
+        if not isinstance(value, bool):
+            raise ValueError("需为布尔值")
 
 
 def _item(db, key: str) -> ConfigItemOut:

app/admin/routers/withdraw.py

@@ -34,6 +34,7 @@ from app.admin.schemas.wallet import (
 from app.core.config import settings
 from app.integrations import wxpay
 from app.models.admin import AdminUser
+from app.repositories import app_config
 from app.repositories import wallet as wallet_repo
 
 router = APIRouter(
@@ -95,7 +96,7 @@ def withdraws_summary(db: AdminDb) -> WithdrawSummaryOut:
     summary="提现配置健康检查",
     dependencies=[Depends(require_role("finance"))],  # 暴露密钥路径/配置,限财务+super
 )
-def withdraw_health_check() -> WxpayHealthCheckOut:
+def withdraw_health_check(db: AdminDb) -> WxpayHealthCheckOut:
     private_path = wxpay._resolve_config_path(settings.WXPAY_MCH_PRIVATE_KEY_PATH)  # noqa: SLF001
     public_path = wxpay._resolve_config_path(settings.WXPAY_PUBLIC_KEY_PATH)  # noqa: SLF001
     issues: list[str] = []
@@ -117,8 +118,15 @@ def withdraw_health_check() -> WxpayHealthCheckOut:
         issues.append("微信支付基础配置不完整")
     if not settings.WXPAY_AUTH_NOTIFY_URL:
         issues.append("免确认授权回调地址未配置")
-    if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
+
-        issues.append("自动对账未开启")
+    # 实际是否自动对账 = env 部署总闸(worker 起没起)AND 运营后台 DB 开关(本轮跑不跑)。
+    worker_running = settings.WITHDRAW_AUTO_RECONCILE_ENABLED
+    daily_on = bool(app_config.get_value(db, "withdraw_auto_reconcile_enabled"))
+    auto_reconcile_enabled = worker_running and daily_on
+    if not worker_running:
+        issues.append("自动对账 worker 未启动(部署侧 env WITHDRAW_AUTO_RECONCILE_ENABLED=false)")
+    elif not daily_on:
+        issues.append("自动对账运营开关已关闭(系统配置页可开)")
 
     return WxpayHealthCheckOut(
         ok=not issues,
@@ -131,7 +139,7 @@ def withdraw_health_check() -> WxpayHealthCheckOut:
         public_key_exists=public_path.exists(),
         public_key_loadable=public_loadable,
         auth_notify_url_configured=bool(settings.WXPAY_AUTH_NOTIFY_URL),
-        auto_reconcile_enabled=settings.WITHDRAW_AUTO_RECONCILE_ENABLED,
+        auto_reconcile_enabled=auto_reconcile_enabled,
         auto_reconcile_interval_sec=settings.WITHDRAW_AUTO_RECONCILE_INTERVAL_SEC,
         auto_reconcile_older_than_minutes=settings.WITHDRAW_AUTO_RECONCILE_OLDER_THAN_MINUTES,
         issues=issues,

app/admin/schemas/config.py

@@ -10,7 +10,7 @@ class ConfigItemOut(BaseModel):
     key: str
     label: str
     group: str
-    type: str  # int / int_list / dict_str_int
+    type: str  # int / int_list / dict_str_int / bool
     help: str | None = None
     default: Any
     value: Any

app/api/v1/ad.py

@@ -87,7 +87,7 @@ def pangle_callback(request: Request, db: DbSession) -> PangleCallbackOut:
 
     params = dict(request.query_params)
 
-    if not pangle.verify_callback_sign(params, settings.PANGLE_REWARD_SECRET):
+    if not pangle.verify_callback_sign_any(params, settings.pangle_reward_secrets):
         logger.warning("pangle callback bad sign trans_id=%s", params.get("trans_id"))
         raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="bad sign")
 

app/api/v1/platform.py

@@ -2,6 +2,7 @@
 
 路由前缀 `/api/v1/platform`:
   GET /stats   首页三统计(帮助用户 / 完成比价 / 累计节省),按运营后台配的模式算。
+  GET /flags   客户端运营 feature flag(比价/领券期广告开关等),客户端拉取后缓存。
 
 展示模式(real/manual/random,每指标独立)与计算逻辑见 app/repositories/ops_stat.py。
 """
@@ -12,9 +13,15 @@ import logging
 from fastapi import APIRouter, Query
 
 from app.api.deps import DbSession
+from app.repositories import app_config
 from app.repositories import ops_marquee as marquee_crud
 from app.repositories import ops_stat as crud
-from app.schemas.platform import PlatformStatsOut, SavingsFeedItem, SavingsFeedOut
+from app.schemas.platform import (
+    AppFlagsOut,
+    PlatformStatsOut,
+    SavingsFeedItem,
+    SavingsFeedOut,
+)
 
 logger = logging.getLogger("shagua.platform")
 
@@ -35,3 +42,12 @@ def stats(db: DbSession) -> PlatformStatsOut:
 def savings_feed(db: DbSession, limit: int = Query(8, ge=1, le=30)) -> SavingsFeedOut:
     items = marquee_crud.get_feed(db, limit=limit)
     return SavingsFeedOut(items=[SavingsFeedItem(**it) for it in items])
+
+
+@router.get("/flags", response_model=AppFlagsOut, summary="客户端运营 feature flag(不鉴权)")
+def flags(db: DbSession) -> AppFlagsOut:
+    """客户端拉取运营开关并缓存(app 启动 / 每场比价开始时刷新)。不鉴权:开关非敏感,
+    且比价无障碍服务取值时未必有登录态。值来自 app_config(admin 可改),空库回退默认。"""
+    return AppFlagsOut(
+        comparing_ad_enabled=bool(app_config.get_value(db, "comparing_ad_enabled")),
+    )

app/core/config.py

@@ -136,8 +136,15 @@ class Settings(BaseSettings):
 
     # ===== 穿山甲激励视频(服务端发奖回调)=====
     # 看完激励视频后穿山甲服务器回调本服务发金币(S2S,客户端被破解也刷不到)。
-    # PANGLE_REWARD_SECRET 是穿山甲后台配置的"奖励校验密钥",验签用,从后台取到后填 .env。
+    # 穿山甲后台配置的"奖励校验密钥"(m-key),验签用。每个 GroMore 广告位 m-key 不同(后台各自
+    # 生成),但共用同一回调 URL → 多个位的 m-key 都要配上。验签时所有非空 m-key 逐个试、任一通过
+    # 即接受(见 pangle.verify_callback_sign_any / 下面的 pangle_reward_secrets)。
     PANGLE_CALLBACK_ENABLED: bool = False
+    # 推荐:每个激励位的 m-key 分开一行配,清晰不混淆(留空的忽略)。
+    PANGLE_REWARD_SECRET_TEST: str = ""            # 测试应用 激励位 104099649
+    PANGLE_REWARD_SECRET_TEST_DEDICATED: str = ""  # 测试应用 专属激励位 104127529
+    PANGLE_REWARD_SECRET_PROD: str = ""            # 正式应用 激励位 104099389
+    # 旧用法:单个或逗号分隔的多个 m-key,仍兼容(会与上面三个命名项合并去重)。
     PANGLE_REWARD_SECRET: str = ""
 
     # ⚠️ 仅本地联调:打开后开放 POST /api/v1/ad/test-grant,让(已登录的)客户端在没部署公网、
@@ -145,10 +152,23 @@ class Settings(BaseSettings):
     # 它让客户端能自助发奖 = 绕过反作弊,**生产必须保持 False**(默认 False;只在本地 .env 设 true)。
     AD_REWARD_TEST_GRANT_ENABLED: bool = False
 
+    @property
+    def pangle_reward_secrets(self) -> list[str]:
+        """汇总所有 m-key 成列表(去空白、去空项、去重保序)。验签时逐个试、任一通过即接受
+        (见 pangle.verify_callback_sign_any)。来源可混用:三个命名项 + 旧的逗号分隔 PANGLE_REWARD_SECRET。"""
+        raw = [
+            *self.PANGLE_REWARD_SECRET.split(","),
+            self.PANGLE_REWARD_SECRET_TEST,
+            self.PANGLE_REWARD_SECRET_TEST_DEDICATED,
+            self.PANGLE_REWARD_SECRET_PROD,
+        ]
+        cleaned = [s.strip() for s in raw if s and s.strip()]
+        return list(dict.fromkeys(cleaned))  # 去重保序
+
     @property
     def pangle_callback_configured(self) -> bool:
-        """回调开关打开且验签密钥已配,才接受发奖回调。"""
+        """回调开关打开且至少配了一个验签密钥,才接受发奖回调。"""
-        return bool(self.PANGLE_CALLBACK_ENABLED and self.PANGLE_REWARD_SECRET)
+        return bool(self.PANGLE_CALLBACK_ENABLED and self.pangle_reward_secrets)
 
     # ===== Pricebot 上游 (领券/比价业务透传目标) =====
     # pricebot-backend 默认跑在 8000。/api/v1/coupon/step 会透传到这里的 /api/coupon/step

app/core/config_schema.py

@@ -11,7 +11,7 @@ from typing import Any
 
 from app.core import rewards as r
 
-# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int
+# type 约定(给前端渲染编辑控件用):int / int_list / dict_str_int / bool
 CONFIG_DEFS: dict[str, dict[str, Any]] = {
     "signin_rewards": {
         "default": list(r.SIGNIN_REWARDS), "label": "签到 7 天金币档位",
@@ -65,4 +65,23 @@ CONFIG_DEFS: dict[str, dict[str, Any]] = {
         "group": "签到", "type": "int",
         "help": "Day1-Day6 签到后看完激励视频额外发放的固定金币;Day7 不展示也不允许膨胀。",
     },
+    "comparing_ad_enabled": {
+        "default": True, "label": "比价/领券期信息流广告",
+        "group": "看广告", "type": "bool",
+        "help": (
+            "开启后,比价进行中 + 领券等候期会在悬浮窗展示穿山甲信息流广告(变现行为);"
+            "关闭则全程不出广告。客户端按 app 启动 / 每场比价开始时拉取并缓存,故为「最终一致」的"
+            "远程开关(下一场比价生效),用于出问题时无需发版即可快速止血。debug 包可用本地开关覆盖。"
+        ),
+    },
+    "withdraw_auto_reconcile_enabled": {
+        "default": True, "label": "提现自动对账",
+        "group": "钱包", "type": "bool",
+        "help": (
+            "开启后后台 worker 每隔一段时间自动扫描超时仍「打款中」的提现单并归一化"
+            "(查微信/退款/撤单)。需部署侧 env WITHDRAW_AUTO_RECONCILE_ENABLED=true 启动 worker "
+            "进程后此开关才起效;扫描间隔/超时阈值仍由 env 控制。关掉只停自动扫描,"
+            "提现页「批量对账」手动按钮不受影响。"
+        ),
+    },
 }

app/core/withdraw_reconcile_worker.py

@@ -14,8 +14,11 @@ from sqlalchemy.exc import SQLAlchemyError
 from app.core.config import settings
 from app.db.session import SessionLocal
 from app.integrations.wxpay import WxPayNotConfiguredError
+from app.repositories import app_config
 from app.repositories import wallet as wallet_repo
 
+_AUTO_RECONCILE_KEY = "withdraw_auto_reconcile_enabled"
+
 logger = logging.getLogger("shagua.withdraw_reconcile")
 _LOCK_PATH = Path(__file__).resolve().parents[2] / "data" / "withdraw_reconcile.lock"
 
@@ -59,8 +62,15 @@ def _single_instance_lock(stale_after_sec: int) -> Iterator[bool]:
                 _LOCK_PATH.unlink()
 
 
-def _reconcile_once(older_than_minutes: int) -> dict:
+def _reconcile_once(older_than_minutes: int) -> dict | None:
+    """读运营开关(app_config):关着返回 None(本轮跳过),开着才真扫单。
+
+    env WITHDRAW_AUTO_RECONCILE_ENABLED 是部署级总闸(决定 worker 起不起);
+    这里的 DB 开关是运营级日常开关,后台一改下一轮即生效、跨进程一致、无需重启。
+    """
     with SessionLocal() as db:
+        if not app_config.get_value(db, _AUTO_RECONCILE_KEY):
+            return None
         return wallet_repo.reconcile_pending_withdraws(db, older_than_minutes=older_than_minutes)
 
 
@@ -78,16 +88,18 @@ async def _run_loop() -> None:
 
 async def _run_locked_loop(interval: int, older_than: int) -> None:
     logger.info(
-        "withdraw auto reconcile started interval=%ss older_than=%sm",
+        "withdraw auto reconcile worker started interval=%ss older_than=%sm "
+        "(runtime on/off via app_config '%s')",
         interval,
         older_than,
+        _AUTO_RECONCILE_KEY,
     )
     try:
         while True:
             try:
                 _touch_lock()
                 result = await asyncio.to_thread(_reconcile_once, older_than)
-                if result["checked"] or result["resolved"]:
+                if result is not None and (result["checked"] or result["resolved"]):
                     logger.info("withdraw auto reconcile result=%s", result)
             except WxPayNotConfiguredError:
                 logger.warning("withdraw auto reconcile skipped: wxpay not configured")
@@ -103,7 +115,8 @@ async def _run_locked_loop(interval: int, older_than: int) -> None:
 
 def start_withdraw_reconcile_worker() -> asyncio.Task | None:
     if not settings.WITHDRAW_AUTO_RECONCILE_ENABLED:
-        logger.info("withdraw auto reconcile disabled")
+        # 部署级总闸关:worker 进程不启动(运营后台开关此时无效,需先在 env 打开)。
+        logger.info("withdraw auto reconcile worker not started (env master switch off)")
         return None
     if not settings.wxpay_configured:
         logger.warning("withdraw auto reconcile enabled but wxpay not configured")

app/integrations/cities.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+"""美团城市字典(地级市)读取。
+
+数据源:app/integrations/data/meituan_cities.json,由 tools/gen_meituan_cities.py 从
+美团官方「城市字典」Excel 生成(359 个地级市:city_id / 城市名 / 省份)。
+实测一个地级市 cityId 已覆盖其下辖县级市供给(徐州 → 邳州 / 新沂 / 睢宁 …),故无区县层级。
+
+- ETL(scripts/pull_meituan_coupons.py)遍历 all_cities() 全量抓取入库。
+- 读取侧将来「按城市过滤」时,也从这里取 city_id ↔ 城市名映射。
+"""
+from __future__ import annotations
+
+import functools
+import json
+from pathlib import Path
+
+_JSON = Path(__file__).resolve().parent / "data" / "meituan_cities.json"
+
+
+@functools.lru_cache(maxsize=1)
+def all_cities() -> list[dict]:
+    """全部城市 [{city_id, name, province}, ...](359 个地级市)。"""
+    return json.loads(_JSON.read_text(encoding="utf-8"))
+
+
+@functools.lru_cache(maxsize=1)
+def _by_id() -> dict[str, dict]:
+    return {c["city_id"]: c for c in all_cities()}
+
+
+def city_name(city_id: str) -> str | None:
+    """city_id → 城市名(查不到返回 None)。"""
+    c = _by_id().get(city_id)
+    return c["name"] if c else None

app/integrations/pangle.py

@@ -42,3 +42,13 @@ def verify_callback_sign(params: dict[str, str], secret: str) -> bool:
     got = params.get("sign") or ""
     expect = build_sign(trans_id, secret)
     return hmac.compare_digest(got, expect)
+
+
+def verify_callback_sign_any(params: dict[str, str], secrets: list[str]) -> bool:
+    """对**多个** m-key 逐个验签,任一通过即接受。
+
+    多广告位场景:每个 GroMore 广告位的 m-key 由后台各自生成、互不相同,但本服务用
+    同一个回调 URL 接所有位的回调。配置里放多个密钥(逗号分隔),回调到达时挨个试。
+    仍然安全:伪造者必须知道其中某个 m-key 才能算出合法 sign;空列表 → 一律失败。
+    """
+    return any(verify_callback_sign(params, s) for s in secrets if s)

app/schemas/platform.py

@@ -22,3 +22,9 @@ class SavingsFeedItem(BaseModel):
 
 class SavingsFeedOut(BaseModel):
     items: list[SavingsFeedItem]
+
+
+class AppFlagsOut(BaseModel):
+    """客户端拉取的运营 feature flag(不鉴权,登录前也能拉)。客户端缓存后按需读。"""
+
+    comparing_ad_enabled: bool  # 比价/领券期是否展示信息流广告(远程 kill-switch)

docs/api/ad-pangle-callback.md

@@ -22,7 +22,7 @@ GroMore 以 GET 回调,关键参数:
 | `ecpm` | string | 本次广告 eCPM（同上,可用于收益分析） |
 | `sign` | string | 签名,见下 |
 
-**验签**:`sign = SHA256("{m-key}:{trans_id}")` 十六进制（只签 `trans_id`，其余参数不参与）。算法细节、m-key 来源、为什么这样设计 → 见集成文档 [integrations/pangle](../integrations/pangle.md)。
+**验签**:`sign = SHA256("{m-key}:{trans_id}")` 十六进制（只签 `trans_id`，其余参数不参与）。多激励位共用同一回调 URL → 服务端把各位的 m-key 都配上,`verify_callback_sign_any` 逐个试、任一过即接受。算法细节、m-key 配置项(`PANGLE_REWARD_SECRET_TEST/_TEST_DEDICATED/_PROD`)、为什么这样设计 → 见集成文档 [integrations/pangle](../integrations/pangle.md)。
 
 ## 出参
 响应 `200`,**响应体必须是 `{"is_verify": bool, "reason": int}`**（GroMore 规范）。

docs/database/app_config.md

@@ -14,8 +14,8 @@
 ## 字段
 | 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
 |---|---|---|---|
-| `key` | String(64) | **PK** | 配置标识,取值见 `config_schema.CONFIG_DEFS`:`signin_rewards` / `min_exchange_coin` / `withdraw_min_cents` / `withdraw_max_cents` / `task_rewards` / `record_milestones` / `ad_reward_coin` / `ad_daily_limit` / `ad_max_coin` / `ad_round_count` / `ad_cooldown_sec` |
+| `key` | String(64) | **PK** | 配置标识,取值见 `config_schema.CONFIG_DEFS`:`signin_rewards` / `min_exchange_coin` / `withdraw_min_cents` / `withdraw_max_cents` / `task_rewards` / `record_milestones` / `ad_reward_coin` / `ad_daily_limit` / `ad_max_coin` / `ad_round_count` / `ad_cooldown_sec` / `signin_boost_coin` / `withdraw_auto_reconcile_enabled` |
-| `value` | JSON(PG: JSONB) | NOT NULL | 配置值,类型随 key(`int` / `int_list` 如签到 14 档 / `dict_str_int` 如 task_rewards) |
+| `value` | JSON(PG: JSONB) | NOT NULL | 配置值,类型随 key(`int` / `int_list` 如签到 14 档 / `dict_str_int` 如 task_rewards / `bool` 如 withdraw_auto_reconcile_enabled) |
 | `updated_by_admin_id` | Integer | nullable | 最后修改的管理员 id(= `admin_user.id`,软引用,无 FK) |
 | `updated_at` | DateTime(tz) | server_default now(), onupdate now() | 最后修改时间 |
 

docs/integrations/pangle.md

@@ -16,12 +16,18 @@
 | 函数 | 说明 |
 |---|---|
 | `build_sign(trans_id, secret) -> str` | 计算 `SHA256("{secret}:{trans_id}")` hex。自验签测试 / 模拟回调脚本共用，保证两端一致 |
-| `verify_callback_sign(params, secret) -> bool` | 校验回调签名。密钥空 / 缺 `trans_id` 一律失败；比较用 `hmac.compare_digest` 定长比较防时序侧信道 |
+| `verify_callback_sign(params, secret) -> bool` | 用**单个** m-key 校验回调签名。密钥空 / 缺 `trans_id` 一律失败；比较用 `hmac.compare_digest` 定长比较防时序侧信道 |
+| `verify_callback_sign_any(params, secrets) -> bool` | 用**一组** m-key 逐个验签、任一通过即接受(多激励位共用同一回调 URL 时用)。空列表 → 失败 |
+
+## 配置(多激励位 = 多 m-key)
+每个 GroMore 激励位的 m-key 由后台各自生成、互不相同,但本服务用**同一个回调 URL** 接所有位的回调,因此把用到的位的 m-key 都配上;验签时 `verify_callback_sign_any` 逐个试、任一过即接受(仍安全:伪造者须知道其中某个 m-key 才能造出合法 sign)。m-key 在后台「GroMore 聚合管理 → 搜广告位 ID → 编辑」处获取。`settings.pangle_reward_secrets` 把下列来源汇总去重:
 
-## 配置
 | 配置项 | 说明 |
 |---|---|
-| `PANGLE_REWARD_SECRET` | m-key（安全密钥）。后台「GroMore 聚合管理 → 搜广告位 ID → 编辑」处获取 |
+| `PANGLE_REWARD_SECRET_TEST` | 测试应用 激励位 104099649 的 m-key |
+| `PANGLE_REWARD_SECRET_TEST_DEDICATED` | 测试应用 专属激励位 104127529 的 m-key |
+| `PANGLE_REWARD_SECRET_PROD` | 正式应用 激励位 104099389 的 m-key |
+| `PANGLE_REWARD_SECRET` | 旧用法:单个或逗号分隔多个 m-key,仍兼容(与上面三个命名项合并去重) |
 
 ## 踩坑
 - **别用联盟代码位那套**：联盟代码位层级用的是另一套 Security Key + `isValid` 响应体，与 GroMore 广告位层级不通用。我们走 GroMore，别接错。

docs/看广告赚金币上线清单.md

@@ -41,18 +41,31 @@
 > 已确认走 **GroMore 广告位层级**回调(客户端 useMediation(true);规范 supportcenter/26240),
 > **不是**联盟代码位层级(5416)。验签算法 / 响应格式 / reward_amount 解析**代码已按 GroMore 规范实现**
 > (2026-05-27),剩下的是后台配置 + 填密钥。
+>
+> 🔴 **2026-06-16 实测的头号阻塞:穿山甲打线上后端会超时**。真机看广告时 SDK 日志
+> `onRewardArrived valid=false errorCode=50002`,errorMsg =
+> `Get "https://app-api.shaguabijia.com/api/v1/ad/pangle-callback...": context deadline exceeded`。
+> 而我们自己 curl 线上是秒回。说明**穿山甲机房到线上后端这条网络路径有问题**(疑似 nginx/云防火墙对
+> 机房/境外 IP 限流或拦截)。**m-key 配对了也救不了——穿山甲根本没把请求送达。** 上线前必须单独验
+> 「线上 `app-api.shaguabijia.com` 对穿山甲机房可达且响应够快」,否则用户看完发不了奖。
+>
+> ✅ **多激励位已支持**(2026-06-16):内部测试期 测试位/专属位/正式位 并存,m-key 各配一行
+> (`PANGLE_REWARD_SECRET_TEST/_TEST_DEDICATED/_PROD`),`verify_callback_sign_any` 逐个试。本地已验通。
 
 - [ ] **GroMore 后台配回调**:GroMore 聚合管理 → 搜广告位ID → 编辑 → 勾选「服务端激励回调」→
       回调 URL 填 `https://app-api.shaguabijia.com/api/v1/ad/pangle-callback`(用域名,别用 IP)。
       ⚠️ 广告位层级配了就**别再在代码位层级重复配**(会导致发奖出问题)。
-- [ ] **拿 m-key(安全密钥)**:就在上面"编辑广告位"页获取,填到生产 `.env` 的 `PANGLE_REWARD_SECRET`。
+- [ ] **拿 m-key(安全密钥)**:就在上面"编辑广告位"页获取(每个激励位各一把),填到生产 `.env` 的命名变量
+      (正式位填 `PANGLE_REWARD_SECRET_PROD`;内部测试期 测试位/专属位 也并存就填 `_TEST`/`_TEST_DEDICATED`)。
       (注:这是 GroMore 广告位的 m-key,**不是**联盟代码位那个 Security Key。)
 - [x] ~~换验签~~:`app/integrations/pangle.py` 已实现 `sign = SHA256("{m-key}:{trans_id}")`(GroMore 真实算法)。
 - [x] ~~响应格式~~:已返回 GroMore 要求的 `{"is_verify": bool, "reason": int}`。
 - [x] ~~reward_amount~~:回调按 `reward_amount` 发金币(`rewards.resolve_ad_reward_coin`,带回退/夹紧);
       **后台广告位"奖励数量"须配成与 `AD_REWARD_COIN`(=100)一致**,保证"广告内展示/进度预告/到账"三者一致。
 - [x] ~~透传 user_id~~:客户端已 `setUserID(userId)` + `setMediaExtra("uid:...")`。
-- [ ] 生产 `.env`:`PANGLE_CALLBACK_ENABLED=true` + `PANGLE_REWARD_SECRET=<m-key>`(配齐才不返 503)
+- [ ] 生产 `.env`:`PANGLE_CALLBACK_ENABLED=true` + 至少一个 m-key 命名变量(配齐才不返 503)+ 确认
+      `AD_REWARD_TEST_GRANT_ENABLED=false`(绕过反作弊,生产红线)。改 `.env` 后**重启 uvicorn**
+      (watchfiles 不监听 .env)。验证:外部 curl 回调 URL 从 `503` 变 `403 bad sign` 即生效。
 
 ## C. 部署 + 包名
 

scripts/pull_meituan_coupons.py

@@ -1,45 +1,74 @@
-"""美团 CPS 券定时抓取入库(北京试点)。
+"""美团 CPS 券定时抓取入库(全国 359 个地级市)。
 
-把 3 路券抓进 meituan_coupon 表,供「销量/佣金排序」从库里捞、本地排序,不再实时打美团:
+把每个城市的 3 路券抓进 meituan_coupon 表,供「智能推荐 / 销量最高」从库里捞、本地排序,
+不再实时打美团:
   1. search_waimai : 到家/外卖, 搜「外卖」          翻到尽头
   2. search_meishi : 到家/外卖, 搜「美食」          翻到尽头
   3. store_supply  : 到店, 多业务线供给(到餐+到综+酒店+门票) 翻到尽头
 
+城市来自 app/integrations/data/meituan_cities.json(美团官方城市字典,359 地级市)。
+实测一个地级市 cityId 已覆盖其下辖县级市(徐州 → 邳州/新沂…),故按地级市抓即可。
+
+并发:城市级并发(每城内部仍顺序跑 3 路),主线程串行入库(Session 不跨线程)。
+实测单城全量 ~110s/~2300 条;15 并发抓完全国一轮 ~50–60min,402 占比 ~3% 且退避全消化。
+mentor 要求每 3h 全量一轮,窗口充裕,故默认 12 并发 + 启动错峰,把瞬时峰值与 402 压更低。
+
 按 (source, product_view_sign) upsert 存最新态;last_seen 每轮刷新。带文件锁,防止
-上一轮没跑完下一轮又起(本地 5~10min、跨进程 cron 都安全)。
+上一轮没跑完下一轮又起。
 
 用法:
-    # 单轮(打通验证 / 给 cron 用,线上每 1h 一次)
+    # 单轮全量(给 cron 用,线上每 3h 一次)
     python -m scripts.pull_meituan_coupons --once
 
-    # 本地循环(默认每 10min 一轮)
+    # 常驻循环(每 3h 一轮)
-    python -m scripts.pull_meituan_coupons --loop --interval 600
+    python -m scripts.pull_meituan_coupons --loop --interval 10800
+
+    # 本地测试:只抓前 N 个城市 / 指定城市
+    python -m scripts.pull_meituan_coupons --once --limit 3
+    python -m scripts.pull_meituan_coupons --once --city-ids OCZOBCJDEXKE7KBN3BD7AYQG2Q
+
+部署(服务器):推荐 cron 跑 --once(每 3h),避免长驻进程孤儿:
+    0 */3 * * *  cd /path/to/app-server && .venv/bin/python -m scripts.pull_meituan_coupons --once >> data/etl.log 2>&1
 """
 from __future__ import annotations
 
 import argparse
 import hashlib
+import logging
 import os
 import re
 import sys
 import tempfile
+import threading
 import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta, timezone
 
-# Windows 控制台按 UTF-8 输出中文/¥
+# Windows 控制台按 UTF-8 输出中文/¥;line_buffering=True 让 print 每行即时 flush——
+# 否则 stdout 重定向到 cron/后台日志文件时是块缓冲,要攒到 ~4KB 或进程退出才落盘,
+# 常驻(--loop)时几乎看不到每轮进度。
 try:
-    sys.stdout.reconfigure(encoding="utf-8")  # type: ignore[attr-defined]
+    sys.stdout.reconfigure(encoding="utf-8", line_buffering=True)  # type: ignore[attr-defined]
 except Exception:  # noqa: BLE001
     pass
 
-from sqlalchemy import delete, func, select
+from sqlalchemy import delete, func, select  # noqa: E402
-from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.dialects.postgresql import insert as pg_insert  # noqa: E402
 
-from app.db.session import SessionLocal
+from app.db.session import SessionLocal, engine  # noqa: E402
-from app.integrations.meituan import MeituanCpsError, _call
+from app.integrations.cities import all_cities  # noqa: E402
-from app.models.meituan_coupon import MeituanCoupon
+from app.integrations.meituan import MeituanCpsError, _call  # noqa: E402
+from app.models.meituan_coupon import MeituanCoupon  # noqa: E402
+
+# dev 默认 create_engine(echo=True) 会逐条打 SQL,本脚本逐城 upsert 会把日志刷爆;
+# ETL 不需要 SQL 日志,显式关掉(入库不受影响;线上 prod 本就 echo=False)。
+engine.echo = False
+
+# 美团调用偶发错误(本机走代理高并发时的 SSL EOF / 上游 code=5 等)会被 meituan._call 的
+# logger.exception 打完整 traceback,并发抓取下单轮可刷数十 KB 日志。ETL 自己用 _STATS 统计
+# 「放弃页数」,无需逐条 traceback,故把美团 logger 压到 CRITICAL(线上直连少见此类错误)。
+logging.getLogger("shagua.meituan").setLevel(logging.CRITICAL)
 
-CITY_BEIJING = "WKV2HMXUEK634WP64CUCUQGM64"
 QUERY_PATH = "/cps_open/common/api/v1/query_coupon"
 PAGE_SIZE = 20
 MAX_PAGES = 80          # 单路安全上限(搜索 ~52 页、供给 ~70 页)
@@ -49,7 +78,12 @@ RETRY = 7
 # 无 sudo 部署时 data/ 常属 root,cps 写不了会导致每轮 PermissionError、cron 抓不进数据。
 # 需要指定位置时用环境变量 MEITUAN_ETL_LOCK 覆盖。
 LOCK_FILE = os.environ.get("MEITUAN_ETL_LOCK") or os.path.join(tempfile.gettempdir(), "meituan_etl.lock")
-LOCK_STALE_SEC = 30 * 60   # 锁超过 30min 视为陈旧(进程异常退出残留),自动接管
+# 多城全量一轮 ~50–60min,锁陈旧阈值放宽到 90min,避免把「正在跑的轮次」误判为残留而接管。
+LOCK_STALE_SEC = 90 * 60
+
+DEFAULT_CONCURRENCY = 12   # 并发城市数(实测 15 并发 402 占 3% 可退避消化;12 更稳,3h 窗口充裕)
+STARTUP_STAGGER = 0.3      # 首批城市启动错峰间隔秒(削平瞬时峰值,实测能压低 402)
+PRUNE_FAIL_RATIO_MAX = 0.05  # 失败城占比超此值则本轮跳过 prune(避免大面积抓取失败误删库)
 
 SOURCES = [
     {"code": "search_waimai", "label": "外卖·搜外卖", "kind": "search", "platform": 1, "keyword": "外卖"},
@@ -58,33 +92,44 @@ SOURCES = [
      "platform": 2, "biz_lines": [1, 2, 3, 4]},
 ]
 
+# 跨线程统计(并发抓取下汇总请求量 / 402 / 放弃页数,供观测限流)
+_STATS_LOCK = threading.Lock()
+_STATS = {"req": 0, "r402": 0, "err": 0}
+
+
+def _bump(key: str) -> None:
+    with _STATS_LOCK:
+        _STATS[key] += 1
+
 
 # ───────────────────────── 美团调用 ─────────────────────────
 
 def _call_retry(body: dict) -> dict | None:
-    """打美团,402/频繁退避重试;其它错误打印并放弃本页。"""
+    """打美团,402/频繁退避重试;其它错误放弃本页。并发下不逐条 print(避免刷屏),计入 _STATS。"""
     for a in range(RETRY):
+        _bump("req")
         try:
             return _call(QUERY_PATH, body)
         except MeituanCpsError as e:
             msg = str(e)
             if "402" in msg or "频繁" in msg:
+                _bump("r402")
                 time.sleep(2.5 * (a + 1))
                 continue
-            print(f"   [warn] meituan: {msg[:80]}")
+            _bump("err")
             return None
-        except Exception as e:  # noqa: BLE001
+        except Exception:  # noqa: BLE001
-            print(f"   [warn] {type(e).__name__}: {str(e)[:60]}")
             time.sleep(2.0 * (a + 1))
+    _bump("err")
     return None
 
 
-def _pull_search(platform: int, keyword: str) -> list[dict]:
+def _pull_search(city_id: str, platform: int, keyword: str) -> list[dict]:
     rows: list[dict] = []
     sid = None
     pg = 1
     while pg <= MAX_PAGES:
-        body = {"platform": platform, "searchText": keyword, "cityId": CITY_BEIJING, "pageSize": PAGE_SIZE}
+        body = {"platform": platform, "searchText": keyword, "cityId": city_id, "pageSize": PAGE_SIZE}
         if sid:
             body["searchId"] = sid
         else:
@@ -102,14 +147,14 @@ def _pull_search(platform: int, keyword: str) -> list[dict]:
     return rows
 
 
-def _pull_supply(platform: int, biz_lines: list[int]) -> list[dict]:
+def _pull_supply(city_id: str, platform: int, biz_lines: list[int]) -> list[dict]:
     rows: list[dict] = []
     sid = None
     biz_param = [{"bizLine": b} for b in biz_lines]
     for _ in range(MAX_PAGES):
         body = {
             "multipleSupplyList": [{"platform": platform, "bizLineParamList": biz_param}],
-            "cityId": CITY_BEIJING,
+            "cityId": city_id,
             "sortField": 2,  # 供给查询 sortField 必填;我们入库后本地再排,这里给个默认
             "pageSize": PAGE_SIZE,
         }
@@ -157,7 +202,19 @@ def _to_cents(yuan) -> int | None:
         return None
 
 
-def _parse_item(item: dict, source: dict) -> dict | None:
+def _strip_nul(v):
+    """递归去掉字符串里的 NUL(0x00):PostgreSQL 的 text / jsonb 字段都不接受该字节,
+    美团偶有脏数据(某券文本含 NUL)会让整批 upsert 抛 DataError。入库前统一清洗。"""
+    if isinstance(v, str):
+        return v.replace(chr(0), "")
+    if isinstance(v, dict):
+        return {k: _strip_nul(x) for k, x in v.items()}
+    if isinstance(v, list):
+        return [_strip_nul(x) for x in v]
+    return v
+
+
+def _parse_item(item: dict, source: dict, city_id: str) -> dict | None:
     cpd = item.get("couponPackDetail") or {}
     br = item.get("brandInfo") or {}
     ci = item.get("commissionInfo") or {}
@@ -186,11 +243,12 @@ def _parse_item(item: dict, source: dict) -> dict | None:
         dist = None
 
     dedup_raw = f"{brand}|{name}|{price_cents}"
-    return {
+    # 入库前递归清洗 NUL(美团脏数据偶含 0x00,PostgreSQL text/jsonb 拒绝整批 → DataError)
+    return _strip_nul({
         "source": source["code"],
         "platform": source["platform"],
         "biz_line": item.get("bizLine") or cpd.get("bizLine"),
-        "city_id": CITY_BEIJING,
+        "city_id": city_id,
         "product_view_sign": str(sign)[:128],
         "sku_view_id": cpd.get("skuViewId"),
         "name": (name[:256] or None),
@@ -207,7 +265,28 @@ def _parse_item(item: dict, source: dict) -> dict | None:
         "delivery_distance_m": dist,
         "dedup_key": hashlib.md5(dedup_raw.encode("utf-8")).hexdigest(),
         "raw": item,
-    }
+    })
+
+
+def _pull_one_city(city: dict, index: int, concurrency: int, stagger: float) -> list[dict]:
+    """抓单个城市的 3 路券并解析。worker 线程内执行(只抓取+解析,不碰 DB)。
+
+    启动错峰:首批并发的 `concurrency` 个城市按 index 错开首请求,削平瞬时峰值(压低 402)。
+    """
+    if stagger:
+        time.sleep((index % concurrency) * stagger)
+    cid = city["city_id"]
+    parsed: list[dict] = []
+    for src in SOURCES:
+        if src["kind"] == "search":
+            items = _pull_search(cid, src["platform"], src["keyword"])
+        else:
+            items = _pull_supply(cid, src["platform"], src["biz_lines"])
+        for it in items:
+            p = _parse_item(it, src, cid)
+            if p:
+                parsed.append(p)
+    return parsed
 
 
 # ───────────────────────── 入库(upsert) ─────────────────────────
@@ -272,30 +351,59 @@ def _release_lock() -> None:
 
 # ───────────────────────── 主流程 ─────────────────────────
 
-def run_once(prune_hours: int = 24) -> None:
+def run_once(
+    prune_hours: int = 24,
+    concurrency: int = DEFAULT_CONCURRENCY,
+    stagger: float = STARTUP_STAGGER,
+    cities: list[dict] | None = None,
+) -> None:
     if not _acquire_lock():
         print(f"[{datetime.now():%H:%M:%S}] 上一轮还在跑(锁占用),跳过本轮")
         return
     t0 = time.time()
     now = datetime.now(timezone.utc)
+    with _STATS_LOCK:
+        _STATS.update(req=0, r402=0, err=0)
+    cities = cities if cities is not None else all_cities()
+    n_city = len(cities)
     db = SessionLocal()
     try:
-        total = 0
+        total_up = 0
-        for src in SOURCES:
+        done = 0
-            ts = time.time()
+        fails: list[str] = []
-            if src["kind"] == "search":
+        # 城市级并发抓取(worker 只抓取+解析),主线程逐城串行入库(Session 不跨线程)。
-                items = _pull_search(src["platform"], src["keyword"])
+        with ThreadPoolExecutor(max_workers=concurrency) as pool:
-            else:
+            futs = {
-                items = _pull_supply(src["platform"], src["biz_lines"])
+                pool.submit(_pull_one_city, city, i, concurrency, stagger): city
-            parsed = [p for p in (_parse_item(it, src) for it in items) if p]
+                for i, city in enumerate(cities)
-            up, dup = _upsert(db, parsed, now)
+            }
-            total += up
+            for fut in as_completed(futs):
-            print(f"  {src['label']:18} 抓{len(items):5} 解析{len(parsed):5} "
+                city = futs[fut]
-                  f"入库{up:5} (本源去重{dup:3}) {time.time() - ts:4.0f}s")
+                try:
-        # 清理长期未再出现的陈旧券(美团 sign 轮换 / 券下架后的残留),默认 24h 宽限。
+                    parsed = fut.result()
-        # 护栏:仅在本轮确有入库(total>0)时才清理。美团整体故障时本轮可能 0 入库
+                except Exception as e:  # noqa: BLE001
-        # (脚本不抛异常、只是抓回空),若仍照常 prune,连续故障会按 last_seen 把全表删空。
+                    fails.append(city["name"])
-        if prune_hours and prune_hours > 0 and total > 0:
+                    print(f"  [fail] {city['name']}: {type(e).__name__}: {str(e)[:60]}")
+                    continue
+                try:
+                    up, _dup = _upsert(db, parsed, now)
+                except Exception as e:  # noqa: BLE001
+                    db.rollback()  # 事务已 abort,必须 rollback 才能继续给下一城复用 Session
+                    fails.append(city["name"])
+                    print(f"  [入库失败] {city['name']}: {type(e).__name__}: {str(e)[:80]}")
+                    continue
+                total_up += up
+                done += 1
+                if done % 20 == 0 or done == n_city:
+                    print(f"  [{done}/{n_city}] {city['name']:10} 抓{len(parsed):5} 入库{up:5} "
+                          f"(累计入库 {total_up}, 用时 {time.time() - t0:.0f}s)")
+
+        # 清理陈旧券(美团 sign 轮换 / 券下架残留),默认 24h 宽限。两道护栏防误删:
+        #  ① total_up>0:本轮 0 入库(疑似上游整体故障,脚本不抛异常只抓回空)时跳过,否则
+        #     会按 last_seen 把全表删空;
+        #  ② 失败城占比 ≤5%:大面积城市抓取失败(限流/网络)时跳过,避免误删还在架上的券。
+        fail_ratio = len(fails) / max(1, n_city)
+        if prune_hours and prune_hours > 0 and total_up > 0 and fail_ratio <= PRUNE_FAIL_RATIO_MAX:
             cutoff = now - timedelta(hours=prune_hours)
             pruned = db.execute(
                 delete(MeituanCoupon).where(MeituanCoupon.last_seen < cutoff)
@@ -303,36 +411,62 @@ def run_once(prune_hours: int = 24) -> None:
             db.commit()
             if pruned:
                 print(f"  清理陈旧券(>{prune_hours}h 未再出现): {pruned} 条")
-        elif prune_hours and prune_hours > 0 and total == 0:
+        elif prune_hours and prune_hours > 0:
-            print("  本轮 0 入库(疑似上游故障),跳过清理以防误删全表")
+            reason = "本轮 0 入库" if total_up == 0 else f"失败城占比 {fail_ratio:.0%}>{PRUNE_FAIL_RATIO_MAX:.0%}"
+            print(f"  跳过 prune({reason},防误删)")
+
         cnt = db.execute(select(func.count()).select_from(MeituanCoupon)).scalar()
-        print(f"[{datetime.now():%H:%M:%S}] 本轮完成: 入库 {total} 条, 表总计 {cnt} 行, "
+        with _STATS_LOCK:
-              f"用时 {time.time() - t0:.0f}s")
+            req, r402, err = _STATS["req"], _STATS["r402"], _STATS["err"]
+        fail_tail = (": " + ",".join(fails[:10])) if fails else ""
+        print(f"[{datetime.now():%H:%M:%S}] 本轮完成: {done}/{n_city} 城, 入库 {total_up} 条, "
+              f"表总计 {cnt} 行, 请求 {req}(402 {r402} / 放弃 {err}), "
+              f"失败城 {len(fails)}{fail_tail}, 用时 {time.time() - t0:.0f}s")
     finally:
         db.close()
         _release_lock()
 
 
+def _select_cities(limit: int, city_ids: str) -> list[dict]:
+    cities = all_cities()
+    if city_ids.strip():
+        want = {c.strip() for c in city_ids.split(",") if c.strip()}
+        return [c for c in cities if c["city_id"] in want]
+    if limit and limit > 0:
+        return cities[:limit]
+    return cities
+
+
 def main() -> None:
-    ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库")
+    ap = argparse.ArgumentParser(description="美团 CPS 券定时抓取入库(全国 359 地级市)")
     ap.add_argument("--once", action="store_true", help="只跑一轮(默认)")
     ap.add_argument("--loop", action="store_true", help="循环跑")
-    ap.add_argument("--interval", type=int, default=600, help="循环间隔秒(默认 600=10min)")
+    ap.add_argument("--interval", type=int, default=10800,
+                    help="循环间隔秒(默认 10800=3h,对齐每 3h 全量一轮)")
     ap.add_argument("--prune-hours", type=int, default=24,
                     help="清理超过 N 小时未再出现的陈旧券(默认 24;0=不清理)")
+    ap.add_argument("--concurrency", type=int, default=DEFAULT_CONCURRENCY,
+                    help=f"并发城市数(默认 {DEFAULT_CONCURRENCY};实测 15 并发 402 占 3%% 可退避消化)")
+    ap.add_argument("--stagger", type=float, default=STARTUP_STAGGER,
+                    help=f"首批城市启动错峰间隔秒(默认 {STARTUP_STAGGER};削平瞬时峰值压低 402)")
+    ap.add_argument("--limit", type=int, default=0, help="只抓前 N 个城市(测试用,0=全部)")
+    ap.add_argument("--city-ids", default="", help="只抓这些 cityId(逗号分隔,测试用,优先于 --limit)")
     args = ap.parse_args()
 
+    cities = _select_cities(args.limit, args.city_ids)
+    print(f"城市数: {len(cities)} / 并发: {args.concurrency} / 错峰: {args.stagger}s / 间隔: {args.interval}s")
+
     if args.loop:
         print(f"循环模式: 每 {args.interval}s 一轮 (Ctrl-C 退出)")
         while True:
             try:
-                run_once(args.prune_hours)
+                run_once(args.prune_hours, args.concurrency, args.stagger, cities)
             except Exception as e:  # noqa: BLE001
                 print(f"[{datetime.now():%H:%M:%S}] 本轮异常: {type(e).__name__}: {e}")
                 _release_lock()
             time.sleep(args.interval)
     else:
-        run_once(args.prune_hours)
+        run_once(args.prune_hours, args.concurrency, args.stagger, cities)
 
 
 if __name__ == "__main__":

tests/test_admin_config.py

@@ -117,6 +117,40 @@ def test_update_ad_limit_takes_effect(admin_client: TestClient, token: str) -> N
         db.close()
 
 
+def test_update_bool_config(admin_client: TestClient, token: str) -> None:
+    # 提现自动对账开关默认 True
+    items = {
+        i["key"]: i
+        for i in admin_client.get("/admin/api/config", headers=_auth(token)).json()
+    }
+    assert items["withdraw_auto_reconcile_enabled"]["type"] == "bool"
+    assert items["withdraw_auto_reconcile_enabled"]["value"] is True
+
+    # 关掉 → DB 落 False、业务读到 False
+    r = admin_client.patch(
+        "/admin/api/config/withdraw_auto_reconcile_enabled",
+        json={"value": False},
+        headers=_auth(token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["value"] is False and r.json()["overridden"] is True
+
+    db = SessionLocal()
+    try:
+        from app.repositories import app_config
+
+        assert app_config.get_value(db, "withdraw_auto_reconcile_enabled") is False
+    finally:
+        db.close()
+
+    # bool 项不接受非布尔值
+    assert admin_client.patch(
+        "/admin/api/config/withdraw_auto_reconcile_enabled",
+        json={"value": 1},
+        headers=_auth(token),
+    ).status_code == 400
+
+
 def test_config_validation(admin_client: TestClient, token: str) -> None:
     # 签到档位长度≠7
     assert admin_client.patch(

tests/test_platform.py

@@ -0,0 +1,13 @@
+"""平台公开端点测试(不鉴权):/api/v1/platform/flags 等。"""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+
+def test_flags_default_comparing_ad_enabled(client: TestClient) -> None:
+    """空配置库下,/flags 返回 comparing_ad_enabled 的默认值 True;不需要鉴权。"""
+    r = client.get("/api/v1/platform/flags")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert "comparing_ad_enabled" in body
+    assert body["comparing_ad_enabled"] is True

tools/gen_meituan_cities.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""从美团「城市字典」Excel 生成随仓库的 JSON(app/integrations/data/meituan_cities.json)。
+
+本地一次性工具:美团每年更新城市字典时,把新 Excel 放进来重跑即可。
+Excel 不入库、不进仓库(二进制、需 openpyxl);生成的 JSON 随仓库提交,部署到服务器后
+ETL(scripts/pull_meituan_coupons.py)直接读它遍历全部城市。
+
+字典口径:359 个地级市,经实测一个地级市 cityId 已覆盖其下辖县级市(如徐州→邳州/新沂),
+故无需区县层级。
+
+用法:
+    python tools/gen_meituan_cities.py ["城市字典xxx.xlsx" 路径]
+    (不传则用默认 e:\\codes\\城市字典2025 (1).xlsx)
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+
+import openpyxl
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+DEFAULT_EXCEL = r"e:\codes\城市字典2025 (1).xlsx"
+OUT = os.path.join(ROOT, "app", "integrations", "data", "meituan_cities.json")
+
+
+def main() -> None:
+    excel = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_EXCEL
+    wb = openpyxl.load_workbook(excel, data_only=True)
+    ws = wb.worksheets[0]
+    rows = list(ws.iter_rows(values_only=True))
+    header = rows[0]
+
+    cities = []
+    seen = set()
+    for r in rows[1:]:
+        cid = str(r[0]).strip() if r[0] else ""
+        name = str(r[1]).strip() if len(r) > 1 and r[1] else ""
+        prov = str(r[2]).strip() if len(r) > 2 and r[2] else ""
+        if not cid or cid in seen:
+            continue
+        seen.add(cid)
+        cities.append({"city_id": cid, "name": name, "province": prov})
+
+    os.makedirs(os.path.dirname(OUT), exist_ok=True)
+    with open(OUT, "w", encoding="utf-8") as f:
+        json.dump(cities, f, ensure_ascii=False, indent=1)
+
+    print(f"表头: {header}")
+    print(f"写出 {len(cities)} 城 → {OUT}")
+    print("样例:", cities[:3])
+
+
+if __name__ == "__main__":
+    main()