Upload 8 files

app.py

@@ -222,6 +222,10 @@ async def on_startup():
     asyncio.create_task(daily_version_check_task())
     logger.info("✅ 定时版本检测任务已挂载")
     
+    # ========== 启动 HF 批量同步定时器 ==========
+    db.start_batch_sync()
+    logger.info("✅ HF 批量同步定时器已启动")
+    
     logger.info("🎉 ComfyUI-Ranking API 启动完成!")
 
 
@@ -229,6 +233,11 @@ async def on_startup():
 async def on_shutdown():
     """优雅关闭，清理资源"""
     logger.info("🛑 ComfyUI-Ranking API 正在关闭...")
+
+    # ========== 关闭前同步所有脏文件 ==========
+    db.flush_sync()
+    logger.info("✅ HF 批量同步已完成")
+
     # 这里可以添加其他清理逻辑（如关闭连接池等）
     logger.info("✅ 关闭完成")
 

notifications.py

@@ -15,7 +15,7 @@ def add_notification(target_account: str, notif_data: dict):
         "type": notif_data.get("type"), 
         "from_user": from_user,
         "from_name": user_info.get("name", from_user),
-        "from_avatar": user_info.get("avatarDataUrl", "https://via.placeholder.com/150"),
+        "from_avatar": user_info.get("avatarDataUrl", "")
         "target_item_id": notif_data.get("target_item_id", ""),
         "target_item_title": notif_data.get("target_item_title", ""),
         "content": notif_data.get("content", ""),

router_comments.py

@@ -42,7 +42,7 @@ async def post_comment(comment: CommentCreate):
     item_comments = comments_db.get(comment.item_id, [])
     new_comment = {
         "id": f"c_{int(time.time())}_{uuid.uuid4().hex[:6]}", "author": comment.author, 
-        "authorName": author_info.get("name", comment.author), "avatar": author_info.get("avatarDataUrl", "https://via.placeholder.com/150"),
+        "authorName": author_info.get("name", comment.author), "avatar": author_info.get("avatarDataUrl", ""),
         "content": comment.content, "replyToUser": comment.reply_to_user, "replyToUserName": reply_name,
         "isDeleted": False, "replies": [], "created_at": int(time.time())
     }

router_items.py

@@ -121,7 +121,7 @@ async def get_creators(sort: str = "downloads", limit: int = 20):
                 for m in months: trend_recommends[m] += history.get(m, 0)
 
         creators.append({
-            "account": account, "name": u.get("name", account), "avatar": u.get("avatarDataUrl", "https://via.placeholder.com/150"),
+            "account": account, "name": u.get("name", account), "avatar": u.get("avatarDataUrl", ""),
             "bannerUrl": u.get("bannerUrl"),  # 🖼️ 个人资料卡背景图
             "shortDesc": u.get("intro") or "这个人很懒，什么都没写...", "fullDesc": u.get("intro") or "这个人很懒，什么都没写...",
             "likes": sum(i.get("likes", 0) for i in u_items), "favorites": sum(i.get("favorites", 0) for i in u_items), 

router_messages.py

@@ -35,7 +35,7 @@ async def publish_announcement(ann: SystemAnnouncement, current_user: str = Depe
         "type": "system",
         "from_user": current_user,  # 使用真实的管理员账号
         "from_name": "官方团队",
-        "from_avatar": "https://via.placeholder.com/150/FF9800/FFFFFF?text=Sys",
+        "from_avatar": ""
         "content": ann.content,
         "created_at": int(time.time())
     }
@@ -163,7 +163,7 @@ async def get_chat_list(account: str):
             chat_list.append({
                 "account": other_account,
                 "name": other_user.get("name", other_account),
-                "avatar": other_user.get("avatarDataUrl", "https://via.placeholder.com/150"),
+                "avatar": other_user.get("avatarDataUrl", "")
                 "last_message": last_msg["content"],
                 "last_time": last_msg["created_at"],
                 "unread_count": unread_count

router_users_auth.py

@@ -17,7 +17,7 @@ import random
 import json
 import 数据库连接 as db
 from models import UserRegister, UserLogin, SendCodeRequest
-from verify_code_engine import VERIFY_CODES, send_email_code, send_sms_code, cleanup_expired_codes
+from verify_code_engine import VERIFY_CODES, send_email_code, send_sms_code, cleanup_expired_codes, check_send_cooldown
 
 # 🔒 P0安全增强：导入密码哈希和 JWT 工具
 from 安全认证 import hash_password, verify_password, create_token, require_password_match
@@ -65,6 +65,9 @@ async def send_verify_code(request: Request, req: SendCodeRequest, bg_tasks: Bac
         if req.contact_type == "phone" and user.get("phone") != req.contact:
             raise HTTPException(status_code=400, detail="填写的手机号与该账号绑定的手机号不一致")
 
+    # 🔒 检查发送频率限制（同一联系方式60秒内只能发送1条）
+    check_send_cooldown(req.contact)
+    
     # 生成6位随机验证码
     code = str(random.randint(100000, 999999))
     
@@ -74,10 +77,10 @@ async def send_verify_code(request: Request, req: SendCodeRequest, bg_tasks: Bac
     # 构建缓存键（联系方式_动作类型）
     cache_key = f"{req.contact}_{req.action_type}"
     
-    # 将验证码存入内存缓存，有效期10分钟
+    # 将验证码存入内存缓存，有效期5分钟
     VERIFY_CODES[cache_key] = {
         "code": code,
-        "expires_at": int(time.time()) + 600  # 当前时间 + 600秒
+        "expires_at": int(time.time()) + 300  # 当前时间 + 300秒(5分钟)
     }
     
     # 根据联系方式类型，添加后台发送任务
@@ -99,14 +102,17 @@ async def send_verify_code(request: Request, req: SendCodeRequest, bg_tasks: Bac
 @router.post("/api/users/send_code")
 async def send_code_api(req: SendCodeRequest):
     """发送验证码接口（同步版本，直接等待发送结果）"""
+    # 🔒 检查发送频率限制（同一联系方式60秒内只能发送1条）
+    check_send_cooldown(req.contact)
+    
     # 生成6位随机验证码
     code = str(random.randint(100000, 999999))
     key = f"{req.contact}_{req.action_type}"
     
-    # 存入缓存
+    # 存入缓存，有效期5分钟
     VERIFY_CODES[key] = {
         "code": code,
-        "expires_at": time.time() + 600
+        "expires_at": time.time() + 300  # 300秒(5分钟)
     }
     
     # 同步发送（会阻塞等待结果）
@@ -161,10 +167,11 @@ async def register_user(request: Request, user: UserRegister):
         if user.phone and existing_user.get("phone") == user.phone: 
             raise HTTPException(status_code=400, detail="该手机号已被绑定")
             
-    # ========== 第二步：验证码校验 ==========
+    # ========== 第二步：验证码校验（原子性获取+删除） ==========
     # 根据注册方式构建缓存键
     cache_key = f"{user.email}_register" if user.email else f"{user.phone}_register"
-    cached = VERIFY_CODES.get(cache_key)
+    # 🔒 P0安全修复：验证码一次性使用，原子性pop防止并发重用
+    cached = VERIFY_CODES.pop(cache_key, None)
     
     # 兼容新老缓存格式（expires_at 或 expires）
     expire_time = cached.get("expires_at", cached.get("expires", 0)) if cached else 0
@@ -186,9 +193,6 @@ async def register_user(request: Request, user: UserRegister):
         raise HTTPException(status_code=400, detail="个人介绍不能超过100个字符")
     
     # ========== 第四步：保存新用户 ==========
-    # 验证通过后，清除已使用的验证码
-    VERIFY_CODES.pop(cache_key, None) 
-    
     # 构建用户数据对象
     new_user = user.dict()
     new_user.pop("code", None)  # 移除验证码字段，不存入数据库
@@ -241,15 +245,23 @@ async def login_user(request: Request, user: UserLogin):
     if not require_password_match(stored_password, user.password):
         raise HTTPException(status_code=401, detail="密码错误")
     
+    # 🔒 P0安全增强：登录成功后，检查是否需要迁移旧密码为bcrypt
+    if not user_data["password"].startswith('$2b$') and not user_data["password"].startswith('$2a$'):
+        # 旧版SHA256密码，自动迁移为bcrypt
+        user_data["password"] = hash_password(user.password)
+        db.save_data("users.json", users_db)
+    
     # 🔒 P0安全增强：生成 JWT Token（替代 mock_token）
-    token = create_token(user.account)
+    # 获取password_version用于Token生成（如不存在则默认为0）
+    password_version = user_data.get("password_version", 0)
+    token = create_token(user.account, extra_data={"pwd_ver": password_version})
     
     return {
         "status": "success", 
         "token": token,  # 🔒 JWT Token
         "account": user.account, 
         "name": user_data["name"], 
-        "avatar": user_data.get("avatarDataUrl", "https://via.placeholder.com/150")
+        "avatar": user_data.get("avatarDataUrl", "")
     }
 
 
@@ -317,9 +329,10 @@ async def reset_password(request: Request):
     if verify_type == "phone" and user.get("phone") != verify_contact:
         raise HTTPException(status_code=400, detail="填写的手机号与该账号绑定的手机号不匹配")
 
-    # 校验验证码
+    # 校验验证码（原子性获取+删除）
     cache_key = f"{verify_contact}_reset"
-    cached = VERIFY_CODES.get(cache_key)
+    # 🔒 P0安全修复：验证码一次性使用，原子性pop防止并发重用
+    cached = VERIFY_CODES.pop(cache_key, None)
     expire_time = cached.get("expires_at", cached.get("expires", 0)) if cached else 0
     
     if not cached or cached["code"] != code or time.time() > expire_time:
@@ -332,11 +345,16 @@ async def reset_password(request: Request):
         raise HTTPException(status_code=400, detail="新密码包含不支持的特殊字符")
     
     # ========== 第四步：更新密码并保存 ==========
-    VERIFY_CODES.pop(cache_key, None)  # 清除已使用的验证码
-    
     # 🔒 P0安全增强：新密码哈希化存储
     user["password"] = hash_password(new_password)
     
+    # 🔒 P0安全增强：更新password_version使旧Token失效
+    import time as time_module
+    user["password_version"] = int(time_module.time())
+    
     db.save_data("users.json", users_db)
     
-    return {"status": "success", "message": "密码修改成功"}
+    # 🔒 P0安全增强：生成新Token返回给前端替换旧Token
+    new_token = create_token(account, extra_data={"pwd_ver": user["password_version"]})
+    
+    return {"status": "success", "message": "密码修改成功", "token": new_token}

数据库连接.py

@@ -25,8 +25,7 @@ import shutil
 import tempfile
 import logging
 from typing import Any, Dict, List, Optional, Union
-from concurrent.futures import ThreadPoolExecutor
-from huggingface_hub import HfApi, hf_hub_download
+from huggingface_hub import HfApi, CommitOperationAdd, hf_hub_download
 
 # 📝 日志配置
 logger = logging.getLogger("ComfyUI-Ranking.DB")
@@ -51,8 +50,11 @@ BACKUP_DIR = os.path.join(LOCAL_DB_DIR, "_backups")
 # HuggingFace API 客户端
 api = HfApi() if HF_TOKEN else None
 
-# 🔧 P3优化：线程池管理上传任务（限制并发数）
-_upload_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="hf_upload")
+# ===== HF 批量同步机制 =====
+_dirty_files = set()                  # 记录自上次同步以来有变更的文件名
+_dirty_files_lock = threading.Lock()
+_BATCH_SYNC_INTERVAL = 300            # 批量同步间隔（秒），默认5分钟
+_batch_sync_timer = None              # 定时器引用
 
 # 确保目录存在
 os.makedirs(LOCAL_DB_DIR, exist_ok=True)
@@ -403,63 +405,89 @@ def save_data(file_name: str, data: Union[Dict, List]) -> bool:
     # ========== 🚀 P1优化：更新内存缓存 ==========
     _set_to_cache(file_name, data, local_path)
     
-    # ========== 第五步：异步同步到云端 ==========
-    # 🔧 P3优化：使用线程池替代直接创建线程
+    # ========== 第五步：标记文件脏，等待批量同步 ==========
     if HF_TOKEN:
-        try:
-            _upload_executor.submit(_background_upload_to_hf, local_path, file_name)
-        except Exception as e:
-            logger.warning(f"提交上传任务失败: {e}")
+        _mark_dirty(file_name)
 
 
 # ==========================================
-# ☁️ 后台上传到 HuggingFace Dataset
+# ☁️ HF 批量同步机制
 # ==========================================
-# 特点：
-#   - 后台线程执行，不阻塞主流程
-#   - 失败自动重试（最多3次）
-#   - 指数退避策略
 
-def _background_upload_to_hf(local_path: str, file_name: str, retries: int = 3):
-    """
-    后台上传文件到 HuggingFace Dataset
-    
-    参数：
-        local_path: 本地文件路径
-        file_name: 远程文件名
-        retries: 最大重试次数
-    """
-    for attempt in range(retries):
-        try:
-            api.upload_file(
-                path_or_fileobj=local_path,
-                path_in_repo=file_name,
+def _mark_dirty(file_name: str):
+    """标记文件为脏，等待下次批量同步"""
+    with _dirty_files_lock:
+        _dirty_files.add(file_name)
+
+
+def _batch_sync_to_hf(schedule_next=True):
+    """批量同步所有脏文件到 HF Dataset（单次 commit）"""
+    # 取出脏文件列表并清空
+    with _dirty_files_lock:
+        files_to_sync = list(_dirty_files)
+        _dirty_files.clear()
+
+    if not files_to_sync:
+        if schedule_next:
+            _schedule_next_sync()
+        return
+
+    try:
+        operations = []
+        for file_name in files_to_sync:
+            local_path = os.path.join(LOCAL_DB_DIR, file_name)
+            if os.path.exists(local_path):
+                operations.append(
+                    CommitOperationAdd(
+                        path_in_repo=file_name,
+                        path_or_fileobj=local_path
+                    )
+                )
+
+        if operations:
+            hf_api = HfApi(token=HF_TOKEN)
+            hf_api.create_commit(
                 repo_id=DATASET_REPO_ID,
                 repo_type="dataset",
-                token=HF_TOKEN,
-                commit_message=f"Auto-update {file_name}"
+                operations=operations,
+                commit_message=f"batch sync: {', '.join(files_to_sync)}"
             )
-            return  # 成功后退出
-            
-        except Exception as e:
-            wait_time = 2 ** attempt  # 指数退避：1s, 2s, 4s
-            if attempt < retries - 1:
-                print(f"⚠️ 上传 {file_name} 失败 (第{attempt+1}次)，{wait_time}秒后重试: {e}")
-                time.sleep(wait_time)
-            else:
-                print(f"🚨 致命错误：重试 {retries} 次后，同步到 HF Dataset 依然失败: {e}")
-                # 最后一次失败，保存到失败队列（可选：后续恢复机制）
-                _save_to_failed_queue(file_name)
+            logger.info(f"✅ 批量同步成功: {len(operations)} 个文件 ({', '.join(files_to_sync)})")
+        else:
+            logger.info("⏭️ 批量同步: 脏文件均不存在，跳过")
+    except Exception as e:
+        logger.error(f"🚨 批量同步失败: {e}")
+        # 失败的文件重新标记为脏，下次重试
+        with _dirty_files_lock:
+            _dirty_files.update(files_to_sync)
+    finally:
+        if schedule_next:
+            _schedule_next_sync()
 
 
-def _save_to_failed_queue(file_name: str):
-    """记录上传失败的文件，供后续重试"""
-    failed_queue_path = os.path.join(BACKUP_DIR, "_upload_failed.txt")
-    try:
-        with open(failed_queue_path, "a", encoding="utf-8") as f:
-            f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} | {file_name}\n")
-    except Exception as e:
-        print(f"⚠️ 记录上传失败队列异常: {e}")
+def _schedule_next_sync():
+    """调度下一轮批量同步"""
+    global _batch_sync_timer
+    _batch_sync_timer = threading.Timer(_BATCH_SYNC_INTERVAL, _batch_sync_to_hf)
+    _batch_sync_timer.daemon = True
+    _batch_sync_timer.start()
+
+
+def start_batch_sync():
+    """启动批量同步定时器（在 app 启动时调用）"""
+    if HF_TOKEN:
+        _schedule_next_sync()
+        logger.info(f"📡 HF 批量同步已启动，间隔 {_BATCH_SYNC_INTERVAL} 秒")
+
+
+def flush_sync():
+    """立即同步所有脏文件（用于服务关闭前）"""
+    global _batch_sync_timer
+    if _batch_sync_timer:
+        _batch_sync_timer.cancel()
+        _batch_sync_timer = None
+    logger.info("🔄 正在执行关闭前同步...")
+    _batch_sync_to_hf(schedule_next=False)
 
 
 # ==========================================
@@ -658,11 +686,8 @@ def atomic_update(file_name: str, updater, default_data=None):
     # ========== 第四步：更新内存缓存 ==========
     _set_to_cache(file_name, data, local_path)
     
-    # ========== 第五步：异步同步到云端 ==========
+    # ========== 第五步：标记文件脏，等待批量同步 ==========
     if HF_TOKEN:
-        try:
-            _upload_executor.submit(_background_upload_to_hf, local_path, file_name)
-        except Exception as e:
-            logger.warning(f"提交上传任务失败: {e}")
+        _mark_dirty(file_name)
     
     return result