document-management/backend/dms/services/category_service.py

"""方案分类树、写操作、路径冗余和占用规则。"""

from __future__ import annotations

from collections import deque
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import and_, exists, func, or_, select, update
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AuditAction,
    AuditTarget,
    CategoryType,
    DocumentStatus,
    DocumentType,
    EnabledStatus,
    RoleCode,
    SECURITY_LEVEL_VALUES,
    SecurityLevel,
    SubjectType,
    VisibilityType,
)
from dms.common.errors import (
    CategoryInUseError,
    CategoryNotLeafError,
    ConflictError,
    InvalidArgumentError,
    ResourceNotFoundError,
)
from dms.common.response import serialize_id
from dms.database.transaction import transaction
from dms.extensions import db
from dms.models import Category, Document, Permission
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.services.authorization_service import _active_organization_ancestor_ids


CREATE_FIELDS = {
    "categoryCode",
    "categoryName",
    "categoryType",
    "parentId",
    "sortNo",
}
UPDATE_FIELDS = {
    "categoryName",
    "categoryType",
    "parentId",
    "sortNo",
    "rowVersion",
}


def _utc_now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _string_id(value: Any, *, field: str, nullable: bool = False) -> int | None:
    if value is None and nullable:
        return None
    if not isinstance(value, str) or not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError(f"{field}必须是正整数形式的字符串ID")
    return int(value)


def _required_text(
    value: Any,
    *,
    field: str,
    max_length: int,
) -> str:
    if not isinstance(value, str) or not value.strip():
        raise InvalidArgumentError(f"{field}不能为空")
    normalized = value.strip()
    if len(normalized) > max_length:
        raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
    return normalized


def _category_type(value: Any) -> str:
    if not isinstance(value, str):
        raise InvalidArgumentError("categoryType必须是字符串")
    try:
        return CategoryType(value).value
    except ValueError as exc:
        raise InvalidArgumentError("categoryType不是有效枚举值") from exc


def _sort_no(value: Any) -> int:
    if type(value) is not int:
        raise InvalidArgumentError("sortNo必须是整数")
    return value


def _row_version(value: Any) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return value


def _exact_payload(payload: Any, expected: set[str]) -> dict[str, Any]:
    if not isinstance(payload, dict) or set(payload) != expected:
        raise InvalidArgumentError(
            "请求体字段必须且只能包含：" + "、".join(sorted(expected))
        )
    return payload


def _path(parent: Category | None, category_name: str) -> str:
    value = (
        f"{parent.category_path.rstrip('/')}/{category_name}"
        if parent is not None
        else f"/{category_name}"
    )
    if len(value) > 1000:
        raise InvalidArgumentError("categoryPath长度超过数据库限制")
    return value


def _dto(
    category: Category,
    *,
    parent_id: int | None = None,
    document_count: int | None = None,
    children: list[dict[str, object]] | None = None,
) -> dict[str, object]:
    return {
        "id": serialize_id(category.id),
        "categoryCode": category.category_code,
        "categoryName": category.category_name,
        "categoryType": category.category_type,
        "parentId": serialize_id(
            category.parent_id if parent_id is None else parent_id
        ),
        "categoryPath": category.category_path,
        "sortNo": category.sort_no,
        "documentCount": (
            category.document_count
            if document_count is None
            else document_count
        ),
        "rowVersion": category.row_version,
        "children": children or [],
    }


def _children_map(categories: list[Category]) -> dict[int | None, list[Category]]:
    children: dict[int | None, list[Category]] = {}
    for category in categories:
        children.setdefault(category.parent_id, []).append(category)
    return children


def _descendants(
    category_id: int,
    children: dict[int | None, list[Category]],
) -> list[Category]:
    result: list[Category] = []
    pending = deque(children.get(category_id, []))
    seen = {category_id}
    while pending:
        category = pending.popleft()
        if category.id in seen:
            raise ConflictError("分类树存在循环关系")
        seen.add(category.id)
        result.append(category)
        pending.extend(children.get(category.id, []))
    return result


def category_tree(
    *,
    keyword: str | None,
    status: str | None,
) -> list[dict[str, object]]:
    if status:
        try:
            normalized_status = EnabledStatus(status).value
        except ValueError as exc:
            raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc
    else:
        normalized_status = EnabledStatus.ENABLED.value

    categories = db.session.scalars(
        select(Category)
        .where(Category.is_deleted.is_(False))
        .order_by(Category.sort_no.asc(), Category.id.asc())
    ).all()
    context = get_auth_context()
    visible_main = select(Document.category_id, func.count(Document.id)).where(
        Document.category_id.is_not(None),
        Document.document_type == DocumentType.MAIN.value,
        Document.is_deleted.is_(False),
    )
    allowed_security_levels = [
        level.value
        for level, rank in SECURITY_LEVEL_VALUES.items()
        if rank <= SECURITY_LEVEL_VALUES[context.security_level]
    ]
    visible_main = visible_main.where(
        Document.security_level.in_(allowed_security_levels)
    )
    if context.role_code == RoleCode.USER:
        ancestor_ids = _active_organization_ancestor_ids(context.organization_id)
        org_permission = exists(
            select(Permission.id).where(
                Permission.document_id == Document.id,
                Permission.is_deleted.is_(False),
                Permission.can_view.is_(True),
                Permission.subject_type == SubjectType.ORG.value,
                Permission.subject_id.in_(ancestor_ids or {-1}),
            )
        )
        user_permission = exists(
            select(Permission.id).where(
                Permission.document_id == Document.id,
                Permission.is_deleted.is_(False),
                Permission.can_view.is_(True),
                Permission.subject_type == SubjectType.USER.value,
                Permission.subject_id == context.user_id,
            )
        )
        visible_main = visible_main.where(
            Document.document_status == DocumentStatus.PUBLISHED.value,
            or_(
                Document.visibility_type == VisibilityType.ALL_AUTHENTICATED.value,
                and_(
                    Document.visibility_type == VisibilityType.ORGANIZATION.value,
                    org_permission,
                ),
                and_(
                    Document.visibility_type == VisibilityType.CUSTOM.value,
                    or_(org_permission, user_permission),
                ),
            ),
        )
    direct_main_counts = dict(
        db.session.execute(
            visible_main.group_by(Document.category_id)
        ).all()
    )
    by_id = {item.id: item for item in categories}
    children = _children_map(categories)
    eligible_ids = {
        item.id for item in categories if item.status == normalized_status
    }

    if keyword and keyword.strip():
        needle = keyword.strip().casefold()
        matched_ids = {
            item.id
            for item in categories
            if item.id in eligible_ids
            and (
                needle in item.category_name.casefold()
                or needle in item.category_code.casefold()
            )
        }
        included_ids = set(matched_ids)
        for matched_id in tuple(matched_ids):
            included_ids.update(
                descendant.id
                for descendant in _descendants(matched_id, children)
                if descendant.id in eligible_ids
            )
            parent_id = by_id[matched_id].parent_id
            while parent_id is not None and parent_id in by_id:
                included_ids.add(parent_id)
                parent_id = by_id[parent_id].parent_id
    else:
        included_ids = eligible_ids

    subtree_counts = {
        category.id: 0
        for category in categories
        if category.status == EnabledStatus.ENABLED.value
    }
    for category_id, count in direct_main_counts.items():
        current_id = category_id
        seen: set[int] = set()
        while current_id in subtree_counts and current_id not in seen:
            seen.add(current_id)
            subtree_counts[current_id] += int(count)
            current_id = by_id[current_id].parent_id

    nodes: dict[int, dict[str, object]] = {}
    for category in categories:
        if category.id in included_ids:
            response_parent_id = (
                category.parent_id if category.parent_id in included_ids else None
            )
            nodes[category.id] = _dto(
                category,
                document_count=subtree_counts.get(category.id, 0),
            )
            nodes[category.id]["parentId"] = serialize_id(response_parent_id)

    roots: list[dict[str, object]] = []
    for category in categories:
        node = nodes.get(category.id)
        if node is None:
            continue
        if category.parent_id in nodes:
            nodes[category.parent_id]["children"].append(node)
        else:
            roots.append(node)
    return roots


def _valid_parent(session, parent_id: int | None) -> Category | None:
    if parent_id is None:
        return None
    parent = session.scalar(
        select(Category)
        .where(
            Category.id == parent_id,
            Category.is_deleted.is_(False),
            Category.status == EnabledStatus.ENABLED.value,
        )
        .with_for_update()
    )
    if parent is None:
        raise InvalidArgumentError("parentId必须指向有效且已启用的分类")
    direct_main_count = session.scalar(
        select(func.count(Document.id)).where(
            Document.category_id == parent.id,
            Document.document_type == DocumentType.MAIN.value,
            Document.is_deleted.is_(False),
        )
    ) or 0
    if direct_main_count:
        child_count = session.scalar(
            select(func.count(Category.id)).where(
                Category.parent_id == parent.id,
                Category.is_deleted.is_(False),
                Category.status == EnabledStatus.ENABLED.value,
            )
        ) or 0
        raise CategoryNotLeafError(
            details={
                "categoryId": serialize_id(parent.id),
                "childCategoryCount": int(child_count),
            }
        )
    return parent


def create_category(payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, CREATE_FIELDS)
    category_code = _required_text(
        body["categoryCode"],
        field="categoryCode",
        max_length=64,
    )
    category_name = _required_text(
        body["categoryName"],
        field="categoryName",
        max_length=128,
    )
    category_type = _category_type(body["categoryType"])
    parent_id = _string_id(body["parentId"], field="parentId", nullable=True)
    sort_no = _sort_no(body["sortNo"])
    actor_id = get_auth_context().user_id

    try:
        with transaction() as session:
            if session.scalar(
                select(Category.id).where(Category.category_code == category_code)
            ) is not None:
                raise ConflictError("分类编码已存在")
            parent = _valid_parent(session, parent_id)
            category = Category(
                category_code=category_code,
                category_name=category_name,
                category_type=category_type,
                parent_id=parent_id,
                category_path=_path(parent, category_name),
                sort_no=sort_no,
                document_count=0,
                status=EnabledStatus.ENABLED.value,
                created_by=actor_id,
                updated_by=actor_id,
            )
            session.add(category)
            session.flush()
            session.add(
                business_audit(
                    action=AuditAction.CREATE_CATEGORY,
                    target=AuditTarget.CATEGORY,
                    target_id=category.id,
                    target_name=category.category_name,
                    detail={
                        "categoryCode": category.category_code,
                        "categoryType": category.category_type,
                        "parentId": serialize_id(category.parent_id),
                        "categoryPath": category.category_path,
                        "sortNo": category.sort_no,
                    },
                )
            )
        return _dto(category)
    except IntegrityError as exc:
        raise ConflictError("分类编码已存在") from exc


def _locked_categories(session) -> list[Category]:
    return session.scalars(
        select(Category)
        .where(Category.is_deleted.is_(False))
        .order_by(Category.id.asc())
        .with_for_update()
    ).all()


def update_category(category_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, UPDATE_FIELDS)
    category_name = _required_text(
        body["categoryName"],
        field="categoryName",
        max_length=128,
    )
    category_type = _category_type(body["categoryType"])
    parent_id = _string_id(body["parentId"], field="parentId", nullable=True)
    sort_no = _sort_no(body["sortNo"])
    expected_version = _row_version(body["rowVersion"])
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        categories = _locked_categories(session)
        by_id = {item.id: item for item in categories}
        category = by_id.get(category_id)
        if category is None:
            raise ResourceNotFoundError("分类不存在")
        if category.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": category.row_version},
            )

        children = _children_map(categories)
        descendants = _descendants(category.id, children)
        descendant_ids = {item.id for item in descendants}
        if parent_id == category.id:
            raise InvalidArgumentError("parentId不能等于当前分类ID")
        if parent_id in descendant_ids:
            raise InvalidArgumentError("分类不能移动到自己的后代节点下")
        parent = None
        if parent_id is not None:
            parent = by_id.get(parent_id)
            if (
                parent is None
                or parent.is_deleted
                or parent.status != EnabledStatus.ENABLED.value
            ):
                raise InvalidArgumentError("parentId必须指向有效且已启用的分类")
            if parent_id != category.parent_id:
                direct_main_count = session.scalar(
                    select(func.count(Document.id)).where(
                        Document.category_id == parent.id,
                        Document.document_type == DocumentType.MAIN.value,
                        Document.is_deleted.is_(False),
                    )
                ) or 0
                if direct_main_count:
                    child_count = session.scalar(
                        select(func.count(Category.id)).where(
                            Category.parent_id == parent.id,
                            Category.is_deleted.is_(False),
                            Category.status == EnabledStatus.ENABLED.value,
                        )
                    ) or 0
                    raise CategoryNotLeafError(
                        details={
                            "categoryId": serialize_id(parent.id),
                            "childCategoryCount": int(child_count),
                        }
                    )

        before = {
            "categoryName": category.category_name,
            "categoryType": category.category_type,
            "parentId": serialize_id(category.parent_id),
            "categoryPath": category.category_path,
            "sortNo": category.sort_no,
            "rowVersion": category.row_version,
        }
        path_changed = (
            category_name != category.category_name
            or parent_id != category.parent_id
        )
        category.category_name = category_name
        category.category_type = category_type
        category.parent_id = parent_id
        category.category_path = _path(parent, category_name)
        category.sort_no = sort_no
        category.updated_by = actor_id
        category.updated_at = now
        category.row_version += 1

        affected_categories = [category]
        if path_changed:
            pending = deque([category])
            while pending:
                current = pending.popleft()
                for child in children.get(current.id, []):
                    child.category_path = _path(current, child.category_name)
                    child.updated_by = actor_id
                    child.updated_at = now
                    child.row_version += 1
                    affected_categories.append(child)
                    pending.append(child)

        affected_document_count = 0
        if path_changed:
            for affected in affected_categories:
                result = session.execute(
                    update(Document)
                    .where(
                        Document.category_id == affected.id,
                        Document.is_deleted.is_(False),
                        Document.document_type.in_(
                            [
                                DocumentType.MAIN.value,
                                DocumentType.SUB_PLAN.value,
                            ]
                        ),
                    )
                    .values(
                        category_name=affected.category_name,
                        category_path=affected.category_path,
                        updated_by=actor_id,
                        updated_at=now,
                        row_version=Document.row_version + 1,
                    )
                )
                affected_document_count += result.rowcount

        after = {
            "categoryName": category.category_name,
            "categoryType": category.category_type,
            "parentId": serialize_id(category.parent_id),
            "categoryPath": category.category_path,
            "sortNo": category.sort_no,
            "rowVersion": category.row_version,
        }
        session.add(
            business_audit(
                action=AuditAction.EDIT_CATEGORY,
                target=AuditTarget.CATEGORY,
                target_id=category.id,
                target_name=category.category_name,
                detail={
                    "before": before,
                    "after": after,
                    "updatedDescendantCount": len(affected_categories) - 1,
                    "updatedDocumentCount": affected_document_count,
                },
            )
        )
    return _dto(category)


def delete_category(category_id: int, row_version: int) -> dict[str, object]:
    actor_id = get_auth_context().user_id
    now = _utc_now()
    with transaction() as session:
        categories = _locked_categories(session)
        by_id = {item.id: item for item in categories}
        category = by_id.get(category_id)
        if category is None:
            raise ResourceNotFoundError("分类不存在")
        if category.row_version != row_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": category.row_version},
            )

        descendants = _descendants(category.id, _children_map(categories))
        child_count = len(descendants)
        document_count = (
            session.scalar(
                select(func.count(Document.id)).where(
                    Document.category_id == category.id,
                    Document.is_deleted.is_(False),
                    Document.document_type.in_(
                        [
                            DocumentType.MAIN.value,
                            DocumentType.SUB_PLAN.value,
                        ]
                    ),
                )
            )
            or 0
        )
        if child_count or document_count:
            raise CategoryInUseError(
                details={
                    "categoryId": serialize_id(category.id),
                    "childCategoryCount": child_count,
                    "documentCount": document_count,
                }
            )

        target_name = category.category_name
        category.is_deleted = True
        category.deleted_at = now
        category.updated_at = now
        category.updated_by = actor_id
        category.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.DELETE_CATEGORY,
                target=AuditTarget.CATEGORY,
                target_id=category.id,
                target_name=target_name,
                detail={
                    "categoryCode": category.category_code,
                    "categoryPath": category.category_path,
                    "rowVersion": category.row_version,
                },
            )
        )
    return {"id": serialize_id(category.id), "deleted": True}


def parse_category_id(value: str) -> int:
    result = _string_id(value, field="id")
    assert result is not None
    return result


def parse_delete_row_version(value: str | None) -> int:
    if value is None or not value.isdecimal():
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return int(value)