"""Create the homepage content domains. Revision ID: 0017_home_content Revises: 0016_concierge_advisors """ from datetime import datetime from alembic import context, op import sqlalchemy as sa from sqlalchemy import inspect revision = "0017_home_content" down_revision = "0016_concierge_advisors" branch_labels = None depends_on = None SEED_EXPERIENCES = ( { "id": "waterfall-descent", "badge": "玩过推荐", "category": "瀑降体验", "title": "悬崖瀑降", "englishTitle": "WATERFALL DESCENT", "image": "https://genk.mediacdn.vn/139269124445442048/2024/4/27/10-23-sinkhole-1714189653945948438879.jpg", "demandKeyword": "悬崖瀑降", "isActive": True, "sortOrder": 0, }, { "id": "cave-exploration", "badge": "玩过推荐", "category": "地心探险", "title": "森林&探洞", "englishTitle": "CAVE EXPLORATION", "image": "https://www.zurnal24.si/media/img/5e/d5/9526a56dba168aa136f3.jpeg", "demandKeyword": "地心探险", "isActive": True, "sortOrder": 1, }, ) SEED_TEAM_BUILDINGS = ( { "id": "wild-challenge", "tag": "户外挑战", "title": "山野挑战，共创极境", "description": "洞穴、瀑降与协作，适合 10-30 人。", "image": "https://dimg04.c-ctrip.com/images/1mh0412000njfr1ot9453_W_640_10000.jpg?proc=autoorient", "demandKeyword": "户外团建", "isActive": True, "sortOrder": 0, }, { "id": "canyon-teamwork", "tag": "团队协作", "title": "峡谷溯溪，默契同行", "description": "溯溪与分组协作，兼顾参与感与安全。", "image": "https://dimg04.c-ctrip.com/images/0EQ5712000ca7t504EC0E_W_640_10000.jpg?proc=autoorient", "demandKeyword": "峡谷团建", "isActive": True, "sortOrder": 1, }, { "id": "village-gathering", "tag": "人文聚会", "title": "苗寨共聚，认识彼此", "description": "夜游、长桌宴与文化体验，适合团建收尾。", "image": "https://p6.itc.cn/q_70/images03/20200918/df728d2b79d943da869333e2ea2c92c8.jpeg", "demandKeyword": "贵州团建", "isActive": True, "sortOrder": 2, }, ) SEED_WILD_ARCHIVES = ( { "id": "hundred-meter-descent", "title": "百米自降", "image": "https://genk.mediacdn.vn/139269124445442048/2024/4/27/10-23-sinkhole-1714189653945948438879.jpg", "demandKeyword": "悬崖瀑降", "isActive": True, "sortOrder": 0, }, { "id": "shilong-cave", "title": "石龙洞", "image": "https://www.zurnal24.si/media/img/5e/d5/9526a56dba168aa136f3.jpeg", "demandKeyword": "地心探险", "isActive": True, "sortOrder": 1, }, { "id": "cliff-current", "title": "绝壁迎流", "image": "https://q9.itc.cn/q_70/images03/20250810/b62f5afc191a4947a66e6b32721b0235.jpeg", "demandKeyword": "峡谷探险", "isActive": True, "sortOrder": 2, }, { "id": "canyon-streaming", "title": "峡谷溯溪", "image": "https://dimg04.c-ctrip.com/images/0EQ5712000ca7t504EC0E_W_640_10000.jpg?proc=autoorient", "demandKeyword": "峡谷溯溪", "isActive": True, "sortOrder": 3, }, ) def _create_table(table_name: str, columns: list[sa.Column]) -> None: bind = op.get_bind() if table_name in set(inspect(bind).get_table_names()): return op.create_table(table_name, *columns, sa.PrimaryKeyConstraint("id")) op.create_index(f"ix_{table_name}_isActive", table_name, ["isActive"]) op.create_index(f"ix_{table_name}_sortOrder", table_name, ["sortOrder"]) def _seed_table(table_name: str, columns: list[sa.Column], rows: tuple[dict, ...]) -> None: table = sa.table(table_name, *[sa.column(column.name, column.type) for column in columns]) timestamp = datetime(2026, 1, 1) op.bulk_insert( table, [{**row, "createdAt": timestamp, "updatedAt": timestamp} for row in rows], ) def upgrade() -> None: experience_columns = [ sa.Column("id", sa.String(), nullable=False), sa.Column("badge", sa.String(), nullable=False), sa.Column("category", sa.String(), nullable=False), sa.Column("title", sa.String(), nullable=False), sa.Column("englishTitle", sa.String(), nullable=False), sa.Column("image", sa.String(), nullable=False), sa.Column("demandKeyword", sa.String(), nullable=False), sa.Column("isActive", sa.Boolean(), nullable=False, server_default=sa.true()), sa.Column("sortOrder", sa.Integer(), nullable=False, server_default="0"), sa.Column("createdAt", sa.DateTime(), nullable=False), sa.Column("updatedAt", sa.DateTime(), nullable=False), ] team_columns = [ sa.Column("id", sa.String(), nullable=False), sa.Column("tag", sa.String(), nullable=False), sa.Column("title", sa.String(), nullable=False), sa.Column("description", sa.Text(), nullable=False), sa.Column("image", sa.String(), nullable=False), sa.Column("demandKeyword", sa.String(), nullable=False), sa.Column("isActive", sa.Boolean(), nullable=False, server_default=sa.true()), sa.Column("sortOrder", sa.Integer(), nullable=False, server_default="0"), sa.Column("createdAt", sa.DateTime(), nullable=False), sa.Column("updatedAt", sa.DateTime(), nullable=False), ] archive_columns = [ sa.Column("id", sa.String(), nullable=False), sa.Column("title", sa.String(), nullable=False), sa.Column("image", sa.String(), nullable=False), sa.Column("demandKeyword", sa.String(), nullable=False), sa.Column("isActive", sa.Boolean(), nullable=False, server_default=sa.true()), sa.Column("sortOrder", sa.Integer(), nullable=False, server_default="0"), sa.Column("createdAt", sa.DateTime(), nullable=False), sa.Column("updatedAt", sa.DateTime(), nullable=False), ] _create_table("HomeExperience", experience_columns) _create_table("HomeTeamBuilding", team_columns) _create_table("HomeWildArchive", archive_columns) if context.is_offline_mode(): _seed_table("HomeExperience", experience_columns, SEED_EXPERIENCES) _seed_table("HomeTeamBuilding", team_columns, SEED_TEAM_BUILDINGS) _seed_table("HomeWildArchive", archive_columns, SEED_WILD_ARCHIVES) return bind = op.get_bind() for table_name, columns, rows in ( ("HomeExperience", experience_columns, SEED_EXPERIENCES), ("HomeTeamBuilding", team_columns, SEED_TEAM_BUILDINGS), ("HomeWildArchive", archive_columns, SEED_WILD_ARCHIVES), ): table = sa.table(table_name, sa.column("id", sa.String())) if bind.execute(sa.select(table.c.id).limit(1)).first() is None: _seed_table(table_name, columns, rows) def downgrade() -> None: bind = op.get_bind() existing_tables = set(inspect(bind).get_table_names()) for table_name in ("HomeWildArchive", "HomeTeamBuilding", "HomeExperience"): if table_name in existing_tables: op.drop_table(table_name)