From 09705a88438d6643318fa4c800fefb70ff2e91d3 Mon Sep 17 00:00:00 2001 From: zikai Date: Wed, 22 Jul 2026 01:25:41 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20SFTP=20=E4=B8=8A=E4=BC=A0=E7=9A=84?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E7=BB=8F=E7=A3=81=E7=9B=98=E6=89=AB=E6=8F=8F?= =?UTF-8?q?=E8=A1=A5=E5=BD=95=E5=90=8E=E5=8F=AF=E5=9C=A8=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E6=B5=8F=E8=A7=88=E9=A1=B5=E5=B1=95=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 上传文件一致）。 --- app/controllers/file_admin_controller.py | 4 +- app/dao/uploaded_file_dao.py | 5 ++ app/services/upload_service.py | 71 ++++++++++++++++++++++++ 3 files changed, 79 insertions(+), 1 deletion(-) 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)