Merge pull request #705 from Fr1ngg/bedolaga/add-alembic-migration-for-system_settings_history-table

Add migration for system settings history audit table

app/database/universal_migration.py

@@ -1998,6 +1998,74 @@ async def create_system_settings_table() -> bool:
         return False
 
 
+async def create_system_settings_history_table() -> bool:
+    table_name = "system_settings_history"
+    table_exists = await check_table_exists(table_name)
+    if table_exists:
+        logger.info("ℹ️ Таблица system_settings_history уже существует")
+        return True
+
+    try:
+        async with engine.begin() as conn:
+            db_type = await get_database_type()
+
+            if db_type == "sqlite":
+                create_sql = """
+                CREATE TABLE system_settings_history (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    key VARCHAR(255) NOT NULL,
+                    old_value TEXT NULL,
+                    new_value TEXT NULL,
+                    changed_by INTEGER NULL,
+                    changed_by_username VARCHAR(255) NULL,
+                    source VARCHAR(50) NOT NULL DEFAULT 'bot',
+                    reason VARCHAR(255) NULL,
+                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+                );
+                CREATE INDEX ix_system_settings_history_key ON system_settings_history(key);
+                """
+            elif db_type == "postgresql":
+                create_sql = """
+                CREATE TABLE system_settings_history (
+                    id SERIAL PRIMARY KEY,
+                    key VARCHAR(255) NOT NULL,
+                    old_value TEXT NULL,
+                    new_value TEXT NULL,
+                    changed_by INTEGER NULL,
+                    changed_by_username VARCHAR(255) NULL,
+                    source VARCHAR(50) NOT NULL DEFAULT 'bot',
+                    reason VARCHAR(255) NULL,
+                    created_at TIMESTAMP DEFAULT NOW()
+                );
+                CREATE INDEX ix_system_settings_history_key ON system_settings_history(key);
+                """
+            else:
+                create_sql = """
+                CREATE TABLE system_settings_history (
+                    id INT AUTO_INCREMENT PRIMARY KEY,
+                    key VARCHAR(255) NOT NULL,
+                    old_value TEXT NULL,
+                    new_value TEXT NULL,
+                    changed_by INT NULL,
+                    changed_by_username VARCHAR(255) NULL,
+                    source VARCHAR(50) NOT NULL DEFAULT 'bot',
+                    reason VARCHAR(255) NULL,
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                ) ENGINE=InnoDB;
+                CREATE INDEX ix_system_settings_history_key ON system_settings_history(key);
+                """
+
+            await conn.execute(text(create_sql))
+            logger.info("✅ Таблица system_settings_history создана")
+            return True
+
+    except Exception as error:
+        logger.error(
+            f"❌ Ошибка создания таблицы system_settings_history: {error}"
+        )
+        return False
+
+
 async def create_web_api_tokens_table() -> bool:
     table_exists = await check_table_exists("web_api_tokens")
     if table_exists:
@@ -2153,6 +2221,13 @@ async def run_universal_migration():
         else:
             logger.warning("⚠️ Проблемы с таблицей system_settings")
 
+        logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ SYSTEM_SETTINGS_HISTORY ===")
+        system_settings_history_ready = await create_system_settings_history_table()
+        if system_settings_history_ready:
+            logger.info("✅ Таблица system_settings_history готова")
+        else:
+            logger.warning("⚠️ Проблемы с таблицей system_settings_history")
+
         logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ WEB_API_TOKENS ===")
         web_api_tokens_ready = await create_web_api_tokens_table()
         if web_api_tokens_ready:

migrations/alembic/versions/2b1c8fd0dc42_add_system_settings_history_table.py

@@ -0,0 +1,69 @@
+"""create system settings history table"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.engine.reflection import Inspector
+
+
+revision: str = "2b1c8fd0dc42"
+down_revision: Union[str, None] = "8fd1e338eb45"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+TABLE_NAME = "system_settings_history"
+KEY_INDEX = "ix_system_settings_history_key"
+
+
+def _table_exists(inspector: Inspector) -> bool:
+    return TABLE_NAME in inspector.get_table_names()
+
+
+def _index_exists(inspector: Inspector, index_name: str) -> bool:
+    try:
+        indexes = inspector.get_indexes(TABLE_NAME)
+    except sa.exc.NoSuchTableError:
+        return False
+    return any(index.get("name") == index_name for index in indexes)
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    table_created = False
+
+    if not _table_exists(inspector):
+        op.create_table(
+            TABLE_NAME,
+            sa.Column("id", sa.Integer(), primary_key=True),
+            sa.Column("key", sa.String(length=255), nullable=False),
+            sa.Column("old_value", sa.Text(), nullable=True),
+            sa.Column("new_value", sa.Text(), nullable=True),
+            sa.Column("changed_by", sa.Integer(), nullable=True),
+            sa.Column("changed_by_username", sa.String(length=255), nullable=True),
+            sa.Column(
+                "source",
+                sa.String(length=50),
+                nullable=False,
+                server_default=sa.text("'bot'"),
+            ),
+            sa.Column("reason", sa.String(length=255), nullable=True),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()),
+        )
+        table_created = True
+
+    if table_created or not _index_exists(inspector, KEY_INDEX):
+        op.create_index(KEY_INDEX, TABLE_NAME, ["key"])
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+
+    if _table_exists(inspector):
+        if _index_exists(inspector, KEY_INDEX):
+            op.drop_index(KEY_INDEX, table_name=TABLE_NAME)
+        op.drop_table(TABLE_NAME)