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']} 笔推广归因订单。" )