chore(scripts): 新增 mock 比价记录灌库脚本(端上走查用)
给指定用户灌一批真实感外卖比价记录,填充首页「上次比价」横幅 / 比价记录页 / 省钱战绩卡。 trace_id 固定幂等,重跑覆盖同号;第 1 条落在 4 分钟新鲜窗口,其余铺近 7 天。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
guke
@@ -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 <id>)")
|
||||
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()
|
||||
Reference in New Issue
Block a user