"""用户界面主题偏好：sys_user 加 ui_theme 列（契约1.20）。

Revision ID: 0027_ui_theme
Revises: 0026_borrow_permission
Create Date: 2026-09-03

变更要点：
1. 主题为账号级漫游偏好：任意设备登录后跟随账号；blue 为缺省
   （存量用户与未设置路径统一回退蓝色）；
2. CHECK 白名单 blue/red/green（convention 渲染名
   ck_sys_user_user_ui_theme，与模型 CheckConstraint(name="user_ui_theme") 一致）；
3. 达梦兼容：探测统一走 SQLAlchemy inspector、约束名大小写不敏感比较，
   删除以库中实际名执行（达梦渲染为大写）；
4. 升级幂等：列/约束先探测，与 Flyway V12 并行执行互不冲突；
   降级删列（偏好可丢失，无审计意义）。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0027_ui_theme"
down_revision: str | None = "0026_borrow_permission"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_TABLE = "sys_user"
_COLUMN = "ui_theme"
# naming convention 渲染后的最终约束名（ck_%(table)s_%(constraint_name)s）
_CHECK_NAME = "ck_sys_user_user_ui_theme"


def _columns() -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {column["name"] for column in inspector.get_columns(_TABLE)}


def _check_exists() -> bool:
    existing = {
        item["name"].lower()
        for item in sa.inspect(op.get_bind()).get_check_constraints(_TABLE)
    }
    return _CHECK_NAME in existing


def _drop_check() -> None:
    # 达梦约束名为大写，以库中实际名执行删除。
    for item in sa.inspect(op.get_bind()).get_check_constraints(_TABLE):
        if item["name"].lower() == _CHECK_NAME:
            op.get_bind().execute(
                sa.text(f"ALTER TABLE {_TABLE} DROP CONSTRAINT {item['name']}")
            )
            return


def upgrade() -> None:
    if _COLUMN not in _columns():
        op.add_column(
            _TABLE,
            sa.Column(
                _COLUMN,
                sa.String(length=12),
                nullable=False,
                server_default=sa.text("'blue'"),
            ),
        )
    if not _check_exists():
        op.create_check_constraint(
            "user_ui_theme",
            _TABLE,
            "ui_theme IN ('blue', 'red', 'green')",
        )


def downgrade() -> None:
    _drop_check()
    if _COLUMN in _columns():
        op.drop_column(_TABLE, _COLUMN)