document-management/backend/migrations/versions/0020_sub_plan_to_attachment.py

"""配套方案并入共享资料库：SUB_PLAN → ATTACHMENT(ACCOMPANYING_PLAN)。

Revision ID: 0020_sub_plan_to_attachment
Revises: 0019_content_extract_error
Create Date: 2026-08-29

变更要点：
1. doc_document 附件类型CHECK扩四值：新名 ck_doc_document_attachment_type_v2
   （旧名 DROP IF EXISTS，新名 ADD IF NOT EXISTS，按名幂等，与 Flyway V5 一致）；
2. 同名消歧：存活子案与存活附件同名、或与更小id存活子案同名者，追加"（id）"后缀；
3. 全部 SUB_PLAN（含软删）→ ATTACHMENT + attachment_type=ACCOMPANYING_PLAN，
   visibility 统一 ALL_AUTHENTICATED（附件可见性口径）；
4. 自有子案关系转挂载：存活子案按 parent_document_id 补 doc_attachment_binding
   （sort_no 沿用原排序，NOT EXISTS 防重）；
5. doc_sub_plan_binding 活跃行转 doc_attachment_binding 后软删（防重跑双转）；
6. 重算全部主案 attachment_count；child_count 清零（子案不再按父子计数）；
7. 降级保护：已产生 ACCOMPANYING_PLAN 数据时拒绝有损回退。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0020_sub_plan_to_attachment"
down_revision: str | None = "0019_content_extract_error"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

_OLD_CHECK = "ck_doc_document_document_attachment_type"
_NEW_CHECK = "ck_doc_document_attachment_type_v2"
_NEW_CHECK_DDL = (
    "ALTER TABLE doc_document ADD CONSTRAINT " + _NEW_CHECK + " "
    "CHECK (attachment_type IS NULL OR attachment_type IN "
    "('SUPPORT_ACHIEVEMENT','TRAINING_MATERIAL','TYPICAL_CASE','ACCOMPANYING_PLAN'))"
)

# 附件shape约束：放宽 security_level（配套方案保留原密级，读路径由密级前置拦截）
_OLD_SHAPE = "ck_doc_document_document_type_shape"
_NEW_SHAPE = "ck_doc_document_type_shape_v2"
_NEW_SHAPE_DDL = (
    "ALTER TABLE doc_document ADD CONSTRAINT " + _NEW_SHAPE + " CHECK ("
    "((document_type = 'MAIN') AND (parent_document_id IS NULL) "
    "AND (category_id IS NOT NULL) AND (attachment_type IS NULL)) "
    "OR ((document_type = 'SUB_PLAN') AND (parent_document_id IS NOT NULL) "
    "AND (root_document_id IS NOT NULL) AND (category_id IS NOT NULL) "
    "AND (attachment_type IS NULL)) "
    "OR ((document_type = 'ATTACHMENT') AND (parent_document_id IS NULL) "
    "AND (root_document_id IS NULL) AND (category_id IS NULL) "
    "AND (attachment_type IS NOT NULL) "
    "AND (visibility_type = 'ALL_AUTHENTICATED')))"
)


def _constraint_exists(name: str) -> bool:
    bind = op.get_bind()
    return bool(
        bind.scalar(
            sa.text(
                "SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS "
                "WHERE TABLE_SCHEMA = DATABASE() "
                "AND TABLE_NAME = 'doc_document' AND CONSTRAINT_NAME = :name"
            ),
            {"name": name},
        )
    )


def _run(statement: str) -> None:
    op.get_bind().execute(sa.text(statement))


def upgrade() -> None:
    # 1) CHECK 约束扩四值（按名幂等）
    if _constraint_exists(_OLD_CHECK):
        _run(f"ALTER TABLE doc_document DROP CHECK {_OLD_CHECK}")
    if not _constraint_exists(_NEW_CHECK):
        _run(_NEW_CHECK_DDL)
    # 1b) shape约束放宽：附件允许任意密级（存量子案密级保留，读路径密级前置）
    if _constraint_exists(_OLD_SHAPE):
        _run(f"ALTER TABLE doc_document DROP CHECK {_OLD_SHAPE}")
    if not _constraint_exists(_NEW_SHAPE):
        _run(_NEW_SHAPE_DDL)

    # 2) 同名消歧（仅存活行；后缀含id保证确定性且可重跑无冲突）。
    #    候选id先落临时表再JOIN更新，规避 MySQL 1093（UPDATE 目标表不可出现在子查询）。
    _run("DROP TEMPORARY TABLE IF EXISTS tmp_0020_sub_dedupe")
    _run(
        "CREATE TEMPORARY TABLE tmp_0020_sub_dedupe (id BIGINT PRIMARY KEY)"
    )
    _run(
        """
        INSERT INTO tmp_0020_sub_dedupe (id)
        SELECT a.id FROM doc_document a
        WHERE a.document_type = 'SUB_PLAN' AND a.is_deleted = 0 AND (
            EXISTS (
                SELECT 1 FROM doc_document b
                WHERE b.is_deleted = 0 AND b.document_type = 'ATTACHMENT'
                  AND b.document_name = a.document_name
            )
            OR EXISTS (
                SELECT 1 FROM doc_document c
                WHERE c.document_type = 'SUB_PLAN'
                  AND c.document_name = a.document_name AND c.id < a.id
            )
        )
        """
    )
    _run(
        """
        UPDATE doc_document d JOIN tmp_0020_sub_dedupe t ON d.id = t.id
        SET d.document_name = CONCAT(d.document_name, '（', d.id, '）')
        """
    )
    _run("DROP TEMPORARY TABLE IF EXISTS tmp_0020_sub_dedupe")

    # 3) 自有子案关系先转挂载（需 parent_document_id，父主案存活的存活子案）
    _run(
        """
        INSERT INTO doc_attachment_binding
            (main_document_id, attachment_document_id, sort_no, created_by, updated_by)
        SELECT d.parent_document_id, d.id, d.sort_no, d.created_by, d.updated_by
        FROM doc_document d
        WHERE d.document_type = 'SUB_PLAN'
          AND d.is_deleted = 0
          AND d.parent_document_id IS NOT NULL
          AND EXISTS (
              SELECT 1 FROM doc_document m
              WHERE m.id = d.parent_document_id AND m.is_deleted = 0
          )
          AND NOT EXISTS (
              SELECT 1 FROM doc_attachment_binding b
              WHERE b.main_document_id = d.parent_document_id
                AND b.attachment_document_id = d.id AND b.is_deleted = 0
          )
        """
    )

    # 4) 子案跨主案挂载关系转附件挂载后软删（防重跑双转）
    _run(
        """
        INSERT INTO doc_attachment_binding
            (main_document_id, attachment_document_id, sort_no, created_by, updated_by)
        SELECT s.main_document_id, s.sub_document_id, s.sort_no, s.created_by, s.updated_by
        FROM doc_sub_plan_binding s
        WHERE s.is_deleted = 0
          AND NOT EXISTS (
              SELECT 1 FROM doc_attachment_binding b
              WHERE b.main_document_id = s.main_document_id
                AND b.attachment_document_id = s.sub_document_id AND b.is_deleted = 0
          )
        """
    )
    _run(
        """
        UPDATE doc_sub_plan_binding
        SET is_deleted = 1, deleted_at = NOW(3), updated_at = NOW(3),
            row_version = row_version + 1
        WHERE is_deleted = 0
        """
    )

    # 5) 类型转换（全部 SUB_PLAN，含软删，重跑自然无目标行）：
    #    置空父/根/分类（附件shape），保留密级（读路径密级前置拦截）
    _run(
        """
        UPDATE doc_document
        SET document_type = 'ATTACHMENT',
            attachment_type = 'ACCOMPANYING_PLAN',
            visibility_type = 'ALL_AUTHENTICATED',
            parent_document_id = NULL,
            root_document_id = NULL,
            category_id = NULL,
            category_name = NULL,
            category_path = NULL
        WHERE document_type = 'SUB_PLAN'
        """
    )

    # 6) 计数重算与清零
    _run(
        """
        UPDATE doc_document m
        SET m.attachment_count = (
            SELECT COUNT(*) FROM doc_attachment_binding b
            WHERE b.main_document_id = m.id AND b.is_deleted = 0
        )
        WHERE m.document_type = 'MAIN'
          AND m.attachment_count <> (
            SELECT COUNT(*) FROM doc_attachment_binding b
            WHERE b.main_document_id = m.id AND b.is_deleted = 0
          )
        """
    )
    _run(
        "UPDATE doc_document SET child_count = 0 "
        "WHERE document_type = 'MAIN' AND child_count <> 0"
    )


def downgrade() -> None:
    bind = op.get_bind()
    converted = bind.scalar(
        sa.text(
            "SELECT COUNT(*) FROM doc_document "
            "WHERE attachment_type = 'ACCOMPANYING_PLAN'"
        )
    )
    if converted:
        raise RuntimeError(
            "0020已产生配套方案（ACCOMPANYING_PLAN）数据，拒绝有损降级；"
            "请先备份并按数据治理流程处理"
        )
    if _constraint_exists(_NEW_CHECK):
        _run(f"ALTER TABLE doc_document DROP CHECK {_NEW_CHECK}")
    if not _constraint_exists(_OLD_CHECK):
        _run(
            "ALTER TABLE doc_document ADD CONSTRAINT " + _OLD_CHECK + " "
            "CHECK (attachment_type IS NULL OR attachment_type IN "
            "('SUPPORT_ACHIEVEMENT','TRAINING_MATERIAL','TYPICAL_CASE'))"
        )
    # shape约束回旧版（附件强制PUBLIC密级）；已拒绝并轨数据，安全恢复
    if _constraint_exists(_NEW_SHAPE):
        _run(f"ALTER TABLE doc_document DROP CHECK {_NEW_SHAPE}")
    if not _constraint_exists(_OLD_SHAPE):
        _run(
            "ALTER TABLE doc_document ADD CONSTRAINT " + _OLD_SHAPE + " "
            "CHECK (((document_type = 'MAIN') AND (parent_document_id IS NULL) "
            "AND (category_id IS NOT NULL) AND (attachment_type IS NULL)) "
            "OR ((document_type = 'SUB_PLAN') AND (parent_document_id IS NOT NULL) "
            "AND (root_document_id IS NOT NULL) AND (category_id IS NOT NULL) "
            "AND (attachment_type IS NULL)) "
            "OR ((document_type = 'ATTACHMENT') AND (parent_document_id IS NULL) "
            "AND (root_document_id IS NULL) AND (category_id IS NULL) "
            "AND (attachment_type IS NOT NULL) "
            "AND (security_level = 'PUBLIC') "
            "AND (visibility_type = 'ALL_AUTHENTICATED')))"
        )