feat:脚本文件预览-backend

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}` | 软删 |
 

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

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.api._resources_common import (
-from backend.schemas.common import DownloadUrlRequest
+    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(

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": {}}

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)