shaguabijia-app-server/scripts/seed_mock_price_reports.py

"""一次性 mock:造几条带「真实可加载截图」的上报更低价记录,供后台「上报审核」页验收图片加载。

为什么要真实图片:验证「截图加载」bug 是否修好,必须让 images 指向**真实存在于 /media 的文件**,
否则 404 是「文件不存在」而非「前端取图地址不对」,无法区分。本脚本:
  1. 在 data/media/price_report/ 生成几张纯色 PNG(手写 PNG 字节,无需 Pillow)作为截图;
  2. 造 3 个 mock 用户 + 覆盖 待审核 / 已通过 / 已拒绝 的上报记录,引用这些图。

幂等:重跑先按固定 mock 手机号清掉旧用户/上报 + 删 mock 图再重建;--clean-only 只清。

  python -m scripts.seed_mock_price_reports
  python -m scripts.seed_mock_price_reports --clean-only

看图:前端「上报审核」页(http://localhost:3001 → 上报审核)。图经 NEXT_PUBLIC_MEDIA_BASE
(本地 = http://localhost:8770)由 App 后端 /media 加载——改过 .env.local 后需重启 next dev,
且 App 后端(:8770)要在跑。
"""
from __future__ import annotations

import argparse
import struct
import sys
import zlib
from datetime import datetime, timedelta
from pathlib import Path

from sqlalchemy import delete, select

from app.core import rewards
from app.core.config import settings
from app.db.session import SessionLocal
from app.models.price_report import PriceReport
from app.models.user import User

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")  # Windows 控制台输出中文/¥

# 固定 mock 手机号:脚本只动这些号,便于幂等重建/清理。
MOCK_PHONES = ["13266660001", "13266660002", "13266660003"]

_REPORT_DIR = Path(settings.MEDIA_ROOT) / "price_report"
_MOCK_IMG_GLOB = "mock_pr_*.png"  # 本脚本生成的图前缀,清理时按此删


def _solid_png(width: int, height: int, rgb: tuple[int, int, int]) -> bytes:
    """生成一张纯色 PNG(truecolor RGB)的字节,无需 Pillow。颜色块即可肉眼判断「图加载出来了」。"""
    def _chunk(typ: bytes, data: bytes) -> bytes:
        body = typ + data
        return struct.pack(">I", len(data)) + body + struct.pack(">I", zlib.crc32(body) & 0xFFFFFFFF)

    ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)  # 8bit/通道, color type 2 = RGB
    row = b"\x00" + bytes(rgb) * width  # 每行前缀 filter byte 0
    idat = zlib.compress(row * height, 9)
    return b"\x89PNG\r\n\x1a\n" + _chunk(b"IHDR", ihdr) + _chunk(b"IDAT", idat) + _chunk(b"IEND", b"")


def _write_mock_image(name: str, rgb: tuple[int, int, int]) -> str:
    """写一张 mock 截图到 media/price_report/,返回其相对 URL(/media/price_report/<name>)。"""
    _REPORT_DIR.mkdir(parents=True, exist_ok=True)
    (_REPORT_DIR / name).write_bytes(_solid_png(320, 320, rgb))
    return f"{settings.MEDIA_URL_PREFIX}/price_report/{name}"


def _bj_now() -> datetime:
    """北京 wall-clock(naive),与 report_repo.create_report 一致(前端按北京时间解析,不加 Z)。"""
    return datetime.now(rewards.CN_TZ).replace(tzinfo=None)


def clean(db) -> int:
    uids = list(db.execute(select(User.id).where(User.phone.in_(MOCK_PHONES))).scalars())
    if uids:
        db.execute(delete(PriceReport).where(PriceReport.user_id.in_(uids)))
        db.execute(delete(User).where(User.id.in_(uids)))
        db.commit()
    # 删 mock 截图文件
    if _REPORT_DIR.exists():
        for f in _REPORT_DIR.glob(_MOCK_IMG_GLOB):
            f.unlink(missing_ok=True)
    return len(uids)


def seed(db) -> list[PriceReport]:
    now = _bj_now()

    # 1) 建 3 个 mock 用户
    users: dict[str, User] = {}
    for i, (phone, nickname) in enumerate(
        zip(MOCK_PHONES, ["省钱小王", "比价老李", "薅羊毛阿珍"])
    ):
        u = User(
            phone=phone,
            nickname=nickname,
            register_channel="sms",
            status="active",
            created_at=now - timedelta(days=8 + i),
            last_login_at=now - timedelta(hours=3),
        )
        db.add(u)
        users[phone] = u
    db.flush()  # 拿 user.id

    # 2) 生成 mock 截图(不同颜色块,便于肉眼区分「都加载出来了」)
    palette = [
        (250, 173, 20),   # 橙
        (82, 196, 26),    # 绿
        (24, 144, 255),   # 蓝
        (114, 46, 209),   # 紫
        (245, 34, 45),    # 红
    ]
    imgs = [_write_mock_image(f"mock_pr_{i}.png", palette[i]) for i in range(len(palette))]

    # 3) 造上报记录(覆盖 pending / approved / rejected 三个 tab,每条都带可加载截图)
    def report(
        phone: str,
        store: str,
        dish: str,
        orig_name: str,
        orig_cents: int,
        rep_id: str,
        rep_name: str,
        rep_cents: int,
        images: list[str],
        *,
        status: str = "pending",
        reward_coins: int | None = None,
        reject_reason: str | None = None,
        created: datetime,
    ) -> PriceReport:
        return PriceReport(
            user_id=users[phone].id,
            comparison_record_id=None,  # 可空:不依赖真实比价记录
            store_name=store,
            dish_summary=dish,
            original_platform_id="meituan-waimai",
            original_platform_name=orig_name,
            original_price_cents=orig_cents,
            reported_platform_id=rep_id,
            reported_platform_name=rep_name,
            reported_price_cents=rep_cents,
            images=images,
            status=status,
            reward_coins=reward_coins,
            reject_reason=reject_reason,
            reviewed_at=(created + timedelta(hours=1)) if status != "pending" else None,
            created_at=created,
        )

    reports = [
        # 待审核(默认 tab,重点验图):2 图 + 1 图各一条
        report("13266660001", "瑞幸咖啡(国贸店)", "生椰拿铁 × 2", "美团外卖", 3980,
               "jd-waimai", "京东外卖", 3200, [imgs[0], imgs[1]],
               created=now - timedelta(minutes=8)),
        report("13266660002", "麦当劳(三里屯店)", "巨无霸套餐", "美团外卖", 4200,
               "taobao-shanguang", "淘宝闪购", 3550, [imgs[2]],
               created=now - timedelta(minutes=35)),
        # 已通过(已发金币)
        report("13266660003", "蜜雪冰城(西单店)", "柠檬水 × 3", "美团外卖", 1800,
               "jd-waimai", "京东外卖", 1500, [imgs[3]],
               status="approved", reward_coins=1000, created=now - timedelta(days=1)),
        # 已拒绝(带理由)
        report("13266660001", "星巴克(望京店)", "美式 × 1", "美团外卖", 3000,
               "jd-waimai", "京东外卖", 2800, [imgs[4]],
               status="rejected", reject_reason="截图不清晰", created=now - timedelta(days=2)),
    ]
    db.add_all(reports)
    db.commit()
    for r in reports:
        db.refresh(r)
    return reports


def _yuan(cents: int | None) -> str:
    return "-" if cents is None else f"¥{cents / 100:.2f}"


def main() -> None:
    parser = argparse.ArgumentParser(description="造上报更低价 mock 数据(后台上报审核页验图用)")
    parser.add_argument("--clean-only", action="store_true", help="只清理 mock 数据,不重建")
    args = parser.parse_args()

    db = SessionLocal()
    try:
        removed = clean(db)
        if removed:
            print(f"🧹 已清理旧 mock:{removed} 个用户及其上报记录 + mock 截图")
        if args.clean_only:
            print("✅ 仅清理,已完成。")
            return

        reports = seed(db)
        label = {"pending": "待审核", "approved": "已通过", "rejected": "已拒绝"}
        print(f"\n✅ 已生成 {len(reports)} 条上报记录(截图落 {_REPORT_DIR}):")
        for r in reports:
            print(
                f"   #{r.id} [{label.get(r.status, r.status)}] {r.store_name} | "
                f"{r.original_platform_name}{_yuan(r.original_price_cents)} → "
                f"{r.reported_platform_name}{_yuan(r.reported_price_cents)} | {len(r.images)}图"
            )
        print(
            "\n👉 打开 http://localhost:3001 → 上报审核 查看(默认「待审核」tab)。"
            "\n   图加载不出来时排查:① 是否重启过 next dev(读 .env.local 的 NEXT_PUBLIC_MEDIA_BASE)"
            "  ② App 后端(:8770)是否在跑(它托管 /media)。"
        )
    finally:
        db.close()


if __name__ == "__main__":
    main()