feat: SFTP 上传的文件经磁盘扫描补录后可在文件浏览页展示
SFTP 上传的文件直接落盘到 upload_root(chroot 根),不经 HTTP 路径,无 DB 记录, 文件浏览页(读 uploaded_file 表)看不到。新增磁盘扫描机制: - UploadService.scan_sftp_files(force=False):遍历 uploads/ 下所有文件(排除 .work/ 与 .part 残品),与 DB 已有 storage_path 比对,磁盘有 DB 无的补录(source="sftp", 流式算 size + sha256)。 - 频率限制:模块级 _last_scan_time,两次扫描间隔 < 3s 则跳过返回 0(只读 DB 缓存), force=True 强制扫。 - file_admin_controller 的 list 接口返回前触发 scan_sftp_files(),文件浏览页打开/ 刷新即自动同步 SFTP 文件(受 3s 频率限制,不会每次翻页都扫磁盘)。 - UploadedFileDAO 加 list_storage_paths() 批量查已记录路径集合供比对。 SFTP 文件补录后可正常下载/删除(与 HTTP 上传文件一致)。
This commit is contained in:
@@ -43,7 +43,8 @@ class BatchDeleteResult(BaseModel):
|
||||
"",
|
||||
response_model=FileListResponse,
|
||||
summary="列出已上传文件(需鉴权)",
|
||||
description="供文件浏览页使用;字段同公开 /api/files,但要求 docs Basic Auth。支持分页。",
|
||||
description="供文件浏览页使用;字段同公开 /api/files,但要求 docs Basic Auth。支持分页。"
|
||||
"返回前会触发磁盘扫描(频率限制 3s),把 SFTP 上传但未入库的文件补录进来。",
|
||||
)
|
||||
def list_files(
|
||||
limit: int = Query(100, ge=1, le=10000, description="每页条数,1-10000"),
|
||||
@@ -51,6 +52,7 @@ def list_files(
|
||||
service: UploadService = Depends(_service),
|
||||
_: str = Depends(require_docs_auth),
|
||||
) -> FileListResponse:
|
||||
service.scan_sftp_files() # 频率限制内置:3s 内重复只返回 DB 缓存
|
||||
total, items = service.list_files(limit=limit, offset=offset)
|
||||
return FileListResponse(total=total, items=items)
|
||||
|
||||
|
||||
@@ -29,6 +29,11 @@ class UploadedFileDAO:
|
||||
"""返回数据库中文件总条数。"""
|
||||
return self.db.scalar(select(func.count()).select_from(UploadedFile)) or 0
|
||||
|
||||
def list_storage_paths(self) -> set[str]:
|
||||
"""返回所有已记录的 storage_path 集合(供磁盘扫描比对,识别未入库文件)。"""
|
||||
stmt = select(UploadedFile.storage_path)
|
||||
return {row for row in self.db.scalars(stmt).all()}
|
||||
|
||||
def list(self, limit: int = 100, offset: int = 0) -> list[UploadedFile]:
|
||||
stmt = (
|
||||
select(UploadedFile)
|
||||
|
||||
@@ -10,6 +10,7 @@ from __future__ import annotations
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
from collections.abc import Iterator
|
||||
from datetime import datetime, timezone
|
||||
@@ -24,6 +25,10 @@ from ..schemas.file import FileUploadResponse, UploadedFileOut
|
||||
|
||||
logger = logging.getLogger("zikai.upload")
|
||||
|
||||
# 磁盘扫描频率限制:两次扫描至少间隔此秒数,否则跳过(只返回 DB 缓存)。
|
||||
_SCAN_MIN_INTERVAL = 3.0
|
||||
_last_scan_time: float = 0.0
|
||||
|
||||
|
||||
def hash_stream(chunks: Iterator[bytes]) -> tuple[int, str]:
|
||||
"""对一段字节块序列流式计算 (size, sha256)。供多条上传路径复用。"""
|
||||
@@ -69,6 +74,72 @@ class UploadService:
|
||||
|
||||
# ---------------- 查询 ----------------
|
||||
|
||||
def scan_sftp_files(self, force: bool = False) -> int:
|
||||
"""扫描 uploads/ 目录,把磁盘上有但 DB 未记录的文件补录入库。
|
||||
|
||||
SFTP 上传的文件直接落盘到 upload_root(chroot 根),不经过 HTTP 路径,
|
||||
因此没有 DB 记录。本方法遍历磁盘文件,与 DB 已有 storage_path 比对,
|
||||
为缺失项创建记录(source="sftp",计算 size + sha256)。
|
||||
|
||||
频率限制:两次扫描间隔 < _SCAN_MIN_INTERVAL(3s)则跳过,force=True 强制扫。
|
||||
返回本次新补录的条数(跳过时返回 0)。
|
||||
"""
|
||||
global _last_scan_time
|
||||
now = time.monotonic()
|
||||
if not force and (now - _last_scan_time) < _SCAN_MIN_INTERVAL:
|
||||
return 0
|
||||
_last_scan_time = now
|
||||
|
||||
known = self.dao.list_storage_paths()
|
||||
new_count = 0
|
||||
# 排除分片会话暂存目录与 .part 残品
|
||||
skip_dirs = {".work"}
|
||||
for abs_path in self.upload_root.rglob("*"):
|
||||
if not abs_path.is_file():
|
||||
continue
|
||||
if abs_path.suffix == ".part":
|
||||
continue
|
||||
# 跳过 .work 目录下的任何文件
|
||||
rel = abs_path.relative_to(self.upload_root)
|
||||
if rel.parts and rel.parts[0] in skip_dirs:
|
||||
continue
|
||||
storage_path = str(rel).replace("\\", "/")
|
||||
if storage_path in known:
|
||||
continue
|
||||
# 磁盘有、DB 无:补录
|
||||
try:
|
||||
size = abs_path.stat().st_size
|
||||
sha256 = self._hash_file(abs_path)
|
||||
except Exception as exc:
|
||||
logger.warning("扫描文件失败 path=%s: %s", abs_path, exc)
|
||||
continue
|
||||
entity = UploadedFile(
|
||||
storage_path=storage_path,
|
||||
original_filename=abs_path.name,
|
||||
content_type="",
|
||||
size_bytes=size,
|
||||
sha256=sha256,
|
||||
source="sftp",
|
||||
uploaded_by="sftp",
|
||||
)
|
||||
try:
|
||||
self.dao.create(entity)
|
||||
new_count += 1
|
||||
known.add(storage_path)
|
||||
except Exception as exc:
|
||||
logger.warning("补录 SFTP 文件失败 path=%s: %s", storage_path, exc)
|
||||
if new_count:
|
||||
logger.info("SFTP 文件扫描补录 %d 个", new_count)
|
||||
return new_count
|
||||
|
||||
def _hash_file(self, path: Path) -> str:
|
||||
"""流式计算文件 sha256(避免大文件一次性读入内存)。"""
|
||||
h = hashlib.sha256()
|
||||
with path.open("rb") as f:
|
||||
while chunk := f.read(self.chunk_bytes):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
def list_files(self, limit: int = 100, offset: int = 0) -> tuple[int, list[UploadedFileOut]]:
|
||||
total = self.dao.count()
|
||||
rows = self.dao.list(limit=limit, offset=offset)
|
||||
|
||||
Reference in New Issue
Block a user