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

"""组织树查询、写操作、路径冗余与占用规则。"""

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.audit_service import business_audit


CREATE_FIELDS = {
    "orgCode",
    "orgName",
    "parentId",
    "sortNo",
    "status",
}
UPDATE_FIELDS = {
    "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:
            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_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:
        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},
            )

        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 = {
            "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_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:
        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)