"""initial schema Revision ID: 20260331_000001 Revises: Create Date: 2026-03-31 12:30:00 """ from typing import Sequence, Union from alembic import context, op import sqlalchemy as sa # revision identifiers, used by Alembic. revision: str = "20260331_000001" down_revision: Union[str, None] = None branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def _table_exists(name: str) -> bool: if context.is_offline_mode(): return False return sa.inspect(op.get_bind()).has_table(name) def upgrade() -> None: if not _table_exists("websites"): op.create_table( "websites", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("site_url", sa.String(length=512), nullable=False), sa.Column("site_name", sa.String(length=255), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.PrimaryKeyConstraint("id"), ) op.create_index("ix_websites_site_url", "websites", ["site_url"], unique=True) if not _table_exists("tryouts"): op.create_table( "tryouts", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("website_id", sa.Integer(), nullable=False), sa.Column("tryout_id", sa.String(length=255), nullable=False), sa.Column("name", sa.String(length=255), nullable=False), sa.Column("description", sa.String(length=1000), nullable=True), sa.Column("scoring_mode", sa.String(length=50), nullable=False), sa.Column("selection_mode", sa.String(length=50), nullable=False), sa.Column("normalization_mode", sa.String(length=50), nullable=False), sa.Column("min_sample_for_dynamic", sa.Integer(), nullable=False), sa.Column("static_rataan", sa.Float(), nullable=False), sa.Column("static_sb", sa.Float(), nullable=False), sa.Column("ai_generation_enabled", sa.Boolean(), nullable=False), sa.Column("hybrid_transition_slot", sa.Integer(), nullable=True), sa.Column("min_calibration_sample", sa.Integer(), nullable=False), sa.Column("theta_estimation_method", sa.String(length=50), nullable=False), sa.Column("fallback_to_ctt_on_error", sa.Boolean(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.ForeignKeyConstraint(["website_id"], ["websites.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("website_id", "tryout_id", name="uq_tryouts_website_id_tryout_id"), sa.CheckConstraint("min_sample_for_dynamic > 0", name="ck_min_sample_positive"), sa.CheckConstraint("static_rataan > 0", name="ck_static_rataan_positive"), sa.CheckConstraint("static_sb > 0", name="ck_static_sb_positive"), sa.CheckConstraint("min_calibration_sample > 0", name="ck_min_calibration_positive"), ) op.create_index("ix_tryouts_website_id", "tryouts", ["website_id"], unique=False) op.create_index("ix_tryouts_tryout_id", "tryouts", ["tryout_id"], unique=False) if not _table_exists("users"): op.create_table( "users", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("wp_user_id", sa.String(length=255), nullable=False), sa.Column("website_id", sa.Integer(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.ForeignKeyConstraint(["website_id"], ["websites.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("wp_user_id", "website_id", name="uq_users_wp_user_id_website_id"), ) op.create_index("ix_users_wp_user_id", "users", ["wp_user_id"], unique=False) op.create_index("ix_users_website_id", "users", ["website_id"], unique=False) if not _table_exists("items"): op.create_table( "items", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("tryout_id", sa.String(length=255), nullable=False), sa.Column("website_id", sa.Integer(), nullable=False), sa.Column("slot", sa.Integer(), nullable=False), sa.Column("level", sa.String(length=50), nullable=False), sa.Column("stem", sa.Text(), nullable=False), sa.Column("options", sa.JSON(), nullable=False), sa.Column("correct_answer", sa.String(length=10), nullable=False), sa.Column("explanation", sa.Text(), nullable=True), sa.Column("ctt_p", sa.Float(), nullable=True), sa.Column("ctt_bobot", sa.Float(), nullable=True), sa.Column("ctt_category", sa.String(length=50), nullable=True), sa.Column("irt_b", sa.Float(), nullable=True), sa.Column("irt_se", sa.Float(), nullable=True), sa.Column("calibrated", sa.Boolean(), nullable=False), sa.Column("calibration_sample_size", sa.Integer(), nullable=False), sa.Column("generated_by", sa.String(length=50), nullable=False), sa.Column("ai_model", sa.String(length=255), nullable=True), sa.Column("basis_item_id", sa.Integer(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.ForeignKeyConstraint(["website_id"], ["websites.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.ForeignKeyConstraint(["basis_item_id"], ["items.id"], ondelete="SET NULL", onupdate="CASCADE"), sa.ForeignKeyConstraint( ["website_id", "tryout_id"], ["tryouts.website_id", "tryouts.tryout_id"], name="fk_items_tryout", ondelete="CASCADE", onupdate="CASCADE", ), sa.PrimaryKeyConstraint("id"), sa.CheckConstraint("irt_b IS NULL OR (irt_b >= -3 AND irt_b <= 3)", name="ck_irt_b_range"), sa.CheckConstraint("ctt_p IS NULL OR (ctt_p >= 0 AND ctt_p <= 1)", name="ck_ctt_p_range"), sa.CheckConstraint("ctt_bobot IS NULL OR (ctt_bobot >= 0 AND ctt_bobot <= 1)", name="ck_ctt_bobot_range"), sa.CheckConstraint("slot > 0", name="ck_slot_positive"), ) op.create_index("ix_items_tryout_id", "items", ["tryout_id"], unique=False) op.create_index("ix_items_website_id", "items", ["website_id"], unique=False) op.create_index("ix_items_basis_item_id", "items", ["basis_item_id"], unique=False) op.create_index("ix_items_calibrated", "items", ["calibrated"], unique=False) op.create_index( "ix_items_tryout_id_website_id_slot", "items", ["tryout_id", "website_id", "slot", "level"], unique=True, ) if not _table_exists("sessions"): op.create_table( "sessions", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("session_id", sa.String(length=255), nullable=False), sa.Column("wp_user_id", sa.String(length=255), nullable=False), sa.Column("website_id", sa.Integer(), nullable=False), sa.Column("tryout_id", sa.String(length=255), nullable=False), sa.Column("start_time", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("end_time", sa.DateTime(timezone=True), nullable=True), sa.Column("is_completed", sa.Boolean(), nullable=False), sa.Column("scoring_mode_used", sa.String(length=50), nullable=False), sa.Column("total_benar", sa.Integer(), nullable=False), sa.Column("total_bobot_earned", sa.Float(), nullable=False), sa.Column("NM", sa.Integer(), quote=True, nullable=True), sa.Column("NN", sa.Integer(), quote=True, nullable=True), sa.Column("theta", sa.Float(), nullable=True), sa.Column("theta_se", sa.Float(), nullable=True), sa.Column("rataan_used", sa.Float(), nullable=True), sa.Column("sb_used", sa.Float(), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.ForeignKeyConstraint(["website_id"], ["websites.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.ForeignKeyConstraint( ["website_id", "tryout_id"], ["tryouts.website_id", "tryouts.tryout_id"], name="fk_sessions_tryout", ondelete="CASCADE", onupdate="CASCADE", ), sa.ForeignKeyConstraint( ["wp_user_id", "website_id"], ["users.wp_user_id", "users.website_id"], name="fk_sessions_user", ondelete="CASCADE", onupdate="CASCADE", ), sa.PrimaryKeyConstraint("id"), sa.UniqueConstraint("session_id"), sa.CheckConstraint('"NM" IS NULL OR ("NM" >= 0 AND "NM" <= 1000)', name="ck_nm_range"), sa.CheckConstraint('"NN" IS NULL OR ("NN" >= 0 AND "NN" <= 1000)', name="ck_nn_range"), sa.CheckConstraint("theta IS NULL OR (theta >= -3 AND theta <= 3)", name="ck_theta_range"), sa.CheckConstraint("total_benar >= 0", name="ck_total_benar_non_negative"), sa.CheckConstraint("total_bobot_earned >= 0", name="ck_total_bobot_non_negative"), ) op.create_index("ix_sessions_session_id", "sessions", ["session_id"], unique=True) op.create_index("ix_sessions_wp_user_id", "sessions", ["wp_user_id"], unique=False) op.create_index("ix_sessions_website_id", "sessions", ["website_id"], unique=False) op.create_index("ix_sessions_tryout_id", "sessions", ["tryout_id"], unique=False) op.create_index("ix_sessions_is_completed", "sessions", ["is_completed"], unique=False) if not _table_exists("tryout_stats"): op.create_table( "tryout_stats", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("website_id", sa.Integer(), nullable=False), sa.Column("tryout_id", sa.String(length=255), nullable=False), sa.Column("participant_count", sa.Integer(), nullable=False), sa.Column("total_nm_sum", sa.Float(), nullable=False), sa.Column("total_nm_sq_sum", sa.Float(), nullable=False), sa.Column("rataan", sa.Float(), nullable=True), sa.Column("sb", sa.Float(), nullable=True), sa.Column("min_nm", sa.Integer(), nullable=True), sa.Column("max_nm", sa.Integer(), nullable=True), sa.Column("last_calculated", sa.DateTime(timezone=True), nullable=True), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.ForeignKeyConstraint(["website_id"], ["websites.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.ForeignKeyConstraint( ["website_id", "tryout_id"], ["tryouts.website_id", "tryouts.tryout_id"], name="fk_tryout_stats_tryout", ondelete="CASCADE", onupdate="CASCADE", ), sa.PrimaryKeyConstraint("id"), sa.CheckConstraint("participant_count >= 0", name="ck_participant_count_non_negative"), sa.CheckConstraint("min_nm IS NULL OR (min_nm >= 0 AND min_nm <= 1000)", name="ck_min_nm_range"), sa.CheckConstraint("max_nm IS NULL OR (max_nm >= 0 AND max_nm <= 1000)", name="ck_max_nm_range"), sa.CheckConstraint( "min_nm IS NULL OR max_nm IS NULL OR min_nm <= max_nm", name="ck_min_max_nm_order", ), ) op.create_index("ix_tryout_stats_website_id", "tryout_stats", ["website_id"], unique=False) op.create_index("ix_tryout_stats_tryout_id", "tryout_stats", ["tryout_id"], unique=False) op.create_index( "ix_tryout_stats_website_id_tryout_id", "tryout_stats", ["website_id", "tryout_id"], unique=True, ) if not _table_exists("user_answers"): op.create_table( "user_answers", sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), sa.Column("session_id", sa.String(length=255), nullable=False), sa.Column("wp_user_id", sa.String(length=255), nullable=False), sa.Column("website_id", sa.Integer(), nullable=False), sa.Column("tryout_id", sa.String(length=255), nullable=False), sa.Column("item_id", sa.Integer(), nullable=False), sa.Column("response", sa.String(length=10), nullable=False), sa.Column("is_correct", sa.Boolean(), nullable=False), sa.Column("time_spent", sa.Integer(), nullable=False), sa.Column("scoring_mode_used", sa.String(length=50), nullable=False), sa.Column("bobot_earned", sa.Float(), nullable=False), sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), sa.ForeignKeyConstraint(["session_id"], ["sessions.session_id"], ondelete="CASCADE", onupdate="CASCADE"), sa.ForeignKeyConstraint(["website_id"], ["websites.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.ForeignKeyConstraint(["item_id"], ["items.id"], ondelete="CASCADE", onupdate="CASCADE"), sa.PrimaryKeyConstraint("id"), sa.CheckConstraint("time_spent >= 0", name="ck_time_spent_non_negative"), sa.CheckConstraint("bobot_earned >= 0", name="ck_bobot_earned_non_negative"), ) op.create_index("ix_user_answers_session_id", "user_answers", ["session_id"], unique=False) op.create_index("ix_user_answers_wp_user_id", "user_answers", ["wp_user_id"], unique=False) op.create_index("ix_user_answers_website_id", "user_answers", ["website_id"], unique=False) op.create_index("ix_user_answers_tryout_id", "user_answers", ["tryout_id"], unique=False) op.create_index("ix_user_answers_item_id", "user_answers", ["item_id"], unique=False) op.create_index( "ix_user_answers_session_id_item_id", "user_answers", ["session_id", "item_id"], unique=True, ) def downgrade() -> None: if _table_exists("user_answers"): op.drop_table("user_answers") if _table_exists("tryout_stats"): op.drop_table("tryout_stats") if _table_exists("sessions"): op.drop_table("sessions") if _table_exists("items"): op.drop_table("items") if _table_exists("users"): op.drop_table("users") if _table_exists("tryouts"): op.drop_table("tryouts") if _table_exists("websites"): op.drop_table("websites")