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

"""B6借阅授权审批单（PDF凭证）的上传、下载与清除。

凭证与授权条目一对一（0026迁移，契约1.19），读写均要求授权配置位
（CONFIG_PERMISSION）并通过密级与密码锁校验；上传为替换语义，旧凭证
文件在数据库事务成功后清理。仅主案可操作（与save_permissions同口径）。
"""

from __future__ import annotations

import hashlib
import logging
import uuid
from pathlib import Path

from flask import current_app, send_file
from sqlalchemy import select
from werkzeug.datastructures import FileStorage

from dms.common.enums import (
    AllowedAction,
    AuditAction,
    AuditTarget,
    DocumentType,
)
from dms.common.errors import (
    AttachmentHasNoAclError,
    PayloadTooLargeError,
    ResourceNotFoundError,
    SubPlanPermissionInheritedError,
    UnsupportedFileTypeError,
)
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, Permission
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.services.document_password_service import (
    DocumentPasswordAction,
    get_request_password,
    require_document_password,
)
from dms.services.authorization_service import (
    require_plan_action,
    require_security_clearance,
)
from dms.services.permission_service import _document, _iso, _now
from dms.storage.paths import (
    ensure_storage_directories,
    resolve_storage_path,
)
from dms.storage.uploads import StagedUpload, _validate

logger = logging.getLogger(__name__)


def _stage_voucher(file: FileStorage) -> StagedUpload:
    """暂存审批单并校验：仅允许PDF，大小受DMS_MAX_FILE_SIZE约束。"""
    original_name = Path(file.filename or "").name
    if not original_name or "." not in original_name:
        raise UnsupportedFileTypeError("审批单文件名缺少扩展名")
    extension = original_name.rsplit(".", 1)[1].lower()
    if extension != "pdf":
        raise UnsupportedFileTypeError("审批单凭证只允许PDF文件")
    root = ensure_storage_directories(current_app.config["DMS_STORAGE_ROOT"])
    temporary = root / "temporary" / f"{uuid.uuid4()}.upload"
    final_name = f"{uuid.uuid4()}.pdf"
    final = root / "voucher" / final_name
    digest = hashlib.sha256()
    size = 0
    try:
        with temporary.open("xb") as stream:
            while chunk := file.stream.read(1024 * 1024):
                size += len(chunk)
                if size > current_app.config["DMS_MAX_FILE_SIZE"]:
                    raise PayloadTooLargeError("审批单超过允许大小")
                digest.update(chunk)
                stream.write(chunk)
        if size == 0:
            raise UnsupportedFileTypeError("不允许上传空文件")
        _validate(temporary, "pdf")
        return StagedUpload(
            temporary_path=temporary,
            final_path=final,
            relative_path=f"voucher/{final_name}",
            original_file_name=original_name,
            extension=extension,
            mime_type="application/pdf",
            file_size=size,
            file_hash=digest.hexdigest(),
        )
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


def _checked_main(
    document_id: int,
    *,
    lock: bool = False,
) -> Document:
    """定位主案并执行与save_permissions同口径的授权校验。"""
    document = _document(document_id, lock=lock)
    if document.document_type == DocumentType.ATTACHMENT.value:
        raise AttachmentHasNoAclError()
    if document.document_type == DocumentType.SUB_PLAN.value:
        raise SubPlanPermissionInheritedError()
    context = get_auth_context()
    require_security_clearance(document, context)
    require_plan_action(document, context, AllowedAction.CONFIG_PERMISSION)
    require_document_password(
        document,
        DocumentPasswordAction.CONFIG_PERMISSION.value,
        get_request_password(),
        context,
    )
    return document


def _locked_entry(source_id: int, entry_id: int) -> Permission:
    statement = (
        select(Permission)
        .where(
            Permission.id == entry_id,
            Permission.document_id == source_id,
            Permission.is_deleted.is_(False),
        )
        .with_for_update()
    )
    entry = db.session.scalar(statement)
    if entry is None:
        raise ResourceNotFoundError("授权条目不存在")
    return entry


def _remove_storage_file(relative_path: str) -> None:
    """清理存储文件；失败仅记录日志，不影响业务结果。"""
    try:
        path = resolve_storage_path(
            relative_path, current_app.config["DMS_STORAGE_ROOT"]
        )
        path.unlink(missing_ok=True)
    except OSError:
        logger.exception("清理审批单文件失败：%s", relative_path)


def upload_voucher(
    document_id: int,
    entry_id: int,
    file: FileStorage,
) -> dict[str, object]:
    staged = _stage_voucher(file)
    context = get_auth_context()
    replaced: list[str] = []

    def operation() -> dict[str, object]:
        with transaction() as session:
            document = _checked_main(document_id, lock=True)
            entry = _locked_entry(document.id, entry_id)
            if entry.voucher_storage_path:
                replaced.append(entry.voucher_storage_path)
            now = _now()
            entry.voucher_file_name = staged.original_file_name
            entry.voucher_storage_path = staged.relative_path
            entry.voucher_file_size = staged.file_size
            entry.voucher_uploaded_at = now
            entry.updated_at = now
            entry.updated_by = context.user_id
            entry.row_version += 1
            session.add(
                business_audit(
                    action=AuditAction.UPLOAD_PERMISSION_VOUCHER,
                    target=AuditTarget.PERMISSION,
                    target_id=document.id,
                    target_name=document.document_name,
                    detail={
                        "mainPlanId": serialize_id(document.id),
                        "entryId": serialize_id(entry.id),
                        "subjectId": serialize_id(entry.subject_id),
                        "subjectName": entry.subject_name,
                        "voucherFileName": staged.original_file_name,
                        "voucherFileSize": staged.file_size,
                    },
                )
            )
            return {
                "documentId": serialize_id(document.id),
                "entryId": serialize_id(entry.id),
                "voucherFileName": staged.original_file_name,
                "voucherFileSize": staged.file_size,
                "voucherUploadedAt": _iso(now),
            }

    try:
        # 文件提升放数据库事务成功之后：死锁重试只重放纯DB操作，
        # 避免promote重入；提交后未及提升的极端窗口可通过重新上传修复。
        result = execute_with_deadlock_retry(operation)
        staged.promote()
    except Exception:
        staged.cleanup_all()
        raise
    for relative_path in set(replaced):
        if relative_path != staged.relative_path:
            _remove_storage_file(relative_path)
    return result


def download_voucher(document_id: int, entry_id: int):
    document = _checked_main(document_id)
    statement = select(Permission).where(
        Permission.id == entry_id,
        Permission.document_id == document.id,
        Permission.is_deleted.is_(False),
    )
    entry = db.session.scalar(statement)
    if entry is None:
        raise ResourceNotFoundError("授权条目不存在")
    if not entry.voucher_storage_path:
        raise ResourceNotFoundError("该授权未上传审批单")
    path = resolve_storage_path(
        entry.voucher_storage_path, current_app.config["DMS_STORAGE_ROOT"]
    )
    if not path.is_file():
        raise ResourceNotFoundError("审批单文件不存在或已被清理")
    response = send_file(
        path,
        mimetype="application/pdf",
        as_attachment=True,
        download_name=entry.voucher_file_name or "voucher.pdf",
        conditional=True,
    )
    response.headers["X-Content-Type-Options"] = "nosniff"
    return response


def delete_voucher(document_id: int, entry_id: int) -> dict[str, object]:
    context = get_auth_context()
    removed: list[str] = []

    def operation() -> dict[str, object]:
        with transaction() as session:
            document = _checked_main(document_id, lock=True)
            entry = _locked_entry(document.id, entry_id)
            if not entry.voucher_storage_path:
                raise ResourceNotFoundError("该授权未上传审批单")
            removed.append(entry.voucher_storage_path)
            now = _now()
            cleared_file_name = entry.voucher_file_name
            entry.voucher_file_name = None
            entry.voucher_storage_path = None
            entry.voucher_file_size = None
            entry.voucher_uploaded_at = None
            entry.updated_at = now
            entry.updated_by = context.user_id
            entry.row_version += 1
            session.add(
                business_audit(
                    action=AuditAction.DELETE_PERMISSION_VOUCHER,
                    target=AuditTarget.PERMISSION,
                    target_id=document.id,
                    target_name=document.document_name,
                    detail={
                        "mainPlanId": serialize_id(document.id),
                        "entryId": serialize_id(entry.id),
                        "subjectId": serialize_id(entry.subject_id),
                        "subjectName": entry.subject_name,
                        "voucherFileName": cleared_file_name,
                    },
                )
            )
            return {
                "documentId": serialize_id(document.id),
                "entryId": serialize_id(entry.id),
            }

    result = execute_with_deadlock_retry(operation)
    for relative_path in set(removed):
        _remove_storage_file(relative_path)
    return result


__all__ = ["delete_voucher", "download_voucher", "upload_voucher"]