document-management/backend/migrations/versions/0003_q3_business_alignment_and_content_search.py

"""Q3业务规则对齐、角色收敛和正文检索字段。

Revision ID: 0003_q3_business_alignment_and_content_search
Revises: 0002_add_restore_audit_action
Create Date: 2026-08-07
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

revision: str = "0003_q3_business_alignment_and_content_search"
down_revision: str | None = "0002_add_restore_audit_action"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

ROLE_CONSTRAINT = "user_role_code"
CONTENT_STATUS_CONSTRAINT = "document_content_extract_status"


def upgrade() -> None:
    # 本阶段按契约使用可读revision名，长度超过Alembic默认VARCHAR(32)。
    op.alter_column(
        "alembic_version",
        "version_num",
        existing_type=sa.String(length=32),
        type_=sa.String(length=128),
        existing_nullable=False,
    )
    op.add_column(
        "doc_document",
        sa.Column("content_text", mysql.LONGTEXT(), nullable=True),
    )
    op.add_column(
        "doc_document",
        sa.Column(
            "content_extract_status",
            sa.String(length=32),
            server_default=sa.text("'UNSUPPORTED'"),
            nullable=False,
        ),
    )
    op.add_column(
        "doc_document",
        sa.Column("content_extracted_at", mysql.DATETIME(fsp=3), nullable=True),
    )
    op.alter_column(
        "doc_document",
        "search_text",
        existing_type=sa.Text(),
        type_=mysql.LONGTEXT(),
        existing_nullable=True,
    )
    op.create_check_constraint(
        CONTENT_STATUS_CONSTRAINT,
        "doc_document",
        "content_extract_status IN "
        "('SUCCESS', 'EMPTY', 'UNSUPPORTED', 'FAILED')",
    )

    # 历史审计员必须先使全部旧Token失效，再禁用并逻辑删除；绝不升级为管理员。
    op.execute(
        sa.text(
            "UPDATE sys_user SET "
            "auth_version = auth_version + 1, "
            "row_version = row_version + 1, "
            "status = 'DISABLED', "
            "is_deleted = 1, "
            "deleted_at = COALESCE(deleted_at, CURRENT_TIMESTAMP(3)), "
            "updated_at = CURRENT_TIMESTAMP(3), "
            "role_code = 'USER' "
            "WHERE role_code = 'AUDITOR'"
        )
    )
    op.drop_constraint(ROLE_CONSTRAINT, "sys_user", type_="check")
    op.create_check_constraint(
        ROLE_CONSTRAINT,
        "sys_user",
        "role_code IN ('USER', 'ADMIN')",
    )


def downgrade() -> None:
    bind = op.get_bind()
    extracted_count = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM doc_document WHERE "
            "content_text IS NOT NULL OR content_extracted_at IS NOT NULL OR "
            "content_extract_status <> 'UNSUPPORTED'"
        )
    )
    oversized_search_count = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM doc_document "
            "WHERE OCTET_LENGTH(search_text) > 65535"
        )
    )
    if extracted_count or oversized_search_count:
        raise RuntimeError(
            "0003已包含正文提取结果或超出TEXT容量的检索文本，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )

    op.drop_constraint(ROLE_CONSTRAINT, "sys_user", type_="check")
    op.create_check_constraint(
        ROLE_CONSTRAINT,
        "sys_user",
        "role_code IN ('USER', 'ADMIN', 'AUDITOR')",
    )
    # 已失效的历史AUDITOR不会被自动复活或重新授权。

    op.drop_constraint(
        CONTENT_STATUS_CONSTRAINT,
        "doc_document",
        type_="check",
    )
    op.alter_column(
        "doc_document",
        "search_text",
        existing_type=mysql.LONGTEXT(),
        type_=sa.Text(),
        existing_nullable=True,
    )
    op.drop_column("doc_document", "content_extracted_at")
    op.drop_column("doc_document", "content_extract_status")
    op.drop_column("doc_document", "content_text")