"""生成 admin「领券记录」本地联调数据。 用法: python scripts/seed_coupon_session_mock.py 脚本只清理 ``mock-coupon-repeat-*`` 前缀的数据并重新生成。打开后台「领券记录」， 日期选今天；分别切换 prod/dev，可验证同一设备同一天多次领券仍各自显示正确分数。 """ from __future__ import annotations import sys from datetime import UTC, datetime from pathlib import Path from zoneinfo import ZoneInfo from sqlalchemy import delete, select ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.db.session import SessionLocal, engine # noqa: E402 from app.models.coupon_state import ( # noqa: E402 CouponClaimEvent, CouponClaimRecord, CouponSession, ) from app.models.user import User # noqa: E402 from app.repositories.coupon_state import record_claims, today_cn # noqa: E402 PREFIX = "mock-coupon-repeat-" CN_TZ = ZoneInfo("Asia/Shanghai") DEVICE_REPEAT = f"{PREFIX}device" DEVICE_CONTROL = f"{PREFIX}control-device" PHONE = "19900009001" USERNAME = "80000009001" PLATFORM_ELAPSED = { "meituan-waimai": 46_800, "taobao-shanguang": 19_500, "jd-waimai": 24_000, } FIRST_RESULTS = [ {"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "success"}, {"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "success"}, {"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"}, {"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "failed", "reason": "模拟失败"}, {"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"}, {"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"}, {"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "success"}, {"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"}, ] SECOND_RESULTS = [ {"coupon_id": "mt_mock_1", "name": "美团模拟券1", "vendor": "meituan", "status": "skipped", "reason": "模拟跳过，不计分母"}, {"coupon_id": "mt_mock_2", "name": "美团模拟券2", "vendor": "meituan", "status": "already_claimed"}, {"coupon_id": "mt_mock_3", "name": "美团模拟券3", "vendor": "meituan", "status": "success"}, {"coupon_id": "mt_mock_4", "name": "美团模拟券4", "vendor": "meituan", "status": "success"}, {"coupon_id": "tb_mock_1", "name": "淘宝模拟券1", "vendor": "taobao", "status": "success"}, {"coupon_id": "tb_mock_2", "name": "淘宝模拟券2", "vendor": "taobao", "status": "success"}, {"coupon_id": "tb_mock_3", "name": "淘宝模拟券3", "vendor": "taobao", "status": "already_claimed"}, {"coupon_id": "jd_mock_1", "name": "京东模拟券", "vendor": "jingdong", "status": "success"}, ] def _started_at(hour: int, minute: int) -> datetime: local = datetime.combine(today_cn(), datetime.min.time()).replace( hour=hour, minute=minute, tzinfo=CN_TZ ) return local.astimezone(UTC) def _session( *, trace_id: str, device_id: str, user_id: int, app_env: str, hour: int, minute: int, status: str = "completed", elapsed_ms: int | None = 91_900, platform_elapsed: dict[str, int] | None = None, ) -> CouponSession: started_at = _started_at(hour, minute) return CouponSession( trace_id=trace_id, device_id=device_id, user_id=user_id, status=status, app_env=app_env, platforms=[], origin_package=None, device_model="Mock Phone", rom="MockOS 1", started_at=started_at, started_date=today_cn(), finished_at=started_at if status != "started" else None, elapsed_ms=elapsed_ms, platform_elapsed=platform_elapsed, platform_success=( ["meituan-waimai", "taobao-shanguang", "jd-waimai"] if status == "completed" else None ), claimed_count=7 if status == "completed" else 0, ) def main() -> None: # 本地旧库 Alembic 版本链可能未同步；仅为联调补建新事件表，正式环境仍走 migration。 CouponClaimEvent.__table__.create(bind=engine, checkfirst=True) with SessionLocal() as db: db.execute(delete(CouponClaimEvent).where( CouponClaimEvent.trace_id.startswith(PREFIX) )) db.execute(delete(CouponClaimRecord).where( CouponClaimRecord.device_id.startswith(PREFIX) )) db.execute(delete(CouponSession).where( CouponSession.trace_id.startswith(PREFIX) )) user = db.execute(select(User).where(User.phone == PHONE)).scalar_one_or_none() if user is None: user = User( phone=PHONE, username=USERNAME, register_channel="sms", nickname="领券重复测试", ) db.add(user) db.flush() first_trace = f"{PREFIX}dev-first" second_trace = f"{PREFIX}prod-second" abandoned_trace = f"{PREFIX}prod-abandoned" control_trace = f"{PREFIX}prod-control" db.add_all([ _session( trace_id=first_trace, device_id=DEVICE_REPEAT, user_id=user.id, app_env="dev", hour=10, minute=0, platform_elapsed=PLATFORM_ELAPSED, ), _session( trace_id=second_trace, device_id=DEVICE_REPEAT, user_id=user.id, app_env="prod", hour=15, minute=0, platform_elapsed=PLATFORM_ELAPSED, ), _session( trace_id=abandoned_trace, device_id=DEVICE_REPEAT, user_id=user.id, app_env="prod", hour=16, minute=0, status="abandoned", elapsed_ms=21_500, platform_elapsed={"meituan-waimai": 20_500}, ), _session( trace_id=control_trace, device_id=DEVICE_CONTROL, user_id=user.id, app_env="prod", hour=17, minute=0, platform_elapsed=PLATFORM_ELAPSED, ), ]) db.commit() record_claims( db, DEVICE_REPEAT, user.id, first_trace, FIRST_RESULTS, app_env="dev" ) record_claims( db, DEVICE_REPEAT, user.id, second_trace, SECOND_RESULTS, app_env="prod" ) record_claims( db, DEVICE_CONTROL, user.id, control_trace, FIRST_RESULTS, app_env="prod" ) print(f"已生成 {today_cn()} 的领券 mock 数据。") print("筛选用户 19900009001。") print("prod 应有：7/7（100.0%）、-、7/8（87.5%）三条。") print("dev 应有：7/8（87.5%）一条。") if __name__ == "__main__": main()