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

"""方案关联关系：预置关系类型加载校验与关联关系读写。"""

from __future__ import annotations

import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import AuditAction, AuditTarget, DocumentType
from dms.common.errors import (
    ConflictError,
    DocumentViewForbiddenError,
    InvalidArgumentError,
    ResourceNotFoundError,
    SecurityLevelForbiddenError,
)
from dms.common.response import serialize_id
from dms.database.transaction import execute_with_deadlock_retry, transaction
from dms.extensions import db
from dms.models import LOCK_DEFER_LOB, Document, DocumentRelation
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.common.enums import AllowedAction
from dms.services.authorization_service import (
    evaluate_plan_access,
    require_plan_action,
    require_security_clearance,
)
from dms.services.document_password_service import (
    DocumentPasswordAction,
    get_request_password,
    require_document_password,
)
from dms.services.document_query_service import PLAN_TYPES

RELATION_FIELDS = {"relations", "rowVersion"}
RELATION_ITEM_FIELDS = {"relatedDocumentId", "relationType"}
MAX_RELATIONS = 50
RELATION_TYPE_MAX_LENGTH = 64


# ─── 预置关系类型：启动期加载与严格校验 ─────────────────────────────────────


@dataclass(frozen=True, slots=True)
class RelationPreset:
    code: str
    label: str
    sort_no: int

    def to_dict(self) -> dict[str, str | int]:
        return {
            "code": self.code,
            "label": self.label,
            "sortNo": self.sort_no,
        }


class DocumentRelationPresetError(RuntimeError):
    """不包含文件路径或原始内容的预置关系配置错误。"""


def _preset_invalid(reason: str) -> DocumentRelationPresetError:
    return DocumentRelationPresetError(f"方案关联关系预置配置无效：{reason}")


def load_relation_presets(config_path: str | Path) -> tuple[RelationPreset, ...]:
    """读取并严格校验预置关系类型清单。"""
    path = Path(config_path)
    try:
        with path.open("r", encoding="utf-8") as source:
            raw = json.load(source)
    except FileNotFoundError:
        raise _preset_invalid("配置文件不存在") from None
    except UnicodeDecodeError:
        raise _preset_invalid("配置不是有效的UTF-8文件") from None
    except json.JSONDecodeError:
        raise _preset_invalid("配置不是有效的JSON") from None
    except OSError:
        raise _preset_invalid("配置文件无法读取") from None

    if not isinstance(raw, dict) or set(raw) != {"presets"}:
        raise _preset_invalid("根节点必须且只能包含presets")
    raw_presets = raw["presets"]
    if not isinstance(raw_presets, list) or not raw_presets:
        raise _preset_invalid("presets必须是非空数组")

    seen_codes: set[str] = set()
    seen_labels: set[str] = set()
    presets: list[RelationPreset] = []
    for index, raw_preset in enumerate(raw_presets):
        if not isinstance(raw_preset, dict) or set(raw_preset) != {
            "code",
            "label",
            "sortNo",
        }:
            raise _preset_invalid(f"presets[{index}]字段不合法")
        code = raw_preset["code"]
        label = raw_preset["label"]
        sort_no = raw_preset["sortNo"]
        if not isinstance(code, str) or not code:
            raise _preset_invalid(f"presets[{index}]包含非法code")
        if code in seen_codes:
            raise _preset_invalid(f"presets[{index}]包含重复code")
        if not isinstance(label, str) or not label.strip():
            raise _preset_invalid(f"presets[{index}]包含空label")
        if label.strip() in seen_labels:
            raise _preset_invalid(f"presets[{index}]包含重复label")
        if type(sort_no) is not int or sort_no < 0:
            raise _preset_invalid(f"presets[{index}]包含非法sortNo")
        seen_codes.add(code)
        seen_labels.add(label.strip())
        presets.append(
            RelationPreset(code=code, label=label.strip(), sort_no=sort_no)
        )
    return tuple(sorted(presets, key=lambda item: (item.sort_no, item.code)))


# ─── 关联关系读取 ──────────────────────────────────────────────────────────


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


def _version(value: Any, name: str) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError(f"{name}必须是非负整数")
    return value


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


def _relation_items(session, document_id: int) -> list[dict[str, str]]:
    rows = session.execute(
        select(DocumentRelation, Document)
        .join(Document, Document.id == DocumentRelation.related_document_id)
        .where(
            DocumentRelation.document_id == document_id,
            DocumentRelation.is_deleted.is_(False),
            Document.is_deleted.is_(False),
        )
        .order_by(
            DocumentRelation.sort_no.asc(), DocumentRelation.id.asc()
        )
    ).all()
    return [
        {
            "relatedDocumentId": serialize_id(relation.related_document_id),
            "relatedDocumentName": document.document_name,
            "relationType": relation.relation_type,
        }
        for relation, document in rows
    ]


def _require_plan_access(document: Document, context) -> None:
    access = evaluate_plan_access(document, context)
    if access.allowed:
        return
    if access.reason == "SECURITY":
        raise SecurityLevelForbiddenError()
    raise DocumentViewForbiddenError()


def list_document_relations(document_id: int) -> dict[str, object]:
    """查看方案声明的关联清单（与文档详情同套访问控制）。"""
    context = get_auth_context()
    document = db.session.scalar(
        select(Document).where(
            Document.id == document_id,
            Document.is_deleted.is_(False),
            Document.document_type.in_(PLAN_TYPES),
        )
    )
    if document is None:
        raise ResourceNotFoundError("方案文档不存在")
    _require_plan_access(document, context)
    return {"items": _relation_items(db.session, document.id)}


# ─── 关联关系整体替换保存 ──────────────────────────────────────────────────


def _relations_payload(payload: Any) -> tuple[list[tuple[int, str]], int]:
    if not isinstance(payload, dict) or set(payload) != RELATION_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含relations和rowVersion"
        )
    raw_relations = payload["relations"]
    if not isinstance(raw_relations, list):
        raise InvalidArgumentError("relations必须是数组")
    if len(raw_relations) > MAX_RELATIONS:
        raise InvalidArgumentError(f"relations最多允许{MAX_RELATIONS}条")
    pairs: list[tuple[int, str]] = []
    seen: set[int] = set()
    for index, raw_item in enumerate(raw_relations):
        if not isinstance(raw_item, dict) or set(raw_item) != RELATION_ITEM_FIELDS:
            raise InvalidArgumentError(
                f"relations[{index}]必须且只能包含relatedDocumentId和relationType"
            )
        related_id = _string_id(
            raw_item["relatedDocumentId"],
            f"relations[{index}].relatedDocumentId",
        )
        relation_type = raw_item["relationType"]
        if not isinstance(relation_type, str):
            raise InvalidArgumentError(
                f"relations[{index}].relationType必须是字符串"
            )
        relation_type = relation_type.strip()
        if not relation_type or len(relation_type) > RELATION_TYPE_MAX_LENGTH:
            raise InvalidArgumentError(
                f"relations[{index}].relationType长度须为1~{RELATION_TYPE_MAX_LENGTH}"
            )
        if related_id in seen:
            raise InvalidArgumentError(
                f"relations[{index}].relatedDocumentId与其他条目重复"
            )
        seen.add(related_id)
        pairs.append((related_id, relation_type))
    return pairs, _version(payload["rowVersion"], "rowVersion")


def _locked_related_plans(
    session,
    related_ids: list[int],
    self_document_id: int,
) -> None:
    """被关联方必须存在、未删且为主案；按ID升序加锁避免死锁。"""
    if not related_ids:
        return
    documents = session.scalars(
        select(Document)
        .where(Document.id.in_(sorted(related_ids)))
        .order_by(Document.id.asc())
        .options(*LOCK_DEFER_LOB)
        .with_for_update()
    ).all()
    by_id = {document.id: document for document in documents}
    for related_id in related_ids:
        if related_id == self_document_id:
            raise InvalidArgumentError("不允许关联方案自身")
        target = by_id.get(related_id)
        if (
            target is None
            or target.is_deleted
            or target.document_type != DocumentType.MAIN.value
        ):
            raise ResourceNotFoundError(
                f"被关联方案不存在或类型不匹配：{related_id}"
            )


def _current_row_version(document_id: int) -> int | None:
    return db.session.scalar(
        select(Document.row_version).where(
            Document.id == document_id,
            Document.is_deleted.is_(False),
        )
    )


def save_document_relations(
    document_id: int,
    payload: Any,
) -> dict[str, object]:
    """整体替换方案的关联清单；同一方案上的写入由方案行锁串行化。"""
    pairs, expected_version = _relations_payload(payload)
    context = get_auth_context()

    def operation() -> dict[str, object]:
        with transaction() as session:
            document = session.scalar(
                select(Document)
                .where(
                    Document.id == document_id,
                    Document.is_deleted.is_(False),
                    Document.document_type.in_(PLAN_TYPES),
                )
                .options(*LOCK_DEFER_LOB)
                .with_for_update()
            )
            if document is None:
                raise ResourceNotFoundError("方案文档不存在")
            require_plan_action(document, context, AllowedAction.EDIT)
            require_security_clearance(document, context)
            require_document_password(
                document,
                DocumentPasswordAction.EDIT.value,
                get_request_password(),
                context,
            )
            if document.row_version != expected_version:
                raise ConflictError(
                    "数据已被其他用户修改，请刷新后重试",
                    details={"currentRowVersion": document.row_version},
                )
            _locked_related_plans(
                session, [related_id for related_id, _ in pairs], document.id
            )

            existing = session.scalars(
                select(DocumentRelation)
                .where(
                    DocumentRelation.document_id == document.id,
                    DocumentRelation.is_deleted.is_(False),
                )
                .order_by(
                    DocumentRelation.sort_no.asc(), DocumentRelation.id.asc()
                )
                .with_for_update()
            ).all()
            existing_pairs = [
                (relation.related_document_id, relation.relation_type)
                for relation in existing
            ]
            if existing_pairs == pairs:
                # 内容未变化：不产生写入、版本号与审计噪音
                return {
                    "relations": _relation_items(session, document.id),
                    "rowVersion": document.row_version,
                }

            now = _now()
            for relation in existing:
                relation.is_deleted = True
                relation.deleted_at = now
                relation.updated_at = now
                relation.updated_by = context.user_id
                relation.row_version += 1
            for offset, (related_id, relation_type) in enumerate(
                pairs, start=1
            ):
                session.add(
                    DocumentRelation(
                        document_id=document.id,
                        related_document_id=related_id,
                        relation_type=relation_type,
                        sort_no=offset * 10,
                        created_by=context.user_id,
                        updated_by=context.user_id,
                    )
                )
            session.flush()
            document.row_version += 1
            document.updated_by = context.user_id
            document.updated_by_name = context.real_name
            document.updated_at = now
            session.add(
                business_audit(
                    action=AuditAction.EDIT_DOCUMENT,
                    target=AuditTarget.DOCUMENT,
                    target_id=document.id,
                    target_name=document.document_name,
                    detail={
                        "scope": "relations",
                        "documentType": document.document_type,
                        "removedCount": len(existing_pairs),
                        "relations": [
                            {
                                "relatedDocumentId": serialize_id(related_id),
                                "relationType": relation_type,
                            }
                            for related_id, relation_type in pairs
                        ],
                        "rowVersion": document.row_version,
                    },
                )
            )
            return {
                "relations": _relation_items(session, document.id),
                "rowVersion": document.row_version,
            }

    try:
        return execute_with_deadlock_retry(operation)
    except IntegrityError as exc:
        db.session.rollback()
        raise ConflictError(
            "关联关系已被并发修改，请刷新后重试",
            details={
                "currentRowVersion": _current_row_version(document_id)
            },
        ) from exc


__all__ = [
    "DocumentRelationPresetError",
    "RelationPreset",
    "list_document_relations",
    "load_relation_presets",
    "save_document_relations",
]