harness_zbt/backend/app/commands/seed_business_data.py

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from hashlib import sha256

from app.commands.seed import build_seed_manifest, stable_id
from app.core.config import Settings
from app.domains.attribution.models import LeadAttribution, OrderAttribution
from app.domains.enrollment.models import (
    EnrollmentDraft,
    EnrollmentOrder,
    PaymentTransaction,
    Policy,
    Quote,
    UserConfirmation,
)
from app.domains.identity.models import H5User
from app.infrastructure.mysql.attribution_repositories import (
    SqlAlchemyAttributionRepository,
)
from app.infrastructure.mysql.enrollment_repositories import (
    SqlAlchemyEnrollmentRepository,
)
from app.infrastructure.mysql.repositories import SqlAlchemyIdentityRepository
from app.infrastructure.mysql.sessions import (
    DatabaseName,
    create_mysql_engine,
    create_session_factory,
)


@dataclass(frozen=True, slots=True)
class BusinessCase:
    applicant: str
    insured: str
    relationship: str
    product_code: str
    plan_code: str
    status: str
    days_ago: int
    policy_status: str | None = None
    payment_status: str | None = None


BASE_CASES: tuple[BusinessCase, ...] = (
    BusinessCase(
        "周明远", "周明远", "SELF", "MED-BASIC", "BASIC", "ISSUED", 0, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "林晓雨",
        "林建国",
        "PARENT",
        "MED-SENIOR",
        "SENIOR_STANDARD",
        "ISSUED",
        1,
        "ACTIVE",
        "SUCCEEDED",
    ),
    BusinessCase(
        "陈子昂", "王思雅", "SPOUSE", "MED-UPGRADE", "STANDARD", "ISSUED", 2, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "赵清和", "赵安安", "CHILD", "ACC-FAMILY", "FAMILY", "ISSUED", 3, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "许文博", "许文博", "SELF", "ACC-FAMILY", "INDIVIDUAL", "ISSUED", 4, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "郑嘉宁",
        "郑国强",
        "PARENT",
        "MED-SENIOR",
        "SENIOR_STANDARD",
        "ISSUED",
        5,
        "ACTIVE",
        "SUCCEEDED",
    ),
    BusinessCase(
        "孙语桐", "方俊杰", "SPOUSE", "MED-UPGRADE", "ENHANCED", "ISSUED", 7, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "何志诚", "何沐阳", "CHILD", "ACC-FAMILY", "FAMILY", "ISSUED", 9, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "蒋若溪", "蒋若溪", "SELF", "MED-UPGRADE", "STANDARD", "ISSUED", 12, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "唐景行",
        "唐德华",
        "PARENT",
        "MED-SENIOR",
        "SENIOR_STANDARD",
        "ISSUED",
        15,
        "ACTIVE",
        "SUCCEEDED",
    ),
    BusinessCase(
        "邓舒涵", "宋嘉树", "SPOUSE", "MED-BASIC", "BASIC", "ISSUED", 20, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "梁书航", "梁小满", "CHILD", "ACC-FAMILY", "FAMILY", "ISSUED", 25, "ACTIVE", "SUCCEEDED"
    ),
    BusinessCase(
        "顾念安", "顾念安", "SELF", "MED-BASIC", "BASIC", "ISSUED", 30, "CANCELLED", "SUCCEEDED"
    ),
    BusinessCase(
        "马致远",
        "马国庆",
        "PARENT",
        "MED-SENIOR",
        "SENIOR_STANDARD",
        "ISSUED",
        430,
        "EXPIRED",
        "SUCCEEDED",
    ),
    BusinessCase(
        "罗婉清", "罗婉清", "SELF", "MED-UPGRADE", "ENHANCED", "PAID", 1, None, "SUCCEEDED"
    ),
    BusinessCase("谢云舟", "沈知夏", "SPOUSE", "MED-BASIC", "BASIC", "PAID", 3, None, "SUCCEEDED"),
    BusinessCase(
        "苏曼青",
        "苏曼青",
        "SELF",
        "ACC-FAMILY",
        "INDIVIDUAL",
        "PENDING_PAYMENT",
        0,
        None,
        "CREATED",
    ),
    BusinessCase(
        "白敬亭",
        "白国良",
        "PARENT",
        "MED-SENIOR",
        "SENIOR_STANDARD",
        "PENDING_PAYMENT",
        2,
        None,
        None,
    ),
    BusinessCase(
        "陆星河", "陆小禾", "CHILD", "ACC-FAMILY", "FAMILY", "PENDING_PAYMENT", 5, None, "CREATED"
    ),
    BusinessCase(
        "秦知意", "秦知意", "SELF", "MED-UPGRADE", "STANDARD", "PENDING_PAYMENT", 8, None, None
    ),
    BusinessCase(
        "夏予川", "顾安然", "SPOUSE", "MED-BASIC", "BASIC", "PENDING_PAYMENT", 11, None, "CREATED"
    ),
    BusinessCase("叶听澜", "叶听澜", "SELF", "MED-BASIC", "BASIC", "CANCELLED", 4, None, "FAILED"),
    BusinessCase(
        "程砚秋",
        "程建军",
        "PARENT",
        "MED-SENIOR",
        "SENIOR_STANDARD",
        "CANCELLED",
        13,
        None,
        "FAILED",
    ),
    BusinessCase("江望舒", "江小鱼", "CHILD", "ACC-FAMILY", "FAMILY", "CANCELLED", 21, None, None),
)

GENERATED_NAMES = (
    "沈亦辰",
    "温书瑶",
    "乔景川",
    "孟知微",
    "裴言澈",
    "楚星晚",
    "傅时安",
    "宋予宁",
    "贺云深",
    "姜晚晴",
    "顾承泽",
    "许清欢",
    "陆闻舟",
    "苏念初",
    "程屿森",
    "叶舒然",
    "周砚北",
    "林语棠",
    "陈遇白",
    "赵南乔",
    "唐慕川",
    "梁初夏",
    "郑怀瑾",
    "孙云舒",
    "何景明",
    "蒋依澜",
    "邓星野",
    "罗芷晴",
    "马明谦",
    "谢知夏",
    "秦远洲",
    "白若宁",
    "夏云川",
    "顾安禾",
    "叶庭深",
    "程书妍",
    "江叙白",
    "周晚意",
    "林嘉树",
    "陈思源",
    "赵安澜",
    "唐清越",
    "梁知行",
    "郑初晴",
    "孙景和",
    "何语心",
    "蒋承安",
    "邓若溪",
)


def _generated_business_cases() -> tuple[BusinessCase, ...]:
    relationships = ("SELF", "PARENT", "SPOUSE", "CHILD")
    product_plans = (
        ("MED-BASIC", "BASIC"),
        ("MED-BASIC", "BASIC"),
        ("MED-UPGRADE", "STANDARD"),
        ("MED-UPGRADE", "ENHANCED"),
        ("MED-SENIOR", "SENIOR_STANDARD"),
        ("ACC-FAMILY", "INDIVIDUAL"),
        ("ACC-FAMILY", "FAMILY"),
    )
    generated: list[BusinessCase] = []
    for offset, applicant in enumerate(GENERATED_NAMES, start=1):
        index = len(BASE_CASES) + offset
        relationship = relationships[index % len(relationships)]
        product_code, plan_code = product_plans[index % len(product_plans)]
        if product_code == "MED-SENIOR":
            relationship = "PARENT"
        selector = index % 20
        if selector in {0, 7, 14, 18}:
            status = "PENDING_PAYMENT"
            policy_status = None
            payment_status = "CREATED" if index % 2 else None
        elif selector in {9, 19}:
            status = "CANCELLED"
            policy_status = None
            payment_status = "FAILED" if index % 2 else None
        elif selector == 11:
            status = "PAID"
            policy_status = None
            payment_status = "SUCCEEDED"
        else:
            status = "ISSUED"
            policy_status = "ACTIVE"
            payment_status = "SUCCEEDED"
        insured = (
            applicant
            if relationship == "SELF"
            else GENERATED_NAMES[(offset + 11) % len(GENERATED_NAMES)]
        )
        generated.append(
            BusinessCase(
                applicant=applicant,
                insured=insured,
                relationship=relationship,
                product_code=product_code,
                plan_code=plan_code,
                status=status,
                days_ago=(index * 7 + (index // 3) * 5) % 120,
                policy_status=policy_status,
                payment_status=payment_status,
            )
        )
    return tuple(generated)


CASES = BASE_CASES + _generated_business_cases()

ATTRIBUTION_PATTERN: tuple[int | None, ...] = (
    0,
    0,
    0,
    1,
    0,
    1,
    2,
    0,
    1,
    2,
    3,
    0,
    1,
    2,
    3,
    4,
    0,
    1,
    2,
    3,
    4,
    5,
    None,
    None,
)

LEAD_ONLY_COUNT = 42

PREMIUMS = {
    ("MED-BASIC", "BASIC"): 23900,
    ("MED-UPGRADE", "STANDARD"): 39900,
    ("MED-UPGRADE", "ENHANCED"): 59900,
    ("MED-SENIOR", "SENIOR_STANDARD"): 19900,
    ("ACC-FAMILY", "INDIVIDUAL"): 9900,
    ("ACC-FAMILY", "FAMILY"): 15900,
}


def run_seed(settings: Settings | None = None) -> dict[str, int]:
    resolved_settings = settings or Settings()
    engine = create_mysql_engine(resolved_settings, DatabaseName.CORE)
    factory = create_session_factory(engine)
    identities = SqlAlchemyIdentityRepository(factory)
    enrollment = SqlAlchemyEnrollmentRepository(factory)
    attribution = SqlAlchemyAttributionRepository(factory)
    manifest = build_seed_manifest()
    for salesperson in manifest.salespersons:
        attribution.save_salesperson(salesperson)
    for seed_code in manifest.promotion_codes:
        attribution.save_promotion_code(seed_code)
    products = {product.product_code: product for product in manifest.products}
    published_versions = {
        product.product_code: next(
            version
            for version in manifest.versions
            if version.product_id == product.id and version.status == "PUBLISHED"
        )
        for product in manifest.products
    }
    anchor = datetime(2026, 7, 26, 16, 30, tzinfo=UTC)
    customer_ids: dict[str, str] = {}

    for index, case in enumerate(CASES, start=1):
        mobile = f"1886001{index:04d}"
        user_id = stable_id(f"business-user:{mobile}")
        customer_ids[mobile] = user_id
        identities.save_h5_user(
            H5User(
                id=user_id,
                mobile=mobile,
                mobile_masked=f"{mobile[:3]}****{mobile[-4:]}",
                display_name=case.applicant,
                status="ACTIVE",
                created_at=anchor - timedelta(days=case.days_ago + 8),
            )
        )
        promotion_index = ATTRIBUTION_PATTERN[(index - 1) % len(ATTRIBUTION_PATTERN)]
        attributed_code = (
            manifest.promotion_codes[promotion_index] if promotion_index is not None else None
        )
        if attributed_code is not None:
            attribution.save_lead(
                LeadAttribution(
                    id=stable_id(f"business-lead:{index}"),
                    h5_user_id=user_id,
                    salesperson_id=attributed_code.salesperson_id,
                    promotion_code_id=attributed_code.id,
                    source_code=attributed_code.code,
                    visit_count=1 + index % 4,
                    first_touch_at=anchor - timedelta(days=case.days_ago + 8),
                    last_touch_at=anchor - timedelta(days=case.days_ago + 1),
                )
            )

        product = products[case.product_code]
        version = published_versions[case.product_code]
        plan = next(item for item in version.plans if item.code == case.plan_code)
        created_at = anchor - timedelta(
            days=case.days_ago,
            hours=(index * 3) % 12,
            minutes=(index * 7) % 60,
        )
        premium_cents = PREMIUMS[(case.product_code, case.plan_code)]
        quote_id = stable_id(f"business-quote:{index}")
        draft_id = stable_id(f"business-draft:{index}")
        confirmation_id = stable_id(f"business-confirmation:{index}")
        order_id = stable_id(f"business-order:{index}")
        id_no_suffix = f"{index:04d}"
        applicant_snapshot = {
            "name": case.applicant,
            "id_no": f"51010419880101{id_no_suffix}",
        }
        insured_snapshot = {
            "name": case.insured,
            "id_no": f"51010419920102{id_no_suffix}",
        }

        enrollment.save_quote(
            Quote(
                id=quote_id,
                user_id=user_id,
                product_id=product.id,
                product_version_id=version.id,
                plan_id=plan.id,
                insured_age=65 if case.relationship == "PARENT" else 35,
                insured_region_code="510100",
                occupation_code="GENERAL",
                relationship=case.relationship,
                premium_cents=premium_cents,
                currency="CNY",
                rule_version="eligibility-v1",
                rate_version="rate-v1",
                status="USED",
                expires_at=created_at + timedelta(minutes=30),
                created_at=created_at - timedelta(minutes=12),
            )
        )
        enrollment.save_draft(
            EnrollmentDraft(
                id=draft_id,
                user_id=user_id,
                quote_id=quote_id,
                applicant=applicant_snapshot,
                insured=insured_snapshot,
                contact={"mobile": mobile},
                status="CONFIRMED",
                expires_at=created_at + timedelta(hours=24),
                created_at=created_at - timedelta(minutes=8),
            )
        )
        enrollment.save_confirmation(
            UserConfirmation(
                id=confirmation_id,
                user_id=user_id,
                draft_id=draft_id,
                token_hash=sha256(f"business-confirmation-token:{index}".encode()).hexdigest(),
                status="USED",
                expires_at=created_at + timedelta(minutes=10),
                created_at=created_at - timedelta(minutes=3),
            )
        )
        order = EnrollmentOrder(
            id=order_id,
            order_no=f"ORD-{created_at:%Y%m%d}-B{index:07d}",
            user_id=user_id,
            quote_id=quote_id,
            draft_id=draft_id,
            confirmation_id=confirmation_id,
            idempotency_key=f"business-order-{index:04d}",
            product_id=product.id,
            product_version_id=version.id,
            plan_id=plan.id,
            applicant_snapshot=applicant_snapshot,
            insured_snapshot=insured_snapshot,
            amount_cents=premium_cents,
            currency="CNY",
            status=case.status,
            created_at=created_at,
        )
        enrollment.save_order(order)
        if attributed_code is not None:
            attribution.save_order_attribution(
                OrderAttribution(
                    id=stable_id(f"business-order-attribution:{index}"),
                    order_id=order.id,
                    h5_user_id=user_id,
                    salesperson_id=attributed_code.salesperson_id,
                    promotion_code_id=attributed_code.id,
                    source_code=attributed_code.code,
                    amount_cents=premium_cents,
                    attributed_at=created_at,
                )
            )

        if case.payment_status is not None:
            payment_id = stable_id(f"business-payment:{index}")
            succeeded_at = (
                created_at + timedelta(minutes=2) if case.payment_status == "SUCCEEDED" else None
            )
            enrollment.save_payment(
                PaymentTransaction(
                    id=payment_id,
                    payment_no=f"PAY-{created_at:%Y%m%d}-B{index:07d}",
                    order_id=order_id,
                    user_id=user_id,
                    transaction_type="PAYMENT",
                    provider="LOCAL_MOCK_PAY",
                    idempotency_key=f"business-payment-{index:04d}",
                    amount_cents=premium_cents,
                    currency="CNY",
                    status=case.payment_status,
                    provider_transaction_no=(
                        f"MP-{created_at:%Y%m%d}-B{index:07d}"
                        if case.payment_status == "SUCCEEDED"
                        else None
                    ),
                    succeeded_at=succeeded_at,
                    created_at=created_at + timedelta(minutes=1),
                )
            )

        if case.policy_status is not None:
            coverage_start = (created_at + timedelta(days=1)).date()
            coverage_end = coverage_start + timedelta(days=365)
            enrollment.save_policy(
                Policy(
                    id=stable_id(f"business-policy:{index}"),
                    policy_no=f"POL-{created_at:%Y%m%d}-B{index:07d}",
                    order_id=order_id,
                    user_id=user_id,
                    product_id=product.id,
                    product_version_id=version.id,
                    plan_id=plan.id,
                    premium_cents=premium_cents,
                    currency="CNY",
                    coverage_start=coverage_start,
                    coverage_end=coverage_end,
                    status=case.policy_status,
                    issued_at=created_at + timedelta(minutes=4),
                )
            )

    for index in range(1, LEAD_ONLY_COUNT + 1):
        mobile = f"1887002{index:04d}"
        user_id = stable_id(f"business-lead-only-user:{mobile}")
        promotion_index = ATTRIBUTION_PATTERN[
            (index * 5 + index // 4) % (len(ATTRIBUTION_PATTERN) - 2)
        ]
        if promotion_index is None:
            continue
        promotion_code = manifest.promotion_codes[promotion_index]
        created_days_ago = (index * 11 + 3) % 150
        identities.save_h5_user(
            H5User(
                id=user_id,
                mobile=mobile,
                mobile_masked=f"{mobile[:3]}****{mobile[-4:]}",
                display_name=f"咨询客户{index:02d}",
                status="ACTIVE",
                created_at=anchor - timedelta(days=created_days_ago),
            )
        )
        attribution.save_lead(
            LeadAttribution(
                id=stable_id(f"business-lead-only:{index}"),
                h5_user_id=user_id,
                salesperson_id=promotion_code.salesperson_id,
                promotion_code_id=promotion_code.id,
                source_code=promotion_code.code,
                visit_count=1 + (index * 3) % 7,
                first_touch_at=anchor - timedelta(days=created_days_ago),
                last_touch_at=anchor - timedelta(days=max(0, created_days_ago - (index % 12))),
            )
        )

    engine.dispose()
    return {
        "customers": len(customer_ids) + LEAD_ONLY_COUNT,
        "orders": len(CASES),
        "policies": sum(case.policy_status is not None for case in CASES),
        "payments": sum(case.payment_status is not None for case in CASES),
        "leads": sum(
            ATTRIBUTION_PATTERN[(index - 1) % len(ATTRIBUTION_PATTERN)] is not None
            for index in range(1, len(CASES) + 1)
        )
        + LEAD_ONLY_COUNT,
        "attributed_orders": sum(
            ATTRIBUTION_PATTERN[(index - 1) % len(ATTRIBUTION_PATTERN)] is not None
            for index in range(1, len(CASES) + 1)
        ),
    }


if __name__ == "__main__":
    result = run_seed()
    print(
        "业务数据写入完成："
        f"{result['customers']} 位投保用户，"
        f"{result['orders']} 笔订单，"
        f"{result['payments']} 条支付流水，"
        f"{result['policies']} 张保单，"
        f"{result['leads']} 条推广线索，"
        f"{result['attributed_orders']} 笔推广归因订单。"
    )