diff --git a/app/controllers/file_admin_controller.py b/app/controllers/file_admin_controller.py index cb0bae5..4caba02 100644 --- a/app/controllers/file_admin_controller.py +++ b/app/controllers/file_admin_controller.py @@ -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) diff --git a/app/dao/uploaded_file_dao.py b/app/dao/uploaded_file_dao.py index 3df2581..ec13292 100644 --- a/app/dao/uploaded_file_dao.py +++ b/app/dao/uploaded_file_dao.py @@ -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) diff --git a/app/services/upload_service.py b/app/services/upload_service.py index b405ae6..0c448be 100644 --- a/app/services/upload_service.py +++ b/app/services/upload_service.py @@ -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)