重构比价结果页取数据逻辑 (#195 )

Reviewed-on: #195
修复：统一用户Draw信息流eCPM统计口径 (#190 )
2026-07-29 01:59:31 +08:00 · 2026-07-28 17:58:23 +08:00 · 2026-07-28 17:57:52 +08:00 · 2026-07-28 14:04:36 +08:00
31 changed files with 712 additions and 1740 deletions
@@ -6,17 +6,8 @@ APP_NAME=shaguabijia-app-server
 APP_DEBUG=true
 # ===== 数据库 =====
-# 本地开发/测试统一用 Docker PostgreSQL:run.bat/run.sh 会自动拉起容器
+# SQLite 本地文件路径。生产环境用 /opt/shaguabijia-app-server/data.db
-# (docker-compose.yml + scripts/ensure_pg.py)。详见 docs/database/postgres-migration.md。
+DATABASE_URL=sqlite:///./data/app.db
 # 生产用原生 PG,由 scripts/init_postgres.py 写入强随机密码的连接串。
 # ⚠️ scheme 必须是 postgresql+psycopg://(psycopg3);不要写成 postgresql://(会去找未装的 psycopg2)。
 DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia
 # Docker 自动定位:run.bat/run.sh 会自动找 Docker Desktop 并启动——优先从 PATH 上的 docker CLI 反推
 # 安装目录(装在 D 盘等非默认盘符也能找到),再退到注册表 / 常见目录。仅当你的安装位置极特殊、自动
 # 探测失败时,才需下面这行显式指到 exe(值可含空格,直接写到行尾即可,无需引号):
 # DOCKER_DESKTOP_EXE=D:\Program Files\Docker\Docker\Docker Desktop.exe
 # 实在不想装/启 Docker → 把上面 DATABASE_URL 改成 sqlite 可降级跑(仅救急,PG 专有 SQL/严格性不被验证):
 # DATABASE_URL=sqlite:///./data/app.db
 # ===== JWT =====
 # 生产部署务必改成随机长字符串,可用:python -c "import secrets; print(secrets.token_urlsafe(64))"
@@ -60,6 +60,3 @@ tests/meituan_coupon_bj.tsv
 tests/meituan_coupon_data.tsv
 tests/meituan_coupon_fz.tsv
 tests/meituan_coupon_xm.tsv
 # git worktrees (superpowers 隔离工作区)
 .worktrees/
@@ -76,8 +76,8 @@ Endpoints under `app/api/internal/` are for server-to-server communication (pric
 ## Database
- **Dev/Test**: Docker PostgreSQL 16 — `run.sh`/`run.bat` auto-start it via `scripts/ensure_pg.py` + `docker-compose.yml`; `.env.example` ships the PG URL by default; pytest uses the same container's `shaguabijia_test` DB. **Local no longer uses SQLite** (the SQLite branch in `db/session.py` is retained as a fallback only).
+- **Dev**: SQLite (`sqlite:///./data/app.db`), `check_same_thread=False`, no connection pool.
- **Prod**: native PostgreSQL — bootstrap with `scripts/init_postgres.py` (no Docker). Pool size 10 + max overflow 20, pool_recycle 3600.
+- **Prod**: PostgreSQL — just change `DATABASE_URL` in `.env`. Pool size 10 + max overflow 20, pool_recycle 3600.
 - **Migrations**: Alembic with `render_as_batch` for SQLite compatibility. ~60+ migration files in `alembic/versions/` (filenames are descriptive, not hex prefixes). Migration chain uses `down_revision` within each file.
 - **New models**: Define in `app/models/`, import in `app/models/__init__.py`, then run `alembic revision --autogenerate`.
@@ -89,7 +89,7 @@ All config via `pydantic-settings` in `app/core/config.py`. Single `Settings` cl
 ## Testing
- `tests/conftest.py`: Sets env vars BEFORE imports, ensures the Docker PG `shaguabijia_test` DB via `scripts/ensure_pg.py`, builds all tables with `Base.metadata.create_all()` (drop+create for a clean start), tears down with `drop_all()`.
+- `tests/conftest.py`: Sets env vars BEFORE imports, creates temp SQLite file, builds all tables with `Base.metadata.create_all()`, tears down with `drop_all()` + unlink.
 - External integrations are monkeypatched in tests (e.g., WeChat Pay, Jiguang, Pangle callbacks) — tests never make real HTTP calls.
 - `TestClient` from FastAPI is used for all tests. Rate limiting is disabled globally in tests.
@@ -0,0 +1,26 @@
 """merge comparison platforms + fail_reason heads
 Revision ID: 6d2309208549
 Revises: comparison_platforms_col, comparison_record_fail_reason
 Create Date: 2026-07-29 01:48:41.868083
 """
 from typing import Sequence, Union
 from alembic import op
 import sqlalchemy as sa
 # revision identifiers, used by Alembic.
 revision: str = '6d2309208549'
 down_revision: Union[str, Sequence[str], None] = ('comparison_platforms_col', 'comparison_record_fail_reason')
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    pass
 def downgrade() -> None:
    pass
@@ -0,0 +1,47 @@
 """add platforms unified array column to comparison_record
 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
 status/is_best/display, 记录页据此直接渲染, 不再靠 comparison_results + 客户端合并 + 前端派生。
 纯新增列, 老记录为空 → 前端回退老 comparison_results。
 Revision ID: comparison_platforms_col
 Revises: user_manual_risk_fields
 """
 from __future__ import annotations
 from collections.abc import Sequence
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 from alembic import op
 revision: str = "comparison_platforms_col"
 down_revision: str | None = "user_manual_risk_fields"
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
 _JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
 def upgrade() -> None:
    # 幂等: 线上为了提前给历史数据补 platforms(2026-07-29), 已手动
    # `ALTER TABLE comparison_record ADD COLUMN IF NOT EXISTS platforms jsonb
    #  NOT NULL DEFAULT '[]'::jsonb`(与本 migration 定义一致)。列已存在时跳过,
    # 否则上线 alembic upgrade head 会撞 DuplicateColumn 直接部署失败。
    bind = op.get_bind()
    cols = {c["name"] for c in sa.inspect(bind).get_columns("comparison_record")}
    if "platforms" in cols:
        return
    with op.batch_alter_table("comparison_record") as batch_op:
        batch_op.add_column(
            sa.Column(
                "platforms", _JSON, nullable=False,
                server_default=sa.text("'[]'"),
            )
        )
 def downgrade() -> None:
    with op.batch_alter_table("comparison_record") as batch_op:
        batch_op.drop_column("platforms")
@@ -0,0 +1,49 @@
 """comparison_record.fail_reason (失败卡展示原因)
 Revision ID: comparison_record_fail_reason
 Revises: user_manual_risk_fields
 Create Date: 2026-07-28 12:00:00.000000
 失败记录的展示原因:information 具体则=它;笼统则由写路径从 platform_results 捞出的
 业务原因;纯系统失败为 None(端侧品牌兜底)。见 repositories.comparison._derive_fail_display。
 """
 from typing import Sequence, Union
 from alembic import op
 import sqlalchemy as sa
 # revision identifiers, used by Alembic.
 revision: str = 'comparison_record_fail_reason'
 down_revision: Union[str, Sequence[str], None] = 'user_manual_risk_fields'
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
        batch_op.add_column(sa.Column('fail_reason', sa.String(length=256), nullable=True))
    # 回填老失败记录:information 具体的直接搬过来(笼统/系统失败留 None → 端侧品牌兜底)。
    # 新记录由写路径 _derive_fail_display 落库(含 platform_results 救援/补判),不走这条。
    # platform_results 只在 raw_payload 里,SQL 里不易解析,故老记录不做救援/补判(可接受:
    # 老 mixed/打烊记录回退品牌兜底);具体 information 的老记录本次即可显示真实原因。
    op.execute(
        """
        UPDATE comparison_record
        SET fail_reason = information
        WHERE status = 'failed'
          AND information IS NOT NULL
          AND information <> ''
          AND information NOT IN (
              '比价过程出错，请稍后重试',
              '比价出错',
              '比价未完成',
              'done 参数缺少可验证的目标平台结果'
          )
        """
    )
 def downgrade() -> None:
    with op.batch_alter_table('comparison_record', schema=None) as batch_op:
        batch_op.drop_column('fail_reason')
@@ -43,7 +43,7 @@ _KNOWN_PROD_BUSINESS_CODE_IDS = frozenset({"104098712", "104099389"})
 _TEST_BUSINESS_CODE_IDS = frozenset({"104127529", "104127626", "104137445"})
-def _business_code_ids(db: Session, app_env: str | None) -> set[str]:
+def business_code_ids(db: Session, app_env: str | None) -> set[str]:
    """返回指定应用环境下可用于业务收益对账的 GroMore 聚合代码位。"""
    prod_config = app_config.get_ad_config(db)
    prod_ids = set(_KNOWN_PROD_BUSINESS_CODE_IDS) | {
@@ -320,10 +320,10 @@ def ad_revenue_report(
    # 业务口径仅保留正式配置/测试业务链路实际使用的代码位。穿山甲“全量”还包含广告测试
    # demo、插屏等没有客户端收益上报的曝光，两边直接比较会天然产生假差额。
-    business_code_ids: set[str] | None = None
+    business_ids: set[str] | None = None
    if revenue_scope == "business":
-        business_code_ids = _business_code_ids(db, app_env)
+        business_ids = business_code_ids(db, app_env)
-        events = [e for e in events if e.get("our_code_id") in business_code_ids]
+        events = [e for e in events if e.get("our_code_id") in business_ids]
    # 排序:time=按时间倒序(新→旧);ecpm=按 eCPM 数值倒序(eCPM 原值是字符串「分」,转数值排;
    # 纯发奖行用其发奖采用的 eCPM,缺失/非法计 0 排末尾)。
@@ -381,7 +381,7 @@ def ad_revenue_report(
            date_from=date_from,
            date_to=date_to,
            app_env=app_env,
-            our_code_ids=business_code_ids,
+            our_code_ids=business_ids,
        )
        if pangle_aggs:
            by_date = {a["date"]: a for a in pangle_aggs}
@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
 from app.core import rewards
 from app.core.config import settings
 from app.models.ad_ecpm import AdEcpmRecord
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.admin import AdminAuditLog
@@ -1257,11 +1258,15 @@ def user_reward_stats(
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    withdraw_source: str | None = None,
    app_env: str | None = None,
    revenue_scope: str = "all",
    feed_scene: str | None = None,
 ) -> dict:
    """提现详情「用户统计区」10 项。窗口作用于除「现金余额」外的所有项(余额是当前快照)。
-    口径:激励视频/信息流只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加);
+    口径:激励视频/信息流奖励数量只统计 granted;数量——视频按条数、信息流按份数(unit_count 累加)。
-    平均 eCPM 用原始分值(分/千次)按记录取算术平均;各「提现」= 该来源累计金币折现。
+    平均 Draw eCPM 与广告收益报表一致:基于 ad_ecpm_record 的全部 draw/feed 展示记录计算，
    不以是否发奖为筛选条件。各「提现」= 该来源累计金币折现。
    传统任务 = 窗口内正向金币中,排除广告(reward_video/feed_ad_reward)与人工调整后的折现。
    """
    withdraw_source_conds = (
@@ -1296,30 +1301,68 @@ def user_reward_stats(
    # 只投影本统计实际使用的列。避免滚动发布或旧本地库尚未补齐无关新列时，
    # SQLAlchemy 因 select(ORM) 自动展开整表字段而让提现详情整体 500。
    business_ids: set[str] | None = None
    if revenue_scope == "business":
        # 与广告收益报表共用正式/测试业务代码位集合，避免两个页面随配置切换后再次漂移。
        from app.admin.repositories.ad_revenue import business_code_ids
        business_ids = business_code_ids(db, app_env)
    rv_conds = [
        AdRewardRecord.user_id == user_id,
        AdRewardRecord.reward_scene == "reward_video",
        AdRewardRecord.status == "granted",
        *_window_conds(AdRewardRecord.created_at, date_from, date_to),
    ]
    if app_env is not None:
        rv_conds.append(AdRewardRecord.app_env == app_env)
    if business_ids is not None:
        rv_conds.append(AdRewardRecord.our_code_id.in_(business_ids))
    rv = db.execute(
        select(AdRewardRecord.ecpm_raw, AdRewardRecord.coin).where(
-            AdRewardRecord.user_id == user_id,
+            *rv_conds,
            AdRewardRecord.reward_scene == "reward_video",
            AdRewardRecord.status == "granted",
            *_window_conds(AdRewardRecord.created_at, date_from, date_to),
        )
    ).all()
    rv_ecpms = [rewards.parse_ecpm_fen(r.ecpm_raw) for r in rv if r.ecpm_raw]
    rv_coins = sum(r.coin for r in rv)
-    feed = db.execute(
+    feed_reward_conds = [
        AdFeedRewardRecord.user_id == user_id,
        AdFeedRewardRecord.status == "granted",
        *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
    ]
    if app_env is not None:
        feed_reward_conds.append(AdFeedRewardRecord.app_env == app_env)
    if feed_scene is not None:
        feed_reward_conds.append(AdFeedRewardRecord.feed_scene == feed_scene)
    if business_ids is not None:
        feed_reward_conds.append(AdFeedRewardRecord.our_code_id.in_(business_ids))
    feed_rewards = db.execute(
        select(
            AdFeedRewardRecord.unit_count,
            AdFeedRewardRecord.ecpm_raw,
            AdFeedRewardRecord.coin,
        ).where(
-            AdFeedRewardRecord.user_id == user_id,
+            *feed_reward_conds,
            AdFeedRewardRecord.status == "granted",
            *_window_conds(AdFeedRewardRecord.created_at, date_from, date_to),
        )
    ).all()
-    feed_ecpms = [rewards.parse_ecpm_fen(f.ecpm_raw) for f in feed if f.ecpm_raw]
+    feed_coins = sum(f.coin for f in feed_rewards)
-    feed_coins = sum(f.coin for f in feed)
+
    feed_impression_conds = [
        AdEcpmRecord.user_id == user_id,
        AdEcpmRecord.ad_type.in_(("draw", "feed")),
        *_window_conds(AdEcpmRecord.created_at, date_from, date_to),
    ]
    if app_env is not None:
        feed_impression_conds.append(AdEcpmRecord.app_env == app_env)
    if feed_scene is not None:
        feed_impression_conds.append(AdEcpmRecord.feed_scene == feed_scene)
    if business_ids is not None:
        feed_impression_conds.append(AdEcpmRecord.our_code_id.in_(business_ids))
    feed_impressions = db.execute(
        select(AdEcpmRecord.ecpm_raw).where(*feed_impression_conds)
    ).all()
    # 与 ad_revenue.category_stats 相同:每次展示权重相同，非法原值按 parse_ecpm_fen 记 0。
    feed_ecpms = [rewards.parse_ecpm_fen(row.ecpm_raw) for row in feed_impressions]
    trad_coins = db.execute(
        select(func.coalesce(func.sum(CoinTransaction.amount), 0)).where(
@@ -1338,7 +1381,7 @@ def user_reward_stats(
        "reward_video_count": len(rv),
        "reward_video_avg_ecpm": round(sum(rv_ecpms) / len(rv_ecpms), 2) if rv_ecpms else 0.0,
        "reward_video_cash_cents": _coins_to_cents(rv_coins),
-        "feed_count": int(sum(f.unit_count for f in feed)),
+        "feed_count": int(sum(f.unit_count for f in feed_rewards)),
        "feed_avg_ecpm": round(sum(feed_ecpms) / len(feed_ecpms), 2) if feed_ecpms else 0.0,
        "feed_cash_cents": _coins_to_cents(feed_coins),
    }
@@ -88,6 +88,11 @@ def get_user_reward_stats(
    withdraw_source: Annotated[
        str | None, Query(pattern="^(coin_cash|invite_cash)$")
    ] = None,
    app_env: Annotated[str | None, Query(pattern="^(prod|test)$")] = None,
    revenue_scope: Annotated[str, Query(pattern="^(business|all)$")] = "all",
    feed_scene: Annotated[
        str | None, Query(pattern="^(comparison|coupon|welfare)$")
    ] = None,
 ) -> UserRewardStats:
    """提现详情抽屉「用户统计区」。date_from/date_to 都不传 = 注册至今(全量)。"""
    if not user_repo.user_exists(db, user_id):
@@ -99,6 +104,9 @@ def get_user_reward_stats(
            date_from=date_from,
            date_to=date_to,
            withdraw_source=withdraw_source,
            app_env=app_env,
            revenue_scope=revenue_scope,
            feed_scene=feed_scene,
        )
    )
@@ -60,7 +60,7 @@ class UserRewardStats(BaseModel):
    reward_video_avg_ecpm: float  # 平均激励视频 eCPM(分/千次)
    reward_video_cash_cents: int  # 激励视频提现(金币折现)
    feed_count: int  # 累计信息流广告数(granted 份数,unit_count 累加)
-    feed_avg_ecpm: float  # 平均信息流广告 eCPM(分/千次)
+    feed_avg_ecpm: float  # 全部 Draw/feed 实际展示的平均 eCPM(分/千次,含未发奖展示)
    feed_cash_cents: int  # 信息流广告提现(金币折现)
@@ -108,7 +108,7 @@ def _harvest_done_blocking(
 def _harvest_abort_blocking(
    trace_id: str, status_hint: str, reason: str | None, trace_url: str | None,
-) -> None:
+) -> int | None:
    with SessionLocal() as db:
        rec = crud_compare.harvest_abort(
            db, trace_id=trace_id, status=status_hint, reason=reason, trace_url=trace_url,
@@ -118,6 +118,7 @@ def _harvest_abort_blocking(
        extra={"phase": "harvest_abort",
               "status": (rec.status if rec else None), "reason": reason},
    )
    return rec.id if rec is not None else None
 async def _forward(
@@ -291,7 +292,10 @@ async def trace_epilogue(
@router.post("/trace/finalize", summary="比价 trace 收尾上云 (透传 + 夭折落库)")
 async def trace_finalize(
-    request: Request, user: OptionalUser, db: DbSession
+    request: Request,
    background_tasks: BackgroundTasks,
    user: OptionalUser,
    db: DbSession,
 ) -> dict[str, Any]:
    _ensure_compare_allowed(user, db)
    # 用户终止 / Phase1 未识别没到 done 帧: pricebot 打包半截上云返回 {trace_url};
@@ -302,12 +306,16 @@ async def trace_finalize(
        request, "/api/trace/finalize", user, harvest_first_frame=False,
    )
    try:
-        await run_in_threadpool(
+        record_id = await run_in_threadpool(
            _harvest_abort_blocking, trace_id,
            (meta.get("status") or "cancelled"),
            (meta.get("reason") or meta.get("information")),
            (resp.get("trace_url") if isinstance(resp, dict) else None),
        )
        if record_id is not None:
            background_tasks.add_task(
                backfill_comparison_llm_cost, record_id, trace_id
            )
    except Exception as e:  # noqa: BLE001
        logger.warning("harvest_abort failed trace=%s: %s", trace_id, e)
    return resp
@@ -102,12 +102,21 @@ class ComparisonRecord(Base):
    # done 帧 information 文案。成功:"在美团找到同店,到手价 ¥X…";
    # 失败:具体原因(如"美团、京东外卖均未找到该商品")。前端在比价失败时当原因展示。
    information: Mapped[str | None] = mapped_column(String(256), nullable=True)
    # 失败卡「原因」行的展示文案(仅 status=failed 时非空):information 具体则=它;笼统则从
    # platform_results 捞出的业务原因(打烊/未起送/找不到店或菜/单点不配送);纯系统失败为 None
    # → 端侧显示品牌兜底「网络开小差…」。写路径(harvest_done / upsert_record)落库时派生。
    # 见 repositories.comparison._derive_fail_display。
    fail_reason: Mapped[str | None] = mapped_column(String(256), nullable=True)
    # ===== 明细(JSON,越详细越好)=====
    # 下单菜品 [{name, qty, specs?}]
    items: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
    # 逐平台对比 [{platform_id, platform_name, package, price, is_source, rank, coupon_saved, coupon_name, applied_coupons}](price/coupon_saved 单位:元,原样存;coupon_name=优惠来源名;applied_coupons=[{name,amount}] 多券明细)
    comparison_results: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
    # 展示模型统一数组(pricebot done.params.platforms 原样存): 每平台一行、自带
    # status/is_best/display/display_order, 记录页据此直接渲染, 不再靠 comparison_results
    # + 客户端合并 + 前端派生。老记录/旧客户端为空 → 前端回退老 comparison_results 渲染。
    platforms: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
    # 目标平台未找到、跳过的菜名
    skipped_dish_names: Mapped[list] = mapped_column(_JSON, nullable=False, default=list)
    # 客户端上报的原始 payload(calibration + done.params 全量),未来取数兜底
@@ -52,6 +52,81 @@ def _product_names_from_items(items: list | None) -> str | None:
    return joined[:500] or None
 # ---- 失败记录的展示文案(记录页失败卡「原因」行)------------------------------
 # information 具体就直出;笼统(_GENERIC_INFO)则从 platform_results 捞一条用户可读的业务
 # 原因;捞不到 → None(端侧显示品牌兜底「网络开小差…」)。pricebot 把 store_closed /
 # no_delivery 漏成了 status=failed,这里按 reason 关键字补判;打烊类 reason 常带一坨脏店名
 # (店名+月售+起送+配送…),统一成简短模板。自动化黑话(搜索失败/读价失败/购物车残留/裸
 # FAILED…)不给用户看 → 归入品牌兜底。
 # pricebot 组不出具体原因时的笼统 information(线上统计的大头),一律走品牌兜底。
 _GENERIC_INFO = {
    "比价过程出错，请稍后重试",
    "比价出错",
    "比价未完成",
    "done 参数缺少可验证的目标平台结果",
 }
 # 干净业务结局 status(直接可信),按展示优先级(越靠前越先选)。
 _BIZ_STATUS_PRIORITY = (
    "below_minimum",
    "no_delivery",
    "store_closed",
    "items_not_found",
    "store_not_found",
 )
 def _store_closed_text(reason: str | None) -> str:
    """打烊/暂停营业/休息类 reason 常带脏店名元数据 → 只留结论,套简短模板。"""
    r = reason or ""
    if "暂停营业" in r:
        state = "暂停营业"
    elif "休息" in r:
        state = "休息中"
    else:
        state = "已打烊"
    return f"门店{state}，无法比价"
 def _target_display_reason(platform_results: dict | None) -> str | None:
    """从逐平台结果里挑一条"可展示给用户"的失败原因;挑不到返回 None。
    ① status 命中干净业务结局集 → 直接采信(打烊套模板,其余用 reason);
    ② 补判 pricebot 漏成 status=failed 的两类:打烊(套模板)、单点不配送(reason 本身干净);
    自动化黑话(搜索失败/读价失败/购物车残留/裸 FAILED…)一律不展示 → None。"""
    pr = platform_results or {}
    targets = [
        v for v in pr.values() if isinstance(v, dict) and not v.get("is_source")
    ]
    for want in _BIZ_STATUS_PRIORITY:  # ① 干净 status 优先
        for v in targets:
            if v.get("status") == want:
                if want == "store_closed":
                    return _store_closed_text(v.get("reason"))
                if v.get("reason"):
                    return v["reason"]
    for v in targets:  # ② 漏成 failed 的业务结局补判
        if v.get("status") != "failed":
            continue
        reason = (v.get("reason") or "").strip()
        if any(k in reason for k in ("打烊", "暂停营业", "休息")):
            return _store_closed_text(reason)
        if "单点不配送" in reason:
            return reason
    return None
 def _derive_fail_display(
    information: str | None, platform_results: dict | None
 ) -> str | None:
    """失败记录展示文案:information 具体则直出;笼统则从 platform_results 捞/补判;
    都拿不到 → None(端侧品牌兜底)。仅在 status=failed 时调用。"""
    info = (information or "").strip()
    text = info if (info and info not in _GENERIC_INFO) else _target_display_reason(
        platform_results
    )
    return text[:256] if text else None
 def _derive(payload: ComparisonRecordIn) -> dict:
    """从上报 payload 派生结构化列(best/saved/is_source_best/status)。"""
    results = payload.comparison_results
@@ -89,8 +164,9 @@ def _derive(payload: ComparisonRecordIn) -> dict:
    is_source_best = best.is_source if best is not None else None
-    # status:客户端显式给了就用;否则有"非源且有价"的结果=success,否则 failed
+    # status:优先 pricebot record_status(区分 below_minimum/store_closed) → 客户端显式 status
-    status = payload.status
+    # → 兜底"非源且有价"=success/否则 failed。record_status 让"未满起送"不再塌缩成 failed。
    status = payload.record_status or payload.status
    if status is None:
        has_valid_target = any(
            (not r.is_source) and r.price is not None for r in results
@@ -105,6 +181,11 @@ def _derive(payload: ComparisonRecordIn) -> dict:
        "saved_amount_cents": saved_amount_cents,
        "is_source_best": is_source_best,
        "status": status,
        "fail_reason": (
            _derive_fail_display(payload.information, _pr)
            if status == "failed"
            else None
        ),
    }
@@ -117,16 +198,35 @@ def upsert_record(
    灰度期老客户端 POST /compare/record 走这条,与后端 harvest 按 trace_id reconcile;
    新客户端不再 POST(改由 compare.py 透传壳 harvest 落库)。
    """
-    derived = _derive(payload)
+    # 单源派生: 与 harvest_done 一致, payload 带 platforms 时从它派生(唯一真相源
    # _derive_from_platforms), 老客户端不带 platforms 时回退 _derive(从 comparison_results)。
    if payload.platforms:
        derived = _derive_from_platforms(payload.platforms, payload.record_status)
        # 对齐 _derive 返回键(#189 fail_reason): 两路径 fields 键集一致, 覆盖已有行时不残留旧值
        derived["fail_reason"] = (
            _derive_fail_display(payload.information, payload.platform_results or {})
            if derived["status"] == "failed"
            else None
        )
        # 单源派生取自 platforms 源行(常无源平台元数据/店名)→ 空则用 payload 兜底不丢字段。
        # 下面 fields 不再显式写这四个键, 统一由 derived 提供(否则 dict(store_name=..., **derived)
        # 与 _derive_from_platforms 同名键撞键 TypeError)。
        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
            if not derived.get(_k):
                derived[_k] = getattr(payload, _k)
    else:
        derived = _derive(payload)
        # _derive 只从 comparison_results 派生, 不含源平台四件套 / store_name → 从 payload 补,
        # 与上面 platforms 分支键集对齐(fields 统一靠 **derived 提供这些列)。
        for _k in ("store_name", "source_platform_id", "source_platform_name", "source_package"):
            derived[_k] = getattr(payload, _k)
    items = [it.model_dump(exclude_none=True) for it in payload.items]
    fields = dict(
        device_id=payload.device_id,
        business_type=payload.business_type,
        store_name=payload.store_name,
        product_names=_product_names_from_items(items),
-        source_platform_id=payload.source_platform_id,
+        # store_name / source_platform_id / source_platform_name / source_package 统一由
-        source_platform_name=payload.source_platform_name,
+        # derived 提供(见上方两分支补齐), 不在此显式写 —— 否则与 _derive_from_platforms 撞键。
        source_package=payload.source_package,
        information=payload.information,
        best_deeplink=payload.best_deeplink,
        trace_url=payload.trace_url,
@@ -134,6 +234,7 @@ def upsert_record(
        skipped_dish_count=payload.skipped_dish_count,
        items=items,
        comparison_results=[r.model_dump() for r in payload.comparison_results],
        platforms=list(payload.platforms or []),
        skipped_dish_names=list(payload.skipped_dish_names),
        # 客户端环境 / 性能(debug,客户端上报;旧客户端为 None)
        device_model=payload.device_model,
@@ -206,7 +307,8 @@ def upsert_record(
 def _derive_from_results(
-    results: list[dict], platform_results: dict | None = None
+    results: list[dict], platform_results: dict | None = None,
    record_status: str | None = None,
 ) -> dict:
    """从 done 帧 comparison_results(pricebot 原始 dict 列表)派生结构化列。
    等价 _derive,但吃原始字段(is_source/price/rank/platform_id/store_name...)而非 pydantic 对象。
@@ -254,7 +356,53 @@ def _derive_from_results(
        "saved_amount_cents": saved_amount_cents,
        "is_source_best": best.get("is_source") if best else None,
        "store_name": (src_row or {}).get("store_name") or None,
-        "status": "success" if has_valid_target else "failed",
+        # 记录级结局: 优先用 pricebot 下发的 record_status(区分 below_minimum/store_closed,
        # 不再把"未满起送"塌缩成 failed → 记录页不再误报"网络开小差"); 旧 pricebot 未下发时
        # 回退老的 success/failed 二态派生, 向后兼容。
        "status": record_status or ("success" if has_valid_target else "failed"),
    }
 def _derive_from_platforms(
    platforms: list, record_status: str | None = None,
 ) -> dict:
    """从 done 帧 platforms(每平台一行、渲染就绪)派生结构化列——**单一真相源**。
    best_* 直接取 platforms 里 is_best 的那一行、source_* 取 role=source 行,与前端读的
    platforms 天然一致(不再像 _derive_from_results 那样从 comparison_results 二次评最优,
    消除"标量列 vs platforms"双源不一致)。platforms 非空时优先走这里;老 pricebot 无
    platforms 时调用方回退 _derive_from_results(向后兼容)。"""
    rows = [p for p in (platforms or []) if isinstance(p, dict)]
    src = next((p for p in rows if p.get("role") == "source"), None)
    best = next((p for p in rows if p.get("is_best")), None)
    source_price_cents = _yuan_to_cents(src.get("price")) if src else None
    best_price_cents = _yuan_to_cents(best.get("price")) if best else None
    saved_amount_cents = None
    if source_price_cents is not None and best_price_cents is not None:
        saved_amount_cents = source_price_cents - best_price_cents
    has_valid_target = any(
        p.get("role") != "source" and p.get("price") is not None for p in rows
    )
    # store_name: 优先源行; recompare 场景源平台自己当目标、源行被目标覆盖(pricebot
    # _build_platform_rows 有意去重, platforms 无 role=source 行)→ 回退 best 行 → 首个有店名
    # 的行(显示现场实际比到的店), 免得记录页店名空掉兜底显示成"比价"。正常比价有源行不走回退。
    store_name = (
        (src or {}).get("store_name")
        or (best or {}).get("store_name")
        or next((p.get("store_name") for p in rows if p.get("store_name")), None)
    )
    return {
        "source_platform_id": (src or {}).get("platform_id"),
        "source_platform_name": (src or {}).get("platform_name"),
        "source_package": (src or {}).get("package"),
        "source_price_cents": source_price_cents,
        "best_platform_id": (best or {}).get("platform_id"),
        "best_platform_name": (best or {}).get("platform_name"),
        "best_price_cents": best_price_cents,
        "saved_amount_cents": saved_amount_cents,
        "is_source_best": (best.get("role") == "source") if best else None,
        "store_name": store_name or None,
        "status": record_status or ("success" if has_valid_target else "failed"),
    }
@@ -413,12 +561,40 @@ def harvest_done(
    返回 (记录, 是否本次**新**落成 success)——供调用方据此幂等发一次邀请奖。
    行不存在(理论上帧0已建;防御)则新建。"""
    results = done_params.get("comparison_results") or []
-    derived = _derive_from_results(results, done_params.get("platform_results"))
+    # 展示模型统一数组(pricebot 新增, 每平台一行自带 status/is_best): 原样存, 记录页据此直渲染。
-    # 菜品:pricebot 已把源单菜品塞进 comparison_results[源行].items
+    # record_status: 记录级结局(success/below_minimum/store_closed/failed), 覆盖老二态派生。
-    items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
+    platforms = done_params.get("platforms") or []
    record_status = done_params.get("record_status")
    # 单源派生: platforms(含 pricebot 权威 is_best)是唯一真相源, best_*/source_*/saved/status
    # 全从它取 → 与前端读的 platforms 天然一致; 菜品也取 platforms 源行。老 pricebot 无
    # platforms 时回退从 comparison_results 派生(向后兼容)。
    if platforms:
        derived = _derive_from_platforms(platforms, record_status)
        # 菜品优先源行; recompare 无源行 → 回退 best 行 → 首个有菜品的行(同 store_name 回退)
        _item_row = (
            next((p for p in platforms if isinstance(p, dict) and p.get("role") == "source"), None)
            or next((p for p in platforms if isinstance(p, dict) and p.get("is_best")), None)
            or next((p for p in platforms if isinstance(p, dict) and p.get("items")), None)
        )
        items = (_item_row or {}).get("items") or []
    else:
        derived = _derive_from_results(
            results, done_params.get("platform_results"), record_status
        )
        # pricebot 已把源单菜品塞进 comparison_results[源行].items
        items = next((r.get("items") or [] for r in results if r.get("is_source")), [])
    # 失败展示原因(#189): platforms / results 两个派生分支的 status 都可能 failed, 统一在此算
    fail_reason = (
        _derive_fail_display(
            done_params.get("information"), done_params.get("platform_results")
        )
        if derived["status"] == "failed"
        else None
    )
    fields = dict(
        business_type=business_type or "food",
        information=done_params.get("information") or None,
        fail_reason=fail_reason,
        # best_deeplink 来自客户端剪贴板采集,harvest 拿不到 → 留空(灰度期 fromComparison 会补;
        # 纯 harvest 行「再次比价」退化为按 package 拉起 App。要精确深链需客户端另传,后续)。
        trace_url=trace_url or done_params.get("trace_url"),
@@ -426,6 +602,7 @@ def harvest_done(
        skipped_dish_count=done_params.get("skipped_dish_count"),
        skipped_dish_names=list(done_params.get("skipped_dish_names") or []),
        comparison_results=results,
        platforms=platforms,
        items=items,
        product_names=_product_names_from_items(items),
        raw_payload=done_params,
@@ -107,6 +107,13 @@ class ComparisonRecordIn(BaseModel):
    # 明细
    items: list[ComparisonItemIn] = Field(default_factory=list)
    comparison_results: list[ComparisonResultIn] = Field(default_factory=list)
    # 展示模型统一数组(pricebot done.params.platforms 原样透传): 每平台一行、自带
    # status/is_best/display/display_order,记录页据此直渲染。宽松 list[dict] 存(结构由
    # pricebot 定,server 只原样落库),前端读它、老记录空时回退 comparison_results。
    platforms: list[dict] = Field(default_factory=list)
    # 记录级结局(pricebot 下发): success/below_minimum/store_closed/failed。让"未满起送"不再
    # 被塌缩成 failed。_derive 优先用它、其次客户端 status、再兜底二态派生。
    record_status: str | None = None
    # 逐平台结局摘要(含失败平台的细分原因 status: store_not_found/items_not_found/below_minimum/
    # unsupported/...)。来自 done.params.platform_results,客户端透传;落 raw_payload(不单列),
    # admin「卡在哪一步」从这里读。dict{platform_id: {...}} 宽松存(结构由 pricebot 定——是
@@ -173,8 +180,13 @@ class ComparisonRecordOut(BaseModel):
    skipped_dish_count: int | None = None
    status: str
    information: str | None = None
    # 失败卡「原因」文案:具体失败给具体原因,纯系统失败为 None(端侧品牌兜底)。见模型 fail_reason。
    fail_reason: str | None = None
    items: list = []
    comparison_results: list = []
    # 展示模型统一数组(每平台一行、自带 status/is_best/display/display_order): 记录页据此
    # 直渲染, 不再靠 comparison_results + 前端派生。老记录为空 → 前端回退 comparison_results。
    platforms: list = []
    skipped_dish_names: list = []
    total_ms: int | None = None
    # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
@@ -135,7 +135,7 @@ def repair_missing_comparison_llm_costs(
                select(ComparisonRecord.id, ComparisonRecord.trace_id)
                .where(
                    *date_conditions,
-                    ComparisonRecord.status.in_(("success", "failed")),
+                    ComparisonRecord.status.in_(("success", "failed", "cancelled")),
                    ComparisonRecord.llm_cost_yuan.is_(None),
                )
                .order_by(ComparisonRecord.created_at.desc(), ComparisonRecord.id.desc())
@@ -1,22 +0,0 @@
 # 本地开发/测试用 PostgreSQL。生产用原生 PG(scripts/init_postgres.py),不使用本文件。
 services:
  postgres:
    image: postgres:16-alpine
    container_name: shaguabijia-pg
    environment:
      POSTGRES_USER: shaguabijia_app
      POSTGRES_PASSWORD: shaguabijia_dev_pw
      POSTGRES_DB: shaguabijia
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./docker/initdb:/docker-entrypoint-initdb.d:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U shaguabijia_app -d shaguabijia"]
      interval: 3s
      timeout: 3s
      retries: 20
 volumes:
  pgdata:
@@ -1,3 +0,0 @@
 -- 仅在 pgdata 卷首次初始化时执行一次(以 shaguabijia_app 连 shaguabijia 库运行)。
 -- 幂等兜底见 scripts/ensure_pg.py 的 _ensure_test_db()。
 CREATE DATABASE shaguabijia_test OWNER shaguabijia_app;
@@ -27,21 +27,7 @@ PG 默认上 16 版(工具链最齐),驱动用 **psycopg3**(SQLAlchemy 2.0 时
 ## 1. 本地起 PG + 跑通空库(半天)
-### 1.0 推荐:Docker 一键起(本地开发/测试)
+### 1.1 装 PG
 本地开发不必手动装 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
@@ -1,740 +0,0 @@
 # 本地开发切 Docker PostgreSQL 实现计划
 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
 **Goal:** 让 app-server 本地开发运行与 pytest 都跑在 Docker 化的 PostgreSQL 16 上,退掉 SQLite,`run.bat`/`run.sh` 启动时自动检测并拉起 PG(必要时先启 Docker Desktop、缺镜像先拉)。
 **Architecture:** 新增 `docker-compose.yml`(声明 PG 服务/卷/健康检查)+ `scripts/ensure_pg.py`(跨平台引导:探测→启 Docker→compose up→等就绪→幂等建测试库),由 `run.sh`/`run.bat`/`tests/conftest.py` 三处共用。`.env.example` 默认切 PG。`app/db/session.py` 与 `alembic/env.py` 已天然支持 PG,无需改。
 **Tech Stack:** Docker Compose、`postgres:16-alpine`、Python 3.10+ 标准库(`socket`/`subprocess`/`urllib.parse`)、psycopg3(已装)、SQLAlchemy 2.0 + Alembic、pytest。
 **工作目录:** 本计划在 worktree `.worktrees/local-dev-postgres-docker`(分支 `chore/local-dev-postgres-docker`,基于 `main`)内执行。下面所有路径相对该 worktree 根(= 仓库根)。
 **设计依据:** [docs/superpowers/specs/2026-07-08-local-dev-postgres-docker-design.md](../specs/2026-07-08-local-dev-postgres-docker-design.md)
 **连接参数(全程固定值):**
 - 镜像 `postgres:16-alpine`,容器名 `shaguabijia-pg`,宿主端口 `5432`
 - 用户 `shaguabijia_app`,dev 密码 `shaguabijia_dev_pw`(本地非机密)
 - 业务库 `shaguabijia`,测试库 `shaguabijia_test`,命名卷 `pgdata`
 - dev URL:`postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia`
 - test URL:`...@localhost:5432/shaguabijia_test`
 **前置:** 执行机已安装 Docker Desktop。
 ---
 ## Task 1: Docker Compose + 测试库 initdb 脚本
 **Files:**
 - Create: `docker-compose.yml`
 - Create: `docker/initdb/01-create-test-db.sql`
 - [ ] **Step 1: 写 `docker-compose.yml`**
 ```yaml
 # 本地开发/测试用 PostgreSQL。生产用原生 PG(scripts/init_postgres.py),不使用本文件。
 services:
  postgres:
    image: postgres:16-alpine
    container_name: shaguabijia-pg
    environment:
      POSTGRES_USER: shaguabijia_app
      POSTGRES_PASSWORD: shaguabijia_dev_pw
      POSTGRES_DB: shaguabijia
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./docker/initdb:/docker-entrypoint-initdb.d:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U shaguabijia_app -d shaguabijia"]
      interval: 3s
      timeout: 3s
      retries: 20
 volumes:
  pgdata:
 ```
 - [ ] **Step 2: 写 `docker/initdb/01-create-test-db.sql`**
 ```sql
 -- 仅在 pgdata 卷首次初始化时执行一次(以 shaguabijia_app 连 shaguabijia 库运行)。
 -- 幂等兜底见 scripts/ensure_pg.py 的 _ensure_test_db()。
 CREATE DATABASE shaguabijia_test OWNER shaguabijia_app;
 ```
 - [ ] **Step 3: 起容器验证**
 Run: `docker compose up -d`
 Expected: 拉取 `postgres:16-alpine`(首次)后 `Container shaguabijia-pg  Started`。
 - [ ] **Step 4: 验证两个库都在 + 健康**
 Run: `docker compose exec -T postgres psql -U shaguabijia_app -d shaguabijia -tAc "SELECT datname FROM pg_database WHERE datname IN ('shaguabijia','shaguabijia_test') ORDER BY 1"`
 Expected 输出:
 ```
 shaguabijia
 shaguabijia_test
 ```
 - [ ] **Step 5: 验证 `.worktrees/` 与 `data/` 忽略不受影响、compose 无落盘到项目目录**
 Run: `git status --short`
 Expected: 只列出本任务新增的 `docker-compose.yml`、`docker/initdb/01-create-test-db.sql`(数据在命名卷 `pgdata`,不在项目目录;`.worktrees/` 已忽略)。
 - [ ] **Step 6: Commit**
 ```bash
 git add docker-compose.yml docker/initdb/01-create-test-db.sql
 git commit -m "feat(dev): docker-compose 起本地 PostgreSQL(含测试库 initdb)"
 ```
 ---
 ## Task 2: `scripts/ensure_pg.py` 引导脚本(TDD)
 **Files:**
 - Create: `scripts/ensure_pg.py`
 - Test: `tests/test_ensure_pg.py`
 > 说明:纯函数(URL 解析 / sqlite 判定 / 端口探测 / 平台命令映射 / sqlite 守卫 / 端口通时短路)走 TDD 单测;真正拉 Docker 的编排 `ensure()` 全链路靠 Task 3/5 的运行来验证(需真 Docker,不做单测)。此时 `conftest.py` 仍是 SQLite,不依赖 PG,单测可独立跑。
 - [ ] **Step 1: 写失败测试 `tests/test_ensure_pg.py`**
 ```python
 """scripts/ensure_pg.py 纯函数单测(不需要 Docker/PG)。"""
 from __future__ import annotations
 import socket
 from scripts.ensure_pg import (
    _docker_desktop_cmd,
    _is_sqlite,
    _parse_host_port,
    _port_open,
    ensure,
 )
 def test_is_sqlite():
    assert _is_sqlite("sqlite:///./data/app.db")
    assert _is_sqlite("  SQLite:///x ")
    assert not _is_sqlite("postgresql+psycopg://u:p@localhost:5432/db")
 def test_parse_host_port_full():
    assert _parse_host_port(
        "postgresql+psycopg://u:p@localhost:5432/shaguabijia"
    ) == ("localhost", 5432)
 def test_parse_host_port_defaults():
    # 缺端口 → 5432
    assert _parse_host_port("postgresql+psycopg://u:p@db.example/x")[1] == 5432
    # 缺 host → localhost
    assert _parse_host_port("postgresql+psycopg:///x") == ("localhost", 5432)
 def test_parse_host_port_testdb():
    assert _parse_host_port(
        "postgresql+psycopg://u:p@localhost:5432/shaguabijia_test"
    ) == ("localhost", 5432)
 def test_port_open_true():
    srv = socket.socket()
    srv.bind(("127.0.0.1", 0))
    srv.listen(1)
    port = srv.getsockname()[1]
    try:
        assert _port_open("127.0.0.1", port, timeout=1.0)
    finally:
        srv.close()
 def test_port_open_false():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
    s.close()  # 释放端口,无人监听 → 连接应失败
    assert not _port_open("127.0.0.1", port, timeout=0.3)
 def test_docker_desktop_cmd_windows():
    cmd = _docker_desktop_cmd("win32", r"C:\Program Files")
    assert cmd is not None
    assert cmd[0].endswith("Docker Desktop.exe")
    assert "Docker" in cmd[0]
 def test_docker_desktop_cmd_darwin():
    assert _docker_desktop_cmd("darwin", "") == ["open", "-a", "Docker"]
 def test_docker_desktop_cmd_linux():
    assert _docker_desktop_cmd("linux", "") is None
 def test_ensure_rejects_sqlite():
    # dev 守卫:sqlite 直接 False(不碰 Docker)
    assert ensure("sqlite:///./data/app.db") is False
 def test_ensure_shortcircuits_when_pg_up(monkeypatch):
    # 端口通 → 直接 True,绝不触碰 docker
    monkeypatch.setattr("scripts.ensure_pg._port_open", lambda *a, **k: True)
    def _boom():
        raise AssertionError("端口通时不应调用 docker")
    monkeypatch.setattr("scripts.ensure_pg._docker_cli_ok", _boom)
    assert ensure("postgresql+psycopg://u:p@localhost:5432/shaguabijia") is True
 ```
 - [ ] **Step 2: 跑测试确认失败**
 Run: `pytest tests/test_ensure_pg.py -v`
 Expected: FAIL —— `ModuleNotFoundError: No module named 'scripts.ensure_pg'`(还没建)。
 - [ ] **Step 3: 写实现 `scripts/ensure_pg.py`**
 ```python
 """确保本地 PostgreSQL 就绪(开发/测试统一用 Docker PG)。
 被三处复用:
  - run.sh / run.bat:`python -m scripts.ensure_pg`(CLI,失败退非 0)
  - tests/conftest.py:`from scripts.ensure_pg import ensure; ensure(test_url)`
 流程:读 DATABASE_URL → TCP 探测 → 没起就(必要时启 Docker Desktop)→
 `docker compose up -d` → 等 PG ready → 幂等确保测试库存在。全程无 SQLite 兜底。
 生产用原生 PG(scripts/init_postgres.py),不走本模块。
 """
 from __future__ import annotations
 import os
 import socket
 import subprocess
 import sys
 import time
 from pathlib import Path
 from urllib.parse import urlsplit
 ROOT = Path(__file__).resolve().parent.parent
 # 日志里可能含 emoji(如 ✅);Windows GBK 控制台(cmd.exe)无法编码会抛 UnicodeEncodeError → 脚本崩、
 # run.bat 误判 ensure_pg 失败。用 backslashreplace 保底:中文仍正常,仅不可编码字符被转义,不崩。
 for _stream in (sys.stdout, sys.stderr):
    try:
        _stream.reconfigure(errors="backslashreplace")
    except (AttributeError, ValueError):
        pass
 APP_DB = "shaguabijia"
 TEST_DB = "shaguabijia_test"
 DB_USER = "shaguabijia_app"
 COMPOSE_SERVICE = "postgres"
 DOCKER_START_TIMEOUT = int(os.environ.get("ENSURE_PG_DOCKER_TIMEOUT", "120"))
 PG_READY_TIMEOUT = int(os.environ.get("ENSURE_PG_READY_TIMEOUT", "60"))
 POLL_INTERVAL = 3.0
 SQLITE_FIX_HINT = (
    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia"
 )
 def _log(msg: str) -> None:
    print(f"[ensure_pg] {msg}", flush=True)
 def _is_sqlite(url: str) -> bool:
    return url.strip().lower().startswith("sqlite")
 def _parse_host_port(url: str) -> tuple[str, int]:
    """从 SQLAlchemy URL 取 host/port,缺省 localhost:5432。"""
    parts = urlsplit(url)
    return (parts.hostname or "localhost"), (parts.port or 5432)
 def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False
 def _docker_desktop_cmd(platform: str, program_files: str) -> list[str] | None:
    """按平台给出启动 Docker Desktop 的命令;Linux 返回 None(daemon 需 sudo,让用户手动)。"""
    if platform.startswith("win"):
        return [str(Path(program_files) / "Docker" / "Docker" / "Docker Desktop.exe")]
    if platform == "darwin":
        return ["open", "-a", "Docker"]
    return None
 def _docker_ok(subcmd: str) -> bool:
    """`docker version`(CLI 在不在)/`docker info`(daemon 起没起)成功与否。"""
    try:
        subprocess.run(
            ["docker", subcmd],
            cwd=ROOT,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=True,
        )
        return True
    except (OSError, subprocess.CalledProcessError):
        return False
 def _docker_cli_ok() -> bool:
    return _docker_ok("version")
 def _docker_daemon_ok() -> bool:
    return _docker_ok("info")
 def _start_docker_daemon() -> bool:
    """守护进程没起时按平台拉起,轮询到就绪。返回是否成功。"""
    if _docker_daemon_ok():
        return True
    cmd = _docker_desktop_cmd(
        sys.platform, os.environ.get("ProgramFiles", r"C:\Program Files")
    )
    if cmd is None:
        _log("Docker 守护进程未运行。Linux 请手动:sudo systemctl start docker,然后重试。")
        return False
    if sys.platform.startswith("win") and not Path(cmd[0]).exists():
        _log(f"找不到 Docker Desktop:{cmd[0]}。请手动启动 Docker Desktop 后重试。")
        return False
    _log(f"启动 Docker Desktop(首次冷启可能 30-60s)…")
    try:
        subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    except OSError as e:
        _log(f"启动 Docker Desktop 失败:{e}")
        return False
    deadline = time.monotonic() + DOCKER_START_TIMEOUT
    while time.monotonic() < deadline:
        if _docker_daemon_ok():
            _log("Docker 守护进程已就绪。")
            return True
        _log("等待 Docker 守护进程…")
        time.sleep(POLL_INTERVAL)
    _log(f"等待 Docker 守护进程超时({DOCKER_START_TIMEOUT}s)。")
    return False
 def _compose_up() -> bool:
    _log("docker compose up -d(镜像缺失会自动拉取,首用约几十秒)…")
    try:
        subprocess.run(["docker", "compose", "up", "-d"], cwd=ROOT, check=True)
        return True
    except (OSError, subprocess.CalledProcessError) as e:
        _log(f"docker compose up 失败:{e}")
        return False
 def _pg_isready() -> bool:
    r = subprocess.run(
        ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
         "pg_isready", "-U", DB_USER, "-d", APP_DB],
        cwd=ROOT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    return r.returncode == 0
 def _wait_pg_ready(host: str, port: int) -> bool:
    deadline = time.monotonic() + PG_READY_TIMEOUT
    while time.monotonic() < deadline:
        if _port_open(host, port) and _pg_isready():
            _log("PostgreSQL 已就绪。")
            return True
        _log("等待 PostgreSQL 就绪…")
        time.sleep(POLL_INTERVAL)
    _log(f"等待 PostgreSQL 就绪超时({PG_READY_TIMEOUT}s)。")
    return False
 def _ensure_test_db() -> None:
    """幂等建测试库(兼容老 pgdata 卷首启没跑 initdb 的情况)。"""
    check = subprocess.run(
        ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
         "psql", "-U", DB_USER, "-d", APP_DB, "-tAc",
         f"SELECT 1 FROM pg_database WHERE datname='{TEST_DB}'"],
        cwd=ROOT, capture_output=True, text=True,
    )
    if check.returncode == 0 and check.stdout.strip() == "1":
        return
    _log(f"建测试库 {TEST_DB}…")
    subprocess.run(
        ["docker", "compose", "exec", "-T", COMPOSE_SERVICE,
         "psql", "-U", DB_USER, "-d", APP_DB, "-c",
         f"CREATE DATABASE {TEST_DB} OWNER {DB_USER}"],
        cwd=ROOT, check=False,
    )
 def ensure(database_url: str | None = None) -> bool:
    """确保 PG 就绪,返回 True/False。database_url 缺省从 settings 读(尊重 .env)。"""
    if database_url is None:
        from app.core.config import settings  # 延迟导入,避免过早固化 settings
        database_url = settings.DATABASE_URL
    if _is_sqlite(database_url):
        _log("检测到 DATABASE_URL 仍是 SQLite。本地开发/测试已切 PostgreSQL,请改成:")
        _log(f"  DATABASE_URL={SQLITE_FIX_HINT}")
        return False
    host, port = _parse_host_port(database_url)
    if _port_open(host, port):
        _log(f"✅ PostgreSQL 已在 {host}:{port} 运行,跳过 Docker。")
        return True
    _log(f"{host}:{port} 无 PostgreSQL,准备用 Docker 拉起…")
    if not _docker_cli_ok():
        _log("未检测到 docker 命令。请先安装 Docker Desktop:")
        _log("  https://www.docker.com/products/docker-desktop/")
        return False
    if not _start_docker_daemon():
        return False
    if not _compose_up():
        return False
    if not _wait_pg_ready(host, port):
        return False
    _ensure_test_db()
    return True
 if __name__ == "__main__":
    sys.exit(0 if ensure() else 1)
 ```
 - [ ] **Step 4: 跑测试确认通过**
 Run: `pytest tests/test_ensure_pg.py -v`
 Expected: 11 passed。
 - [ ] **Step 5: 手动冒烟(PG 已在跑时应秒过短路)**
 Run: `python -m scripts.ensure_pg`
 Expected: 打印 `[ensure_pg] ✅ PostgreSQL 已在 localhost:5432 运行,跳过 Docker。`,退出码 0。
 - [ ] **Step 6: Commit**
 ```bash
 git add scripts/ensure_pg.py tests/test_ensure_pg.py
 git commit -m "feat(dev): scripts/ensure_pg.py 探测/拉起本地 Docker PostgreSQL"
 ```
 ---
 ## Task 3: `run.sh` / `run.bat` 接入 ensure_pg
 **Files:**
 - Modify: `run.sh`(在 `alembic upgrade head` 前插一步)
 - Modify: `run.bat`(同上)
 - [ ] **Step 1: 改 `run.sh`**
 在 `mkdir -p data` 之后、`"$PY" -m alembic upgrade head` 之前插入:
 ```bash
 "$PY" -m scripts.ensure_pg   # 确保本地 Docker PostgreSQL 就绪(没起会自动拉起;失败即退出)
 ```
 (`set -e` 已在文件顶部,ensure_pg 失败会自动终止脚本。)
 - [ ] **Step 2: 改 `run.bat`**
 在 `if not exist data mkdir data` 之后、`call "%PY%" -m alembic upgrade head` 之前插入:
 ```bat
 REM 确保本地 Docker PostgreSQL 就绪(没起会自动拉起 Docker + PG 容器)
 call "%PY%" -m scripts.ensure_pg
 if errorlevel 1 (
    echo [X] ensure_pg failed ^(PostgreSQL 未就绪^)
    exit /b %errorlevel%
 )
 ```
 - [ ] **Step 3: 验证 `run.sh`(PG 已在跑,应短路后继续 alembic + uvicorn)**
 Run(Git Bash):`bash run.sh 8770`
 Expected: 依次出现 `[ensure_pg] ✅ PostgreSQL 已在 localhost:5432 运行` → alembic 无报错 → uvicorn `Application startup complete`。`Ctrl-C` 停。
 - [ ] **Step 4: 验证 `run.bat`(同上,Windows 原生)**
 Run(cmd/PowerShell):`.\run.bat 8770`
 Expected: 同 Step 3。`Ctrl-C` 停。
 - [ ] **Step 5: Commit**
 ```bash
 git add run.sh run.bat
 git commit -m "feat(dev): run.sh/run.bat 启动前确保 Docker PostgreSQL 就绪"
 ```
 ---
 ## Task 4: `.env.example` 默认切 PostgreSQL
 **Files:**
 - Modify: `.env.example`(第 8-10 行「数据库」段)
 - [ ] **Step 1: 改 `.env.example` 的 DATABASE_URL**
 把:
 ```ini
 # ===== 数据库 =====
 # SQLite 本地文件路径。生产环境用 /opt/shaguabijia-app-server/data.db
 DATABASE_URL=sqlite:///./data/app.db
 ```
 改成:
 ```ini
 # ===== 数据库 =====
 # 本地开发/测试统一用 Docker PostgreSQL:run.bat/run.sh 会自动拉起容器
 # (docker-compose.yml + scripts/ensure_pg.py)。详见 docs/database/postgres-migration.md。
 # 生产用原生 PG,由 scripts/init_postgres.py 写入强随机密码的连接串。
 # ⚠️ scheme 必须是 postgresql+psycopg://(psycopg3);不要写成 postgresql://(会去找未装的 psycopg2)。
 DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia
 ```
 - [ ] **Step 2: 验证(新 .env 从模板复制后能起服务)**
 Run: `cp .env.example /tmp/env.check && grep '^DATABASE_URL=' /tmp/env.check`
 Expected: `DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia`
 - [ ] **Step 3: Commit**
 ```bash
 git add .env.example
 git commit -m "feat(dev): .env.example 默认 DATABASE_URL 切 Docker PostgreSQL"
 ```
 ---
 ## Task 5: `tests/conftest.py` 切 PostgreSQL 测试库
 **Files:**
 - Modify: `tests/conftest.py`(整体替换:去掉临时 SQLite,改指 `shaguabijia_test` + 调 ensure_pg + fixture 改 drop/create)
 - [ ] **Step 1: 整体替换 `tests/conftest.py`**
 ```python
 """测试用 fixtures。
 测试库用 Docker PG 的 shaguabijia_test(与 dev 业务库 shaguabijia 隔离)。
 顺序(必须):设 test DATABASE_URL(在 import app.* 之前)→ ensure PG 就绪 →
 import app → 建表。持久卷可能残留上次的表 → session 开头先 drop 再 create。
 """
 from __future__ import annotations
 import os
 from collections.abc import Iterator
 # 1) 测试库连接串——必须在 import app.* 之前设好(app.db.session 在 import 期建 engine)
 _TEST_DB_URL = (
    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test"
 )
 os.environ["DATABASE_URL"] = _TEST_DB_URL
 os.environ.setdefault("JWT_SECRET_KEY", "test-secret-please-ignore-this-is-only-for-pytest-not-real")
 os.environ.setdefault("ADMIN_JWT_SECRET", "test-admin-secret-please-ignore-only-for-pytest-not-real")
 os.environ.setdefault("JG_APP_KEY", "test-key")
 os.environ.setdefault("JG_MASTER_SECRET", "test-secret")
 os.environ.setdefault("SMS_MOCK", "true")
 os.environ.setdefault("APP_ENV", "dev")
 os.environ.setdefault("APP_DEBUG", "false")  # 测试不打 SQL 日志
 os.environ.setdefault("WECHAT_APP_ID", "wxtest0000000000")
 os.environ.setdefault("WECHAT_APP_SECRET", "test-secret")
 os.environ.setdefault("WXPAY_MCH_ID", "test-mch")
 os.environ.setdefault("WXPAY_MCH_SERIAL_NO", "test-serial")
 os.environ.setdefault("WXPAY_PUBLIC_KEY_ID", "test-pubkey-id")
 os.environ.setdefault("RATE_LIMIT_ENABLED", "false")
 os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
 os.environ.setdefault("PANGLE_REWARD_SECRET", "test-pangle-secret-only-for-pytest")
 # 2) 保证 Docker PG 就绪 + 测试库存在(必须在 import app.db.session 建 engine 之前)
 from scripts.ensure_pg import ensure
 if not ensure(_TEST_DB_URL):
    raise RuntimeError(
        "测试需要 Docker PostgreSQL 就绪。请确认已装 Docker Desktop;"
        "或先跑一次 run.bat/run.sh 把 PG 拉起,再重试 pytest。"
    )
 import pytest
 from fastapi.testclient import TestClient
 from app.db.base import Base
 from app.db.session import engine
 from app.main import app
@pytest.fixture(scope="session", autouse=True)
 def _setup_db() -> Iterator[None]:
    # 持久卷可能残留上次跑崩后的表/数据 → 先 drop 再 create,保证干净起点
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)
@pytest.fixture()
 def client() -> TestClient:
    return TestClient(app)
 ```
 - [ ] **Step 2: 验证 conftest 能引导 PG 并收集用例(选一个不涉 DB 的测试文件)**
 Run: `pytest tests/test_ensure_pg.py -v`
 Expected: conftest 先打印 `[ensure_pg] ✅ PostgreSQL 已在 localhost:5432 运行`(或拉起过程),随后 12 passed。说明「测试走 PG 引导」链路通、且纯函数测试不受影响。
 - [ ] **Step 3: 验证建表落到 PG 测试库(跑一个 DB 相关用例)**
 Run: `pytest tests/test_invite.py -v`
 Expected: 用例在 `shaguabijia_test` 上建表并执行(可能有个别红,留待 Task 6);关键是不再出现 SQLite 临时文件、engine 连的是 PG。
 - [ ] **Step 4: Commit**
 ```bash
 git add tests/conftest.py
 git commit -m "test(dev): conftest 切 shaguabijia_test(Docker PG),引导+drop/create"
 ```
 ---
 ## Task 6: 全量跑 pytest on PG,逐个修红用例
 > SQLite 宽松、PG 严格,切库会暴露一批真 bug(迁移指南 §2.2 已列)。本任务是**发现驱动**:先跑全量、按类别归因、按下述配方修,直到全绿。修改范围限被测业务/模型代码,不改测试来掩盖真 bug(除非测试本身依赖 SQLite 特性,如秒级时间精度)。
 **Files:**
 - Modify: 视失败而定(常见:`app/models/*.py`、`app/**/repositories/*.py`、少量 `tests/*.py`)
 - [ ] **Step 1: 全量跑,拿到失败清单**
 Run: `pytest -q`
 Expected: 大部分通过;记录所有 FAIL 的用例名与报错文本,按下面类别归因。
 - [ ] **Step 2: 修「naive datetime / 时区」类**
 定位:`git grep -n "utcnow()" app/`。把 `datetime.utcnow()` 改成 `datetime.now(timezone.utc)`(并 `from datetime import timezone`)。
 症状:PG `TIMESTAMPTZ` 与 naive datetime 比较/写入报错或结果错位;`tests/test_cps_admin.py` 已注释过 SQLite 忽略 tzinfo 的行为。
 例:
 ```python
 # 改前
 from datetime import datetime
 ts = datetime.utcnow()
 # 改后
 from datetime import datetime, timezone
 ts = datetime.now(timezone.utc)
 ```
 - [ ] **Step 3: 修「字符串/整数隐式比较」类**
 症状:SQLite 允许 `WHERE phone = 13800138000`(自动转型),PG 直接报类型错。定位报错用例引用的查询,确保比较两侧类型一致(手机号等一律按字符串传参 `:phone`,不要传裸 int)。
 - [ ] **Step 3.5: 修「事务已中止」类**
 症状:某用例后续报 `current transaction is aborted, commands ignored until end of transaction block`,根因是前一句 SQL 出错后业务代码缺 `db.rollback()`/`db.commit()` 边界。补上正确的 commit/rollback。
 - [ ] **Step 4: 修「测试依赖 SQLite 特性」类(仅此类可改测试)**
 症状:测试断言依赖 SQLite 秒级时间精度或 FK 不强制(见 `test_invite.py:328`、`test_compare_harvest.py:153` 的注释)。PG 下时间精度更高/FK 更严——调整测试数据(如手动拉开时间间隔、用合法 FK)使断言在 PG 下成立,不改业务逻辑。
 - [ ] **Step 5: 反复跑到全绿**
 Run: `pytest -q`
 Expected: `N passed`(0 failed)。若仍有红,回到 Step 2-4 继续归因。
 - [ ] **Step 6: Commit**
 ```bash
 git add -A
 git commit -m "fix(db): 测试套件切 PostgreSQL 后修复严格性暴露的用例"
 ```
 ---
 ## Task 7: 文档更新
 **Files:**
 - Modify: `docs/database/postgres-migration.md`(§1 增「本地 Docker 一键起」小节)
 - Modify: `CLAUDE.md`(DB 段注明 dev/test = Docker PG)
 - Modify: `scripts/init_postgres.py`(顶部注释区分生产/本地)
 - [ ] **Step 1: `postgres-migration.md` 在「## 1. 本地起 PG」开头插入推荐做法**
 在 `## 1. 本地起 PG + 跑通空库(半天)` 标题下、`### 1.1 装 PG` 之前插入:
 ```markdown
 ### 1.0 推荐:Docker 一键起(本地开发/测试)
 本地开发不必手动装 PG。已提供 `docker-compose.yml` + `scripts/ensure_pg.py`:
 ```bash
 cp .env.example .env    # DATABASE_URL 默认已是 Docker PG 连接串
 ./run.sh                # 或 run.bat;会自动:探测 PG → 没起则启 Docker → 起 PG 容器 → 建库 → alembic → uvicorn
 pytest                  # conftest 自动引导同一容器的 shaguabijia_test 库
 ```
 容器:`postgres:16-alpine`(名 `shaguabijia-pg`,端口 5432,命名卷 `pgdata` 持久化),
 首启即建业务库 `shaguabijia` 与测试库 `shaguabijia_test`。下面 1.1-1.5 的手动装 PG 步骤仅在
 不用 Docker 时才需要;生产仍走 §4 的原生 PG。
 ```
 - [ ] **Step 2: `CLAUDE.md` DB 段补充**
 找到 DB 相关行(`**Prod**: PostgreSQL — just change DATABASE_URL...`)所在段,在其上方加一行:
 ```markdown
 - **Dev/Test**: Docker PostgreSQL 16 — `run.sh`/`run.bat` 经 `scripts/ensure_pg.py` + `docker-compose.yml` 自动拉起;`.env.example` 默认即 PG 连接串;pytest 用同容器的 `shaguabijia_test` 库。**本地不再用 SQLite**。
 ```
 - [ ] **Step 3: `scripts/init_postgres.py` 顶部注释区分场景**
 把模块 docstring 第一行下方(`新机器初始化用。前置:...` 那行)改为:
 ```python
 新机器初始化用(面向【生产原生 PG】:apt/systemd 装好的 PostgreSQL)。
 本地开发/测试请改用 docker-compose.yml + scripts/ensure_pg.py(run.sh/run.bat 自动拉起),不必跑本脚本。
 前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
 ```
 - [ ] **Step 4: 验证无坏链接/格式**
 Run: `git diff --stat`
 Expected: 三个文档文件有改动,无其他文件被误改。
 - [ ] **Step 5: Commit**
 ```bash
 git add docs/database/postgres-migration.md CLAUDE.md scripts/init_postgres.py
 git commit -m "docs(dev): 记录本地 Docker PostgreSQL 用法,区分生产原生 PG 路径"
 ```
 ---
 ## 完成标准(对齐 spec §8 验收)
 - [ ] 全新机器(装了 Docker Desktop、`.env` 从 `.env.example` 复制)跑 `run.bat`/`run.sh` 全自动拉起 PG 并起服务,无手动装 PG。
 - [ ] `docker ps` 见 `shaguabijia-pg` healthy;`shaguabijia` 与 `shaguabijia_test` 两库都在。
 - [ ] PG 已在跑时再跑 `run`,ensure_pg 秒过短路。
 - [ ] `pytest -q` 全绿(连 `shaguabijia_test`)。
 - [ ] `.env` 改回 sqlite 时,`python -m scripts.ensure_pg` 硬失败并打印正确 PG 串。
 - [ ] 能在 `admin/repositories` 写一段 PG 专有聚合(如 `count(*) FILTER (WHERE ...)`),`run` 手动跑通且相应 pytest 通过。
 ## Self-Review 记录(计划作者已核)
 - **Spec 覆盖:** spec §9 待实现清单 8 项 → Task 1(compose+initdb)、Task 2(ensure_pg)、Task 3(run 接线)、Task 4(.env.example)、Task 5(conftest)、Task 6(修红用例)、Task 7(文档);「session.py 无需改」在 Header 与 spec §4.7 说明;「data/ 忽略」在 Task 1 Step 5 验证。无遗漏。
 - **占位符:** 全部步骤含真实代码/命令/期望输出。Task 6 是发现驱动,已用「类别+具体转换配方+定位命令」代替不可预知的逐条 diff——非占位。
 - **类型/命名一致:** `ensure(database_url=None)` 签名在 Task 2 定义,Task 5 以 `ensure(_TEST_DB_URL)` 调用一致;库名/用户/密码/端口全程为 Header 固定值;compose 服务名 `postgres` 与 ensure_pg `COMPOSE_SERVICE` 一致;`shaguabijia_test` 在 initdb SQL、`_ensure_test_db()`、conftest 三处一致。
@@ -1,281 +0,0 @@
 # 本地开发切 Docker PostgreSQL —— 设计文档
 > 让本地开发与测试统一跑在 Docker 化的 PostgreSQL 上,彻底退掉 SQLite。
 > `run.bat` / `run.sh` 启动时自动检测本机 PG,没起就拉起 Docker → 起 PG 容器(镜像缺失先拉),
 > 目的是让开发/大模型能放心用 PG 专有的高效聚合函数,不再为兼容 SQLite 而退化成"取基础数据后内存聚合"。
 >
 > 状态:已定稿(待用户复核)。作者对话日期:2026-07-08。
 > 关联:[postgres-migration.md](../../database/postgres-migration.md)(切引擎完整步骤)、`scripts/init_postgres.py`(生产原生 PG 初始化)。
 > ⚠️ **2026-07-27 增补(见 §10)**:D4 已从「sqlite 硬失败」松为「显式 SQLite 逃生舱」。§1-9 描述的是初版「彻底退掉 SQLite」设计;凡涉及「无 Docker / DATABASE_URL 是 sqlite 时如何处理」,**以 §10 为准**(测试仍只跑 PG 不变)。
 ---
 ## 1. 背景与目标
 ### 问题
 当前开发环境默认用 SQLite(`DATABASE_URL=sqlite:///./data/app.db`),生产用 PostgreSQL 16。两套引擎并存,导致写数据访问代码时(尤其 `app/admin/repositories/` 的报表聚合)为了"两边都能跑",放弃 PG 专有能力(窗口函数、`FILTER`、`JSONB` 操作符、`GROUPING SETS` 等),改成"先查基础数据、再在 Python 内存里聚合"——既慢又啰嗦。
 ### 目标
 本地开发与测试都跑在 PG 上,SQLite 退出本地开发闭环。之后写 PG 专有 SQL 时:
 - 开发运行时(`run.bat`/`run.sh`)直接连 PG,手动验证可行;
 - `pytest` 也连 PG,PG 专有 SQL 在被测代码路径里也安全,不会因 SQLite 而挂——**这是"双库兼容代码彻底消失"的必要条件**。
 ### 非目标(本期不做)
 - 不动**生产**部署(生产仍是原生 PG16 + systemd,无 Docker;`init_postgres.py` 保持不变)。
 - 不做数据搬迁(MVP 阶段无真实用户数据,详见迁移指南背景假设)。
 - 不接 CI(仓库当前无 `.github/workflows`;若将来加 CI,再单独让 CI 起 PG service)。
 - 不引入 Redis / testcontainers / 连接池中间件。
 ---
 ## 2. 决策记录(本次对话已拍板)
 | # | 决策点 | 结论 | 理由 |
 |---|---|---|---|
 | D1 | PG 覆盖范围 | **dev 运行 + 测试都切 PG** | 只切运行时的话,被 SQLite 测试覆盖的代码路径(如 `test_cps_admin.py` 覆盖的 `admin/repositories/cps.py`)仍不能用 PG 专有 SQL,双库代码不会真正消失 |
 | D2 | 打包方式 | **方案 A:Compose + `scripts/ensure_pg.py`** | 唯一真正需要定制的部分(启动 Docker 守护进程、等 PG 就绪)集中到一个跨平台模块,`run.bat`/`run.sh`/`conftest.py` 共用;声明式的容器/卷/健康检查交给 Compose |
 | D3 | 宿主端口 | **5432**(与生产/文档一致) | 边界:若本机已有原生 PG 占 5432,`ensure_pg` 会探测到"PG 已在"直接复用它(可能连到不带业务库的实例)——见 §7 风险,文档提示 |
 | D4 | dev 下 `DATABASE_URL` 仍是 sqlite | **硬失败**(打印一行 fix 后非 0 退出) | 彻底断掉 SQLite 退路,符合"让大家都用 PG"的目标 |
 | D5 | dev 数据库密码 | 固定 `shaguabijia_dev_pw`,写进 compose + `.env.example` | 本地容器仅绑 `localhost`,非机密;保证 `.env.example` 复制即可用。生产密码另由 `init_postgres.py` 强随机生成,不复用 |
 | D6 | 改哪些启动脚本 | `run.bat` 与 `run.sh` **都改** | 仓库一贯保持两者同步 |
 | D7 | 镜像 | `postgres:16-alpine` | 对齐生产 PG16;alpine 体积小 |
 ---
 ## 3. 现状(改动前)
 - **配置**:`app/core/config.py` `DATABASE_URL` 默认 `sqlite:///./data/app.db`,pydantic-settings 从 `.env` 读(环境变量优先级高于 `.env` 文件)。
 - **引擎**:`app/db/session.py` 用 `_is_sqlite = DATABASE_URL.startswith("sqlite")` 分流——SQLite 加 `check_same_thread=False`、不建池;非 SQLite 加 `pool_size=10/max_overflow=20/pool_recycle=3600`。**已天然支持 PG,无需改。**
 - **迁移**:`alembic/env.py` 从 `settings.DATABASE_URL` 读连接串,`render_as_batch` 仅对 sqlite 开;PG 下自动关。**无 psycopg2 硬编码,切 PG 无需改。**
 - **驱动**:`pyproject.toml` 已装 `psycopg[binary]>=3.1`(psycopg3)。URL scheme 必须 `postgresql+psycopg://`(裸 `postgresql://` 会被 SQLAlchemy 路由到未安装的 psycopg2 → ModuleNotFoundError)。
 - **测试**:`tests/conftest.py` 在 import app 前把 `DATABASE_URL` 设成临时文件 SQLite;session 级 autouse fixture 做 `Base.metadata.create_all(engine)` / 结束 `drop_all`(schema 来自 model 而非 alembic,无逐用例 rollback,全会话共享一个库)。
 - **启动脚本**:`run.bat` / `run.sh` 均为:校验 `.env` 存在 → `mkdir data` → `alembic upgrade head` → uvicorn 监听 `0.0.0.0:8770`(`.sh` 带 `--reload --reload-dir app`)。
 - **现有 PG 资产**:`scripts/init_postgres.py`(交互式:建用户/建库/授权/写 .env/跑迁移,面向**已装好的原生 PG**)、`docs/database/postgres-migration.md`(切引擎完整步骤,含 §2 测试切 PG、§2.2 会暴露的真 bug 清单)。
 - **CI**:无(`.github/workflows` 不存在),故测试切 PG 无 CI 联动负担。
 ---
 ## 4. 方案详解
 ### 4.1 新增 `docker-compose.yml`(app-server 根目录)
 ```yaml
 services:
  postgres:
    image: postgres:16-alpine
    container_name: shaguabijia-pg
    environment:
      POSTGRES_USER: shaguabijia_app
      POSTGRES_PASSWORD: shaguabijia_dev_pw
      POSTGRES_DB: shaguabijia
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./docker/initdb:/docker-entrypoint-initdb.d:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U shaguabijia_app -d shaguabijia"]
      interval: 3s
      timeout: 3s
      retries: 20
 volumes:
  pgdata:
 ```
 - `POSTGRES_USER` 设定后,该用户以超级用户身份创建并拥有 `POSTGRES_DB`,故能再建测试库。
 - 首启 initdb 脚本建测试库(见 4.2)。命名卷 `pgdata` 让数据跨重启留存。
 ### 4.2 新增 `docker/initdb/01-create-test-db.sql`
 ```sql
 -- 仅在 pgdata 卷首次初始化时执行一次。以 shaguabijia_app(超级用户)连 shaguabijia 库运行。
 CREATE DATABASE shaguabijia_test OWNER shaguabijia_app;
 ```
 ### 4.3 新增 `scripts/ensure_pg.py`(纯标准库 + docker CLI,跨平台)
 对外同时暴露**可导入函数** `ensure()`(供 `conftest.py` 直接调)和 **CLI 入口** `if __name__ == "__main__": sys.exit(0 if ensure() else 1)`(供 `run` 以 `python -m scripts.ensure_pg` 跑)。`ensure()` 从 `app.core.config.settings` 读 `DATABASE_URL`,解析 host/port,主流程:
 1. **sqlite 守卫**:若 `DATABASE_URL` 以 `sqlite` 开头 → 打印"dev 已切 PG,请把 .env 的 DATABASE_URL 改成 `postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia`"→ 非 0 退出(D4)。
 2. **TCP 探测** `host:port`(stdlib `socket`,超时 1s)。通 → 打印"✅ PG 已就绪"直接返回(幂等:PG 已在跑时开销≈一次握手)。
 3. 不通 → `docker version` 探 CLI;缺失 → 中文报错"请先安装 Docker Desktop:https://www.docker.com/products/docker-desktop/" → 非 0 退出。
 4. `docker info` 探守护进程;不通 → 按平台启动:
   - Windows:`start "" "%ProgramFiles%\Docker\Docker\Docker Desktop.exe"`(找不到则报错让用户手动开)
   - macOS:`open -a Docker`
   - Linux:不自动 sudo,打印 `sudo systemctl start docker` 让用户执行后重试
   然后轮询 `docker info` 直到就绪或超时(默认 120s,每 3s 一次,打印进度)。
 5. `docker compose up -d`(Compose 在镜像缺失时**自动拉取**,首用拉 alpine ~90MB;有进度输出)。
 6. 轮询 healthcheck(`docker inspect` 的 health 状态)/ TCP 直到 PG 接受连接(默认 60s 超时)。
 7. **幂等确保测试库存在**(兼容"老 pgdata 卷没跑过 initdb"的情况):
   `docker compose exec -T postgres psql -U shaguabijia_app -tc "SELECT 1 FROM pg_database WHERE datname='shaguabijia_test'"`,不存在则 `CREATE DATABASE shaguabijia_test OWNER shaguabijia_app`。
 失败即清晰中文报错 + 非 0 退出,**全程不回退 SQLite**。所有超时可用环境变量覆盖(如 `ENSURE_PG_DOCKER_TIMEOUT`)。
 ### 4.4 `run.bat` / `run.sh` 接线
 在 `alembic upgrade head` **之前**插一行调用,失败即退出:
 - `run.sh`:`"$PY" -m scripts.ensure_pg`(`set -e` 已在,失败自动退出)
 - `run.bat`:`call "%PY%" -m scripts.ensure_pg` + `if errorlevel 1 exit /b 1`
 其余逻辑不动(`mkdir data` 保留给 media 等落盘目录)。
 ### 4.5 `.env.example` 默认切 PG
 把
 ```ini
 DATABASE_URL=sqlite:///./data/app.db
 ```
 改为
 ```ini
 # 本地开发/测试统一用 Docker PG(run.bat/run.sh 会自动拉起容器;详见 docs/database/postgres-migration.md §本地 Docker 一键起)。
 # 生产用原生 PG,由 scripts/init_postgres.py 写入强随机密码的连接串。
 DATABASE_URL=postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia
 ```
 ### 4.6 `tests/conftest.py` 切 PG
 调整顶部顺序(仍必须在 `import app.*` 之前完成 env 设定):
 1. 设 `os.environ["DATABASE_URL"] = "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test"`(测试库,永不碰 dev 业务库)。
 2. 调 `scripts.ensure_pg.ensure()`(保证容器在 + 测试库在;PG 已在时几乎零开销)。
 3. `import app...`。
 session 级 autouse fixture:改为 **`Base.metadata.drop_all(engine)` → `create_all(engine)`(开头先清干净,防持久卷里上一次跑残留的表/数据)→ yield → 结束 `drop_all`**;删掉临时 SQLite 文件相关代码。
 > 预期:部分用例会因 PG 的严格性变红(SQLite 宽松、PG 严格),按迁移指南 §2.2 逐个修——常见为:字符串/整数隐式比较、`datetime.utcnow()` naive vs `TIMESTAMPTZ`、事务边界(`current transaction is aborted`)。这既是工作量也是本次改造的**直接收益**(暴露真 bug)。实现阶段需为"跑 pytest 并修红用例"单列步骤。
 ### 4.7 `app/db/session.py`
 **无需改动**——`_is_sqlite` 为假时自动走 PG 池化分支。
 ### 4.8 文档
 - `docs/database/postgres-migration.md` 增一节「本地 Docker 一键起 PG(推荐)」,指向 compose + `ensure_pg`,并说明它替代了 §1.1 的手动 brew/apt 装 PG。
 - `CLAUDE.md` 的 DB 段注明:dev/test = Docker PG(`run` 自动拉起);prod = 原生 PG(`init_postgres.py`)。
 - `scripts/init_postgres.py` 顶部注释补一句"本脚本面向生产原生 PG;本地开发用 docker-compose + scripts/ensure_pg"。
 ---
 ## 5. 连接参数汇总
 | 项 | 值 |
 |---|---|
 | 镜像 | `postgres:16-alpine` |
 | 容器名 | `shaguabijia-pg` |
 | 宿主端口 | `5432` |
 | 超级/业务用户 | `shaguabijia_app` |
 | dev 密码 | `shaguabijia_dev_pw`(本地非机密) |
 | 业务库(dev 运行) | `shaguabijia` |
 | 测试库(pytest) | `shaguabijia_test` |
 | dev `DATABASE_URL` | `postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia` |
 | test `DATABASE_URL` | `postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test` |
 | 数据持久化 | 命名卷 `pgdata` |
 ---
 ## 6. 失败处理矩阵(无 SQLite 兜底)
 | 情形 | ensure_pg 行为 |
 |---|---|
 | `DATABASE_URL` 是 sqlite | 打印应改成的 PG 串 → 非 0 退出 |
 | PG 已在跑(TCP 通) | 打印"已就绪" → 返回 0(跳过 docker) |
 | 无 docker CLI | 提示装 Docker Desktop + 官网链接 → 非 0 退出 |
 | docker 守护进程未起 | 尝试按平台启动 Docker Desktop,轮询到就绪;超时则报错 → 非 0 退出 |
 | 镜像缺失 | `docker compose up -d` 自动拉取(不额外处理) |
 | 容器起了但 PG 未 ready | 轮询 healthcheck 到超时;超时报错 → 非 0 退出 |
 | 老 pgdata 卷缺测试库 | 幂等 `CREATE DATABASE shaguabijia_test` |
 ---
 ## 7. 风险与边界
 - **端口占用(原生 PG 撞 5432)**:D3 选了 5432。若开发机已有原生 PG 监听 5432,step 2 的 TCP 探测会判"PG 已在"并复用它——但那个实例可能没有 `shaguabijia`/`shaguabijia_test` 库或用户,后续 `alembic upgrade head` / 测试会报连不上库或认证失败。**缓解**:文档提示"本机别再单独跑原生 PG";报错信息里提示检查是不是撞了原生 PG。
 - **首次启动慢**:首用需 Docker Desktop 冷启(~30–60s)+ 拉镜像(~数十秒~数分钟,视网络)。`ensure_pg` 全程打印进度,超时可配。
 - **`DATABASE_URL` 环境变量优先级**:pydantic-settings 里 shell 环境变量优先于 `.env`。若开发者 shell 残留旧的 `DATABASE_URL`(如指向 sqlite),会盖过 `.env`。sqlite 守卫(D4)能挡住 sqlite 残留;但若残留的是另一个 PG 串,则以它为准——文档提示。
 - **持久卷脏状态**:测试用 drop_all→create_all 开头清库,避免上次崩溃残留污染;dev 业务库随卷留存(符合预期)。
 - **Docker 未安装/公司网络拉镜像受限**:硬失败并给出明确指引;不提供 SQLite 退路是刻意选择(D4/目标)。
 ---
 ## 8. 验收标准
 1. 全新机器(装了 Docker Desktop、`.env` 从 `.env.example` 复制)执行 `run.bat`(或 `run.sh`):自动拉起 Docker→起 PG 容器→建库→`alembic upgrade head`→uvicorn 起在 8770,无手动装 PG 步骤。
 2. `docker ps` 见 `shaguabijia-pg` 健康;`psql`/客户端能连 `shaguabijia` 与 `shaguabijia_test` 两个库。
 3. PG 已在跑时再次 `run`,`ensure_pg` 秒过(不重复拉容器)。
 4. `pytest` 连 `shaguabijia_test` 跑;红用例全部修绿(PG 严格性暴露的问题)。
 5. `.env` 的 `DATABASE_URL` 改回 sqlite 时,`run`/`pytest` 硬失败并打印正确的 PG 串。
 6. 能在 `admin/repositories/` 里写一段 PG 专有聚合 SQL(如带 `FILTER (WHERE ...)` 的聚合),`run` 下手动跑通、相应 pytest 也通过——即"双库兼容负担消失"的实证。
 ---
 ## 9. 待实现清单(供 writing-plans 拆解)
 - [ ] 新增 `docker-compose.yml`
 - [ ] 新增 `docker/initdb/01-create-test-db.sql`
 - [ ] 新增 `scripts/ensure_pg.py`(TCP 探测 / 启 Docker Desktop 轮询 / compose up / 等 healthy / 幂等建测试库 / sqlite 守卫)
 - [ ] `run.sh`、`run.bat` 接入 `ensure_pg`
 - [ ] `.env.example` 的 `DATABASE_URL` 切 PG
 - [ ] `tests/conftest.py` 切 `shaguabijia_test` + 调 `ensure_pg` + fixture 改 drop/create
 - [ ] 跑 `pytest`,按迁移指南 §2.2 修红用例
 - [ ] 文档:`postgres-migration.md` 增「本地 Docker 一键起」节;`CLAUDE.md` DB 段;`init_postgres.py` 注释
 - [ ] `.gitignore` 确认 `data/` 已忽略(compose 用命名卷,不落项目目录,无需额外忽略)
 ---
 ## 10. 增补(2026-07-27):D4 反转 —— 显式 SQLite 逃生舱
 > 背景:§2 的 D4 定为「dev 下 `DATABASE_URL` 仍是 sqlite → 硬失败」,目的是彻底断掉 SQLite 退路。实践中这对「本机装不了 Docker」的开发者过于刚性——直接被卡死、连跑都跑不起来。本次(2026-07-27 对话)把 D4 从「硬失败」松成「**显式逃生舱**」:工具**从不替你静默切库**,但会在没 Docker 时告诉你怎么手动降级,且降级时每次启动都醒目告警。
 ### 10.1 决策更新
 | # | 原决策 | 新决策 | 理由 |
 |---|---|---|---|
 | D4′ | dev sqlite URL → 硬失败退出 | **放行 + 每次打印醒目降级横幅**(仍非静默) | 已手动改 `.env`=sqlite = 开发者的显式选择,尊重它;但吼一嗓子防止忘了自己在降级、把 PG 专有 SQL 提交上去 |
 | D8(新) | (无) | 无 docker CLI 时,报错里**追加逃生舱指路**(改 `.env`=sqlite),但仍非 0 退出 | 「显式」的关键:工具不替你切库,只指路;开发者改完 `.env` 再跑一次才真正降级 |
 **未变**:D1(测试仍只跑 PG)、D2-D3、D5-D7 全部保留。逃生舱**只作用于 `run.sh`/`run.bat` 运行时**;`pytest` 仍写死连 PG 测试库(`conftest.py` 传 PG URL,sqlite 分支根本不触发),没 Docker 就 `raise`、跑不了完整套件——这正是 D1「测试上 PG 才能暴露真 bug」的初衷,刻意不给逃生舱。
 ### 10.2 代码改动(仅 `scripts/ensure_pg.py` 的 `ensure()`)
 1. **sqlite 分支**(原 `return False`)→ 打印多行降级横幅后 `return True`。横幅点明:PG 专有 SQL/严格类型在此模式**不被验证**、提交前须在有 Docker 的机器上用 PG 复跑、装好 Docker 后把 `DATABASE_URL` 改回 PG 串。
 2. **无 docker CLI 分支**(原仅提示装 Docker + `return False`)→ 追加一句「装不了 Docker?把 `.env` 的 `DATABASE_URL` 改成 `sqlite:///./data/app.db` 可降级运行」;**仍 `return False`**(run 脚本照常退出,开发者需显式改 .env 再跑)。
 3. **常量**:新增 `SQLITE_URL = "sqlite:///./data/app.db"`(逃生舱指路用);`SQLITE_FIX_HINT` 重命名 `PG_URL`(降级横幅"改回 PG"引用)。
 4. 更新模块 docstring 中「全程无 SQLite 兜底」一句,改述为「无 Docker/sqlite URL 时【显式】降级 SQLite(带醒目告警),测试侧不降级」。
 **其余全不动**:`run.sh`/`run.bat`(sqlite 下 `ensure` 返 True → 照常 `alembic upgrade head` + uvicorn)、`docker-compose.yml`、`app/db/session.py`(SQLite 引擎分支本就保留为 fallback)、`tests/conftest.py`、`.env.example`(默认仍 PG)。
 ### 10.3 改完后行为矩阵(覆盖用户列的 5 场景)
 | 场景 | `DATABASE_URL` | ensure_pg 行为 |
 |---|---|---|
 | ① 无 Docker | PG(默认) | 报错 + 指逃生舱 → 退出;开发者改 `.env`=sqlite → 再跑 → **放行 + 降级横幅**,alembic/uvicorn 跑 SQLite |
 | ② 有 Docker 未启动 | PG | 启 Docker Desktop → `compose up` → 等 ready → 建测试库(**不变**) |
 | ③ 有 Docker 已启动 | PG | `compose up` → 等 ready(**不变**) |
 | ④ PG 已在跑 | PG | TCP 通 → 秒过跳过 Docker(**不变**) |
 | ⑤ PG 起来后 | 任意 | run 脚本 `alembic upgrade head`(**不变**;SQLite 走 `render_as_batch`) |
 ### 10.4 风险
 - **降级被忽视**:横幅仅在 `run` 启动时打印一次;若开发者用 IDE 直接起 uvicorn(绕过 run 脚本)则看不到。缓解:横幅足够醒目 + 文档强调;**不**引入 app 启动期重复告警(YAGNI)。
 - **测试无 Docker 跑不了**:刻意保留(D1)。文档提示无 Docker 者:要么装 Docker 跑全量测试,要么只在 CI/有 Docker 的机器上验证 PG 相关改动。
 ### 10.5 Redis 前瞻(不在本次)
 §2 未涉及 Redis。②③ 场景未来若加 Redis 实例:在 `docker-compose.yml` 增 `redis` 服务即可,`docker compose up -d` 天然带起;仅当启动期有组件依赖 Redis 才需给 `ensure_pg` 加 redis readiness 探测。本次不做,方案对它友好。
 ### 10.6 附带修复:`_docker_cli_ok` 守护进程误判(2026-07-27)
 诊断「装了 Docker Desktop 却报未检测到 docker」时发现的真 bug:`_docker_cli_ok()` 原用 `docker version`
 判断 CLI 是否存在,但该命令**要连 daemon**,守护进程没起时退非零 → 把「Docker 装了但没启动」
 误判成「没装 CLI」,`ensure()` 直接打印"请安装 Docker Desktop"并 `return False`,**绕过了专为需求②
 写的 `_start_docker_daemon()` 自动拉起逻辑**——需求②(有 Docker 未启动 → 自动启动)因此从未真正生效。
 修复:改用 `docker --version`(纯客户端、不连 daemon、退 0)。`_docker_daemon_ok()` 仍用 `docker info`
 (正确,该检查本就依赖 daemon)。实测机器:Docker Desktop 20.10.12 已装但引擎未起,修复前 `_docker_cli_ok()`
 误报 False,修复后 True。
 ### 10.7 附带修复:固定 compose 项目名 + 清理残留同名容器(2026-07-27)
 诊断「`docker compose up` 报 `container name "/shaguabijia-pg" already in use`」时发现的又一 bug:compose
 项目名默认取运行目录 basename,在不同目录/worktree(如 `local-dev-postgres-docker` vs `shaguabijia-app-server`)
 之间切换会各自成一个项目;而 `docker-compose.yml` 写死了 `container_name: shaguabijia-pg`(全局唯一名),
 于是新项目 `up` 时要创建同名容器 → 撞上旧项目留下的那个 → 冲突。副作用:`pgdata` 卷也按项目名分裂
 成 `local-dev-postgres-docker_pgdata` / `shaguabijia-app-server_pgdata`,数据被切成两半。
 修复(均在 `scripts/ensure_pg.py`,`docker-compose.yml` 不动、容器名仍是 `shaguabijia-pg`):
 1. 模块级 `os.environ.setdefault("COMPOSE_PROJECT_NAME", "shaguabijia")` —— 钉死项目名,无论从哪个
   目录/worktree 跑都是同一个项目、同一个卷 `shaguabijia_pgdata`,所有 `docker compose up/exec` 一致。
 2. `_compose_up()` 前置 `_remove_stale_container()`:若存在「同名但不属于本项目」的残留容器,先 `docker rm -f`
   再 up(靠 `docker ps --filter name/label` 判归属;数据在命名卷里,删容器不丢)。旧目录/worktree 留下的
   残留容器就此自动清掉,不需手动干预。
 影响:本次修复后首跑,旧的 `shaguabijia-pg`(属项目 `local-dev-postgres-docker`)会被自动删除、在项目
 `shaguabijia` 下重建,挂载全新的 `shaguabijia_pgdata`(空库,`alembic upgrade head` 重建表)。旧数据仍留在
 `local-dev-postgres-docker_pgdata` 卷里(未删,可恢复);确认不需要后可 `docker volume rm` 清理两个旧卷。
@@ -30,14 +30,7 @@ if not exist .env (
 if not exist data mkdir data
-REM Ensure local Docker PostgreSQL is up (auto-starts Docker + PG container if needed)
+REM Build/upgrade SQLite schema (idempotent; no-op if already at head)
 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
@@ -18,8 +18,7 @@ if [ ! -f .env ]; then
  exit 1
 fi
-mkdir -p data            # 运行期落盘目录(媒体上传等)
+mkdir -p data            # sqlite 文件所在目录
 "$PY" -m scripts.ensure_pg      # 确保本地 Docker PostgreSQL 就绪(没起会自动拉起;失败即退出)
 "$PY" -m alembic upgrade head   # 确保表已建（幂等，已是最新则 no-op）
 # --reload 只盯源码目录 app/:别去监视 logs/(日志写入触发"检测→再写日志"回环)和
@@ -1,50 +0,0 @@
@echo off
 REM Admin backend startup (Windows) - the :8771 peer of run.bat.
 REM
 REM Usage:
 REM   cd shaguabijia-app-server
 REM   run8771.bat
 REM
 REM Runs the ADMIN FastAPI app (app.admin.main:admin_app) on 127.0.0.1:8771 —
 REM a SEPARATE process from run.bat (which runs app.main:app on 8770). The admin
 REM web frontend (Next.js :3001) points at http://localhost:8771. Auto-reload on
 REM code change.
 REM
 REM Prerequisite (first time):
 REM   conda activate pricebot ^&^& pip install -e .
 REM   copy .env.example .env  ^&^& fill JWT_SECRET_KEY
 REM
 REM Tip: shaguabijia-admin-web\start.bat starts user-api(8770) + admin-api(8771)
 REM + frontend(3001) in one go, if you prefer a single command.
 cd /d "%~dp0"
 REM Prefer the project virtualenv (.venv) so we never inherit a wrong
 REM global/conda interpreter. FastAPI<0.115 on Pydantic 2.12 crashes at import
 REM with "'FieldInfo' object has no attribute 'in_'". Falls back to PATH python.
 set "PY=python"
 if exist "%~dp0.venv\Scripts\python.exe" set "PY=%~dp0.venv\Scripts\python.exe"
 if not exist .env (
    echo [X] Missing .env. Run: copy .env.example .env  and fill JWT_SECRET_KEY ^(plus MT_CPS_* if you test Meituan^)
    exit /b 1
 )
 if not exist data mkdir data
 REM Ensure local Docker PostgreSQL is up (auto-starts Docker + PG container if needed)
 call "%PY%" -m scripts.ensure_pg
 if errorlevel 1 (
    echo [X] ensure_pg failed ^(PostgreSQL not ready^)
    exit /b %errorlevel%
 )
 REM Build/upgrade schema (idempotent; no-op if already at head)
 call "%PY%" -m alembic upgrade head
 if errorlevel 1 (
    echo [X] alembic upgrade head failed
    exit /b %errorlevel%
 )
 REM Long-running foreground process. Ctrl+C to stop.
 "%PY%" -m uvicorn app.admin.main:admin_app --host 127.0.0.1 --port 8771 --reload
@@ -1,394 +0,0 @@
 """确保本地 PostgreSQL 就绪(开发/测试统一用 Docker PG)。
 被三处复用:
  - run.sh / run.bat:`python -m scripts.ensure_pg`(CLI,失败退非 0)
  - tests/conftest.py:`from scripts.ensure_pg import ensure; ensure(test_url)`
 流程:读 DATABASE_URL → TCP 探测 → 没起就(必要时启 Docker Desktop)→
 `docker compose up -d` → 等 PG ready → 幂等确保测试库存在。
 运行时(run.sh/run.bat)支持【显式】SQLite 逃生舱:DATABASE_URL 设为 sqlite → 放行并打印
 醒目降级横幅(绝不静默替你切库);无 docker CLI 时报错里也指路该逃生舱。测试侧
 (conftest 传 PG URL)不降级——sqlite 分支不触发,没 PG 直接 raise。详见设计文档 §10。
 生产用原生 PG(scripts/init_postgres.py),不走本模块。
 """
 from __future__ import annotations
 import os
 import shutil
 import socket
 import subprocess
 import sys
 import time
 from collections.abc import Mapping
 from pathlib import Path, PureWindowsPath
 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 _win_docker_desktop_candidates(
    env: Mapping[str, str], docker_cli: str | None
 ) -> list[PureWindowsPath]:
    """Windows 上 Docker Desktop.exe 的候选路径(按优先级)。纯函数:不碰文件系统、不读注册表。
    用 PureWindowsPath 解析,故在任意 OS 上跑单测都按 Windows 语义(反斜杠分隔),行为确定。
    优先级:
      ① 环境变量 DOCKER_DESKTOP_EXE 显式指定(终极逃生舱,盘符随你);
      ② 从 PATH 上的 docker CLI 反推——Docker Desktop 的 CLI 在
         <安装目录>\\resources\\bin\\docker.exe,往上几级即安装目录,天然跟随实际盘符
         (装在 D 盘就反推出 D 盘,不再写死 C 盘);多取几级容忍未来目录布局微调;
      ③ 各 Program Files 变体下的标准安装路径兜底(覆盖常规 C 盘装)。
    调用方按序取第一个真实存在的。
    """
    out: list[PureWindowsPath] = []
    override = (env.get("DOCKER_DESKTOP_EXE") or "").strip().strip('"')
    if override:
        out.append(PureWindowsPath(override))
    if docker_cli:
        for parent in list(PureWindowsPath(docker_cli).parents)[:4]:
            out.append(parent / "Docker Desktop.exe")
    for var in ("ProgramFiles", "ProgramW6432", "ProgramFiles(x86)"):
        root = env.get(var)
        if root:
            out.append(PureWindowsPath(root) / "Docker" / "Docker" / "Docker Desktop.exe")
    return out
 def _docker_desktop_from_registry() -> Path | None:
    """从注册表尽力取 Docker Desktop.exe 位置(best-effort;非 Windows / 任何异常都当没找到)。
    比路径猜测更权威且完全跟随实际盘符。探两处:
      - App Paths\\Docker Desktop.exe 的默认值(通常就是 exe 全路径);
      - Uninstall\\Docker Desktop 的 InstallLocation(安装目录,需再拼 exe 名)。
    """
    try:
        import winreg
    except ImportError:  # 非 Windows
        return None
    probes = (
        (winreg.HKEY_LOCAL_MACHINE,
         r"SOFTWARE\Microsoft\Windows\CurrentVersion\App Paths\Docker Desktop.exe", "", False),
        (winreg.HKEY_LOCAL_MACHINE,
         r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\Docker Desktop",
         "InstallLocation", True),
    )
    for hive, subkey, value_name, join_exe in probes:
        try:
            with winreg.OpenKey(hive, subkey) as key:
                val, _ = winreg.QueryValueEx(key, value_name)
        except OSError:
            continue  # 键不存在/无权限 → 下一个
        if not val:
            continue
        exe = Path(val) / "Docker Desktop.exe" if join_exe else Path(val)
        if exe.exists():
            return exe
    return None
 def _find_docker_desktop_exe() -> Path | None:
    """Windows 上尽力定位【真实存在】的 Docker Desktop.exe;遍历候选 + 注册表兜底,找不到返回 None。"""
    for cand in _win_docker_desktop_candidates(os.environ, shutil.which("docker")):
        if Path(cand).exists():
            return Path(cand)
    return _docker_desktop_from_registry()
 def _docker_desktop_cmd(platform: str) -> list[str] | None:
    """按平台给出启动 Docker Desktop 的命令。
    Windows:智能定位 exe(见 _find_docker_desktop_exe),找不到 → None。
    macOS:交给 `open -a Docker`。Linux:None(daemon 需 sudo,让用户手动)。
    """
    if platform.startswith("win"):
        exe = _find_docker_desktop_exe()
        return [str(exe)] if exe else None
    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)
    if cmd is None:
        if sys.platform.startswith("win"):
            # docker CLI 在 PATH 上(否则走不到这)、却定位不到 Docker Desktop.exe:多为非标准安装位置
            _log("找不到 Docker Desktop.exe(已试:PATH 上 docker CLI 反推、注册表、常见安装目录)。")
            _log("  确已安装 → 设环境变量 DOCKER_DESKTOP_EXE=<Docker Desktop.exe 全路径> 再重试,"
                 "或先手动启动 Docker Desktop。")
            _log(f"  不想折腾 → 把 .env 的 DATABASE_URL 改成 {SQLITE_URL} 可降级用 SQLite 跑(仅救急)。")
        else:
            _log("Docker 守护进程未运行。Linux 请手动:sudo systemctl start docker,然后重试。")
        return False
    _log(f"启动 Docker Desktop(首次冷启可能 30-60s):{cmd[0]}")
    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)
@@ -1,8 +1,6 @@
 """Bootstrap PostgreSQL: 建用户 + 建库 + 授权 + 写 .env + 跑迁移。
-新机器初始化用(面向【生产原生 PG】:apt/systemd 装好的 PostgreSQL)。
+新机器初始化用。前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
 本地开发/测试请改用 docker-compose.yml + scripts/ensure_pg.py(run.sh/run.bat 自动拉起),不必跑本脚本。
 前置:已装 PostgreSQL 16 + 知道 postgres 超级用户密码。
 用法:
    python scripts/init_postgres.py
@@ -1,19 +1,22 @@
 """测试用 fixtures。
-测试库用 Docker PostgreSQL 的 shaguabijia_test(与 dev 业务库 shaguabijia 隔离)。
+测试 DB 用临时文件 SQLite。
-顺序(必须):设 test DATABASE_URL(在 import app.* 之前)→ ensure PG 就绪 + 测试库存在 →
+  - 不用 in-memory:in-memory 默认 per-connection,跨连接看不到表。
-import app → 建表。持久卷可能残留上次的表 → session 开头先 drop 再 create。
+  - 用临时文件保证 SessionLocal 每次新连都看到同一份 schema。
 顺序:set env(必须在 import app.* 之前) → import app → 建表 → TestClient
 """
 from __future__ import annotations
 import os
 import tempfile
 from collections.abc import Iterator
-# 1) 测试库连接串——必须在 import app.* 之前设好(app.db.session 在 import 期就建 engine)
+# 临时 db 文件路径,进程退出后清理
-_TEST_DB_URL = (
+_tmp_db = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
-    "postgresql+psycopg://shaguabijia_app:shaguabijia_dev_pw@localhost:5432/shaguabijia_test"
+_tmp_db.close()
-)
+
-os.environ["DATABASE_URL"] = _TEST_DB_URL
+os.environ["DATABASE_URL"] = f"sqlite:///{_tmp_db.name}"
 os.environ.setdefault("JWT_SECRET_KEY", "test-secret-please-ignore-this-is-only-for-pytest-not-real")
 os.environ.setdefault("ADMIN_JWT_SECRET", "test-admin-secret-please-ignore-only-for-pytest-not-real")
 os.environ.setdefault("JG_APP_KEY", "test-key")
@@ -32,18 +35,6 @@ os.environ.setdefault("RATE_LIMIT_ENABLED", "false")  # 限流内存计数会跨
 os.environ.setdefault("PANGLE_CALLBACK_ENABLED", "true")
 os.environ.setdefault("PANGLE_REWARD_SECRET", "test-pangle-secret-only-for-pytest")
 # 2) 保证 Docker PG 就绪(必须在 import app.db.session 建 engine 之前)。
 #    ensure() 在「端口已通」时会短路、不建测试库,故随后再显式补一刀 ensure_test_db()。
 from scripts.ensure_pg import ensure, ensure_test_db
 if not ensure(_TEST_DB_URL):
    raise RuntimeError(
        "测试需要 Docker PostgreSQL 就绪。请确认已装 Docker Desktop;"
        "或先跑一次 run.bat/run.sh 把 PG 拉起,再重试 pytest。"
    )
 # best-effort 兜底建测试库(PG 已在跑但测试库缺失=老卷)。真缺库时下面 create_all 会明确报错。
 ensure_test_db()
 import pytest
 from fastapi.testclient import TestClient
@@ -54,11 +45,13 @@ from app.main import app
@pytest.fixture(scope="session", autouse=True)
 def _setup_db() -> Iterator[None]:
    # 持久卷可能残留上次跑崩后的表/数据 → 先 drop 再 create,保证干净起点
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)
    try:
        os.unlink(_tmp_db.name)
    except OSError:
        pass
@pytest.fixture()
@@ -219,6 +219,115 @@ def test_user_reward_stats_can_scope_withdrawals_by_account(
    assert invite.json()["cash_balance_cents"] == 456
 def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions(
    admin_client: TestClient, admin_token: str
 ) -> None:
    """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。"""
    from app.models.ad_ecpm import AdEcpmRecord
    from app.models.ad_feed_reward import AdFeedRewardRecord
    uid = _seed_user_with_data("13800000024")
    created_at = datetime(2038, 1, 15, 4, tzinfo=UTC)
    db = SessionLocal()
    try:
        db.add_all(
            [
                AdFeedRewardRecord(
                    client_event_id="reward-stats-granted-high",
                    ad_session_id="reward-stats-granted-high",
                    user_id=uid,
                    reward_date="2038-01-15",
                    duration_seconds=10,
                    unit_count=1,
                    ecpm_raw="9000",
                    ad_type="draw",
                    feed_scene="coupon",
                    app_env="prod",
                    our_code_id="104098712",
                    coin=9,
                    status="granted",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-impression-low",
                    app_env="prod",
                    our_code_id="104098712",
                    ecpm_raw="1000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="feed",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-impression-mid",
                    app_env="prod",
                    our_code_id="104098712",
                    ecpm_raw="3000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="comparison",
                    ad_session_id="reward-stats-other-scene",
                    app_env="prod",
                    our_code_id="104098712",
                    ecpm_raw="7000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-test-env",
                    app_env="test",
                    our_code_id="104127529",
                    ecpm_raw="8000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
                AdEcpmRecord(
                    user_id=uid,
                    ad_type="draw",
                    feed_scene="coupon",
                    ad_session_id="reward-stats-non-business",
                    app_env="prod",
                    our_code_id="demo-slot",
                    ecpm_raw="9000",
                    report_date="2038-01-15",
                    created_at=created_at,
                ),
            ]
        )
        db.commit()
    finally:
        db.close()
    response = admin_client.get(
        f"/admin/api/users/{uid}/reward-stats",
        params={
            "date_from": "2038-01-15T00:00:00Z",
            "date_to": "2038-01-15T23:59:59Z",
            "app_env": "prod",
            "revenue_scope": "business",
            "feed_scene": "coupon",
        },
        headers=_auth(admin_token),
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["feed_count"] == 1
    # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。
    assert data["feed_avg_ecpm"] == 2000.0
 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None:
    """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。"""
    naive = datetime(2038, 1, 1, 8, 0)
@@ -18,7 +18,6 @@ from sqlalchemy import select
 from app.db.session import SessionLocal
 from app.models.comparison import ComparisonRecord
 from app.repositories import comparison as crud
 from app.repositories.user import get_user_by_phone
 from app.schemas.compare_record import ComparisonRecordIn
@@ -47,17 +46,6 @@ def _get(db, trace_id: str) -> ComparisonRecord | None:
    ).scalar_one_or_none()
 def _make_user(client, phone: str) -> int:
    """登录建号并返回其真实 user_id。
    PG 强制 comparison_record.user_id → user.id 外键,须引用真实存在的用户;
    用本用例自己登录出的用户,不会与别的用例撞。
    """
    client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
    with SessionLocal() as db:
        return get_user_by_phone(db, phone).id
 # ============================================================
 # repo 层
 # ============================================================
@@ -114,6 +102,7 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
        assert rec.is_source_best is False
        assert rec.store_name == "测试店"
        assert rec.information == "美团更便宜"
        assert rec.fail_reason is None               # 成功记录不派生失败原因
        assert rec.items == [{"name": "肥牛饭", "qty": 1}]
        assert rec.trace_url.endswith("/done/")
        # 再来一次(重试 done)→ 已 success,newly_success=False(发奖不重复触发)
@@ -122,6 +111,35 @@ def test_harvest_done_derives_and_newly_success_once(client) -> None:
        assert newly2 is False
 def test_harvest_done_failed_derives_fail_reason(client) -> None:
    """failed 记录:记录级 information 笼统,但 fail_reason 从 platform_results 救出具体原因
    (id 3030 型:美团系统失败 + 京东 items_not_found → 展示京东那条)。"""
    tid = _tid()
    done_failed = {
        "comparison_results": [
            {"platform_id": "taobao_flash", "platform_name": "淘宝闪购",
             "package": "com.taobao.taobao", "price": 23.04, "is_source": True, "rank": 1,
             "items": [{"name": "肥牛饭", "qty": 1}]},
        ],
        "platform_results": {
            "taobao_flash": {"is_source": True, "status": "source", "price": 23.04},
            "meituan_waimai": {"is_source": False, "status": "failed",
                               "reason": "搜索店铺失败, 无法跳转到搜索页"},
            "jd_waimai_standalone": {"is_source": False, "status": "items_not_found",
                                     "reason": "京东外卖此店内未找到这些菜品"},
        },
        "information": "比价过程出错，请稍后重试",
    }
    with SessionLocal() as db:
        crud.harvest_running(db, trace_id=tid, user_id=None)
        rec, newly = crud.harvest_done(db, trace_id=tid, user_id=None,
                                       done_params=done_failed)
        assert newly is False                        # 没落成 success
        assert rec.status == "failed"
        assert rec.fail_reason == "京东外卖此店内未找到这些菜品"
        assert rec.information == "比价过程出错，请稍后重试"  # 原文案仍留存
 def test_harvest_abort_cancels_running(client) -> None:
    tid = _tid()
    with SessionLocal() as db:
@@ -155,18 +173,17 @@ def test_harvest_abort_missing_row_returns_none(client) -> None:
 def test_upsert_record_no_downgrade_after_harvest_success(client) -> None:
    """harvest 落 success 后,老客户端 fromFailure 的 cancelled 上报不许把它盖回去。"""
    tid = _tid()
    # PG 强制 comparison_record.user_id → user.id 外键(SQLite 不强制,老写法用合成 id
    # 987654)。用本用例自己登录出的真实用户,既满足外键、又不与别的用例撞。
    uid = _make_user(client, "13800007701")
    with SessionLocal() as db:
        crud.harvest_done(db, trace_id=tid, user_id=None, done_params=_done_params())
        payload = ComparisonRecordIn(
            trace_id=tid, business_type="food", status="cancelled",
            information="用户终止", comparison_results=[],
        )
-        rec = crud.upsert_record(db, user_id=uid, payload=payload)
+        # 用一个不会与顺序自增用户撞的合成 id(SQLite 测试库 FK 不强制;别用小整数,
        # 否则会撞上别的测试 login 出来的真实 user_id → 记录混进那个用户的列表)。
        rec = crud.upsert_record(db, user_id=987654, payload=payload)
        assert rec.status == "success"              # 不降级
-        assert rec.user_id == uid                    # 但补上了 user_id(原为 None)
+        assert rec.user_id == 987654                 # 但补上了 user_id(原为 None)
 # ============================================================
@@ -246,7 +263,9 @@ def test_trace_finalize_harvests_abort(client) -> None:
    with SessionLocal() as db:                        # 先有 running 行(帧0建的)
        crud.harvest_running(db, trace_id=tid, user_id=None)
    p, _cap = _mock_pricebot({"trace_url": "https://price.shaguabijia.com/traces/fin/"})
-    with p:
+    with p, patch(
        "app.api.v1.compare.backfill_comparison_llm_cost"
    ) as backfill:
        r = client.post("/api/v1/trace/finalize",
                        json={"trace_id": tid, "status": "cancelled", "reason": "用户终止"})
    assert r.status_code == 200
@@ -254,6 +273,7 @@ def test_trace_finalize_harvests_abort(client) -> None:
        rec = _get(db, tid)
        assert rec is not None and rec.status == "cancelled"
        assert rec.trace_url.endswith("/fin/")
        backfill.assert_called_once_with(rec.id, tid)
 def test_price_step_binds_user_when_authed(client) -> None:
@@ -72,6 +72,7 @@ def test_backfill_retries_then_persists_cost(monkeypatch):
 def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
    missing_id = _record("llm-repair-missing")
    cancelled_id = _record("llm-repair-cancelled", status="cancelled")
    running_id = _record("llm-repair-running", status="running")
    calls = [
        {
@@ -96,12 +97,15 @@ def test_repair_batch_only_targets_terminal_missing_rows(monkeypatch):
        )
        assert result["repaired"] >= 1
        assert "llm-repair-missing" in seen
        assert "llm-repair-cancelled" in seen
        assert "llm-repair-running" not in seen
        with SessionLocal() as db:
            assert db.get(ComparisonRecord, missing_id).llm_cost_yuan is not None
            assert db.get(ComparisonRecord, cancelled_id).llm_cost_yuan is not None
            assert db.get(ComparisonRecord, running_id).llm_cost_yuan is None
    finally:
        _delete(missing_id)
        _delete(cancelled_id)
        _delete(running_id)
@@ -1,129 +0,0 @@
 """scripts/ensure_pg.py 纯函数单测(不需要 Docker/PG)。"""
 from __future__ import annotations
 import socket
 from scripts.ensure_pg import (
    _docker_desktop_cmd,
    _docker_desktop_from_registry,
    _is_sqlite,
    _parse_host_port,
    _port_open,
    _win_docker_desktop_candidates,
    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_win_candidates_override_wins():
    env = {
        "DOCKER_DESKTOP_EXE": r"X:\custom\Docker Desktop.exe",
        "ProgramFiles": r"C:\Program Files",
    }
    cands = [str(p) for p in _win_docker_desktop_candidates(env, None)]
    assert cands[0] == r"X:\custom\Docker Desktop.exe"
 def test_win_candidates_follow_cli_drive():
    # 核心:docker CLI 在 D 盘 → 反推出 D 盘的 Docker Desktop.exe(不再写死 C 盘)
    env = {"ProgramFiles": r"C:\Program Files"}
    cli = r"D:\Docker\Docker\resources\bin\docker.exe"
    cands = [str(p) for p in _win_docker_desktop_candidates(env, cli)]
    assert r"D:\Docker\Docker\Docker Desktop.exe" in cands
    # 兜底的 C 盘常见路径也仍在
    assert r"C:\Program Files\Docker\Docker\Docker Desktop.exe" in cands
 def test_win_candidates_no_cli_uses_program_files():
    env = {"ProgramFiles": r"C:\Program Files"}
    cands = [str(p) for p in _win_docker_desktop_candidates(env, None)]
    assert cands == [r"C:\Program Files\Docker\Docker\Docker Desktop.exe"]
 def test_docker_desktop_cmd_windows_found(monkeypatch, tmp_path):
    exe = tmp_path / "Docker Desktop.exe"
    exe.write_text("")  # 真实存在
    monkeypatch.setattr("scripts.ensure_pg._find_docker_desktop_exe", lambda: exe)
    assert _docker_desktop_cmd("win32") == [str(exe)]
 def test_docker_desktop_cmd_windows_not_found(monkeypatch):
    monkeypatch.setattr("scripts.ensure_pg._find_docker_desktop_exe", lambda: None)
    assert _docker_desktop_cmd("win32") is None
 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_registry_probe_never_raises():
    # best-effort:无论平台/有无键,只返回 Path 或 None,绝不抛
    r = _docker_desktop_from_registry()
    assert r is None or hasattr(r, "exists")
 def test_ensure_sqlite_escape_hatch(monkeypatch):
    # sqlite 是【显式降级逃生舱】:打印横幅、返回 True,且绝不触碰 docker
    def _boom():
        raise AssertionError("sqlite 分支不应调用 docker")
    monkeypatch.setattr("scripts.ensure_pg._docker_cli_ok", _boom)
    assert ensure("sqlite:///./data/app.db") is True
 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
@@ -0,0 +1,122 @@
 """失败卡展示原因派生(repositories.comparison._derive_fail_display)单元测试。
 用例取自线上真实 failed 记录(platform_results 形态),覆盖:
  - information 具体 → 直出
  - information 笼统 + platform_results 有干净业务结局 → 救援出该原因(id 3030/2964 型)
  - information 笼统 + 仅系统失败(搜索失败等黑话)→ None(端侧品牌兜底,id 3027 型)
  - store_closed / no_delivery 被 pricebot 漏成 status=failed → 按 reason 关键字补判
  - 打烊类脏店名 blob → 统一简短模板
  - platform_results 为空 / 非对象 → None
 """
 from __future__ import annotations
 from app.repositories import comparison as crud
 def test_specific_information_passthrough() -> None:
    # information 本身具体(未达起送/找不到菜等)→ 直出,不看 platform_results
    assert (
        crud._derive_fail_display("淘宝闪购未达起送门槛，可加菜凑单后下单", {})
        == "淘宝闪购未达起送门槛，可加菜凑单后下单"
    )
    assert crud._derive_fail_display("未识别到商品", {}) == "未识别到商品"
 def test_generic_info_rescued_from_items_not_found() -> None:
    # id 3030 型:美团系统失败 + 京东 items_not_found,记录级 information 笼统 → 救出京东那条
    pr = {
        "eleme": {"is_source": True, "status": "source", "price": 23.04},
        "meituan_waimai": {
            "is_source": False, "status": "failed",
            "reason": "搜索店铺失败, 无法跳转到搜索页",
        },
        "jd_waimai_standalone": {
            "is_source": False, "status": "items_not_found",
            "reason": "京东外卖此店内未找到这些菜品",
        },
    }
    assert (
        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
        == "京东外卖此店内未找到这些菜品"
    )
 def test_generic_info_rescued_from_store_not_found() -> None:
    # id 2964 型:美团系统失败 + 京东 store_not_found → 救出京东相似店铺文案
    pr = {
        "taobao_flash": {"is_source": True, "status": "source", "price": 127.98},
        "meituan": {
            "is_source": False, "status": "failed",
            "reason": "比价过程出错，请稍后重试",
        },
        "jd_waimai": {
            "is_source": False, "status": "store_not_found",
            "reason": "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺",
        },
    }
    assert (
        crud._derive_fail_display("比价过程出错，请稍后重试", pr)
        == "未在京东找到「黔珍味·贵州牛肉蘸水健康菜 (望京店)」相似店铺"
    )
 def test_generic_info_pure_system_failure_returns_none() -> None:
    # id 3027 型:唯一目标平台是自动化黑话失败 → 不给用户看 → None(端侧品牌兜底)
    pr = {
        "eleme": {"is_source": True, "status": "source", "price": 18.83},
        "meituan_waimai": {
            "is_source": False, "status": "failed",
            "reason": "搜索店铺失败, 无法跳转到搜索页",
        },
    }
    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) is None
 def test_store_closed_leaked_to_failed_is_rescued_and_cleaned() -> None:
    # 打烊被漏成 status=failed;reason 常带脏店名 blob → 统一简短模板
    pr = {
        "jd_waimai": {"is_source": True, "status": "source", "price": 25},
        "taobao_flash": {
            "is_source": False, "status": "failed",
            "reason": "淘宝闪购「沙胆彪炭炉牛杂煲(...)，蜂鸟准时达，月售300+，起送¥20」本店已休息,无法比价",
        },
    }
    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "门店休息中，无法比价"
    pr2 = {
        "taobao_flash": {"is_source": True, "status": "source", "price": 31.83},
        "meituan": {
            "is_source": False, "status": "failed",
            "reason": "美团「奈雪的茶（北京王府井奥莱·香江」门店已打烊,无法比价",
        },
    }
    assert crud._derive_fail_display("比价出错", pr2) == "门店已打烊，无法比价"
 def test_no_delivery_leaked_to_failed_is_rescued() -> None:
    # 单点不配送被漏成 status=failed;reason 本身干净 → 直接用
    reason = "京东外卖该商家所选商品单点不配送，无法进入结算比价"
    pr = {
        "taobao_flash": {"is_source": True, "status": "source", "price": 20.1},
        "jd_waimai_standalone": {
            "is_source": False, "status": "failed", "reason": reason,
        },
    }
    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == reason
 def test_empty_or_missing_platform_results_returns_none() -> None:
    # 「比价出错」+ 空 {} / None / 非对象:引擎早夭,无可展示原因 → None
    assert crud._derive_fail_display("比价出错", {}) is None
    assert crud._derive_fail_display("比价过程出错，请稍后重试", None) is None
    assert crud._derive_fail_display("比价出错", []) is None  # 老 array 形态,防御
 def test_clean_status_wins_over_priority_order() -> None:
    # 多个业务结局同现时按 _BIZ_STATUS_PRIORITY 选(below_minimum 优先于 store_not_found)
    pr = {
        "src": {"is_source": True, "status": "source", "price": 30},
        "a": {"is_source": False, "status": "store_not_found", "reason": "未找到店铺A"},
        "b": {"is_source": False, "status": "below_minimum", "reason": "B未达起送门槛"},
    }
    assert crud._derive_fail_display("比价过程出错，请稍后重试", pr) == "B未达起送门槛"
Author	SHA1	Message	Date
marco	4bd4e66678	重构比价结果页取数据逻辑 (#195 ) Reviewed-on: #195	2026-07-29 01:59:31 +08:00
linkeyu	90c6fe599a	修复：统一用户Draw信息流eCPM统计口径 (#190 ) ## 问题业务收益详情的平均 Draw eCPM 仅平均成功发奖记录，会排除未发奖的真实展示，导致数值系统性偏高，且与广告收益页口径不一致。 ## 修复 - `feed_avg_ecpm` 改为从 `ad_ecpm_record` 的全部 `draw/feed` 实际展示计算 - 成功发奖、未发奖展示均纳入，每次展示等权 - 日期、正式/测试环境、业务代码位、领券/比价场景支持与广告收益页对齐 - 奖励份数仍基于成功发奖表，不混用展示数据源 - 复用广告收益报表的业务代码位集合 ## 线上数据复算 2026-07-25、正式业务、用户 #33： - 旧口径（只看成功发奖）：`29.9117 元/千次` - 新口径（333 次真实展示）：`19.9926 元/千次` - 新值与广告收益报表一致 ## 验证 - 新增成功/未发奖、场景、环境、业务代码位回归用例 - `tests/test_admin_read.py` + `tests/test_admin_ad_revenue_scope.py`：29 项全通过 - Ruff 改动文件检查通过 ## 上线顺序本 PR 需先于管理后台配套 PR 上线。 --------- Co-authored-by: guke <guke@wonderable.ai> Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #190 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-28 17:58:23 +08:00
linkeyu	e529112a90	修复中途退出比价的 LLM 成本回填 (#191 ) ## 问题比价记录进入中途退出后未触发 LLM 成本回填，周期补偿也未扫描 cancelled，导致实际已有 LLM 调用的记录长期显示成本、LLM、TOKEN 为空。 ## 修改 - finalize 落库后立即追加 LLM 成本回填 - 周期补偿范围加入 cancelled - 保持无有效调用和全调用失败记录不伪造成本 - 增加即时回填和周期补偿回归测试 ## 验证 - ruff 检查通过 - 相关测试 28 项通过 - 全仓 626 项通过；主干既有失败已在未修改的 origin/main 复现 --------- Co-authored-by: unknown <798648091@qq.com> Reviewed-on: #191 Co-authored-by: linkeyu <linkeyu@wonderable.ai> Co-committed-by: linkeyu <linkeyu@wonderable.ai>	2026-07-28 17:57:52 +08:00
guke	50da718e35	比价记录失败卡展示具体原因(新增 fail_reason) (#189 ) 失败记录不再一律「网络开小差」:新增记录级 fail_reason 派生列——information 具体则直出,笼统则从 platform_results 救出业务原因(找不到店/菜、未起送、打烊、单点不配送等),纯系统失败为 None → 端侧品牌兜底。store_closed/no_delivery 被 pricebot 漏成 status=failed 的按 reason 补判,打烊脏店名统一简短模板。接入 harvest_done 与灰度期 upsert_record 两条写路径。 - models: comparison_record.fail_reason 列 - repositories: _derive_fail_display + 补判/清洗 helper,两条写路径接入 - schemas: ComparisonRecordOut 暴露 fail_reason - alembic: 加列 + 回填老 specific 失败记录 - tests: _derive_fail_display 单测(8 例)+ harvest 失败落库集成测试 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: guke <guke@autohome.com.cn> Reviewed-on: #189	2026-07-28 14:04:36 +08:00