Upload 3 files

router_items.py

@@ -7,6 +7,7 @@ import os
 import urllib.request
 import urllib.error
 import json
+import asyncio
 import 数据库连接 as db
 from models import ItemCreate, ItemUpdate, RatingRequest
 from 安全认证 import require_auth, check_ownership
@@ -15,6 +16,13 @@ from db_utils import record_view, sort_cache
 
 router = APIRouter()
 
+def _get_version_str(versions_db: dict, item_id: str) -> str:
+    """兼容新旧格式获取版本hash字符串"""
+    val = versions_db.get(item_id, "")
+    if isinstance(val, dict):
+        return val.get("hash", "") or ""
+    return val or ""
+
 def get_last_6_months():
     res = []
     today = datetime.date.today()
@@ -69,7 +77,7 @@ async def get_items(type: str = "tool", sort: str = "time", limit: int = 50): #
     for item in filtered_items:
         item["commentsData"] = comments_db.get(item["id"], [])
         item["comments"] = len(item["commentsData"])
-        item["latest_version"] = versions_db.get(item["id"], "")
+        item["latest_version"] = _get_version_str(versions_db, item["id"])
         
         # 🔴 【绝对核心防线】：在下发给前端前，强行在内存中抹除敏感信息！
         item["has_private_token"] = bool(item.get("github_token"))
@@ -331,7 +339,10 @@ async def update_item(item_id: str, update_data: ItemUpdate, current_user: str =
                 if update_data.title is not None: item["title"] = update_data.title
                 if update_data.shortDesc is not None: item["shortDesc"] = update_data.shortDesc
                 if update_data.fullDesc is not None: item["fullDesc"] = update_data.fullDesc
+                old_link = item.get("link", "")
                 if update_data.link is not None: item["link"] = update_data.link
+                result_holder["link_changed"] = old_link != item.get("link", "")
+                result_holder["new_link"] = item.get("link", "")
                 if update_data.coverUrl is not None: item["coverUrl"] = update_data.coverUrl
                 if update_data.imageUrls is not None: item["imageUrls"] = update_data.imageUrls  # 🖼️ 效果展示图列表
                 
@@ -383,6 +394,49 @@ async def update_item(item_id: str, update_data: ItemUpdate, current_user: str =
     sort_cache.invalidate("items:")
     invalidate_cache("items.json")
     
+    # 🔄 [缓存刷新] 资源 link 变更时清除 ZIP 缓存元信息
+    if result_holder.get("link_changed"):
+        try:
+            versions_db = db.load_data("versions.json", default_data={})
+            entry = versions_db.get(item_id)
+            if entry is not None:
+                if isinstance(entry, str):
+                    entry = {"hash": entry}
+                had_meta = "cached_at" in entry or "zip_size" in entry
+                entry.pop("cached_at", None)
+                entry.pop("zip_size", None)
+                if had_meta:
+                    versions_db[item_id] = entry
+                    db.save_data("versions.json", versions_db)
+                    print(f"[缓存刷新] 已清除 item {item_id} 的 ZIP 缓存元信息")
+            
+            # 可选：触发异步预缓存
+            new_link = result_holder.get("new_link", "")
+            if new_link and new_link.startswith("https://github.com/"):
+                try:
+                    from 云端_定时版本检测引擎 import precache_github_zip
+                    # 获取当前 item 的 token 和 version_hash
+                    items_db = db.load_data("items.json", default_data=[])
+                    token = None
+                    for it in items_db:
+                        if it["id"] == item_id:
+                            token = it.get("github_token")
+                            break
+                    
+                    versions_db = db.load_data("versions.json", default_data={})
+                    entry = versions_db.get(item_id, {})
+                    version_hash = entry.get("hash", "") if isinstance(entry, dict) else entry
+                    
+                    if version_hash:
+                        asyncio.create_task(precache_github_zip(new_link, token, item_id, version_hash))
+                        print(f"[缓存刷新] 已触发 item {item_id} 的异步预缓存")
+                except ImportError:
+                    print(f"[缓存刷新] 预缓存模块导入失败，跳过异步预缓存")
+                except Exception as e:
+                    print(f"[缓存刷新] 触发异步预缓存失败: {e}")
+        except Exception as e:
+            print(f"[缓存刷新] 缓存刷新失败（不影响主流程）: {e}")
+    
     result = {"status": "success"}
     if result_holder.get("price_change_info"):
         price_change_info = result_holder["price_change_info"]
@@ -472,7 +526,7 @@ async def get_item_by_id(item_id: str):
             # 添加关联数据
             item["commentsData"] = comments_db.get(item_id, [])
             item["comments"] = len(item["commentsData"])
-            item["latest_version"] = versions_db.get(item_id, "")
+            item["latest_version"] = _get_version_str(versions_db, item_id)
             
             # 🔴 【安全防线】：抹除敏感信息
             item["has_private_token"] = bool(item.get("github_token"))
@@ -488,7 +542,7 @@ async def get_item_by_id(item_id: str):
 async def get_item_version(item_id: str):
     """获取单个资源的最新版本号"""
     versions_db = db.load_data("versions.json", default_data={})
-    return {"status": "success", "version": versions_db.get(item_id, "")}
+    return {"status": "success", "version": _get_version_str(versions_db, item_id)}
 
 @router.delete("/api/items/{item_id}")
 async def delete_item(item_id: str, current_user: str = Depends(require_auth)):

router_proxy.py

@@ -57,6 +57,64 @@ async def proxy_github_zip(req_data: ProxyGithubZipRequest, db: Session = Depend
         if not owned:
             return JSONResponse(content={"error": "🚨 拒绝访问：未找到购买记录！"}, status_code=403)
 
+    # ===== ZIP 缓存优先机制（鉴权后、GitHub请求前） =====
+    try:
+        versions_db = json_db.load_data("versions.json", default_data={})
+        item_version = versions_db.get(req_data.item_id)
+        
+        # 兼容新格式：{"item_id": {"hash": "...", "cached_at": "...", "zip_size": ...}}
+        if isinstance(item_version, dict) and item_version.get("cached_at"):
+            cache_filename = f"zips/{req_data.item_id}_{item_version['hash']}.zip"
+            cache_url = f"https://huggingface.co/datasets/ZHIWEI666/ComfyUI-Ranking/resolve/main/{cache_filename}"
+            
+            cache_client = httpx.AsyncClient(follow_redirects=True)
+            head_resp = None
+            try:
+                head_resp = await cache_client.head(cache_url, timeout=10, follow_redirects=True)
+            except Exception as head_err:
+                print(f"[ZIP缓存] HEAD验证失败: {head_err}")
+                await cache_client.aclose()
+                raise
+            
+            if head_resp.status_code == 200:
+                print(f"[ZIP缓存] 命中: {req_data.item_id}, 大小: {item_version.get('zip_size', 'unknown')}")
+                
+                content_length = head_resp.headers.get('content-length', '') or str(item_version.get('zip_size', ''))
+                
+                try:
+                    stream_resp = await cache_client.send(
+                        cache_client.build_request("GET", cache_url),
+                        stream=True
+                    )
+                except Exception as stream_err:
+                    print(f"[ZIP缓存] 流请求失败: {stream_err}")
+                    await cache_client.aclose()
+                    raise
+                
+                cache_headers = {}
+                if content_length:
+                    cache_headers["Content-Length"] = content_length
+                
+                async def cache_stream_generator():
+                    try:
+                        if stream_resp.status_code != 200:
+                            yield b"CACHE_DOWNLOAD_FAILED"
+                            return
+                        async for chunk in stream_resp.aiter_bytes():
+                            yield chunk
+                    except Exception:
+                        yield b"CACHE_DOWNLOAD_FAILED"
+                    finally:
+                        await stream_resp.aclose()
+                        await cache_client.aclose()
+                
+                return StreamingResponse(cache_stream_generator(), media_type="application/zip", headers=cache_headers)
+            
+            # 缓存未命中：关闭 client 后继续原有逻辑
+            await cache_client.aclose()
+    except Exception as e:
+        print(f"[ZIP缓存] 查询失败，fallback到GitHub直连: {e}")
+
     # 2. 解析 GitHub 仓库信息
     repo_url = item.get("link", "").rstrip("/")
     if not repo_url.startswith("https://github.com/"):

云端_定时版本检测引擎.py

@@ -3,13 +3,46 @@ import asyncio
 import datetime
 import httpx
 import logging
+import os
+import tempfile
+import zipfile
 import 数据库连接 as db
 from notifications import add_notification
 from database_sql import SessionLocal
 from models_sql import Ownership
+from huggingface_hub import upload_file, HfApi
 
 logger = logging.getLogger("ComfyUI-Ranking.VersionCheck")
 
+# ==========================================
+# 预缓存配置常量
+# ==========================================
+HF_REPO_ID = "ZHIWEI666/ComfyUI-Ranking"
+ZIP_MAX_SIZE = 500 * 1024 * 1024  # 500MB
+CACHE_EXPIRE_DAYS = 30
+PRECACHE_BATCH_SIZE = 5
+PRECACHE_INTERVAL = 10
+
+
+def get_version_hash(versions_db: dict, item_id: str) -> str | None:
+    """兼容新旧格式获取 hash"""
+    val = versions_db.get(item_id)
+    if isinstance(val, str):
+        return val
+    elif isinstance(val, dict):
+        return val.get("hash")
+    return None
+
+
+def _normalize_version_entry(versions_db: dict, item_id: str):
+    """将旧格式字符串转换为新格式字典（原地修改）"""
+    val = versions_db.get(item_id)
+    if isinstance(val, str):
+        versions_db[item_id] = {"hash": val}
+    elif not isinstance(val, dict):
+        versions_db[item_id] = {"hash": ""}
+
+
 async def fetch_latest_github_hash(repo_url, token):
     """
     请求 GitHub API 获取最新版的 Commit Hash
@@ -151,6 +184,218 @@ async def trigger_update_notifications(updated_items: list):
             session.close()
 
 
+async def precache_github_zip(repo_url: str, token: str | None, item_id: str, version_hash: str):
+    """预下载 GitHub ZIP 并上传到 HF Dataset 缓存"""
+    try:
+        parts = repo_url.rstrip("/").split("/")
+        if len(parts) < 2:
+            logger.warning(f"[预缓存] 无效的仓库地址: {repo_url}")
+            return False
+        owner, repo = parts[-2], parts[-1].replace(".git", "")
+        zip_url = f"https://api.github.com/repos/{owner}/{repo}/zipball"
+
+        headers = {
+            "Accept": "application/vnd.github.v3+json",
+            "User-Agent": "ComfyUI-Hub"
+        }
+        if token:
+            headers["Authorization"] = f"token {token}"
+
+        hf_token = os.environ.get("HF_TOKEN")
+        if not hf_token:
+            logger.warning("[预缓存] HF_TOKEN 环境变量未设置，跳过预缓存")
+            return False
+
+        # 下载到临时文件（stream 模式，超时 300s）
+        temp_path = None
+        async with httpx.AsyncClient(follow_redirects=True, timeout=httpx.Timeout(300.0, connect=30.0)) as client:
+            logger.info(f"[预缓存] 开始下载 {item_id} ZIP: {zip_url}")
+            async with client.stream("GET", zip_url, headers=headers) as resp:
+                if resp.status_code != 200:
+                    logger.warning(f"[预缓存] 下载 ZIP 失败，状态码: {resp.status_code}, URL: {zip_url}")
+                    return False
+
+                fd, temp_path = tempfile.mkstemp(suffix=".zip", prefix=f"precache_{item_id}_")
+                try:
+                    total_size = 0
+                    oversized = False
+                    with os.fdopen(fd, "wb") as f:
+                        async for chunk in resp.aiter_bytes(chunk_size=8192):
+                            f.write(chunk)
+                            total_size += len(chunk)
+                            if total_size > ZIP_MAX_SIZE:
+                                logger.warning(
+                                    f"[预缓存] ZIP 超过 {ZIP_MAX_SIZE // (1024*1024)}MB 限制，跳过: "
+                                    f"{item_id} (已下载 {total_size} 字节)"
+                                )
+                                oversized = True
+                                break
+
+                    if oversized:
+                        if temp_path and os.path.exists(temp_path):
+                            try:
+                                os.remove(temp_path)
+                            except Exception:
+                                pass
+                        return False
+                except Exception:
+                    # 清理临时文件
+                    if temp_path and os.path.exists(temp_path):
+                        try:
+                            os.remove(temp_path)
+                        except Exception:
+                            pass
+                    raise
+
+        if not temp_path or not os.path.exists(temp_path):
+            logger.warning(f"[预缓存] 临时文件未生成: {item_id}")
+            return False
+
+        # 验证 ZIP 完整性
+        if not zipfile.is_zipfile(temp_path):
+            logger.warning(f"[预缓存] ZIP 文件验证失败（非有效 ZIP）: {item_id}")
+            os.remove(temp_path)
+            return False
+
+        zip_size = os.path.getsize(temp_path)
+        logger.info(f"[预缓存] ZIP 下载完成: {item_id}, 大小 {zip_size} 字节")
+
+        # 上传到 HF Dataset
+        path_in_repo = f"zips/{item_id}_{version_hash}.zip"
+        try:
+            upload_file(
+                path_or_fileobj=temp_path,
+                path_in_repo=path_in_repo,
+                repo_id=HF_REPO_ID,
+                repo_type="dataset",
+                token=hf_token,
+            )
+            logger.info(f"[预缓存] 上传成功: {path_in_repo}")
+        except Exception as e:
+            logger.warning(f"[预缓存] 上传 HF 失败 ({item_id}): {e}")
+            os.remove(temp_path)
+            return False
+
+        # 更新 versions.json 缓存元信息
+        versions_db = db.load_data("versions.json", default_data={})
+        entry = versions_db.get(item_id, {})
+        if isinstance(entry, str):
+            entry = {"hash": entry}
+        entry["cached_at"] = datetime.datetime.utcnow().isoformat() + "Z"
+        entry["zip_size"] = zip_size
+        versions_db[item_id] = entry
+        db.save_data("versions.json", versions_db)
+        logger.info(f"[预缓存] 缓存元信息已更新: {item_id}")
+
+        # 清理临时文件
+        os.remove(temp_path)
+        return True
+
+    except Exception as e:
+        logger.warning(f"[预缓存] 预缓存异常 ({item_id}): {e}")
+        return False
+
+
+async def _delete_old_zip_cache(item_id: str, old_hash: str):
+    """删除指定 item_id + old_hash 对应的 HF Dataset 缓存 ZIP"""
+    try:
+        hf_token = os.environ.get("HF_TOKEN")
+        if not hf_token:
+            return
+        hf_api = HfApi(token=hf_token)
+        path_in_repo = f"zips/{item_id}_{old_hash}.zip"
+        try:
+            hf_api.delete_file(
+                path_in_repo=path_in_repo,
+                repo_id=HF_REPO_ID,
+                repo_type="dataset",
+            )
+            logger.info(f"[预缓存] 已删除旧版本缓存: {path_in_repo}")
+        except Exception as e:
+            error_msg = str(e)
+            if "404" in error_msg or "Not Found" in error_msg:
+                logger.info(f"[预缓存] 旧缓存文件已不存在: {path_in_repo}")
+            else:
+                logger.warning(f"[预缓存] 删除旧缓存失败: {path_in_repo}, {e}")
+    except Exception as e:
+        logger.warning(f"[预缓存] 删除旧缓存异常 ({item_id}): {e}")
+
+
+async def cleanup_expired_cache(versions_db: dict):
+    """清理过期的 ZIP 缓存文件"""
+    try:
+        hf_token = os.environ.get("HF_TOKEN")
+        if not hf_token:
+            logger.warning("[预缓存] HF_TOKEN 未设置，跳过缓存清理")
+            return
+
+        hf_api = HfApi(token=hf_token)
+        now = datetime.datetime.utcnow()
+        expired_items = []
+
+        for item_id, val in list(versions_db.items()):
+            if isinstance(val, str):
+                continue
+            if not isinstance(val, dict):
+                continue
+
+            old_hash = val.get("hash")
+            cached_at_str = val.get("cached_at")
+
+            # 规则1: cached_at 超过 30 天的条目
+            if cached_at_str:
+                try:
+                    cached_at = datetime.datetime.fromisoformat(cached_at_str.replace("Z", "+00:00"))
+                    if (now - cached_at).days > CACHE_EXPIRE_DAYS:
+                        expired_items.append((item_id, old_hash, "过期"))
+                        continue
+                except ValueError:
+                    pass
+
+            # 规则2: 版本更新时（hash 变化）——在 daily_version_check_task 中单独处理
+            # 本函数仅处理过期时间
+
+        if not expired_items:
+            return
+
+        deleted_count = 0
+        modified_local = False
+        for item_id, old_hash, reason in expired_items:
+            if old_hash:
+                path_in_repo = f"zips/{item_id}_{old_hash}.zip"
+                try:
+                    hf_api.delete_file(
+                        path_in_repo=path_in_repo,
+                        repo_id=HF_REPO_ID,
+                        repo_type="dataset",
+                    )
+                    logger.info(f"[预缓存] 已删除过期缓存 ({reason}): {path_in_repo}")
+                    deleted_count += 1
+                except Exception as e:
+                    error_msg = str(e)
+                    if "404" in error_msg or "Not Found" in error_msg:
+                        logger.info(f"[预缓存] 缓存文件已不存在: {path_in_repo}")
+                    else:
+                        logger.warning(f"[预缓存] 删除缓存失败: {path_in_repo}, {e}")
+
+            # 清除本地 cached_at 和 zip_size（保留 hash）
+            entry = versions_db.get(item_id, {})
+            if isinstance(entry, dict):
+                had_meta = "cached_at" in entry or "zip_size" in entry
+                entry.pop("cached_at", None)
+                entry.pop("zip_size", None)
+                versions_db[item_id] = entry
+                if had_meta:
+                    modified_local = True
+
+        if deleted_count > 0 or modified_local:
+            db.save_data("versions.json", versions_db)
+            logger.info(f"[预缓存] 过期缓存清理完成，共删除 {deleted_count} 个文件")
+
+    except Exception as e:
+        logger.warning(f"[预缓存] 缓存清理异常: {e}")
+
+
 async def daily_version_check_task():
     """每日 02:00 定时执行的守护进程"""
     while True:
@@ -168,6 +413,9 @@ async def daily_version_check_task():
         items_db = db.load_data("items.json", default_data=[])
         versions_db = db.load_data("versions.json", default_data={})
         
+        # 先清理过期缓存（cached_at 超过 30 天）
+        await cleanup_expired_cache(versions_db)
+        
         updated_count = 0
         updated_items = []  # 记录有更新的插件信息
         
@@ -176,14 +424,23 @@ async def daily_version_check_task():
             if "github.com" in link:
                 token = item.get("github_token")
                 latest_hash = await fetch_latest_github_hash(link, token)
-                if latest_hash and versions_db.get(item["id"]) != latest_hash:
-                    versions_db[item["id"]] = latest_hash
+                old_hash = get_version_hash(versions_db, item["id"])
+                if latest_hash and old_hash != latest_hash:
+                    # 版本更新时，删除旧 hash 对应的缓存文件
+                    if old_hash:
+                        await _delete_old_zip_cache(item["id"], old_hash)
+                    
+                    # 统一转换为新格式并写入新 hash
+                    _normalize_version_entry(versions_db, item["id"])
+                    versions_db[item["id"]]["hash"] = latest_hash
                     updated_count += 1
-                    # 记录更新的插件信息
+                    # 记录更新的插件信息（包含 link/token 供预缓存使用）
                     updated_items.append({
                         "id": item["id"],
                         "title": item.get("title", "未知插件"),
-                        "version_hash": latest_hash
+                        "version_hash": latest_hash,
+                        "link": link,
+                        "github_token": token
                     })
         
         if updated_count > 0:
@@ -193,6 +450,26 @@ async def daily_version_check_task():
             await trigger_update_notifications(updated_items)
         else:
             print("✅ 版本检测任务完成，暂无任何新版本发现。")
+        
+        # 对更新的 items 执行预缓存（最多 5 个，每个间隔 10 秒）
+        if updated_items:
+            precache_count = 0
+            batch = updated_items[:PRECACHE_BATCH_SIZE]
+            for idx, item in enumerate(batch):
+                success = await precache_github_zip(
+                    item["link"],
+                    item.get("github_token"),
+                    item["id"],
+                    item["version_hash"]
+                )
+                if success:
+                    precache_count += 1
+                # 间隔 10 秒（最后一个不需要等待）
+                if idx < len(batch) - 1:
+                    await asyncio.sleep(PRECACHE_INTERVAL)
+            logger.info(
+                f"[预缓存] 本次预缓存完成: {precache_count}/{len(batch)} 成功"
+            )
 
 
 async def reset_daily_views_task():