diff --git a/README.md b/README.md index 5a5abf7..79cc3b6 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,7 @@ - **Aiogram 3.x:** Асинхронный фреймворк для Telegram ботов. - **aiohttp:** Для запуска веб-сервера (вебхуки). - **SQLAlchemy 2.x & asyncpg:** Асинхронная работа с базой данных PostgreSQL. +- **Alembic:** Миграции схемы базы данных. - **YooKassa, FreeKassa API, Platega, SeverPay, aiocryptopay:** Интеграции с платежными системами. - **Pydantic:** Для управления настройками из `.env` файла. - **Docker & Docker Compose:** Для контейнеризации и развертывания. @@ -182,6 +183,15 @@ > 💡 Если включена проверка подписки (`REQUIRED_CHANNEL_SUBSCRIBE_TO_USE=true`), добавьте бота администратором в канал из `REQUIRED_CHANNEL_ID`. Пользователь увидит кнопку «Проверить подписку», и после успешного подтверждения доступ продолжится. +### Миграции БД (Alembic) + +- При запуске `python main.py` миграции применяются автоматически до `head`. +- Для ручного запуска используйте: + +```bash +alembic upgrade head +``` + ## Подробная инструкция для развертывания на сервере с панелью Remnawave ### 1. Клонирование репозитория diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..a97945f --- /dev/null +++ b/alembic.ini @@ -0,0 +1,38 @@ +[alembic] +script_location = alembic +prepend_sys_path = . +sqlalchemy.url = + +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARN +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARN +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..5bcb21f --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import asyncio +import os +from logging.config import fileConfig + +from alembic import context +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config + +from db.models import Base + +config = context.config + +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def _get_database_url() -> str: + configured_url = config.get_main_option("sqlalchemy.url") + if configured_url: + return configured_url + + env_url = os.getenv("DATABASE_URL") + if env_url: + return env_url + + user = os.getenv("POSTGRES_USER", "postgres") + password = os.getenv("POSTGRES_PASSWORD", "postgres") + host = os.getenv("POSTGRES_HOST", "localhost") + port = os.getenv("POSTGRES_PORT", "5432") + db_name = os.getenv("POSTGRES_DB", "postgres") + return f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{db_name}" + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode.""" + + context.configure( + url=_get_database_url(), + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + compare_type=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure( + connection=connection, + target_metadata=target_metadata, + compare_type=True, + ) + + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + configuration = config.get_section(config.config_ini_section) or {} + configuration["sqlalchemy.url"] = _get_database_url() + + connectable = async_engine_from_config( + configuration, + prefix="sqlalchemy.", + poolclass=pool.NullPool, + ) + + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + + await connectable.dispose() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode.""" + + connectable = config.attributes.get("connection", None) + if connectable is None: + asyncio.run(run_async_migrations()) + else: + do_run_migrations(connectable) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..5866ecf --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,26 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/0001_initial_schema.py b/alembic/versions/0001_initial_schema.py new file mode 100644 index 0000000..3e6d9fc --- /dev/null +++ b/alembic/versions/0001_initial_schema.py @@ -0,0 +1,280 @@ +"""initial schema + +Revision ID: 0001_initial_schema +Revises: +Create Date: 2026-02-08 00:00:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "0001_initial_schema" +down_revision: Union[str, Sequence[str], None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "users", + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("username", sa.String(), nullable=True), + sa.Column("first_name", sa.String(), nullable=True), + sa.Column("last_name", sa.String(), nullable=True), + sa.Column("language_code", sa.String(), nullable=True), + sa.Column("registration_date", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("is_banned", sa.Boolean(), nullable=True), + sa.Column("panel_user_uuid", sa.String(), nullable=True), + sa.Column("referral_code", sa.String(length=16), nullable=True), + sa.Column("referred_by_id", sa.BigInteger(), nullable=True), + sa.Column("channel_subscription_verified", sa.Boolean(), nullable=True), + sa.Column("channel_subscription_checked_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("channel_subscription_verified_for", sa.BigInteger(), nullable=True), + sa.ForeignKeyConstraint(["referred_by_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("user_id"), + sa.UniqueConstraint("panel_user_uuid"), + sa.UniqueConstraint("referral_code"), + ) + op.create_index("ix_users_username", "users", ["username"], unique=False) + + op.create_table( + "promo_codes", + sa.Column("promo_code_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("code", sa.String(), nullable=False), + sa.Column("promo_type", sa.String(), nullable=False), + sa.Column("bonus_days", sa.Integer(), nullable=True), + sa.Column("discount_percentage", sa.Integer(), nullable=True), + sa.Column("max_activations", sa.Integer(), nullable=False), + sa.Column("current_activations", sa.Integer(), nullable=True), + sa.Column("is_active", sa.Boolean(), nullable=True), + sa.Column("created_by_admin_id", sa.BigInteger(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("valid_until", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("promo_code_id"), + sa.UniqueConstraint("code"), + ) + op.create_index("idx_promo_codes_promo_type", "promo_codes", ["promo_type"], unique=False) + + op.create_table( + "ad_campaigns", + sa.Column("ad_campaign_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("source", sa.String(), nullable=False), + sa.Column("start_param", sa.String(), nullable=False), + sa.Column("cost", sa.Float(), nullable=False), + sa.Column("is_active", sa.Boolean(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.PrimaryKeyConstraint("ad_campaign_id"), + sa.UniqueConstraint("start_param"), + ) + op.create_index("ix_ad_campaigns_source", "ad_campaigns", ["source"], unique=False) + op.create_index("ix_ad_campaigns_is_active", "ad_campaigns", ["is_active"], unique=False) + + op.create_table( + "subscriptions", + sa.Column("subscription_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("panel_user_uuid", sa.String(), nullable=False), + sa.Column("panel_subscription_uuid", sa.String(), nullable=True), + sa.Column("start_date", sa.DateTime(timezone=True), nullable=True), + sa.Column("end_date", sa.DateTime(timezone=True), nullable=False), + sa.Column("duration_months", sa.Integer(), nullable=True), + sa.Column("is_active", sa.Boolean(), nullable=True), + sa.Column("status_from_panel", sa.String(), nullable=True), + sa.Column("traffic_limit_bytes", sa.BigInteger(), nullable=True), + sa.Column("traffic_used_bytes", sa.BigInteger(), nullable=True), + sa.Column("last_notification_sent", sa.DateTime(timezone=True), nullable=True), + sa.Column("provider", sa.String(), nullable=True), + sa.Column("skip_notifications", sa.Boolean(), nullable=True), + sa.Column("auto_renew_enabled", sa.Boolean(), nullable=True), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("subscription_id"), + sa.UniqueConstraint("panel_subscription_uuid"), + ) + op.create_index("ix_subscriptions_user_id", "subscriptions", ["user_id"], unique=False) + op.create_index("ix_subscriptions_panel_user_uuid", "subscriptions", ["panel_user_uuid"], unique=False) + op.create_index("ix_subscriptions_end_date", "subscriptions", ["end_date"], unique=False) + op.create_index("ix_subscriptions_is_active", "subscriptions", ["is_active"], unique=False) + op.create_index("ix_subscriptions_auto_renew_enabled", "subscriptions", ["auto_renew_enabled"], unique=False) + + op.create_table( + "payments", + sa.Column("payment_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("yookassa_payment_id", sa.String(), nullable=True), + sa.Column("provider_payment_id", sa.String(), nullable=True), + sa.Column("provider", sa.String(), nullable=False), + sa.Column("idempotence_key", sa.String(), nullable=True), + sa.Column("amount", sa.Float(), nullable=False), + sa.Column("original_amount", sa.Float(), nullable=True), + sa.Column("discount_applied", sa.Float(), nullable=True), + sa.Column("currency", sa.String(), nullable=False), + sa.Column("status", sa.String(), nullable=False), + sa.Column("description", sa.String(), nullable=True), + sa.Column("subscription_duration_months", sa.Integer(), nullable=True), + sa.Column("promo_code_id", sa.Integer(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["promo_code_id"], ["promo_codes.promo_code_id"]), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("payment_id"), + sa.UniqueConstraint("idempotence_key"), + sa.UniqueConstraint("provider_payment_id"), + sa.UniqueConstraint("yookassa_payment_id"), + ) + op.create_index("ix_payments_user_id", "payments", ["user_id"], unique=False) + op.create_index("ix_payments_provider", "payments", ["provider"], unique=False) + op.create_index("ix_payments_status", "payments", ["status"], unique=False) + + op.create_table( + "user_billing", + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("yookassa_payment_method_id", sa.String(), nullable=True), + sa.Column("card_last4", sa.String(), nullable=True), + sa.Column("card_network", sa.String(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("user_id"), + sa.UniqueConstraint("yookassa_payment_method_id"), + ) + + op.create_table( + "user_payment_methods", + sa.Column("method_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("provider", sa.String(), nullable=False), + sa.Column("provider_payment_method_id", sa.String(), nullable=False), + sa.Column("card_last4", sa.String(), nullable=True), + sa.Column("card_network", sa.String(), nullable=True), + sa.Column("is_default", sa.Boolean(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("method_id"), + sa.UniqueConstraint("provider_payment_method_id"), + sa.UniqueConstraint("user_id", "provider_payment_method_id", name="uq_user_provider_method"), + ) + op.create_index("ix_user_payment_methods_user_id", "user_payment_methods", ["user_id"], unique=False) + op.create_index("ix_user_payment_methods_provider", "user_payment_methods", ["provider"], unique=False) + op.create_index("ix_user_payment_methods_is_default", "user_payment_methods", ["is_default"], unique=False) + + op.create_table( + "promo_code_activations", + sa.Column("activation_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("promo_code_id", sa.Integer(), nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("activated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("payment_id", sa.Integer(), nullable=True), + sa.ForeignKeyConstraint(["payment_id"], ["payments.payment_id"]), + sa.ForeignKeyConstraint(["promo_code_id"], ["promo_codes.promo_code_id"]), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("activation_id"), + sa.UniqueConstraint("promo_code_id", "user_id", name="uq_promo_user_activation"), + ) + + op.create_table( + "active_discounts", + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("promo_code_id", sa.Integer(), nullable=False), + sa.Column("discount_percentage", sa.Integer(), nullable=False), + sa.Column("activated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["promo_code_id"], ["promo_codes.promo_code_id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("user_id"), + ) + + op.create_table( + "message_logs", + sa.Column("log_id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=True), + sa.Column("telegram_username", sa.String(), nullable=True), + sa.Column("telegram_first_name", sa.String(), nullable=True), + sa.Column("event_type", sa.String(), nullable=False), + sa.Column("content", sa.Text(), nullable=True), + sa.Column("raw_update_preview", sa.Text(), nullable=True), + sa.Column("timestamp", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("is_admin_event", sa.Boolean(), nullable=True), + sa.Column("target_user_id", sa.BigInteger(), nullable=True), + sa.ForeignKeyConstraint(["target_user_id"], ["users.user_id"]), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("log_id"), + ) + op.create_index("ix_message_logs_user_id", "message_logs", ["user_id"], unique=False) + op.create_index("ix_message_logs_event_type", "message_logs", ["event_type"], unique=False) + op.create_index("ix_message_logs_timestamp", "message_logs", ["timestamp"], unique=False) + op.create_index("ix_message_logs_target_user_id", "message_logs", ["target_user_id"], unique=False) + + op.create_table( + "panel_sync_status", + sa.Column("id", sa.Integer(), autoincrement=False, nullable=False), + sa.Column("last_sync_time", sa.DateTime(timezone=True), nullable=True), + sa.Column("status", sa.String(), nullable=True), + sa.Column("details", sa.Text(), nullable=True), + sa.Column("users_processed_from_panel", sa.Integer(), nullable=True), + sa.Column("subscriptions_synced", sa.Integer(), nullable=True), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("id"), + ) + + op.create_table( + "ad_attributions", + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("ad_campaign_id", sa.Integer(), nullable=False), + sa.Column("first_start_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=True), + sa.Column("trial_activated_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["ad_campaign_id"], ["ad_campaigns.ad_campaign_id"]), + sa.ForeignKeyConstraint(["user_id"], ["users.user_id"]), + sa.PrimaryKeyConstraint("user_id"), + ) + op.create_index("ix_ad_attributions_ad_campaign_id", "ad_attributions", ["ad_campaign_id"], unique=False) + + +def downgrade() -> None: + op.drop_index("ix_ad_attributions_ad_campaign_id", table_name="ad_attributions") + op.drop_table("ad_attributions") + + op.drop_table("panel_sync_status") + + op.drop_index("ix_message_logs_target_user_id", table_name="message_logs") + op.drop_index("ix_message_logs_timestamp", table_name="message_logs") + op.drop_index("ix_message_logs_event_type", table_name="message_logs") + op.drop_index("ix_message_logs_user_id", table_name="message_logs") + op.drop_table("message_logs") + + op.drop_table("active_discounts") + + op.drop_table("promo_code_activations") + + op.drop_index("ix_user_payment_methods_is_default", table_name="user_payment_methods") + op.drop_index("ix_user_payment_methods_provider", table_name="user_payment_methods") + op.drop_index("ix_user_payment_methods_user_id", table_name="user_payment_methods") + op.drop_table("user_payment_methods") + + op.drop_table("user_billing") + + op.drop_index("ix_payments_status", table_name="payments") + op.drop_index("ix_payments_provider", table_name="payments") + op.drop_index("ix_payments_user_id", table_name="payments") + op.drop_table("payments") + + op.drop_index("ix_subscriptions_auto_renew_enabled", table_name="subscriptions") + op.drop_index("ix_subscriptions_is_active", table_name="subscriptions") + op.drop_index("ix_subscriptions_end_date", table_name="subscriptions") + op.drop_index("ix_subscriptions_panel_user_uuid", table_name="subscriptions") + op.drop_index("ix_subscriptions_user_id", table_name="subscriptions") + op.drop_table("subscriptions") + + op.drop_index("ix_ad_campaigns_is_active", table_name="ad_campaigns") + op.drop_index("ix_ad_campaigns_source", table_name="ad_campaigns") + op.drop_table("ad_campaigns") + + op.drop_index("idx_promo_codes_promo_type", table_name="promo_codes") + op.drop_table("promo_codes") + + op.drop_index("ix_users_username", table_name="users") + op.drop_table("users") diff --git a/db/alembic_runner.py b/db/alembic_runner.py new file mode 100644 index 0000000..4797420 --- /dev/null +++ b/db/alembic_runner.py @@ -0,0 +1,277 @@ +import logging +from pathlib import Path +from typing import Tuple + +from alembic import command +from alembic.config import Config +from sqlalchemy import inspect, text +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import AsyncEngine + +from config.settings import Settings + + +_BASELINE_REVISION = "0001_initial_schema" + + +def _build_alembic_config(settings: Settings) -> Config: + project_root = Path(__file__).resolve().parents[1] + config = Config(str(project_root / "alembic.ini")) + config.set_main_option("script_location", str(project_root / "alembic")) + config.set_main_option("sqlalchemy.url", settings.DATABASE_URL) + return config + + +def _inspect_database_state(connection: Connection) -> Tuple[bool, bool, bool]: + db_inspector = inspect(connection) + has_alembic_version = db_inspector.has_table("alembic_version") + has_users_table = db_inspector.has_table("users") + has_legacy_migrator_table = db_inspector.has_table("schema_migrations") + return has_alembic_version, has_users_table, has_legacy_migrator_table + + +def _run_legacy_migrator_compatibility(connection: Connection) -> None: + db_inspector = inspect(connection) + if not db_inspector.has_table("users"): + return + + users_columns = { + column["name"] + for column in db_inspector.get_columns("users") + } + user_alter_statements = [] + + if "channel_subscription_verified" not in users_columns: + user_alter_statements.append( + "ALTER TABLE users ADD COLUMN channel_subscription_verified BOOLEAN" + ) + if "channel_subscription_checked_at" not in users_columns: + user_alter_statements.append( + "ALTER TABLE users ADD COLUMN channel_subscription_checked_at TIMESTAMPTZ" + ) + if "channel_subscription_verified_for" not in users_columns: + user_alter_statements.append( + "ALTER TABLE users ADD COLUMN channel_subscription_verified_for BIGINT" + ) + if "referral_code" not in users_columns: + user_alter_statements.append( + "ALTER TABLE users ADD COLUMN referral_code VARCHAR(16)" + ) + + for statement in user_alter_statements: + connection.execute(text(statement)) + + users_columns = { + column["name"] + for column in inspect(connection).get_columns("users") + } + if "referral_code" in users_columns: + connection.execute( + text( + """ + WITH generated_codes AS ( + SELECT + user_id, + UPPER( + SUBSTRING( + md5( + user_id::text + || clock_timestamp()::text + || random()::text + ) + FROM 1 FOR 9 + ) + ) AS referral_code + FROM users + WHERE referral_code IS NULL OR referral_code = '' + ) + UPDATE users AS u + SET referral_code = g.referral_code + FROM generated_codes AS g + WHERE u.user_id = g.user_id + """ + ) + ) + connection.execute( + text( + """ + CREATE UNIQUE INDEX IF NOT EXISTS uq_users_referral_code + ON users (referral_code) + WHERE referral_code IS NOT NULL + """ + ) + ) + connection.execute( + text( + """ + UPDATE users + SET referral_code = UPPER(referral_code) + WHERE referral_code IS NOT NULL + AND referral_code <> UPPER(referral_code) + """ + ) + ) + + db_inspector = inspect(connection) + if db_inspector.has_table("payments"): + payments_columns = { + column["name"] + for column in db_inspector.get_columns("payments") + } + if "original_amount" not in payments_columns: + connection.execute(text("ALTER TABLE payments ADD COLUMN original_amount FLOAT")) + if "discount_applied" not in payments_columns: + connection.execute(text("ALTER TABLE payments ADD COLUMN discount_applied FLOAT")) + + db_inspector = inspect(connection) + has_promo_codes = db_inspector.has_table("promo_codes") + if has_promo_codes: + promo_columns = { + column["name"] + for column in db_inspector.get_columns("promo_codes") + } + if "promo_type" not in promo_columns: + connection.execute( + text( + "ALTER TABLE promo_codes ADD COLUMN promo_type VARCHAR NOT NULL DEFAULT 'bonus_days'" + ) + ) + if "discount_percentage" not in promo_columns: + connection.execute( + text("ALTER TABLE promo_codes ADD COLUMN discount_percentage INTEGER") + ) + if "bonus_days" in promo_columns: + connection.execute( + text("ALTER TABLE promo_codes ALTER COLUMN bonus_days DROP NOT NULL") + ) + connection.execute( + text( + "CREATE INDEX IF NOT EXISTS idx_promo_codes_promo_type ON promo_codes (promo_type)" + ) + ) + + db_inspector = inspect(connection) + has_active_discounts = db_inspector.has_table("active_discounts") + if not has_active_discounts and has_promo_codes: + connection.execute( + text( + """ + CREATE TABLE IF NOT EXISTS active_discounts ( + user_id BIGINT PRIMARY KEY, + promo_code_id INTEGER NOT NULL, + discount_percentage INTEGER NOT NULL, + activated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT fk_active_discounts_user + FOREIGN KEY (user_id) REFERENCES users (user_id) ON DELETE CASCADE, + CONSTRAINT fk_active_discounts_promo_code + FOREIGN KEY (promo_code_id) REFERENCES promo_codes (promo_code_id) ON DELETE CASCADE + ) + """ + ) + ) + has_active_discounts = True + + if has_active_discounts and has_promo_codes: + connection.execute( + text( + "DELETE FROM active_discounts ad " + "WHERE NOT EXISTS (SELECT 1 FROM users u WHERE u.user_id = ad.user_id) " + "OR NOT EXISTS (SELECT 1 FROM promo_codes p WHERE p.promo_code_id = ad.promo_code_id)" + ) + ) + connection.execute( + text( + "ALTER TABLE active_discounts " + "DROP CONSTRAINT IF EXISTS active_discounts_user_id_fkey" + ) + ) + connection.execute( + text( + "ALTER TABLE active_discounts " + "DROP CONSTRAINT IF EXISTS fk_active_discounts_user" + ) + ) + connection.execute( + text( + "ALTER TABLE active_discounts " + "DROP CONSTRAINT IF EXISTS active_discounts_promo_code_id_fkey" + ) + ) + connection.execute( + text( + "ALTER TABLE active_discounts " + "DROP CONSTRAINT IF EXISTS fk_active_discounts_promo_code" + ) + ) + connection.execute( + text( + "ALTER TABLE active_discounts " + "ADD CONSTRAINT fk_active_discounts_user " + "FOREIGN KEY (user_id) REFERENCES users (user_id) ON DELETE CASCADE" + ) + ) + connection.execute( + text( + "ALTER TABLE active_discounts " + "ADD CONSTRAINT fk_active_discounts_promo_code " + "FOREIGN KEY (promo_code_id) REFERENCES promo_codes (promo_code_id) ON DELETE CASCADE" + ) + ) + elif has_active_discounts and not has_promo_codes: + logging.warning( + "Alembic legacy compatibility: skipped active_discounts FK repair " + "because promo_codes table is missing." + ) + + +def _run_stamp(connection: Connection, alembic_config: Config, revision: str) -> None: + alembic_config.attributes["connection"] = connection + command.stamp(alembic_config, revision) + + +def _run_upgrade(connection: Connection, alembic_config: Config) -> None: + alembic_config.attributes["connection"] = connection + command.upgrade(alembic_config, "head") + + +async def run_alembic_migrations(settings: Settings, async_engine: AsyncEngine) -> None: + """Apply Alembic migrations with bootstrap for existing installations.""" + + alembic_config = _build_alembic_config(settings) + + async with async_engine.begin() as async_connection: + ( + has_alembic_version, + has_users_table, + has_legacy_migrator_table, + ) = await async_connection.run_sync( + _inspect_database_state + ) + + if not has_alembic_version and has_users_table: + if not has_legacy_migrator_table: + raise RuntimeError( + "Alembic bootstrap refused: found existing users table without " + "alembic_version and without legacy schema_migrations marker. " + "Cannot safely determine migration baseline." + ) + + logging.info( + "Alembic: applying legacy migrator compatibility fixes before stamp." + ) + await async_connection.run_sync(_run_legacy_migrator_compatibility) + + logging.info( + "Alembic: existing schema detected without alembic_version; stamping %s.", + _BASELINE_REVISION, + ) + await async_connection.run_sync( + _run_stamp, + alembic_config, + _BASELINE_REVISION, + ) + + logging.info("Alembic: running upgrade to head...") + await async_connection.run_sync(_run_upgrade, alembic_config) + + logging.info("Alembic: migrations applied successfully.") diff --git a/db/database_setup.py b/db/database_setup.py index ba084fa..3d89ecc 100644 --- a/db/database_setup.py +++ b/db/database_setup.py @@ -4,8 +4,7 @@ from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sess from sqlalchemy.orm import sessionmaker from config.settings import Settings -from .models import Base -from .migrator import run_database_migrations +from .alembic_runner import run_alembic_migrations async_engine = None @@ -46,7 +45,7 @@ def init_db_connection(settings: Settings) -> sessionmaker: autoflush=False, ) logging.info( - f"SQLAlchemy Async Engine and SessionFactory configured for PostgreSQL." + "SQLAlchemy Async Engine and SessionFactory configured for PostgreSQL." ) return local_async_session_factory @@ -77,12 +76,8 @@ async def init_db(settings: Settings, session_factory: sessionmaker): "async_engine is not initialized. Call init_db_connection and get session_factory first." ) - async with async_engine.begin() as conn: - await conn.run_sync(Base.metadata.create_all) - await conn.run_sync(run_database_migrations) - logging.info( - "PostgreSQL database initialized/checked successfully using SQLAlchemy." - ) + await run_alembic_migrations(settings, async_engine) + logging.info("PostgreSQL database migrations checked/applied via Alembic.") async with session_factory() as session: from .dal.panel_sync_dal import get_panel_sync_status, update_panel_sync_status diff --git a/db/migrator.py b/db/migrator.py deleted file mode 100644 index 18bb209..0000000 --- a/db/migrator.py +++ /dev/null @@ -1,280 +0,0 @@ -import logging -from dataclasses import dataclass -from typing import Callable, List, Set - -from sqlalchemy import inspect, text -from sqlalchemy.engine import Connection - - -@dataclass(frozen=True) -class Migration: - id: str - description: str - upgrade: Callable[[Connection], None] - - -def _ensure_migrations_table(connection: Connection) -> None: - connection.execute( - text( - """ - CREATE TABLE IF NOT EXISTS schema_migrations ( - id VARCHAR(255) PRIMARY KEY, - applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW() - ) - """ - ) - ) - - -def _migration_0001_add_channel_subscription_fields(connection: Connection) -> None: - inspector = inspect(connection) - columns: Set[str] = {col["name"] for col in inspector.get_columns("users")} - statements: List[str] = [] - - if "channel_subscription_verified" not in columns: - statements.append( - "ALTER TABLE users ADD COLUMN channel_subscription_verified BOOLEAN" - ) - if "channel_subscription_checked_at" not in columns: - statements.append( - "ALTER TABLE users ADD COLUMN channel_subscription_checked_at TIMESTAMPTZ" - ) - if "channel_subscription_verified_for" not in columns: - statements.append( - "ALTER TABLE users ADD COLUMN channel_subscription_verified_for BIGINT" - ) - - for stmt in statements: - connection.execute(text(stmt)) - - -def _migration_0002_add_referral_code(connection: Connection) -> None: - inspector = inspect(connection) - columns: Set[str] = {col["name"] for col in inspector.get_columns("users")} - - if "referral_code" not in columns: - connection.execute( - text("ALTER TABLE users ADD COLUMN referral_code VARCHAR(16)") - ) - - connection.execute( - text( - """ - WITH generated_codes AS ( - SELECT - user_id, - UPPER( - SUBSTRING( - md5( - user_id::text - || clock_timestamp()::text - || random()::text - ) - FROM 1 FOR 9 - ) - ) AS referral_code - FROM users - WHERE referral_code IS NULL OR referral_code = '' - ) - UPDATE users AS u - SET referral_code = g.referral_code - FROM generated_codes AS g - WHERE u.user_id = g.user_id - """ - ) - ) - - connection.execute( - text( - """ - CREATE UNIQUE INDEX IF NOT EXISTS uq_users_referral_code - ON users (referral_code) - WHERE referral_code IS NOT NULL - """ - ) - ) - - -def _migration_0003_normalize_referral_codes(connection: Connection) -> None: - inspector = inspect(connection) - columns: Set[str] = {col["name"] for col in inspector.get_columns("users")} - if "referral_code" not in columns: - return - - connection.execute( - text( - """ - UPDATE users - SET referral_code = UPPER(referral_code) - WHERE referral_code IS NOT NULL - AND referral_code <> UPPER(referral_code) - """ - ) - ) - - -def _migration_0004_add_discount_promo_codes(connection: Connection) -> None: - inspector = inspect(connection) - - # 1. Добавить поля в payments - payment_columns: Set[str] = {col["name"] for col in inspector.get_columns("payments")} - if "original_amount" not in payment_columns: - connection.execute(text("ALTER TABLE payments ADD COLUMN original_amount FLOAT")) - if "discount_applied" not in payment_columns: - connection.execute(text("ALTER TABLE payments ADD COLUMN discount_applied FLOAT")) - - # 2. Модифицировать promo_codes - promo_columns: Set[str] = {col["name"] for col in inspector.get_columns("promo_codes")} - - if "promo_type" not in promo_columns: - connection.execute( - text( - "ALTER TABLE promo_codes ADD COLUMN promo_type VARCHAR NOT NULL DEFAULT 'bonus_days'" - ) - ) - - if "discount_percentage" not in promo_columns: - connection.execute( - text("ALTER TABLE promo_codes ADD COLUMN discount_percentage INTEGER") - ) - - # Изменить bonus_days на nullable (если еще не nullable) - connection.execute( - text("ALTER TABLE promo_codes ALTER COLUMN bonus_days DROP NOT NULL") - ) - - # Создать индекс на promo_type - connection.execute( - text( - "CREATE INDEX IF NOT EXISTS idx_promo_codes_promo_type ON promo_codes (promo_type)" - ) - ) - - # 3. Создать таблицу active_discounts - connection.execute( - text( - """ - CREATE TABLE IF NOT EXISTS active_discounts ( - user_id BIGINT PRIMARY KEY, - promo_code_id INTEGER NOT NULL, - discount_percentage INTEGER NOT NULL, - activated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - CONSTRAINT fk_active_discounts_user - FOREIGN KEY (user_id) REFERENCES users (user_id) ON DELETE CASCADE, - CONSTRAINT fk_active_discounts_promo_code - FOREIGN KEY (promo_code_id) REFERENCES promo_codes (promo_code_id) ON DELETE CASCADE - ) - """ - ) - ) - - -def _migration_0005_fix_active_discounts_fk_cascade(connection: Connection) -> None: - inspector = inspect(connection) - if not inspector.has_table("active_discounts"): - return - - connection.execute( - text( - "DELETE FROM active_discounts ad " - "WHERE NOT EXISTS (SELECT 1 FROM users u WHERE u.user_id = ad.user_id) " - "OR NOT EXISTS (SELECT 1 FROM promo_codes p WHERE p.promo_code_id = ad.promo_code_id)" - ) - ) - - connection.execute( - text("ALTER TABLE active_discounts DROP CONSTRAINT IF EXISTS active_discounts_user_id_fkey") - ) - connection.execute( - text("ALTER TABLE active_discounts DROP CONSTRAINT IF EXISTS fk_active_discounts_user") - ) - connection.execute( - text("ALTER TABLE active_discounts DROP CONSTRAINT IF EXISTS active_discounts_promo_code_id_fkey") - ) - connection.execute( - text("ALTER TABLE active_discounts DROP CONSTRAINT IF EXISTS fk_active_discounts_promo_code") - ) - - connection.execute( - text( - "ALTER TABLE active_discounts " - "ADD CONSTRAINT fk_active_discounts_user " - "FOREIGN KEY (user_id) REFERENCES users (user_id) ON DELETE CASCADE" - ) - ) - connection.execute( - text( - "ALTER TABLE active_discounts " - "ADD CONSTRAINT fk_active_discounts_promo_code " - "FOREIGN KEY (promo_code_id) REFERENCES promo_codes (promo_code_id) ON DELETE CASCADE" - ) - ) - -MIGRATIONS: List[Migration] = [ - Migration( - id="0001_add_channel_subscription_fields", - description="Add columns to track required channel subscription verification", - upgrade=_migration_0001_add_channel_subscription_fields, - ), - Migration( - id="0002_add_referral_code", - description="Store short referral codes for users and backfill existing rows", - upgrade=_migration_0002_add_referral_code, - ), - Migration( - id="0003_normalize_referral_codes", - description="Normalize referral codes to uppercase for consistent lookups", - upgrade=_migration_0003_normalize_referral_codes, - ), - Migration( - id="0004_add_discount_promo_codes", - description="Add support for percentage discount promo codes", - upgrade=_migration_0004_add_discount_promo_codes, - ), - Migration( - id="0005_fix_active_discounts_fk_cascade", - description="Ensure active_discounts FKs cascade on user/promo delete", - upgrade=_migration_0005_fix_active_discounts_fk_cascade, - ), -] - - -def run_database_migrations(connection: Connection) -> None: - """ - Apply pending migrations sequentially. Already applied revisions are skipped. - """ - _ensure_migrations_table(connection) - - applied_revisions: Set[str] = { - row[0] - for row in connection.execute( - text("SELECT id FROM schema_migrations") - ) - } - - for migration in MIGRATIONS: - if migration.id in applied_revisions: - continue - - logging.info( - "Migrator: applying %s – %s", migration.id, migration.description - ) - try: - with connection.begin_nested(): - migration.upgrade(connection) - connection.execute( - text( - "INSERT INTO schema_migrations (id) VALUES (:revision)" - ), - {"revision": migration.id}, - ) - except Exception as exc: - logging.error( - "Migrator: failed to apply %s (%s)", - migration.id, - migration.description, - exc_info=True, - ) - raise exc - else: - logging.info("Migrator: migration %s applied successfully", migration.id) diff --git a/requirements.txt b/requirements.txt index 9207699..5937431 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,4 +7,5 @@ httpx>=0.27.0 pydantic_settings==2.12.0 sqlalchemy[asyncio]==2.0.45 asyncpg==0.31.0 +alembic==1.18.3 aiocryptopay==0.4.8