document-management/backend/dms/seed_showcase.py

"""幂等补充业务展示分类、方案、附件、挂载和权限。"""

from __future__ import annotations

from pathlib import Path

from flask import Flask, current_app
from sqlalchemy import func, select

from dms import init_dms
from dms.common.enums import (
    AttachmentType,
    CategoryType,
    DocumentStatus,
    DocumentType,
    EnabledStatus,
    SecurityLevel,
)
from dms.extensions import db
from dms.models import (
    AttachmentBinding,
    Category,
    Document,
    Permission,
    User,
)
from dms.seed_documents import _binding, _document, _permission, _required
from dms.storage.paths import ensure_storage_directories


def _category(
    *,
    code: str,
    name: str,
    category_type: CategoryType,
    sort_no: int,
    parent: Category | None = None,
) -> tuple[Category, bool]:
    existing = db.session.scalar(
        select(Category).where(Category.category_code == code)
    )
    if existing is not None:
        if existing.is_deleted or existing.status != EnabledStatus.ENABLED.value:
            raise RuntimeError(f"展示分类{code}已存在但不可用")
        return existing, False

    category_path = (
        f"{parent.category_path.rstrip('/')}/{name}"
        if parent is not None
        else f"/{name}"
    )
    category = Category(
        category_code=code,
        category_name=name,
        category_type=category_type.value,
        parent_id=parent.id if parent is not None else None,
        category_path=category_path,
        sort_no=sort_no,
        document_count=0,
        status=EnabledStatus.ENABLED.value,
    )
    db.session.add(category)
    db.session.flush()
    return category, True


def _refresh_counts(categories: list[Category], mains: list[Document]) -> None:
    for main in mains:
        main.child_count = db.session.scalar(
            select(func.count(Document.id)).where(
                Document.parent_document_id == main.id,
                Document.document_type == DocumentType.SUB_PLAN.value,
                Document.is_deleted.is_(False),
            )
        ) or 0
        main.attachment_count = db.session.scalar(
            select(func.count(AttachmentBinding.id)).where(
                AttachmentBinding.main_document_id == main.id,
                AttachmentBinding.is_deleted.is_(False),
            )
        ) or 0

    for category in categories:
        category.document_count = db.session.scalar(
            select(func.count(Document.id)).where(
                Document.category_id == category.id,
                Document.document_type.in_(
                    [DocumentType.MAIN.value, DocumentType.SUB_PLAN.value]
                ),
                Document.is_deleted.is_(False),
            )
        ) or 0


def seed_showcase() -> tuple[int, int, int, int, int]:
    root = ensure_storage_directories(current_app.config["DMS_STORAGE_ROOT"])
    actor = _required(
        User,
        User.username == "admin",
        User.is_deleted.is_(False),
        description="有效admin用户",
    )
    ordinary_user = _required(
        User,
        User.username == "user",
        User.is_deleted.is_(False),
        description="有效user用户",
    )

    category_specs = (
        ("DEMO_EMERGENCY", "应急预案体系", CategoryType.SCENE, 10, None),
        ("DEMO_NATURAL", "自然灾害应急", CategoryType.STYLE, 10, "DEMO_EMERGENCY"),
        ("DEMO_PUBLIC", "公共安全保障", CategoryType.STYLE, 20, "DEMO_EMERGENCY"),
        ("DEMO_SUPPORT", "保障资源体系", CategoryType.SCENE, 20, None),
        ("DEMO_COMMS", "通信保障", CategoryType.SITUATION, 10, "DEMO_SUPPORT"),
        ("DEMO_MATERIAL", "物资保障", CategoryType.SITUATION, 20, "DEMO_SUPPORT"),
        ("DEMO_RULES", "制度与模板", CategoryType.SCENE, 30, None),
        ("DEMO_STANDARD", "工作规范", CategoryType.OTHER, 10, "DEMO_RULES"),
        ("DEMO_TEMPLATE", "表单模板", CategoryType.OTHER, 20, "DEMO_RULES"),
    )
    categories: dict[str, Category] = {}
    created_categories = 0
    for code, name, category_type, sort_no, parent_code in category_specs:
        category, created = _category(
            code=code,
            name=name,
            category_type=category_type,
            sort_no=sort_no,
            parent=categories.get(parent_code) if parent_code else None,
        )
        categories[code] = category
        created_categories += int(created)

    new_files: list[Path] = []
    created_documents = 0
    existing_documents = 0
    created_bindings = 0
    created_permissions = 0
    try:
        main_specs = (
            (
                "防汛抗旱专项应急预案",
                "面向汛情、城市内涝和旱情的监测预警、响应处置与恢复重建方案。",
                "DEMO_NATURAL",
                SecurityLevel.INTERNAL,
                ["防汛", "抗旱", "应急响应"],
                ("监测预警子方案", "人员转移安置子方案", "抢险救援子方案"),
            ),
            (
                "重大活动通信保障方案",
                "保障重大活动期间指挥链路、无线通信和应急通信资源连续可用。",
                "DEMO_COMMS",
                SecurityLevel.SECRET,
                ["通信", "重大活动", "链路保障"],
                ("核心网络保障子方案", "应急通信车调度子方案"),
            ),
            (
                "应急物资储备与调拨方案",
                "规范应急物资储备布局、库存盘点、需求汇总和跨区域调拨流程。",
                "DEMO_MATERIAL",
                SecurityLevel.INTERNAL,
                ["物资", "储备", "调拨"],
                ("仓储盘点子方案", "跨区域调拨子方案"),
            ),
            (
                "大型活动现场安全保障方案",
                "覆盖人员疏导、重点区域管控、突发事件处置和多部门协同。",
                "DEMO_PUBLIC",
                SecurityLevel.INTERNAL,
                ["现场安全", "活动保障", "协同处置"],
                ("人员疏导子方案", "重点区域管控子方案"),
            ),
        )

        mains: list[Document] = []
        for name, summary, category_code, security, tags, children in main_specs:
            main, created, failed = _document(
                root=root,
                name=name,
                summary=summary,
                document_type=DocumentType.MAIN,
                status=DocumentStatus.PUBLISHED,
                security=security,
                category=categories[category_code],
                actor=actor,
                tags=tags,
                new_files=new_files,
            )
            if failed or main is None:
                raise RuntimeError(f"同名展示主案已逻辑删除：{name}")
            created_documents += int(created)
            existing_documents += int(not created)
            mains.append(main)

            for child_name in children:
                child, child_created, child_failed = _document(
                    root=root,
                    name=child_name,
                    summary=f"《{name}》配套的{child_name}，明确任务分工、实施步骤和保障要求。",
                    document_type=DocumentType.SUB_PLAN,
                    status=DocumentStatus.PUBLISHED,
                    security=security,
                    category=categories[category_code],
                    actor=actor,
                    tags=["子方案", *tags[:2]],
                    new_files=new_files,
                    parent=main,
                    root_document=main,
                )
                if child_failed or child is None:
                    raise RuntimeError(f"同名展示子方案已逻辑删除：{child_name}")
                created_documents += int(child_created)
                existing_documents += int(not child_created)

        attachment_specs = (
            ("防汛应急响应流程图", "防汛响应等级、指挥关系和处置流程示意。", AttachmentType.TYPICAL_CASE),
            ("应急物资需求统计表", "应急物资需求、库存和缺口统计模板。", AttachmentType.SUPPORT_ACHIEVEMENT),
            ("现场处置工作规范", "现场警戒、人员疏导和信息报送工作规范。", AttachmentType.TRAINING_MATERIAL),
            ("应急通讯录模板", "应急联系人、值班电话和协同单位通讯录模板。", AttachmentType.TRAINING_MATERIAL),
            ("跨部门协同处置流程", "跨部门会商、指令下达和反馈闭环流程。", AttachmentType.TYPICAL_CASE),
        )
        attachments: list[Document] = []
        for name, summary, attachment_type in attachment_specs:
            attachment, created, failed = _document(
                root=root,
                name=name,
                summary=summary,
                document_type=DocumentType.ATTACHMENT,
                status=DocumentStatus.PUBLISHED,
                security=SecurityLevel.PUBLIC,
                category=None,
                actor=actor,
                tags=["共享附件", "业务模板"],
                new_files=new_files,
                attachment_type=attachment_type,
            )
            if failed or attachment is None:
                raise RuntimeError(f"同名展示附件已逻辑删除：{name}")
            created_documents += int(created)
            existing_documents += int(not created)
            attachments.append(attachment)

        # 借阅语义（0026）：授权仅面向用户，种子演示保留普通用户的借阅样例
        permission_specs = (
            (mains[0], ordinary_user.id, ordinary_user.real_name),
            (mains[2], ordinary_user.id, ordinary_user.real_name),
        )
        for main, subject_id, subject_name in permission_specs:
            existing_permission = db.session.scalar(
                select(Permission.id).where(
                    Permission.document_id == main.id,
                    Permission.subject_id == subject_id,
                )
            )
            _permission(
                main,
                subject_id=subject_id,
                subject_name=subject_name,
                actor_id=actor.id,
            )
            created_permissions += int(existing_permission is None)

        binding_specs = (
            (mains[0], attachments[0], 10),
            (mains[0], attachments[1], 20),
            (mains[0], attachments[4], 30),
            (mains[1], attachments[3], 10),
            (mains[1], attachments[4], 20),
            (mains[2], attachments[1], 10),
            (mains[2], attachments[3], 20),
            (mains[3], attachments[2], 10),
            (mains[3], attachments[4], 20),
        )
        for main, attachment, sort_no in binding_specs:
            existing_binding = db.session.scalar(
                select(AttachmentBinding.id).where(
                    AttachmentBinding.main_document_id == main.id,
                    AttachmentBinding.attachment_document_id == attachment.id,
                )
            )
            _binding(
                main,
                attachment,
                sort_no=sort_no,
                actor_id=actor.id,
            )
            created_bindings += int(existing_binding is None)

        _refresh_counts(list(categories.values()), mains)
        db.session.commit()
    except Exception:
        db.session.rollback()
        for path in new_files:
            path.unlink(missing_ok=True)
        raise

    return (
        created_categories,
        created_documents,
        existing_documents,
        created_bindings,
        created_permissions,
    )


def main() -> None:
    app = Flask("dms-seed-showcase")
    init_dms(app)
    with app.app_context():
        result = seed_showcase()
    print(
        "业务展示数据初始化完成："
        f"新增分类{result[0]}个，新增文档{result[1]}个，"
        f"已有文档{result[2]}个，新增挂载{result[3]}条，"
        f"新增权限{result[4]}条。"
    )


if __name__ == "__main__":
    main()