feat(coin-history): 信息流广告奖励按点位场景拆流水文案(比价/领券)

grant_feed_reward 按 feed_scene 落不同 biz_type/remark:
comparison→feed_ad_reward_comparison(比价奖励)、coupon→feed_ad_reward_coupon
(领券奖励),其余(welfare/空/旧端不传)维持通用 feed_ad_reward(信息流广告奖励)。
客户端按 biz_type 直显固定文案,remark 仅作后台留痕/兜底。

附 scripts/seed_coinhistory_labels_test.py:dev-only 文案验收脚手架,往测试号
塞每种 bizType 各一条流水(ref_id 前缀 TESTDOC),--clean 按前缀精确清理。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.env.example

@@ -6,11 +6,8 @@ APP_NAME=shaguabijia-app-server
 APP_DEBUG=true
 
 # ===== 数据库 =====
-# 本地开发/测试统一用 Docker PostgreSQL:run.bat/run.sh 会自动拉起容器
-# (docker-compose.yml + scripts/ensure_pg.py)。详见 docs/database/postgres-migration.md。
-# 生产用原生 PG,由 scripts/init_postgres.py 写入强随机密码的连接串。
-# ⚠️ scheme 必须是 postgresql+psycopg://(psycopg3);不要写成 postgresql://(会去找未装的 psycopg2)。
-DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia
+# SQLite 本地文件路径。生产环境用 /opt/shaguabijia-app-server/data.db
+DATABASE_URL=sqlite:///./data/app.db
 
 # ===== JWT =====
 # 生产部署务必改成随机长字符串,可用:python -c "import secrets; print(secrets.token_urlsafe(64))"

.gitignore

@@ -57,6 +57,3 @@ tests/meituan_coupon_bj.tsv
 tests/meituan_coupon_data.tsv
 tests/meituan_coupon_fz.tsv
 tests/meituan_coupon_xm.tsv
-
-# git worktrees (superpowers 隔离工作区)
-.worktrees/

CLAUDE.md

@@ -76,8 +76,8 @@ Endpoints under `app/api/internal/` are for server-to-server communication (pric
 
 ## Database
 
-- **Dev/Test**: Docker PostgreSQL 16 — `run.sh`/`run.bat` auto-start it via `scripts/ensure_pg.py` + `docker-compose.yml`; `.env.example` ships the PG URL by default; pytest uses the same container's `shaguabijia_test` DB. **Local no longer uses SQLite** (the SQLite branch in `db/session.py` is retained as a fallback only).
-- **Prod**: native PostgreSQL — bootstrap with `scripts/init_postgres.py` (no Docker). Pool size 10 + max overflow 20, pool_recycle 3600.
+- **Dev**: SQLite (`sqlite:///./data/app.db`), `check_same_thread=False`, no connection pool.
+- **Prod**: PostgreSQL — just change `DATABASE_URL` in `.env`. Pool size 10 + max overflow 20, pool_recycle 3600.
 - **Migrations**: Alembic with `render_as_batch` for SQLite compatibility. ~60+ migration files in `alembic/versions/` (filenames are descriptive, not hex prefixes). Migration chain uses `down_revision` within each file.
 - **New models**: Define in `app/models/`, import in `app/models/__init__.py`, then run `alembic revision --autogenerate`.
 
@@ -89,7 +89,7 @@ All config via `pydantic-settings` in `app/core/config.py`. Single `Settings` cl
 
 ## Testing
 
-- `tests/conftest.py`: Sets env vars BEFORE imports, ensures the Docker PG `shaguabijia_test` DB via `scripts/ensure_pg.py`, builds all tables with `Base.metadata.create_all()` (drop+create for a clean start), tears down with `drop_all()`.
+- `tests/conftest.py`: Sets env vars BEFORE imports, creates temp SQLite file, builds all tables with `Base.metadata.create_all()`, tears down with `drop_all()` + unlink.
 - External integrations are monkeypatched in tests (e.g., WeChat Pay, Jiguang, Pangle callbacks) — tests never make real HTTP calls.
 - `TestClient` from FastAPI is used for all tests. Rate limiting is disabled globally in tests.
 

app/admin/repositories/ad_revenue.py

@@ -385,27 +385,6 @@ def ad_revenue_report(
         for k, v in type_map.items()
     }
 
-    # 分场景小计(按 feed_scene:展示条数 + 预估收益),同 type_stats 基于全量 events——
-    # 供数据大盘「领券广告 / 比价广告」卡用。此前大盘是在分页 items 里按 feed_scene 现算,
-    # 2026-07-02 起信息流逐条展示行(唯一带收益 + 场景的行)不再进主表 items,现算恒为 0;
-    # 改为服务端在全量上聚合下发(也顺带不受 limit 分页截断影响)。feed_scene 为空(激励视频 /
-    # 旧数据)不计入任何场景桶。
-    scene_map: dict[str, dict] = {}
-    for e in events:
-        sc = e.get("feed_scene")
-        if not sc:
-            continue
-        s = scene_map.get(sc)
-        if s is None:
-            s = {"impressions": 0, "revenue_yuan": 0.0}
-            scene_map[sc] = s
-        s["impressions"] += e["impressions"]
-        s["revenue_yuan"] += e["revenue_yuan"]
-    scene_stats = {
-        k: {"impressions": v["impressions"], "revenue_yuan": round(v["revenue_yuan"], 6)}
-        for k, v in scene_map.items()
-    }
-
     # DAU:复用数据大盘活跃用户口径(登录 + 开始比价 + 开始领券,按用户去重),按所选日期区间
     # 统计(含今日),历史 / 多天区间同样有值。ARPU = 区间预估收益 ÷ 区间活跃用户。全局口径,
     # 不随 user / ad_type / feed_scene / app_env 筛选变化(活跃用户口径无这些维度)。
@@ -439,7 +418,6 @@ def ad_revenue_report(
         "daily": daily,
         "hourly": hourly,
         "type_stats": type_stats,
-        "scene_stats": scene_stats,
         "dau": dau,
         "items": main_rows[offset:offset + limit],
     }

app/admin/repositories/cps.py

@@ -51,27 +51,7 @@ def _yuan_to_cents(v: object) -> int | None:
 
 def _ts_to_dt(ts: object) -> datetime | None:
     """秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
-    if ts is None:
-        return None
-    if isinstance(ts, datetime):
-        return ts if ts.tzinfo else ts.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc)
-    s = str(ts).strip()
-    if not s or s.lower() == "null":
-        return None
-    try:
-        seconds = float(Decimal(s))
-    except (InvalidOperation, ValueError):
-        return None
-    if seconds == 0:
-        return None
-    # 美团文档是秒级时间戳，这里顺手兼容毫秒/微秒，避免上游格式变化导致时间再次落空。
-    if abs(seconds) > 10_000_000_000_000:
-        seconds /= 1_000_000
-    elif abs(seconds) > 10_000_000_000:
-        seconds /= 1_000
-    try:
-        return datetime.fromtimestamp(seconds, tz=timezone.utc)
-    except (OverflowError, OSError, ValueError):
+    if not ts:
         return None
 
 
@@ -103,6 +83,10 @@ def _pick(row: dict[str, Any], *keys: str) -> Any:
         if key in row and row[key] is not None:
             return row[key]
     return None
+    try:
+        return datetime.fromtimestamp(int(ts), tz=timezone.utc)
+    except (ValueError, OSError, TypeError):
+        return None
 
 
 # ───────────── 群 ─────────────

app/admin/repositories/queries.py

@@ -11,7 +11,6 @@ from zoneinfo import ZoneInfo
 from sqlalchemy import Select, asc, case, desc, func, or_, select
 from sqlalchemy.orm import Session
 
-from app.admin.repositories.stats import COMPARE_START_EVENT, COUPON_START_EVENT
 from app.core import rewards
 from app.core.config import settings
 from app.models.ad_feed_reward import AdFeedRewardRecord
@@ -19,22 +18,12 @@ from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
 from app.models.analytics_event import AnalyticsEvent
 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.onboarding import OnboardingCompletion
 from app.models.price_report import PriceReport
 from app.models.user import User
-from app.models.wallet import (
-    CashTransaction,
-    CoinAccount,
-    CoinTransaction,
-    InviteCashTransaction,
-    WithdrawOrder,
-)
-
-# 「最近活跃」计入的行为事件(与大盘 DAU/留存活跃口径一致:开始比价 + 开始领券)
-_ACTIVE_EVENTS = (COMPARE_START_EVENT, COUPON_START_EVENT)
+from app.models.wallet import CashTransaction, CoinAccount, CoinTransaction, WithdrawOrder
 
 # 折算成可提现现金时,非广告金币来源的排除集(广告单独统计、人工调整不算"赚取")
 _NON_TASK_BIZ_TYPES = ("reward_video", "feed_ad_reward", "admin_grant", "admin_deduct")
@@ -87,86 +76,6 @@ def offset_paginate(
     return items, next_cursor, total
 
 
-def _last_active_parts():
-    """「最近活跃」的两个按 user_id 预聚合派生表(最近开始比价/领券事件、最近领券发起)。
-
-    活跃口径与大盘 DAU/留存一致(2026-07-05 产品定:进入 App≈登录 last_login_at +
-    发起比价 real_compare_start + 发起领券 real_coupon_start/claim_started)。
-    用 LEFT JOIN 预聚合而非相关标量子查询:后者在 PG 上对 users 每行各跑一个 SubPlan
-    (排序键、range 筛选、offset_paginate 的 count 三处叠加),埋点表大了会拖垮列表接口;
-    预聚合借 analytics_event.event 索引只扫两类 start 事件,每次查询聚合一次。
-    """
-    ev_agg = (
-        select(
-            AnalyticsEvent.user_id.label("user_id"),
-            func.max(AnalyticsEvent.created_at).label("last_at"),
-        )
-        .where(
-            AnalyticsEvent.user_id.is_not(None),
-            AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
-        )
-        .group_by(AnalyticsEvent.user_id)
-        .subquery()
-    )
-    eng_agg = (
-        select(
-            CouponPromptEngagement.user_id.label("user_id"),
-            func.max(CouponPromptEngagement.created_at).label("last_at"),
-        )
-        .where(
-            CouponPromptEngagement.user_id.is_not(None),
-            CouponPromptEngagement.engage_type == "claim_started",
-        )
-        .group_by(CouponPromptEngagement.user_id)
-        .subquery()
-    )
-    return ev_agg, eng_agg
-
-
-def _norm_utc(dt: datetime | None) -> datetime | None:
-    """naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
-    if dt is None:
-        return None
-    return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
-
-
-def _attach_last_active(db: Session, users: list[User]) -> None:
-    """给本页用户瞬态挂 last_active_at(非 DB 列,供 AdminUserListItem from_attributes 读)。
-
-    口径同 [_last_active_expr];按本页 user_id 批量两次 GROUP BY,防 N+1。
-    """
-    uids = [u.id for u in users]
-    if not uids:
-        return
-    ev_map = dict(
-        db.execute(
-            select(AnalyticsEvent.user_id, func.max(AnalyticsEvent.created_at))
-            .where(
-                AnalyticsEvent.user_id.in_(uids),
-                AnalyticsEvent.event.in_(_ACTIVE_EVENTS),
-            )
-            .group_by(AnalyticsEvent.user_id)
-        ).all()
-    )
-    eng_map = dict(
-        db.execute(
-            select(CouponPromptEngagement.user_id, func.max(CouponPromptEngagement.created_at))
-            .where(
-                CouponPromptEngagement.user_id.in_(uids),
-                CouponPromptEngagement.engage_type == "claim_started",
-            )
-            .group_by(CouponPromptEngagement.user_id)
-        ).all()
-    )
-    for u in users:
-        candidates = [
-            _norm_utc(u.last_login_at),
-            _norm_utc(ev_map.get(u.id)),
-            _norm_utc(eng_map.get(u.id)),
-        ]
-        u.last_active_at = max((c for c in candidates if c is not None), default=None)
-
-
 def list_users(
     db: Session,
     *,
@@ -178,34 +87,16 @@ def list_users(
     created_to: datetime | None = None,
     last_login_from: datetime | None = None,
     last_login_to: datetime | None = None,
-    last_active_from: datetime | None = None,
-    last_active_to: datetime | None = None,
     sort_by: str = "id",
     sort_order: str = "desc",
     limit: int = 20,
     cursor: int | None = None,
 ) -> tuple[list[User], int | None, int]:
-    """用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录·最近活跃
-    时间范围筛选,按 id·注册时间·最近登录·最近活跃排序;每页附带计算列 last_active_at
-    (口径见 [_last_active_expr])。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
+    """用户列表(admin 全量)。支持手机号前缀 / 渠道 / 状态 / 昵称模糊 / 注册·最近登录时间范围筛选,
+    按 id·注册时间·最近登录排序。**offset 分页**(cursor=offset):任意列排序下游标语义统一,
     代价是翻页期间数据变动可能错位一条——admin 低频场景可接受(同 [list_all_withdraw_orders])。
     日期入参统一转 tz-aware UTC 比较(列为 timestamptz,见 _as_utc)。"""
-    # 最近活跃 = max(最近登录, 最近行为事件, 最近领券发起)。PG 用 GREATEST;SQLite 标量 max()
-    # 任一参数 NULL 即返回 NULL,故 LEFT JOIN 未命中侧 coalesce 到 last_login_at 兜底
-    # (注册即登录,该列恒非空)。派生表 1:1(按 user_id 聚合),outerjoin 不会放大行数,
-    # offset_paginate 的 count 不受影响。
-    ev_agg, eng_agg = _last_active_parts()
-    greatest = func.greatest if db.get_bind().dialect.name == "postgresql" else func.max
-    last_active = greatest(
-        User.last_login_at,
-        func.coalesce(ev_agg.c.last_at, User.last_login_at),
-        func.coalesce(eng_agg.c.last_at, User.last_login_at),
-    )
-    stmt = (
-        select(User)
-        .outerjoin(ev_agg, ev_agg.c.user_id == User.id)
-        .outerjoin(eng_agg, eng_agg.c.user_id == User.id)
-    )
+    stmt = select(User)
     if phone:
         stmt = stmt.where(User.phone.like(f"{phone}%"))  # 前缀匹配
     if register_channel:
@@ -222,25 +113,16 @@ def list_users(
         stmt = stmt.where(User.last_login_at >= _as_utc(last_login_from))
     if last_login_to is not None:
         stmt = stmt.where(User.last_login_at <= _as_utc(last_login_to))
-    if last_active_from is not None:
-        stmt = stmt.where(last_active >= _as_utc(last_active_from))
-    if last_active_to is not None:
-        stmt = stmt.where(last_active <= _as_utc(last_active_to))
 
     sort_cols = {
         "id": User.id,
         "created_at": User.created_at,
         "last_login_at": User.last_login_at,
-        "last_active_at": last_active,
     }
     sort_col = sort_cols.get(sort_by, User.id)
     order_fn = asc if sort_order == "asc" else desc
     id_order = asc(User.id) if sort_order == "asc" else desc(User.id)
-    items, next_cursor, total = offset_paginate(
-        db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor
-    )
-    _attach_last_active(db, items)
-    return items, next_cursor, total
+    return offset_paginate(db, stmt, (order_fn(sort_col), id_order), limit=limit, cursor=cursor)
 
 
 def _attach_user_info(db: Session, records: list[ComparisonRecord | Feedback | PriceReport]) -> None:
@@ -852,19 +734,26 @@ def withdraw_risk_flags(
     return flags, score
 
 
-def _check_withdraw_ledger_side(
-    orders: list[WithdrawOrder], txns: list, *, withdraw_biz: str, refund_biz: str
-) -> dict:
-    """对某一本账(普通现金 / 邀请奖励金)做提现单 ↔ 流水的交叉校验。
+def withdraw_ledger_check(db: Session) -> dict:
+    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()
+    )
 
-    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}
+    orders = list(db.execute(select(WithdrawOrder)).scalars().all())
+    cash_txns = list(
+        db.execute(
+            select(CashTransaction).where(
+                CashTransaction.biz_type.in_(("withdraw", "withdraw_refund"))
+            )
+        ).scalars().all()
+    )
+    withdraw_refs = {txn.ref_id for txn in cash_txns if txn.biz_type == "withdraw"}
     refund_counts: dict[str, int] = {}
-    for txn in txns:
-        if txn.biz_type == refund_biz and txn.ref_id:
+    for txn in cash_txns:
+        if txn.biz_type == "withdraw_refund" and txn.ref_id:
             refund_counts[txn.ref_id] = refund_counts.get(txn.ref_id, 0) + 1
 
     missing_withdraw = 0
@@ -879,95 +768,24 @@ def _check_withdraw_ledger_side(
         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
-
+    duplicate_refund = sum(1 for count in refund_counts.values() if count > 1)
+    diff = cash_balance_total - cash_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())
+        diff == 0
+        and missing_withdraw == 0
+        and missing_refund == 0
+        and duplicate_refund == 0
+        and refund_on_non_terminal == 0
     )
     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"],
+        "balance_diff_cents": diff,
+        "missing_withdraw_txn_count": missing_withdraw,
+        "missing_refund_txn_count": missing_refund,
+        "duplicate_refund_txn_count": duplicate_refund,
+        "refund_txn_on_non_terminal_count": refund_on_non_terminal,
     }
 
 

app/admin/repositories/stats.py

@@ -5,23 +5,17 @@ user.last_login_at / comparison_record.status / withdraw_order.status)要加索
 """
 from __future__ import annotations
 
-from collections import Counter
 from datetime import date, datetime, time, timedelta, timezone
 from decimal import Decimal, InvalidOperation
 
-from sqlalchemy import case, func, select
+from sqlalchemy import func, select
 from sqlalchemy.orm import Session
 
-from app.admin.repositories.coupon_data import _percentile
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.analytics_event import AnalyticsEvent
 from app.models.comparison import ComparisonRecord
-from app.models.coupon_state import (
-    CouponClaimRecord,
-    CouponPromptEngagement,
-    CouponSession,
-)
+from app.models.coupon_state import CouponPromptEngagement
 from app.models.cps_order import CpsOrder
 from app.models.feedback import Feedback
 from app.models.savings import SavingsRecord
@@ -305,12 +299,12 @@ def dashboard_overview(
         period_from=period_from,
         period_to=period_to,
     )
-    # 留存口径(2026-07-05 产品改):次日留存——窗口内每天 D,取 **D-1 日(前日)新增**用户,
-    # 统计其 D 日活跃(登录/开始比价/开始领券)比例,逐日累加。默认窗口=昨日单天,即
-    # 「前日新增用户的昨日留存」。原口径(窗口内新增∩窗口内活跃)在单日窗口下≈100% 无意义
-    # (注册即登录,当天新增必然当天活跃)。逐日 cohort 在下方 trend 循环内顺带累计。
-    retention_cohort_total = 0
-    retention_retained_total = 0
+    period_retained_new_user_ids = period_new_user_ids & period_active_user_ids
+    period_retention_rate = (
+        round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4)
+        if period_new_user_ids
+        else None
+    )
     trend_points: list[dict] = []
     for cur_date in _date_range(period_from, period_to):
         day_start_utc, day_end_utc, day_start_local, day_end_local = _period_bounds(
@@ -320,26 +314,18 @@ def dashboard_overview(
             ComparisonRecord.created_at >= day_start_local,
             ComparisonRecord.created_at < day_end_local,
         )
-        daily_active_user_ids = _period_active_user_ids(
-            db,
-            start_utc=day_start_utc,
-            end_utc=day_end_utc,
-            period_from=cur_date,
-            period_to=cur_date,
-        )
-        # 次日留存:cohort = 前一日(D-1)新增用户,留存 = 其中当日(D)活跃者(口径见上)。
-        cohort_ids = _user_id_set(
-            select(User.id).where(
-                User.created_at >= day_start_utc - timedelta(days=1),
-                User.created_at < day_end_utc - timedelta(days=1),
-            )
-        )
-        retention_cohort_total += len(cohort_ids)
-        retention_retained_total += len(cohort_ids & daily_active_user_ids)
         trend_points.append(
             {
                 "date": cur_date,
-                "active_users": len(daily_active_user_ids),
+                "active_users": len(
+                    _period_active_user_ids(
+                        db,
+                        start_utc=day_start_utc,
+                        end_utc=day_end_utc,
+                        period_from=cur_date,
+                        period_to=cur_date,
+                    )
+                ),
                 "new_users": _count(
                     User,
                     User.created_at >= day_start_utc,
@@ -348,11 +334,6 @@ def dashboard_overview(
                 "comparisons": _count(ComparisonRecord, *daily_comparison_conds),
             }
         )
-    period_retention_rate = (
-        round(retention_retained_total / retention_cohort_total, 4)
-        if retention_cohort_total
-        else None
-    )
 
     period_coin_conds = (
         CoinTransaction.created_at >= start_local,
@@ -436,98 +417,6 @@ def dashboard_overview(
         else None
     )
 
-    # ===== 领券核心数据(2026-07-05 产品新增)=====
-    # 数据源:coupon_session(一次领券一行,started_date 北京自然日)+ coupon_claim_record
-    # (一券/点位一天一条终态,claim_date 北京自然日)。点位与 session 不按 trace_id 关联——
-    # record_claims 更新路径不覆盖 trace_id(同设备同券同日重跑归第一次的 trace),按
-    # (device_id, 自然日) 桶关联才可靠;同桶多次发起共享同一份点位终态。
-    period_coupon_sessions = db.execute(
-        select(
-            CouponSession.device_id,
-            CouponSession.started_date,
-            CouponSession.status,
-            CouponSession.elapsed_ms,
-        ).where(
-            CouponSession.started_date >= period_from,
-            CouponSession.started_date <= period_to,
-            # 只统计正式环境,同「领券数据」页默认口径(防 debug 包调试数据串台;
-            # 命中 ix_coupon_session_date_env)。点位表无 app_env 列,但点位指标只经
-            # 下方 prod session 触达的 (device, 日) 桶进入统计,随之收敛到 prod。
-            CouponSession.app_env == "prod",
-        )
-    ).all()
-    coupon_started = len(period_coupon_sessions)
-    coupon_completed_elapsed = sorted(
-        s.elapsed_ms
-        for s in period_coupon_sessions
-        if s.status == "completed" and s.elapsed_ms is not None
-    )
-    # 点位桶:(device, 日) → (点位总数, 成功点位数)。成功口径与「我的」页累计领券一致
-    # (sum_claimed_count,2026-06-15 产品定):success + already_claimed(已领过=持有券)都算成功。
-    point_buckets: dict[tuple[str, date], tuple[int, int]] = {
-        (dev, d): (int(total), int(succ or 0))
-        for dev, d, total, succ in db.execute(
-            select(
-                CouponClaimRecord.device_id,
-                CouponClaimRecord.claim_date,
-                func.count(),
-                func.sum(
-                    case(
-                        (CouponClaimRecord.status.in_(("success", "already_claimed")), 1),
-                        else_=0,
-                    )
-                ),
-            )
-            .where(
-                CouponClaimRecord.claim_date >= period_from,
-                # 上界放宽一天:跨零点场次(23:5x 发起)的点位 claim_date 落在发起日+1,
-                # 桶只经下方 session 触达的键参与计数,放宽不会引入无关数据。
-                CouponClaimRecord.claim_date <= period_to + timedelta(days=1),
-            )
-            .group_by(CouponClaimRecord.device_id, CouponClaimRecord.claim_date)
-        ).all()
-    }
-    # 全部领成功的次数:completed 且其 (device, 日) 桶内点位全部成功(桶为空不算)。
-    coupon_all_success = 0
-    completed_bucket_totals: list[int] = []
-    session_bucket_keys: set[tuple[str, date]] = set()
-    for s in period_coupon_sessions:
-        key = (s.device_id, s.started_date)
-        if key not in point_buckets:
-            # 跨零点回退:发起日桶不存在(点位终态全部落在次日)时取 (device, 发起日+1)。
-            # 仅在发起日桶完全缺失时回退,避免抢占该设备次日 session 自己的桶。
-            next_key = (s.device_id, s.started_date + timedelta(days=1))
-            if next_key in point_buckets:
-                key = next_key
-        bucket = point_buckets.get(key)
-        if bucket is not None:
-            session_bucket_keys.add(key)
-        if s.status != "completed" or bucket is None:
-            continue
-        total, succ = bucket
-        completed_bucket_totals.append(total)
-        if total > 0 and succ == total:
-            coupon_all_success += 1
-    # 每次发起的应领点位数:取「完成过的领券」实际点位数的众数(done 帧会给所有点位终态,
-    # 完成场的点位数=当前配置的全量点位数;数据自校准,配置改点位数无需改代码)。本期无完成场
-    # 时给不出,点位成功率置空。
-    coupon_points_per_session = (
-        Counter(completed_bucket_totals).most_common(1)[0][0]
-        if completed_bucket_totals
-        else None
-    )
-    # 成功点位数:本期 session 触达过的 (device, 日) 桶内成功点位之和(桶级去重,同桶重试不重复计)。
-    coupon_point_success = sum(point_buckets[k][1] for k in session_bucket_keys)
-    # 点位成功率 = 成功点位数 / (发起数 × 应领点位数):中途退出未跑到的点位不产生记录,
-    # 但发起数×点位数把它们计入分母 → 视为失败,符合产品口径;重试会拉低该率(分母按次数计)。
-    coupon_point_success_rate = (
-        round(
-            min(1.0, coupon_point_success / (coupon_started * coupon_points_per_session)), 4
-        )
-        if coupon_started and coupon_points_per_session
-        else None
-    )
-
     return {
         "users": {
             "total": _count(User),
@@ -591,13 +480,11 @@ def dashboard_overview(
             "users": {
                 "new": len(period_new_user_ids),
                 "active": len(period_active_user_ids),
-                "retained_new_users": retention_retained_total,
-                "retention_cohort": retention_cohort_total,
+                "retained_new_users": len(period_retained_new_user_ids),
                 "retention_rate": period_retention_rate,
                 "retention_note": (
-                    "次日留存:窗口内每天取前一日新增用户,统计其当日活跃"
-                    "(登录/开始比价/开始领券,按用户去重)比例,逐日累加;"
-                    "默认窗口=昨日,即前日新增用户的昨日留存"
+                    "口径:登录(last_login_at)+开始比价(real_compare_start)+"
+                    "开始领券(real_coupon_start/claim_started),按用户去重"
                 ),
             },
             "comparison": {
@@ -608,17 +495,6 @@ def dashboard_overview(
                 "average_duration_ms": period_avg_duration_ms,
                 "average_saved_cents": period_avg_saved_cents,
             },
-            "coupon": {
-                "started": coupon_started,
-                "all_success": coupon_all_success,
-                "success_rate": (
-                    round(coupon_all_success / coupon_started, 4) if coupon_started else None
-                ),
-                "point_success": coupon_point_success,
-                "points_per_session": coupon_points_per_session,
-                "point_success_rate": coupon_point_success_rate,
-                "median_elapsed_ms": _percentile(coupon_completed_elapsed, 50),
-            },
             "coins": {
                 "granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
                 "reward_video_coin_total": period_reward_video_coin_total,

app/admin/routers/ad_revenue.py

@@ -91,7 +91,6 @@ def get_ad_revenue_report(
         daily=[AdRevenueDaily(**d) for d in result["daily"]],
         hourly=[AdRevenueHourly(**h) for h in result["hourly"]],
         type_stats={k: AdRevenueTypeStat(**v) for k, v in result["type_stats"].items()},
-        scene_stats={k: AdRevenueTypeStat(**v) for k, v in result["scene_stats"].items()},
         dau=result["dau"],
         total=result["total"],
         truncated=result["truncated"],

app/admin/routers/ops_marquee_seed.py

@@ -57,15 +57,9 @@ def list_seeds(db: AdminDb) -> list[OpsMarqueeSeedOut]:
 def preview_feed(
     db: AdminDb,
     limit: Annotated[int, Query(ge=1, le=30)] = 8,
-    mode: Annotated[str | None, Query(description="mixed/real/seed;不传=当前持久化模式")] = None,
 ) -> OpsSavingsFeedPreviewOut:
-    """返回客户端实际会看到的 feed(真实记录会插队、种子随机抽取/金额随机/名字合成),供运营对效果。
-
-    mode 显式指定则预览该模式(**不改持久化配置**,供前端切换开关时实时预览);不传则用当前持久化模式。
-    """
-    if mode is not None and mode not in ops_marquee.FEED_MODES:
-        raise HTTPException(status_code=400, detail="mode 需为 mixed / real / seed")
-    return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit, mode=mode))
+    """返回客户端实际会看到的 feed(真实记录会插队、种子随机抽取/金额随机/名字合成),供运营对效果。"""
+    return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit))
 
 
 # 注:/mode 两个端点须在 /{seed_id} 之前注册,否则 PATCH /mode 会被 /{seed_id} 抢先按 id 解析。

app/admin/routers/users.py

@@ -42,12 +42,7 @@ def list_users(
     created_to: Annotated[datetime | None, Query()] = None,
     last_login_from: Annotated[datetime | None, Query()] = None,
     last_login_to: Annotated[datetime | None, Query()] = None,
-    # 最近活跃(登录/发起比价/发起领券取最大,见 queries._last_active_expr)筛选与排序
-    last_active_from: Annotated[datetime | None, Query()] = None,
-    last_active_to: Annotated[datetime | None, Query()] = None,
-    sort_by: Annotated[
-        str, Query(pattern="^(id|created_at|last_login_at|last_active_at)$")
-    ] = "id",
+    sort_by: Annotated[str, Query(pattern="^(id|created_at|last_login_at)$")] = "id",
     sort_order: Annotated[str, Query(pattern="^(asc|desc)$")] = "desc",
     limit: Annotated[int, Query(ge=1, le=100)] = 20,
     cursor: Annotated[int | None, Query()] = None,
@@ -56,7 +51,6 @@ def list_users(
         db, phone=phone, register_channel=register_channel, status=status,
         nickname=nickname, created_from=created_from, created_to=created_to,
         last_login_from=last_login_from, last_login_to=last_login_to,
-        last_active_from=last_active_from, last_active_to=last_active_to,
         sort_by=sort_by, sort_order=sort_order, limit=limit, cursor=cursor,
     )
     return CursorPage(

app/admin/schemas/ad_revenue.py

@@ -136,11 +136,6 @@ class AdRevenueReportOut(BaseModel):
         default_factory=dict,
         description="按广告类型(ad_type)小计 {ad_type: {impressions, revenue_yuan}};前端取 draw / reward_video 做分类大盘",
     )
-    scene_stats: dict[str, AdRevenueTypeStat] = Field(
-        default_factory=dict,
-        description="按信息流场景(feed_scene)小计 {comparison/coupon/welfare: {impressions, revenue_yuan}};"
-        "全量统计(不受分页截断),供数据大盘「领券广告 / 比价广告」卡;feed_scene 为空的事件不计入",
-    )
     dau: int | None = Field(
         None,
         description="所选日期区间的去重活跃用户数(口径同数据大盘 period.users.active:登录 + 开始比价 + "

app/admin/schemas/dashboard.py

@@ -43,10 +43,7 @@ class DashboardComparison(BaseModel):
 class DashboardPeriodUsers(BaseModel):
     new: int
     active: int
-    # 次日留存(2026-07-05 起):retained_new_users = 窗口内逐日「前一日新增且当日活跃」用户数之和,
-    # retention_cohort = 对应的前一日新增基数之和,retention_rate = 两者之比。
     retained_new_users: int
-    retention_cohort: int = 0
     retention_rate: float | None = None
     retention_note: str
 
@@ -60,24 +57,6 @@ class DashboardPeriodComparison(BaseModel):
     average_saved_cents: int | None = None
 
 
-class DashboardPeriodCoupon(BaseModel):
-    """领券核心数据(2026-07-05 产品新增)。点位=一张券(coupon_claim_record 一天一条终态);
-    成功口径 success+already_claimed(与「我的」页累计领券一致)。"""
-
-    started: int = 0
-    # 全部领成功的次数:completed 且当日该设备全部点位成功
-    all_success: int = 0
-    success_rate: float | None = None
-    # 本期 session 触达的点位中成功的条数(同设备同日去重)
-    point_success: int = 0
-    # 每次发起的应领点位数(本期完成场实际点位数的众数;无完成场为空)
-    points_per_session: int | None = None
-    # 点位成功率 = point_success / (started × points_per_session);未跑到的点位计入分母视为失败
-    point_success_rate: float | None = None
-    # 耗时中位数(仅 completed 的 elapsed_ms,同「领券数据」页口径)
-    median_elapsed_ms: int | None = None
-
-
 class DashboardPeriodCoins(BaseModel):
     granted_total: int
     reward_video_coin_total: int = 0
@@ -106,7 +85,6 @@ class DashboardPeriod(BaseModel):
     date_to: date
     users: DashboardPeriodUsers
     comparison: DashboardPeriodComparison
-    coupon: DashboardPeriodCoupon = DashboardPeriodCoupon()
     coins: DashboardPeriodCoins
     cash: DashboardPeriodCash
     trend: list[DashboardTrendPoint] = []

app/admin/schemas/user.py

@@ -20,9 +20,6 @@ class AdminUserListItem(BaseModel):
     wechat_nickname: str | None = None
     created_at: datetime
     last_login_at: datetime
-    # 最近活跃 = max(最近登录, 最近发起比价, 最近发起领券);列表页由 queries._attach_last_active
-    # 瞬态挂上。其他复用本 schema 的入口(用户 360 等)没挂该属性 → None(前端显示 '-')。
-    last_active_at: datetime | None = None
 
 
 class AdminUserOverview(BaseModel):

app/admin/schemas/wallet.py

@@ -130,7 +130,6 @@ class WithdrawBulkResult(BaseModel):
 
 class WithdrawLedgerCheckOut(BaseModel):
     ok: bool
-    # 普通现金账(coin_cash:金币兑换的现金)
     cash_balance_total_cents: int
     cash_transaction_total_cents: int
     balance_diff_cents: int
@@ -138,14 +137,6 @@ class WithdrawLedgerCheckOut(BaseModel):
     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):

app/repositories/ad_feed_reward.py

@@ -176,10 +176,19 @@ def grant_feed_reward(
         )
         return _commit_record(db, rec, client_event_id)
 
+    # 按点位场景拆流水文案(2026-07):比价等候期看的广告 vs 领券时看的广告,在收益明细里分开显示。
+    # feed_scene=comparison→比价奖励 / coupon→领券奖励;其它(welfare/空/旧端不带)维持通用「信息流广告奖励」。
+    # 客户端按此 biz_type 直显固定文案(见 CoinHistoryViewModel.coinTitle),故 remark 只作后台留痕/兜底。
+    if feed_scene == "comparison":
+        reward_biz, reward_remark = "feed_ad_reward_comparison", "比价奖励"
+    elif feed_scene == "coupon":
+        reward_biz, reward_remark = "feed_ad_reward_coupon", "领券奖励"
+    else:
+        reward_biz, reward_remark = "feed_ad_reward", "信息流广告奖励"
     crud_wallet.grant_coins(
         db, user_id, coin,
-        biz_type="feed_ad_reward", ref_id=client_event_id,
-        remark="信息流广告奖励",
+        biz_type=reward_biz, ref_id=client_event_id,
+        remark=reward_remark,
     )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,

app/repositories/ops_marquee.py

@@ -30,7 +30,6 @@ from app.models.comparison import ComparisonRecord
 from app.models.ops_marquee_seed import OpsMarqueeSeed
 from app.models.user import User
 from app.repositories import app_config
-from app.repositories.user import is_default_nickname
 
 # 首页轮播数据源模式(存 app_config.marquee_feed_mode):
 #   mixed=真实优先+种子补位+合成兜底(默认,原行为);real=只真实(不足则少/空);seed=只种子+合成兜底。
@@ -141,12 +140,9 @@ def _synth_masked_name(rng: random.Random) -> str:
 
 def _mask_real(nickname: str | None, user_id: int) -> str:
     """真实用户脱敏(对齐 PRD「用户标识打码规则」):设过昵称→昵称脱敏(中英文皆可);
-    没昵称→「用户」+5星+id 后 2 位(用户*****08),按 user_id 稳定、刷新不变脸。
-
-    创建时自动分配的默认昵称(「用户」+9 位随机,见 user.is_default_nickname)不算用户主动设的昵称,
-    按「无昵称」处理走 id 规则(产品决策 2026-07:默认昵称归入「没昵称」档)。"""
+    没昵称→「用户」+5星+id 后 2 位(用户*****08),按 user_id 稳定、刷新不变脸。"""
     nick = (nickname or "").strip()
-    if nick and not is_default_nickname(nick):
+    if nick:
         return _mask_nickname(nick)
     return _mask_anon(user_id)
 
@@ -217,11 +213,9 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
     return out
 
 
-def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]:
+def get_feed(db: Session, limit: int = 8) -> list[dict]:
     """返回最多 limit 条 {masked_user, saved_amount_cents, time(HH:MM:SS 北京)}。
 
-    mode:显式传入(admin 预览指定模式)则用它、**不改持久化配置**;不传(客户端 /savings-feed)读
-    持久化的 marquee_feed_mode;非法值一律回退到持久化模式。
     真实条:success 且 0 < saved ≤ 上限,按 user 去重(同一用户只取最新一条,避免单人刷屏)。
     不足用启用的种子补齐——**公平随机抽取** need 个(而非固定取前 N),让所有种子都有机会露出;
     种子用户名留空则随机合成(避开撞名),金额取**长尾随机**(小额居多、偶尔大额,更像真实分布)。
@@ -229,8 +223,7 @@ def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]
     展示时间统一「刷新」成相对现在的最近时刻(从 now 往前**随机抖动**递减),保证轮播永远像刚发生、
     节奏自然不机械(真实用户/金额不变,只换展示时间——避免旧测试数据 / 低谷期记录显示成过时时间)。
     """
-    # 预览可显式指定模式(所见=选中模式,不依赖 PATCH 落库时序);None/非法 → 读持久化配置。
-    mode = mode if mode in FEED_MODES else get_feed_mode(db)  # mixed / real / seed
+    mode = get_feed_mode(db)  # mixed / real / seed(运营可在「首页轮播种子」页切换)
 
     items: list[dict] = []
     used_names: set[str] = set()

app/repositories/user.py

@@ -42,22 +42,6 @@ def _gen_nickname() -> str:
     )
 
 
-def is_default_nickname(nickname: str | None) -> bool:
-    """是否为创建时自动分配的默认昵称(= "用户" + 9 位字母数字,见 [_gen_nickname])。
-
-    这类不是用户主动设置的昵称,展示脱敏时按「无昵称」处理(走 id 规则,见 ops_marquee._mask_real)。
-    精确匹配生成格式(前缀 + 定长字母数字集),不误伤真人以「用户」开头的昵称(如「用户体验师」含汉字、
-    长度也不符)。用户改过昵称即不再匹配。"""
-    if not nickname:
-        return False
-    s = nickname.strip()
-    return (
-        len(s) == len(_NICKNAME_PREFIX) + _NICKNAME_LEN
-        and s.startswith(_NICKNAME_PREFIX)
-        and all(c in _NICKNAME_ALPHABET for c in s[len(_NICKNAME_PREFIX):])
-    )
-
-
 def get_user_by_username(db: Session, username: str) -> User | None:
     return db.execute(
         select(User).where(User.username == username)

docker-compose.yml

@@ -1,22 +0,0 @@
-# 本地开发/测试用 PostgreSQL。生产用原生 PG(scripts/init_postgres.py),不使用本文件。
-services:
-  postgres:
-    image: postgres:16-alpine
-    container_name: shaguabijia-pg
-    environment:
-      POSTGRES_USER: shaguabijia_app
-      POSTGRES_PASSWORD: shaguabijia_dev_pw
-      POSTGRES_DB: shaguabijia
-    ports:
-      - "5432:5432"
-    volumes:
-      - pgdata:/var/lib/postgresql/data
-      - ./docker/initdb:/docker-entrypoint-initdb.d:ro
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U shaguabijia_app -d shaguabijia"]
-      interval: 3s
-      timeout: 3s
-      retries: 20
-
-volumes:
-  pgdata:

docker/initdb/01-create-test-db.sql

@@ -1,3 +0,0 @@
--- 仅在 pgdata 卷首次初始化时执行一次(以 shaguabijia_app 连 shaguabijia 库运行)。
--- 幂等兜底见 scripts/ensure_pg.py 的 _ensure_test_db()。
-CREATE DATABASE shaguabijia_test OWNER shaguabijia_app;

docs/database/postgres-migration.md

@@ -27,21 +27,7 @@ PG 默认上 16 版(工具链最齐),驱动用 **psycopg3**(SQLAlchemy 2.0 时
 
 ## 1. 本地起 PG + 跑通空库(半天)
 
-### 1.0 推荐:Docker 一键起(本地开发/测试)
-
-本地开发不必手动装 PG。已提供 `docker-compose.yml` + `scripts/ensure_pg.py`:
-
-```bash
-cp .env.example .env    # DATABASE_URL 默认已是 Docker PG 连接串
-./run.sh                # 或 run.bat;会自动:探测 PG → 没起则启 Docker → 起 PG 容器 → 建库 → alembic → uvicorn
-pytest                  # conftest 自动引导同一容器的 shaguabijia_test 库
-```
-
-容器:`postgres:16-alpine`(名 `shaguabijia-pg`,端口 5432,命名卷 `pgdata` 持久化),
-首启即建业务库 `shaguabijia` 与测试库 `shaguabijia_test`。下面 1.1-1.5 的手动装 PG 步骤仅在
-不用 Docker 时才需要;生产仍走 §4 的原生 PG。
-
-### 1.1 装 PG(不用 Docker 时的手动方式)
+### 1.1 装 PG
 
 macOS:
 ```bash

docs/superpowers/plans/2026-07-08-local-dev-postgres-docker.md

@@ -1,740 +0,0 @@
-# 本地开发切 Docker PostgreSQL 实现计划
-
-> **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:** 让 app-server 本地开发运行与 pytest 都跑在 Docker 化的 PostgreSQL 16 上,退掉 SQLite,`run.bat`/`run.sh` 启动时自动检测并拉起 PG(必要时先启 Docker Desktop、缺镜像先拉)。
-
-**Architecture:** 新增 `docker-compose.yml`(声明 PG 服务/卷/健康检查)+ `scripts/ensure_pg.py`(跨平台引导:探测→启 Docker→compose up→等就绪→幂等建测试库),由 `run.sh`/`run.bat`/`tests/conftest.py` 三处共用。`.env.example` 默认切 PG。`app/db/session.py` 与 `alembic/env.py` 已天然支持 PG,无需改。
-
-**Tech Stack:** Docker Compose、`postgres:16-alpine`、Python 3.10+ 标准库(`socket`/`subprocess`/`urllib.parse`)、psycopg3(已装)、SQLAlchemy 2.0 + Alembic、pytest。
-
-**工作目录:** 本计划在 worktree `.worktrees/local-dev-postgres-docker`(分支 `chore/local-dev-postgres-docker`,基于 `main`)内执行。下面所有路径相对该 worktree 根(= 仓库根)。
-
-**设计依据:** [docs/superpowers/specs/2026-07-08-local-dev-postgres-docker-design.md](../specs/2026-07-08-local-dev-postgres-docker-design.md)
-
-**连接参数(全程固定值):**
-- 镜像 `postgres:16-alpine`,容器名 `shaguabijia-pg`,宿主端口 `5432`
-- 用户 `shaguabijia_app`,dev 密码 `shaguabijia_dev_pw`(本地非机密)
-- 业务库 `shaguabijia`,测试库 `shaguabijia_test`,命名卷 `pgdata`
-- dev URL:`postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia`
-- test URL:`...@localhost:5432/shaguabijia_test`
-
-**前置:** 执行机已安装 Docker Desktop。
-
----
-
-## Task 1: Docker Compose + 测试库 initdb 脚本
-
-**Files:**
-- Create: `docker-compose.yml`
-- Create: `docker/initdb/01-create-test-db.sql`
-
-- [ ] **Step 1: 写 `docker-compose.yml`**
-
-```yaml
-# 本地开发/测试用 PostgreSQL。生产用原生 PG(scripts/init_postgres.py),不使用本文件。
-services:
-  postgres:
-    image: postgres:16-alpine
-    container_name: shaguabijia-pg
-    environment:
-      POSTGRES_USER: shaguabijia_app
-      POSTGRES_PASSWORD: shaguabijia_dev_pw
-      POSTGRES_DB: shaguabijia
-    ports:
-      - "5432:5432"
-    volumes:
-      - pgdata:/var/lib/postgresql/data
-      - ./docker/initdb:/docker-entrypoint-initdb.d:ro
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U shaguabijia_app -d shaguabijia"]
-      interval: 3s
-      timeout: 3s
-      retries: 20
-
-volumes:
-  pgdata:
-```
-
-- [ ] **Step 2: 写 `docker/initdb/01-create-test-db.sql`**
-
-```sql
--- 仅在 pgdata 卷首次初始化时执行一次(以 shaguabijia_app 连 shaguabijia 库运行)。
--- 幂等兜底见 scripts/ensure_pg.py 的 _ensure_test_db()。
-CREATE DATABASE shaguabijia_test OWNER shaguabijia_app;
-```
-
-- [ ] **Step 3: 起容器验证**
-
-Run: `docker compose up -d`
-Expected: 拉取 `postgres:16-alpine`(首次)后 `Container shaguabijia-pg  Started`。
-
-- [ ] **Step 4: 验证两个库都在 + 健康**
-
-Run: `docker compose exec -T postgres psql -U shaguabijia_app -d shaguabijia -tAc "SELECT datname FROM pg_database WHERE datname IN ('shaguabijia','shaguabijia_test') ORDER BY 1"`
-Expected 输出:
-```
-shaguabijia
-shaguabijia_test
-```
-
-- [ ] **Step 5: 验证 `.worktrees/` 与 `data/` 忽略不受影响、compose 无落盘到项目目录**
-
-Run: `git status --short`
-Expected: 只列出本任务新增的 `docker-compose.yml`、`docker/initdb/01-create-test-db.sql`(数据在命名卷 `pgdata`,不在项目目录;`.worktrees/` 已忽略)。
-
-- [ ] **Step 6: Commit**
-
-```bash
-git add docker-compose.yml docker/initdb/01-create-test-db.sql
-git commit -m "feat(dev): docker-compose 起本地 PostgreSQL(含测试库 initdb)"
-```
-
----
-
-## Task 2: `scripts/ensure_pg.py` 引导脚本(TDD)
-
-**Files:**
-- Create: `scripts/ensure_pg.py`
-- Test: `tests/test_ensure_pg.py`
-
-> 说明:纯函数(URL 解析 / sqlite 判定 / 端口探测 / 平台命令映射 / sqlite 守卫 / 端口通时短路)走 TDD 单测;真正拉 Docker 的编排 `ensure()` 全链路靠 Task 3/5 的运行来验证(需真 Docker,不做单测)。此时 `conftest.py` 仍是 SQLite,不依赖 PG,单测可独立跑。
-
-- [ ] **Step 1: 写失败测试 `tests/test_ensure_pg.py`**
-
-```python
-"""scripts/ensure_pg.py 纯函数单测(不需要 Docker/PG)。"""
-from __future__ import annotations
-
-import socket
-
-from scripts.ensure_pg import (
-    _docker_desktop_cmd,
-    _is_sqlite,
-    _parse_host_port,
-    _port_open,
-    ensure,
-)
-
-
-def test_is_sqlite():
-    assert _is_sqlite("sqlite:///./data/app.db")
-    assert _is_sqlite("  SQLite:///x ")
-    assert not _is_sqlite("postgresql+psycopg://u:p@localhost:5432/db")
-
-
-def test_parse_host_port_full():
-    assert _parse_host_port(
-        "postgresql+psycopg://u:p@localhost:5432/shaguabijia"
-    ) == ("localhost", 5432)
-
-
-def test_parse_host_port_defaults():
-    # 缺端口 → 5432
-    assert _parse_host_port("postgresql+psycopg://u:p@db.example/x")[1] == 5432
-    # 缺 host → localhost
-    assert _parse_host_port("postgresql+psycopg:///x") == ("localhost", 5432)
-
-
-def test_parse_host_port_testdb():
-    assert _parse_host_port(
-        "postgresql+psycopg://u:p@localhost:5432/shaguabijia_test"
-    ) == ("localhost", 5432)
-
-
-def test_port_open_true():
-    srv = socket.socket()
-    srv.bind(("127.0.0.1", 0))
-    srv.listen(1)
-    port = srv.getsockname()[1]
-    try:
-        assert _port_open("127.0.0.1", port, timeout=1.0)
-    finally:
-        srv.close()
-
-
-def test_port_open_false():
-    s = socket.socket()
-    s.bind(("127.0.0.1", 0))
-    port = s.getsockname()[1]
-    s.close()  # 释放端口,无人监听 → 连接应失败
-    assert not _port_open("127.0.0.1", port, timeout=0.3)
-
-
-def test_docker_desktop_cmd_windows():
-    cmd = _docker_desktop_cmd("win32", r"C:\Program Files")
-    assert cmd is not None
-    assert cmd[0].endswith("Docker Desktop.exe")
-    assert "Docker" in cmd[0]
-
-
-def test_docker_desktop_cmd_darwin():
-    assert _docker_desktop_cmd("darwin", "") == ["open", "-a", "Docker"]
-
-
-def test_docker_desktop_cmd_linux():
-    assert _docker_desktop_cmd("linux", "") is None
-
-
-def test_ensure_rejects_sqlite():
-    # dev 守卫:sqlite 直接 False(不碰 Docker)
-    assert ensure("sqlite:///./data/app.db") is False
-
-
-def test_ensure_shortcircuits_when_pg_up(monkeypatch):
-    # 端口通 → 直接 True,绝不触碰 docker
-    monkeypatch.setattr("scripts.ensure_pg._port_open", lambda *a, **k: True)
-
-    def _boom():
-        raise AssertionError("端口通时不应调用 docker")
-
-    monkeypatch.setattr("scripts.ensure_pg._docker_cli_ok", _boom)
-    assert ensure("postgresql+psycopg://u:p@localhost:5432/shaguabijia") is True
-```
-
-- [ ] **Step 2: 跑测试确认失败**
-
-Run: `pytest tests/test_ensure_pg.py -v`
-Expected: FAIL —— `ModuleNotFoundError: No module named 'scripts.ensure_pg'`(还没建)。
-
-- [ ] **Step 3: 写实现 `scripts/ensure_pg.py`**
-
-```python
-"""确保本地 PostgreSQL 就绪(开发/测试统一用 Docker PG)。
-
-被三处复用:
-  - run.sh / run.bat:`python -m scripts.ensure_pg`(CLI,失败退非 0)
-  - tests/conftest.py:`from scripts.ensure_pg import ensure; ensure(test_url)`
-
-流程:读 DATABASE_URL → TCP 探测 → 没起就(必要时启 Docker Desktop)→
-`docker compose up -d` → 等 PG ready → 幂等确保测试库存在。全程无 SQLite 兜底。
-
-生产用原生 PG(scripts/init_postgres.py),不走本模块。
-"""
-from __future__ import annotations
-
-import os
-import socket
-import subprocess
-import sys
-import time
-from pathlib import Path
-from urllib.parse import urlsplit
-
-ROOT = Path(__file__).resolve().parent.parent
-
-# 日志里可能含 emoji(如 ✅);Windows GBK 控制台(cmd.exe)无法编码会抛 UnicodeEncodeError → 脚本崩、
-# run.bat 误判 ensure_pg 失败。用 backslashreplace 保底:中文仍正常,仅不可编码字符被转义,不崩。
-for _stream in (sys.stdout, sys.stderr):
-    try:
-        _stream.reconfigure(errors="backslashreplace")
-    except (AttributeError, ValueError):
-        pass
-
-APP_DB = "shaguabijia"
-TEST_DB = "shaguabijia_test"
-DB_USER = "shaguabijia_app"
-COMPOSE_SERVICE = "postgres"
-
-DOCKER_START_TIMEOUT = int(os.environ.get("ENSURE_PG_DOCKER_TIMEOUT", "120"))
-PG_READY_TIMEOUT = int(os.environ.get("ENSURE_PG_READY_TIMEOUT", "60"))
-POLL_INTERVAL = 3.0
-
-SQLITE_FIX_HINT = (
-    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia"
-)
-
-
-def _log(msg: str) -> None:
-    print(f"[ensure_pg] {msg}", flush=True)
-
-
-def _is_sqlite(url: str) -> bool:
-    return url.strip().lower().startswith("sqlite")
-
-
-def _parse_host_port(url: str) -> tuple[str, int]:
-    """从 SQLAlchemy URL 取 host/port,缺省 localhost:5432。"""
-    parts = urlsplit(url)
-    return (parts.hostname or "localhost"), (parts.port or 5432)
-
-
-def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
-    try:
-        with socket.create_connection((host, port), timeout=timeout):
-            return True
-    except OSError:
-        return False
-
-
-def _docker_desktop_cmd(platform: str, program_files: str) -> list[str] | None:
-    """按平台给出启动 Docker Desktop 的命令;Linux 返回 None(daemon 需 sudo,让用户手动)。"""
-    if platform.startswith("win"):
-        return [str(Path(program_files) / "Docker" / "Docker" / "Docker Desktop.exe")]
-    if platform == "darwin":
-        return ["open", "-a", "Docker"]
-    return None
-
-
-def _docker_ok(subcmd: str) -> bool:
-    """`docker version`(CLI 在不在)/`docker info`(daemon 起没起)成功与否。"""
-    try:
-        subprocess.run(
-            ["docker", subcmd],
-            cwd=ROOT,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=True,
-        )
-        return True
-    except (OSError, subprocess.CalledProcessError):
-        return False
-
-
-def _docker_cli_ok() -> bool:
-    return _docker_ok("version")
-
-
-def _docker_daemon_ok() -> bool:
-    return _docker_ok("info")
-
-
-def _start_docker_daemon() -> bool:
-    """守护进程没起时按平台拉起,轮询到就绪。返回是否成功。"""
-    if _docker_daemon_ok():
-        return True
-    cmd = _docker_desktop_cmd(
-        sys.platform, os.environ.get("ProgramFiles", r"C:\Program Files")
-    )
-    if cmd is None:
-        _log("Docker 守护进程未运行。Linux 请手动:sudo systemctl start docker,然后重试。")
-        return False
-    if sys.platform.startswith("win") and not Path(cmd[0]).exists():
-        _log(f"找不到 Docker Desktop:{cmd[0]}。请手动启动 Docker Desktop 后重试。")
-        return False
-    _log(f"启动 Docker Desktop(首次冷启可能 30-60s)…")
-    try:
-        subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-    except OSError as e:
-        _log(f"启动 Docker Desktop 失败:{e}")
-        return False
-    deadline = time.monotonic() + DOCKER_START_TIMEOUT
-    while time.monotonic() < deadline:
-        if _docker_daemon_ok():
-            _log("Docker 守护进程已就绪。")
-            return True
-        _log("等待 Docker 守护进程…")
-        time.sleep(POLL_INTERVAL)
-    _log(f"等待 Docker 守护进程超时({DOCKER_START_TIMEOUT}s)。")
-    return False
-
-
-def _compose_up() -> bool:
-    _log("docker compose up -d(镜像缺失会自动拉取,首用约几十秒)…")
-    try:
-        subprocess.run(["docker", "compose", "up", "-d"], cwd=ROOT, check=True)
-        return True
-    except (OSError, subprocess.CalledProcessError) as e:
-        _log(f"docker compose up 失败:{e}")
-        return False
-
-
-def _pg_isready() -> bool:
-    r = subprocess.run(
-        ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
-         "pg_isready", "-U", DB_USER, "-d", APP_DB],
-        cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-    )
-    return r.returncode == 0
-
-
-def _wait_pg_ready(host: str, port: int) -> bool:
-    deadline = time.monotonic() + PG_READY_TIMEOUT
-    while time.monotonic() < deadline:
-        if _port_open(host, port) and _pg_isready():
-            _log("PostgreSQL 已就绪。")
-            return True
-        _log("等待 PostgreSQL 就绪…")
-        time.sleep(POLL_INTERVAL)
-    _log(f"等待 PostgreSQL 就绪超时({PG_READY_TIMEOUT}s)。")
-    return False
-
-
-def _ensure_test_db() -> None:
-    """幂等建测试库(兼容老 pgdata 卷首启没跑 initdb 的情况)。"""
-    check = subprocess.run(
-        ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
-         "psql", "-U", DB_USER, "-d", APP_DB, "-tAc",
-         f"SELECT 1 FROM pg_database WHERE datname='{TEST_DB}'"],
-        cwd=ROOT, capture_output=True, text=True,
-    )
-    if check.returncode == 0 and check.stdout.strip() == "1":
-        return
-    _log(f"建测试库 {TEST_DB}…")
-    subprocess.run(
-        ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
-         "psql", "-U", DB_USER, "-d", APP_DB, "-c",
-         f"CREATE DATABASE {TEST_DB} OWNER {DB_USER}"],
-        cwd=ROOT, check=False,
-    )
-
-
-def ensure(database_url: str | None = None) -> bool:
-    """确保 PG 就绪,返回 True/False。database_url 缺省从 settings 读(尊重 .env)。"""
-    if database_url is None:
-        from app.core.config import settings  # 延迟导入,避免过早固化 settings
-
-        database_url = settings.DATABASE_URL
-
-    if _is_sqlite(database_url):
-        _log("检测到 DATABASE_URL 仍是 SQLite。本地开发/测试已切 PostgreSQL,请改成:")
-        _log(f"  DATABASE_URL={SQLITE_FIX_HINT}")
-        return False
-
-    host, port = _parse_host_port(database_url)
-
-    if _port_open(host, port):
-        _log(f"✅ PostgreSQL 已在 {host}:{port} 运行,跳过 Docker。")
-        return True
-
-    _log(f"{host}:{port} 无 PostgreSQL,准备用 Docker 拉起…")
-
-    if not _docker_cli_ok():
-        _log("未检测到 docker 命令。请先安装 Docker Desktop:")
-        _log("  https://www.docker.com/products/docker-desktop/")
-        return False
-    if not _start_docker_daemon():
-        return False
-    if not _compose_up():
-        return False
-    if not _wait_pg_ready(host, port):
-        return False
-
-    _ensure_test_db()
-    return True
-
-
-if __name__ == "__main__":
-    sys.exit(0 if ensure() else 1)
-```
-
-- [ ] **Step 4: 跑测试确认通过**
-
-Run: `pytest tests/test_ensure_pg.py -v`
-Expected: 11 passed。
-
-- [ ] **Step 5: 手动冒烟(PG 已在跑时应秒过短路)**
-
-Run: `python -m scripts.ensure_pg`
-Expected: 打印 `[ensure_pg] ✅ PostgreSQL 已在 localhost:5432 运行,跳过 Docker。`,退出码 0。
-
-- [ ] **Step 6: Commit**
-
-```bash
-git add scripts/ensure_pg.py tests/test_ensure_pg.py
-git commit -m "feat(dev): scripts/ensure_pg.py 探测/拉起本地 Docker PostgreSQL"
-```
-
----
-
-## Task 3: `run.sh` / `run.bat` 接入 ensure_pg
-
-**Files:**
-- Modify: `run.sh`(在 `alembic upgrade head` 前插一步)
-- Modify: `run.bat`(同上)
-
-- [ ] **Step 1: 改 `run.sh`**
-
-在 `mkdir -p data` 之后、`"$PY" -m alembic upgrade head` 之前插入:
-
-```bash
-"$PY" -m scripts.ensure_pg   # 确保本地 Docker PostgreSQL 就绪(没起会自动拉起;失败即退出)
-```
-
-(`set -e` 已在文件顶部,ensure_pg 失败会自动终止脚本。)
-
-- [ ] **Step 2: 改 `run.bat`**
-
-在 `if not exist data mkdir data` 之后、`call "%PY%" -m alembic upgrade head` 之前插入:
-
-```bat
-REM 确保本地 Docker PostgreSQL 就绪(没起会自动拉起 Docker + PG 容器)
-call "%PY%" -m scripts.ensure_pg
-if errorlevel 1 (
-    echo [X] ensure_pg failed ^(PostgreSQL 未就绪^)
-    exit /b %errorlevel%
-)
-```
-
-- [ ] **Step 3: 验证 `run.sh`(PG 已在跑,应短路后继续 alembic + uvicorn)**
-
-Run(Git Bash):`bash run.sh 8770`
-Expected: 依次出现 `[ensure_pg] ✅ PostgreSQL 已在 localhost:5432 运行` → alembic 无报错 → uvicorn `Application startup complete`。`Ctrl-C` 停。
-
-- [ ] **Step 4: 验证 `run.bat`(同上,Windows 原生)**
-
-Run(cmd/PowerShell):`.\run.bat 8770`
-Expected: 同 Step 3。`Ctrl-C` 停。
-
-- [ ] **Step 5: Commit**
-
-```bash
-git add run.sh run.bat
-git commit -m "feat(dev): run.sh/run.bat 启动前确保 Docker PostgreSQL 就绪"
-```
-
----
-
-## Task 4: `.env.example` 默认切 PostgreSQL
-
-**Files:**
-- Modify: `.env.example`(第 8-10 行「数据库」段)
-
-- [ ] **Step 1: 改 `.env.example` 的 DATABASE_URL**
-
-把:
-```ini
-# ===== 数据库 =====
-# SQLite 本地文件路径。生产环境用 /opt/shaguabijia-app-server/data.db
-DATABASE_URL=sqlite:///./data/app.db
-```
-改成:
-```ini
-# ===== 数据库 =====
-# 本地开发/测试统一用 Docker PostgreSQL:run.bat/run.sh 会自动拉起容器
-# (docker-compose.yml + scripts/ensure_pg.py)。详见 docs/database/postgres-migration.md。
-# 生产用原生 PG,由 scripts/init_postgres.py 写入强随机密码的连接串。
-# ⚠️ scheme 必须是 postgresql+psycopg://(psycopg3);不要写成 postgresql://(会去找未装的 psycopg2)。
-DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia
-```
-
-- [ ] **Step 2: 验证(新 .env 从模板复制后能起服务)**
-
-Run: `cp .env.example /tmp/env.check && grep '^DATABASE_URL=' /tmp/env.check`
-Expected: `DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia`
-
-- [ ] **Step 3: Commit**
-
-```bash
-git add .env.example
-git commit -m "feat(dev): .env.example 默认 DATABASE_URL 切 Docker PostgreSQL"
-```
-
----
-
-## Task 5: `tests/conftest.py` 切 PostgreSQL 测试库
-
-**Files:**
-- Modify: `tests/conftest.py`(整体替换:去掉临时 SQLite,改指 `shaguabijia_test` + 调 ensure_pg + fixture 改 drop/create)
-
-- [ ] **Step 1: 整体替换 `tests/conftest.py`**
-
-```python
-"""测试用 fixtures。
-
-测试库用 Docker PG 的 shaguabijia_test(与 dev 业务库 shaguabijia 隔离)。
-顺序(必须):设 test DATABASE_URL(在 import app.* 之前)→ ensure PG 就绪 →
-import app → 建表。持久卷可能残留上次的表 → session 开头先 drop 再 create。
-"""
-from __future__ import annotations
-
-import os
-from collections.abc import Iterator
-
-# 1) 测试库连接串——必须在 import app.* 之前设好(app.db.session 在 import 期建 engine)
-_TEST_DB_URL = (
-    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test"
-)
-os.environ["DATABASE_URL"] = _TEST_DB_URL
-os.environ.setdefault("JWT_SECRET_KEY", "test-secret-please-ignore-this-is-only-for-pytest-not-real")
-os.environ.setdefault("ADMIN_JWT_SECRET", "test-admin-secret-please-ignore-only-for-pytest-not-real")
-os.environ.setdefault("JG_APP_KEY", "test-key")
-os.environ.setdefault("JG_MASTER_SECRET", "test-secret")
-os.environ.setdefault("SMS_MOCK", "true")
-os.environ.setdefault("APP_ENV", "dev")
-os.environ.setdefault("APP_DEBUG", "false")  # 测试不打 SQL 日志
-os.environ.setdefault("WECHAT_APP_ID", "wxtest0000000000")
-os.environ.setdefault("WECHAT_APP_SECRET", "test-secret")
-os.environ.setdefault("WXPAY_MCH_ID", "test-mch")
-os.environ.setdefault("WXPAY_MCH_SERIAL_NO", "test-serial")
-os.environ.setdefault("WXPAY_PUBLIC_KEY_ID", "test-pubkey-id")
-os.environ.setdefault("RATE_LIMIT_ENABLED", "false")
-os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
-os.environ.setdefault("PANGLE_REWARD_SECRET", "test-pangle-secret-only-for-pytest")
-
-# 2) 保证 Docker PG 就绪 + 测试库存在(必须在 import app.db.session 建 engine 之前)
-from scripts.ensure_pg import ensure
-
-if not ensure(_TEST_DB_URL):
-    raise RuntimeError(
-        "测试需要 Docker PostgreSQL 就绪。请确认已装 Docker Desktop;"
-        "或先跑一次 run.bat/run.sh 把 PG 拉起,再重试 pytest。"
-    )
-
-import pytest
-from fastapi.testclient import TestClient
-
-from app.db.base import Base
-from app.db.session import engine
-from app.main import app
-
-
-@pytest.fixture(scope="session", autouse=True)
-def _setup_db() -> Iterator[None]:
-    # 持久卷可能残留上次跑崩后的表/数据 → 先 drop 再 create,保证干净起点
-    Base.metadata.drop_all(engine)
-    Base.metadata.create_all(engine)
-    yield
-    Base.metadata.drop_all(engine)
-
-
-@pytest.fixture()
-def client() -> TestClient:
-    return TestClient(app)
-```
-
-- [ ] **Step 2: 验证 conftest 能引导 PG 并收集用例(选一个不涉 DB 的测试文件)**
-
-Run: `pytest tests/test_ensure_pg.py -v`
-Expected: conftest 先打印 `[ensure_pg] ✅ PostgreSQL 已在 localhost:5432 运行`(或拉起过程),随后 12 passed。说明「测试走 PG 引导」链路通、且纯函数测试不受影响。
-
-- [ ] **Step 3: 验证建表落到 PG 测试库(跑一个 DB 相关用例)**
-
-Run: `pytest tests/test_invite.py -v`
-Expected: 用例在 `shaguabijia_test` 上建表并执行(可能有个别红,留待 Task 6);关键是不再出现 SQLite 临时文件、engine 连的是 PG。
-
-- [ ] **Step 4: Commit**
-
-```bash
-git add tests/conftest.py
-git commit -m "test(dev): conftest 切 shaguabijia_test(Docker PG),引导+drop/create"
-```
-
----
-
-## Task 6: 全量跑 pytest on PG,逐个修红用例
-
-> SQLite 宽松、PG 严格,切库会暴露一批真 bug(迁移指南 §2.2 已列)。本任务是**发现驱动**:先跑全量、按类别归因、按下述配方修,直到全绿。修改范围限被测业务/模型代码,不改测试来掩盖真 bug(除非测试本身依赖 SQLite 特性,如秒级时间精度)。
-
-**Files:**
-- Modify: 视失败而定(常见:`app/models/*.py`、`app/**/repositories/*.py`、少量 `tests/*.py`)
-
-- [ ] **Step 1: 全量跑,拿到失败清单**
-
-Run: `pytest -q`
-Expected: 大部分通过;记录所有 FAIL 的用例名与报错文本,按下面类别归因。
-
-- [ ] **Step 2: 修「naive datetime / 时区」类**
-
-定位:`git grep -n "utcnow()" app/`。把 `datetime.utcnow()` 改成 `datetime.now(timezone.utc)`(并 `from datetime import timezone`)。
-症状:PG `TIMESTAMPTZ` 与 naive datetime 比较/写入报错或结果错位;`tests/test_cps_admin.py` 已注释过 SQLite 忽略 tzinfo 的行为。
-
-例:
-```python
-# 改前
-from datetime import datetime
-ts = datetime.utcnow()
-# 改后
-from datetime import datetime, timezone
-ts = datetime.now(timezone.utc)
-```
-
-- [ ] **Step 3: 修「字符串/整数隐式比较」类**
-
-症状:SQLite 允许 `WHERE phone = 13800138000`(自动转型),PG 直接报类型错。定位报错用例引用的查询,确保比较两侧类型一致(手机号等一律按字符串传参 `:phone`,不要传裸 int)。
-
-- [ ] **Step 3.5: 修「事务已中止」类**
-
-症状:某用例后续报 `current transaction is aborted, commands ignored until end of transaction block`,根因是前一句 SQL 出错后业务代码缺 `db.rollback()`/`db.commit()` 边界。补上正确的 commit/rollback。
-
-- [ ] **Step 4: 修「测试依赖 SQLite 特性」类(仅此类可改测试)**
-
-症状:测试断言依赖 SQLite 秒级时间精度或 FK 不强制(见 `test_invite.py:328`、`test_compare_harvest.py:153` 的注释)。PG 下时间精度更高/FK 更严——调整测试数据(如手动拉开时间间隔、用合法 FK)使断言在 PG 下成立,不改业务逻辑。
-
-- [ ] **Step 5: 反复跑到全绿**
-
-Run: `pytest -q`
-Expected: `N passed`(0 failed)。若仍有红,回到 Step 2-4 继续归因。
-
-- [ ] **Step 6: Commit**
-
-```bash
-git add -A
-git commit -m "fix(db): 测试套件切 PostgreSQL 后修复严格性暴露的用例"
-```
-
----
-
-## Task 7: 文档更新
-
-**Files:**
-- Modify: `docs/database/postgres-migration.md`(§1 增「本地 Docker 一键起」小节)
-- Modify: `CLAUDE.md`(DB 段注明 dev/test = Docker PG)
-- Modify: `scripts/init_postgres.py`(顶部注释区分生产/本地)
-
-- [ ] **Step 1: `postgres-migration.md` 在「## 1. 本地起 PG」开头插入推荐做法**
-
-在 `## 1. 本地起 PG + 跑通空库(半天)` 标题下、`### 1.1 装 PG` 之前插入:
-
-```markdown
-### 1.0 推荐:Docker 一键起(本地开发/测试)
-
-本地开发不必手动装 PG。已提供 `docker-compose.yml` + `scripts/ensure_pg.py`:
-
-```bash
-cp .env.example .env    # DATABASE_URL 默认已是 Docker PG 连接串
-./run.sh                # 或 run.bat;会自动:探测 PG → 没起则启 Docker → 起 PG 容器 → 建库 → alembic → uvicorn
-pytest                  # conftest 自动引导同一容器的 shaguabijia_test 库
-```
-
-容器:`postgres:16-alpine`(名 `shaguabijia-pg`,端口 5432,命名卷 `pgdata` 持久化),
-首启即建业务库 `shaguabijia` 与测试库 `shaguabijia_test`。下面 1.1-1.5 的手动装 PG 步骤仅在
-不用 Docker 时才需要;生产仍走 §4 的原生 PG。
-```
-
-- [ ] **Step 2: `CLAUDE.md` DB 段补充**
-
-找到 DB 相关行(`**Prod**: PostgreSQL — just change DATABASE_URL...`)所在段,在其上方加一行:
-
-```markdown
-- **Dev/Test**: Docker PostgreSQL 16 — `run.sh`/`run.bat` 经 `scripts/ensure_pg.py` + `docker-compose.yml` 自动拉起;`.env.example` 默认即 PG 连接串;pytest 用同容器的 `shaguabijia_test` 库。**本地不再用 SQLite**。
-```
-
-- [ ] **Step 3: `scripts/init_postgres.py` 顶部注释区分场景**
-
-把模块 docstring 第一行下方(`新机器初始化用。前置:...` 那行)改为:
-
-```python
-新机器初始化用(面向【生产原生 PG】:apt/systemd 装好的 PostgreSQL)。
-本地开发/测试请改用 docker-compose.yml + scripts/ensure_pg.py(run.sh/run.bat 自动拉起),不必跑本脚本。
-前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
-```
-
-- [ ] **Step 4: 验证无坏链接/格式**
-
-Run: `git diff --stat`
-Expected: 三个文档文件有改动,无其他文件被误改。
-
-- [ ] **Step 5: Commit**
-
-```bash
-git add docs/database/postgres-migration.md CLAUDE.md scripts/init_postgres.py
-git commit -m "docs(dev): 记录本地 Docker PostgreSQL 用法,区分生产原生 PG 路径"
-```
-
----
-
-## 完成标准(对齐 spec §8 验收)
-
-- [ ] 全新机器(装了 Docker Desktop、`.env` 从 `.env.example` 复制)跑 `run.bat`/`run.sh` 全自动拉起 PG 并起服务,无手动装 PG。
-- [ ] `docker ps` 见 `shaguabijia-pg` healthy;`shaguabijia` 与 `shaguabijia_test` 两库都在。
-- [ ] PG 已在跑时再跑 `run`,ensure_pg 秒过短路。
-- [ ] `pytest -q` 全绿(连 `shaguabijia_test`)。
-- [ ] `.env` 改回 sqlite 时,`python -m scripts.ensure_pg` 硬失败并打印正确 PG 串。
-- [ ] 能在 `admin/repositories` 写一段 PG 专有聚合(如 `count(*) FILTER (WHERE ...)`),`run` 手动跑通且相应 pytest 通过。
-
-## Self-Review 记录(计划作者已核)
-
-- **Spec 覆盖:** spec §9 待实现清单 8 项 → Task 1(compose+initdb)、Task 2(ensure_pg)、Task 3(run 接线)、Task 4(.env.example)、Task 5(conftest)、Task 6(修红用例)、Task 7(文档);「session.py 无需改」在 Header 与 spec §4.7 说明;「data/ 忽略」在 Task 1 Step 5 验证。无遗漏。
-- **占位符:** 全部步骤含真实代码/命令/期望输出。Task 6 是发现驱动,已用「类别+具体转换配方+定位命令」代替不可预知的逐条 diff——非占位。
-- **类型/命名一致:** `ensure(database_url=None)` 签名在 Task 2 定义,Task 5 以 `ensure(_TEST_DB_URL)` 调用一致;库名/用户/密码/端口全程为 Header 固定值;compose 服务名 `postgres` 与 ensure_pg `COMPOSE_SERVICE` 一致;`shaguabijia_test` 在 initdb SQL、`_ensure_test_db()`、conftest 三处一致。

docs/superpowers/specs/2026-07-08-local-dev-postgres-docker-design.md

@@ -1,281 +0,0 @@
-# 本地开发切 Docker PostgreSQL —— 设计文档
-
-> 让本地开发与测试统一跑在 Docker 化的 PostgreSQL 上,彻底退掉 SQLite。
-> `run.bat` / `run.sh` 启动时自动检测本机 PG,没起就拉起 Docker → 起 PG 容器(镜像缺失先拉),
-> 目的是让开发/大模型能放心用 PG 专有的高效聚合函数,不再为兼容 SQLite 而退化成"取基础数据后内存聚合"。
->
-> 状态:已定稿(待用户复核)。作者对话日期:2026-07-08。
-> 关联:[postgres-migration.md](../../database/postgres-migration.md)(切引擎完整步骤)、`scripts/init_postgres.py`(生产原生 PG 初始化)。
-> ⚠️ **2026-07-27 增补(见 §10)**:D4 已从「sqlite 硬失败」松为「显式 SQLite 逃生舱」。§1-9 描述的是初版「彻底退掉 SQLite」设计;凡涉及「无 Docker / DATABASE_URL 是 sqlite 时如何处理」,**以 §10 为准**(测试仍只跑 PG 不变)。
-
----
-
-## 1. 背景与目标
-
-### 问题
-当前开发环境默认用 SQLite(`DATABASE_URL=sqlite:///./data/app.db`),生产用 PostgreSQL 16。两套引擎并存,导致写数据访问代码时(尤其 `app/admin/repositories/` 的报表聚合)为了"两边都能跑",放弃 PG 专有能力(窗口函数、`FILTER`、`JSONB` 操作符、`GROUPING SETS` 等),改成"先查基础数据、再在 Python 内存里聚合"——既慢又啰嗦。
-
-### 目标
-本地开发与测试都跑在 PG 上,SQLite 退出本地开发闭环。之后写 PG 专有 SQL 时:
-- 开发运行时(`run.bat`/`run.sh`)直接连 PG,手动验证可行;
-- `pytest` 也连 PG,PG 专有 SQL 在被测代码路径里也安全,不会因 SQLite 而挂——**这是"双库兼容代码彻底消失"的必要条件**。
-
-### 非目标(本期不做)
-- 不动**生产**部署(生产仍是原生 PG16 + systemd,无 Docker;`init_postgres.py` 保持不变)。
-- 不做数据搬迁(MVP 阶段无真实用户数据,详见迁移指南背景假设)。
-- 不接 CI(仓库当前无 `.github/workflows`;若将来加 CI,再单独让 CI 起 PG service)。
-- 不引入 Redis / testcontainers / 连接池中间件。
-
----
-
-## 2. 决策记录(本次对话已拍板)
-
-| # | 决策点 | 结论 | 理由 |
-|---|---|---|---|
-| D1 | PG 覆盖范围 | **dev 运行 + 测试都切 PG** | 只切运行时的话,被 SQLite 测试覆盖的代码路径(如 `test_cps_admin.py` 覆盖的 `admin/repositories/cps.py`)仍不能用 PG 专有 SQL,双库代码不会真正消失 |
-| D2 | 打包方式 | **方案 A:Compose + `scripts/ensure_pg.py`** | 唯一真正需要定制的部分(启动 Docker 守护进程、等 PG 就绪)集中到一个跨平台模块,`run.bat`/`run.sh`/`conftest.py` 共用;声明式的容器/卷/健康检查交给 Compose |
-| D3 | 宿主端口 | **5432**(与生产/文档一致) | 边界:若本机已有原生 PG 占 5432,`ensure_pg` 会探测到"PG 已在"直接复用它(可能连到不带业务库的实例)——见 §7 风险,文档提示 |
-| D4 | dev 下 `DATABASE_URL` 仍是 sqlite | **硬失败**(打印一行 fix 后非 0 退出) | 彻底断掉 SQLite 退路,符合"让大家都用 PG"的目标 |
-| D5 | dev 数据库密码 | 固定 `shaguabijia_dev_pw`,写进 compose + `.env.example` | 本地容器仅绑 `localhost`,非机密;保证 `.env.example` 复制即可用。生产密码另由 `init_postgres.py` 强随机生成,不复用 |
-| D6 | 改哪些启动脚本 | `run.bat` 与 `run.sh` **都改** | 仓库一贯保持两者同步 |
-| D7 | 镜像 | `postgres:16-alpine` | 对齐生产 PG16;alpine 体积小 |
-
----
-
-## 3. 现状(改动前)
-
-- **配置**:`app/core/config.py` `DATABASE_URL` 默认 `sqlite:///./data/app.db`,pydantic-settings 从 `.env` 读(环境变量优先级高于 `.env` 文件)。
-- **引擎**:`app/db/session.py` 用 `_is_sqlite = DATABASE_URL.startswith("sqlite")` 分流——SQLite 加 `check_same_thread=False`、不建池;非 SQLite 加 `pool_size=10/max_overflow=20/pool_recycle=3600`。**已天然支持 PG,无需改。**
-- **迁移**:`alembic/env.py` 从 `settings.DATABASE_URL` 读连接串,`render_as_batch` 仅对 sqlite 开;PG 下自动关。**无 psycopg2 硬编码,切 PG 无需改。**
-- **驱动**:`pyproject.toml` 已装 `psycopg[binary]>=3.1`(psycopg3)。URL scheme 必须 `postgresql+psycopg://`(裸 `postgresql://` 会被 SQLAlchemy 路由到未安装的 psycopg2 → ModuleNotFoundError)。
-- **测试**:`tests/conftest.py` 在 import app 前把 `DATABASE_URL` 设成临时文件 SQLite;session 级 autouse fixture 做 `Base.metadata.create_all(engine)` / 结束 `drop_all`(schema 来自 model 而非 alembic,无逐用例 rollback,全会话共享一个库)。
-- **启动脚本**:`run.bat` / `run.sh` 均为:校验 `.env` 存在 → `mkdir data` → `alembic upgrade head` → uvicorn 监听 `0.0.0.0:8770`(`.sh` 带 `--reload --reload-dir app`)。
-- **现有 PG 资产**:`scripts/init_postgres.py`(交互式:建用户/建库/授权/写 .env/跑迁移,面向**已装好的原生 PG**)、`docs/database/postgres-migration.md`(切引擎完整步骤,含 §2 测试切 PG、§2.2 会暴露的真 bug 清单)。
-- **CI**:无(`.github/workflows` 不存在),故测试切 PG 无 CI 联动负担。
-
----
-
-## 4. 方案详解
-
-### 4.1 新增 `docker-compose.yml`(app-server 根目录)
-```yaml
-services:
-  postgres:
-    image: postgres:16-alpine
-    container_name: shaguabijia-pg
-    environment:
-      POSTGRES_USER: shaguabijia_app
-      POSTGRES_PASSWORD: shaguabijia_dev_pw
-      POSTGRES_DB: shaguabijia
-    ports:
-      - "5432:5432"
-    volumes:
-      - pgdata:/var/lib/postgresql/data
-      - ./docker/initdb:/docker-entrypoint-initdb.d:ro
-    healthcheck:
-      test: ["CMD-SHELL", "pg_isready -U shaguabijia_app -d shaguabijia"]
-      interval: 3s
-      timeout: 3s
-      retries: 20
-volumes:
-  pgdata:
-```
-- `POSTGRES_USER` 设定后,该用户以超级用户身份创建并拥有 `POSTGRES_DB`,故能再建测试库。
-- 首启 initdb 脚本建测试库(见 4.2)。命名卷 `pgdata` 让数据跨重启留存。
-
-### 4.2 新增 `docker/initdb/01-create-test-db.sql`
-```sql
--- 仅在 pgdata 卷首次初始化时执行一次。以 shaguabijia_app(超级用户)连 shaguabijia 库运行。
-CREATE DATABASE shaguabijia_test OWNER shaguabijia_app;
-```
-
-### 4.3 新增 `scripts/ensure_pg.py`(纯标准库 + docker CLI,跨平台)
-对外同时暴露**可导入函数** `ensure()`(供 `conftest.py` 直接调)和 **CLI 入口** `if __name__ == "__main__": sys.exit(0 if ensure() else 1)`(供 `run` 以 `python -m scripts.ensure_pg` 跑)。`ensure()` 从 `app.core.config.settings` 读 `DATABASE_URL`,解析 host/port,主流程:
-1. **sqlite 守卫**:若 `DATABASE_URL` 以 `sqlite` 开头 → 打印"dev 已切 PG,请把 .env 的 DATABASE_URL 改成 `postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia`"→ 非 0 退出(D4)。
-2. **TCP 探测** `host:port`(stdlib `socket`,超时 1s)。通 → 打印"✅ PG 已就绪"直接返回(幂等:PG 已在跑时开销≈一次握手)。
-3. 不通 → `docker version` 探 CLI;缺失 → 中文报错"请先安装 Docker Desktop:https://www.docker.com/products/docker-desktop/" → 非 0 退出。
-4. `docker info` 探守护进程;不通 → 按平台启动:
-   - Windows:`start "" "%ProgramFiles%\Docker\Docker\Docker Desktop.exe"`(找不到则报错让用户手动开)
-   - macOS:`open -a Docker`
-   - Linux:不自动 sudo,打印 `sudo systemctl start docker` 让用户执行后重试
-   然后轮询 `docker info` 直到就绪或超时(默认 120s,每 3s 一次,打印进度)。
-5. `docker compose up -d`(Compose 在镜像缺失时**自动拉取**,首用拉 alpine ~90MB;有进度输出)。
-6. 轮询 healthcheck(`docker inspect` 的 health 状态)/ TCP 直到 PG 接受连接(默认 60s 超时)。
-7. **幂等确保测试库存在**(兼容"老 pgdata 卷没跑过 initdb"的情况):
-   `docker compose exec -T postgres psql -U shaguabijia_app -tc "SELECT 1 FROM pg_database WHERE datname='shaguabijia_test'"`,不存在则 `CREATE DATABASE shaguabijia_test OWNER shaguabijia_app`。
-
-失败即清晰中文报错 + 非 0 退出,**全程不回退 SQLite**。所有超时可用环境变量覆盖(如 `ENSURE_PG_DOCKER_TIMEOUT`)。
-
-### 4.4 `run.bat` / `run.sh` 接线
-在 `alembic upgrade head` **之前**插一行调用,失败即退出:
-- `run.sh`:`"$PY" -m scripts.ensure_pg`(`set -e` 已在,失败自动退出)
-- `run.bat`:`call "%PY%" -m scripts.ensure_pg` + `if errorlevel 1 exit /b 1`
-
-其余逻辑不动(`mkdir data` 保留给 media 等落盘目录)。
-
-### 4.5 `.env.example` 默认切 PG
-把
-```ini
-DATABASE_URL=sqlite:///./data/app.db
-```
-改为
-```ini
-# 本地开发/测试统一用 Docker PG(run.bat/run.sh 会自动拉起容器;详见 docs/database/postgres-migration.md §本地 Docker 一键起)。
-# 生产用原生 PG,由 scripts/init_postgres.py 写入强随机密码的连接串。
-DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia
-```
-
-### 4.6 `tests/conftest.py` 切 PG
-调整顶部顺序(仍必须在 `import app.*` 之前完成 env 设定):
-1. 设 `os.environ["DATABASE_URL"] = "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test"`(测试库,永不碰 dev 业务库)。
-2. 调 `scripts.ensure_pg.ensure()`(保证容器在 + 测试库在;PG 已在时几乎零开销)。
-3. `import app...`。
-session 级 autouse fixture:改为 **`Base.metadata.drop_all(engine)` → `create_all(engine)`(开头先清干净,防持久卷里上一次跑残留的表/数据)→ yield → 结束 `drop_all`**;删掉临时 SQLite 文件相关代码。
-
-> 预期:部分用例会因 PG 的严格性变红(SQLite 宽松、PG 严格),按迁移指南 §2.2 逐个修——常见为:字符串/整数隐式比较、`datetime.utcnow()` naive vs `TIMESTAMPTZ`、事务边界(`current transaction is aborted`)。这既是工作量也是本次改造的**直接收益**(暴露真 bug)。实现阶段需为"跑 pytest 并修红用例"单列步骤。
-
-### 4.7 `app/db/session.py`
-**无需改动**——`_is_sqlite` 为假时自动走 PG 池化分支。
-
-### 4.8 文档
-- `docs/database/postgres-migration.md` 增一节「本地 Docker 一键起 PG(推荐)」,指向 compose + `ensure_pg`,并说明它替代了 §1.1 的手动 brew/apt 装 PG。
-- `CLAUDE.md` 的 DB 段注明:dev/test = Docker PG(`run` 自动拉起);prod = 原生 PG(`init_postgres.py`)。
-- `scripts/init_postgres.py` 顶部注释补一句"本脚本面向生产原生 PG;本地开发用 docker-compose + scripts/ensure_pg"。
-
----
-
-## 5. 连接参数汇总
-
-| 项 | 值 |
-|---|---|
-| 镜像 | `postgres:16-alpine` |
-| 容器名 | `shaguabijia-pg` |
-| 宿主端口 | `5432` |
-| 超级/业务用户 | `shaguabijia_app` |
-| dev 密码 | `shaguabijia_dev_pw`(本地非机密) |
-| 业务库(dev 运行) | `shaguabijia` |
-| 测试库(pytest) | `shaguabijia_test` |
-| dev `DATABASE_URL` | `postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia` |
-| test `DATABASE_URL` | `postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test` |
-| 数据持久化 | 命名卷 `pgdata` |
-
----
-
-## 6. 失败处理矩阵(无 SQLite 兜底)
-
-| 情形 | ensure_pg 行为 |
-|---|---|
-| `DATABASE_URL` 是 sqlite | 打印应改成的 PG 串 → 非 0 退出 |
-| PG 已在跑(TCP 通) | 打印"已就绪" → 返回 0(跳过 docker) |
-| 无 docker CLI | 提示装 Docker Desktop + 官网链接 → 非 0 退出 |
-| docker 守护进程未起 | 尝试按平台启动 Docker Desktop,轮询到就绪;超时则报错 → 非 0 退出 |
-| 镜像缺失 | `docker compose up -d` 自动拉取(不额外处理) |
-| 容器起了但 PG 未 ready | 轮询 healthcheck 到超时;超时报错 → 非 0 退出 |
-| 老 pgdata 卷缺测试库 | 幂等 `CREATE DATABASE shaguabijia_test` |
-
----
-
-## 7. 风险与边界
-
-- **端口占用(原生 PG 撞 5432)**:D3 选了 5432。若开发机已有原生 PG 监听 5432,step 2 的 TCP 探测会判"PG 已在"并复用它——但那个实例可能没有 `shaguabijia`/`shaguabijia_test` 库或用户,后续 `alembic upgrade head` / 测试会报连不上库或认证失败。**缓解**:文档提示"本机别再单独跑原生 PG";报错信息里提示检查是不是撞了原生 PG。
-- **首次启动慢**:首用需 Docker Desktop 冷启(~30–60s)+ 拉镜像(~数十秒~数分钟,视网络)。`ensure_pg` 全程打印进度,超时可配。
-- **`DATABASE_URL` 环境变量优先级**:pydantic-settings 里 shell 环境变量优先于 `.env`。若开发者 shell 残留旧的 `DATABASE_URL`(如指向 sqlite),会盖过 `.env`。sqlite 守卫(D4)能挡住 sqlite 残留;但若残留的是另一个 PG 串,则以它为准——文档提示。
-- **持久卷脏状态**:测试用 drop_all→create_all 开头清库,避免上次崩溃残留污染;dev 业务库随卷留存(符合预期)。
-- **Docker 未安装/公司网络拉镜像受限**:硬失败并给出明确指引;不提供 SQLite 退路是刻意选择(D4/目标)。
-
----
-
-## 8. 验收标准
-
-1. 全新机器(装了 Docker Desktop、`.env` 从 `.env.example` 复制)执行 `run.bat`(或 `run.sh`):自动拉起 Docker→起 PG 容器→建库→`alembic upgrade head`→uvicorn 起在 8770,无手动装 PG 步骤。
-2. `docker ps` 见 `shaguabijia-pg` 健康;`psql`/客户端能连 `shaguabijia` 与 `shaguabijia_test` 两个库。
-3. PG 已在跑时再次 `run`,`ensure_pg` 秒过(不重复拉容器)。
-4. `pytest` 连 `shaguabijia_test` 跑;红用例全部修绿(PG 严格性暴露的问题)。
-5. `.env` 的 `DATABASE_URL` 改回 sqlite 时,`run`/`pytest` 硬失败并打印正确的 PG 串。
-6. 能在 `admin/repositories/` 里写一段 PG 专有聚合 SQL(如带 `FILTER (WHERE ...)` 的聚合),`run` 下手动跑通、相应 pytest 也通过——即"双库兼容负担消失"的实证。
-
----
-
-## 9. 待实现清单(供 writing-plans 拆解)
-
-- [ ] 新增 `docker-compose.yml`
-- [ ] 新增 `docker/initdb/01-create-test-db.sql`
-- [ ] 新增 `scripts/ensure_pg.py`(TCP 探测 / 启 Docker Desktop 轮询 / compose up / 等 healthy / 幂等建测试库 / sqlite 守卫)
-- [ ] `run.sh`、`run.bat` 接入 `ensure_pg`
-- [ ] `.env.example` 的 `DATABASE_URL` 切 PG
-- [ ] `tests/conftest.py` 切 `shaguabijia_test` + 调 `ensure_pg` + fixture 改 drop/create
-- [ ] 跑 `pytest`,按迁移指南 §2.2 修红用例
-- [ ] 文档:`postgres-migration.md` 增「本地 Docker 一键起」节;`CLAUDE.md` DB 段;`init_postgres.py` 注释
-- [ ] `.gitignore` 确认 `data/` 已忽略(compose 用命名卷,不落项目目录,无需额外忽略)
-
----
-
-## 10. 增补(2026-07-27):D4 反转 —— 显式 SQLite 逃生舱
-
-> 背景:§2 的 D4 定为「dev 下 `DATABASE_URL` 仍是 sqlite → 硬失败」,目的是彻底断掉 SQLite 退路。实践中这对「本机装不了 Docker」的开发者过于刚性——直接被卡死、连跑都跑不起来。本次(2026-07-27 对话)把 D4 从「硬失败」松成「**显式逃生舱**」:工具**从不替你静默切库**,但会在没 Docker 时告诉你怎么手动降级,且降级时每次启动都醒目告警。
-
-### 10.1 决策更新
-
-| # | 原决策 | 新决策 | 理由 |
-|---|---|---|---|
-| D4′ | dev sqlite URL → 硬失败退出 | **放行 + 每次打印醒目降级横幅**(仍非静默) | 已手动改 `.env`=sqlite = 开发者的显式选择,尊重它;但吼一嗓子防止忘了自己在降级、把 PG 专有 SQL 提交上去 |
-| D8(新) | (无) | 无 docker CLI 时,报错里**追加逃生舱指路**(改 `.env`=sqlite),但仍非 0 退出 | 「显式」的关键:工具不替你切库,只指路;开发者改完 `.env` 再跑一次才真正降级 |
-
-**未变**:D1(测试仍只跑 PG)、D2-D3、D5-D7 全部保留。逃生舱**只作用于 `run.sh`/`run.bat` 运行时**;`pytest` 仍写死连 PG 测试库(`conftest.py` 传 PG URL,sqlite 分支根本不触发),没 Docker 就 `raise`、跑不了完整套件——这正是 D1「测试上 PG 才能暴露真 bug」的初衷,刻意不给逃生舱。
-
-### 10.2 代码改动(仅 `scripts/ensure_pg.py` 的 `ensure()`)
-
-1. **sqlite 分支**(原 `return False`)→ 打印多行降级横幅后 `return True`。横幅点明:PG 专有 SQL/严格类型在此模式**不被验证**、提交前须在有 Docker 的机器上用 PG 复跑、装好 Docker 后把 `DATABASE_URL` 改回 PG 串。
-2. **无 docker CLI 分支**(原仅提示装 Docker + `return False`)→ 追加一句「装不了 Docker?把 `.env` 的 `DATABASE_URL` 改成 `sqlite:///./data/app.db` 可降级运行」;**仍 `return False`**(run 脚本照常退出,开发者需显式改 .env 再跑)。
-3. **常量**:新增 `SQLITE_URL = "sqlite:///./data/app.db"`(逃生舱指路用);`SQLITE_FIX_HINT` 重命名 `PG_URL`(降级横幅"改回 PG"引用)。
-4. 更新模块 docstring 中「全程无 SQLite 兜底」一句,改述为「无 Docker/sqlite URL 时【显式】降级 SQLite(带醒目告警),测试侧不降级」。
-
-**其余全不动**:`run.sh`/`run.bat`(sqlite 下 `ensure` 返 True → 照常 `alembic upgrade head` + uvicorn)、`docker-compose.yml`、`app/db/session.py`(SQLite 引擎分支本就保留为 fallback)、`tests/conftest.py`、`.env.example`(默认仍 PG)。
-
-### 10.3 改完后行为矩阵(覆盖用户列的 5 场景)
-
-| 场景 | `DATABASE_URL` | ensure_pg 行为 |
-|---|---|---|
-| ① 无 Docker | PG(默认) | 报错 + 指逃生舱 → 退出;开发者改 `.env`=sqlite → 再跑 → **放行 + 降级横幅**,alembic/uvicorn 跑 SQLite |
-| ② 有 Docker 未启动 | PG | 启 Docker Desktop → `compose up` → 等 ready → 建测试库(**不变**) |
-| ③ 有 Docker 已启动 | PG | `compose up` → 等 ready(**不变**) |
-| ④ PG 已在跑 | PG | TCP 通 → 秒过跳过 Docker(**不变**) |
-| ⑤ PG 起来后 | 任意 | run 脚本 `alembic upgrade head`(**不变**;SQLite 走 `render_as_batch`) |
-
-### 10.4 风险
-
-- **降级被忽视**:横幅仅在 `run` 启动时打印一次;若开发者用 IDE 直接起 uvicorn(绕过 run 脚本)则看不到。缓解:横幅足够醒目 + 文档强调;**不**引入 app 启动期重复告警(YAGNI)。
-- **测试无 Docker 跑不了**:刻意保留(D1)。文档提示无 Docker 者:要么装 Docker 跑全量测试,要么只在 CI/有 Docker 的机器上验证 PG 相关改动。
-
-### 10.5 Redis 前瞻(不在本次)
-
-§2 未涉及 Redis。②③ 场景未来若加 Redis 实例:在 `docker-compose.yml` 增 `redis` 服务即可,`docker compose up -d` 天然带起;仅当启动期有组件依赖 Redis 才需给 `ensure_pg` 加 redis readiness 探测。本次不做,方案对它友好。
-
-### 10.6 附带修复:`_docker_cli_ok` 守护进程误判(2026-07-27)
-
-诊断「装了 Docker Desktop 却报未检测到 docker」时发现的真 bug:`_docker_cli_ok()` 原用 `docker version`
-判断 CLI 是否存在,但该命令**要连 daemon**,守护进程没起时退非零 → 把「Docker 装了但没启动」
-误判成「没装 CLI」,`ensure()` 直接打印"请安装 Docker Desktop"并 `return False`,**绕过了专为需求②
-写的 `_start_docker_daemon()` 自动拉起逻辑**——需求②(有 Docker 未启动 → 自动启动)因此从未真正生效。
-修复:改用 `docker --version`(纯客户端、不连 daemon、退 0)。`_docker_daemon_ok()` 仍用 `docker info`
-(正确,该检查本就依赖 daemon)。实测机器:Docker Desktop 20.10.12 已装但引擎未起,修复前 `_docker_cli_ok()`
-误报 False,修复后 True。
-
-### 10.7 附带修复:固定 compose 项目名 + 清理残留同名容器(2026-07-27)
-
-诊断「`docker compose up` 报 `container name "/shaguabijia-pg" already in use`」时发现的又一 bug:compose
-项目名默认取运行目录 basename,在不同目录/worktree(如 `local-dev-postgres-docker` vs `shaguabijia-app-server`)
-之间切换会各自成一个项目;而 `docker-compose.yml` 写死了 `container_name: shaguabijia-pg`(全局唯一名),
-于是新项目 `up` 时要创建同名容器 → 撞上旧项目留下的那个 → 冲突。副作用:`pgdata` 卷也按项目名分裂
-成 `local-dev-postgres-docker_pgdata` / `shaguabijia-app-server_pgdata`,数据被切成两半。
-
-修复(均在 `scripts/ensure_pg.py`,`docker-compose.yml` 不动、容器名仍是 `shaguabijia-pg`):
-1. 模块级 `os.environ.setdefault("COMPOSE_PROJECT_NAME", "shaguabijia")` —— 钉死项目名,无论从哪个
-   目录/worktree 跑都是同一个项目、同一个卷 `shaguabijia_pgdata`,所有 `docker compose up/exec` 一致。
-2. `_compose_up()` 前置 `_remove_stale_container()`:若存在「同名但不属于本项目」的残留容器,先 `docker rm -f`
-   再 up(靠 `docker ps --filter name/label` 判归属;数据在命名卷里,删容器不丢)。旧目录/worktree 留下的
-   残留容器就此自动清掉,不需手动干预。
-
-影响:本次修复后首跑,旧的 `shaguabijia-pg`(属项目 `local-dev-postgres-docker`)会被自动删除、在项目
-`shaguabijia` 下重建,挂载全新的 `shaguabijia_pgdata`(空库,`alembic upgrade head` 重建表)。旧数据仍留在
-`local-dev-postgres-docker_pgdata` 卷里(未删,可恢复);确认不需要后可 `docker volume rm` 清理两个旧卷。

run.bat

@@ -30,14 +30,7 @@ if not exist .env (
 
 if not exist data mkdir data
 
-REM Ensure local Docker PostgreSQL is up (auto-starts Docker + PG container if needed)
-call "%PY%" -m scripts.ensure_pg
-if errorlevel 1 (
-    echo [X] ensure_pg failed ^(PostgreSQL not ready^)
-    exit /b %errorlevel%
-)
-
-REM Build/upgrade schema (idempotent; no-op if already at head)
+REM Build/upgrade SQLite schema (idempotent; no-op if already at head)
 call "%PY%" -m alembic upgrade head
 if errorlevel 1 (
     echo [X] alembic upgrade head failed

run.sh

@@ -18,8 +18,7 @@ if [ ! -f .env ]; then
   exit 1
 fi
 
-mkdir -p data            # 运行期落盘目录(媒体上传等)
-"$PY" -m scripts.ensure_pg      # 确保本地 Docker PostgreSQL 就绪(没起会自动拉起;失败即退出)
+mkdir -p data            # sqlite 文件所在目录
 "$PY" -m alembic upgrade head   # 确保表已建（幂等，已是最新则 no-op）
 
 # --reload 只盯源码目录 app/:别去监视 logs/(日志写入触发"检测→再写日志"回环)和

run8771.bat

@@ -1,50 +0,0 @@
-@echo off
-REM Admin backend startup (Windows) - the :8771 peer of run.bat.
-REM
-REM Usage:
-REM   cd shaguabijia-app-server
-REM   run8771.bat
-REM
-REM Runs the ADMIN FastAPI app (app.admin.main:admin_app) on 127.0.0.1:8771 —
-REM a SEPARATE process from run.bat (which runs app.main:app on 8770). The admin
-REM web frontend (Next.js :3001) points at http://localhost:8771. Auto-reload on
-REM code change.
-REM
-REM Prerequisite (first time):
-REM   conda activate pricebot ^&^& pip install -e .
-REM   copy .env.example .env  ^&^& fill JWT_SECRET_KEY
-REM
-REM Tip: shaguabijia-admin-web\start.bat starts user-api(8770) + admin-api(8771)
-REM + frontend(3001) in one go, if you prefer a single command.
-
-cd /d "%~dp0"
-
-REM Prefer the project virtualenv (.venv) so we never inherit a wrong
-REM global/conda interpreter. FastAPI<0.115 on Pydantic 2.12 crashes at import
-REM with "'FieldInfo' object has no attribute 'in_'". Falls back to PATH python.
-set "PY=python"
-if exist "%~dp0.venv\Scripts\python.exe" set "PY=%~dp0.venv\Scripts\python.exe"
-
-if not exist .env (
-    echo [X] Missing .env. Run: copy .env.example .env  and fill JWT_SECRET_KEY ^(plus MT_CPS_* if you test Meituan^)
-    exit /b 1
-)
-
-if not exist data mkdir data
-
-REM Ensure local Docker PostgreSQL is up (auto-starts Docker + PG container if needed)
-call "%PY%" -m scripts.ensure_pg
-if errorlevel 1 (
-    echo [X] ensure_pg failed ^(PostgreSQL not ready^)
-    exit /b %errorlevel%
-)
-
-REM Build/upgrade schema (idempotent; no-op if already at head)
-call "%PY%" -m alembic upgrade head
-if errorlevel 1 (
-    echo [X] alembic upgrade head failed
-    exit /b %errorlevel%
-)
-
-REM Long-running foreground process. Ctrl+C to stop.
-"%PY%" -m uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771 --reload

scripts/ensure_pg.py

@@ -1,317 +0,0 @@
-"""确保本地 PostgreSQL 就绪(开发/测试统一用 Docker PG)。
-
-被三处复用:
-  - run.sh / run.bat:`python -m scripts.ensure_pg`(CLI,失败退非 0)
-  - tests/conftest.py:`from scripts.ensure_pg import ensure; ensure(test_url)`
-
-流程:读 DATABASE_URL → TCP 探测 → 没起就(必要时启 Docker Desktop)→
-`docker compose up -d` → 等 PG ready → 幂等确保测试库存在。
-
-运行时(run.sh/run.bat)支持【显式】SQLite 逃生舱:DATABASE_URL 设为 sqlite → 放行并打印
-醒目降级横幅(绝不静默替你切库);无 docker CLI 时报错里也指路该逃生舱。测试侧
-(conftest 传 PG URL)不降级——sqlite 分支不触发,没 PG 直接 raise。详见设计文档 §10。
-
-生产用原生 PG(scripts/init_postgres.py),不走本模块。
-"""
-from __future__ import annotations
-
-import os
-import socket
-import subprocess
-import sys
-import time
-from pathlib import Path
-from urllib.parse import urlsplit
-
-ROOT = Path(__file__).resolve().parent.parent
-
-# 日志里可能含 emoji(如 ✅);Windows GBK 控制台(cmd.exe)无法编码会抛 UnicodeEncodeError → 脚本崩、
-# run.bat 误判 ensure_pg 失败。用 backslashreplace 保底:中文仍正常,仅不可编码字符被转义,不崩。
-for _stream in (sys.stdout, sys.stderr):
-    try:
-        _stream.reconfigure(errors="backslashreplace")
-    except (AttributeError, ValueError):
-        pass
-
-APP_DB = "shaguabijia"
-TEST_DB = "shaguabijia_test"
-DB_USER = "shaguabijia_app"
-COMPOSE_SERVICE = "postgres"
-CONTAINER_NAME = "shaguabijia-pg"  # 必须与 docker-compose.yml 的 container_name 一致
-
-# 钉死 compose 项目名:否则它默认取运行目录 basename,在不同目录/worktree 之间切会各自
-# 成一个项目 → 同一个固定 container_name 撞名报错、pgdata 卷还会按项目名分裂成多份。
-# 钉成 app 名后,无论从哪个目录/worktree 跑都是同一个项目、同一个卷。setdefault:尊重外部覆盖。
-os.environ.setdefault("COMPOSE_PROJECT_NAME", "shaguabijia")
-
-DOCKER_START_TIMEOUT = int(os.environ.get("ENSURE_PG_DOCKER_TIMEOUT", "120"))
-PG_READY_TIMEOUT = int(os.environ.get("ENSURE_PG_READY_TIMEOUT", "60"))
-# 单条 docker 探测/exec 命令的超时:防 Docker 守护进程半死(尤其 Windows 冷启)时
-# docker info / exec 无限挂起、绕过上面的总超时。
-DOCKER_CMD_TIMEOUT = int(os.environ.get("ENSURE_PG_CMD_TIMEOUT", "15"))
-POLL_INTERVAL = 3.0
-
-PG_URL = (
-    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia"
-)
-SQLITE_URL = "sqlite:///./data/app.db"  # 无 Docker 时的显式降级逃生舱(仅 run 运行时)
-
-
-def _log(msg: str) -> None:
-    print(f"[ensure_pg] {msg}", flush=True)
-
-
-def _is_sqlite(url: str) -> bool:
-    return url.strip().lower().startswith("sqlite")
-
-
-def _parse_host_port(url: str) -> tuple[str, int]:
-    """从 SQLAlchemy URL 取 host/port,缺省 localhost:5432。"""
-    parts = urlsplit(url)
-    return (parts.hostname or "localhost"), (parts.port or 5432)
-
-
-def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
-    try:
-        with socket.create_connection((host, port), timeout=timeout):
-            return True
-    except OSError:
-        return False
-
-
-def _docker_desktop_cmd(platform: str, program_files: str) -> list[str] | None:
-    """按平台给出启动 Docker Desktop 的命令;Linux 返回 None(daemon 需 sudo,让用户手动)。"""
-    if platform.startswith("win"):
-        return [str(Path(program_files) / "Docker" / "Docker" / "Docker Desktop.exe")]
-    if platform == "darwin":
-        return ["open", "-a", "Docker"]
-    return None
-
-
-def _docker_ok(subcmd: str) -> bool:
-    """`docker --version`(CLI 在不在,纯客户端)/`docker info`(daemon 起没起)成功与否。"""
-    try:
-        subprocess.run(
-            ["docker", subcmd],
-            cwd=ROOT,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=True,
-            timeout=DOCKER_CMD_TIMEOUT,
-        )
-        return True
-    except (OSError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
-        return False
-
-
-def _docker_cli_ok() -> bool:
-    # 必须用 `docker --version`(纯客户端,不连 daemon)而非 `docker version`
-    # (后者要连 daemon,守护进程没起时退非零)——否则「装了 Docker 但没启动」
-    # 会被误判成「没装 CLI」,直接绕过下面 _start_docker_daemon() 的自动拉起(需求②)。
-    return _docker_ok("--version")
-
-
-def _docker_daemon_ok() -> bool:
-    return _docker_ok("info")
-
-
-def _start_docker_daemon() -> bool:
-    """守护进程没起时按平台拉起,轮询到就绪。返回是否成功。"""
-    if _docker_daemon_ok():
-        return True
-    cmd = _docker_desktop_cmd(
-        sys.platform, os.environ.get("ProgramFiles", r"C:\Program Files")
-    )
-    if cmd is None:
-        _log("Docker 守护进程未运行。Linux 请手动:sudo systemctl start docker,然后重试。")
-        return False
-    if sys.platform.startswith("win") and not Path(cmd[0]).exists():
-        _log(f"找不到 Docker Desktop:{cmd[0]}。请手动启动 Docker Desktop 后重试。")
-        return False
-    _log("启动 Docker Desktop(首次冷启可能 30-60s)…")
-    try:
-        subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-    except OSError as e:
-        _log(f"启动 Docker Desktop 失败:{e}")
-        return False
-    deadline = time.monotonic() + DOCKER_START_TIMEOUT
-    while time.monotonic() < deadline:
-        if _docker_daemon_ok():
-            _log("Docker 守护进程已就绪。")
-            return True
-        _log("等待 Docker 守护进程…")
-        time.sleep(POLL_INTERVAL)
-    _log(f"等待 Docker 守护进程超时({DOCKER_START_TIMEOUT}s)。")
-    return False
-
-
-def _ps_names(*filters: str) -> str:
-    """docker ps -a 按 filter 查容器名(每行一个);失败返回空串。"""
-    args = ["docker", "ps", "-a", "--format", "{{.Names}}"]
-    for f in filters:
-        args += ["--filter", f]
-    try:
-        r = subprocess.run(
-            args, cwd=ROOT, capture_output=True, text=True, timeout=DOCKER_CMD_TIMEOUT,
-        )
-    except (OSError, subprocess.TimeoutExpired):
-        return ""
-    return r.stdout if r.returncode == 0 else ""
-
-
-def _remove_stale_container() -> None:
-    """删掉「同名但不属于本 compose 项目」的残留容器(旧目录/worktree 建的)。
-
-    固定的 container_name 是全局唯一名:若旧项目留下一个同名容器,`docker compose up`
-    会因撞名报 "container name already in use" 而失败。这里在 up 之前主动清掉它。
-    数据在命名卷(<project>_pgdata)里,删容器不删卷、不丢数据。
-    """
-    project = os.environ.get("COMPOSE_PROJECT_NAME", "")
-    name_filter = f"name=^{CONTAINER_NAME}$"
-    if CONTAINER_NAME not in _ps_names(name_filter).split():
-        return  # 没有同名容器
-    ours = _ps_names(name_filter, f"label=com.docker.compose.project={project}")
-    if CONTAINER_NAME in ours.split():
-        return  # 就是本项目的容器,compose 会自己 start/复用,别删
-    _log(f"发现残留同名容器 {CONTAINER_NAME}(非本项目 '{project}'),删除以避免撞名"
-         f"(数据在卷里,不丢)…")
-    try:
-        subprocess.run(
-            ["docker", "rm", "-f", CONTAINER_NAME], cwd=ROOT,
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=DOCKER_CMD_TIMEOUT,
-        )
-    except (OSError, subprocess.TimeoutExpired):
-        _log(f"⚠️ 删除残留容器失败,可手动: docker rm -f {CONTAINER_NAME}")
-
-
-def _compose_up() -> bool:
-    _remove_stale_container()
-    _log("docker compose up -d(镜像缺失会自动拉取,首用约几十秒)…")
-    try:
-        subprocess.run(["docker", "compose", "up", "-d"], cwd=ROOT, check=True)
-        return True
-    except (OSError, subprocess.CalledProcessError) as e:
-        _log(f"docker compose up 失败:{e}")
-        return False
-
-
-def _pg_isready() -> bool:
-    try:
-        r = subprocess.run(
-            ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
-             "pg_isready", "-U", DB_USER, "-d", APP_DB],
-            cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-            timeout=DOCKER_CMD_TIMEOUT,
-        )
-    except (OSError, subprocess.TimeoutExpired):
-        return False
-    return r.returncode == 0
-
-
-def _wait_pg_ready(host: str, port: int) -> bool:
-    deadline = time.monotonic() + PG_READY_TIMEOUT
-    while time.monotonic() < deadline:
-        if _port_open(host, port) and _pg_isready():
-            _log("PostgreSQL 已就绪。")
-            return True
-        _log("等待 PostgreSQL 就绪…")
-        time.sleep(POLL_INTERVAL)
-    _log(f"等待 PostgreSQL 就绪超时({PG_READY_TIMEOUT}s)。")
-    return False
-
-
-def _test_db_exists() -> bool:
-    """测试库是否已存在(连业务库 shaguabijia 查 pg_database)。"""
-    try:
-        r = subprocess.run(
-            ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
-             "psql", "-U", DB_USER, "-d", APP_DB, "-tAc",
-             f"SELECT 1 FROM pg_database WHERE datname='{TEST_DB}'"],
-            cwd=ROOT, capture_output=True, text=True, timeout=DOCKER_CMD_TIMEOUT,
-        )
-    except (OSError, subprocess.TimeoutExpired):
-        return False
-    return r.returncode == 0 and r.stdout.strip() == "1"
-
-
-def ensure_test_db() -> bool:
-    """幂等建测试库(兼容老 pgdata 卷首启没跑 initdb 的情况)。返回测试库是否就绪。
-
-    公开给 conftest 单独调用:ensure() 在「端口已通」时会短路返回、不建测试库,
-    所以测试侧需在 ensure() 之后再显式补一刀(best-effort)。
-    """
-    if _test_db_exists():
-        return True
-    _log(f"建测试库 {TEST_DB}…")
-    try:
-        create = subprocess.run(
-            ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
-             "psql", "-U", DB_USER, "-d", APP_DB, "-c",
-             f"CREATE DATABASE {TEST_DB} OWNER {DB_USER}"],
-            cwd=ROOT, capture_output=True, text=True, timeout=DOCKER_CMD_TIMEOUT,
-        )
-    except (OSError, subprocess.TimeoutExpired) as e:
-        _log(f"⚠️ 建测试库 {TEST_DB} 失败:{e}")
-        return False
-    # returncode==0=建成功;非 0 但库已存在=与并发创建者竞争失败(42P04),仍算就绪
-    if create.returncode == 0 or _test_db_exists():
-        return True
-    _log(f"⚠️ 建测试库 {TEST_DB} 失败:{(create.stderr or '').strip()}")
-    return False
-
-
-def _warn_sqlite_degraded() -> None:
-    """DATABASE_URL 是 SQLite 时打印醒目降级横幅(显式逃生舱,非静默切库)。"""
-    for line in (
-        "⚠️ ================= 降级模式(SQLite) =================",
-        "⚠️  DATABASE_URL 是 SQLite,不是 PostgreSQL。",
-        "⚠️  PG 专有 SQL(窗口函数/FILTER/JSONB)与严格类型在此模式【不被验证】。",
-        "⚠️  提交前请在装了 Docker 的机器上用 PG 复跑;装好后把 DATABASE_URL 改回:",
-        f"⚠️    {PG_URL}",
-        "⚠️ ===================================================",
-    ):
-        _log(line)
-
-
-def ensure(database_url: str | None = None) -> bool:
-    """确保 PG 就绪,返回 True/False。database_url 缺省从 settings 读(尊重 .env)。
-
-    运行时若 DATABASE_URL 是 SQLite → 打印降级横幅并返回 True(显式逃生舱);
-    conftest 传的是 PG URL,故测试侧永不走此分支。
-    """
-    if database_url is None:
-        from app.core.config import settings  # 延迟导入,避免过早固化 settings
-
-        database_url = settings.DATABASE_URL
-
-    if _is_sqlite(database_url):
-        _warn_sqlite_degraded()
-        return True
-
-    host, port = _parse_host_port(database_url)
-
-    if _port_open(host, port):
-        _log(f"✅ PostgreSQL 已在 {host}:{port} 运行,跳过 Docker。")
-        return True
-
-    _log(f"{host}:{port} 无 PostgreSQL,准备用 Docker 拉起…")
-
-    if not _docker_cli_ok():
-        _log("未检测到 docker 命令。请先安装 Docker Desktop:")
-        _log("  https://www.docker.com/products/docker-desktop/")
-        _log(f"装不了 Docker?把 .env 的 DATABASE_URL 改成  {SQLITE_URL}  可降级用 SQLite 跑")
-        _log("  (PG 专有 SQL/严格性不被验证,仅救急);改完重跑 run.sh/run.bat。")
-        return False
-    if not _start_docker_daemon():
-        return False
-    if not _compose_up():
-        return False
-    if not _wait_pg_ready(host, port):
-        return False
-    if not ensure_test_db():
-        return False
-    return True
-
-
-if __name__ == "__main__":
-    sys.exit(0 if ensure() else 1)

scripts/init_postgres.py

@@ -1,8 +1,6 @@
 """Bootstrap PostgreSQL: 建用户 + 建库 + 授权 + 写 .env + 跑迁移。
 
-新机器初始化用(面向【生产原生 PG】:apt/systemd 装好的 PostgreSQL)。
-本地开发/测试请改用 docker-compose.yml + scripts/ensure_pg.py(run.sh/run.bat 自动拉起),不必跑本脚本。
-前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
+新机器初始化用。前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
 
 用法:
     python scripts/init_postgres.py

scripts/seed_coinhistory_labels_test.py

@@ -0,0 +1,166 @@
+"""收益明细「金币记录」文案验证脚手架(2026-07 文案改版验收用)。
+
+问题:客户端按 bizType 显示固定文案(路线B,强制覆盖后端 remark),但账号若没有对应
+bizType 的流水,收益明细页就空着,无从验证。本脚本往指定测试用户塞每种 bizType 各一条
+流水,让你在手机收益明细页一屏核对全部新文案;验收完 --clean 一键删除,不污染数据。
+
+⚠️ 仅 dev 库用(APP_ENV=dev 时才允许 --seed/--clean)。所有测试流水 ref_id 前缀 TESTDOC,
+按前缀精确清理,不会误删真实流水。
+
+用法(pricebot env 直调,见项目 CLAUDE.md):
+  D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --show
+  D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --seed
+  D:/miniconda/envs/pricebot/python.exe scripts/seed_coinhistory_labels_test.py --clean
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+
+from app.core.config import settings
+from app.core.rewards import CN_TZ
+from datetime import datetime
+
+from app.db.session import SessionLocal
+from app.models.user import User
+from app.models.wallet import CoinAccount, CoinTransaction
+
+TEST_PHONE = "11111111111"
+REF_PREFIX = "TESTDOC"  # 所有本脚本造的流水都带这个 ref_id 前缀,便于精确清理
+
+# 每条 = (bizType, 故意写错/留空的 remark, 金币数)。
+# remark 故意填「错的」→ 若客户端仍显示新文案 = 证明路线B强制覆盖生效(无视后端 remark)。
+# task_ 一条留空 remark → 走客户端兜底映射。顺序即手机上从新到旧的展示顺序(后塞的在最上)。
+CASES: list[tuple[str, str, int]] = [
+    ("signin", "每日签到 第99天(旧文案,应被覆盖)", 220),
+    ("signin_boost", "签到膨胀 第99天", 3000),
+    ("reward_video", "看视频奖励金币(旧文案,应被覆盖)", 200),
+    ("feed_ad_reward_comparison", "", 50),   # 后端新拆:比价场景 → 比价奖励
+    ("feed_ad_reward_coupon", "", 50),       # 后端新拆:领券场景 → 领券奖励
+    ("feed_ad_reward", "信息流广告奖励(welfare/旧数据兜底)", 50),
+    ("price_report_reward", "上报更低价审核通过(旧文案,应被覆盖)", 1000),
+    ("feedback_reward", "意见反馈被采纳(旧文案,应被覆盖)", 10000),
+    ("task_enable_notification", "", 750),   # remark 留空 → 客户端「打开消息提醒奖励」
+    # 下两条现实中不会进金币记录(invite 发现金进邀请钱包 / compare_milestone 后端死代码不发钱),
+    # 仅用于验证「杀掉好友比价奖励 + 兜底改任务奖励」:两条都应显示「任务奖励」(remark 被强制无视)。
+    ("invite", "好友比价奖励(旧文案,应被杀→任务奖励)", 200),
+    ("compare_milestone", "", 120),
+]
+
+
+def _client_coin_title(biz_type: str, remark: str | None) -> str:
+    """复刻 CoinHistoryViewModel.coinTitle 的最新逻辑(路线B),用于 --show 预览。
+
+    ⚠️ 必须与客户端保持一致;客户端改了这里也要同步,否则预览会骗人。
+    """
+    fixed = {
+        "exchange_out": "金币兑换现金",
+        "signin": "每日签到奖励",
+        "signin_boost": "签到膨胀奖励",
+        "reward_video": "看视频赚金币",
+        "ad_reward": "看视频赚金币",
+        "feed_ad_reward_comparison": "比价奖励",
+        "feed_ad_reward_coupon": "领券奖励",
+        "feed_ad_reward": "信息流广告奖励",
+        "price_report_reward": "爆料奖励",
+        "feedback_reward": "反馈奖励",
+        "invite": "任务奖励",  # 杀掉"好友比价奖励",归兜底
+    }
+    if biz_type in fixed:
+        return fixed[biz_type]
+    if remark:
+        return remark
+    if biz_type == "task_enable_notification":
+        return "打开消息提醒奖励"
+    return "任务奖励"
+
+
+def _get_user(db) -> User:
+    u = db.query(User).filter(User.phone == TEST_PHONE).first()
+    if not u:
+        print(f"✗ 库里没有测试号 {TEST_PHONE} —— 先在手机上用这个号登录一次再跑本脚本。")
+        sys.exit(1)
+    return u
+
+
+def cmd_show() -> None:
+    """只打印:每种 bizType 经客户端映射后会显示成什么(不写库)。"""
+    print("bizType 造流水后,收益明细页预期显示的文案:\n")
+    print(f"  {'bizType':32} {'后端remark(故意填的)':32} → 手机显示")
+    print("  " + "-" * 90)
+    for biz, remark, _coin in CASES:
+        shown = _client_coin_title(biz, remark or None)
+        rk = (remark or "(空)")
+        print(f"  {biz:32} {rk:32} → {shown}")
+    print("\n注:remark 列是故意填的『旧/错』文案;'手机显示'若为新文案 = 强制覆盖生效。")
+    print("invite / compare_milestone 已从客户端映射删除:invite 有 remark 故显示原样,")
+    print("compare_milestone remark 空故落兜底『奖励』—— 两者都不再有专属新文案(符合『去掉』)。")
+
+
+def cmd_seed() -> None:
+    if settings.APP_ENV != "dev":
+        print(f"✗ 拒绝:APP_ENV={settings.APP_ENV},本脚本只在 dev 库造测试数据。")
+        sys.exit(1)
+    db = SessionLocal()
+    u = _get_user(db)
+    acc = db.query(CoinAccount).filter(CoinAccount.user_id == u.id).first()
+    if acc is None:
+        acc = CoinAccount(user_id=u.id, coin_balance=0, cash_balance_cents=0)
+        db.add(acc)
+        db.flush()
+    now = datetime.now(CN_TZ).replace(tzinfo=None)
+    made = 0
+    for i, (biz, remark, coin) in enumerate(CASES):
+        ref = f"{REF_PREFIX}:{biz}:{i}"
+        exists = db.query(CoinTransaction).filter(CoinTransaction.ref_id == ref).first()
+        if exists:
+            continue
+        acc.coin_balance += coin
+        db.add(CoinTransaction(
+            user_id=u.id, amount=coin, balance_after=acc.coin_balance,
+            biz_type=biz, ref_id=ref, remark=remark or None, created_at=now,
+        ))
+        made += 1
+    db.commit()
+    print(f"✓ 已给 user_id={u.id}({TEST_PHONE})造 {made} 条测试流水,当前金币余额 {acc.coin_balance}。")
+    print("  → 打开手机 App「收益明细 / 金币记录」下拉刷新,逐条核对文案。")
+    print("  → 验收完跑 --clean 删除这些测试流水。")
+
+
+def cmd_clean() -> None:
+    if settings.APP_ENV != "dev":
+        print(f"✗ 拒绝:APP_ENV={settings.APP_ENV}。")
+        sys.exit(1)
+    db = SessionLocal()
+    u = _get_user(db)
+    rows = db.query(CoinTransaction).filter(
+        CoinTransaction.user_id == u.id,
+        CoinTransaction.ref_id.like(f"{REF_PREFIX}:%"),
+    ).all()
+    total = sum(r.amount for r in rows)
+    for r in rows:
+        db.delete(r)
+    acc = db.query(CoinAccount).filter(CoinAccount.user_id == u.id).first()
+    if acc is not None:
+        acc.coin_balance -= total  # 把造流水时加的余额扣回,还原
+    db.commit()
+    print(f"✓ 已删除 {len(rows)} 条 TESTDOC 测试流水,余额回扣 {total},当前 {acc.coin_balance if acc else 0}。")
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="收益明细金币文案验证脚手架")
+    g = ap.add_mutually_exclusive_group(required=True)
+    g.add_argument("--show", action="store_true", help="只打印每种 bizType 的预期显示文案,不写库")
+    g.add_argument("--seed", action="store_true", help="往测试号造每种 bizType 各一条流水")
+    g.add_argument("--clean", action="store_true", help="删除本脚本造的所有测试流水")
+    args = ap.parse_args()
+    if args.show:
+        cmd_show()
+    elif args.seed:
+        cmd_seed()
+    elif args.clean:
+        cmd_clean()
+
+
+if __name__ == "__main__":
+    main()

tests/conftest.py

@@ -1,19 +1,22 @@
 """测试用 fixtures。
 
-测试库用 Docker PostgreSQL 的 shaguabijia_test(与 dev 业务库 shaguabijia 隔离)。
-顺序(必须):设 test DATABASE_URL(在 import app.* 之前)→ ensure PG 就绪 + 测试库存在 →
-import app → 建表。持久卷可能残留上次的表 → session 开头先 drop 再 create。
+测试 DB 用临时文件 SQLite。
+  - 不用 in-memory:in-memory 默认 per-connection,跨连接看不到表。
+  - 用临时文件保证 SessionLocal 每次新连都看到同一份 schema。
+
+顺序:set env(必须在 import app.* 之前) → import app → 建表 → TestClient
 """
 from __future__ import annotations
 
 import os
+import tempfile
 from collections.abc import Iterator
 
-# 1) 测试库连接串——必须在 import app.* 之前设好(app.db.session 在 import 期就建 engine)
-_TEST_DB_URL = (
-    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test"
-)
-os.environ["DATABASE_URL"] = _TEST_DB_URL
+# 临时 db 文件路径,进程退出后清理
+_tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+_tmp_db.close()
+
+os.environ["DATABASE_URL"] = f"sqlite:///{_tmp_db.name}"
 os.environ.setdefault("JWT_SECRET_KEY", "test-secret-please-ignore-this-is-only-for-pytest-not-real")
 os.environ.setdefault("ADMIN_JWT_SECRET", "test-admin-secret-please-ignore-only-for-pytest-not-real")
 os.environ.setdefault("JG_APP_KEY", "test-key")
@@ -32,18 +35,6 @@ os.environ.setdefault("RATE_LIMIT_ENABLED", "false")  # 限流内存计数会跨
 os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
 os.environ.setdefault("PANGLE_REWARD_SECRET", "test-pangle-secret-only-for-pytest")
 
-# 2) 保证 Docker PG 就绪(必须在 import app.db.session 建 engine 之前)。
-#    ensure() 在「端口已通」时会短路、不建测试库,故随后再显式补一刀 ensure_test_db()。
-from scripts.ensure_pg import ensure, ensure_test_db
-
-if not ensure(_TEST_DB_URL):
-    raise RuntimeError(
-        "测试需要 Docker PostgreSQL 就绪。请确认已装 Docker Desktop;"
-        "或先跑一次 run.bat/run.sh 把 PG 拉起,再重试 pytest。"
-    )
-# best-effort 兜底建测试库(PG 已在跑但测试库缺失=老卷)。真缺库时下面 create_all 会明确报错。
-ensure_test_db()
-
 import pytest
 from fastapi.testclient import TestClient
 
@@ -54,11 +45,13 @@ from app.main import app
 
 @pytest.fixture(scope="session", autouse=True)
 def _setup_db() -> Iterator[None]:
-    # 持久卷可能残留上次跑崩后的表/数据 → 先 drop 再 create,保证干净起点
-    Base.metadata.drop_all(engine)
     Base.metadata.create_all(engine)
     yield
     Base.metadata.drop_all(engine)
+    try:
+        os.unlink(_tmp_db.name)
+    except OSError:
+        pass
 
 
 @pytest.fixture()

tests/test_compare_harvest.py

@@ -18,7 +18,6 @@ import httpx
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
-from app.repositories.user import get_user_by_phone
 from app.schemas.compare_record import ComparisonRecordIn
 from sqlalchemy import select
 
@@ -48,17 +47,6 @@ def _get(db, trace_id: str) -> ComparisonRecord | None:
     ).scalar_one_or_none()
 
 
-def _make_user(client, phone: str) -> int:
-    """登录建号并返回其真实 user_id。
-
-    PG 强制 comparison_record.user_id → user.id 外键,须引用真实存在的用户;
-    用本用例自己登录出的用户,不会与别的用例撞。
-    """
-    client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
-    with SessionLocal() as db:
-        return get_user_by_phone(db, phone).id
-
-
 # ============================================================
 # repo 层
 # ============================================================
@@ -156,18 +144,17 @@ def test_harvest_abort_missing_row_returns_none(client) -> None:
 def test_upsert_record_no_downgrade_after_harvest_success(client) -> None:
     """harvest 落 success 后,老客户端 fromFailure 的 cancelled 上报不许把它盖回去。"""
     tid = _tid()
-    # PG 强制 comparison_record.user_id → user.id 外键(SQLite 不强制,老写法用合成 id
-    # 987654)。用本用例自己登录出的真实用户,既满足外键、又不与别的用例撞。
-    uid = _make_user(client, "13800007701")
     with SessionLocal() as db:
         crud.harvest_done(db, trace_id=tid, user_id=None, done_params=_done_params())
         payload = ComparisonRecordIn(
             trace_id=tid, business_type="food", status="cancelled",
             information="用户终止", comparison_results=[],
         )
-        rec = crud.upsert_record(db, user_id=uid, payload=payload)
+        # 用一个不会与顺序自增用户撞的合成 id(SQLite 测试库 FK 不强制;别用小整数,
+        # 否则会撞上别的测试 login 出来的真实 user_id → 记录混进那个用户的列表)。
+        rec = crud.upsert_record(db, user_id=987654, payload=payload)
         assert rec.status == "success"              # 不降级
-        assert rec.user_id == uid                    # 但补上了 user_id(原为 None)
+        assert rec.user_id == 987654                 # 但补上了 user_id(原为 None)
 
 
 # ============================================================

tests/test_ensure_pg.py

@@ -1,87 +0,0 @@
-"""scripts/ensure_pg.py 纯函数单测(不需要 Docker/PG)。"""
-from __future__ import annotations
-
-import socket
-
-from scripts.ensure_pg import (
-    _docker_desktop_cmd,
-    _is_sqlite,
-    _parse_host_port,
-    _port_open,
-    ensure,
-)
-
-
-def test_is_sqlite():
-    assert _is_sqlite("sqlite:///./data/app.db")
-    assert _is_sqlite("  SQLite:///x ")
-    assert not _is_sqlite("postgresql+psycopg://u:p@localhost:5432/db")
-
-
-def test_parse_host_port_full():
-    assert _parse_host_port(
-        "postgresql+psycopg://u:p@localhost:5432/shaguabijia"
-    ) == ("localhost", 5432)
-
-
-def test_parse_host_port_defaults():
-    # 缺端口 → 5432
-    assert _parse_host_port("postgresql+psycopg://u:p@db.example/x")[1] == 5432
-    # 缺 host → localhost
-    assert _parse_host_port("postgresql+psycopg:///x") == ("localhost", 5432)
-
-
-def test_parse_host_port_testdb():
-    assert _parse_host_port(
-        "postgresql+psycopg://u:p@localhost:5432/shaguabijia_test"
-    ) == ("localhost", 5432)
-
-
-def test_port_open_true():
-    srv = socket.socket()
-    srv.bind(("127.0.0.1", 0))
-    srv.listen(1)
-    port = srv.getsockname()[1]
-    try:
-        assert _port_open("127.0.0.1", port, timeout=1.0)
-    finally:
-        srv.close()
-
-
-def test_port_open_false():
-    s = socket.socket()
-    s.bind(("127.0.0.1", 0))
-    port = s.getsockname()[1]
-    s.close()  # 释放端口,无人监听 → 连接应失败
-    assert not _port_open("127.0.0.1", port, timeout=0.3)
-
-
-def test_docker_desktop_cmd_windows():
-    cmd = _docker_desktop_cmd("win32", r"C:\Program Files")
-    assert cmd is not None
-    assert cmd[0].endswith("Docker Desktop.exe")
-    assert "Docker" in cmd[0]
-
-
-def test_docker_desktop_cmd_darwin():
-    assert _docker_desktop_cmd("darwin", "") == ["open", "-a", "Docker"]
-
-
-def test_docker_desktop_cmd_linux():
-    assert _docker_desktop_cmd("linux", "") is None
-
-
-def test_ensure_rejects_sqlite():
-    # dev 守卫:sqlite 直接 False(不碰 Docker)
-    assert ensure("sqlite:///./data/app.db") is False
-
-
-def test_ensure_shortcircuits_when_pg_up(monkeypatch):
-    # 端口通 → 直接 True,绝不触碰 docker
-    monkeypatch.setattr("scripts.ensure_pg._port_open", lambda *a, **k: True)
-
-    def _boom():
-        raise AssertionError("端口通时不应调用 docker")
-
-    monkeypatch.setattr("scripts.ensure_pg._docker_cli_ok", _boom)
-    assert ensure("postgresql+psycopg://u:p@localhost:5432/shaguabijia") is True

tests/test_withdraw_ledger_check.py

@@ -1,150 +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",
-        json={"amount_cents": 200, "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"]