From 807870e90df51ead3a871b3c380afc9f97f70a7e Mon Sep 17 00:00:00 2001 From: xiaozhu <2395895331@qq.com> Date: Mon, 31 Aug 2026 18:51:17 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E8=84=9A=E6=9C=AC=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E9=A2=84=E8=A7=88-backend?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- API.md | 2 + backend/src/backend/api/_resources_common.py | 130 +++++++++++ backend/src/backend/api/resources.py | 176 +------------- backend/src/backend/api/resources_content.py | 231 +++++++++++++++++++ backend/src/backend/main.py | 2 + 5 files changed, 374 insertions(+), 167 deletions(-) create mode 100644 backend/src/backend/api/_resources_common.py create mode 100644 backend/src/backend/api/resources_content.py diff --git a/API.md b/API.md index fadd101..efae065 100644 --- a/API.md +++ b/API.md @@ -508,6 +508,8 @@ queued ──→ running ──┬─→ succeeded | `POST` | `/api/v1/data-resources/uploads/{upload_id}/bind` | 绑定已完成上传为数据资源 | | `GET` | `/api/v1/data-resources` | 列表(owner 作用域 + visibility 过滤) | | `GET` | `/api/v1/data-resources/{id}` | 详情 | +| `GET` | `/api/v1/data-resources/{id}/content` | 同源流式读取文件字节(预览/下载) | +| `GET` | `/api/v1/data-resources/{id}/preview` | 表格抽样预览(csv/tsv, `limit` 默认 100) | | `POST` | `/api/v1/data-resources/{id}/download-url` | 生成 presigned GET URL | | `DELETE` | `/api/v1/data-resources/{id}` | 软删 | diff --git a/backend/src/backend/api/_resources_common.py b/backend/src/backend/api/_resources_common.py new file mode 100644 index 0000000..e35d2a3 --- /dev/null +++ b/backend/src/backend/api/_resources_common.py @@ -0,0 +1,130 @@ +"""数据资源路由共享的 payload / 可见性辅助。""" + +from __future__ import annotations + +import os +from pathlib import Path, PurePosixPath +from typing import Any + +from common.db.models import DataResources, StorageObjects +from common.storage import workspaces_root +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.api.dependencies import RequestContext +from backend.api.scripts import _escape_like_pattern, normalize_user_path + + +def build_list_resources_descendant_prefix(parent_path: str) -> str: + """Return the escaped materialized-path prefix for direct children.""" + normalized = normalize_user_path(parent_path) + escaped = _escape_like_pattern(normalized) + return f"{escaped}/" if escaped else "" + + +def compute_jupyter_relative_path(script_path: str, resource_relative: str) -> str: + """从当前脚本所在目录算到资源文件的 Jupyter 相对路径。""" + script_dir = PurePosixPath(script_path).parent.as_posix() + if not script_dir or script_dir == ".": + return resource_relative + return os.path.relpath(resource_relative, start=script_dir) + + +def resource_directory( + object_key: str, + workspace_id: str, + owner_user_id: str, +) -> str: + """从 object_key 解析资源所在目录(相对于用户根目录,根目录返回 "")。""" + prefix = f"{workspace_id}/{owner_user_id}/" + if not object_key.startswith(prefix): + return "" + tail = object_key[len(prefix):] + directory, _, _ = tail.rpartition("/") + return directory + + +def resource_payload( + resource: DataResources, + storage_object: StorageObjects, + owner_display_name: str | None = None, +) -> dict[str, Any]: + workspace_prefix = f"{resource.workspace_id}/" + user_prefix = f"{resource.owner_user_id}/" + jupyter_accessible_path = "" + absolute_path = "" + if storage_object.object_key and storage_object.object_key.startswith( + workspace_prefix + ): + remainder = storage_object.object_key[len(workspace_prefix):] + if remainder.startswith(user_prefix): + tail = remainder[len(user_prefix):] + jupyter_accessible_path = tail + absolute_path = ( + workspaces_root() + / resource.workspace_id + / resource.owner_user_id + / Path(tail) + ).as_posix() + else: + jupyter_accessible_path = remainder + elif storage_object.object_key: + jupyter_accessible_path = storage_object.object_key + return { + "resource_id": resource.resource_id, + "workspace_id": resource.workspace_id, + "storage_object_id": resource.storage_object_id, + "owner_user_id": resource.owner_user_id, + "owner_display_name": owner_display_name, + "resource_name": resource.resource_name, + "description": resource.description, + "visibility": resource.visibility, + "status": resource.status, + "created_at": resource.created_at.isoformat(), + "updated_at": resource.updated_at.isoformat(), + "file": { + "file_name": storage_object.file_name, + "file_extension": storage_object.file_extension, + "mime_type": storage_object.mime_type, + "size_bytes": storage_object.size_bytes, + "content_hash": storage_object.content_hash, + "object_status": storage_object.object_status, + }, + "jupyter_accessible_path": jupyter_accessible_path, + "absolute_path": absolute_path, + } + + +def can_view(resource: DataResources, context: RequestContext) -> bool: + if resource.owner_user_id == context.user.user_id: + return True + if resource.visibility in {"workspace", "public"}: + return True + return context.is_admin + + +async def get_visible_resource( + resource_id: str, + context: RequestContext, + session: AsyncSession, +) -> tuple[DataResources, StorageObjects]: + row = ( + await session.execute( + select(DataResources, StorageObjects) + .join( + StorageObjects, + StorageObjects.storage_object_id + == DataResources.storage_object_id, + ) + .where( + DataResources.resource_id == resource_id, + DataResources.workspace_id + == context.workspace.workspace_id, + DataResources.status == "active", + ) + ) + ).one_or_none() + if row is None or not can_view(row[0], context): + raise HTTPException(status.HTTP_404_NOT_FOUND, "resource not found") + return row diff --git a/backend/src/backend/api/resources.py b/backend/src/backend/api/resources.py index 8a16e6e..a1fef77 100644 --- a/backend/src/backend/api/resources.py +++ b/backend/src/backend/api/resources.py @@ -7,17 +7,13 @@ from __future__ import annotations -import os from datetime import UTC, datetime -from pathlib import Path, PurePosixPath from typing import Any from common.db.models import DataResources, StorageObjects, Users from common.ids import new_ulid -from common.storage import workspaces_root from common.storage.schemas import ( CreateUploadRequest, - DownloadUrlRequest, ) from fastapi import APIRouter, Depends, Header, HTTPException, Query, Request, status from sqlalchemy import func, or_, select @@ -28,8 +24,15 @@ from backend.api.dependencies import ( database_session, request_context, ) -from backend.api.scripts import _escape_like_pattern, normalize_user_path -from backend.schemas.common import DownloadUrlRequest +from backend.api._resources_common import ( + build_list_resources_descendant_prefix as _build_list_resources_descendant_prefix, + can_view, + compute_jupyter_relative_path, + get_visible_resource, + resource_directory, + resource_payload, +) +from backend.api.scripts import _escape_like_pattern from backend.schemas.resources import ( CompleteResourceUploadRequest, CreateResourceUploadRequest, @@ -37,7 +40,6 @@ from backend.schemas.resources import ( ) from backend.services.storage import ( acquire_named_lock, - create_download_url_payload, create_upload_record, release_named_lock, soft_delete_object, @@ -47,119 +49,6 @@ from backend.services.storage import ( router = APIRouter(prefix="/api/v1/data-resources", tags=["data-resources"]) -def _build_list_resources_descendant_prefix(parent_path: str) -> str: - """Return the escaped materialized-path prefix for direct children - of ``parent_path`` against ``StorageObjects.object_key``. - - The full object_key is ``{ws_id}/{owner_user_id}/{jupyter_path}``. - ``list_resources`` prepends ``{ws_id}/{owner_user_id}`` (the requester - by default, or the ``owner_user_id`` query param) to this prefix and - applies ``LIKE '{ws_id}/{owner}/{prefix}%' AND NOT LIKE '...%/%'`` so - only that owner's direct children under ``parent_path`` match. LIKE - wildcards in parent_path are escaped so folder names containing ``_`` - or ``%`` do not act as wildcards. - """ - normalized = normalize_user_path(parent_path) - escaped = _escape_like_pattern(normalized) - return f"{escaped}/" if escaped else "" - - -def compute_jupyter_relative_path(script_path: str, resource_relative: str) -> str: - """从当前脚本所在目录算到资源文件的 Jupyter 相对路径。 - - script_path / resource_relative 都是相对于 user root_dir 的 POSIX 路径。 - """ - script_dir = PurePosixPath(script_path).parent.as_posix() - if not script_dir or script_dir == ".": - return resource_relative - return os.path.relpath(resource_relative, start=script_dir) - - -def resource_directory( - object_key: str, - workspace_id: str, - owner_user_id: str, -) -> str: - """从 object_key 解析资源所在目录(相对于用户根目录,根目录返回 "")。 - - object_key 形如 ``{ws_id}/{user_id}/{target_path}/{file_name}``; - 不匹配该前缀的键(如无 ws/user 前缀的旧数据)统一视为根目录。 - """ - prefix = f"{workspace_id}/{owner_user_id}/" - if not object_key.startswith(prefix): - return "" - tail = object_key[len(prefix):] - directory, _, _ = tail.rpartition("/") - return directory - - -def resource_payload( - resource: DataResources, - storage_object: StorageObjects, - owner_display_name: str | None = None, -) -> dict[str, Any]: - # ``object_key`` now follows ``{ws_id}/{user_id}/{target_path}/{file_name}`` - # (target_path may be empty). Legacy objects still live under - # ``.resources/{file_name}`` and must remain readable. Both shapes share - # the same derivation: strip the workspace/user prefix and use the rest - # as the Jupyter-relative path. - workspace_prefix = f"{resource.workspace_id}/" - user_prefix = f"{resource.owner_user_id}/" - jupyter_accessible_path = "" - absolute_path = "" - if storage_object.object_key and storage_object.object_key.startswith( - workspace_prefix - ): - remainder = storage_object.object_key[len(workspace_prefix):] - if remainder.startswith(user_prefix): - tail = remainder[len(user_prefix):] - jupyter_accessible_path = tail - absolute_path = ( - workspaces_root() - / resource.workspace_id - / resource.owner_user_id - / Path(tail) - ).as_posix() - else: - jupyter_accessible_path = remainder - elif storage_object.object_key: - jupyter_accessible_path = storage_object.object_key - return { - "resource_id": resource.resource_id, - "workspace_id": resource.workspace_id, - "storage_object_id": resource.storage_object_id, - "owner_user_id": resource.owner_user_id, - "owner_display_name": owner_display_name, - "resource_name": resource.resource_name, - "description": resource.description, - "visibility": resource.visibility, - "status": resource.status, - "created_at": resource.created_at.isoformat(), - "updated_at": resource.updated_at.isoformat(), - "file": { - "file_name": storage_object.file_name, - "file_extension": storage_object.file_extension, - "mime_type": storage_object.mime_type, - "size_bytes": storage_object.size_bytes, - "content_hash": storage_object.content_hash, - "object_status": storage_object.object_status, - }, - "jupyter_accessible_path": jupyter_accessible_path, - "absolute_path": absolute_path, - } - - -def can_view(resource: DataResources, context: RequestContext) -> bool: - # 同一 workspace 内：owner 永远可见自己的资源（含 private）； - # 其他成员只见 visibility in {workspace, public} 的资源； - # admin 全部可见。 - if resource.owner_user_id == context.user.user_id: - return True - if resource.visibility in {"workspace", "public"}: - return True - return context.is_admin - - # 根据当前脚本位置计算资源的相对路径，便于 Notebook 中用相对路径读取文件。 @router.post("/{resource_id}/jupyter-relative-path") async def resource_jupyter_relative_path( @@ -465,31 +354,6 @@ async def list_resources( } -async def get_visible_resource( - resource_id: str, - context: RequestContext, - session: AsyncSession, -) -> tuple[DataResources, StorageObjects]: - row = ( - await session.execute( - select(DataResources, StorageObjects) - .join( - StorageObjects, - StorageObjects.storage_object_id - == DataResources.storage_object_id, - ) - .where( - DataResources.resource_id == resource_id, - DataResources.workspace_id - == context.workspace.workspace_id, - DataResources.status == "active", - ) - ) - ).one_or_none() - if row is None or not can_view(row[0], context): - raise HTTPException(status.HTTP_404_NOT_FOUND, "resource not found") - return row - # 查询单个数据资源的元数据与其关联文件信息。 @router.get("/{resource_id}") @@ -510,28 +374,6 @@ async def get_resource( } -# 为资源文件生成带时效的下载链接。 -@router.post("/{resource_id}/download-url") -async def resource_download_url( - resource_id: str, - payload: DownloadUrlRequest, - request: Request, - context: RequestContext = Depends(request_context), - session: AsyncSession = Depends(database_session), -) -> dict[str, Any]: - resource, _ = await get_visible_resource( - resource_id, - context, - session, - ) - data = await create_download_url_payload( - await session.get(StorageObjects, resource.storage_object_id), - DownloadUrlRequest(expires_seconds=payload.expires_seconds), - request, - ) - return {"request_id": context.request_id, "data": data["data"], "meta": {}} - - # 软删除数据资源及其关联对象，遵循存储层的回收站策略。 @router.delete("/{resource_id}") async def delete_resource( diff --git a/backend/src/backend/api/resources_content.py b/backend/src/backend/api/resources_content.py new file mode 100644 index 0000000..7789a28 --- /dev/null +++ b/backend/src/backend/api/resources_content.py @@ -0,0 +1,231 @@ +"""数据资源下载、同源内容流与表格抽样预览接口。 + +与 ``resources.py`` 共用前缀 ``/api/v1/data-resources``，在 ``main`` 中并列挂载。 +""" + +from __future__ import annotations + +import csv +import io +from typing import Any +from urllib.parse import quote + +from common.db.models import StorageObjects +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status +from fastapi.responses import StreamingResponse +from sqlalchemy.ext.asyncio import AsyncSession + +from backend.api.dependencies import ( + RequestContext, + database_session, + request_context, +) +from backend.api._resources_common import get_visible_resource +from backend.schemas.common import DownloadUrlRequest +from backend.services.storage import create_download_url_payload + +router = APIRouter(prefix="/api/v1/data-resources", tags=["data-resources"]) + +_PREVIEW_MAX_BYTES = 2 * 1024 * 1024 +_PREVIEW_DEFAULT_LIMIT = 100 +_PREVIEW_MAX_LIMIT = 500 +_TABLE_EXTENSIONS = {".csv", ".tsv"} + + +def _extension_of(file_name: str | None, resource_name: str | None) -> str: + for candidate in (file_name, resource_name): + if not candidate: + continue + lower = candidate.lower() + for ext in _TABLE_EXTENSIONS: + if lower.endswith(ext): + return ext + return "" + + +def _decode_preview_text(raw: bytes) -> str: + for encoding in ("utf-8-sig", "utf-8", "gb18030"): + try: + return raw.decode(encoding) + except UnicodeDecodeError: + continue + return raw.decode("utf-8", errors="replace") + + +async def _read_prefix_bytes(store: Any, object_key: str, max_bytes: int) -> bytes: + chunks: list[bytes] = [] + total = 0 + async for chunk in store.get_stream(object_key): + if not chunk: + continue + chunks.append(chunk) + total += len(chunk) + if total >= max_bytes: + break + data = b"".join(chunks) + return data[:max_bytes] + + +def _parse_table_preview( + text: str, + *, + delimiter: str, + limit: int, + byte_truncated: bool, +) -> dict[str, Any]: + reader = csv.reader(io.StringIO(text), delimiter=delimiter) + try: + header = next(reader) + except StopIteration: + return { + "kind": "table", + "columns": [], + "rows": [], + "row_count": 0, + "truncated": byte_truncated, + "delimiter": delimiter, + } + + columns = [str(cell) if cell is not None else "" for cell in header] + if not any(columns): + columns = [f"col_{index + 1}" for index in range(max(len(header), 1))] + + rows: list[list[str]] = [] + truncated = byte_truncated + for row in reader: + if len(rows) >= limit: + truncated = True + break + cells = [str(cell) if cell is not None else "" for cell in row] + if len(cells) < len(columns): + cells.extend([""] * (len(columns) - len(cells))) + elif len(cells) > len(columns): + cells = cells[: len(columns)] + rows.append(cells) + + return { + "kind": "table", + "columns": columns, + "rows": rows, + "row_count": len(rows), + "truncated": truncated, + "delimiter": delimiter, + } + + +@router.post("/{resource_id}/download-url") +async def resource_download_url( + resource_id: str, + payload: DownloadUrlRequest, + request: Request, + context: RequestContext = Depends(request_context), + session: AsyncSession = Depends(database_session), +) -> dict[str, Any]: + resource, _ = await get_visible_resource( + resource_id, + context, + session, + ) + data = await create_download_url_payload( + await session.get(StorageObjects, resource.storage_object_id), + DownloadUrlRequest(expires_seconds=payload.expires_seconds), + request, + ) + return {"request_id": context.request_id, "data": data["data"], "meta": {}} + + +@router.get("/{resource_id}/content") +async def resource_content( + resource_id: str, + request: Request, + context: RequestContext = Depends(request_context), + session: AsyncSession = Depends(database_session), +) -> StreamingResponse: + """同源流式读取资源字节，供前端预览器加载。""" + resource, storage_object = await get_visible_resource( + resource_id, + context, + session, + ) + if storage_object.object_status != "available": + raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found") + if ( + not storage_object.bucket_name + or not storage_object.object_key + or storage_object.bucket_name not in request.app.state.object_stores + ): + raise HTTPException( + status.HTTP_409_CONFLICT, + "object does not support content download", + ) + + file_name = storage_object.file_name or resource.resource_name or "file" + media_type = storage_object.mime_type or "application/octet-stream" + store = request.app.state.object_stores[storage_object.bucket_name] + stream = store.get_stream(storage_object.object_key) + headers = { + "Content-Disposition": f"inline; filename*=UTF-8''{quote(file_name)}", + "Cache-Control": "private, no-store", + } + if storage_object.size_bytes is not None: + headers["Content-Length"] = str(storage_object.size_bytes) + + return StreamingResponse( + stream, + media_type=media_type, + headers=headers, + ) + + +@router.get("/{resource_id}/preview") +async def resource_preview( + resource_id: str, + request: Request, + context: RequestContext = Depends(request_context), + session: AsyncSession = Depends(database_session), + limit: int = Query(default=_PREVIEW_DEFAULT_LIMIT, ge=1, le=_PREVIEW_MAX_LIMIT), +) -> dict[str, Any]: + """表格类数据资源抽样预览（csv / tsv）。""" + resource, storage_object = await get_visible_resource( + resource_id, + context, + session, + ) + if storage_object.object_status != "available": + raise HTTPException(status.HTTP_404_NOT_FOUND, "object not found") + if ( + not storage_object.bucket_name + or not storage_object.object_key + or storage_object.bucket_name not in request.app.state.object_stores + ): + raise HTTPException( + status.HTTP_409_CONFLICT, + "object does not support preview", + ) + + extension = _extension_of(storage_object.file_name, resource.resource_name) + if extension not in _TABLE_EXTENSIONS: + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, + "仅支持预览 .csv / .tsv 表格文件", + ) + + store = request.app.state.object_stores[storage_object.bucket_name] + raw = await _read_prefix_bytes( + store, + storage_object.object_key, + _PREVIEW_MAX_BYTES, + ) + byte_truncated = ( + storage_object.size_bytes is not None + and storage_object.size_bytes > len(raw) + ) or len(raw) >= _PREVIEW_MAX_BYTES + text = _decode_preview_text(raw) + delimiter = "\t" if extension == ".tsv" else "," + payload = _parse_table_preview( + text, + delimiter=delimiter, + limit=limit, + byte_truncated=byte_truncated, + ) + return {"request_id": context.request_id, "data": payload, "meta": {}} diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index cd8eab5..6c6ad3a 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -38,6 +38,7 @@ from backend.api.auth import router as auth_router from backend.api.jupyter import router as jupyter_router from backend.api.platform import router as platform_router from backend.api.resources import router as resources_router +from backend.api.resources_content import router as resources_content_router from backend.api.schedules.runs import router as schedule_runs_router from backend.api.schedules.schedules import router as schedules_router from backend.api.scripts import router as scripts_router @@ -107,6 +108,7 @@ app = create_service_app( app.include_router(auth_router) app.include_router(jupyter_router) app.include_router(resources_router) +app.include_router(resources_content_router) app.include_router(schedule_runs_router) app.include_router(schedules_router) app.include_router(scripts_router)