shaguabijia-app-server/scripts/seed_coupon_session_mock.py

"""生成 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()