diff --git a/scripts/mock_compare_records.py b/scripts/mock_compare_records.py new file mode 100644 index 0000000..cb8fe61 --- /dev/null +++ b/scripts/mock_compare_records.py @@ -0,0 +1,189 @@ +"""给指定用户灌一批 mock 比价记录(开发 / 真机走查用)。 + +为什么:首页「上次比价」横幅(4 分钟新鲜窗口)、比价记录页、「我的」省钱战绩卡都读 +`comparison_record`。dev 库里这个用户没数据时这些 UI 是空的,本脚本灌一批真实感的外卖 +比价记录,方便端上走查。 + +用法(项目根、已 pip install -e . 的环境): + python scripts/mock_compare_records.py # 默认 user 昵称后缀 DlSKoc5S7, 10 条 + python scripts/mock_compare_records.py --user DlSKoc5S7 # 按 昵称后缀/用户名/邀请码/手机号 + python scripts/mock_compare_records.py --user 5 # 直接按 user.id + python scripts/mock_compare_records.py --user 5 --count 6 + +要点: + - 幂等:trace_id 固定为 mock-<标识>-NN,重跑覆盖同号记录(不会越灌越多)。 + - 第 1 条 created_at = 跑脚本当下(4 分钟内),其余铺在近 7 天;重跑会刷新时间戳 + (所以重跑一次即可让「上次比价」横幅重新进入 4 分钟窗口)。 + - best_* / saved / status 由真实的 repositories.comparison.upsert_record 从 comparison_results 派生。 +""" +from __future__ import annotations + +import argparse +from datetime import datetime, timedelta + +from sqlalchemy import or_, select + +from app.core.rewards import CN_TZ +from app.db.session import SessionLocal +from app.models.user import User +from app.repositories.comparison import upsert_record +from app.schemas.compare_record import ComparisonRecordIn + +# 平台名 → id(展示用;横幅按 platform_name 渲染,id 仅留痕) +_PLATFORM_ID = {"美团": "meituan", "京东": "jd", "淘宝闪购": "taobao", "饿了么": "ele"} + +# 10 单真实感外卖订单:source=美团,targets=(平台, 到手价/元, 优惠来源名|None, 红包额/元)。 +# failed=True 表示目标平台没采到价(targets 价置 None → 派生 status=failed)。 +_CATALOG = [ + dict(store="瑞幸咖啡(国贸店)", items=[("生椰拿铁(大杯)", 2)], + source=("美团", 39.8), + targets=[("淘宝闪购", 29.9, "平台红包", 6.0), ("京东", 33.0, None, 0)]), + dict(store="茶百道(大悦城店)", items=[("杨梅冰萃(大杯)", 1), ("脆波波奶茶", 1)], + source=("美团", 31.0), + targets=[("京东", 24.5, "百亿补贴", 4.5), ("饿了么", 27.9, None, 0)]), + dict(store="麦当劳(西单店)", items=[("板烧鸡腿堡套餐", 1)], + source=("美团", 32.0), + targets=[("京东", 25.9, "神券", 3.0), ("淘宝闪购", 28.0, None, 0)]), + dict(store="肯德基(朝阳大悦城店)", items=[("疯狂星期四全家桶", 1)], + source=("美团", 89.0), + targets=[("京东", 79.9, "百亿补贴", 9.0)]), + dict(store="张亮麻辣烫(双井店)", items=[("自选麻辣烫", 1)], + source=("美团", 35.0), + targets=[("京东", None, None, 0), ("淘宝闪购", None, None, 0)], + failed=True, fail_reason="京东外卖、淘宝闪购均未找到该店"), + dict(store="必胜客(王府井店)", items=[("超级至尊比萨(9寸)", 1), ("香辣鸡翅", 1)], + source=("美团", 78.0), + targets=[("淘宝闪购", 62.0, "跨店满减", 10.0), ("京东", 69.0, None, 0)]), + dict(store="蜜雪冰城(中关村店)", items=[("多肉葡萄", 2), ("冰鲜柠檬水", 1)], + source=("美团", 21.0), + targets=[("京东", 16.5, None, 0)]), + dict(store="海底捞外送(三里屯店)", items=[("番茄锅底", 1), ("鲜毛肚", 2), ("虾滑", 1)], + source=("美团", 168.0), + targets=[("京东", 155.0, "大额神券", 13.0), ("饿了么", 162.0, None, 0)]), + dict(store="华莱士(回龙观店)", items=[("全鸡汉堡套餐", 1)], + source=("美团", 26.0), + targets=[("京东", None, None, 0)], + failed=True, fail_reason="该商品在京东外卖未上架"), + dict(store="星巴克(国贸店)", items=[("燕麦拿铁(大杯)", 2)], + source=("美团", 66.0), + targets=[("饿了么", 58.0, "会员券", 8.0), ("京东", 61.0, None, 0)]), +] + +# 第 2..N 条的 created_at 相对当下的回退量(都 > 4 分钟,确保只有第 1 条落在新鲜窗口)。 +_OFFSETS = [ + timedelta(minutes=25), timedelta(hours=2, minutes=10), timedelta(hours=6), + timedelta(days=1, hours=3), timedelta(days=1, hours=20), timedelta(days=2, hours=9), + timedelta(days=3, hours=14), timedelta(days=5, hours=7), timedelta(days=7, hours=2), +] + + +def _resolve_user(db, ident: str) -> User | None: + """按 id(纯数字)/ 用户名 / 邀请码 / 手机号 / 昵称(或昵称去掉「用户」前缀的后缀)解析用户。""" + if ident.isdigit(): + u = db.get(User, int(ident)) + if u is not None: + return u + return db.execute( + select(User).where( + or_( + User.username == ident, + User.invite_code == ident, + User.phone == ident, + User.nickname == ident, + User.nickname == f"用户{ident}", + User.nickname.like(f"%{ident}"), + ) + ) + ).scalars().first() + + +def _build_results(order: dict) -> tuple[list[dict], str]: + """造 comparison_results + information 文案;返回 (results, information)。""" + src_name, src_price = order["source"] + failed = order.get("failed", False) + rows: list[dict] = [dict( + platform_id=_PLATFORM_ID.get(src_name), platform_name=src_name, + price=src_price, is_source=True, status="success", + )] + for pname, price, cname, csaved in order["targets"]: + row = dict( + platform_id=_PLATFORM_ID.get(pname), platform_name=pname, + price=(None if failed else price), is_source=False, + status=("store_not_found" if failed else "success"), + ) + if not failed and csaved: + row["coupon_saved"] = csaved + row["coupon_name"] = cname + rows.append(row) + + # rank:有价的按升序 1..N;无价的不排 + priced = sorted((r for r in rows if r["price"] is not None), key=lambda r: r["price"]) + for i, r in enumerate(priced, start=1): + r["rank"] = i + + if failed: + return rows, order.get("fail_reason", "目标平台未找到该商品") + best = priced[0] + saved = src_price - best["price"] + if best["is_source"]: + info = f"{src_name}已是最低价,本单未比出更低" + else: + info = f"已在{best['platform_name']}比出最低价 ¥{best['price']:.2f}(比{src_name}省 ¥{saved:.2f})" + return rows, info + + +def main() -> None: + parser = argparse.ArgumentParser(description="给用户灌一批 mock 比价记录") + parser.add_argument("--user", default="DlSKoc5S7", help="user.id / 用户名 / 邀请码 / 手机号 / 昵称后缀") + parser.add_argument("--count", type=int, default=10, help="条数(1..10),默认 10") + args = parser.parse_args() + + count = max(1, min(args.count, len(_CATALOG))) + now = datetime.now(CN_TZ).replace(tzinfo=None) + + db = SessionLocal() + try: + user = _resolve_user(db, args.user) + if user is None: + raise SystemExit(f"❌ 找不到用户:{args.user}(试试 --user )") + print(f"→ 用户 id={user.id} nickname={user.nickname} phone={user.phone}") + + ok = failed = 0 + for idx in range(count): + order = _CATALOG[idx] + src_name, src_price = order["source"] + results, info = _build_results(order) + payload = ComparisonRecordIn( + trace_id=f"mock-{args.user}-{idx:02d}", + business_type="food", + device_id="mock-seed", + store_name=order["store"], + source_platform_id=_PLATFORM_ID.get(src_name), + source_platform_name=src_name, + source_price=src_price, + items=[{"name": n, "qty": q} for n, q in order["items"]], + comparison_results=results, + total_dish_count=sum(q for _, q in order["items"]), + information=info, + app_version="mock", + ) + rec = upsert_record(db, user_id=user.id, payload=payload) + # created_at:第 1 条 = 当下(4 分钟新鲜窗口);其余铺近 7 天。重跑刷新时间戳。 + rec.created_at = now if idx == 0 else now - _OFFSETS[idx - 1] + db.commit() + if rec.status == "success": + ok += 1 + else: + failed += 1 + tag = "🆕now" if idx == 0 else f"-{_OFFSETS[idx - 1]}" + print(f" [{rec.status:7}] id={rec.id} {order['store']} best={rec.best_platform_name} " + f"¥{(rec.best_price_cents or 0)/100:.2f} saved=¥{(rec.saved_amount_cents or 0)/100:.2f} {tag}") + + print(f"✅ 已为 user id={user.id} 写入 {count} 条(success {ok} / failed {failed});" + f"第 1 条 created_at=当下,4 分钟内进 app 可见「上次比价」横幅。") + finally: + db.close() + + +if __name__ == "__main__": + main()