Foxel/domain/virtual_fs/api.py

from typing import Annotated
import hashlib
import json
import mimetypes
from urllib.parse import quote

from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
from fastapi.responses import Response

from api.response import success
from domain.audit import AuditAction, audit
from domain.auth import User, get_current_active_user
from domain.permission import require_path_permission
from domain.permission.types import PathAction
from .service import VirtualFSService
from .types import MkdirRequest, MoveRequest
from .chunk_uploads import ChunkUploadCreate, ChunkUploadService

router = APIRouter(prefix="/api/fs", tags=["virtual-fs"])


@router.post("/uploads")
async def create_chunk_upload(
    body: ChunkUploadCreate,
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return success(await ChunkUploadService.create(body, current_user.id))


@router.put("/uploads/{upload_id}/parts/{part}")
async def upload_chunk(
    upload_id: str,
    part: int,
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return success(await ChunkUploadService.put_part(upload_id, part, current_user.id, request.stream()))


@router.get("/uploads/{upload_id}")
async def chunk_upload_status(
    upload_id: str,
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return success(await ChunkUploadService.status(upload_id, current_user.id))


@router.post("/uploads/{upload_id}/complete")
@audit(action=AuditAction.UPLOAD, description="Complete chunked upload")
async def complete_chunk_upload(
    upload_id: str,
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    overwrite: bool | None = Query(None),
):
    return success(await ChunkUploadService.complete(upload_id, current_user.id, overwrite))


@router.delete("/uploads/{upload_id}")
async def abort_chunk_upload(
    upload_id: str,
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return success(await ChunkUploadService.abort(upload_id, current_user.id))


@router.get("/download/{full_path:path}")
@audit(action=AuditAction.DOWNLOAD, description="Download original file")
@require_path_permission(PathAction.READ, "full_path")
async def download_original(
    full_path: str,
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return await _download_original_response(full_path, request)


@router.get("/download-public/{token}/{filename}")
@audit(action=AuditAction.DOWNLOAD, description="Download original file through temporary link")
async def download_original_public(token: str, filename: str, request: Request):
    path = await VirtualFSService.verify_temp_link_token(token)
    return await _download_original_response(path, request)


async def _download_original_response(full_path: str, request: Request):
    full_path = VirtualFSService._normalize_path(full_path)
    try:
        stat = await VirtualFSService.stat_file(full_path)
    except FileNotFoundError:
        raise HTTPException(404, detail="File not found")
    if stat.get("is_dir"):
        raise HTTPException(400, detail="Path is a directory")
    # Pin the metadata version across browser range requests. Providers such as
    # S3 additionally pin their native object version while fetching each range.
    etag = '"' + hashlib.sha256(json.dumps(stat, sort_keys=True, default=str).encode()).hexdigest() + '"'
    if request.headers.get("If-Match") not in (None, etag):
        raise HTTPException(412, detail="File changed during download")
    adapter, _, root, rel = await VirtualFSService.resolve_adapter_and_rel(full_path)
    stream = getattr(adapter, "stream_file", None)
    if callable(stream):
        response = await stream(root, rel, request.headers.get("Range"))
    else:
        data = await VirtualFSService.read_file(full_path)
        response = Response(data, media_type=mimetypes.guess_type(full_path)[0] or "application/octet-stream")
    response.headers["ETag"] = etag
    response.headers["Content-Disposition"] = "attachment; filename*=UTF-8''" + quote(full_path.rstrip("/").split("/")[-1], safe="")
    return response


@router.get("/file/{full_path:path}")
@audit(action=AuditAction.DOWNLOAD, description="获取文件")
@require_path_permission(PathAction.READ, "full_path")
async def get_file(
    full_path: str,
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
):
    return await VirtualFSService.serve_file(full_path, request.headers.get("Range"))


@router.get("/thumb/{full_path:path}")
@audit(action=AuditAction.READ, description="获取缩略图")
async def get_thumb(
    full_path: str,
    request: Request,
    w: int = Query(256, ge=8, le=1024),
    h: int = Query(256, ge=8, le=1024),
    fit: str = Query("cover"),
):
    return await VirtualFSService.get_thumbnail(full_path, w, h, fit)


@router.get("/stream/{full_path:path}")
@audit(action=AuditAction.DOWNLOAD, description="流式读取文件")
async def stream_endpoint(
    full_path: str,
    request: Request,
):
    return await VirtualFSService.stream_response(full_path, request.headers.get("Range"))


@router.get("/temp-link/{full_path:path}")
@audit(action=AuditAction.SHARE, description="创建临时链接")
@require_path_permission(PathAction.READ, "full_path")
async def get_temp_link(
    full_path: str,
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    expires_in: int = Query(3600, description="有效时间(秒), 0或负数表示永久"),
):
    data = await VirtualFSService.create_temp_link(full_path, expires_in)
    return success(data)


@router.get("/public/{token}")
@audit(action=AuditAction.DOWNLOAD, description="访问临时链接文件")
async def access_public_file(
    token: str,
    request: Request,
):
    return await VirtualFSService.access_public_file(token, request.headers.get("Range"))


@router.get("/public/{token}/{filename}")
@audit(action=AuditAction.DOWNLOAD, description="访问临时链接文件")
async def access_public_file_with_name(
    token: str,
    filename: str,
    request: Request,
):
    return await VirtualFSService.access_public_file(token, request.headers.get("Range"))


@router.get("/stat/{full_path:path}")
@audit(action=AuditAction.READ, description="查看文件信息")
@require_path_permission(PathAction.READ, "full_path")
async def get_file_stat(
    full_path: str,
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    verbose: bool = Query(False, description="是否返回扩展元数据"),
):
    stat = await VirtualFSService.stat(full_path, verbose=verbose)
    return success(stat)


@router.post("/file/{full_path:path}")
@audit(action=AuditAction.UPLOAD, description="上传文件")
@require_path_permission(PathAction.WRITE, "full_path")
async def put_file(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    full_path: str,
    file: UploadFile = File(...),
):
    data = await file.read()
    result = await VirtualFSService.write_uploaded_file(full_path, data)
    return success(result)


@router.post("/mkdir")
@audit(action=AuditAction.CREATE, description="创建目录", body_fields=["path"])
@require_path_permission(PathAction.WRITE, "body.path")
async def api_mkdir(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    body: MkdirRequest,
):
    result = await VirtualFSService.mkdir(body.path)
    return success(result)


@router.post("/move")
@audit(action=AuditAction.UPDATE, description="移动路径", body_fields=["src", "dst"])
@require_path_permission(PathAction.WRITE, "body.dst")
@require_path_permission(PathAction.DELETE, "body.src")
async def api_move(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    body: MoveRequest,
    overwrite: bool = Query(False, description="是否允许覆盖已存在目标"),
):
    result = await VirtualFSService.move(body.src, body.dst, overwrite)
    return success(result)


@router.post("/rename")
@audit(action=AuditAction.UPDATE, description="重命名路径", body_fields=["src", "dst"])
@require_path_permission(PathAction.WRITE, "body.src")
async def api_rename(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    body: MoveRequest,
    overwrite: bool = Query(False, description="是否允许覆盖已存在目标"),
):
    result = await VirtualFSService.rename(body.src, body.dst, overwrite)
    return success(result)


@router.post("/copy")
@audit(action=AuditAction.CREATE, description="复制路径", body_fields=["src", "dst"])
@require_path_permission(PathAction.WRITE, "body.dst")
@require_path_permission(PathAction.READ, "body.src")
async def api_copy(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    body: MoveRequest,
    overwrite: bool = Query(False, description="是否覆盖已存在目标"),
):
    result = await VirtualFSService.copy(body.src, body.dst, overwrite)
    return success(result)


@router.post("/upload/{full_path:path}")
@audit(action=AuditAction.UPLOAD, description="流式上传文件")
@require_path_permission(PathAction.WRITE, "full_path")
async def upload_stream(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    full_path: str,
    file: UploadFile = File(...),
    overwrite: bool = Query(True, description="是否覆盖已存在文件"),
    chunk_size: int = Query(1024 * 1024, ge=8 * 1024, le=8 * 1024 * 1024, description="单次读取块大小"),
):
    result = await VirtualFSService.upload_stream_from_upload_file(full_path, file, chunk_size, overwrite)
    return success(result)


@router.put("/upload-raw/{full_path:path}")
@audit(action=AuditAction.UPLOAD, description="原始流上传文件")
@require_path_permission(PathAction.WRITE, "full_path")
async def upload_raw_stream(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    full_path: str,
    overwrite: bool = Query(True, description="是否覆盖已存在文件"),
):
    result = await VirtualFSService.upload_raw_stream(full_path, request, overwrite)
    return success(result)


@router.get("/{full_path:path}")
@audit(action=AuditAction.READ, description="浏览目录")
@require_path_permission(PathAction.READ, "full_path")
async def browse_fs(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    full_path: str,
    page_num: int = Query(1, alias="page", ge=1, description="页码"),
    page_size: int = Query(50, ge=1, le=500, description="每页条数"),
    sort_by: str = Query("name", description="按字段排序: name, size, mtime"),
    sort_order: str = Query("asc", description="排序顺序: asc, desc"),
    cursor: str | None = Query(None, description="游标分页位置"),
):
    data = await VirtualFSService.list_directory_with_permission(
        full_path, current_user.id, page_num, page_size, sort_by, sort_order, cursor
    )
    return success(data)


@router.delete("/{full_path:path}")
@audit(action=AuditAction.DELETE, description="删除路径")
@require_path_permission(PathAction.DELETE, "full_path")
async def api_delete(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    full_path: str,
):
    result = await VirtualFSService.delete(full_path)
    return success(result)


@router.get("/")
@audit(action=AuditAction.READ, description="浏览根目录")
async def root_listing(
    request: Request,
    current_user: Annotated[User, Depends(get_current_active_user)],
    page_num: int = Query(1, alias="page", ge=1, description="页码"),
    page_size: int = Query(50, ge=1, le=500, description="每页条数"),
    sort_by: str = Query("name", description="按字段排序: name, size, mtime"),
    sort_order: str = Query("asc", description="排序顺序: asc, desc"),
    cursor: str | None = Query(None, description="游标分页位置"),
):
    # 根目录不需要权限检查，但需要过滤无权限的子目录
    data = await VirtualFSService.list_directory_with_permission(
        "/", current_user.id, page_num, page_size, sort_by, sort_order, cursor
    )
    return success(data)