init

api_routes.py

@@ -178,6 +178,8 @@ from utils import (
     compress_image_by_file_size,
     convert_image_format,
     upload_file_to_bos,
+    ensure_bos_resources,
+    download_bos_directory,
 )
 from cleanup_scheduler import get_cleanup_status, manual_cleanup
 
@@ -1763,6 +1765,103 @@ async def get_available_models():
     }
 
 
+@api_router.post("/sync_resources", tags=["系统维护"])
+@log_api_params
+async def sync_bos_resources(
+    force_download: bool = Query(False, description="是否强制重新下载已存在的文件"),
+    include_background: bool = Query(
+        False, description="是否同步配置中标记为后台的资源"
+    ),
+    bos_prefix: str | None = Query(
+        None, description="自定义 BOS 前缀，例如 20220620/models"
+    ),
+    destination_dir: str | None = Query(
+        None, description="自定义本地目录，例如 /opt/models/custom"
+    ),
+    background: bool = Query(
+        False, description="与自定义前缀搭配使用时，是否在后台异步下载"
+    ),
+):
+    """
+    手动触发 BOS 资源同步。
+    - 若提供 bos_prefix 与 destination_dir，则按指定路径同步；
+    - 否则根据配置的 BOS_DOWNLOAD_TARGETS 执行批量同步。
+    """
+    start_time = time.perf_counter()
+
+    if (bos_prefix and not destination_dir) or (destination_dir and not bos_prefix):
+        raise HTTPException(status_code=400, detail="bos_prefix 和 destination_dir 需要同时提供")
+
+    if bos_prefix and destination_dir:
+        dest_path = os.path.abspath(os.path.expanduser(destination_dir.strip()))
+
+        async def _sync_single():
+            return await asyncio.to_thread(
+                download_bos_directory,
+                bos_prefix.strip(),
+                dest_path,
+                force_download=force_download,
+            )
+
+        if background:
+            async def _background_task():
+                success = await _sync_single()
+                if success:
+                    logger.info(
+                        "后台 BOS 下载完成: prefix=%s -> %s", bos_prefix, dest_path
+                    )
+                else:
+                    logger.warning(
+                        "后台 BOS 下载失败: prefix=%s -> %s", bos_prefix, dest_path
+                    )
+
+            asyncio.create_task(_background_task())
+            elapsed = time.perf_counter() - start_time
+            return {
+                "success": True,
+                "force_download": force_download,
+                "include_background": False,
+                "bos_prefix": bos_prefix,
+                "destination_dir": dest_path,
+                "elapsed_seconds": round(elapsed, 3),
+                "message": "后台下载任务已启动",
+            }
+
+        success = await _sync_single()
+        elapsed = time.perf_counter() - start_time
+        return {
+            "success": bool(success),
+            "force_download": force_download,
+            "include_background": False,
+            "bos_prefix": bos_prefix,
+            "destination_dir": dest_path,
+            "elapsed_seconds": round(elapsed, 3),
+            "message": "资源同步完成" if success else "资源同步失败，请查看日志",
+        }
+
+    # 未指定前缀时，按配置批量同步
+    success = await asyncio.to_thread(
+        ensure_bos_resources,
+        force_download,
+        include_background,
+    )
+    elapsed = time.perf_counter() - start_time
+    message = (
+        "后台下载任务已启动，将在后台继续运行"
+        if not include_background
+        else "资源同步完成"
+    )
+    return {
+        "success": bool(success),
+        "force_download": force_download,
+        "include_background": include_background,
+        "elapsed_seconds": round(elapsed, 3),
+        "message": message,
+        "bos_prefix": None,
+        "destination_dir": None,
+    }
+
+
 @api_router.post("/restore")
 @log_api_params
 async def restore_old_photo(

config.py

@@ -306,11 +306,12 @@ CLEANUP_AGE_HOURS = float(os.environ.get("CLEANUP_AGE_HOURS", 12.0))  # 清理
 
 # BOS 自动同步清单：定义 BOS 路径和本地目录的映射，启动时可迭代该结构完成批量下载
 BOS_DOWNLOAD_TARGETS = [
-    {
-        "description": "明星图库数据集",
-        "bos_prefix": BOS_CELEBRITY_PREFIX,
-        "destination": CELEBRITY_DATASET_DIR,
-    },
+    # {
+    #     "description": "明星图库数据集",
+    #     "bos_prefix": BOS_CELEBRITY_PREFIX,
+    #     "destination": CELEBRITY_DATASET_DIR,
+    #     "background": True,
+    # },
     {
         "description": "AI 模型权重",
         "bos_prefix": BOS_MODELS_PREFIX,

utils.py

@@ -37,6 +37,8 @@ _BOS_CLIENT_INITIALIZED = False
 _BOS_CLIENT_LOCK = threading.Lock()
 _BOS_DOWNLOAD_LOCK = threading.Lock()
 _BOS_DOWNLOAD_COMPLETED = False
+_BOS_BACKGROUND_EXECUTOR = None
+_BOS_BACKGROUND_FUTURES = []
 _IMAGES_DIR_ABS = os.path.abspath(os.path.expanduser(IMAGES_DIR))
 
 
@@ -236,16 +238,24 @@ def download_bos_directory(prefix: str, destination_dir: str, *, force_download:
     return downloaded > 0 or skipped > 0
 
 
-def ensure_bos_resources(force_download: bool = False) -> bool:
+def _get_background_executor() -> ThreadPoolExecutor:
+    global _BOS_BACKGROUND_EXECUTOR
+    if _BOS_BACKGROUND_EXECUTOR is None:
+        _BOS_BACKGROUND_EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix="bos-bg")
+    return _BOS_BACKGROUND_EXECUTOR
+
+
+def ensure_bos_resources(force_download: bool = False, include_background: bool = False) -> bool:
     """
     根据配置的 BOS_DOWNLOAD_TARGETS 同步启动所需的模型与数据资源。
     :param force_download: 是否强制重新同步所有资源
+    :param include_background: 是否将标记为后台任务的目标也同步为阻塞任务
     :return: 资源是否已准备就绪
     """
-    global _BOS_DOWNLOAD_COMPLETED
+    global _BOS_DOWNLOAD_COMPLETED, _BOS_BACKGROUND_FUTURES
 
     with _BOS_DOWNLOAD_LOCK:
-        if _BOS_DOWNLOAD_COMPLETED and not force_download:
+        if _BOS_DOWNLOAD_COMPLETED and not force_download and not include_background:
             return True
 
         targets = BOS_DOWNLOAD_TARGETS or []
@@ -255,6 +265,7 @@ def ensure_bos_resources(force_download: bool = False) -> bool:
             return True
 
         download_jobs = []
+        background_jobs = []
         for target in targets:
             if not isinstance(target, dict):
                 logger.warning("无效的 BOS 下载配置项: %r", target)
@@ -263,34 +274,87 @@ def ensure_bos_resources(force_download: bool = False) -> bool:
             prefix = target.get("bos_prefix")
             destination = target.get("destination")
             description = target.get("description") or prefix or "<unnamed>"
+            background_flag = bool(target.get("background"))
 
             if not prefix or not destination:
                 logger.warning("缺少必要字段，无法处理 BOS 下载配置: %r", target)
                 continue
 
-            download_jobs.append(
-                {
-                    "description": description,
-                    "prefix": prefix,
-                    "destination": destination,
-                }
-            )
+            job = {
+                "description": description,
+                "prefix": prefix,
+                "destination": destination,
+            }
 
-        if not download_jobs:
-            logger.info("未找到有效的 BOS 下载目标，跳过资源同步")
-            _BOS_DOWNLOAD_COMPLETED = True
-            return True
+            if background_flag and not include_background:
+                background_jobs.append(job)
+            else:
+                download_jobs.append(job)
 
         results = []
-        max_workers = min(len(download_jobs), max(os.cpu_count() or 1, 1))
-        if max_workers <= 0:
-            max_workers = 1
+        if download_jobs:
+            max_workers = min(len(download_jobs), max(os.cpu_count() or 1, 1))
+            if max_workers <= 0:
+                max_workers = 1
+
+            with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="bos-sync") as executor:
+                future_to_job = {}
+                for job in download_jobs:
+                    logger.info(
+                        "准备同步 BOS 资源: %s (prefix=%s -> %s)",
+                        job["description"],
+                        job["prefix"],
+                        job["destination"],
+                    )
+                    future = executor.submit(
+                        download_bos_directory,
+                        job["prefix"],
+                        job["destination"],
+                        force_download=force_download,
+                    )
+                    future_to_job[future] = job
+
+                for future in as_completed(future_to_job):
+                    job = future_to_job[future]
+                    description = job["description"]
+                    try:
+                        success = future.result()
+                    except Exception as exc:
+                        logger.warning("BOS 资源同步异常: %s (%s)", description, exc)
+                        success = False
+
+                    if success:
+                        logger.info("BOS 资源已就绪: %s", description)
+                    else:
+                        logger.warning("BOS 资源同步失败: %s", description)
+                    results.append(success)
 
-        with ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix="bos-sync") as executor:
-            future_to_job = {}
-            for job in download_jobs:
+        all_ready = all(results) if results else True
+        if all_ready:
+            _BOS_DOWNLOAD_COMPLETED = True
+
+        if background_jobs:
+            executor = _get_background_executor()
+
+            def _make_callback(description: str):
+                def _background_done(fut):
+                    try:
+                        success = fut.result()
+                        if success:
+                            logger.info("后台 BOS 资源已就绪: %s", description)
+                        else:
+                            logger.warning("后台 BOS 资源同步失败: %s", description)
+                    except Exception as exc:
+                        logger.warning("后台 BOS 资源同步异常: %s (%s)", description, exc)
+                    finally:
+                        with _BOS_DOWNLOAD_LOCK:
+                            if fut in _BOS_BACKGROUND_FUTURES:
+                                _BOS_BACKGROUND_FUTURES.remove(fut)
+                return _background_done
+
+            for job in background_jobs:
                 logger.info(
-                    "准备同步 BOS 资源: %s (prefix=%s -> %s)",
+                    "后台同步 BOS 资源: %s (prefix=%s -> %s)",
                     job["description"],
                     job["prefix"],
                     job["destination"],
@@ -301,26 +365,8 @@ def ensure_bos_resources(force_download: bool = False) -> bool:
                     job["destination"],
                     force_download=force_download,
                 )
-                future_to_job[future] = job
-
-            for future in as_completed(future_to_job):
-                job = future_to_job[future]
-                description = job["description"]
-                try:
-                    success = future.result()
-                except Exception as exc:
-                    logger.warning("BOS 资源同步异常: %s (%s)", description, exc)
-                    success = False
-
-                if success:
-                    logger.info("BOS 资源已就绪: %s", description)
-                else:
-                    logger.warning("BOS 资源同步失败: %s", description)
-                results.append(success)
-
-        all_ready = all(results) if results else True
-        if all_ready:
-            _BOS_DOWNLOAD_COMPLETED = True
+                future.add_done_callback(_make_callback(job["description"]))
+                _BOS_BACKGROUND_FUTURES.append(future)
 
         return all_ready