from dataclasses import dataclass from datetime import UTC, datetime, timedelta from hashlib import sha256 from app.core.config import Settings from app.core.passwords import PasswordService from app.domains.attribution.models import PromotionCode, Salesperson from app.domains.catalog.models import Plan, Product, ProductChangeLog, ProductVersion from app.domains.identity.models import AdminUser from app.infrastructure.mysql.attribution_repositories import ( SqlAlchemyAttributionRepository, ) from app.infrastructure.mysql.repositories import ( SqlAlchemyCatalogRepository, SqlAlchemyIdentityRepository, ) from app.infrastructure.mysql.sessions import ( DatabaseName, create_mysql_engine, create_session_factory, ) INITIAL_ADMIN_PASSWORD = "zaq1XSW@" @dataclass(frozen=True, slots=True) class SeedManifest: admin_users: tuple[AdminUser, ...] products: tuple[Product, ...] versions: tuple[ProductVersion, ...] salespersons: tuple[Salesperson, ...] promotion_codes: tuple[PromotionCode, ...] @dataclass(frozen=True, slots=True) class PlanConfig: summary: str premium_cents: int coverage_amount_cents: int min_age: int max_age: int def stable_id(value: str) -> str: return sha256(value.encode("utf-8")).hexdigest()[:26].upper() def build_seed_manifest(password_hash: str = "HASHED_DURING_DATABASE_SEED") -> SeedManifest: products = ( _product("MED-BASIC", "蓉惠基础医疗险", "MEDICAL"), _product("MED-UPGRADE", "蓉惠升级医疗险", "MEDICAL"), _product("MED-SENIOR", "银龄守护医疗险", "MEDICAL"), _product("ACC-FAMILY", "家庭意外保障险", "ACCIDENT"), ) product_ids = {product.product_code: product.id for product in products} versions = ( _version(product_ids, "MED-BASIC", "1.0.0", "EXPIRED", 2023, 2024, ("BASIC",)), _version(product_ids, "MED-BASIC", "2.0.0", "PUBLISHED", 2025, 2029, ("BASIC",)), _version(product_ids, "MED-BASIC", "3.0.0", "DRAFT", 2030, None, ("BASIC",)), _version( product_ids, "MED-UPGRADE", "1.0.0", "PUBLISHED", 2025, 2029, ("STANDARD", "ENHANCED"), ), _version( product_ids, "MED-UPGRADE", "1.1.0", "DRAFT", 2030, None, ("STANDARD", "ENHANCED"), ), _version( product_ids, "MED-SENIOR", "1.0.0", "PUBLISHED", 2025, 2029, ("SENIOR_STANDARD",), ), _version( product_ids, "ACC-FAMILY", "1.0.0", "PUBLISHED", 2025, 2029, ("INDIVIDUAL", "FAMILY"), ), ) admin_users = ( _admin("admin", "超级管理员", "SUPER_ADMIN", "ALL", ("*",), password_hash), _admin( "operator01", "运营人员", "OPERATOR", "MASKED_ALL", ("dashboard:read", "order:read", "policy:read", "user:read"), password_hash, ), _admin( "reviewer01", "审核人员", "REVIEWER", "READ_ONLY", ("order:read", "policy:read"), password_hash, ), _admin( "sales_a", "张伟", "SALESPERSON", "SELF", ( "dashboard:read", "order:read", "policy:read", "product:read", "attribution:read", ), password_hash, ), _admin( "sales_b", "李娜", "SALESPERSON", "SELF", ( "dashboard:read", "order:read", "policy:read", "product:read", "attribution:read", ), password_hash, ), _admin( "sales_c", "王磊", "SALESPERSON", "SELF", ( "dashboard:read", "order:read", "policy:read", "product:read", "attribution:read", ), password_hash, ), _admin( "sales_d", "陈晨", "SALESPERSON", "SELF", ( "dashboard:read", "order:read", "policy:read", "product:read", "attribution:read", ), password_hash, ), _admin( "sales_e", "刘洋", "SALESPERSON", "SELF", ( "dashboard:read", "order:read", "policy:read", "product:read", "attribution:read", ), password_hash, ), _admin( "sales_f", "赵敏", "SALESPERSON", "SELF", ( "dashboard:read", "order:read", "policy:read", "product:read", "attribution:read", ), password_hash, ), ) created_at = datetime(2026, 1, 1, tzinfo=UTC) salespersons = ( Salesperson( id=stable_id("salesperson:sales_a"), admin_user_id=stable_id("admin:sales_a"), code="SALES-A", name="张伟", status="ACTIVE", created_at=created_at, ), Salesperson( id=stable_id("salesperson:sales_b"), admin_user_id=stable_id("admin:sales_b"), code="SALES-B", name="李娜", status="ACTIVE", created_at=created_at, ), Salesperson( id=stable_id("salesperson:sales_c"), admin_user_id=stable_id("admin:sales_c"), code="SALES-C", name="王磊", status="ACTIVE", created_at=created_at, ), Salesperson( id=stable_id("salesperson:sales_d"), admin_user_id=stable_id("admin:sales_d"), code="SALES-D", name="陈晨", status="ACTIVE", created_at=created_at, ), Salesperson( id=stable_id("salesperson:sales_e"), admin_user_id=stable_id("admin:sales_e"), code="SALES-E", name="刘洋", status="ACTIVE", created_at=created_at, ), Salesperson( id=stable_id("salesperson:sales_f"), admin_user_id=stable_id("admin:sales_f"), code="SALES-F", name="赵敏", status="ACTIVE", created_at=created_at, ), ) promotion_codes = ( PromotionCode( id=stable_id("promotion:ZBT-SALES-A"), code="ZBT-SALES-A", salesperson_id=salespersons[0].id, channel="GROUND_PROMOTION", status="ACTIVE", created_at=created_at, ), PromotionCode( id=stable_id("promotion:ZBT-SALES-B"), code="ZBT-SALES-B", salesperson_id=salespersons[1].id, channel="COMMUNITY", status="ACTIVE", created_at=created_at, ), PromotionCode( id=stable_id("promotion:ZBT-SALES-C"), code="ZBT-SALES-C", salesperson_id=salespersons[2].id, channel="ENTERPRISE", status="ACTIVE", created_at=created_at, ), PromotionCode( id=stable_id("promotion:ZBT-SALES-D"), code="ZBT-SALES-D", salesperson_id=salespersons[3].id, channel="PARTNER", status="ACTIVE", created_at=created_at, ), PromotionCode( id=stable_id("promotion:ZBT-SALES-E"), code="ZBT-SALES-E", salesperson_id=salespersons[4].id, channel="ONLINE_REFERRAL", status="ACTIVE", created_at=created_at, ), PromotionCode( id=stable_id("promotion:ZBT-SALES-F"), code="ZBT-SALES-F", salesperson_id=salespersons[5].id, channel="STORE", status="ACTIVE", created_at=created_at, ), ) return SeedManifest( admin_users=admin_users, products=products, versions=versions, salespersons=salespersons, promotion_codes=promotion_codes, ) def run_seed(settings: Settings | None = None) -> SeedManifest: resolved_settings = settings or Settings() password_hash = PasswordService().hash(INITIAL_ADMIN_PASSWORD) manifest = build_seed_manifest(password_hash) engine = create_mysql_engine(resolved_settings, DatabaseName.CORE) factory = create_session_factory(engine) identities = SqlAlchemyIdentityRepository(factory) catalog = SqlAlchemyCatalogRepository(factory) attribution = SqlAlchemyAttributionRepository(factory) for admin_user in manifest.admin_users: identities.save_admin_user(admin_user) for product in manifest.products: catalog.save_product(product) for version in manifest.versions: catalog.save_version(version) audit_anchor = datetime(2026, 7, 1, 9, 0, tzinfo=UTC) for index, product in enumerate(manifest.products): catalog.save_change_log( ProductChangeLog( id=stable_id(f"product-log:created:{product.id}"), product_id=product.id, version_id=None, action="PRODUCT_CREATED", actor_id="SYSTEM", actor_name="系统初始化", detail={"product_code": product.product_code}, created_at=audit_anchor - timedelta(days=120 - index), ) ) for index, version in enumerate(manifest.versions): catalog.save_change_log( ProductChangeLog( id=stable_id(f"product-log:version:{version.id}"), product_id=version.product_id, version_id=version.id, action=( "VERSION_DRAFT_CREATED" if version.status == "DRAFT" else "VERSION_IMPORTED" ), actor_id="SYSTEM", actor_name="系统初始化", detail={"version_no": version.version_no, "status": version.status}, created_at=audit_anchor - timedelta(days=40 - index * 3), ) ) for salesperson in manifest.salespersons: attribution.save_salesperson(salesperson) for promotion_code in manifest.promotion_codes: attribution.save_promotion_code(promotion_code) engine.dispose() return manifest def _product(code: str, name: str, category: str) -> Product: summaries = { "MED-BASIC": "覆盖常见住院医疗费用，为年轻家庭提供简洁实用的基础保障。", "MED-UPGRADE": "提高医疗费用保障额度，并提供标准与升级两档方案灵活选择。", "MED-SENIOR": "面向中老年人群的住院医疗保障，覆盖50至75周岁的投保需求。", "ACC-FAMILY": "覆盖日常意外风险，可按个人或家庭方案为家人统一配置保障。", } return Product( id=stable_id(f"product:{code}"), product_code=code, name=name, category=category, summary=summaries[code], status="ACTIVE", ) def _version( product_ids: dict[str, str], product_code: str, version_no: str, status: str, start_year: int, end_year: int | None, plan_codes: tuple[str, ...], ) -> ProductVersion: version_key = f"{product_code}:{version_no}" return ProductVersion( id=stable_id(f"version:{version_key}"), product_id=product_ids[product_code], version_no=version_no, status=status, effective_from=datetime(start_year, 1, 1, tzinfo=UTC), effective_to=( datetime(end_year, 12, 31, 23, 59, 59, tzinfo=UTC) if end_year is not None else None ), terms_summary=( "本产品为一年期保障，等待期、责任免除及赔付比例以电子保险条款为准；" "投保前请如实完成健康告知并确认被保人信息。" ), plans=tuple( _plan(version_key, code) for code in plan_codes ), ) def _plan_name(code: str) -> str: return { "BASIC": "基础计划", "STANDARD": "标准计划", "ENHANCED": "升级计划", "SENIOR_STANDARD": "银龄标准计划", "INDIVIDUAL": "个人计划", "FAMILY": "家庭计划", }[code] def _plan(version_key: str, code: str) -> Plan: config = _plan_config(code) return Plan( id=stable_id(f"plan:{version_key}:{code}"), code=code, name=_plan_name(code), summary=config.summary, premium_cents=config.premium_cents, coverage_amount_cents=config.coverage_amount_cents, min_age=config.min_age, max_age=config.max_age, ) def _plan_config(code: str) -> PlanConfig: return { "BASIC": PlanConfig( "覆盖住院医疗及约定门诊费用，适合首次配置医疗保障的人群。", 23900, 200_000_000, 0, 65, ), "STANDARD": PlanConfig( "提升医疗保障额度，兼顾常见疾病住院和重大疾病医疗支出。", 39900, 300_000_000, 0, 65, ), "ENHANCED": PlanConfig( "提供更高医疗额度和扩展责任，适合重视全面保障的家庭。", 59900, 600_000_000, 0, 65, ), "SENIOR_STANDARD": PlanConfig( "面向中老年人群的住院医疗计划，覆盖高发疾病住院费用。", 19900, 100_000_000, 50, 75, ), "INDIVIDUAL": PlanConfig( "覆盖个人日常意外身故、伤残及意外医疗责任。", 9900, 50_000_000, 0, 70, ), "FAMILY": PlanConfig( "一份计划覆盖家庭成员的常见意外风险，适合家庭统一配置。", 15900, 100_000_000, 0, 70, ), }[code] def _admin( username: str, display_name: str, role: str, data_scope: str, permissions: tuple[str, ...], password_hash: str, ) -> AdminUser: return AdminUser( id=stable_id(f"admin:{username}"), username=username, password_hash=password_hash, display_name=display_name, status="ACTIVE", roles=(role,), permissions=permissions, data_scope=data_scope, ) if __name__ == "__main__": seeded = run_seed() print( f"已写入 {len(seeded.admin_users)} 个后台账号、" f"{len(seeded.products)} 个产品、{len(seeded.versions)} 个产品版本、" f"{len(seeded.promotion_codes)} 个推广码。" )