"""组织树查询、写操作、路径冗余与占用规则。""" from __future__ import annotations from collections import deque from datetime import datetime, timezone from typing import Any from sqlalchemy import func, select, update from sqlalchemy.exc import IntegrityError from dms.common.enums import AuditAction, AuditTarget, EnabledStatus from dms.common.errors import ( ConflictError, InvalidArgumentError, OrganizationInUseError, ResourceNotFoundError, ) from dms.common.response import serialize_id from dms.database.transaction import transaction from dms.extensions import db from dms.models import Document, Organization, OrgIpAssignment, User from dms.security.auth_context import get_auth_context from dms.services.admin_permission import ( ensure_org_in_managed_scope, require_organization_editor, ) from dms.services.audit_service import business_audit CREATE_FIELDS = { "orgCode", "orgName", "parentId", "sortNo", "status", } UPDATE_FIELDS = { "orgCode", "orgName", "parentId", "sortNo", "status", "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 _status_value(value: Any) -> str: if not isinstance(value, str): raise InvalidArgumentError("status必须是字符串") try: return EnabledStatus(value).value except ValueError as exc: raise InvalidArgumentError("status必须是ENABLED或DISABLED") 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 _build_path(parent_path: str | None, org_name: str) -> str: """根据父节点路径构建当前节点的 org_path,并校验长度上限。""" value = ( f"{parent_path.rstrip('/')}/{org_name}" if parent_path is not None else f"/{org_name}" ) if len(value) > 1000: raise InvalidArgumentError("orgPath长度超过数据库限制") return value def _org_path(parent: Organization | None, org_name: str) -> str: return _build_path(parent.org_path if parent is not None else None, org_name) def _dto( organization: Organization, *, children: list[dict[str, object]] | None = None, ) -> dict[str, object]: return { "id": serialize_id(organization.id), "parentId": serialize_id(organization.parent_id), "orgCode": organization.org_code, "orgName": organization.org_name, "orgPath": organization.org_path, "sortNo": organization.sort_no, "status": organization.status, "rowVersion": organization.row_version, "children": children or [], } def _children_map( organizations: list[Organization], ) -> dict[int | None, list[Organization]]: children: dict[int | None, list[Organization]] = {} for organization in organizations: children.setdefault(organization.parent_id, []).append(organization) return children def _descendants( organization_id: int, children: dict[int | None, list[Organization]], ) -> list[Organization]: result: list[Organization] = [] pending = deque(children.get(organization_id, [])) seen = {organization_id} while pending: organization = pending.popleft() if organization.id in seen: raise ConflictError("机构树存在循环关系") seen.add(organization.id) result.append(organization) pending.extend(children.get(organization.id, [])) return result def organization_tree( *, keyword: str | None, status: str | None ) -> list[dict[str, object]]: organizations = db.session.scalars( select(Organization) .where(Organization.is_deleted.is_(False)) .order_by(Organization.sort_no.asc(), Organization.id.asc()) ).all() by_id = {item.id: item for item in organizations} 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 base_ids = {item.id for item in organizations if item.status == normalized_status} if keyword and keyword.strip(): needle = keyword.strip().casefold() base_ids = { item.id for item in organizations if item.id in base_ids and (needle in item.org_name.casefold() or needle in item.org_code.casefold()) } included_ids = set(base_ids) for item_id in tuple(base_ids): parent_id = by_id[item_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 nodes = { item.id: _dto(item) for item in organizations if item.id in included_ids } roots: list[dict[str, object]] = [] for item in organizations: if item.id not in nodes: continue node = nodes[item.id] if item.parent_id in nodes: nodes[item.parent_id]["children"].append(node) else: roots.append(node) return roots def _locked_organizations(session) -> list[Organization]: return session.scalars( select(Organization) .where(Organization.is_deleted.is_(False)) .order_by(Organization.id.asc()) .with_for_update() ).all() def _valid_parent( session, parent_id: int | None ) -> Organization | None: if parent_id is None: return None parent = session.scalar( select(Organization) .where( Organization.id == parent_id, Organization.is_deleted.is_(False), ) .with_for_update() ) if parent is None: raise InvalidArgumentError("parentId必须指向有效机构") if parent.status != EnabledStatus.ENABLED.value: raise InvalidArgumentError("parentId必须指向已启用的机构") return parent def create_organization(payload: Any) -> dict[str, object]: body = _exact_payload(payload, CREATE_FIELDS) org_code = _required_text(body["orgCode"], field="orgCode", max_length=64) org_name = _required_text(body["orgName"], field="orgName", max_length=128) parent_id = _string_id(body["parentId"], field="parentId", nullable=True) sort_no = _sort_no(body["sortNo"]) status = _status_value(body["status"]) actor_id = get_auth_context().user_id try: with transaction() as session: actor = require_organization_editor(session) ensure_org_in_managed_scope(session, actor, parent_id) if ( session.scalar( select(Organization.id) .where( Organization.org_code == org_code, Organization.is_deleted.is_(False), ) ) is not None ): raise ConflictError("机构编码已存在") parent = _valid_parent(session, parent_id) organization = Organization( org_code=org_code, org_name=org_name, parent_id=parent_id, org_path=_org_path(parent, org_name), sort_no=sort_no, status=status, created_by=actor_id, updated_by=actor_id, ) session.add(organization) session.flush() session.add( business_audit( action=AuditAction.CREATE_ORGANIZATION, target=AuditTarget.ORGANIZATION, target_id=organization.id, target_name=organization.org_name, detail={ "orgCode": organization.org_code, "parentId": serialize_id(organization.parent_id), "orgPath": organization.org_path, "sortNo": organization.sort_no, "status": organization.status, }, ) ) return _dto(organization) except IntegrityError as exc: raise ConflictError("机构编码已存在") from exc def update_organization(organization_id: int, payload: Any) -> dict[str, object]: body = _exact_payload(payload, UPDATE_FIELDS) org_code = _required_text(body["orgCode"], field="orgCode", max_length=64) org_name = _required_text(body["orgName"], field="orgName", max_length=128) parent_id = _string_id(body["parentId"], field="parentId", nullable=True) sort_no = _sort_no(body["sortNo"]) status = _status_value(body["status"]) expected_version = _row_version(body["rowVersion"]) actor_id = get_auth_context().user_id now = _utc_now() with transaction() as session: actor = require_organization_editor(session) ensure_org_in_managed_scope(session, actor, organization_id) organizations = _locked_organizations(session) by_id = {item.id: item for item in organizations} organization = by_id.get(organization_id) if organization is None: raise ResourceNotFoundError("机构不存在") if organization.row_version != expected_version: raise ConflictError( "数据已被其他用户修改,请刷新后重试", details={"currentRowVersion": organization.row_version}, ) # 机构编码可编辑:保存时查重(排除自身,口径与创建一致——仅比对未删除机构) if ( session.scalar( select(Organization.id) .where( Organization.org_code == org_code, Organization.id != organization.id, Organization.is_deleted.is_(False), ) ) is not None ): raise ConflictError("机构编码已存在") children = _children_map(organizations) descendants = _descendants(organization.id, children) descendant_ids = {item.id for item in descendants} if parent_id == organization.id: raise InvalidArgumentError("parentId不能等于当前机构ID") if parent_id in descendant_ids: raise InvalidArgumentError("机构不能移动到自己的后代节点下") parent: Organization | None = 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必须指向有效且已启用的机构") before = { "orgCode": organization.org_code, "orgName": organization.org_name, "parentId": serialize_id(organization.parent_id), "orgPath": organization.org_path, "sortNo": organization.sort_no, "status": organization.status, "rowVersion": organization.row_version, } path_changed = ( org_name != organization.org_name or parent_id != organization.parent_id ) organization.org_code = org_code organization.org_name = org_name organization.parent_id = parent_id organization.org_path = _org_path(parent, org_name) organization.sort_no = sort_no organization.status = status organization.updated_by = actor_id organization.updated_at = now organization.row_version += 1 affected_organizations = [organization] if path_changed: # 阶段1:预先计算所有受影响节点的新路径 # 在修改任何对象前发现长度溢出,配合 transaction() 自动回滚保证一致性 new_child_paths: dict[int, str] = {} validation_queue: deque[tuple[Organization, str]] = deque( [(organization, organization.org_path)] ) while validation_queue: current, current_new_path = validation_queue.popleft() for child in children.get(current.id, []): new_child_path = _build_path(current_new_path, child.org_name) new_child_paths[child.id] = new_child_path validation_queue.append((child, new_child_path)) # 阶段2:所有路径已校验通过,按 BFS 顺序应用变更 pending = deque([organization]) while pending: current = pending.popleft() for child in children.get(current.id, []): child.org_path = new_child_paths[child.id] child.updated_by = actor_id child.updated_at = now child.row_version += 1 affected_organizations.append(child) pending.append(child) if path_changed: for affected in affected_organizations: session.execute( update(User) .where( User.organization_id == affected.id, User.is_deleted.is_(False), ) .values( organization_name=affected.org_name, updated_by=actor_id, updated_at=now, row_version=User.row_version + 1, ) ) for affected in affected_organizations: session.execute( update(Document) .where( Document.organization_id == affected.id, Document.is_deleted.is_(False), ) .values( organization_name=affected.org_name, updated_by=actor_id, updated_at=now, row_version=Document.row_version + 1, ) ) after = { "orgName": organization.org_name, "parentId": serialize_id(organization.parent_id), "orgPath": organization.org_path, "sortNo": organization.sort_no, "status": organization.status, "rowVersion": organization.row_version, } session.add( business_audit( action=AuditAction.EDIT_ORGANIZATION, target=AuditTarget.ORGANIZATION, target_id=organization.id, target_name=organization.org_name, detail={ "before": before, "after": after, "updatedDescendantCount": len(affected_organizations) - 1, }, ) ) return _dto(organization) def delete_organization(organization_id: int, row_version: int) -> dict[str, object]: actor_id = get_auth_context().user_id now = _utc_now() with transaction() as session: actor = require_organization_editor(session) ensure_org_in_managed_scope(session, actor, organization_id) organizations = _locked_organizations(session) by_id = {item.id: item for item in organizations} organization = by_id.get(organization_id) if organization is None: raise ResourceNotFoundError("机构不存在") if organization.row_version != row_version: raise ConflictError( "数据已被其他用户修改,请刷新后重试", details={"currentRowVersion": organization.row_version}, ) children = _children_map(organizations) descendants = _descendants(organization.id, children) child_count = len(descendants) user_count = ( session.scalar( select(func.count(User.id)).where( User.organization_id == organization.id, User.is_deleted.is_(False), ) ) or 0 ) if child_count or user_count: raise OrganizationInUseError( details={ "organizationId": serialize_id(organization.id), "childOrganizationCount": child_count, "userCount": user_count, } ) target_name = organization.org_name organization.is_deleted = True organization.deleted_at = now organization.updated_at = now organization.updated_by = actor_id organization.row_version += 1 # 部门删除后其IP分配行同事务软删,避免残留限制语义 for row in session.scalars( select(OrgIpAssignment).where( OrgIpAssignment.organization_id == organization.id, OrgIpAssignment.is_deleted.is_(False), ) ).all(): row.is_deleted = True row.deleted_at = now row.updated_by = actor_id row.updated_at = now session.add( business_audit( action=AuditAction.DELETE_ORGANIZATION, target=AuditTarget.ORGANIZATION, target_id=organization.id, target_name=target_name, detail={ "orgCode": organization.org_code, "orgPath": organization.org_path, "rowVersion": organization.row_version, }, ) ) return {"id": serialize_id(organization.id), "deleted": True} def parse_organization_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)