"""方案分类树、写操作、路径冗余和占用规则。""" 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)