feat(dev): ensure_pg 加显式 SQLite 逃生舱 + 修 Docker 检测/项目名撞名;新增 run8771 起 admin

- ensure_pg: 无 Docker 时【显式】降级 SQLite(带醒目降级横幅),不再硬失败(反转设计 D4)
- ensure_pg: _docker_cli_ok 改用 `docker --version`(纯客户端),修「装了 Docker 但没启动」
  被 `docker version`(要连 daemon)误判成「没装 CLI」而绕过自动拉起(需求②)
- ensure_pg: 钉死 COMPOSE_PROJECT_NAME=shaguabijia + 清「同名但非本项目」残留容器,
  修跨目录/worktree 切换时 container_name 撞名 + pgdata 卷分裂
- run8771.bat: 新增 admin 后端(:8771)启动脚本,跑 app.admin.main:admin_app(run.bat 对等版)
- 设计文档 §10 记录以上 D4 反转与三处修复

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

.env.example

@@ -6,8 +6,11 @@ APP_NAME=shaguabijia-app-server
 APP_DEBUG=true
 
 # ===== 数据库 =====
-# SQLite 本地文件路径。生产环境用 /opt/shaguabijia-app-server/data.db
-DATABASE_URL=sqlite:///./data/app.db
+# 本地开发/测试统一用 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
 
 # ===== JWT =====
 # 生产部署务必改成随机长字符串,可用:python -c "import secrets; print(secrets.token_urlsafe(64))"

.gitignore

@@ -57,3 +57,6 @@ 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**: 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.
+- **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.
 - **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, creates temp SQLite file, builds all tables with `Base.metadata.create_all()`, tears down with `drop_all()` + unlink.
+- `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()`.
 - 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.
 

alembic/versions/admin_user_pages_override.py

@@ -1,35 +0,0 @@
-"""admin_user 加 pages_override 列(「自定义」权限:按人存专属可见页 key 列表)
-
-权限管理页新增「自定义」角色:选它时该成员的可见页不跟随任何共享角色,而由逐页勾选决定,
-存这个人专属的一份页 key 列表。仅 role == "custom" 时有效;普通角色为 None(可见页跟随角色)。
-PG 用 JSONB,SQLite 退化为通用 JSON(同 admin_audit_log.detail / comparison_record.raw_payload)。
-
-Revision ID: admin_user_pages_override
-Revises: admin_user_plain_password
-Create Date: 2026-07-08 00:00:00.000000
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects.postgresql import JSONB
-
-from alembic import op
-
-revision: str = "admin_user_pages_override"
-down_revision: str | Sequence[str] | None = "admin_user_plain_password"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-# 与 app/models/admin.py 的 _JSON 一致:PG JSONB / 其它 JSON
-_JSON = sa.JSON().with_variant(JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("admin_user", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("pages_override", _JSON, nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("admin_user", schema=None) as batch_op:
-        batch_op.drop_column("pages_override")

alembic/versions/coupon_claim_app_env.py

@@ -1,32 +0,0 @@
-"""coupon_claim_record 加 app_env 列(领券所属 session 环境;每券成功率表按它过滤 prod/dev)
-
-Revision ID: coupon_claim_app_env
-Revises: coupon_session_platform_success
-Create Date: 2026-07-08 00:00:00.000000
-
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-
-from alembic import op
-
-revision: str = "coupon_claim_app_env"
-down_revision: str | Sequence[str] | None = "coupon_session_platform_success"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("coupon_claim_record", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("app_env", sa.String(length=16), nullable=True))
-        batch_op.create_index(
-            "ix_coupon_claim_record_app_env", ["app_env"], unique=False
-        )
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("coupon_claim_record", schema=None) as batch_op:
-        batch_op.drop_index("ix_coupon_claim_record_app_env")
-        batch_op.drop_column("app_env")

alembic/versions/coupon_session_platform_success.py

@@ -1,37 +0,0 @@
-"""coupon_session 加 platform_success 列(本次至少领到一张的平台 id 列表)
-
-供 admin「领券数据」算 ②整单成功率 / ③点位成功率(平台粒度)。数据落点:服务端 /step 逐帧
-按 trace_id 并集写入(见 app/repositories/coupon_state.merge_session_platform_success)。旧行 NULL
-视作空集,已建表环境靠它补列、全新环境顺序应用不重复加列。设计:docs/guides/领券成功率指标-设计与埋点.md。
-
-Revision ID: coupon_session_platform_success
-Revises: admin_user_plain_password
-Create Date: 2026-07-07 00:00:00.000000
-
-"""
-
-from collections.abc import Sequence
-
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-from alembic import op
-
-# revision identifiers, used by Alembic.
-revision: str = "coupon_session_platform_success"
-down_revision: str | Sequence[str] | None = "admin_user_plain_password"
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-# PG 用 JSONB,SQLite(本地/测试)退化为通用 JSON(同 model 的 _JSON variant / 建表迁移)。
-_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
-
-
-def upgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("platform_success", _JSON, nullable=True))
-
-
-def downgrade() -> None:
-    with op.batch_alter_table("coupon_session", schema=None) as batch_op:
-        batch_op.drop_column("platform_success")

alembic/versions/merge_pages_override_coupon_slot.py

@@ -1,28 +0,0 @@
-"""合并两个 alembic head:admin_user_pages_override(#126 权限)+ coupon_claim_app_env(领券成功率)。
-
-两条迁移都从 admin_user_plain_password 分叉——#126 经 pull main 进入本分支,领券成功率为本分支新增——
-于是出现两个 head。本迁移仅把二者收敛成单 head,让 `alembic upgrade head`(单数,部署/run.sh 用)
-恢复正常;**不含任何表结构 / 数据改动**(纯 merge)。
-
-Revision ID: merge_pages_override_coupon_slot
-Revises: admin_user_pages_override, coupon_claim_app_env
-Create Date: 2026-07-09 00:00:00.000000
-"""
-
-from collections.abc import Sequence
-
-revision: str = "merge_pages_override_coupon_slot"
-down_revision: str | Sequence[str] | None = (
-    "admin_user_pages_override",
-    "coupon_claim_app_env",
-)
-branch_labels: str | Sequence[str] | None = None
-depends_on: str | Sequence[str] | None = None
-
-
-def upgrade() -> None:
-    """纯合并 head,无 schema 改动。"""
-
-
-def downgrade() -> None:
-    """拆回两个 head,无 schema 改动。"""

app/admin/permissions.py

@@ -9,9 +9,6 @@ super_admin 为内建全权角色,恒可见全部页(effective_pages 特判)。
 from __future__ import annotations
 
 SUPER_ADMIN_ROLE = "super_admin"
-# 「自定义」哨兵角色:不是 admin_role 表里的行,而是标记「这个人的可见页由 pages_override 决定」。
-# admin_user.role == CUSTOM_ROLE 时,有效可见页取 admin_user.pages_override(见 auth._admin_out_with_pages)。
-CUSTOM_ROLE = "custom"
 
 # 分组镜像前端导航(app/(main)/layout.tsx 的 NAV_GROUPS);key = 路由一级
 PERMISSION_CATALOG: list[dict] = [

app/admin/repositories/admin_user.py

@@ -26,17 +26,14 @@ def create_admin(
     password: str,
     role: str = "operator",
     plain_password: str | None = None,
-    pages_override: list[str] | None = None,
 ) -> AdminUser:
     """建管理员。plain_password 非空则额外留存明文(后台 UI 建的账号传,供权限管理页复看);
-    脚本/起后台建账号不传(留 None → 前端不显示密码)。
-    pages_override:role=="custom" 时传专属可见页 key 列表;普通角色为 None。"""
+    脚本/起后台建账号不传(留 None → 前端不显示密码)。"""
     admin = AdminUser(
         username=username,
         password_hash=hash_password(password),
         role=role,
         plain_password=plain_password,
-        pages_override=pages_override,
     )
     db.add(admin)
     db.commit()

app/admin/repositories/coupon_data.py

@@ -5,20 +5,17 @@
 - 发起数 = 区间内全部 session(含 started/completed/failed/abandoned),= 流失统计的基数。
 - 完成数 / 耗时均值 / 分位 = 仅 status==completed 子集(成功跑完才有可比的"领券耗时")。
 - summary/daily/hourly/total 在全量上算,不受分页;items 为排序后当前页。
-- 另含 coupon_slot_report(数据源 coupon_claim_record):按 coupon_id「按券成功率」表,见设计 §13。
 """
 from __future__ import annotations
 
-from datetime import UTC, datetime
-from datetime import date as _date
+from datetime import UTC, date as _date, datetime
 
-from sqlalchemy import case, func, or_, select
+from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from app.core import rewards
-from app.models.coupon_state import CouponClaimRecord, CouponSession
+from app.models.coupon_state import CouponSession
 from app.models.user import User
-from app.repositories.coupon_state import DEFAULT_PLATFORMS, coupon_id_to_platform
 
 
 def _cn_hour(dt: datetime) -> int:
@@ -45,46 +42,6 @@ def _avg(vals: list[int]) -> int | None:
     return round(sum(vals) / len(vals)) if vals else None
 
 
-def _success_rates(rows: list) -> dict:
-    """平台粒度成功率(见 docs/guides/领券成功率指标-设计与埋点.md §3/§12):
-
-    - sel(s) = 勾选平台(`platforms` 空 → 全领三档 DEFAULT_PLATFORMS);
-    - succ(s) = `platform_success` ∩ sel(至少领到一张的平台);
-    - ② 整单成功率 = #{sel⊆succ 且 sel≠∅} / 发起数;
-    - ③ 点位成功率 = Σ|succ| / Σ|sel|;per_platform[p] = 勾了 p 且成功 / 勾了 p。
-    基数含全部 session(started/completed/failed/abandoned),与「发起数」同基数。
-    """
-    started = len(rows)
-    full_success = 0
-    point_success = 0
-    point_total = 0
-    per_succ = {p: 0 for p in DEFAULT_PLATFORMS}
-    per_total = {p: 0 for p in DEFAULT_PLATFORMS}
-    for r in rows:
-        sel = set(r.platforms) if r.platforms else set(DEFAULT_PLATFORMS)
-        succ = set(r.platform_success or []) & sel
-        point_success += len(succ)
-        point_total += len(sel)
-        if sel and succ == sel:
-            full_success += 1
-        for p in sel:
-            if p in per_total:  # 只统计三档已知平台;未知/非法平台 id 不进 per_platform
-                per_total[p] += 1
-                if p in succ:
-                    per_succ[p] += 1
-    return {
-        "full_success_count": full_success,
-        "full_success_rate": round(full_success / started, 4) if started else None,
-        "point_success_count": point_success,
-        "point_total_count": point_total,
-        "point_success_rate": round(point_success / point_total, 4) if point_total else None,
-        "per_platform": {
-            p: (round(per_succ[p] / per_total[p], 4) if per_total[p] else None)
-            for p in DEFAULT_PLATFORMS
-        },
-    }
-
-
 def _session_to_row(r, phone: str | None = None, nickname: str | None = None) -> dict:
     """CouponSession ORM → 明细行 dict(主表「领券数据」与「用户全部领券」抽屉共用)。"""
     return {
@@ -127,7 +84,6 @@ def coupon_data_report(
     date_to: str,
     user: str | None = None,
     app_env: str | None = None,
-    statuses: list[str] | None = None,
     granularity: str = "day",
     limit: int = 500,
     offset: int = 0,
@@ -137,8 +93,6 @@ def coupon_data_report(
 
     - user:手机号/昵称模糊搜(匹配不到任何用户 → 空结果)。
     - app_env:prod/dev 精确;None=全部。
-    - statuses:领券状态多选(started/completed/failed/abandoned);None/空=全部。整个视图
-      (汇总/成功率/趋势/明细)按选中状态算,与 app_env 同级过滤(方案 A)。
     - sort:time=发起时刻倒序(默认) / elapsed=全程耗时倒序(None 末尾)。
     """
     by_hour = granularity == "hour"
@@ -161,8 +115,6 @@ def coupon_data_report(
     )
     if app_env is not None:
         stmt = stmt.where(CouponSession.app_env == app_env)
-    if statuses:
-        stmt = stmt.where(CouponSession.status.in_(statuses))
     if user_ids is not None:
         stmt = stmt.where(CouponSession.user_id.in_(user_ids))
     rows = list(db.execute(stmt).scalars())
@@ -179,7 +131,6 @@ def coupon_data_report(
         "p50_ms": _percentile(completed_elapsed, 50),
         "p95_ms": _percentile(completed_elapsed, 95),
         "p99_ms": _percentile(completed_elapsed, 99),
-        **_success_rates(rows),
     }
 
     # ── 按天趋势(柱=发起/完成数,线=平均耗时)──
@@ -272,52 +223,3 @@ def coupon_user_records(db: Session, *, user_id: int, limit: int = 100) -> dict:
         select(func.count()).select_from(CouponSession).where(CouponSession.user_id == user_id)
     ).scalar_one()
     return {"items": [_session_to_row(r) for r in rows], "total": int(total)}
-
-
-_SLOT_OK = ("success", "already_claimed")
-_SLOT_TRIED = ("success", "already_claimed", "failed")
-
-
-def coupon_slot_report(
-    db: Session, *, date_from: str, date_to: str, app_env: str | None = None
-) -> dict:
-    """按 coupon_id(具体券)聚合成功率(见 docs/guides/领券成功率指标-设计与埋点.md §13)。
-
-    数据源 coupon_claim_record(粒度=设备-天,唯一键 device+coupon+day)。
-    - 尝试 = status ∈ {success, already_claimed, failed}(skipped 排除);
-    - 成功 = status ∈ {success, already_claimed};成功率 = 成功/尝试;
-    - claim_date 区间 + app_env(None=全部)过滤;按 tried 倒序返回。
-    """
-    d_from = _date.fromisoformat(date_from)
-    d_to = _date.fromisoformat(date_to)
-    ok = case((CouponClaimRecord.status.in_(_SLOT_OK), 1), else_=0)
-    stmt = (
-        select(
-            CouponClaimRecord.coupon_id,
-            func.max(CouponClaimRecord.coupon_name).label("coupon_name"),
-            func.count().label("tried"),
-            func.sum(ok).label("succeeded"),
-        )
-        .where(
-            CouponClaimRecord.claim_date >= d_from,
-            CouponClaimRecord.claim_date <= d_to,
-            CouponClaimRecord.status.in_(_SLOT_TRIED),
-        )
-        .group_by(CouponClaimRecord.coupon_id)
-    )
-    if app_env is not None:
-        stmt = stmt.where(CouponClaimRecord.app_env == app_env)
-    items = []
-    for coupon_id, coupon_name, tried, succeeded in db.execute(stmt).all():
-        tried = int(tried or 0)
-        succeeded = int(succeeded or 0)
-        items.append({
-            "coupon_id": coupon_id,
-            "coupon_name": coupon_name,
-            "platform": coupon_id_to_platform(coupon_id),
-            "tried": tried,
-            "succeeded": succeeded,
-            "success_rate": round(succeeded / tried, 4) if tried else None,
-        })
-    items.sort(key=lambda x: (-x["tried"], x["coupon_id"]))
-    return {"items": items}

app/admin/repositories/stats.py

@@ -30,17 +30,11 @@ from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
 _BEIJING = timezone(timedelta(hours=8))
-REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
-# 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
-# coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
-# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
-# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
-COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
+COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
 COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
 EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
 UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
 REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
-    *REWARD_VIDEO_BIZ_TYPES,
     *COUPON_REWARD_BIZ_TYPES,
     *COMPARISON_REWARD_BIZ_TYPES,
     *EXCLUDED_REWARD_BIZ_TYPES,
@@ -368,7 +362,7 @@ def dashboard_overview(
     period_reward_video_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
-        CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
+        CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
     )
     period_feed_ad_coin_total = _sum(
         CoinTransaction.amount,
@@ -390,29 +384,15 @@ def dashboard_overview(
         *period_coin_conds,
         CoinTransaction.biz_type.like("task_%"),
     )
-    # 领券/比价奖励金币 = biz_type 桶(历史空,兜底)+ 该场景信息流广告实发金币
-    # (ad_feed_reward_record.feed_scene,granted;reward_date 是北京日期串,与 period 同自然日窗口)。
     period_coupon_reward_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
-    ) + _sum(
-        AdFeedRewardRecord.coin,
-        AdFeedRewardRecord.status == "granted",
-        AdFeedRewardRecord.feed_scene == "coupon",
-        AdFeedRewardRecord.reward_date >= period_from.isoformat(),
-        AdFeedRewardRecord.reward_date <= period_to.isoformat(),
     )
     period_comparison_reward_coin_total = _sum(
         CoinTransaction.amount,
         *period_coin_conds,
         CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
-    ) + _sum(
-        AdFeedRewardRecord.coin,
-        AdFeedRewardRecord.status == "granted",
-        AdFeedRewardRecord.feed_scene == "comparison",
-        AdFeedRewardRecord.reward_date >= period_from.isoformat(),
-        AdFeedRewardRecord.reward_date <= period_to.isoformat(),
     )
     period_regular_task_coin_total = _sum(
         CoinTransaction.amount,
@@ -563,7 +543,7 @@ def dashboard_overview(
             "reward_video_coin_total": _sum(
                 CoinTransaction.amount,
                 CoinTransaction.amount > 0,
-                CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
+                CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
             ),
             "reward_video_watch_count": _count(
                 AdRewardRecord,

app/admin/routers/admins.py

@@ -5,7 +5,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request
 
 from app.admin.audit import write_audit
 from app.admin.deps import AdminDb, CurrentAdmin, get_client_ip, require_role
-from app.admin.permissions import CUSTOM_ROLE, SUPER_ADMIN_ROLE, sanitize_pages
+from app.admin.permissions import SUPER_ADMIN_ROLE
 from app.admin.repositories import admin_role as role_repo
 from app.admin.repositories import admin_user as admin_repo
 from app.admin.schemas.admin import AdminCreateRequest, AdminUpdateRequest
@@ -26,22 +26,14 @@ def _active_super_count(db: AdminDb) -> int:
 
 
 def _validate_role(db: AdminDb, role: str) -> None:
-    """角色必须是 super_admin / custom(自定义) / admin_role 表里已存在的角色,否则 400。"""
-    if role in (SUPER_ADMIN_ROLE, CUSTOM_ROLE):
+    """角色必须是 super_admin 或 admin_role 表里已存在的角色,否则 400。"""
+    if role == SUPER_ADMIN_ROLE:
         return
     role_repo.ensure_builtin_roles(db)  # 空表(测试/全新库)兜底播种,再校验
     if role_repo.get_role(db, role) is None:
         raise HTTPException(status_code=400, detail=f"角色不存在: {role}")
 
 
-def _clean_override(role: str, pages_override: list[str] | None) -> list[str] | None:
-    """按最终角色算出该存的 pages_override:custom → 勾选集(过滤非法 key,可空列表);
-    非 custom → None(切回普通角色即清空自定义页)。"""
-    if role == CUSTOM_ROLE:
-        return sanitize_pages(pages_override)
-    return None
-
-
 @router.get("", response_model=list[AdminOut], summary="管理员列表(含明文密码,super_admin 专属)")
 def list_admins(db: AdminDb) -> list[AdminOut]:
     out: list[AdminOut] = []
@@ -59,16 +51,13 @@ def create_admin(
     if admin_repo.get_by_username(db, body.username) is not None:
         raise HTTPException(status_code=409, detail="用户名已存在")
     _validate_role(db, body.role)
-    override = _clean_override(body.role, body.pages_override)
     new = admin_repo.create_admin(
         db, username=body.username, password=body.password, role=body.role,
         plain_password=body.password,  # UI 建的账号留存明文,供权限管理页复看
-        pages_override=override,
     )
     write_audit(
         db, admin, action="admin.create", target_type="admin", target_id=new.id,
-        detail={"username": new.username, "role": new.role, "pages_override": override},
-        ip=get_client_ip(request), commit=True,
+        detail={"username": new.username, "role": new.role}, ip=get_client_ip(request), commit=True,
     )
     return AdminOut.model_validate(new)
 
@@ -100,8 +89,7 @@ def update_admin(
         _validate_role(db, body.role)
 
     changes: dict = {}
-    role_changed = body.role is not None and body.role != target.role
-    if role_changed:
+    if body.role is not None and body.role != target.role:
         changes["role"] = {"before": target.role, "after": body.role}
         target.role = body.role
     if body.status is not None and body.status != target.status:
@@ -111,17 +99,6 @@ def update_admin(
         changes["password"] = "reset"
         target.password_hash = hash_password(body.password)
         target.plain_password = body.password  # 同步留存明文,权限管理页复看保持一致
-
-    # 自定义可见页:按「最终角色」(target.role,已应用完角色变更)决定该存什么。
-    # - 切到/维持 custom 且传了 pages_override → 用勾选集;切到 custom 没传 → 清成空列表。
-    # - 切回普通角色 → 清空 override(_clean_override 返回 None)。
-    # - 角色没变、只传 pages_override(编辑现有 custom 用户的勾选)→ 也更新。
-    if role_changed or body.pages_override is not None:
-        new_override = _clean_override(target.role, body.pages_override)
-        if new_override != target.pages_override:
-            changes["pages_override"] = {"before": target.pages_override, "after": new_override}
-            target.pages_override = new_override
-
     if not changes:
         raise HTTPException(status_code=400, detail="无任何变更字段")
     db.commit()

app/admin/routers/auth.py

@@ -6,7 +6,6 @@ import logging
 from fastapi import APIRouter, Depends, HTTPException
 
 from app.admin.deps import AdminDb, CurrentAdmin
-from app.admin.permissions import CUSTOM_ROLE, sanitize_pages
 from app.admin.repositories import admin_role as role_repo
 from app.admin.repositories import admin_user as admin_repo
 from app.admin.schemas.auth import AdminLoginRequest, AdminLoginResponse, AdminOut
@@ -20,13 +19,9 @@ router = APIRouter(prefix="/admin/api/auth", tags=["admin-auth"])
 
 
 def _admin_out_with_pages(admin, db: AdminDb) -> AdminOut:  # noqa: ANN001
-    """AdminOut + 有效可见页(前端左侧导航按此过滤)。
-    role=="custom" → 用这个人的 pages_override(按人自定义,过滤悬空 key);其余走角色解析。"""
+    """AdminOut + 当前角色有效可见页(前端左侧导航按此过滤)。"""
     out = AdminOut.model_validate(admin)
-    if admin.role == CUSTOM_ROLE:
-        out.pages = sanitize_pages(admin.pages_override)
-    else:
-        out.pages = role_repo.effective_pages_of(db, admin.role)
+    out.pages = role_repo.effective_pages_of(db, admin.role)
     return out
 
 

app/admin/routers/coupon_data.py

@@ -18,8 +18,6 @@ from app.admin.schemas.coupon_data import (
     CouponDataOut,
     CouponDataRow,
     CouponDataSummary,
-    CouponSlotRow,
-    CouponSlotsOut,
     CouponUserRecordsOut,
 )
 from app.core.rewards import cn_today
@@ -54,10 +52,6 @@ def get_coupon_data(
     date_to: Annotated[str | None, Query(description="结束日 北京 YYYY-MM-DD,闭区间,默认=date_from")] = None,
     user: Annotated[str | None, Query(description="用户手机号/昵称模糊搜;不传=全部")] = None,
     app_env: Annotated[str, Query(description="prod(默认) / dev / all(全部环境)")] = "prod",
-    status: Annotated[
-        list[str] | None,
-        Query(description="领券状态多选 started/completed/failed/abandoned;不传=全部"),
-    ] = None,
     granularity: Annotated[
         str, Query(description="day=按天 / hour=按小时(北京);区间>1 天建议 day")
     ] = "day",
@@ -79,7 +73,7 @@ def get_coupon_data(
     env = None if app_env == "all" else app_env
     result = coupon_data.coupon_data_report(
         db, date_from=d_from.isoformat(), date_to=d_to.isoformat(),
-        user=user, app_env=env, statuses=status, granularity=granularity,
+        user=user, app_env=env, granularity=granularity,
         limit=limit, offset=offset, sort=sort,
     )
     return CouponDataOut(
@@ -93,35 +87,6 @@ def get_coupon_data(
     )
 
 
-@router.get(
-    "/coupons",
-    response_model=CouponSlotsOut,
-    summary="按券成功率(coupon_id 粒度;成功/(成功+失败),skipped 排除,设备-天口径)",
-)
-def get_coupon_slots(
-    db: AdminDb,
-    date_from: Annotated[str | None, Query(description="起始日 北京 YYYY-MM-DD,默认今天")] = None,
-    date_to: Annotated[str | None, Query(description="结束日 北京 YYYY-MM-DD,闭区间,默认=date_from")] = None,
-    app_env: Annotated[str, Query(description="prod(默认) / dev / all(全部环境)")] = "prod",
-) -> CouponSlotsOut:
-    today = cn_today()
-    d_from = _parse_day(date_from, field="date_from", default=today)
-    d_to = _parse_day(date_to, field="date_to", default=d_from)
-    if d_to < d_from:
-        raise HTTPException(status_code=422, detail="date_to 不能早于 date_from")
-    if (d_to - d_from).days + 1 > _MAX_RANGE_DAYS:
-        raise HTTPException(status_code=422, detail=f"区间最长 {_MAX_RANGE_DAYS} 天")
-    env = None if app_env == "all" else app_env
-    result = coupon_data.coupon_slot_report(
-        db, date_from=d_from.isoformat(), date_to=d_to.isoformat(), app_env=env
-    )
-    return CouponSlotsOut(
-        date_from=d_from.isoformat(),
-        date_to=d_to.isoformat(),
-        items=[CouponSlotRow(**r) for r in result["items"]],
-    )
-
-
 @router.get(
     "/user-records",
     response_model=CouponUserRecordsOut,

app/admin/routers/ops_marquee_seed.py

@@ -19,8 +19,6 @@ from app.admin.schemas.ops_marquee_seed import (
     OpsMarqueeSeedCreate,
     OpsMarqueeSeedOut,
     OpsMarqueeSeedUpdate,
-    OpsRealRecordItem,
-    OpsRealRecordsOut,
     OpsSavingsFeedPreviewOut,
 )
 from app.models.admin import AdminUser
@@ -70,22 +68,6 @@ def preview_feed(
     return OpsSavingsFeedPreviewOut(items=ops_marquee.get_feed(db, limit=limit, mode=mode))
 
 
-@router.get("/real-records", response_model=OpsRealRecordsOut, summary="分页浏览当前模式下可展示的记录(审核用)")
-def list_real_records(
-    db: AdminDb,
-    mode: Annotated[str | None, Query(description="mixed/real/seed;不传=当前持久化模式")] = None,
-    offset: Annotated[int, Query(ge=0)] = 0,
-    limit: Annotated[int, Query(ge=1, le=50)] = 8,
-) -> OpsRealRecordsOut:
-    """分页列出**当前模式**下可在 app 轮播展示的全部记录(**不去重**):只真实=真实记录;只种子=各启用
-    种子按生成逻辑各出一行;混播=真实+种子。与 app 同口径洗牌+去连簇,固定种子→翻页稳定、能翻遍全部。
-    item.user_id=0 表示种子行。"""
-    if mode is not None and mode not in ops_marquee.FEED_MODES:
-        raise HTTPException(status_code=400, detail="mode 需为 mixed / real / seed")
-    items, total = ops_marquee.list_real_records(db, mode=mode, offset=offset, limit=limit)
-    return OpsRealRecordsOut(items=[OpsRealRecordItem(**it) for it in items], total=total)
-
-
 # 注:/mode 两个端点须在 /{seed_id} 之前注册,否则 PATCH /mode 会被 /{seed_id} 抢先按 id 解析。
 @router.get("/mode", summary="首页轮播数据源模式(mixed/real/seed)")
 def get_feed_mode(db: AdminDb) -> dict:

app/admin/schemas/admin.py

@@ -13,18 +13,14 @@ class AdminCreateRequest(BaseModel):
     username: str = Field(..., min_length=3, max_length=64)
     password: str = Field(..., min_length=8, max_length=72)  # bcrypt ≤72 字节
     role: str = Field("operator", min_length=1, max_length=32)
-    # 仅当 role == "custom":这个人专属可见页 key 列表(逐页勾选结果)。其余角色不传/忽略。
-    pages_override: list[str] | None = None
 
 
 class AdminUpdateRequest(BaseModel):
-    """改角色 / 启用禁用 / 重置密码 / 自定义可见页,字段都可选(只改传了的)。"""
+    """改角色 / 启用禁用 / 重置密码,字段都可选(只改传了的)。"""
 
     role: str | None = Field(None, min_length=1, max_length=32)
     status: Literal["active", "disabled"] | None = None
     password: str | None = Field(None, min_length=8, max_length=72)
-    # 改成/更新「自定义」可见页;role 切回普通角色时后端会清空 override(见路由)。
-    pages_override: list[str] | None = None
 
 
 class AdminAuditLogOut(BaseModel):

app/admin/schemas/auth.py

@@ -21,11 +21,8 @@ class AdminOut(BaseModel):
     created_at: datetime
     last_login_at: datetime | None = None
     # 该管理员当前角色的有效可见页(= 左侧导航项 key);仅登录 / /me 填充,列表接口默认空。
-    # 前端据此过滤左侧导航(super_admin = 全部页;role=="custom" = pages_override)。见 permissions.py。
+    # 前端据此过滤左侧导航(super_admin = 全部页)。见 app/admin/permissions.py。
     pages: list[str] = []
-    # 「自定义」可见页原始勾选集(role=="custom" 时非空);列表接口下发,供权限管理页编辑回填勾选。
-    # 普通角色为 None。from_attributes 自动从 ORM 列取。
-    pages_override: list[str] | None = None
     # 明文登录密码:仅「管理员账号列表」(super_admin 专属路由)填充,供权限管理页编辑时复看;
     # 无留存(脚本建的超管 / 旧账号)为 None → 前端不显示。登录 / /me 不下发(保持 None)。
     password: str | None = None

app/admin/schemas/coupon_data.py

@@ -19,16 +19,6 @@ class CouponDataSummary(BaseModel):
     p50_ms: int | None = Field(None, description="耗时 50 分位(ms,中位数)")
     p95_ms: int | None = Field(None, description="耗时 95 分位(ms)")
     p99_ms: int | None = Field(None, description="耗时 99 分位(ms)")
-    # 平台粒度成功率(见 docs/guides/领券成功率指标-设计与埋点.md):基数含全部 session。
-    full_success_count: int = Field(0, description="整单成功数(勾选平台全部领到的 session 数)")
-    full_success_rate: float | None = Field(None, description="整单成功率②=整单成功数/发起数;无数据为空")
-    point_success_count: int = Field(0, description="成功平台点位数(Σ 每次成功的平台数)")
-    point_total_count: int = Field(0, description="总平台点位数(Σ 每次勾选平台数;空勾选=全领三档)")
-    point_success_rate: float | None = Field(None, description="点位成功率③=成功点位/总点位;无数据为空")
-    per_platform: dict[str, float | None] = Field(
-        default_factory=dict,
-        description="分平台点位成功率 {平台id: rate|None};恒含美团/淘宝/京东三档,区间内无人勾选的平台为 None",
-    )
 
 
 class CouponDataDaily(BaseModel):
@@ -91,22 +81,3 @@ class CouponUserRecordsOut(BaseModel):
 
     items: list[CouponDataRow]
     total: int
-
-
-class CouponSlotRow(BaseModel):
-    """按券成功率一行(§13):粒度=设备-天;成功率=成功/(成功+失败),skipped 排除。"""
-
-    coupon_id: str
-    coupon_name: str | None = None
-    platform: str | None = Field(None, description="美团/淘宝/京东 平台 id;无法识别为空")
-    tried: int = Field(..., description="尝试数(success+already_claimed+failed 的设备-天数)")
-    succeeded: int = Field(..., description="成功数(success+already_claimed)")
-    success_rate: float | None = Field(None, description="成功率=成功/尝试")
-
-
-class CouponSlotsOut(BaseModel):
-    """按券成功率表响应(§13)。"""
-
-    date_from: str
-    date_to: str
-    items: list[CouponSlotRow]

app/admin/schemas/ops_marquee_seed.py

@@ -62,16 +62,3 @@ class OpsSavingsFeedPreviewItem(BaseModel):
 class OpsSavingsFeedPreviewOut(BaseModel):
     """运营预览:实际混播出来的 feed(真实记录会插队,与客户端一致)。"""
     items: list[OpsSavingsFeedPreviewItem]
-
-
-class OpsRealRecordItem(BaseModel):
-    masked_user: str          # 脱敏后展示名(与 app 一致)
-    saved_amount_cents: int   # 节省金额(分)
-    created_at: str           # 比价记录时间(YYYY-MM-DD HH:MM)
-    user_id: int              # 真实 user_id(供运营核对,不下发客户端)
-
-
-class OpsRealRecordsOut(BaseModel):
-    """分页浏览「全部可展示的真实记录」(success+省>0,不去重、稳定顺序)。"""
-    items: list[OpsRealRecordItem]
-    total: int                # 满足条件的真实记录总数(算页数用)

app/api/v1/coupon.py

@@ -81,16 +81,7 @@ def _record_claims_blocking(
     device_id: str, user_id: int | None, trace_id: str | None, results: list[dict]
 ) -> None:
     with SessionLocal() as db:
-        # 取本次 session 环境,给 coupon_claim_record 打 app_env 标(每券成功率表按它过滤;设计 §13)。
-        app_env = coupon_repo.session_app_env(db, trace_id)
-        coupon_repo.record_claims(db, device_id, user_id, trace_id, results, app_env=app_env)
-        # 顺带把本帧「成功平台」并入 coupon_session.platform_success(admin 领券数据 ②整单/③点位成功率;
-        # 设计 route B,见 docs/guides/领券成功率指标-设计与埋点.md)。复用同一 SessionLocal、紧接 record_claims,
-        # 不新增连接;并集幂等(无新平台不写),trace_id 缺失或 session 行未落库则跳过。
-        if trace_id:
-            coupon_repo.merge_session_platform_success(
-                db, trace_id, coupon_repo.succeeded_platforms(results)
-            )
+        coupon_repo.record_claims(db, device_id, user_id, trace_id, results)
 
 
 def _mark_completed_blocking(

app/models/admin.py

@@ -28,11 +28,8 @@ class AdminUser(Base):
     # 明文登录密码:仅「后台 UI 创建/重置」的管理员留存,供超管在权限管理页复看转交。
     # 脚本/起后台时建的超管账号不写(为 None → 前端「不显示密码」)。⚠️ 内部工具便利取舍,见 create/list。
     plain_password: Mapped[str | None] = mapped_column(String(128), nullable=True)
-    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)/ custom(按人自定义)
+    # super_admin(全权+管账号)/ finance(钱:提现+金币)/ operator(用户+反馈+大盘)
     role: Mapped[str] = mapped_column(String(20), nullable=False, default="operator")
-    # 「自定义」权限:仅当 role == "custom" 时有效,存这个人专属的可见页 key 列表(不共享给他人)。
-    # 非 custom 用户为 None → 可见页跟随角色。登录/`/me` 下发有效页时,非空即优先用它(见 auth._admin_out_with_pages)。
-    pages_override: Mapped[list | None] = mapped_column(_JSON, nullable=True)
     # active / disabled
     status: Mapped[str] = mapped_column(String(20), nullable=False, default="active")
 

app/models/coupon_state.py

@@ -66,9 +66,6 @@ class CouponClaimRecord(Base):
 
     # success / already_claimed / failed / skipped(原样取 pricebot coupon 结果)
     status: Mapped[str] = mapped_column(String(24), nullable=False)
-    # 领券所属 session 的环境 prod/dev(/step 按 trace_id 查 coupon_session.app_env 打标)。
-    # 旧行 NULL(不回填)。admin「按券成功率」表据此过滤环境。见设计 §13。
-    app_env: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
     vendor: Mapped[str | None] = mapped_column(String(48), nullable=True)
     coupon_name: Mapped[str | None] = mapped_column(String(128), nullable=True)
     # 这张领到几张(pricebot display_count;给不出时为 None)
@@ -243,10 +240,6 @@ class CouponSession(Base):
     platform_elapsed: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
     # 领到总张数(收尾帧带)。
     claimed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    # 本次 session 至少领到一张(status∈{success,already_claimed})的平台 id 列表,如 ["meituan-waimai","jd-waimai"]。
-    # admin「领券数据」据此算整单成功率(②)/点位成功率(③);服务端 /step 逐帧按 trace_id 并集写入
-    # (见 coupon_state.merge_session_platform_success)。旧行=NULL → 视作空集。
-    platform_success: Mapped[list | None] = mapped_column(_JSON, nullable=True)
     # pricebot done 帧回传的公网调试链接(price.shaguabijia.com/traces/{dir});含落盘时分秒、拼不出,只能存
     # (同 ComparisonRecord.trace_url)。admin「领券数据」明细据此渲染可点 trace 链接;未到 done(failed/abandoned)为空。
     trace_url: Mapped[str | None] = mapped_column(String(512), nullable=True)

app/repositories/ad_feed_reward.py

@@ -176,19 +176,10 @@ 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=reward_biz, ref_id=client_event_id,
-        remark=reward_remark,
+        biz_type="feed_ad_reward", ref_id=client_event_id,
+        remark="信息流广告奖励",
     )
     rec = AdFeedRewardRecord(
         client_event_id=client_event_id,

app/repositories/coupon_state.py

@@ -147,22 +147,12 @@ def reset_today_completion(db: Session, device_id: str) -> int:
 
 # ===== 领券记录(coupon_claim_record)=====
 
-def session_app_env(db: Session, trace_id: str | None) -> str | None:
-    """按 trace_id 取 coupon_session.app_env(每券成功率表打环境标用);无 trace_id / 查不到 → None。"""
-    if not trace_id:
-        return None
-    return db.execute(
-        select(CouponSession.app_env).where(CouponSession.trace_id == trace_id)
-    ).scalar_one_or_none()
-
-
 def record_claims(
     db: Session,
     device_id: str,
     user_id: int | None,
     trace_id: str | None,
     results: list[dict],
-    app_env: str | None = None,
 ) -> int:
     """一批券领取结果幂等写入,返回写入(新增 + 更新)条数。
 
@@ -196,15 +186,12 @@ def record_claims(
                 row.user_id = user_id
             if count is not None:
                 row.claimed_count = count
-            if app_env is not None:
-                row.app_env = app_env
             row.extra = r
         else:
             db.add(CouponClaimRecord(
                 device_id=device_id, user_id=user_id,
                 coupon_id=coupon_id, claim_date=today,
-                status=status, app_env=app_env,
-                vendor=r.get("vendor"), coupon_name=r.get("name"),
+                status=status, vendor=r.get("vendor"), coupon_name=r.get("name"),
                 claimed_count=count, trace_id=trace_id, reason=r.get("reason"),
                 extra=r,
             ))
@@ -248,47 +235,6 @@ def sum_claimed_count(db: Session, user_id: int) -> int:
     return int(total or 0)
 
 
-# ===== 领券平台推导(coupon_id → 平台;成功平台集)=====
-
-# 成功语义:success + already_claimed 算成功(pricebot 代码 emit already_claimed,协议 enum 漏了);
-# failed / skipped 不算。与 sum_claimed_count 同口径。
-_SUCCESS_STATUSES = frozenset({"success", "already_claimed"})
-
-# 三档平台 id 及固定序(美团→淘宝→京东),与客户端 DEFAULT_PLATFORM_ORDER 对齐。
-DEFAULT_PLATFORMS: tuple[str, ...] = ("meituan-waimai", "taobao-shanguang", "jd-waimai")
-
-
-def coupon_id_to_platform(coupon_id: str | None) -> str | None:
-    """coupon_id 前缀 → 平台 id;无法识别 / 空 → None。
-
-    与客户端 `CouponForegroundService.couponIdToPlatform` 同词表:
-    mt_→美团外卖 / tb_·ele_·elm_→淘宝闪购 / jd_→京东外卖。
-    """
-    if not coupon_id:
-        return None
-    if coupon_id.startswith("mt_"):
-        return "meituan-waimai"
-    if coupon_id.startswith(("tb_", "ele_", "elm_")):
-        return "taobao-shanguang"
-    if coupon_id.startswith("jd_"):
-        return "jd-waimai"
-    return None
-
-
-def succeeded_platforms(results: list[dict]) -> list[str]:
-    """一批券结果 → 至少领到一张的平台集(按 DEFAULT_PLATFORMS 去重保序)。
-
-    只取 status∈{success, already_claimed} 的券;失败/跳过、无法识别平台的券跳过。
-    """
-    ok: set[str] = set()
-    for r in results:
-        if r.get("status") in _SUCCESS_STATUSES:
-            platform = coupon_id_to_platform(r.get("coupon_id"))
-            if platform is not None:
-                ok.add(platform)
-    return [p for p in DEFAULT_PLATFORMS if p in ok]
-
-
 # ===== 领券任务流水(coupon_session,admin「领券数据」看板数据源)=====
 
 def upsert_coupon_session(
@@ -375,31 +321,3 @@ def upsert_coupon_session(
     except IntegrityError:
         # 并发下另一请求刚插了同 trace_id → 唯一约束撞,回滚忽略(本就幂等)。
         db.rollback()
-
-
-def merge_session_platform_success(
-    db: Session, trace_id: str, platforms: list[str]
-) -> None:
-    """把本帧「成功平台」并入 coupon_session.platform_success(按 trace_id,并集幂等,按 DEFAULT_PLATFORMS 保序)。
-
-    - 领券 /step 每逢带券结果的帧调一次(平台成败布尔,跨帧取并集天然幂等,不重复计)。
-    - 读不到该 trace_id 的行 → **静默跳过**(不建兜底行;设计 §5:started 帧几乎必先落库)。
-    - 并集无变化(该平台已记过)→ 不写库,省一次 UPDATE。
-    - fire-and-forget:调用方已吞异常;并发唯一冲突回滚忽略。
-    """
-    if not platforms:
-        return
-    row = db.execute(
-        select(CouponSession).where(CouponSession.trace_id == trace_id)
-    ).scalar_one_or_none()
-    if row is None:
-        return
-    merged = set(row.platform_success or []) | set(platforms)
-    new_list = [p for p in DEFAULT_PLATFORMS if p in merged]
-    if new_list == (row.platform_success or []):
-        return  # 幂等:无新平台,不写
-    row.platform_success = new_list
-    try:
-        db.commit()
-    except IntegrityError:
-        db.rollback()

app/repositories/ops_marquee.py

@@ -217,28 +217,12 @@ def _recent_real_rows(db: Session) -> list[tuple[int, int, str | None]]:
     return out
 
 
-def _shuffle_declustered(rows: list, rng: random.Random | None = None) -> list:
-    """洗牌 + 「去连簇」:先洗牌,再贪心重排让相邻两条尽量不是同一 user_id(元素 [0] 即 user_id)。
-    rng=None → 用全局 _rng(feed 每次新随机);传入 rng(如固定种子 Random)→ 排列确定(admin 稳定分页)。
-    减少同一用户连续出现;只有少数几个用户时 best-effort。"""
-    rng = rng or _rng
-    pool = list(rows)
-    rng.shuffle(pool)
-    result: list = []
-    while pool:
-        prev_uid = result[-1][0] if result else None
-        # 优先挑与上一条不同 user 的;挑不到(只剩同 user)才取第一个
-        idx = next((i for i, r in enumerate(pool) if r[0] != prev_uid), 0)
-        result.append(pool.pop(idx))
-    return result
-
-
 def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]:
     """返回最多 limit 条 {masked_user, saved_amount_cents, time(HH:MM:SS 北京)}。
 
     mode:显式传入(admin 预览指定模式)则用它、**不改持久化配置**;不传(客户端 /savings-feed)读
     持久化的 marquee_feed_mode;非法值一律回退到持久化模式。
-    真实条:success 且 0 < saved ≤ 上限,**不按 user 去重**(打乱 + 去连簇:相邻尽量不同用户、减少单人连刷)后取前 limit。
+    真实条:success 且 0 < saved ≤ 上限,按 user 去重(同一用户只取最新一条,避免单人刷屏)。
     不足用启用的种子补齐——**公平随机抽取** need 个(而非固定取前 N),让所有种子都有机会露出;
     种子用户名留空则随机合成(避开撞名),金额取**长尾随机**(小额居多、偶尔大额,更像真实分布)。
     真实 + 种子仍不满 limit → 内置合成条**补满**,保证轮播既不空也不稀疏。
@@ -251,13 +235,19 @@ def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]
     items: list[dict] = []
     used_names: set[str] = set()
 
-    # 真实条(mixed / real):**不按 user 去重**——打乱 + 去连簇(相邻尽量不同用户、减少单人连刷)后取前
-    # limit;金额超上限的异常值已在查询剔除。同一用户可多条露出,但被打散、尽量不连续。
+    # 真实条(mixed / real):取较多近期记录(带 ~30s 缓存)后按 user 去重;金额超上限的异常值已在查询剔除。
     if mode != "seed":
-        for uid, sc, nick in _shuffle_declustered(_recent_real_rows(db))[:limit]:
+        rows = _recent_real_rows(db)
+        seen_users: set[int] = set()
+        for uid, sc, nick in rows:
+            if uid in seen_users:
+                continue
+            seen_users.add(uid)
             name = _mask_real(nick, uid)
-            used_names.add(name)  # 同一用户可多条,名字重复无害(used_names 仅供种子避重)
+            used_names.add(name)  # 真实名按昵称/id 稳定;偶发撞名可接受
             items.append({"masked_user": name, "saved_amount_cents": int(sc)})
+            if len(items) >= limit:
+                break
 
     # 种子补位 + 合成兜底(mixed / seed):mixed 下补真实不足的部分,seed 下全量用种子/合成。
     # real 模式**跳过**——只出真实,不掺任何假数据(真实不足则少于 limit,为 0 时返回空)。
@@ -310,78 +300,6 @@ def get_feed(db: Session, limit: int = 8, mode: str | None = None) -> list[dict]
     return items
 
 
-# ===== admin 侧:分页浏览「当前模式下可展示的记录」(审核用,不去重) =====
-_REAL_BROWSE_CAP = 1000   # 真实记录一次最多纳入这么多去洗牌+分页(足够审核;防超大库全量洗牌)
-_BROWSE_SEED = 20260707   # 固定洗牌种子:同一批数据下排列恒定 → 翻页稳定、能翻遍全部
-
-
-def _seed_browse_row(seed: OpsMarqueeSeed) -> dict:
-    """把一条种子按其「生成逻辑」**确定性**生成一行浏览项(名字/金额按 seed.id 派生固定种子 → 翻页稳定)。
-    名字:运营手填的非模板名原样用,否则本地合成;金额:区间内确定性长尾取值。user_id=0(种子无真实用户)。"""
-    r = random.Random(_BROWSE_SEED * 1_000_003 + int(seed.id))
-    fixed = (seed.masked_user or "").strip()
-    name = fixed if (fixed and not fixed.startswith("用户****")) else _synth_name(r)
-    lo = max(0, int(seed.min_cents))
-    hi = max(lo, int(seed.max_cents))
-    amt = lo if hi <= lo else lo + int(round((hi - lo) * (r.random() ** 2.2)))
-    return {"masked_user": name, "saved_amount_cents": amt, "created_at": "", "user_id": 0}
-
-
-def list_real_records(
-    db: Session, mode: str | None = None, offset: int = 0, limit: int = 8
-) -> tuple[list[dict], int]:
-    """分页浏览「**当前模式**下可在 app 轮播展示的全部记录」,供 admin 逐页审核(**不去重**):
-      - 只真实:全部 success+省>0 的真实记录;
-      - 只种子:每条启用种子按生成逻辑各出一行;
-      - 混播:真实 + 种子 全部合在一起。
-    与 app 轮播同口径:先洗牌 + 去连簇(相邻尽量不同 user;种子各自独立、不算同 user);**固定种子** →
-    同批数据下排列恒定,翻页不跳、能翻遍全部。返回 (items, total);item.user_id=0 表示种子。"""
-    mode = mode if mode in FEED_MODES else get_feed_mode(db)
-    # pool: [(cluster_key, item)];cluster_key 供去连簇——真实=user_id、种子=各自唯一负数(互不聚簇)
-    pool: list[tuple[int, dict]] = []
-    if mode != "seed":
-        rows = db.execute(
-            select(
-                ComparisonRecord.user_id,
-                ComparisonRecord.saved_amount_cents,
-                User.nickname,
-                ComparisonRecord.created_at,
-            )
-            .join(User, User.id == ComparisonRecord.user_id)
-            .where(
-                ComparisonRecord.status == "success",
-                ComparisonRecord.saved_amount_cents > 0,
-                ComparisonRecord.saved_amount_cents <= _REAL_MAX_CENTS,
-            )
-            .order_by(ComparisonRecord.created_at.desc())
-            .limit(_REAL_BROWSE_CAP)
-        ).all()
-        for uid, sc, nick, ca in rows:
-            pool.append((
-                int(uid),
-                {
-                    "masked_user": _mask_real(nick, int(uid)),
-                    "saved_amount_cents": int(sc),
-                    "created_at": str(ca)[:16] if ca is not None else "",
-                    "user_id": int(uid),
-                },
-            ))
-    if mode != "real":
-        seeds = (
-            db.execute(select(OpsMarqueeSeed).where(OpsMarqueeSeed.enabled.is_(True)))
-            .scalars()
-            .all()
-        )
-        for i, s in enumerate(seeds):
-            pool.append((-(i + 1), _seed_browse_row(s)))  # 每个种子唯一 key → 互不聚簇
-    total = len(pool)
-    # 每次用同一固定种子新建 Random → 同批数据排列恒定(翻页稳定);同时相邻尽量不同 user。
-    ordered = _shuffle_declustered(pool, random.Random(_BROWSE_SEED))
-    off = max(0, offset)
-    items = [item for _key, item in ordered[off : off + limit]]
-    return items, total
-
-
 # ===== 运营侧:种子 CRUD =====
 def list_seeds(db: Session) -> list[OpsMarqueeSeed]:
     return list(

docker-compose.yml

@@ -0,0 +1,22 @@
+# 本地开发/测试用 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

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

docs/database/coupon_state.md

@@ -1,4 +1,4 @@
-# coupon_state — 领券状态表(今日状态三张 + 领券流水 coupon_session)
+# coupon_state — 领券今日状态三张表(弹窗频控 / 首页置灰 / 领券记录)
 
 > 模型 `app/models/coupon_state.py` · 仓库 `app/repositories/coupon_state.py` · 接口 `app/api/v1/coupon.py`(prefix `/api/v1/coupon`) · [← 索引](./README.md) · [总览](./OVERVIEW.md)
 
@@ -7,9 +7,8 @@
 - **`coupon_prompt_engagement`** — 弹窗频控源。按 `(device, App 包名, 自然日)` 记「今天**这个 App** 是否对领券引导窗表达过**意向**」（弹出即记 `shown` / 点「一键领取」=`claim_started` / 点拒绝关闭=`dismissed` 都算）。切到外卖 App 时据此决定弹不弹：今天**该 App** engage 过就不再弹该 App。频控维度自 2026-06-14 起含 `package`，美团/淘宝/京东各自独立、互不压制。
 - **`coupon_daily_completion`** — 首页置灰源。按 `(device, 自然日)` 记「今天是否已**跑完整轮**领券（到 done 帧）」。首页「去领取」卡据此置灰：今天跑完了就不能再领。
 - **`coupon_claim_record`** — 资产沉淀层。按 `(device, 券, 自然日)` 记每张券的领取结果（success/already_claimed/failed/skipped），**纯沉淀**（资产/画像/排查/CPS 归因），当前**不参与**「要不要领 / 弹不弹」的判断。
-- **`coupon_session`** — admin「领券数据」看板数据源（**独立流水表,不是「今日状态」表**）。按 `trace_id` 一次领券一行，走 `POST /api/v1/coupon/session` 两段上报（发起/收尾），记全程耗时 + 各平台耗时 + `platform_success`（成功平台,算整单②/点位③成功率）。详见下方专节。
 
-前三张「今日状态」表的共同口径：
+三表共同口径：
 - **判断维度是 `device_id`，不是 `user_id`**：券发到的是设备上登录的那个外卖账号，device 比 user 更贴近「哪个登录环境」，且 `device_id` 全链路现成、不依赖领券鉴权（领券 MVP 阶段 `/coupon/step` 不鉴权）。客户端 `getOrCreateDeviceId` 生成存 SP，**卸载重装会变 → 当新设备重新弹一次**（产品预期）。
 - **日期 = `Asia/Shanghai` 自然日**（`claim_date` / `engage_date` / `complete_date`，`repositories/coupon_state.today_cn()`）。每日可领的券（签到/天天红包）靠这天然每天一条。
 - **`user_id` 可空**：领券登录态有就记（资产/画像），可空、**不进唯一键、不阻塞判断**。
@@ -91,7 +90,7 @@
 ### 用在哪 / 增删改查
 - **C / U（幂等 upsert）**：`record_claims`，由 `POST /api/v1/coupon/step` 写入。一帧的券结果来自 pricebot 的 `last_coupon_result`（最后一张）+ `action.params.coupon_results`（全量）——**会重复带同一张券**，端点 `_extract_coupon_results` 先**按 `coupon_id` 去重**（全量覆盖单张），仓库再靠唯一键幂等：已有则更新 `status`/`reason`/`claimed_count`/`extra`（以最后一次为准），否则插入。
 - **U / D**：无业务删除。
-- **R**：`GET /admin/api/coupon-data/coupons`（`coupon_slot_report`）—— admin「按券成功率」表,按 `coupon_id` 聚合 成功/(成功+失败)（`skipped` 排除,设备-天口径,按 `app_env` 过滤）。见设计 §13。
+- **R**：**当前无读取端点**（纯写入沉淀，未来做去重/归因/画像时再用）。
 
 ### 字段
 | 列 | 类型 | 约束 / 默认 | 说明(取值 / join) |
@@ -102,7 +101,6 @@
 | `coupon_id` | String(64) | NOT NULL | 券标识（取自 pricebot 结果） |
 | `claim_date` | **Date** | NOT NULL | **北京时间**自然日（`today_cn()`）；每日可领的券靠它天然每天一条 |
 | `status` | String(24) | NOT NULL | `success` / `already_claimed` / `failed` / `skipped`（原样取 pricebot coupon 结果） |
-| `app_env` | String(16) | index, 可空 | 领券所属 session 环境 `prod`/`dev`（`/step` 按 `trace_id` 取 `coupon_session.app_env` 打标）；旧行 NULL（不回填）。admin「按券成功率」表按它过滤。见设计 §13 |
 | `vendor` | String(48) | 可空 | 券提供方 |
 | `coupon_name` | String(128) | 可空 | 取 pricebot `name` |
 | `claimed_count` | Integer | 可空 | 这张领到几张（pricebot `display_count`，给不出时 None；兼容 `claimed_count`） |
@@ -122,49 +120,7 @@
 
 ---
 
-## coupon_session — 领券任务流水（一次领券一行,admin「领券数据」看板数据源）
-
-`trace_id` 唯一,一次领券一行。与上面三张「今日状态」表不同:本表走 `POST /api/v1/coupon/session`(客户端**两段上报**:发起 `started` 建行、收尾 `completed`/`failed`/`abandoned` 按 `trace_id` 更新同一行),记从发起到收尾的全程耗时 + 各平台耗时 + 机型/ROM。发起即落库 → admin 可算「发起数」与中途流失（started 无终态 = 未完成）。
-
-### 用在哪 / 增删改查
-- **C / U（幂等 upsert）**：`upsert_coupon_session`,由 `POST /api/v1/coupon/session` 两段上报。**状态只前进**（started 帧重复到不覆盖已有终态）;终态补 `finished_at`。
-- **U（并集写）**：`merge_session_platform_success`,由 `POST /api/v1/coupon/step` 每逢**带券结果的帧**调用——把本帧「成功平台」（`status∈{success,already_claimed}` 的券 → `coupon_id` 前缀映射平台）**并入** `platform_success`（并集幂等,无新平台不写;读不到该 trace 行则跳过）。复用 `record_claims` 的同一 `SessionLocal`,不新增连接。
-- **R**：admin `GET /admin/api/coupon-data`（`coupon_data_report`）—— 发起/完成数、耗时分位、**整单成功率②/点位成功率③**、按天/小时趋势、逐条明细;`GET /admin/api/coupon-data/user-records` 某用户全部领券。
-
-### 字段
-| 列 | 类型 | 约束 / 默认 | 说明 |
-|---|---|---|---|
-| `id` | Integer | PK, autoincrement | |
-| `trace_id` | String(64) | NOT NULL, UNIQUE | 一次领券唯一 id（客户端 UUID,全程贯穿）,upsert 键 |
-| `device_id` | String(64) | NOT NULL | |
-| `user_id` | Integer | index, 可空 | 登录态才带（admin join 用户表出手机号/昵称）;匿名领券为空 |
-| `status` | String(16) | NOT NULL | `started` / `completed` / `failed` / `abandoned`;started 无终态 = 中途流失 |
-| `app_env` | String(16) | index, 可空 | `prod` / `dev`;admin 报表默认只看 prod（防测试串台） |
-| `platforms` | JSON | 可空 | 发起勾选平台 `["meituan-waimai",…]`（空 = 全领三档）;**③点位成功率的分母来源** |
-| `origin_package` | String(64) | 可空 | 发起来源外卖 App 包名;null = App 内（傻瓜比价首页）发起,非空 = 外卖侧弹券 |
-| `device_model` | String(128) | 可空 | Build.MANUFACTURER + MODEL |
-| `rom` | String(64) | 可空 | OemDetector,如 "ColorOS 14" |
-| `started_at` | DateTime(tz) | NOT NULL | 发起时刻（客户端墙钟）;明细「时间」列、趋势 X 轴 |
-| `started_date` | **Date** | NOT NULL | 发起的**北京**自然日;admin 按天聚合/筛选（索引） |
-| `finished_at` | DateTime(tz) | 可空 | 收尾时刻（服务端 now）;未收尾（流失）为空 |
-| `elapsed_ms` | Integer | 可空 | 全程耗时（ms,客户端点发起→收尾）;均值/分位只统计 completed |
-| `platform_elapsed` | JSON | 可空 | 各平台领券耗时 `{"meituan-waimai":3200,…}`（ms） |
-| `claimed_count` | Integer | 可空 | 领到总张数（收尾帧带） |
-| `platform_success` | JSON（PG JSONB） | 可空 | **本次至少领到一张（`status∈{success,already_claimed}`）的平台 id 列表**,如 `["meituan-waimai","jd-waimai"]`。`/step` 逐帧按 `trace_id` **并集**写入（`merge_session_platform_success`）;旧行 NULL 视作空集。admin 据此算整单成功率②（`platforms`⊆`platform_success`）/点位成功率③（Σ交集/Σ勾选）。设计:[领券成功率指标](../guides/领券成功率指标-设计与埋点.md) |
-| `trace_url` | String(512) | 可空 | pricebot done 帧回传的公网 trace 链接;未到 done（failed/abandoned）为空 |
-| `created_at` | DateTime(tz) | server_default now() | |
-| `updated_at` | DateTime(tz) | server_default now(), onupdate now() | |
-
-### 索引与约束
-- PK `id`；index `user_id`、`app_env`；UNIQUE(`trace_id`) = `uq_coupon_session_trace`；Index(`started_date`, `app_env`) = `ix_coupon_session_date_env`（admin 主聚合/筛选）。
-
-### 注意
-- `platform_success` 是**布尔性质**的平台集,跨帧**并集**天然幂等 → `/step` 每帧并入不重复计;失败/中途退出的 session 也能拿到崩溃前已成的平台。写放大 ≈ 领券券数（仅带券结果的帧写）。
-- 成功率**基数 = 区间全部 session**（含 abandoned/failed）,与「发起数」同基数（设计 §3）。`coupon_id → 平台` 用前缀（`mt_`/`tb_`·`ele_`·`elm_`/`jd_`）,与客户端 `couponIdToPlatform` 同词表。
-
----
-
-## 今日状态三表共性小结
+## 三表共性小结
 - 数据流向：客户端 → `POST /api/v1/coupon/step`（透传给 pricebot）→ 结果回写这三张表（best-effort，写库失败不影响领券）。
 - 唯一键都含 `device_id` + 某个北京自然日列（engagement 还含 `package`，按 App 频控）；`user_id` 永远是可空旁路（资产留痕，不进唯一键、不阻塞判断）。
 - 无硬外键：`user_id` 软指 `user.id`、`trace_id` 软指 pricebot work_logs（详见 [OVERVIEW → 表间关系 & Join Key](./OVERVIEW.md)）。

docs/database/postgres-migration.md

@@ -27,7 +27,21 @@ PG 默认上 16 版(工具链最齐),驱动用 **psycopg3**(SQLAlchemy 2.0 时
 
 ## 1. 本地起 PG + 跑通空库(半天)
 
-### 1.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 时的手动方式)
 
 macOS:
 ```bash

docs/guides/领券成功率指标-设计与埋点.md

@@ -1,294 +0,0 @@
-# 领券「整单成功率」与「点位成功率」指标 — 设计与埋点
-
-- 日期：2026-07-07
-- 状态：待评审
-- 涉及仓库：`shaguabijia-app-server`（**纯服务端**；客户端零改动）
-- 数据源表：`coupon_session`（admin「领券数据」看板）
-
-## 1. 背景与目标
-
-admin「领券数据」看板（数据源 `coupon_session`，见 `app/admin/repositories/coupon_data.py`）当前能算：**领券发起数、完成数、全程耗时均值/分位**。产品还想要两个成功率指标，现有埋点算不出来：
-
-- **② 整单成功率** = 一次发起里勾选的平台**全部**领到券的次数 / 领券发起数
-- **③ 点位成功率** = 平台维度的领取成功率（每个平台「点位」成没成功）
-
-> 口径决定（2026-07-07）：「点位」= **平台粒度**（美团 / 淘宝闪购 / 京东），不是「每张券」。
-
-耗时中位数、发起数已由 `coupon_session.elapsed_ms` / `status=started` 计数满足，本设计只补 ②③。
-
-## 2. 关键结论：per-slot 信号已在库，缺的是「按 session 可靠归因 + 看板可过滤」
-
-服务端 `/api/v1/coupon/step`（`app/api/v1/coupon.py`）每帧都调 `record_claims`，把**每张券**的结果（`status ∈ success / already_claimed / failed / skipped`）写进 `coupon_claim_record`，还带 `trace_id`。原始成败信号**已经落库**。
-
-但该表**不能**直接支撑本指标：
-
-1. 唯一键是 `(device_id, coupon_id, claim_date)`，**不含 trace_id**；且 `record_claims` 冲突更新时**不更新 trace_id**（`app/repositories/coupon_state.py` 的 `record_claims`，仅 INSERT 时写 trace_id）。→ 同一张券当天跨多次 session 会折叠成一行、只归属**最早**那次 → **按 session 归因不可靠**（直接砸 ②「整单全成功」）。
-2. `coupon_claim_record` 无 `app_env` / `origin_package` → 无法像看板那样只看 prod、也无法拆 Path A（App 内发起）/ Path B（外卖侧弹券）。
-
-因此采用 **route B**：在 `/step` 里服务端推导「本次 session 哪些平台成功」，直接写到 `coupon_session` 行——该表按 `trace_id` 唯一、已带 `app_env` / `origin_package` / `platforms`，指标干净可过滤、可拆路径。
-
-## 3. 指标口径（平台粒度，已定）
-
-记一次 session 为 `s`：
-
-- `sel(s)` = **勾选平台集** = `coupon_session.platforms`；为空表示「全领」→ 取默认 `{meituan-waimai, taobao-shanguang, jd-waimai}`。`|sel(s)|` = 该次「点位数」。
-- `succ(s)` = **成功平台集** = 本次 session 里**至少领到一张**（`status ∈ {success, already_claimed}`）的平台集合。
-
-指标：
-
-- **③ 点位成功率** = `Σ_s |succ(s) ∩ sel(s)|` / `Σ_s |sel(s)|`
-  （分母即「发起数 × 各自点位数」；全部全领时等于 发起数 × 3）
-- **② 整单成功率** = `#{ s : sel(s) ⊆ succ(s) 且 sel(s) ≠ ∅ }` / `发起数`
-
-已定边界：
-
-1. **③ 分母用「勾选平台」`sel(s)`**。勾了淘宝但淘宝没领到 = 该点位未成功（不特判「平台没券」）。若日后要「真没券的平台不计入分母」，再引入 `platform_attempted`（见 §8）。
-2. **②③ 基数 = 区间内全部 session**（含 `started` / `failed` / `abandoned`），与看板「发起数」同基数。**失败 / 中途退出的 session 的 `succ(s)` 取它崩溃前真领到的平台**（不一律算 0）。
-3. **成功语义**：`status ∈ {success, already_claimed}` = 成功；`failed` / `skipped` = 未成功。与 `sum_claimed_count`（`app/repositories/coupon_state.py`）一致。
-   > pricebot 协议文档把 `status` enum 写作 `success|failed|skipped`，但**代码实际还会 emit `already_claimed`**（pricebot `app/services/coupon_provider.py` 等多处）——以代码为准，含 already_claimed 是对的。`skipped` 目前 MVP 阶段基本不出现。
-
-## 4. 数据模型改动
-
-`coupon_session`（`app/models/coupon_state.py::CouponSession`）新增一列：
-
-| 列 | 类型 | 说明 |
-|----|------|------|
-| `platform_success` | `_JSON`（PG→JSONB / SQLite→JSON），nullable | 本次 session **至少领到一张**的平台 id 列表，如 `["meituan-waimai","jd-waimai"]`。旧行 = `NULL` → 视作空集。 |
-
-- alembic 新迁移：`add_column coupon_session.platform_success`，nullable、无 server_default。
-- **分母 `sel(s)` 复用已有 `platforms` 列，不新增字段。** 净新增仅此一列。
-- 同步更新表字典 `docs/database/coupon_state.md`。
-
-## 5. 服务端推导逻辑（`/step`）
-
-在 `app/api/v1/coupon.py::coupon_step` 内（已有 `results = _extract_coupon_results(resp_json)`）新增：
-
-1. 对每条 result 求平台：**按 `coupon_id` 前缀映射**（与客户端 `CouponForegroundService.couponIdToPlatform` 对齐，且与 `platforms` / `platform_elapsed` 用同一套平台 id 词表）：
-   - `mt_` → `meituan-waimai`
-   - `tb_` / `ele_` / `elm_` → `taobao-shanguang`
-   - `jd_` → `jd-waimai`
-   - 其余 → 跳过（无法识别）
-   > 不用券的 `vendor` 字段做映射：vendor 词表未必等于这三档平台 id，而 `sel(s)` 用的就是这三档，`succ(s)` 必须同词表。
-2. 收集 `status ∈ {success, already_claimed}` 的平台集合 `ok_platforms`。
-3. 若 `device_id` 且 `ok_platforms` 非空：调用新 repo 函数把 `ok_platforms` **并入** `coupon_session.platform_success`（按 `trace_id`）。
-
-新增 `app/repositories/coupon_state.py::merge_session_platform_success(db, trace_id, ok_platforms)`：
-
-- 读现有行 → `platform_success = 现有 ∪ ok_platforms`（去重、保序）→ 写回、commit；`IntegrityError` 回滚忽略（同现有 upsert 兜底）。
-- **并集幂等**：跨帧多次并入同一平台不会重复；done 帧的全量 `coupon_results` 保证完整；失败 / 中途退出的 session 靠中间帧 `last_coupon_result` 已并入的平台拿到「部分成功」。
-- **fire-and-forget**：包 `run_in_threadpool` + 整段 try/except 只 log，绝不连累 `/step` 返回（与现有 `record_claims` / `mark_completed` 同规格）。
-
-**建行 / 时序（已定：行不存在则跳过本次并入）**：
-
-- 客户端 `/session started` 在 `start()`（任务发起那刻）就发，而首个带券的 `/step` 要等领券循环跑起来、晚几秒；到 `/step` 有 `ok_platforms` 时，`coupon_session` 行几乎必然已存在。故 `merge_session_platform_success` **读不到行就跳过**，不建兜底行——实现最简，也不引入 `started_at` 不精确的脏行。
-- 残留丢数窗口：仅当「`started` 上报丢失」**且**「`/step` done 先于 `/session` terminal 落库」两者同时成立，该 session 的 `platform_success` 才会缺（terminal 帧会建行但那之后没有 `/step` 再并入）。两条件叠加概率极低，且指标是聚合口径、可容忍个别缺失。
-- 若上线后观测到该缺失不可忽略，再降级为「读不到行则 upsert 建最小兜底行（`started_at=now()`）」——届时改 `merge_session_platform_success` 一处即可，不影响其余设计。
-
-**写放大（重要，非每 step）**：`merge_session_platform_success` **只在「本帧带券结果」时触发**——即 pricebot 在**单券完成帧**给 `last_coupon_result`、**最终 done 帧**给全量 `coupon_results` 的那些帧；领单张券途中的导航/点击帧（占 step 大头）`_extract_coupon_results` 返回空 → **不写**。所以写频次 ≈ **本次领的券数**（通常个位数），且落在**服务端今天已有的** `record_claims` 写的**同一批帧**上，不新增写的帧。
-
-实现：把 merge 放进**现有 `_record_claims_blocking` 的同一个 `SessionLocal`**（紧接 `record_claims`），边际成本 = 每张券完成时多一条 `UPDATE coupon_session`（按 `trace_id` 唯一索引），不新增连接 / 不新增 `run_in_threadpool` 调用。
-
-可选降级（若要「一次 session 只写一次」）：只在 done 帧写 `platform_success`（全量 `coupon_results` 一次算完）。代价：`failed` / `abandoned`（没 done 帧）拿不到「崩溃前已成平台」→ 失败单部分成功丢失，与 §3「失败单取实际成的平台」相悖。**默认取每券帧并入**（失败单也如实统计），此降级留作观测到写压力后再启用。
-
-## 6. admin 聚合与呈现
-
-`app/admin/repositories/coupon_data.py::coupon_data_report` 的 `summary` 增加（沿用「全量拉区间 → Python 聚合」风格，与分位一致）：
-
-- `full_success_rate`（②）、`point_success_rate`（③）
-- 可选 `per_platform`：`{platform: rate}`（各平台点位成功率，拆美团/淘宝/京东）
-
-计算：对区间内 sessions，`sel = platforms or 默认三档`，`succ = set(platform_success) ∩ sel`；按 §3 公式汇总。
-
-- 过滤：默认 `app_env == 'prod'`（同现有分位口径，防测试串台）。
-- 拆路径：`origin_package` 已在表上 → 看板后续可加「Path A / Path B」筛选项（本设计不含前端图表细节）。
-- schema：`app/admin/schemas/coupon_data.py` 的 summary 加对应字段（+ 可选 `per_platform`）。
-
-## 7. 改动清单
-
-- [x] `app/models/coupon_state.py`：`CouponSession` 加 `platform_success`
-- [x] `alembic/versions/`：新迁移 add column `coupon_session.platform_success`
-- [x] `app/repositories/coupon_state.py`：新增 `merge_session_platform_success`
-- [x] `app/api/v1/coupon.py`：`/step` 推导 `ok_platforms` 并 union（前缀映射 + fire-and-forget）
-- [x] `app/admin/repositories/coupon_data.py`：`summary` 加 ②③（+ 可选 `per_platform`）
-- [x] `app/admin/schemas/coupon_data.py`：`summary` schema 加字段
-- [x] `docs/database/coupon_state.md`：补 `platform_success` 列说明（并补 coupon_session 整节）
-- [x] `tests/`：`tests/test_coupon_platform_success.py`（10 测试，见 §9）
-
-## 8. 不做（YAGNI / 边界）
-
-- **客户端不改、历史不回填**：`platform_success` 只对新 session 生效（route B 的固有取舍，用户已接受）。
-- **不引入 `platform_attempted`**：③ 分母用勾选平台。若日后要「排除真没券的平台」，注意 `platform_elapsed.keys()` 已近似「被处理过的平台」，可作 attempted 的现成来源，多半仍不必加列。
-- **不动 `coupon_claim_record`** 的去重 / 归因：本指标绕开它，避免牵动频控 / 资产 / CPS 语义。
-- **不做券级（每张券）成功率**：已选平台粒度。
-
-## 9. 测试口径要点
-
-- **union 幂等**：同一 `trace_id` 多帧并入同一平台，`platform_success` 不重复、保序。
-- **失败单部分成功**：session `failed`，但美团已成 → `succ = {meituan-waimai}`，计入 ③ 分子；② 仅当 `sel ⊆ succ` 才算整单成功。
-- **空 `platforms` → `sel` 取默认三档**（全领）。
-- **成功语义**：`already_claimed` 计成功；`skipped` 不计。
-- **基数**：`abandoned` 计入 ②③ 基数，`succ` 取实际成的平台。
-- **prod 过滤**：`dev` 环境 session 不进指标。
-
-## 10. 对 pricebot-backend 的影响
-
-**结论：不需要改 pricebot，也不改发往 pricebot 的请求 / 不加调用 / 不加负载。**
-
-- `/step` 仍原样透传请求 bytes 给 pricebot；本设计只**多解析 pricebot 的响应**（`coupon_results` / `last_coupon_result`），而这两个字段服务端**今天已在** `_extract_coupon_results` / `record_claims` 里解析。零新增字段需求、零额外上游调用、pricebot 负载不变。
-- **只读依赖（既有耦合，非新引入）**：平台映射靠 pricebot 的 `coupon_id` 前缀约定（`mt_` / `tb_` / `ele_` / `elm_` / `jd_`）。客户端 `couponIdToPlatform` 早就依赖同一套；本设计只是加了这份映射的第二个消费者。维护耦合：pricebot 若改 `coupon_id` 前缀，客户端与本指标会**一起**失效——但这是既有风险，依赖方向不变。无法识别前缀的券按「跳过」处理（与客户端一致）。
-- 认账的 pricebot 事实（源：`app/models/response.py` + `docs/projects/领券-客户端对接协议.md`）：
-  - `CouponResult = {coupon_id, name, vendor, status, reason?, duration_ms?}`；`coupon_results` 仅最终 done 帧全量，中间帧走 `last_coupon_result`（单张）。
-  - `vendor` 是来源标签（`meituan_internal` / `dianping_cps` …），**不等于**三档平台 id；且 `mt_dianping_xxx`（大众点评 CPS）也带 `mt_` 前缀归美团 → 印证「用 `coupon_id` 前缀、不用 `vendor`」正确。
-
----
-
-## 11. 实现状态（交付记录 · 2026-07-07）
-
-**状态:实现完成、TDD 全绿、未提交、迁移未应用。** 纯服务端（shaguabijia-app-server），客户端 / pricebot 未动。
-
-### 已交付改动
-| 文件 | 改动 |
-|---|---|
-| `app/models/coupon_state.py` | `CouponSession` 加 `platform_success`（`_JSON`, nullable） |
-| `app/repositories/coupon_state.py` | `coupon_id_to_platform` / `succeeded_platforms` / `merge_session_platform_success` + 常量 `DEFAULT_PLATFORMS` / `_SUCCESS_STATUSES` |
-| `app/api/v1/coupon.py` | `_record_claims_blocking` 内、`record_claims` 之后并入本帧成功平台（同一 `SessionLocal`） |
-| `app/admin/repositories/coupon_data.py` | `_success_rates(rows)` → summary 加 `full_success_rate②` / `point_success_rate③` + 3 个支撑计数 |
-| `app/admin/schemas/coupon_data.py` | `CouponDataSummary` 加 5 字段 |
-| `alembic/versions/coupon_session_platform_success.py` | add column;revision=`coupon_session_platform_success`，down=`admin_user_plain_password`（当前 head） |
-| `docs/database/coupon_state.md` | 补 `coupon_session` 整节（表原本无文档）+ 新列 |
-| `tests/test_coupon_platform_success.py` | 10 个测试（TDD） |
-
-> §7 里「可选 `per_platform`」原标 YAGNI 延后；**已在 §12（2026-07-08）补做**（随 admin 前端看板卡一并接入，见下）。
-
-### 测试与验证
-- 本特性:`Set-Location e:\project\shaguabijia-app-server; & .\.venv\Scripts\python.exe -m pytest tests\test_coupon_platform_success.py -q` → **10 passed**。
-- 全量 `pytest -q`:**306 passed / 5 failed**。5 个失败**全部预存、与本次无关**:
-  - `test_coupon_proxy.py::test_coupon_step_passes_body_through`（测试断言 `json=` 但 handler 用 `content=` 转发；写代码前 sanity run 就红）。
-  - `test_invite.py` + `test_invite_compare_reward.py` 共 4 个（单独跑也红；本次改动集零 invite 文件）。
-- 迁移:临时库 `alembic upgrade head` 通过、列已建、单一 head。
-- lint:新增代码 `ruff` 全清;`coupon_state.py` 剩 3 处 pre-existing UP017（`upsert_coupon_session` 的 `timezone.utc`）未动。
-
-### 待办（在后端项目里继续）
-1. **应用迁移**:`alembic upgrade head`（DDL 已验；线上只前进统计、历史不回填）。
-2. **提交**:尚未提交;建议先开分支再提交。
-3. **admin 前端图表**:后端指标已就绪（summary 的 `full_success_rate` / `point_success_rate` 等），看板卡 / 趋势展示待接前端。
-4. （可选）修预存 `test_coupon_step_passes_body_through`（一行:capture `content` 而非 `json`）。
-
-### 续开发须知
-- `platform_success` 只在**带券结果的帧**写（`/step` 里 `succeeded_platforms(results)` 非空才 merge），≈ 领券券数量级、非每 step;复用 `record_claims` 同一 `SessionLocal`、不新增连接。
-- merge **读不到 session 行则跳过**（不建兜底行）;并集幂等、无新平台不写。
-- 口径:成功=`status∈{success,already_claimed}`;基数含 `abandoned`/`failed`;③ 分母=`platforms`（空→全领三档）;`coupon_id`→平台走前缀，与客户端 `couponIdToPlatform` 同词表。
-
----
-
-## 12. 续做:admin 前端看板卡 + 分平台点位成功率（设计 · 2026-07-08）
-
-承 §11 待办 #3（前端图表)与 §7「可选 `per_platform`」。本轮把 ②③ 接入 admin「领券数据」页,并把 ③ 按平台拆(`per_platform`)。**改前端 + 后端 + 测试**;客户端 / pricebot 仍零改动。
-
-- 状态:设计已评审通过(2026-07-08),待实现。
-- 涉及仓库:`shaguabijia-app-server`(后端)+ `shaguabijia-admin-web`(admin 前端,Next.js + antd)。
-
-### 12.1 后端:`per_platform` 分平台点位成功率
-
-`app/admin/repositories/coupon_data.py::_success_rates(rows)` 在现有合计基础上,对每个 `p ∈ DEFAULT_PLATFORMS`(美团/淘宝/京东)累加:
-
-- 分母 `per_total[p]` = 勾选了 p 的 session 数(`p ∈ sel`,空勾选 `sel` 按全领三档);
-- 分子 `per_succ[p]` = 其中 `p ∈ succ`(该平台至少领到一张)的 session 数;
-- `per_platform[p]` = `round(per_succ[p] / per_total[p], 4)`;分母 0 → `None`。
-
-**不变量**:`Σ_p per_succ[p] == point_success_count`、`Σ_p per_total[p] == point_total_count`(三档词表下恒成立;非三档平台 id 不计入 `per_platform`,由 `if p in per_total` 守卫)。用 §9 / `test_coupon_data_success_rates` 数据自检:美团 3/4=0.75、淘宝 2/3=0.6667、京东 1/2=0.5;合计仍 6/9=0.6667。
-
-- summary 加一项 `per_platform`:**恒含三档键**,如 `{"meituan-waimai":0.75,"taobao-shanguang":0.6667,"jd-waimai":0.5}`(区间内无人勾选的平台 → 值 `None`)。
-- schema `app/admin/schemas/coupon_data.py::CouponDataSummary` 加 `per_platform: dict[str, float | None]`。
-
-### 12.2 前端:admin-web「领券数据」汇总卡补一段
-
-`shaguabijia-admin-web/src/app/(main)/coupon-data/page.tsx`(单文件,`CouponDataSummary` 为该页内联类型):
-
-- 内联 `CouponDataSummary` 补:`full_success_count` / `full_success_rate` / `point_success_count` / `point_total_count` / `point_success_rate` + `per_platform: Record<string, number | null>`。
-- 新 helper `fmtPct(v) = v == null ? '-' : ${(v*100).toFixed(1)}%`(沿用本页 `-` 空值风格,数学同大盘 `pct`);antd 导入补 `Tooltip`,新增 `import { InfoCircleOutlined } from '@ant-design/icons'`。
-- 汇总卡「耗时分位」行之后,`Divider` + 两行 `Statistic`(各 `Col flex="1 1 0"`):
-  - 行1:**整单成功率** `fmtPct(full_success_rate)` · **点位成功率(合计)** `fmtPct(point_success_rate)`;标题各带 ⓘ `Tooltip`(口径说明;合计率注明「平台粒度、= 三档之和,与『数据大盘』券粒度口径不同」)。
-  - 行2:**美团 / 淘宝 / 京东 点位成功率** `fmtPct(per_platform['meituan-waimai' | 'taobao-shanguang' | 'jd-waimai'])`(平台名同「美团耗时」列既有叫法)。
-- 不显示支撑数(合计与分平台均纯百分比);tooltip 不带分母。
-
-### 12.3 测试
-
-- 扩 `tests/test_coupon_platform_success.py::test_coupon_data_success_rates`:断言 `per_platform == {"meituan-waimai":0.75,"taobao-shanguang":round(2/3,4),"jd-waimai":0.5}`,并断言和不变量(`Σ 分子 == point_success_count == 6`、`Σ 分母 == point_total_count == 9`)。
-- 前端 `shaguabijia-admin-web` lint / type-check 通过。
-- 端到端:随下一步「真实 /step 实测」在跑起来的 admin-web + 后端页面上核对卡片渲染。
-
-### 12.4 不做(YAGNI)
-
-- 成功率**趋势线**:后端 `daily` / `hourly` 不含率字段,加趋势要另改聚合,超出本轮范围。
-- 合计 / 分平台的**支撑数副文本**、tooltip 带分母。
-- 客户端 / pricebot 改动;历史回填。
-
----
-
-## 13. 续做:每券(coupon_id)成功率明细表(设计 · 2026-07-08)
-
-产品要更细粒度:到**具体领券点位**(如「美团外卖红包天天领」「美团甄选好店」),即按 `coupon_id` 算成功/失败率,比 §12 的平台粒度再细一层。**改前端 + 后端 + 测试**;客户端 / pricebot 仍零改动。
-
-- 状态:设计已评审通过(2026-07-08),待实现。
-- 关键取舍(已定):数据源用 **`coupon_claim_record`**(它本就是「单张券一天一条」的系统记录,已带 `coupon_name` / `status` / `vendor` / `trace_id`),唯一缺 `app_env` → 补一列即可。§2 当初绕开它是因为**平台指标要按 session 归因**;而**每券成功率是全局聚合、不需要 session 归因**,`(device,券,天)` 折叠反而天然去重防刷,故这里用它是对的。
-
-### 13.1 指标口径(已定)
-
-- **成功** = `status ∈ {success, already_claimed}`;**尝试(分母)** = `status ∈ {success, already_claimed, failed}`;**`skipped` 排除**(无券可领/不适用,不算尝试、不进分母、不展示)。
-- **成功率** = 成功 / 尝试(某券区间内无 tried 行 → 不出现在表里,无除零)。
-- **粒度 = 「设备-天」**(非「每次点击」):`coupon_claim_record` 唯一键 `(device, coupon_id, claim_date)`,同设备当天同券只留最后状态。所以「尝试」= 有多少**设备-天**尝试过该券,「成功」= 其中最终领到的。要每次点击级须换源(route B / 原始事件),本轮不做。
-- **环境**:跟随页面「环境」(prod/dev/全部),按 `app_env` 过滤;旧行 `app_env=NULL` **不回填** → 仅「全部」视图可见。
-- **日期**:按 `claim_date`(Asia/Shanghai 自然日),与页面日期范围一致。
-
-### 13.2 数据模型
-
-`CouponClaimRecord`(`app/models/coupon_state.py`)加一列:
-
-| 列 | 类型 | 说明 |
-|----|------|------|
-| `app_env` | `String(16)`,index,nullable | prod / dev。`/step` 落库时按 session 的 app_env 打标;旧行 NULL(不回填)。 |
-
-- alembic 新迁移:`add_column coupon_claim_record.app_env`,nullable + index,无 server_default。
-- 同步更新 `docs/database/coupon_state.md` 的 coupon_claim_record 节。
-
-### 13.3 写路径(`/step`)
-
-- `app/repositories/coupon_state.py::record_claims` 加参数 `app_env: str | None = None`;INSERT 时写入,UPDATE 时 `if app_env is not None: row.app_env = app_env`(不用 None 覆盖已有)。
-- 新增轻量 repo 助手 `session_app_env(db, trace_id) -> str | None`(按 trace_id 取 `coupon_session.app_env`,查不到返回 None)。
-- `app/api/v1/coupon.py::_record_claims_blocking`:`trace_id` 存在则先 `app_env = coupon_repo.session_app_env(db, trace_id)`,传给 `record_claims(..., app_env=app_env)`;`merge_session_platform_success` 保持原样(其自身 select 不变)。查不到 session / 无 trace_id → `app_env=None`(行为同旧)。
-  > 写放大:仅「带券结果的帧」触发(同 §5),每帧多一次 `session_app_env` 小查询(按 trace_id 唯一索引),可忽略。
-
-### 13.4 admin 聚合(新 repo 函数)
-
-`app/admin/repositories/coupon_data.py` 新增 `coupon_slot_report(db, *, date_from, date_to, app_env)`(与 `coupon_user_records` 同居本文件,同属「领券数据」看板;更新模块 docstring 注明本文件现读 `coupon_session` + `coupon_claim_record` 两源):
-
-- `SELECT coupon_id, MAX(coupon_name) AS coupon_name, COUNT(*) AS tried, SUM(CASE WHEN status IN (success,already_claimed) THEN 1 ELSE 0 END) AS succeeded`
-  `WHERE claim_date ∈ [from,to] AND status IN (success,already_claimed,failed)` (+ `AND app_env = :env` 当 env 非「全部」) `GROUP BY coupon_id`。
-- 每行:`platform = coupon_id_to_platform(coupon_id)`(复用前缀映射,无法识别→None),`success_rate = round(succeeded/tried, 4)`。
-- 按 `tried` 倒序、`coupon_id` 次序返回 `{"items": [...]}`。
-
-### 13.5 接口 + schema(新子端点)
-
-- 路由 `app/admin/routers/coupon_data.py`:加 `@router.get("/coupons")` → `get_coupon_slots`,参数 `date_from` / `date_to` / `app_env`(同主 report 的解析与 `_MAX_RANGE_DAYS` 校验,`app_env="all"→None`),调 `coupon_slot_report`。路径全称 `/admin/api/coupon-data/coupons`(与现有 `/coupon-data/user-records` 同款子端点)。
-- schema `app/admin/schemas/coupon_data.py`:
-  - `CouponSlotRow{coupon_id: str, coupon_name: str|None, platform: str|None, tried: int, succeeded: int, success_rate: float|None}`
-  - `CouponSlotsOut{date_from: str, date_to: str, items: list[CouponSlotRow]}`
-
-### 13.6 前端
-
-`shaguabijia-admin-web/src/app/(main)/coupon-data/page.tsx`:
-
-- 汇总卡下方(或趋势图下方)新增一张「按券成功率」表:列 **券名**(`coupon_name || coupon_id`)/ **平台**(美团/淘宝/京东/其他)/ **尝试** / **成功** / **成功率**(`fmtPct`)。默认按尝试倒序;支持 antd 列排序。
-- 点「查询」时,除主 report 外并行 `api.get('/admin/api/coupon-data/coupons', {params:{date_from,date_to,app_env}})`;空则不显示表。
-- 新增内联类型 `CouponSlotRow`;平台名复用映射(`meituan-waimai→美团` 等,null→其他)。
-
-### 13.7 测试
-
-- 后端:`record_claims` stamp `app_env`(INSERT/UPDATE 两路);`session_app_env` 助手;`/step` 集成写入 `app_env`;`coupon_slot_report` 聚合(多券 × success/already_claimed/failed/skipped × 两 env:断言 tried/succeeded/rate、**skipped 排除**、env 过滤、排序);schema 契约。
-- 前端:`tsc --noEmit`。
-
-### 13.8 不做(YAGNI)
-
-- 每次点击级成功率(需换数据源);成功率趋势线;`app_env` 历史回填(可选一次性 `join trace_id→session.app_env`,默认不做);券级 tooltip / 明细下钻。
-- 券表为**全局聚合**,**不随页面「用户」搜索框过滤**(用户维度非本需求;`coupon_claim_record` 虽有 `user_id`,YAGNI)。与主明细表按用户过滤的行为不同,属有意为之。

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

@@ -0,0 +1,740 @@
+# 本地开发切 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

@@ -0,0 +1,281 @@
+# 本地开发切 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,7 +30,14 @@ if not exist .env (
 
 if not exist data mkdir data
 
-REM Build/upgrade SQLite schema (idempotent; no-op if already at head)
+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

run.sh

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

run8771.bat

@@ -0,0 +1,50 @@
+@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

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

scripts/seed_coinhistory_labels_test.py

@@ -1,166 +0,0 @@
-"""收益明细「金币记录」文案验证脚手架(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,22 +1,19 @@
 """测试用 fixtures。
 
-测试 DB 用临时文件 SQLite。
-  - 不用 in-memory:in-memory 默认 per-connection,跨连接看不到表。
-  - 用临时文件保证 SessionLocal 每次新连都看到同一份 schema。
-
-顺序:set env(必须在 import app.* 之前) → import app → 建表 → TestClient
+测试库用 Docker PostgreSQL 的 shaguabijia_test(与 dev 业务库 shaguabijia 隔离)。
+顺序(必须):设 test DATABASE_URL(在 import app.* 之前)→ ensure PG 就绪 + 测试库存在 →
+import app → 建表。持久卷可能残留上次的表 → session 开头先 drop 再 create。
 """
 from __future__ import annotations
 
 import os
-import tempfile
 from collections.abc import Iterator
 
-# 临时 db 文件路径,进程退出后清理
-_tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
-_tmp_db.close()
-
-os.environ["DATABASE_URL"] = f"sqlite:///{_tmp_db.name}"
+# 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")
@@ -35,6 +32,18 @@ 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
 
@@ -45,13 +54,11 @@ 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_admin_read.py

@@ -320,92 +320,3 @@ def test_read_apis_require_auth(admin_client: TestClient) -> None:
         "/admin/api/feedbacks",
     ]:
         assert admin_client.get(path).status_code == 401, path
-
-
-def test_period_comparison_reward_coin_from_feed_scene(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """比价奖励金币口径:按 ad_feed_reward_record.feed_scene='comparison' 的实发金币
-    (status=granted)汇总,而非查从不写入的 biz_type 桶(修复大盘该卡恒 0)。
-    too_short(未发奖)不计。"""
-    from app.models.ad_feed_reward import AdFeedRewardRecord
-
-    d = "2021-06-15"  # 独立历史日,隔离其它用例数据
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(db, phone="13800008801", register_channel="sms").id
-        db.add(AdFeedRewardRecord(
-            client_event_id="cmp-fs-granted", user_id=uid, reward_date=d,
-            ecpm_raw="0", coin=123, feed_scene="comparison", status="granted",
-        ))
-        db.add(AdFeedRewardRecord(
-            client_event_id="cmp-fs-tooshort", user_id=uid, reward_date=d,
-            ecpm_raw="0", coin=99, feed_scene="comparison", status="too_short",
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    r = admin_client.get(
-        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    assert r.json()["period"]["coins"]["comparison_reward_coin_total"] == 123
-
-
-def test_period_coupon_reward_coin_from_feed_scene(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """领券奖励金币口径:按 ad_feed_reward_record.feed_scene='coupon' 的实发金币汇总。"""
-    from app.models.ad_feed_reward import AdFeedRewardRecord
-
-    d = "2021-06-16"
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(db, phone="13800008802", register_channel="sms").id
-        db.add(AdFeedRewardRecord(
-            client_event_id="cpn-fs-granted", user_id=uid, reward_date=d,
-            ecpm_raw="0", coin=456, feed_scene="coupon", status="granted",
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    r = admin_client.get(
-        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    assert r.json()["period"]["coins"]["coupon_reward_coin_total"] == 456
-
-
-def test_period_coupon_reward_excludes_reward_video(
-    admin_client: TestClient, admin_token: str
-) -> None:
-    """领券奖励金币不再把激励视频金币算进来(reward_video/ad_reward 从领券桶拆出);
-    激励视频仍单独计入 reward_video_coin_total。"""
-    from datetime import datetime
-
-    from app.models.wallet import CoinTransaction
-
-    d = "2021-06-17"
-    db = SessionLocal()
-    try:
-        uid = user_repo.upsert_user_for_login(db, phone="13800008803", register_channel="sms").id
-        db.add(CoinTransaction(
-            user_id=uid, amount=50, balance_after=50, biz_type="reward_video",
-            ref_id="rv-split-1", created_at=datetime(2021, 6, 17, 12, 0, 0),
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    r = admin_client.get(
-        "/admin/api/stats/overview", params={"date_from": d, "date_to": d},
-        headers=_auth(admin_token),
-    )
-    assert r.status_code == 200, r.text
-    coins = r.json()["period"]["coins"]
-    assert coins["coupon_reward_coin_total"] == 0        # 激励视频不计入领券奖励
-    assert coins["reward_video_coin_total"] == 50        # 仍计入激励视频卡

tests/test_admin_roles.py

@@ -150,80 +150,3 @@ def test_create_admin_rejects_unknown_role(admin_client, super_token) -> None:
         headers=_auth(super_token),
     )
     assert r.status_code == 400
-
-
-def _login_pages(username: str, password: str = "pass1234") -> list[str]:
-    """以某账号登录,取下发的有效可见页(左侧导航过滤依据)。"""
-    c = TestClient(admin_app)
-    return c.post(
-        "/admin/api/auth/login", json={"username": username, "password": password}
-    ).json()["admin"]["pages"]
-
-
-def test_custom_role_create_pages_take_effect(admin_client, super_token) -> None:
-    # 建「自定义」账号:role=custom + 勾选页(含一个非法 key,应被过滤)
-    r = admin_client.post(
-        "/admin/api/admins",
-        json={
-            "username": "cust_user", "password": "pass1234", "role": "custom",
-            "pages_override": ["dashboard", "withdraws", "xx-bad"],
-        },
-        headers=_auth(super_token),
-    )
-    assert r.status_code == 200, r.text
-    # 登录下发的 pages == 勾选集(非法 key 过滤),真正驱动左侧导航
-    assert set(_login_pages("cust_user")) == {"dashboard", "withdraws"}
-    # 账号列表回显原始勾选集(供编辑回填),同样已过滤
-    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
-    assert set(admins["cust_user"]["pages_override"]) == {"dashboard", "withdraws"}
-    assert admins["cust_user"]["role"] == "custom"
-
-
-def test_custom_role_update_and_switch_back_clears_override(admin_client, super_token) -> None:
-    admin_client.post(
-        "/admin/api/admins",
-        json={
-            "username": "cust_sw", "password": "pass1234", "role": "custom",
-            "pages_override": ["dashboard"],
-        },
-        headers=_auth(super_token),
-    )
-    aid = next(
-        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
-        if a["username"] == "cust_sw"
-    )
-    # 只改勾选集(角色不变)→ 生效
-    u = admin_client.patch(
-        f"/admin/api/admins/{aid}", json={"pages_override": ["dashboard", "feedbacks"]},
-        headers=_auth(super_token),
-    )
-    assert u.status_code == 200, u.text
-    assert set(_login_pages("cust_sw")) == {"dashboard", "feedbacks"}
-    # 切回普通角色 operator → override 清空,pages 跟随角色(运营页集,且不含 admins)
-    u2 = admin_client.patch(
-        f"/admin/api/admins/{aid}", json={"role": "operator"}, headers=_auth(super_token)
-    )
-    assert u2.status_code == 200, u2.text
-    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
-    assert admins["cust_sw"]["pages_override"] is None
-    pages = set(_login_pages("cust_sw"))
-    assert "feedbacks" in pages and "admins" not in pages  # 运营口径,自定义页已不生效
-
-
-def test_switch_operator_to_custom_sets_override(admin_client, super_token) -> None:
-    admin_client.post(
-        "/admin/api/admins",
-        json={"username": "op2cust", "password": "pass1234", "role": "operator"},
-        headers=_auth(super_token),
-    )
-    aid = next(
-        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
-        if a["username"] == "op2cust"
-    )
-    u = admin_client.patch(
-        f"/admin/api/admins/{aid}",
-        json={"role": "custom", "pages_override": ["config"]},
-        headers=_auth(super_token),
-    )
-    assert u.status_code == 200, u.text
-    assert set(_login_pages("op2cust")) == {"config"}

tests/test_compare_harvest.py

@@ -18,6 +18,7 @@ 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
 
@@ -47,6 +48,17 @@ 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 层
 # ============================================================
@@ -144,17 +156,18 @@ 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=[],
         )
-        # 用一个不会与顺序自增用户撞的合成 id(SQLite 测试库 FK 不强制;别用小整数,
-        # 否则会撞上别的测试 login 出来的真实 user_id → 记录混进那个用户的列表)。
-        rec = crud.upsert_record(db, user_id=987654, payload=payload)
+        rec = crud.upsert_record(db, user_id=uid, payload=payload)
         assert rec.status == "success"              # 不降级
-        assert rec.user_id == 987654                 # 但补上了 user_id(原为 None)
+        assert rec.user_id == uid                    # 但补上了 user_id(原为 None)
 
 
 # ============================================================

tests/test_coupon_platform_success.py

@@ -1,299 +0,0 @@
-"""领券「平台成功率」埋点:coupon_id→平台映射 / 成功平台推导 / session platform_success 并集。
-
-设计:docs/guides/领券成功率指标-设计与埋点.md
-"""
-from __future__ import annotations
-
-from datetime import UTC, date, datetime
-
-from sqlalchemy import delete, func, select
-
-from app.admin.repositories.coupon_data import coupon_data_report
-from app.db.session import SessionLocal
-from app.models.coupon_state import CouponSession
-from app.repositories.coupon_state import (
-    coupon_id_to_platform,
-    merge_session_platform_success,
-    succeeded_platforms,
-)
-
-
-def _agg_session(
-    trace: str, platforms, platform_success, *, status: str = "completed"
-) -> CouponSession:
-    """构造一条聚合测试用 session(started_date 固定 2020-01-02、app_env=prod,不 commit)。"""
-    return CouponSession(
-        trace_id=trace,
-        device_id="d-agg",
-        status=status,
-        app_env="prod",
-        platforms=platforms,
-        platform_success=platform_success,
-        started_at=datetime(2020, 1, 2, tzinfo=UTC),
-        started_date=date(2020, 1, 2),
-    )
-
-
-def _make_session(db, trace_id: str, **kw) -> CouponSession:
-    row = CouponSession(
-        trace_id=trace_id,
-        device_id="dev-merge",
-        status=kw.pop("status", "started"),
-        started_at=datetime(2020, 1, 1, tzinfo=UTC),
-        started_date=date(2020, 1, 1),
-        **kw,
-    )
-    db.add(row)
-    db.commit()
-    return row
-
-
-def test_coupon_id_to_platform_prefix_mapping() -> None:
-    """coupon_id 前缀 → 三档平台 id(与客户端 couponIdToPlatform 同词表)。"""
-    assert coupon_id_to_platform("mt_banjia_zhoumo") == "meituan-waimai"
-    assert coupon_id_to_platform("mt_cps_waimai_redpacket") == "meituan-waimai"  # 大众点评CPS 也挂 mt_ → 归美团
-    assert coupon_id_to_platform("tb_vip_shangou_voucher") == "taobao-shanguang"
-    assert coupon_id_to_platform("ele_hongbao") == "taobao-shanguang"
-    assert coupon_id_to_platform("elm_hongbao") == "taobao-shanguang"
-    assert coupon_id_to_platform("jd_redpacket") == "jd-waimai"
-
-
-def test_coupon_id_to_platform_unknown_returns_none() -> None:
-    """无法识别的前缀 / 空 → None(调用方跳过,不计入平台)。"""
-    assert coupon_id_to_platform("weird_xxx") is None
-    assert coupon_id_to_platform("") is None
-    assert coupon_id_to_platform(None) is None  # type: ignore[arg-type]
-
-
-def test_succeeded_platforms_filters_status_and_dedups() -> None:
-    """只取 status∈{success,already_claimed} 的券,映射平台后去重;失败/跳过/无法识别的不计。"""
-    results = [
-        {"coupon_id": "mt_a", "status": "success"},
-        {"coupon_id": "mt_b", "status": "already_claimed"},  # 也算成功 → 仍是美团
-        {"coupon_id": "mt_c", "status": "failed"},           # 不算
-        {"coupon_id": "tb_a", "status": "success"},
-        {"coupon_id": "jd_a", "status": "skipped"},          # 不算
-        {"coupon_id": "weird", "status": "success"},         # 无法识别平台 → 跳过
-    ]
-    assert set(succeeded_platforms(results)) == {"meituan-waimai", "taobao-shanguang"}
-
-
-def test_succeeded_platforms_empty() -> None:
-    assert succeeded_platforms([]) == []
-
-
-def test_merge_platform_success_union_idempotent() -> None:
-    """按 trace_id 把成功平台并入 platform_success:并集去重 + 按 DEFAULT_PLATFORMS 保序 + 重复并入幂等。"""
-    db = SessionLocal()
-    trace = "merge-union-1"
-    try:
-        _make_session(db, trace)
-        merge_session_platform_success(db, trace, ["meituan-waimai"])
-        merge_session_platform_success(db, trace, ["jd-waimai", "meituan-waimai"])  # 并集 + 已有幂等
-        db.expire_all()
-        row = db.execute(
-            select(CouponSession).where(CouponSession.trace_id == trace)
-        ).scalar_one()
-        # 美团→淘宝→京东固定序;只含实际成功的美团/京东
-        assert row.platform_success == ["meituan-waimai", "jd-waimai"]
-    finally:
-        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
-        db.commit()
-        db.close()
-
-
-def test_merge_platform_success_missing_row_skips() -> None:
-    """trace_id 无对应行 → 静默跳过:不建兜底行、不抛异常(设计 §5)。"""
-    db = SessionLocal()
-    try:
-        merge_session_platform_success(db, "no-such-trace", ["meituan-waimai"])
-        n = db.execute(
-            select(func.count()).select_from(CouponSession).where(
-                CouponSession.trace_id == "no-such-trace"
-            )
-        ).scalar_one()
-        assert n == 0
-    finally:
-        db.close()
-
-
-def test_coupon_data_success_rates() -> None:
-    """admin 聚合 ②整单成功率 / ③点位成功率:平台粒度,含空 platforms→全领三档、abandoned 入基数。
-
-    A 勾美团+淘宝、两个都成 → 整单成功;点位 2/2
-    B 勾三档、只成美团       → 非整单;点位 1/3
-    C 全领(platforms空→3)、三档全成 → 整单成功;点位 3/3
-    D abandoned、勾美团、无成功平台 → 非整单;点位 0/1(仍进基数)
-    发起数=4;整单成功=2 → 0.5;点位 6/9 → 0.6667
-    """
-    db = SessionLocal()
-    try:
-        db.add_all([
-            _agg_session("agg-A", ["meituan-waimai", "taobao-shanguang"],
-                         ["meituan-waimai", "taobao-shanguang"]),
-            _agg_session("agg-B", ["meituan-waimai", "taobao-shanguang", "jd-waimai"],
-                         ["meituan-waimai"]),
-            _agg_session("agg-C", [], ["meituan-waimai", "taobao-shanguang", "jd-waimai"]),
-            _agg_session("agg-D", ["meituan-waimai"], None, status="abandoned"),
-        ])
-        db.flush()  # 同会话可见,不 commit(finally 回滚保持隔离)
-        s = coupon_data_report(
-            db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod"
-        )["summary"]
-        assert s["started_count"] == 4
-        assert s["full_success_count"] == 2
-        assert s["full_success_rate"] == 0.5
-        assert s["point_success_count"] == 6
-        assert s["point_total_count"] == 9
-        assert s["point_success_rate"] == round(6 / 9, 4)
-        # ③ 分平台点位成功率(§12):美团 3/4、淘宝 2/3、京东 1/2。
-        # 分平台(成功,总)= 美团(3,4)+淘宝(2,3)+京东(1,2) = (6,9),
-        # 其和正好等于上面已断言的 point_success_count=6 / point_total_count=9(和不变量)。
-        assert s["per_platform"] == {
-            "meituan-waimai": 0.75,
-            "taobao-shanguang": round(2 / 3, 4),
-            "jd-waimai": 0.5,
-        }
-    finally:
-        db.rollback()
-        db.close()
-
-
-def test_coupon_data_success_rates_empty_range() -> None:
-    """区间无 session → 发起数 0、两个率为 None(不除零)。"""
-    db = SessionLocal()
-    try:
-        s = coupon_data_report(
-            db, date_from="2019-01-01", date_to="2019-01-01", app_env="prod"
-        )["summary"]
-        assert s["started_count"] == 0
-        assert s["full_success_rate"] is None
-        assert s["point_success_rate"] is None
-    finally:
-        db.close()
-
-
-def test_coupon_data_summary_schema_exposes_rates() -> None:
-    """schema 契约:CouponDataSummary 暴露 ②③ 字段,键名与 repo 输出一致(router 直接 **summary 构造)。"""
-    from app.admin.schemas.coupon_data import CouponDataSummary
-
-    db = SessionLocal()
-    try:
-        db.add(_agg_session("agg-schema-1", ["meituan-waimai"], ["meituan-waimai"]))
-        db.flush()
-        summary = coupon_data_report(
-            db, date_from="2020-01-02", date_to="2020-01-02", app_env="prod"
-        )["summary"]
-        dumped = CouponDataSummary(**summary).model_dump()
-        assert dumped["full_success_rate"] == summary["full_success_rate"]
-        assert dumped["point_success_rate"] == summary["point_success_rate"]
-        assert dumped["full_success_count"] == summary["full_success_count"]
-        assert dumped["point_success_count"] == summary["point_success_count"]
-        assert dumped["point_total_count"] == summary["point_total_count"]
-        assert dumped["per_platform"] == summary["per_platform"]
-    finally:
-        db.rollback()
-        db.close()
-
-
-def test_step_writes_platform_success_to_session(client) -> None:
-    """/step 集成:pricebot 返 coupon_results → 本 trace 的 coupon_session.platform_success 落库(仅成功平台)。"""
-    from unittest.mock import MagicMock, patch
-
-    import httpx
-
-    from app.models.coupon_state import CouponClaimRecord, CouponDailyCompletion
-
-    trace = "int-trace-1"
-    device = "dev-int-1"
-
-    # 预置一条 started session(模拟 /session started 已先落库)
-    db = SessionLocal()
-    try:
-        db.add(CouponSession(
-            trace_id=trace, device_id=device, status="started", app_env="dev",
-            platforms=["meituan-waimai", "taobao-shanguang"],
-            started_at=datetime(2020, 1, 5, tzinfo=UTC),
-            started_date=date(2020, 1, 5),
-        ))
-        db.commit()
-    finally:
-        db.close()
-
-    fake_resp = {
-        "success": True,
-        "action": {"command": "done", "params": {
-            "information": "已领 1 张",
-            "coupon_results": [
-                {"coupon_id": "mt_x", "name": "美团券", "vendor": "meituan_internal", "status": "success"},
-                {"coupon_id": "tb_y", "name": "淘宝券", "vendor": "taobao", "status": "failed"},
-            ],
-        }},
-        "continue": False,
-    }
-
-    async def fake_post(self, url, **kw):
-        m = MagicMock()
-        m.status_code = 200
-        m.json = lambda: fake_resp
-        return m
-
-    body = {
-        "device_id": device, "trace_id": trace, "step": 5,
-        "screen_state": {"screen": {"width": 1080, "height": 2340, "density": 3.0},
-                         "foreground": {"package": "x", "activity": ""}, "windows": []},
-    }
-
-    try:
-        with patch.object(httpx.AsyncClient, "post", fake_post):
-            r = client.post("/api/v1/coupon/step", json=body)
-        assert r.status_code == 200, r.text
-
-        db = SessionLocal()
-        try:
-            row = db.execute(
-                select(CouponSession).where(CouponSession.trace_id == trace)
-            ).scalar_one()
-            assert row.platform_success == ["meituan-waimai"]  # tb 失败不计入
-            claims = db.execute(
-                select(CouponClaimRecord).where(CouponClaimRecord.device_id == device)
-            ).scalars().all()
-            assert claims and all(c.app_env == "dev" for c in claims)  # session app_env 打标
-        finally:
-            db.close()
-    finally:
-        db = SessionLocal()
-        try:
-            db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
-            db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == device))
-            db.execute(delete(CouponDailyCompletion).where(CouponDailyCompletion.device_id == device))
-            db.commit()
-        finally:
-            db.close()
-
-
-def test_coupon_data_status_filter() -> None:
-    """状态多选过滤(方案 A):整个视图按选中状态算;None/空=全部。"""
-    db = SessionLocal()
-    try:
-        db.add_all([
-            _agg_session("st-A", ["meituan-waimai"], ["meituan-waimai"], status="completed"),
-            _agg_session("st-B", ["meituan-waimai"], ["meituan-waimai"], status="started"),
-            _agg_session("st-C", ["meituan-waimai"], None, status="failed"),
-            _agg_session("st-D", ["meituan-waimai"], ["meituan-waimai"], status="abandoned"),
-        ])
-        db.flush()
-        base = dict(date_from="2020-01-02", date_to="2020-01-02", app_env="prod")
-        # None = 全部 4 发起
-        assert coupon_data_report(db, **base)["summary"]["started_count"] == 4
-        # 排除 started → 3 发起(整个视图,发起数也随之变)
-        sub = coupon_data_report(db, **base, statuses=["completed", "failed", "abandoned"])["summary"]
-        assert sub["started_count"] == 3
-        assert sub["completed_count"] == 1
-        # 只 completed → 发起数=1、整单成功率基数=1(该 completed 是整单成功)
-        comp = coupon_data_report(db, **base, statuses=["completed"])["summary"]
-        assert comp["started_count"] == 1
-        assert comp["full_success_rate"] == 1.0
-    finally:
-        db.rollback()
-        db.close()

tests/test_coupon_slots.py

@@ -1,137 +0,0 @@
-"""每券成功率(§13):app_env 落库 + coupon_slot_report 聚合。"""
-from __future__ import annotations
-
-from datetime import UTC, date, datetime
-
-from sqlalchemy import delete, select
-
-from app.admin.repositories.coupon_data import coupon_slot_report
-from app.db.session import SessionLocal
-from app.models.coupon_state import CouponClaimRecord, CouponSession
-from app.repositories.coupon_state import record_claims, session_app_env
-
-
-def test_session_app_env_lookup() -> None:
-    db = SessionLocal()
-    trace = "slot-env-1"
-    try:
-        db.add(CouponSession(
-            trace_id=trace, device_id="d-slot", status="started", app_env="dev",
-            started_at=datetime(2020, 2, 1, tzinfo=UTC), started_date=date(2020, 2, 1),
-        ))
-        db.commit()
-        assert session_app_env(db, trace) == "dev"
-        assert session_app_env(db, "no-such-trace") is None
-        assert session_app_env(db, None) is None
-    finally:
-        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
-        db.commit()
-        db.close()
-
-
-def test_record_claims_stamps_app_env() -> None:
-    db = SessionLocal()
-    dev = "d-slot-stamp"
-    try:
-        record_claims(db, dev, None, "t-stamp",
-                      [{"coupon_id": "mt_x", "status": "success", "name": "美团券"}],
-                      app_env="prod")
-        row = db.execute(select(CouponClaimRecord).where(
-            CouponClaimRecord.device_id == dev)).scalar_one()
-        assert row.app_env == "prod"
-        # UPDATE 路:app_env=None 不覆盖已有值
-        record_claims(db, dev, None, "t-stamp",
-                      [{"coupon_id": "mt_x", "status": "already_claimed", "name": "美团券"}],
-                      app_env=None)
-        db.expire_all()
-        row = db.execute(select(CouponClaimRecord).where(
-            CouponClaimRecord.device_id == dev)).scalar_one()
-        assert row.app_env == "prod"
-        assert row.status == "already_claimed"
-    finally:
-        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == dev))
-        db.commit()
-        db.close()
-
-
-def _claim(db, device, coupon_id, status, app_env, name=None, d=date(2020, 2, 2)):
-    db.add(CouponClaimRecord(
-        device_id=device, coupon_id=coupon_id, claim_date=d,
-        status=status, coupon_name=name, app_env=app_env,
-    ))
-
-
-def test_coupon_slot_report_aggregates() -> None:
-    """按 coupon_id 聚合:tried=成功+失败(skipped 排除)、rate、env 过滤、tried 倒序。"""
-    db = SessionLocal()
-    try:
-        # mt_a(prod):dev1 success + dev2 already_claimed + dev3 failed → 2/3;dev4 skipped 不计
-        _claim(db, "sdev1", "mt_a", "success", "prod", "美团外卖红包天天领")
-        _claim(db, "sdev2", "mt_a", "already_claimed", "prod", "美团外卖红包天天领")
-        _claim(db, "sdev3", "mt_a", "failed", "prod", "美团外卖红包天天领")
-        _claim(db, "sdev4", "mt_a", "skipped", "prod", "美团外卖红包天天领")
-        _claim(db, "sdev1", "tb_b", "success", "prod", "淘宝券")          # 1/1
-        _claim(db, "sdev9", "mt_a", "failed", "dev", "美团外卖红包天天领")  # 仅 dev/全部
-        db.commit()
-
-        prod = coupon_slot_report(
-            db, date_from="2020-02-02", date_to="2020-02-02", app_env="prod"
-        )["items"]
-        by_id = {r["coupon_id"]: r for r in prod}
-        assert by_id["mt_a"]["tried"] == 3            # skipped 排除
-        assert by_id["mt_a"]["succeeded"] == 2
-        assert by_id["mt_a"]["success_rate"] == round(2 / 3, 4)
-        assert by_id["mt_a"]["platform"] == "meituan-waimai"
-        assert by_id["mt_a"]["coupon_name"] == "美团外卖红包天天领"
-        assert by_id["tb_b"]["success_rate"] == 1.0
-        assert by_id["tb_b"]["platform"] == "taobao-shanguang"
-        assert [r["coupon_id"] for r in prod] == ["mt_a", "tb_b"]  # tried 倒序
-
-        dev = coupon_slot_report(
-            db, date_from="2020-02-02", date_to="2020-02-02", app_env="dev"
-        )["items"]
-        assert {r["coupon_id"]: r["tried"] for r in dev} == {"mt_a": 1}
-        allenv = coupon_slot_report(
-            db, date_from="2020-02-02", date_to="2020-02-02", app_env=None
-        )["items"]
-        assert {r["coupon_id"]: r["tried"] for r in allenv}["mt_a"] == 4
-    finally:
-        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.claim_date == date(2020, 2, 2)))
-        db.commit()
-        db.close()
-
-
-def test_coupon_slots_endpoint() -> None:
-    from fastapi.testclient import TestClient
-
-    from app.admin.main import admin_app
-    from app.admin.repositories import admin_user as admin_repo
-    from app.admin.security import create_admin_token
-
-    db = SessionLocal()
-    try:
-        _claim(db, "epdev1", "mt_ep", "success", "prod", "端点券", d=date(2020, 2, 3))
-        _claim(db, "epdev2", "mt_ep", "failed", "prod", "端点券", d=date(2020, 2, 3))
-        admin = admin_repo.create_admin(
-            db, username="slot_admin", password="pass1234", role="super_admin"
-        )
-        token, _exp = create_admin_token(admin_id=admin.id, role=admin.role)
-        db.commit()
-    finally:
-        db.close()
-    try:
-        c = TestClient(admin_app)
-        r = c.get(
-            "/admin/api/coupon-data/coupons",
-            params={"date_from": "2020-02-03", "date_to": "2020-02-03", "app_env": "prod"},
-            headers={"Authorization": f"Bearer {token}"},
-        )
-        assert r.status_code == 200, r.text
-        row = next(x for x in r.json()["items"] if x["coupon_id"] == "mt_ep")
-        assert row["tried"] == 2 and row["succeeded"] == 1 and row["success_rate"] == 0.5
-        assert row["coupon_name"] == "端点券" and row["platform"] == "meituan-waimai"
-    finally:
-        db = SessionLocal()
-        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.claim_date == date(2020, 2, 3)))
-        db.commit()
-        db.close()

tests/test_ensure_pg.py

@@ -0,0 +1,87 @@
+"""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