Blame for apps/contracts/api/folder_binding_api.py

"""合同文件夹绑定 API。"""

from __future__ import annotations

import logging
from typing import Any

from django.http import HttpRequest
from django.utils.translation import gettext_lazy as _
from ninja import Router

from apps.contracts.schemas import (
    FolderBindingCreateSchema,
    FolderBindingResponseSchema,
    FolderBrowseEntrySchema,
    FolderBrowseResponseSchema,
)
from apps.core.exceptions import PermissionDenied
from apps.core.security import get_request_access_context

logger = logging.getLogger("apps.contracts.api")
router = Router()


def _get_folder_binding_service() -> Any:
    """
    工厂函数:创建 FolderBindingService 实例

    Returns:
        FolderBindingService 实例
    """
    from apps.contracts.services import FolderBindingService
    from apps.core.dependencies.documents import build_document_template_binding_service

    return FolderBindingService(document_template_binding_service=build_document_template_binding_service())


def _require_contract_access(request: HttpRequest, contract_id: int) -> None:
    from apps.contracts.services.contract import wiring

    ctx = get_request_access_context(request)
    wiring.get_contract_query_facade().get_contract_ctx(contract_id=contract_id, ctx=ctx)


def _require_admin(request: HttpRequest) -> None:
    user = getattr(request, "user", None)
    if not user or not getattr(user, "is_authenticated", False):
        raise PermissionDenied(_("需要登录"))


@router.post("/{contract_id}/folder-binding", response=FolderBindingResponseSchema)
def create_folder_binding(request: HttpRequest, contract_id: int, data: FolderBindingCreateSchema) -> Any:
    """
    创建或更新文件夹绑定

    Args:
        contract_id: 合同ID
        data: 绑定数据,包含文件夹路径

    Returns:
        FolderBindingResponseSchema: 绑定信息

    Raises:
        ValidationException: 路径无效或合同不存在

    Requirements: 9.1, 9.4
    """
    _require_admin(request)
    _require_contract_access(request, contract_id)
    service = _get_folder_binding_service()
    ctx = get_request_access_context(request)

    binding = service.create_binding(owner_id=contract_id, folder_path=data.folder_path)

    display_path = service.format_path_for_display(binding.folder_path)
    is_accessible = service.check_folder_accessible(binding.folder_path)

    logger.info(
        "contract_folder_binding_upsert",
        extra={
            "action": "contract_folder_binding_upsert",
            "contract_id": contract_id,
            "folder_path": str(getattr(binding, "folder_path", "") or ""),
            "display_path": str(display_path or ""),
            "is_accessible": bool(is_accessible),
            "user_id": getattr(getattr(ctx, "user", None), "id", None),
        },
    )

    return FolderBindingResponseSchema.from_binding(
        binding,
        is_accessible=is_accessible,
        display_path=display_path,
    )


@router.get("/{contract_id}/folder-binding", response=FolderBindingResponseSchema | None)
def get_folder_binding(request: HttpRequest, contract_id: int) -> Any:
    """
    获取文件夹绑定信息

    Args:
        contract_id: 合同ID

    Returns:
        Optional[FolderBindingResponseSchema]: 绑定信息,如果未绑定则返回 None

    Requirements: 9.2, 9.4
    """
    _require_contract_access(request, contract_id)
    service = _get_folder_binding_service()

    binding = service.get_binding(owner_id=contract_id)
    if not binding:
        return None

    # 格式化显示路径
    display_path = service.format_path_for_display(binding.folder_path)

    # 检查文件夹是否可访问(只检查,不创建)
    is_accessible = service.check_folder_accessible(binding.folder_path)

    return FolderBindingResponseSchema.from_binding(binding, is_accessible=is_accessible, display_path=display_path)


@router.delete("/{contract_id}/folder-binding")
def delete_folder_binding(request: HttpRequest, contract_id: int) -> Any:
    """
    删除文件夹绑定

    Args:
        contract_id: 合同ID

    Returns:
        dict: 删除结果

    Requirements: 9.3, 9.5
    """
    _require_admin(request)
    _require_contract_access(request, contract_id)
    service = _get_folder_binding_service()

    ctx = get_request_access_context(request)
    success = service.delete_binding(owner_id=contract_id)

    logger.info(
        "contract_folder_binding_delete",
        extra={
            "action": "contract_folder_binding_delete",
            "contract_id": contract_id,
            "success": bool(success),
            "user_id": getattr(getattr(ctx, "user", None), "id", None),
        },
    )

    return {"success": success, "message": _("绑定已删除") if success else _("绑定不存在")}


@router.get("/folder-browse", response=FolderBrowseResponseSchema)
def browse_folders(request: HttpRequest, path: str | None = None, include_hidden: bool = False) -> Any:
    _require_admin(request)
    service = _get_folder_binding_service()
    ctx = get_request_access_context(request)

    if not path or not str(path).strip():
        # 默认进入用户下载文件夹
        default_path = service.get_default_browse_path()
        if default_path:
            path = str(default_path)
        else:
            # 降级：显示根目录列表
            roots = service.get_browse_roots()
            entries = [FolderBrowseEntrySchema(name=(p.name or str(p)), path=str(p)) for p in roots]
            logger.info(
                "contract_folder_browse_roots",
                extra={
                    "action": "contract_folder_browse_roots",
                    "include_hidden": bool(include_hidden),
                    "roots_count": len(roots),
                    "user_id": getattr(getattr(ctx, "user", None), "id", None),
                },
            )
            return FolderBrowseResponseSchema(
                browsable=True,
                message=None,
                path=None,
                parent_path=None,
                entries=entries,
            )

    browsable, browse_message = service.is_browsable_path(str(path))
    if not browsable:
        return FolderBrowseResponseSchema(
            browsable=False,
            message=browse_message,
            path=str(path).strip(),
            parent_path=None,
            entries=[],
        )

    resolved = service.resolve_under_allowed_roots(str(path))
    entries = [
        FolderBrowseEntrySchema(**item) for item in service.list_subdirs(str(path), include_hidden=include_hidden)
    ]
    parent_path: str | None = service.compute_parent_path(resolved)

    logger.info(
        "contract_folder_browse",
        extra={
            "action": "contract_folder_browse",
            "path": str(path).strip(),
            "resolved_path": str(resolved),
            "include_hidden": bool(include_hidden),
            "entries_count": len(entries),
            "user_id": getattr(getattr(ctx, "user", None), "id", None),
        },
    )

    return FolderBrowseResponseSchema(
        browsable=True,
        message=None,
        path=str(resolved),
        parent_path=parent_path,
        entries=entries,
    )