Upload 91 files

app.py

@@ -20,6 +20,11 @@ import threading
 import 数据库连接 as db
 import asyncio
 from 安全认证 import require_auth
+from profile_io import start_profile_io, close_profile_io
+from script_execution import start_admin_scripts, stop_admin_scripts, close_admin_scripts
+from comment_store import start_comment_reads, stop_comment_reads, close_comment_reads
+from periodic_tasks import start_periodic_task, request_periodic_stop, wait_periodic_tasks
+from owned_thread import run_owned_thread
 
 
 # ==========================================
@@ -228,61 +233,78 @@ def detailed_health_check():
 @app.on_event("startup")
 async def on_startup():
     logger.info("🚀 ComfyUI-Ranking API 启动中...")
-    
-    # ========== 预热检查：SQL 数据库 ==========
+    if getattr(app.state, "_ranking_starting", False) or getattr(app.state, "_ranking_closers", 0):
+        raise RuntimeError("application lifecycle is already in progress")
+    if getattr(app.state, "_ranking_started", False):
+        return
+    app.state._ranking_starting = True
     try:
-        init_sql_db()
-        logger.info("✅ SQL 数据库初始化完成")
-    except Exception as e:
-        logger.error(f"❌ SQL 数据库初始化失败: error_type={type(e).__name__}")
-        # 不抛出异常，允许降级运行
+        start_profile_io()
+        try:
+            start_comment_reads()
+            start_admin_scripts()
 
-    # ========== 加载第三方插件注册表（依赖 SQL 数据库） ==========
-    try:
-        from router_open_api import load_plugins_from_db
-        load_plugins_from_db()
-        logger.info("✅ 第三方插件注册表已加载")
-    except Exception as e:
-        logger.warning(f"⚠️ 第三方插件注册表加载失败: error_type={type(e).__name__}")
-    
-    # ========== 预热检查：JSON 数据库 ==========
-    try:
-        db.load_data("users.json", default_data={})
-        db.load_data("items.json", default_data=[])
-        logger.info("✅ JSON 数据库访问正常")
-    except Exception as e:
-        logger.warning(f"⚠️ JSON 数据库访问异常: error_type={type(e).__name__}")
-    
-    # SQL资金已冻结但JSON发布失败的任务，重启后继续投影；持有句柄供停机收尾。
-    from task_publication import publication_worker
-    from database_sql import engine
-    from db_utils import sort_cache
-    app.state.task_publication_stop = asyncio.Event()
-    app.state.task_publication_worker = asyncio.create_task(publication_worker(
-        engine, db, sort_cache.invalidate, app.state.task_publication_stop))
-
-    from resource_publication import publication_worker as resource_publication_worker
-    app.state.resource_publication_stop = asyncio.Event()
-    app.state.resource_publication_worker = asyncio.create_task(resource_publication_worker(
-        engine, db, sort_cache.invalidate, app.state.resource_publication_stop))
-
-    from resource_deletion import resource_deletion_worker
-    app.state.resource_deletion_stop = asyncio.Event()
-    app.state.resource_deletion_worker = asyncio.create_task(resource_deletion_worker(
-        db, app.state.resource_deletion_stop, sort_cache.invalidate))
-
-    # ========== 启动后台任务 ==========
-    asyncio.create_task(daily_version_check_task())
-    logger.info("✅ 定时版本检测任务已挂载")
-    
-    asyncio.create_task(reset_daily_views_task())
-    logger.info("✅ daily_views 每日重置任务已挂载")
-    
-    # ========== 启动 HF 批量同步定时器 ==========
-    db.start_batch_sync()
-    logger.info("✅ HF 同步调度初始化已完成")
-    
-    logger.info("🎉 ComfyUI-Ranking API 启动完成!")
+            # ========== 预热检查：SQL 数据库 ==========
+            try:
+                init_sql_db()
+                logger.info("✅ SQL 数据库初始化完成")
+            except Exception as e:
+                logger.error(f"❌ SQL 数据库初始化失败: error_type={type(e).__name__}")
+                # 不抛出异常，允许降级运行
+
+            # ========== 加载第三方插件注册表（依赖 SQL 数据库） ==========
+            try:
+                from router_open_api import load_plugins_from_db
+                load_plugins_from_db()
+                logger.info("✅ 第三方插件注册表已加载")
+            except Exception as e:
+                logger.warning(f"⚠️ 第三方插件注册表加载失败: error_type={type(e).__name__}")
+
+            # ========== 预热检查：JSON 数据库 ==========
+            try:
+                db.load_data("users.json", default_data={})
+                db.load_data("items.json", default_data=[])
+                logger.info("✅ JSON 数据库访问正常")
+            except Exception as e:
+                logger.warning(f"⚠️ JSON 数据库访问异常: error_type={type(e).__name__}")
+
+            # SQL资金已冻结但JSON发布失败的任务，重启后继续投影；持有句柄供停机收尾。
+            from task_publication import publication_worker
+            from database_sql import engine
+            from db_utils import sort_cache
+            start_periodic_task(app.state, "task_publication",
+                lambda stop: publication_worker(engine, db, sort_cache.invalidate, stop))
+
+            from resource_publication import publication_worker as resource_publication_worker
+            start_periodic_task(app.state, "resource_publication",
+                lambda stop: resource_publication_worker(engine, db, sort_cache.invalidate, stop))
+
+            from resource_deletion import resource_deletion_worker
+            start_periodic_task(app.state, "resource_deletion",
+                lambda stop: resource_deletion_worker(db, stop, sort_cache.invalidate))
+
+            # ========== 启动后台任务 ==========
+            start_periodic_task(app.state, "daily_version", daily_version_check_task)
+            logger.info("✅ 定时版本检测任务已挂载")
+
+            start_periodic_task(app.state, "daily_views", reset_daily_views_task)
+            logger.info("✅ daily_views 每日重置任务已挂载")
+
+            # ========== 启动 HF 批量同步定时器 ==========
+            db.start_batch_sync()
+            logger.info("✅ HF 同步调度初始化已完成")
+
+            app.state._ranking_started = True
+            logger.info("🎉 ComfyUI-Ranking API 启动完成!")
+        except BaseException:
+            # 启动可能只登记了部分句柄；正常关闭按已登记对象收尾。
+            try:
+                await on_shutdown()
+            except BaseException as error:
+                logger.error("启动收尾失败: error_type=%s", type(error).__name__)
+            raise
+    finally:
+        app.state._ranking_starting = False
 
 
 @app.on_event("shutdown")
@@ -290,33 +312,43 @@ async def on_shutdown():
     """优雅关闭，清理资源"""
     logger.info("🛑 ComfyUI-Ranking API 正在关闭...")
 
-    publication_stop = getattr(app.state, "task_publication_stop", None)
-    publication_task = getattr(app.state, "task_publication_worker", None)
-    if publication_stop is not None:
-        publication_stop.set()
-    if publication_task is not None:
-        await publication_task
-
-    resource_stop = getattr(app.state, "resource_publication_stop", None)
-    resource_task = getattr(app.state, "resource_publication_worker", None)
-    if resource_stop is not None:
-        resource_stop.set()
-    if resource_task is not None:
-        await resource_task
-
-    deletion_stop = getattr(app.state, "resource_deletion_stop", None)
-    deletion_task = getattr(app.state, "resource_deletion_worker", None)
-    if deletion_stop is not None:
-        deletion_stop.set()
-    if deletion_task is not None:
-        await deletion_task
-
-    # ========== 关闭前同步所有脏文件 ==========
-    await asyncio.to_thread(db.flush_sync)
-    logger.info("✅ HF 关闭前同步尝试已结束")
-
-    # 这里可以添加其他清理逻辑（如关闭连接池等）
-    logger.info("✅ 关闭完成")
+    app.state._ranking_closers = getattr(app.state, "_ranking_closers", 0) + 1
+    try:
+        app.state._ranking_started = False
+        stop_admin_scripts()
+        stop_comment_reads()
+        workers = request_periodic_stop(app.state, (
+            "daily_version", "daily_views", "task_publication", "resource_publication", "resource_deletion"))
+
+        # 关闭等待被取消也先收尾全部所属工作，之后传播取消；不提前最终同步。
+        cancelled = None
+        try:
+            await close_profile_io()
+        except asyncio.CancelledError as error:
+            cancelled = error
+        try:
+            await close_admin_scripts()
+        except asyncio.CancelledError as error:
+            cancelled = error
+        try:
+            await close_comment_reads()
+        except asyncio.CancelledError as error:
+            cancelled = error
+        try:
+            await wait_periodic_tasks(workers)
+        except asyncio.CancelledError as error:
+            cancelled = error
+        if cancelled is not None:
+            raise cancelled
+
+        # ========== 关闭前同步所有脏文件 ==========
+        await run_owned_thread(db.flush_sync)
+        logger.info("✅ HF 关闭前同步尝试已结束")
+
+        # 这里可以添加其他清理逻辑（如关闭连接池等）
+        logger.info("✅ 关闭完成")
+    finally:
+        app.state._ranking_closers -= 1
 
 # 🛡️ CORS 配置：默认通配符（插件前端运行于用户本地 ComfyUI 页面，origin 为
 # 127.0.0.1/localhost/局域网 IP/自定义端口等无法穷举；本 API 通过 Authorization 头
@@ -418,7 +450,7 @@ def _is_admin(account: str) -> bool:
 # 配置存储文件
 SYSTEM_CONFIG_FILE = "/tmp/system_config.json"  # 保留作为降级方案
 
-def _load_system_config_from_db(db: Session) -> dict:
+def _load_system_config_from_db(db: Session, *, strict: bool = False) -> dict:
     """从数据库加载所有系统配置"""
     try:
         configs = db.query(SystemConfig).all()
@@ -431,10 +463,18 @@ def _load_system_config_from_db(db: Session) -> dict:
         return result
     except Exception as e:
         logger.warning(f"从数据库加载系统配置失败，降级到文件: error_type={type(e).__name__}")
+        if strict:
+            return _load_system_config_from_file(strict=True)
         return _load_system_config_from_file()
 
-def _load_system_config_from_file() -> dict:
-    """降级方案：从 JSON 文件加载系统配置"""
+def _load_system_config_from_file(*, strict: bool = False) -> dict:
+    """降级方案；严格读取不把故障或缺失文件冒充空配置。"""
+    if strict:
+        with open(SYSTEM_CONFIG_FILE, "r", encoding="utf-8") as stream:
+            config = json.load(stream)
+        if not isinstance(config, dict):
+            raise ValueError("invalid system configuration root")
+        return config
     try:
         if os.path.exists(SYSTEM_CONFIG_FILE):
             with open(SYSTEM_CONFIG_FILE, "r") as f:
@@ -443,18 +483,25 @@ def _load_system_config_from_file() -> dict:
         logger.debug(f"加载系统配置文件失败: error_type={type(e).__name__}")
     return {}
 
-def _save_system_config_to_db(db: Session, config_key: str, value, updated_by: str = None):
+def _save_system_config_to_db(db: Session, config_key: str, value, updated_by: str = None, *, strict: bool = False):
     """保存单个配置项到数据库"""
     try:
         cfg = db.query(SystemConfig).filter(SystemConfig.config_key == config_key).first()
         if not cfg:
             cfg = SystemConfig(config_key=config_key)
             db.add(cfg)
-        cfg.config_value = json.dumps(value, ensure_ascii=False)
+        cfg.config_value = json.dumps(value, ensure_ascii=False, allow_nan=not strict)
         cfg.updated_by = updated_by
         cfg.updated_at = datetime.datetime.utcnow()
         db.commit()
     except Exception as e:
+        if strict:
+            # commit异常可能发生在真实提交之后；回滚不能证明尚未保存。
+            try:
+                db.rollback()
+            except Exception:
+                logger.warning("SYSTEM_CONFIG_ROLLBACK_UNCONFIRMED")
+            raise
         db.rollback()
         logger.warning(f"保存系统配置到数据库失败，降级到文件: error_type={type(e).__name__}")
         # 降级：保存到文件
@@ -467,8 +514,16 @@ def _save_system_config_to_db(db: Session, config_key: str, value, updated_by: s
             logger.warning(f"降级保存系统配置文件也失败: error_type={type(fe).__name__}")
 
 # 兼容旧调用的包装函数
-def _load_system_config(db: Session = None) -> dict:
-    """加载系统配置（优先数据库，降级文件）"""
+def _load_system_config(db: Session = None, *, strict: bool = False) -> dict:
+    """加载系统配置（优先数据库，降级文件）；严格模式保留故障。"""
+    if strict:
+        if db is not None:
+            return _load_system_config_from_db(db, strict=True)
+        temp_db = SessionLocal()
+        try:
+            return _load_system_config_from_db(temp_db, strict=True)
+        finally:
+            temp_db.close()
     if db:
         return _load_system_config_from_db(db)
     # 无 db session 时（如公开端点），创建临时 session
@@ -504,8 +559,15 @@ async def get_system_config(config_key: str, current_user: str = Depends(require
     if not _is_admin(current_user):
         raise HTTPException(status_code=403, detail="无权访问系统配置，仅管理员可操作")
     
-    config = _load_system_config(db)
-    return {"status": "success", "data": config.get(config_key)}
+    def reject_constant(_value):
+        raise ValueError("invalid configuration constant")
+    try:
+        # 管理回读用于确认SQL保存结果，不能用另一个来源的旧值代替。
+        row = db.query(SystemConfig).filter(SystemConfig.config_key == config_key).first()
+        value = None if row is None else json.loads(row.config_value, parse_constant=reject_constant)
+    except Exception:
+        raise HTTPException(status_code=503, detail="config_read_unavailable") from None
+    return {"status": "success", "data": value}
 
 
 # 🏷️ 版本配置默认值
@@ -528,20 +590,29 @@ async def get_public_project_version(db: Session = Depends(get_db)):
 @app.get("/api/public/banner_config")
 async def get_public_banner_config(db: Session = Depends(get_db)):
     """公开接口：获取广告横幅配置（所有已登录/未登录用户均可访问）"""
-    config = _load_system_config(db)
-    banner = config.get("banner_config")
-    if not banner or not banner.get("enabled"):
-        return {"status": "success", "data": None}
-    # 只返回展示所需��字段，不暴露管理内部数据
-    return {"status": "success", "data": {
-        "enabled": banner.get("enabled", False),
-        "bannerImage": banner.get("bannerImage", ""),
-        "title": banner.get("title", ""),
-        "description": banner.get("description", ""),
-        "detailImage": banner.get("detailImage", ""),
-        "detailContent": banner.get("detailContent", ""),
-        "link": banner.get("link", "")
-    }}
+    try:
+        config = _load_system_config(db, strict=True)
+        if not isinstance(config, dict):
+            raise ValueError("invalid system configuration root")
+        banner = config.get("banner_config")
+        if banner is None:
+            return {"status": "success", "data": None}
+        if not isinstance(banner, dict) or type(banner.get("enabled")) is not bool:
+            raise ValueError("invalid banner configuration")
+        if not banner["enabled"]:
+            return {"status": "success", "data": None}
+        # 只投影展示字段，异常配置不能触发浏览器清除旧横幅。
+        result = {"enabled": True}
+        for field in ("bannerImage", "title", "description", "detailImage", "detailContent", "link"):
+            value = banner.get(field, "")
+            if not isinstance(value, str):
+                raise ValueError("invalid banner display field")
+            result[field] = value
+        if not result["bannerImage"].strip():
+            raise ValueError("missing banner image")
+        return {"status": "success", "data": result}
+    except Exception:
+        raise HTTPException(status_code=503, detail="banner_config_unavailable") from None
 
 
 @app.put("/api/admin/config/{config_key}")
@@ -550,7 +621,10 @@ async def set_system_config(config_key: str, value: dict, current_user: str = De
     # 🔒 权限验证
     if not _is_admin(current_user):
         raise HTTPException(status_code=403, detail="无权修改系统配置，仅管理员可操作")
-    _save_system_config_to_db(db, config_key, value, updated_by=current_user)
+    try:
+        _save_system_config_to_db(db, config_key, value, updated_by=current_user, strict=True)
+    except Exception:
+        raise HTTPException(status_code=503, detail="config_save_unconfirmed") from None
     return {"status": "success", "message": "配置已更新"}
 
 

comment_store.py

@@ -1,12 +1,59 @@
 """评论记录的锁内修改；路由负责认证、内容存在性及响应契约。"""
 import copy
+import asyncio
 import logging
+import threading
+from owned_thread import run_owned_thread
 from fastapi import HTTPException
 import 数据库连接 as db
 from resource_deletion import resource_mutation_lock, assert_comment_target_active
 from resource_store import ResourceStoreError
 
 
+COMMENT_MAX_LENGTH = 5000
+MAX_COMMENT_READS = 8
+_comment_read_capacity = threading.BoundedSemaphore(MAX_COMMENT_READS)
+_comment_read_gate = threading.Lock()
+_comment_read_accepting = True
+_comment_read_pending = 0
+_comment_read_closers = 0
+_comment_read_drained = threading.Event()
+_comment_read_drained.set()
+
+
+def start_comment_reads():
+    global _comment_read_accepting
+    with _comment_read_gate:
+        if _comment_read_pending or _comment_read_closers:
+            raise RuntimeError("comment reads are still pending")
+        _comment_read_accepting = True
+
+
+def stop_comment_reads():
+    global _comment_read_accepting
+    with _comment_read_gate:
+        _comment_read_accepting = False
+
+
+async def close_comment_reads():
+    global _comment_read_accepting, _comment_read_closers
+    with _comment_read_gate:
+        _comment_read_accepting = False
+        _comment_read_closers += 1
+    try:
+        cancelled = False
+        while not _comment_read_drained.is_set():
+            try:
+                await asyncio.sleep(.01)
+            except asyncio.CancelledError:
+                cancelled = True
+        if cancelled:
+            raise asyncio.CancelledError
+    finally:
+        with _comment_read_gate:
+            _comment_read_closers -= 1
+
+
 def _item_comments(data, item_id):
     if not isinstance(data, dict):
         raise HTTPException(status_code=503, detail="评论数据暂不可用")
@@ -100,6 +147,60 @@ def _visible_count(records, depth=0):
     return count
 
 
+def _validate_comment_text(records):
+    for record in records:
+        if not isinstance(record.get("author"), str) or not isinstance(record.get("content"), str):
+            raise ValueError("Invalid comment text")
+        _validate_comment_text(record.get("replies", []))
+
+
+def read_comment_thread(item_id):
+    """严格读取评论快照；存储故障不冒充成功空列表，不改历史主记录。"""
+    try:
+        snapshot = db.load_data("comments.json", default_data={}, skip_cache=True, strict=True)
+        records = _item_comments(snapshot, item_id)
+        _visible_count(records)
+        _validate_comment_text(records)
+        return copy.deepcopy(records)
+    except Exception:
+        logging.getLogger(__name__).warning("评论列表无法核实")
+        raise HTTPException(status_code=503, detail="评论暂时无法读取") from None
+
+
+def _read_comment_view(item_id):
+    records = read_comment_thread(item_id)
+    users = db.load_data("users.json", default_data={})
+    result = []
+    for record in records:
+        author_info = users.get(record["author"], {})
+        result.append({**record, "author_name": author_info.get("name", record["author"]),
+                       "author_avatar": author_info.get("avatarDataUrl", "")})
+    return result
+
+
+async def read_comment_view(item_id):
+    """整次评论读取在所属后台工作中执行；繁忙不提交线程池队列。"""
+    global _comment_read_pending
+    with _comment_read_gate:
+        if not _comment_read_accepting or not _comment_read_capacity.acquire(blocking=False):
+            raise HTTPException(status_code=503, detail="评论暂时无法读取")
+        _comment_read_pending += 1
+        _comment_read_drained.clear()
+    try:
+        return await run_owned_thread(_read_comment_view, item_id)
+    except HTTPException:
+        raise
+    except Exception:
+        logging.getLogger(__name__).warning("评论展示无法核实")
+        raise HTTPException(status_code=503, detail="评论暂时无法读取") from None
+    finally:
+        with _comment_read_gate:
+            _comment_read_capacity.release()
+            _comment_read_pending -= 1
+            if not _comment_read_pending:
+                _comment_read_drained.set()
+
+
 def attach_comment_counts(records):
     """只校对响应副本，以评论主记录恢复展示计数，不写内容或资金数据。"""
     try:

db_utils.py

@@ -12,6 +12,8 @@
 
 from typing import Any, Dict, List, Optional, Callable, Union
 import copy
+from collections import OrderedDict
+import threading
 import 数据库连接 as db
 
 
@@ -19,6 +21,36 @@ import 数据库连接 as db
 # 📖 查询工具函数
 # ==========================================
 
+def read_public_source(store, file_name, default_data):
+    """公开读取拒绝故障默认值及异常集合；保留已验证文件缓存。"""
+    try:
+        data = store.load_data(file_name, default_data=default_data, strict=True)
+        if type(data) is not type(default_data):
+            raise ValueError("invalid public source root")
+        if isinstance(data, list):
+            if any(not isinstance(row, dict) or not row for row in data):
+                raise ValueError("invalid public source records")
+        elif file_name == "users.json":
+            if any(not isinstance(key, str) or not key or not isinstance(row, dict)
+                   for key, row in data.items()):
+                raise ValueError("invalid public user records")
+        elif file_name == "comments.json":
+            if any(not isinstance(rows, list) or
+                   any(not isinstance(row, dict) or not row for row in rows)
+                   for rows in data.values()):
+                raise ValueError("invalid public comment records")
+        elif file_name == "versions.json":
+            if any(row is not None and not isinstance(row, (str, dict)) or
+                   isinstance(row, dict) and row.get("hash") is not None and
+                   not isinstance(row.get("hash"), str)
+                   for row in data.values()):
+                raise ValueError("invalid public version records")
+        return data
+    except Exception:
+        from fastapi import HTTPException
+        raise HTTPException(status_code=503, detail="ranking_data_unavailable") from None
+
+
 def get_by_id(file_name: str, item_id: str, id_field: str = "id") -> Optional[Dict]:
     """
     根据 ID 获取单个记录
@@ -478,17 +510,38 @@ class SortCache:
             sort_cache.invalidate("items")
     """
     
-    def __init__(self, ttl: int = 300):
+    def __init__(self, ttl: int = 300, max_entries: int = 128, max_ids: int = 100000):
         """
         初始化排序缓存
         
         参数：
             ttl: 缓存过期时间（秒），默认 300 秒（5分钟）
+            max_entries: 最多保留的缓存条目，0 表示不保留
+            max_ids: 所有缓存合计保留的 ID 数量，0 表示不保留
         """
-        self._cache: Dict[str, tuple] = {}  # {cache_key: (sorted_ids, timestamp)}
+        if any(type(value) is not int or value < 0 for value in (max_entries, max_ids)):
+            raise ValueError("sort cache limits must be nonnegative integers")
+        self._cache = OrderedDict()  # {cache_key: (sorted_ids, timestamp)}
+        self._max_entries = max_entries
+        self._max_ids = max_ids
+        self._cached_ids = 0
         self._ttl = ttl
-    
-    def get_sorted(self, cache_key: str, items: List[Dict], sort_fn: Callable[[List[Dict]], None]) -> List[Dict]:
+        self._lock = threading.Lock()
+        self._pending_sorts: Dict[str, object] = {}
+        self._read_epoch = object()
+    
+    def capture_read_epoch(self):
+        """在读取排序源数据前捕获；只标识本进程失效代次，不是文件版本。"""
+        with self._lock:
+            return self._read_epoch
+
+    def _remove_cached(self, key: str):
+        # 调用者持有 _lock；失效、过期和容量淘汰共用计数维护。
+        cached = self._cache.pop(key, None)
+        if cached is not None:
+            self._cached_ids -= len(cached[0])
+
+    def get_sorted(self, cache_key: str, items: List[Dict], sort_fn: Callable[[List[Dict]], None], *, read_epoch=None, fresh: bool = False) -> List[Dict]:
         """
         获取排序后的数据（带缓存）
         
@@ -496,30 +549,65 @@ class SortCache:
             cache_key: 缓存键，应包含数据文件和排序参数
             items: 原始数据列表（来自 load_data 的最新数据）
             sort_fn: 排序函数，接收 items 列表并原地排序
+            read_epoch: 读取前捕获的代次；已失效输入只排序返回，不读写缓存
+            fresh: 按本次输入直接排序，不读写排序缓存或发布标记
         
         返回：
             排序后的 items 列表
         """
+        if fresh:
+            # 公开榜单源数据可能由其它进程更新，旧ID顺序不可作为当前排序依据。
+            # 直接排序比逐条序列化/摘要校验更轻；不触碰其它调用的缓存与代次。
+            sort_fn(items)
+            return items
         now = time.time()
-        
-        # 检查缓存是否有效
-        if cache_key in self._cache:
-            sorted_ids, cached_time = self._cache[cache_key]
-            if now - cached_time < self._ttl:
-                # 缓存有效，用缓存的顺序重排当前数据
-                id_order = {id_: idx for idx, id_ in enumerate(sorted_ids)}
-                # 按缓存的顺序排序，新数据（不在缓存中的）放在最后
-                return sorted(items, key=lambda x: id_order.get(x.get("id"), float('inf')))
-        
-        # 缓存无效或不存在，执行排序
-        sort_fn(items)
-        
-        # 缓存排序后的 ID 列表
-        sorted_ids = [item.get("id") for item in items]
-        self._cache[cache_key] = (sorted_ids, now)
-        
-        return items
-    
+        with self._lock:
+            stale_read = read_epoch is not None and read_epoch is not self._read_epoch
+            hit = False
+            if not stale_read:
+                cached = self._cache.get(cache_key)
+                hit = cached is not None and now - cached[1] < self._ttl
+                if hit:
+                    self._cache.move_to_end(cache_key)
+                else:
+                    # 未命中时机会性回收过期ID列表，不新增后台线程。
+                    for key, (_, cached_time) in list(self._cache.items()):
+                        if now - cached_time >= self._ttl:
+                            self._remove_cached(key)
+                    token = object()
+                    # 同键后开始的排序取得发布权；旧计算只能返回自己的结果。
+                    self._pending_sorts[cache_key] = token
+        if stale_read:
+            # 失效前读取的输入不能命中新代缓存，也不能撤销新代发布标记。
+            sort_fn(items)
+            return items
+
+        if hit:
+            sorted_ids, cached_time = cached
+            id_order = {id_: idx for idx, id_ in enumerate(sorted_ids)}
+            return sorted(items, key=lambda x: id_order.get(x.get("id"), float('inf')))
+
+        completed = False
+        try:
+            # 耗时排序及业务回调均在锁外，失效/其它键不会被阻塞。
+            sort_fn(items)
+            # 超预算仍返回原排序结果，不构造/保留额外ID列表，也不淘汰有效旧缓存。
+            sorted_ids = ([item.get("id") for item in items]
+                          if self._max_entries and len(items) <= self._max_ids else [])
+            completed = True
+            return items
+        finally:
+            with self._lock:
+                if self._pending_sorts.get(cache_key) is token:
+                    del self._pending_sorts[cache_key]
+                    if completed and sorted_ids:
+                        self._remove_cached(cache_key)
+                        while (len(self._cache) >= self._max_entries or
+                               self._cached_ids + len(sorted_ids) > self._max_ids):
+                            self._remove_cached(next(iter(self._cache)))
+                        self._cache[cache_key] = (sorted_ids, now)
+                        self._cached_ids += len(sorted_ids)
+
     def invalidate(self, prefix: str = ""):
         """
         清除缓存
@@ -527,23 +615,32 @@ class SortCache:
         参数：
             prefix: 缓存键前缀，如果指定则只清除匹配的缓存，否则清除所有缓存
         """
-        if prefix:
-            keys_to_remove = [k for k in self._cache if k.startswith(prefix)]
-            for k in keys_to_remove:
-                del self._cache[k]
-        else:
-            self._cache.clear()
-    
+        with self._lock:
+            self._read_epoch = object()
+            if prefix:
+                for key in list(self._cache):
+                    if key.startswith(prefix):
+                        self._remove_cached(key)
+                for key in list(self._pending_sorts):
+                    if key.startswith(prefix):
+                        del self._pending_sorts[key]
+            else:
+                self._cache.clear()
+                self._cached_ids = 0
+                self._pending_sorts.clear()
+
     def get_stats(self) -> Dict[str, Any]:
         """
         获取缓存统计信息（用于调试）
         """
         now = time.time()
-        valid_count = sum(1 for _, ts in self._cache.values() if now - ts < self._ttl)
+        with self._lock:
+            snapshot = tuple(self._cache.values())
+        valid_count = sum(1 for _, ts in snapshot if now - ts < self._ttl)
         return {
-            "total_cached": len(self._cache),
+            "total_cached": len(snapshot),
             "valid": valid_count,
-            "expired": len(self._cache) - valid_count,
+            "expired": len(snapshot) - valid_count,
             "ttl": self._ttl
         }
 

models.py

@@ -69,6 +69,17 @@ class UserUpdate(BaseModel):
     avatarDataUrl: Optional[str] = None
     bannerUrl: Optional[str] = None         # 🚀 个人资料卡背景图云端URL
 
+class SettingsPrivacy(BaseModel):
+    model_config = {"extra": "forbid"}
+    follows: bool = Field(strict=True)
+    likes: bool = Field(strict=True)
+    favorites: bool = Field(strict=True)
+    downloads: bool = Field(strict=True)
+
+class UserSettings(UserUpdate):
+    model_config = {"extra": "forbid"}
+    privacy: SettingsPrivacy
+
 # 🧹 P2清理：移除零引用的 PasswordReset 模型（重置密码实际走 UserUpdate/专用端点）
 
 class CommentCreate(BaseModel):

owned_thread.py

@@ -1,33 +1,75 @@
-"""等待所属同步工作结束；取消不能遗留仍持有文件/SDK操作的线程。"""
+"""等待所属同步工作真实终态；内部取消也不能遗留仍读写资源的线程。"""
 import asyncio
+import threading
 
 
 async def run_owned_thread(function, *args, **kwargs):
+    ownership = threading.Lock()
+    finished = threading.Event()
+    started = False
+    outcome = None
+
+    def abandon_unstarted(result):
+        nonlocal outcome
+        with ownership:
+            if not started and not finished.is_set():
+                outcome = result
+                finished.set()
+
+    def execute():
+        nonlocal started, outcome
+        with ownership:
+            if finished.is_set():
+                return
+            started = True
+        try:
+            result = (True, function(*args, **kwargs))
+        except BaseException as error:
+            # 同步异常作为数据交回；取消shield不暴露私有异常正文。
+            result = (False, error)
+        with ownership:
+            outcome = result
+            finished.set()
+
     async def finish():
-        # 被取消的shield在部分Python版本会立即记录内层异常正文。
-        # 先把常规线程结果封装为数据，所属调用再决定传播异常或取消。
         try:
-            return True, await asyncio.to_thread(function, *args, **kwargs)
-        except Exception as error:
-            return False, error
+            await asyncio.to_thread(execute)
+        except asyncio.CancelledError:
+            raise
+        except BaseException as error:
+            abandon_unstarted((False, error))
+
+    coroutine = finish()
+    try:
+        task = asyncio.create_task(coroutine)
+    except BaseException:
+        coroutine.close()
+        raise
+
+    def consume(completed):
+        if completed.cancelled():
+            # 取消和线程启动竞争时，只有尚未开始的工作能被撤销。
+            abandon_unstarted((False, asyncio.CancelledError()))
+        else:
+            error = completed.exception()
+            if error is not None:
+                abandon_unstarted((False, error))
+    task.add_done_callback(consume)
 
-    task = asyncio.create_task(finish())
     cancelled = None
-    while not task.done():
+    while not finished.is_set():
         try:
-            await asyncio.shield(task)
+            if not task.done():
+                await asyncio.shield(task)
+            else:
+                # 内部task已取消但线程仍在运行，不占线程池等待其终态。
+                await asyncio.sleep(0.01)
         except asyncio.CancelledError as error:
             cancelled = error
-        except Exception:
-            break
+    with ownership:
+        succeeded, result = outcome
     if cancelled is not None:
-        # 消费线程结果/异常，取消仍是本次调用的终态。
-        try:
-            task.result()
-        except Exception:
-            pass
         raise cancelled
-    succeeded, result = task.result()
     if not succeeded:
         raise result
     return result

periodic_tasks.py

@@ -0,0 +1,59 @@
+"""后台任务的所属启动、停止信号与取消安全等待（保留周期任务导出名）。"""
+import asyncio
+import logging
+
+logger = logging.getLogger("ComfyUI-Ranking.PeriodicTasks")
+
+
+def start_periodic_task(state, name, factory):
+    current = getattr(state, name + "_worker", None)
+    if current is not None and not current.done():
+        return current
+    stop = asyncio.Event()
+    coroutine = factory(stop)
+    try:
+        task = asyncio.create_task(coroutine)
+    except BaseException:
+        coroutine.close()
+        raise
+    setattr(state, name + "_stop", stop)
+    setattr(state, name + "_worker", task)
+
+    def consume(completed):
+        if not completed.cancelled():
+            error = completed.exception()
+            if error is not None:
+                logger.warning("后台任务结束异常: task=%s | error_type=%s", name, type(error).__name__)
+    task.add_done_callback(consume)
+    return task
+
+
+def request_periodic_stop(state, names):
+    tasks = []
+    for name in names:
+        stop = getattr(state, name + "_stop", None)
+        if stop is not None:
+            stop.set()
+        task = getattr(state, name + "_worker", None)
+        if task is not None:
+            tasks.append(task)
+    return tasks
+
+
+async def wait_periodic_tasks(tasks):
+    cancelled = None
+    for task in tasks:
+        while not task.done():
+            try:
+                await asyncio.shield(task)
+            except asyncio.CancelledError as error:
+                # worker自身取消与关闭等待者取消分别处理。
+                if asyncio.current_task().cancelling() or not task.cancelled():
+                    cancelled = error
+            except Exception:
+                break
+        if not task.cancelled():
+            # 业务异常已有所属回调诊断，不阻止其它任务收尾和最终同步。
+            task.exception()
+    if cancelled is not None:
+        raise cancelled

profile_io.py

@@ -0,0 +1,106 @@
+"""资料文件I/O在后台线程执行；准入在排队前，已开始工作取消不提前释放容量。"""
+import asyncio
+import threading
+from fastapi import HTTPException
+
+MAX_PROFILE_IO = 8
+
+class ProfileIO:
+    def __init__(self, capacity=MAX_PROFILE_IO):
+        self.capacity = threading.BoundedSemaphore(capacity)
+        self._gate = threading.Lock()
+        self._accepting = True
+        self._pending = 0
+        self._closers = 0
+        self._drained = threading.Event()
+        self._drained.set()
+
+    def start(self):
+        """同进程重新启动仅在上一代所属工作全部终态后开放准入。"""
+        with self._gate:
+            if self._pending or self._closers:
+                raise RuntimeError("profile work is still pending")
+            self._accepting = True
+
+    async def close(self):
+        """关闭准入并等真实工作终态；取消关闭等待也不提前宣称排空。"""
+        with self._gate:
+            self._accepting = False
+            self._closers += 1
+        try:
+            cancelled = False
+            while not self._drained.is_set():
+                try:
+                    # 不向共用线程池提交等待任务，避免池饱和时相互等待。
+                    await asyncio.sleep(0.01)
+                except asyncio.CancelledError:
+                    cancelled = True
+            if cancelled:
+                raise asyncio.CancelledError
+        finally:
+            with self._gate:
+                self._closers -= 1
+
+    def _finish(self):
+        with self._gate:
+            self.capacity.release()
+            self._pending -= 1
+            if not self._pending:
+                self._drained.set()
+
+    async def run(self, function, *args, **kwargs):
+        with self._gate:
+            if not self._accepting or not self.capacity.acquire(blocking=False):
+                raise HTTPException(status_code=503, detail="用户数据暂不可用")
+            self._pending += 1
+            self._drained.clear()
+
+        # 内部任务也可能在宿主关闭时被取消。锁区分未开始与已开始的工作，
+        # 保证取消和线程启动竞争时只释放一次，且取消后不再开始业务写入。
+        ownership = threading.Lock()
+        started = False
+        released = False
+
+        def release_unstarted():
+            nonlocal released
+            with ownership:
+                if not started and not released:
+                    released = True
+                    self._finish()
+
+        def execute():
+            nonlocal started, released
+            with ownership:
+                if released:
+                    return None
+                started = True
+            try:
+                return function(*args, **kwargs)
+            finally:
+                with ownership:
+                    released = True
+                    self._finish()
+
+        coroutine = asyncio.to_thread(execute)
+        try:
+            task = asyncio.create_task(coroutine)
+        except BaseException:
+            coroutine.close(); release_unstarted()
+            raise
+
+        def consume(completed):
+            if completed.cancelled():
+                release_unstarted()
+            else:
+                # 提交默认线程池本身也可能失败，尚未开始的准入必须收回。
+                release_unstarted()
+                completed.exception()
+        task.add_done_callback(consume)
+        # 外层请求取消不取消所属任务；已开始的工作只在线程终态释放容量。
+        return await asyncio.shield(task)
+
+_profile_io = ProfileIO()
+run_profile_io = _profile_io.run
+
+start_profile_io = _profile_io.start
+close_profile_io = _profile_io.close

proxy_resources.py

@@ -17,6 +17,21 @@ async def _await_close(task):
         raise cancelled
 
 
+async def close_redirect_response(response):
+    """中间跳转响应关闭一次；完整等待后取消优先，普通关闭失败仍向外传播。"""
+    async def finish():
+        try:
+            await response.aclose()
+        except Exception as error:
+            logger.warning("PROXY_REDIRECT_CLOSE_FAIL | error_type=%s", type(error).__name__)
+            return error
+    task = asyncio.create_task(finish())
+    await _await_close(task)
+    error = task.result()
+    if error is not None:
+        raise error
+
+
 class ProxyResources:
     def __init__(self, client):
         self.client = client

rate_limit.py

@@ -48,8 +48,9 @@ def _consume_shared_limit(identity: str, scope: str, limit: int) -> None:
     now = int(time.time())
     window_start = now - (now % 60)
     bucket_key = hashlib.sha256(f"{identity}\0{scope}".encode("utf-8")).hexdigest()
-    session = SessionLocal()
+    session = None
     try:
+        session = SessionLocal()
         dialect = engine.dialect.name
         values = {
             "bucket_key": bucket_key,
@@ -117,14 +118,22 @@ def _consume_shared_limit(identity: str, scope: str, limit: int) -> None:
             raise HTTPException(status_code=429, detail=f"请求过于频繁，限流 {limit}/分钟，请稍后再试")
         _cleanup_old_buckets_if_due()
     except HTTPException:
-        session.rollback()
+        if session is not None:
+            try: session.rollback()
+            except Exception: logger.error("SHARED_RATE_LIMIT_ROLLBACK_FAIL")
         raise
     except Exception as exc:
-        session.rollback()
+        if session is not None:
+            try: session.rollback()
+            except Exception: logger.error("SHARED_RATE_LIMIT_ROLLBACK_FAIL")
         logger.error("SHARED_RATE_LIMIT_DB_FAIL | scope=%s | error_type=%s", scope, type(exc).__name__)
         raise HTTPException(status_code=503, detail="限流服务暂不可用") from exc
     finally:
-        session.close()
+        if session is not None:
+            try: session.close()
+            except Exception as error:
+                logger.error("SHARED_RATE_LIMIT_CLOSE_FAIL | error_type=%s", type(error).__name__)
+                raise HTTPException(status_code=503, detail="限流服务暂不可用") from None
 
 
 def _cleanup_old_buckets_if_due() -> None:

resource_deletion.py

@@ -1,5 +1,6 @@
 """资源删除的本地恢复记录与关联清理；不访问或修改资金数据。"""
 import asyncio
+from owned_thread import run_owned_thread
 import copy
 import datetime
 import logging
@@ -162,7 +163,7 @@ def delete_resource(store, item_id, account, can_moderate=False):
         raise ResourceStoreError(503, '资源删除暂时无法核实，请稍后核对') from None
 
 
-def repair_resource_deletions(store, cursor=0, limit=50, should_stop=lambda: False):
+def repair_resource_deletions(store, cursor=0, limit=50, should_stop=lambda: False, *, on_uncertain=None):
     queue = _read_queue(store)
     ids = list(queue)
     if not ids:
@@ -179,6 +180,11 @@ def repair_resource_deletions(store, cursor=0, limit=50, should_stop=lambda: Fal
                 _, updated = _cleanup(store, queue[item_id])
                 changed += bool(updated)
         except Exception as error:
+            if on_uncertain is not None:
+                try:
+                    on_uncertain()
+                except Exception as notification_error:
+                    logger.warning('资源删除恢复缓存通知失败: error_type=%s', type(notification_error).__name__)
             logger.warning('资源删除恢复失败: error_type=%s', type(error).__name__)
         processed += 1
     return (start + processed) % len(ids), changed
@@ -188,18 +194,41 @@ async def resource_deletion_worker(store, stop, invalidate=None, interval=30):
     initialized = False
     cursor = 0
     while not stop.is_set():
+        attempted = threading.Event()
+        changed_data = threading.Event()
+        uncertain = threading.Event()
+        cancelled = False
+
+        def repair():
+            attempted.set()
+            try:
+                result = repair_resource_deletions(store, cursor, 50, stop.is_set, on_uncertain=uncertain.set)
+            except Exception:
+                uncertain.set()
+                raise
+            if result[1]:
+                changed_data.set()
+            return result
+
         try:
             if not initialized:
-                await asyncio.to_thread(store.atomic_update, DELETION_FILE, _queue, default_data={})
+                await run_owned_thread(store.atomic_update, DELETION_FILE, _queue, default_data={})
                 initialized = True
-            cursor, changed = await asyncio.to_thread(repair_resource_deletions, store, cursor, 50, stop.is_set)
-            if changed and invalidate:
+                if stop.is_set():
+                    break
+            cursor, changed = await run_owned_thread(repair)
+        except asyncio.CancelledError:
+            cancelled = True
+            raise
+        except Exception as error:
+            logger.warning('资源删除恢复轮次失败: error_type=%s', type(error).__name__)
+        finally:
+            # 所属等待已确认真实线程终态，取消也不能跳过已发布数据的失效。
+            if invalidate and (changed_data.is_set() or uncertain.is_set() or (cancelled and attempted.is_set())):
                 try:
                     invalidate('items:')
                 except Exception:
                     logger.warning('资源删除恢复后的排序失效失败')
-        except Exception as error:
-            logger.warning('资源删除恢复轮次失败: error_type=%s', type(error).__name__)
         if stop.is_set():
             break
         try:

resource_publication.py

@@ -1,5 +1,6 @@
 """持久发布身份先入SQL；JSON失败可恢复，同次重试不制造新资源。"""
 import asyncio
+from owned_thread import run_owned_thread
 import copy
 import hashlib
 import json
@@ -179,7 +180,7 @@ async def publication_worker(bind, store, invalidate, stop, interval=30):
     cursor = ''
     while not stop.is_set():
         try:
-            cursor = await asyncio.to_thread(repair_publications, bind, store, invalidate, cursor)
+            cursor = await run_owned_thread(repair_publications, bind, store, invalidate, cursor)
         except Exception:
             logging.getLogger(__name__).warning('资源发布恢复轮次暂未完成')
         try:

router_comments.py

@@ -4,7 +4,7 @@ import time
 import uuid
 from html import escape
 import 数据库连接 as db
-from comment_store import append_comment, delete_comment
+from comment_store import COMMENT_MAX_LENGTH, append_comment, delete_comment
 from notifications import add_notification
 from models import InteractionToggle, CommentCreate
 from 安全认证 import require_auth, is_admin
@@ -70,7 +70,7 @@ async def toggle_interaction(request: Request, interaction: InteractionToggle, c
 async def post_comment(request: Request, comment: CommentCreate, current_user: str = Depends(require_auth)):
     comment.author = current_user  # 强制使用认证用户身份，防止伪造
     # 评论内容长度校验
-    if not comment.content or len(comment.content) > 5000:
+    if not comment.content or len(comment.content) > COMMENT_MAX_LENGTH:
         raise HTTPException(status_code=400, detail="评论长度必须在1-5000字符之间")
     # HTML 转义，防止 XSS
     comment.content = escape(comment.content)

router_items.py

@@ -14,7 +14,7 @@ import 数据库连接 as db
 from models import ItemCreate, ItemUpdate, RatingRequest
 from 安全认证 import require_auth, check_ownership, is_admin
 from 数据库连接 import invalidate_cache
-from db_utils import record_view, sort_cache
+from db_utils import read_public_source, record_view, sort_cache
 from github_source import repository_parts, select_github_token
 from github_versions import refresh_resource_version, invalidate_source
 from resource_store import publish_resource, published_response, edit_resource, ResourceStoreError
@@ -70,9 +70,10 @@ data_file = "items.json"
 
 @router.get("/api/items")
 async def get_items(type: str = "tool", sort: str = "time", limit: int = 50): # 优化：默认限制调大至 50，提升前端列表体验
-    items_db = db.load_data("items.json", default_data=[])
-    comments_db = db.load_data("comments.json", default_data={})
-    versions_db = db.load_data("versions.json", default_data={}) # 读取版本库
+    read_epoch = sort_cache.capture_read_epoch()
+    items_db = read_public_source(db, "items.json", default_data=[])
+    comments_db = read_public_source(db, "comments.json", default_data={})
+    versions_db = read_public_source(db, "versions.json", default_data={}) # 读取版本库
     
     # 如果是推荐榜，匹配所有 recommend 开头的子类型
     if type == "recommend":
@@ -81,28 +82,32 @@ async def get_items(type: str = "tool", sort: str = "time", limit: int = 50): #
         filtered_items = [item for item in items_db if item.get("type") == type]
     
     # 🗂️ 使用排序缓存优化排序性能
-    cache_key = f"items:{type}:{sort}"
+    cache_sort = sort if sort in ("time", "likes", "favorites", "downloads", "tips", "views", "daily_views", "rating") else "time"
+    cache_key = f"items:{type}:{cache_sort}"
     
-    def sort_fn(data):
+    current_month = datetime.date.today().strftime("%Y-%m") if sort == "tips" else None
+    def order_key(item):
         if sort == "likes":
-            data.sort(key=lambda x: x.get("likes", 0), reverse=True)
+            return item.get("likes", 0)
         elif sort == "favorites":
-            data.sort(key=lambda x: x.get("favorites", 0), reverse=True)
+            return item.get("favorites", 0)
         elif sort == "downloads":
-            data.sort(key=lambda x: x.get("uses", 0), reverse=True)
+            return item.get("uses", 0)
         elif sort == "tips": # 🚀 按近期打赏排序
-            current_month = datetime.date.today().strftime("%Y-%m")
-            data.sort(key=lambda x: x.get("tip_history", {}).get(current_month, 0), reverse=True)
+            return item.get("tip_history", {}).get(current_month, 0)
         elif sort == "views": # 👁️ 按总访问量排序
-            data.sort(key=lambda x: x.get("views", 0), reverse=True)
+            return item.get("views", 0)
         elif sort == "daily_views": # 👁️ 按日访问量排序
-            data.sort(key=lambda x: x.get("daily_views", 0), reverse=True)
+            return item.get("daily_views", 0)
         elif sort == "rating": # ⭐ 按评分排序
-            data.sort(key=lambda x: (x.get("rating_avg", 0), x.get("rating_count", 0)), reverse=True)
+            return (item.get("rating_avg", 0), item.get("rating_count", 0))
         else: # time 或其他默认
-            data.sort(key=lambda x: x.get("created_at", 0), reverse=True)
+            return item.get("created_at", 0)
+
+    def sort_fn(data):
+        data.sort(key=order_key, reverse=True)
     
-    filtered_items = sort_cache.get_sorted(cache_key, filtered_items, sort_fn)
+    filtered_items = sort_cache.get_sorted(cache_key, filtered_items, sort_fn, read_epoch=read_epoch, fresh=True)
         
     result_items = []
     for item in filtered_items:
@@ -162,8 +167,8 @@ async def search_creators(keyword: str, sort: str = "downloads", limit: int = 50
     搜索创作者
     根据关键词搜索创作者（name、account、shortDesc 不区分大小写子串匹配）
     """
-    users_db = db.load_data("users.json", default_data={})
-    items_db = db.load_data("items.json", default_data=[])
+    users_db = read_public_source(db, "users.json", default_data={})
+    items_db = read_public_source(db, "items.json", default_data=[])
     
     # 🚀 P1性能优化：预构建 author->items 索引
     author_items_index = {}
@@ -235,8 +240,8 @@ async def get_creators(sort: str = "downloads", limit: int = 100):
     🚀 P1性能优化：预构建 author->items 索引，避免 N+1 查询
     ⚡ P2性能优化：移除大字段（trendData、commentsData、tip_board），移至详情接口按需加载
     """
-    users_db = db.load_data("users.json", default_data={})
-    items_db = db.load_data("items.json", default_data=[])
+    users_db = read_public_source(db, "users.json", default_data={})
+    items_db = read_public_source(db, "items.json", default_data=[])
     
     # 🚀 P1性能优化：预构建 author->items 索引，复杂��从 O(n*m) 降到 O(n+m)
     author_items_index = {}
@@ -297,9 +302,9 @@ async def get_creator_details(account: str):
     供前端展开卡片时按需加载
     包含：trendData、commentsData、tip_board
     """
-    users_db = db.load_data("users.json", default_data={})
-    items_db = db.load_data("items.json", default_data=[])
-    comments_db = db.load_data("comments.json", default_data={})
+    users_db = read_public_source(db, "users.json", default_data={})
+    items_db = read_public_source(db, "items.json", default_data=[])
+    comments_db = read_public_source(db, "comments.json", default_data={})
     
     # 检查用户是否存在
     if account not in users_db:
@@ -464,9 +469,9 @@ async def check_github_updates(request: Request, current_user: str = Depends(req
 @router.get("/api/items/{item_id}")
 async def get_item_by_id(item_id: str):
     """根据ID获取单个资源的详细信息"""
-    items_db = db.load_data("items.json", default_data=[])
-    comments_db = db.load_data("comments.json", default_data={})
-    versions_db = db.load_data("versions.json", default_data={})
+    items_db = read_public_source(db, "items.json", default_data=[])
+    comments_db = read_public_source(db, "comments.json", default_data={})
+    versions_db = read_public_source(db, "versions.json", default_data={})
     
     for item in items_db:
         if item["id"] == item_id:
@@ -490,7 +495,7 @@ async def get_item_by_id(item_id: str):
 @router.get("/api/item/{item_id}/version")
 async def get_item_version(item_id: str):
     """获取单个资源的最新版本号"""
-    versions_db = db.load_data("versions.json", default_data={})
+    versions_db = read_public_source(db, "versions.json", default_data={})
     return {"status": "success", "version": _get_version_str(versions_db, item_id)}
 
 @router.delete("/api/items/{item_id}")

router_messages.py

@@ -6,6 +6,8 @@ import asyncio
 import time
 import uuid
 import subprocess
+import logging
+from script_execution import run_admin_script_process, ScriptExecutionBusy
 import os
 import 数据库连接 as db
 import chat_store
@@ -15,6 +17,7 @@ from 安全认证 import require_auth, is_admin
 
 # 🔒 频次限制（与 router_users_auth.py / router_wallet.py / router_prompts.py 同一写法）
 
+logger = logging.getLogger("ComfyUI-Ranking.AdminScripts")
 router = APIRouter()
 
 # ==========================================
@@ -96,20 +99,7 @@ async def run_admin_script(request: Request, req: AdminScriptRequest, current_us
         sensitive_prefixes = ('GITHUB', 'JWT', 'HF_', 'DATABASE', 'ALIPAY', 'PASSWORD')
         clean_env = {k: v for k, v in os.environ.items()
                      if not any(k.startswith(prefix) for prefix in sensitive_prefixes)}
-        # 执行脚本，超时缩短至30秒
-        # ⚡ 性能修复：subprocess.run 会同步等待子进程结束（最长 30 秒），
-        # 在 async 处理器里直接调用会占死唯一的事件循环，脚本跑完前整个服务无法响应任何请求；
-        # 交给线程池执行，TimeoutExpired 等异常仍由下方原有 except 分支按原样处理。
-        result = await asyncio.to_thread(
-            subprocess.run,
-            ["python", script_path],
-            capture_output=True,
-            text=True,
-            timeout=30,
-            cwd=current_dir,
-            encoding="utf-8",
-            env=clean_env
-        )
+        result = await run_admin_script_process(script_path, cwd=current_dir, env=clean_env)
         
         output = ""
         if result.stdout:
@@ -129,15 +119,18 @@ async def run_admin_script(request: Request, req: AdminScriptRequest, current_us
             "output": output
         }
         
+    except ScriptExecutionBusy:
+        raise HTTPException(status_code=503, detail="脚本执行繁忙，请先确认当前执行状态") from None
     except subprocess.TimeoutExpired:
         return {
             "status": "error",
             "output": "❌ 脚本执行超时 (30秒)"
         }
-    except Exception as e:
+    except Exception as error:
+        logger.error("ADMIN_SCRIPT_ERROR | error_type=%s", type(error).__name__)
         return {
             "status": "error",
-            "output": f"❌ 执行异常: {str(e)}"
+            "output": "脚本执行暂不可用，请先确认当前状态"
         }
 
 # ==========================================

router_posts.py

@@ -10,9 +10,9 @@ from fastapi import APIRouter, HTTPException, Depends, Request
 from sqlalchemy.orm import Session
 from models import PostCreate, PostUpdate, RatingRequest
 import 数据库连接 as db
-from comment_store import append_comment, increment_comment_count, attach_comment_counts
+from comment_store import COMMENT_MAX_LENGTH, append_comment, increment_comment_count, attach_comment_counts, read_comment_view
 from 安全认证 import require_auth
-from db_utils import record_view, sort_cache
+from db_utils import read_public_source, record_view, sort_cache
 from database_sql import get_db
 from models_sql import Wallet, Transaction
 from notifications import add_notification
@@ -60,32 +60,37 @@ async def get_posts(page: int = 1, limit: int = 20, sort: str = "latest"):
     - sort=daily_views: 按日访问量降序
     - sort=tips: 按打赏总额降序
     """
-    posts_db = db.load_data("posts.json", default_data=[])
-    users_db = db.load_data("users.json", default_data={})
+    read_epoch = sort_cache.capture_read_epoch()
+    posts_db = read_public_source(db, "posts.json", default_data=[])
+    users_db = read_public_source(db, "users.json", default_data={})
     
     # users_db 已经是 {account: user_info} 格式，直接使用
     user_map = users_db
     
     # 🗂️ 使用排序缓存优化排序性能
-    cache_key = f"posts:{sort}"
+    cache_sort = sort if sort in ("latest", "likes", "favorites", "views", "daily_views", "tips", "rating") else "latest"
+    cache_key = f"posts:{cache_sort}"
     
-    def sort_fn(data):
+    def order_key(item):
         if sort == "likes":
-            data.sort(key=lambda x: x.get("likes", 0), reverse=True)
+            return item.get("likes", 0)
         elif sort == "favorites":
-            data.sort(key=lambda x: x.get("favorites", 0), reverse=True)
+            return item.get("favorites", 0)
         elif sort == "views":
-            data.sort(key=lambda x: x.get("views", 0), reverse=True)
+            return item.get("views", 0)
         elif sort == "daily_views":
-            data.sort(key=lambda x: x.get("daily_views", 0), reverse=True)
+            return item.get("daily_views", 0)
         elif sort == "tips":
-            data.sort(key=lambda x: sum(t.get("amount", 0) for t in x.get("tip_board", [])), reverse=True)
+            return sum(t.get("amount", 0) for t in item.get("tip_board", []))
         elif sort == "rating":  # ⭐ 按评分排序
-            data.sort(key=lambda x: (x.get("rating_avg", 0), x.get("rating_count", 0)), reverse=True)
+            return (item.get("rating_avg", 0), item.get("rating_count", 0))
         else:  # latest 或其他默认
-            data.sort(key=lambda x: x.get("created_at", 0), reverse=True)
+            return item.get("created_at", 0)
+
+    def sort_fn(data):
+        data.sort(key=order_key, reverse=True)
     
-    sorted_posts = sort_cache.get_sorted(cache_key, posts_db, sort_fn)
+    sorted_posts = sort_cache.get_sorted(cache_key, posts_db, sort_fn, read_epoch=read_epoch, fresh=True)
     
     # 分页
     start = (page - 1) * limit
@@ -439,6 +444,8 @@ async def tip_post(request: Request, post_id: str, amount: int, is_anon: bool =
             Transaction.tx_type == "TIP_OUT",
             Transaction.amount == -amount,
             Transaction.related_account == author,
+            Transaction.item_id == post_id,
+            Transaction.item_type == "post",
             Transaction.created_at >= recent_cutoff
         ).with_for_update().first()
         if duplicate_tx:
@@ -567,29 +574,8 @@ async def tip_post(request: Request, post_id: str, amount: int, is_anon: bool =
 
 @router.get("/api/posts/{post_id}/comments")
 async def get_post_comments(post_id: str):
-    """
-    获取帖子评论
-    """
-    comments_db = db.load_data("comments.json", default_data={})
-    users_db = db.load_data("users.json", default_data={})
-    
-    # users_db 已经是 {account: user_info} 格式，直接使用
-    user_map = users_db
-    
-    # comments_db 是 {item_id: [comments]} 格式
-    post_comments = comments_db.get(post_id, [])
-    
-    # 附加用户信息
-    result = []
-    for c in post_comments:
-        author_info = user_map.get(c.get("author"), {})
-        result.append({
-            **c,
-            "author_name": author_info.get("name", c.get("author")),
-            "author_avatar": author_info.get("avatarDataUrl", "")
-        })
-    
-    return {"status": "success", "data": result}
+    """获取评论；有界后台读取并保留原作者附加字段。"""
+    return {"status": "success", "data": await read_comment_view(post_id)}
 
 @router.post("/api/posts/{post_id}/comments")
 @shared_limit(10)
@@ -599,6 +585,8 @@ async def add_post_comment(request: Request, post_id: str, content: str, current
     """
     if not content or not content.strip():
         raise HTTPException(status_code=400, detail="评论内容不能为空")
+    if len(content) > COMMENT_MAX_LENGTH:
+        raise HTTPException(status_code=400, detail="评论长度必须在1-5000字符之间")
     
     posts_db = db.load_data("posts.json", default_data=[])
     

router_prompts.py

@@ -18,9 +18,9 @@ from sqlalchemy.orm import Session
 from models import PromptCreate, PromptUpdate, PromptCategoryAdd, PromptPurchaseRequest
 from typing import Optional
 import 数据库连接 as db
-from comment_store import append_comment, increment_comment_count, attach_comment_counts
+from comment_store import COMMENT_MAX_LENGTH, append_comment, increment_comment_count, attach_comment_counts, read_comment_view
 from 安全认证 import require_auth, optional_auth, is_admin
-from db_utils import record_view, sort_cache
+from db_utils import read_public_source, record_view, sort_cache
 from database_sql import get_db
 from models_sql import Wallet, Transaction, Ownership
 from notifications import add_notification
@@ -232,28 +232,33 @@ async def get_prompts(prompt_type: str = None, category: str = None,
     - sort: latest / likes / favorites / views / daily_views / tips
     ⚠️ 列表接口永不下发 prompt_text（付费内容脱敏）
     """
+    read_epoch = sort_cache.capture_read_epoch()
     prompts_db = await _prompt_operation(read_prompts, db)
-    users_db = db.load_data("users.json", default_data={})
+    users_db = read_public_source(db, "users.json", default_data={})
     user_map = users_db
 
     # 🗂️ 使用排序缓存优化排序性能
-    cache_key = f"prompts:{sort}"
+    cache_sort = sort if sort in ("latest", "likes", "favorites", "views", "daily_views", "tips") else "latest"
+    cache_key = f"prompts:{cache_sort}"
 
-    def sort_fn(data):
+    def order_key(item):
         if sort == "likes":
-            data.sort(key=lambda x: x.get("likes", 0), reverse=True)
+            return item.get("likes", 0)
         elif sort == "favorites":
-            data.sort(key=lambda x: x.get("favorites", 0), reverse=True)
+            return item.get("favorites", 0)
         elif sort == "views":
-            data.sort(key=lambda x: x.get("views", 0), reverse=True)
+            return item.get("views", 0)
         elif sort == "daily_views":
-            data.sort(key=lambda x: x.get("daily_views", 0), reverse=True)
+            return item.get("daily_views", 0)
         elif sort == "tips":
-            data.sort(key=lambda x: sum(t.get("amount", 0) for t in x.get("tip_board", [])), reverse=True)
+            return sum(t.get("amount", 0) for t in item.get("tip_board", []))
         else:  # latest 或其他默认
-            data.sort(key=lambda x: x.get("created_at", 0), reverse=True)
+            return item.get("created_at", 0)
+
+    def sort_fn(data):
+        data.sort(key=order_key, reverse=True)
 
-    sorted_prompts = sort_cache.get_sorted(cache_key, prompts_db, sort_fn)
+    sorted_prompts = sort_cache.get_sorted(cache_key, prompts_db, sort_fn, read_epoch=read_epoch, fresh=True)
 
     # 类型与分类过滤
     if prompt_type in VALID_PROMPT_TYPES:
@@ -669,6 +674,8 @@ async def tip_prompt(request: Request, prompt_id: str, amount: int, is_anon: boo
             Transaction.tx_type == "TIP_OUT",
             Transaction.amount == -amount,
             Transaction.related_account == author,
+            Transaction.item_id == prompt_id,
+            Transaction.item_type == "prompt",
             Transaction.created_at >= recent_cutoff
         ).with_for_update().first()
         if duplicate_tx:
@@ -905,28 +912,8 @@ async def purchase_prompt(request: Request, prompt_id: str,
 
 @router.get("/api/prompts/{prompt_id}/comments")
 async def get_prompt_comments(prompt_id: str):
-    """
-    获取提示词评论
-    """
-    comments_db = db.load_data("comments.json", default_data={})
-    users_db = db.load_data("users.json", default_data={})
-    user_map = users_db
-
-    # comments_db 是 {item_id: [comments]} 格式
-    prompt_comments = comments_db.get(prompt_id, [])
-
-    # 附加用户信息
-    result = []
-    for c in prompt_comments:
-        author_info = user_map.get(c.get("author"), {})
-        result.append({
-            **c,
-            "author_name": author_info.get("name", c.get("author")),
-            "author_avatar": author_info.get("avatarDataUrl", "")
-        })
-
-    return {"status": "success", "data": result}
-
+    """获取评论；有界后台读取并保留原作者附加字段。"""
+    return {"status": "success", "data": await read_comment_view(prompt_id)}
 
 @router.post("/api/prompts/{prompt_id}/comments")
 @shared_limit(10)
@@ -936,6 +923,8 @@ async def add_prompt_comment(request: Request, prompt_id: str, content: str, cur
     """
     if not content or not content.strip():
         raise HTTPException(status_code=400, detail="评论内容不能为空")
+    if len(content) > COMMENT_MAX_LENGTH:
+        raise HTTPException(status_code=400, detail="评论长度必须在1-5000字符之间")
 
     prompts_db = db.load_data("prompts.json", default_data=[])
 

router_proxy.py

@@ -26,6 +26,7 @@ from safe_http_transport import ValidatedAsyncHTTPTransport
 from workflow_download_service import prepare_workflow_download
 from zip_stream import ZIP_DOWNLOAD_MAX_BYTES
 from github_zip_service import prepare_github_zip
+from proxy_resources import close_redirect_response
 
 router = APIRouter()
 MAX_PROXY_DOWNLOAD_BYTES = int(os.environ.get("MAX_PROXY_DOWNLOAD_BYTES", str(256 * 1024 * 1024)))
@@ -55,7 +56,9 @@ async def _validate_public_http_target(url: str) -> str:
         except OSError:
             raise HTTPException(status_code=400, detail="下载地址无法解析")
         addresses = [ipaddress.ip_address(info[4][0].split("%", 1)[0]) for info in infos]
-    if not addresses or any(not address.is_global for address in addresses):
+    if not addresses or any(not address.is_global or address.is_multicast
+                            or (getattr(address, 'ipv4_mapped', None) is not None
+                                and address.ipv4_mapped.is_multicast) for address in addresses):
         raise HTTPException(status_code=400, detail="下载地址解析到非公网地址")
     return host
 
@@ -92,7 +95,7 @@ async def _send_with_validated_redirects(client, method: str, url: str, headers=
         if response.status_code not in (301, 302, 303, 307, 308):
             return response
         location = response.headers.get("location")
-        await response.aclose()
+        await close_redirect_response(response)
         if not location or redirect_count >= max_redirects:
             raise HTTPException(status_code=502, detail="下载源重定向无效或次数过多")
         current_url = urljoin(current_url, location)
@@ -167,10 +170,11 @@ class ProxyDownloadRequest(BaseModel):
         # 检查是否是私有IP
         try:
             ip = ipaddress.ip_address(parsed.hostname)
-            if ip.is_private or ip.is_loopback or ip.is_reserved:
-                raise ValueError('不允许访问内网地址')
         except ValueError:
-            pass  # hostname不是IP，允许继续
+            return v  # 域名仍由实际请求的逐跳DNS与传输层校验。
+        # 拒绝异常必须在解析异常处理之外，否则会被当作域名吞掉。
+        if ip.is_private or ip.is_loopback or ip.is_reserved:
+            raise ValueError('不允许访问内网地址')
         
         return v
 

router_tasks.py

@@ -16,7 +16,7 @@ from rate_limit import shared_limit
 from sqlalchemy.orm import Session
 from models import TaskCreate, TaskUpdate, TaskSubmit, TaskDispute, TaskDisputeResponse  # 🔧 P2修复：提交/申诉/回应端点改为 body 模型，匹配前端 JSON 提交
 import 数据库连接 as db
-from comment_store import append_comment, increment_comment_count, attach_comment_counts
+from comment_store import COMMENT_MAX_LENGTH, append_comment, increment_comment_count, attach_comment_counts, read_comment_view
 from 安全认证 import require_auth, optional_auth, is_admin
 from notifications import add_notification
 from task_publication import freeze_task, project_publications, InsufficientTaskBalance
@@ -932,6 +932,8 @@ async def tip_task(request: Request, task_id: str, amount: int, is_anon: bool =
             Transaction.tx_type == "TIP_OUT",
             Transaction.amount == -amount,
             Transaction.related_account == publisher,
+            Transaction.item_id == task_id,
+            Transaction.item_type == "task",
             Transaction.created_at >= recent_cutoff
         ).with_for_update().first()
         if duplicate_tx:
@@ -1060,29 +1062,8 @@ async def tip_task(request: Request, task_id: str, amount: int, is_anon: bool =
 
 @router.get("/api/tasks/{task_id}/comments")
 async def get_task_comments(task_id: str):
-    """
-    获取任务评论
-    """
-    comments_db = db.load_data("comments.json", default_data={})
-    users_db = db.load_data("users.json", default_data={})
-    
-    # users_db 已经是 {account: user_info} 格式，直接使用
-    user_map = users_db
-    
-    # comments_db 是 {item_id: [comments]} 格式
-    task_comments = comments_db.get(task_id, [])
-    
-    # 附加用户信息
-    result = []
-    for c in task_comments:
-        author_info = user_map.get(c.get("author"), {})
-        result.append({
-            **c,
-            "author_name": author_info.get("name", c.get("author")),
-            "author_avatar": author_info.get("avatarDataUrl", "")
-        })
-    
-    return {"status": "success", "data": result}
+    """获取评论；有界后台读取并保留原作者附加字段。"""
+    return {"status": "success", "data": await read_comment_view(task_id)}
 
 @router.post("/api/tasks/{task_id}/comments")
 @shared_limit(10)
@@ -1092,6 +1073,8 @@ async def add_task_comment(request: Request, task_id: str, content: str, current
     """
     if not content or not content.strip():
         raise HTTPException(status_code=400, detail="评论内容不能为空")
+    if len(content) > COMMENT_MAX_LENGTH:
+        raise HTTPException(status_code=400, detail="评论长度必须在1-5000字符之间")
     
     tasks_db = db.load_data("tasks.json", default_data=[])
     

router_users_profile.py

@@ -15,10 +15,11 @@
 from fastapi import APIRouter, HTTPException, Depends, Request
 from rate_limit import shared_limit
 import 数据库连接 as db
-from models import UserUpdate
+from models import UserUpdate, UserSettings
 from 安全认证 import require_auth, optional_auth
-from user_store import update_profile, project_profile, read_user, attach_profile_counts
+from user_store import update_settings, update_profile, project_profile, read_user, attach_profile_counts
 import logging
+from profile_io import run_profile_io
 
 logger = logging.getLogger(__name__)
 
@@ -49,6 +50,10 @@ async def get_user_profile(account: str, current_user: str = Depends(optional_au
         - receivedFavorites: 收到的总收藏数
         - receivedUses: 发布内容被使用的总次数
     """
+    return await run_profile_io(_read_profile_response, account, current_user)
+
+
+def _read_profile_response(account, current_user):
     user_data = read_user(account)
     # 作品数据读取失败不当作空列表清零既有计数，也不阻断有效资料。
     try:
@@ -101,5 +106,20 @@ async def update_user_profile(request: Request, account: str, update_data: UserU
         raise HTTPException(status_code=400, detail="个人介绍不能超过100个字符")
     dump = getattr(update_data, "model_dump", None) or update_data.dict
     # 读取、账号存在性检查、字段更新及保存均在同一文件锁内。
-    user = update_profile(account, dump(exclude_unset=True))
+    user = await run_profile_io(update_profile, account, dump(exclude_unset=True))
+    return {"status": "success", "data": user}
+
+
+@router.put("/api/users/{account}/settings")
+@shared_limit(10)
+async def update_user_settings(request: Request, account: str, update_data: UserSettings, current_user: str = Depends(require_auth)):
+    """一次用户文件发布共同保存资料和完整隐私设置；旧独立接口保持。"""
+    if account != current_user:
+        raise HTTPException(status_code=403, detail="只能更新自己的资料")
+    if update_data.intro and len(update_data.intro) > 100:
+        raise HTTPException(status_code=400, detail="个人介绍不能超过100个字符")
+    dump = getattr(update_data, "model_dump", None) or update_data.dict
+    payload = dump(exclude_unset=True)
+    privacy = payload.pop("privacy")
+    user = await run_profile_io(update_settings, account, payload, privacy)
     return {"status": "success", "data": user}

router_wallet.py

@@ -354,6 +354,16 @@ async def tip_user(request: Request, req: TipRequest, current_user: str = Depend
     # 🔒 P1并发安全：使用悲观锁+异常处理+事务回滚防止双花问题
     try:
         sender_wallet, target_wallet = lock_tip_wallets(db, req.sender_account, req.target_account)
+        # 如果有关联资源，获取资源信息
+        item_title = None
+        item_type = None
+        if req.item_id:
+            items_db = json_db.load_data("items.json", default_data=[])
+            item = next((i for i in items_db if i["id"] == req.item_id), None)
+            if item:
+                item_title = item.get("title")
+                item_type = item.get("type")
+        
         # 🔒 P1幂等性防护：检查最近30秒内是否存在相同交易
         recent_cutoff = datetime.datetime.utcnow() - datetime.timedelta(seconds=30)
         duplicate_tx = db.query(Transaction).filter(
@@ -361,6 +371,8 @@ async def tip_user(request: Request, req: TipRequest, current_user: str = Depend
             Transaction.tx_type == "TIP_OUT",
             Transaction.amount == -req.amount,
             Transaction.related_account == req.target_account,
+            Transaction.item_id == req.item_id,
+            Transaction.item_type == item_type,
             Transaction.created_at >= recent_cutoff
         ).with_for_update().first()
         if duplicate_tx:
@@ -391,16 +403,6 @@ async def tip_user(request: Request, req: TipRequest, current_user: str = Depend
         target_user_name = target_user_info.get("name", req.target_account)
         sender_user_name = sender_user_info.get("name", req.sender_account)
         
-        # 如果有关联资源，获取资源信息
-        item_title = None
-        item_type = None
-        if req.item_id:
-            items_db = json_db.load_data("items.json", default_data=[])
-            item = next((i for i in items_db if i["id"] == req.item_id), None)
-            if item:
-                item_title = item.get("title")
-                item_type = item.get("type")
-        
         # 发送方交易记录 (字段名为 related_account，与 models_sql.py 中 Transaction 模型保持一致)
         tx_sender = Transaction(
             tx_id=tx_id_sender, account=req.sender_account, tx_type="TIP_OUT", amount=-req.amount,

safe_http_transport.py

@@ -28,7 +28,9 @@ async def _resolve_public_ips(host: str, port: int) -> tuple[str, ...]:
             for record in records
         ]
 
-    if not addresses or any(not address.is_global for address in addresses):
+    if not addresses or any(not address.is_global or address.is_multicast
+                            or (getattr(address, 'ipv4_mapped', None) is not None
+                                and address.ipv4_mapped.is_multicast) for address in addresses):
         raise httpx.ConnectError("目标域名解析到非公网地址")
     return tuple(dict.fromkeys(str(address) for address in addresses))
 

script_execution.py

@@ -0,0 +1,160 @@
+"""管理脚本的有界输出、每进程准入及所属线程终态等待。"""
+import asyncio
+import logging
+import os
+import subprocess
+import sys
+import threading
+import time
+
+from owned_thread import run_owned_thread
+
+
+class ScriptExecutionBusy(Exception):
+    pass
+
+
+_SCRIPT_TIMEOUT = 30
+_OUTPUT_LIMIT = 64 * 1024
+_READ_CHUNK = 16 * 1024
+logger = logging.getLogger("ComfyUI-Ranking.AdminScripts")
+
+
+def _pipe_reader(stream):
+    descriptor = stream.fileno()
+    if os.name == "nt":
+        import ctypes
+        from ctypes import wintypes
+        import msvcrt
+        peek = ctypes.WinDLL("kernel32", use_last_error=True).PeekNamedPipe
+        peek.argtypes = [wintypes.HANDLE, wintypes.LPVOID, wintypes.DWORD,
+                         ctypes.POINTER(wintypes.DWORD), ctypes.POINTER(wintypes.DWORD),
+                         ctypes.POINTER(wintypes.DWORD)]
+        peek.restype = wintypes.BOOL
+        handle = msvcrt.get_osfhandle(descriptor)
+
+        def read():
+            available = wintypes.DWORD()
+            if not peek(handle, None, 0, None, ctypes.byref(available), None):
+                error = ctypes.get_last_error()
+                if error in (109, 232):
+                    return b""
+                raise ctypes.WinError(error)
+            if not available.value:
+                return None
+            return os.read(descriptor, min(_READ_CHUNK, available.value))
+    else:
+        os.set_blocking(descriptor, False)
+
+        def read():
+            try:
+                return os.read(descriptor, _READ_CHUNK)
+            except BlockingIOError:
+                return None
+    return read
+
+
+def _run_script(script_path, *, cwd, env):
+    command = [sys.executable, script_path]
+    deadline = time.monotonic() + _SCRIPT_TIMEOUT
+    process = subprocess.Popen(
+        command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        bufsize=0, cwd=cwd, env=env,
+        creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0,
+    )
+    streams = (process.stdout, process.stderr)
+    buffers = [bytearray(), bytearray()]
+    truncated = [False, False]
+    try:
+        readers = [_pipe_reader(stream) for stream in streams]
+        active = {0, 1}
+        while active or process.poll() is None:
+            if time.monotonic() >= deadline:
+                raise subprocess.TimeoutExpired(command, _SCRIPT_TIMEOUT)
+            received = False
+            for index in tuple(active):
+                chunk = readers[index]()
+                if chunk is None:
+                    continue
+                if not chunk:
+                    active.remove(index)
+                    continue
+                received = True
+                remaining = _OUTPUT_LIMIT - len(buffers[index])
+                buffers[index].extend(chunk[:remaining])
+                truncated[index] |= len(chunk) > remaining
+            if not received:
+                time.sleep(min(.01, max(0, deadline - time.monotonic())))
+        outputs = [buffer.decode("utf-8", errors="replace").replace("\r\n", "\n").replace("\r", "\n") + ("\n...（输出已截断）" if cut else "")
+                   for buffer, cut in zip(buffers, truncated)]
+        return subprocess.CompletedProcess(command, process.returncode, *outputs)
+    finally:
+        try:
+            if process.poll() is None:
+                process.kill()
+        finally:
+            try:
+                process.wait()
+            finally:
+                for stream in streams:
+                    try:
+                        stream.close()
+                    except Exception as error:
+                        logger.error("ADMIN_SCRIPT_PIPE_CLOSE_ERROR | error_type=%s", type(error).__name__)
+
+
+class ScriptExecution:
+    def __init__(self):
+        self._gate = threading.Lock()
+        self._accepting = True
+        self._pending = False
+        self._closers = 0
+        self._drained = threading.Event()
+        self._drained.set()
+
+    def start(self):
+        with self._gate:
+            if self._pending or self._closers:
+                raise RuntimeError("script work is still pending")
+            self._accepting = True
+
+    def stop(self):
+        with self._gate:
+            self._accepting = False
+
+    async def close(self):
+        with self._gate:
+            self._accepting = False
+            self._closers += 1
+        cancelled = None
+        try:
+            while not self._drained.is_set():
+                try:
+                    await asyncio.sleep(.01)
+                except asyncio.CancelledError as error:
+                    cancelled = error
+            if cancelled is not None:
+                raise cancelled
+        finally:
+            with self._gate:
+                self._closers -= 1
+
+    async def run(self, script_path, *, cwd, env):
+        with self._gate:
+            if not self._accepting or self._pending:
+                raise ScriptExecutionBusy()
+            self._pending = True
+            self._drained.clear()
+        try:
+            return await run_owned_thread(_run_script, script_path, cwd=cwd, env=env)
+        finally:
+            with self._gate:
+                self._pending = False
+                self._drained.set()
+
+
+_script_execution = ScriptExecution()
+run_admin_script_process = _script_execution.run
+start_admin_scripts = _script_execution.start
+stop_admin_scripts = _script_execution.stop
+close_admin_scripts = _script_execution.close

task_projection.py

@@ -1,5 +1,7 @@
 """任务状态恢复逐条隔离，并用轮转游标避免坏记录占满每一批。"""
 import asyncio
+import threading
+from owned_thread import run_owned_thread
 import logging
 from sqlalchemy.orm import Session
 
@@ -37,7 +39,7 @@ def report_pending(stage, error, task_id=None):
         pass
 
 
-def recover_stage(bind, store, model, project, after=None, limit=50, should_stop=lambda: False):
+def recover_stage(bind, store, model, project, after=None, limit=50, should_stop=lambda: False, *, on_uncertain=None):
     """只读选取独立任务；单条故障不确认、不删除，也不阻断下一条。"""
     if type(limit) is not int or not 1 <= limit <= 500:
         raise ValueError('invalid projection batch limit')
@@ -54,6 +56,12 @@ def recover_stage(bind, store, model, project, after=None, limit=50, should_stop
         try:
             count += project(bind, store, limit=1, task_id=task_id)
         except Exception as error:
+            # 投影可先发布JSON再在SQL确认/同步处失败；零计数不代表未写入。
+            if on_uncertain is not None:
+                try:
+                    on_uncertain()
+                except Exception as notification_error:
+                    report_pending("sort_cache_notification", notification_error)
             report_pending(model.__tablename__, error, task_id)
         # 失败也推进游标；下一轮走后续任务，到末尾再回头重试。
         cursor = task_id
@@ -67,29 +75,50 @@ async def recovery_worker(bind, store, invalidate, stop, interval=30):
     repair_cursor = None
     while not stop.is_set():
         count = 0
-        for model, project in stages:
-            if stop.is_set():
-                break
-            try:
-                recovered, cursor = await asyncio.to_thread(
-                    recover_stage, bind, store, model, project, cursors.get(model), 50, stop.is_set)
-                cursors[model] = cursor
-                count += recovered
-            except Exception as error:
-                # 表/连接/批次读取故障仅影响该阶段；之后的阶段仍可恢复。
-                report_pending(model.__tablename__, error)
-        if not stop.is_set():
-            try:
-                recovered, repair_cursor = await asyncio.to_thread(
-                    recover_snapshots, bind, store, repair_cursor, 50, stop.is_set)
-                count += recovered
-            except Exception as error:
-                report_pending('task_snapshots', error)
-        if count:
+        attempted = threading.Event()
+        changed_data = threading.Event()
+        uncertain = threading.Event()
+        cancelled = False
+
+        def recover(function, *args):
+            attempted.set()
             try:
-                invalidate('tasks:')
-            except Exception as error:
-                report_pending('sort_cache', error)
+                result = function(*args, on_uncertain=uncertain.set)
+            except Exception:
+                uncertain.set()
+                raise
+            if result[0]:
+                changed_data.set()
+            return result
+
+        try:
+            for model, project in stages:
+                if stop.is_set():
+                    break
+                try:
+                    recovered, cursor = await run_owned_thread(
+                        recover, recover_stage, bind, store, model, project, cursors.get(model), 50, stop.is_set)
+                    cursors[model] = cursor
+                    count += recovered
+                except Exception as error:
+                    # 表/连接/批次读取故障仅影响该阶段；之后的阶段仍可恢复。
+                    report_pending(model.__tablename__, error)
+            if not stop.is_set():
+                try:
+                    recovered, repair_cursor = await run_owned_thread(
+                        recover, recover_snapshots, bind, store, repair_cursor, 50, stop.is_set)
+                    count += recovered
+                except Exception as error:
+                    report_pending('task_snapshots', error)
+        except asyncio.CancelledError:
+            cancelled = True
+            raise
+        finally:
+            if count or changed_data.is_set() or uncertain.is_set() or (cancelled and attempted.is_set()):
+                try:
+                    invalidate('tasks:')
+                except Exception as error:
+                    report_pending('sort_cache', error)
         if stop.is_set():
             break
         try:

task_repair.py

@@ -1,106 +1,112 @@
-"""定期核对已确认任务投影：仅修复有SQL事实的字段，不重新执行资金操作。"""
-import copy
-from sqlalchemy.orm import Session
-from models_sql import TaskDisputeRecord, TaskDisputeResponseRecord, TaskResolution
-from task_read import MODELS, merge_task_source
-from task_dispute_read import merged_dispute
-from task_projection import report_pending
-from tip_transaction import lock_wallet_accounts
-
-
-TASK_FIELDS = ('title', 'description', 'reference_images', 'reference_link', 'total_price',
-               'frozen_amount', 'deposit_ratio', 'deposit_amount', 'deadline', 'created_at',
-               'status', 'assignee', 'assigned_at', 'deposit_paid', 'deliverables', 'submit_note',
-               'submitted_at', 'feedback', 'completed_at', 'dispute_id', 'refunded', 'refund_amount')
-DISPUTE_FIELDS = ('task_title', 'initiator', 'initiator_role', 'reason', 'evidence', 'created_at',
-                  'status', 'response', 'response_evidence', 'responded_at', 'resolution',
-                  'resolution_ratio', 'resolution_note', 'resolved_by', 'resolved_at')
-
-
-def source_record(store, filename, identifier):
-    source = store.load_data(filename, default_data=[], strict=True, skip_cache=True)
-    if (not isinstance(source, list) or any(not isinstance(r, dict) or not isinstance(r.get('id'), str)
-            or not r['id'] for r in source) or len({r['id'] for r in source}) != len(source)):
-        raise ValueError('invalid repair source')
-    return next((r for r in source if r['id'] == identifier), None)
-
-
-def repair_record(store, filename, canonical, expected, fields):
-    identifier = canonical['id']
-    patch = {k: copy.deepcopy(canonical[k]) for k in fields if k in canonical}
-    if expected is not None and all(k in expected and expected[k] == v for k, v in patch.items()):
-        return 0
-    def update(records):
-        if (not isinstance(records, list) or any(not isinstance(r, dict) or not isinstance(r.get('id'), str)
-                or not r['id'] for r in records) or len({r['id'] for r in records}) != len(records)):
-            raise ValueError('invalid atomic repair source')
-        current = next((r for r in records if r['id'] == identifier), None)
-        if expected is None:
-            if current is not None:
-                raise ValueError('record appeared during repair')
-            records.append(copy.deepcopy(canonical))
-        else:
-            if (current is None or any(current.get(k) != expected.get(k) or (k in current) != (k in expected)
-                    for k in ('publisher', 'assignee', 'task_id', *fields))):
-                raise ValueError('record changed during repair')
-            current.update(patch)  # 报名、互动和扩展字段保留回调中最新值。
-    store.atomic_update(filename, update, default_data=[])
-    return 1
-
-
-def repair_task(bind, store, task_id):
-    with Session(bind=bind, autoflush=False) as session:
-        try:
-            owner = None
-            for model in MODELS:
-                row = session.query(model).filter(model.task_id == task_id).first()
-                if row is not None:
-                    owner = row.publisher
-                    break
-            if owner is None:
-                raise ValueError('durable task missing')
-            # 与编辑/交付/资金提交串行，读取与文件写入之间不会发生新SQL版本。
-            lock_wallet_accounts(session, (owner,))
-            expected = source_record(store, 'tasks.json', task_id)
-            tasks = merge_task_source(session, [] if expected is None else [expected], task_id)
-            if len(tasks) != 1 or tasks[0].get('publisher') != owner:
-                raise ValueError('repair task identity mismatch')
-            canonical = tasks[0]
-            count = repair_record(store, 'tasks.json', canonical, expected, TASK_FIELDS)
-            durable_dispute = None
-            for model in (TaskDisputeRecord, TaskDisputeResponseRecord, TaskResolution):
-                row = session.query(model).filter(model.task_id == task_id).populate_existing().one_or_none()
-                if row is not None:
-                    if durable_dispute is not None and durable_dispute != row.dispute_id:
-                        raise ValueError('conflicting repair disputes')
-                    durable_dispute = row.dispute_id
-            if durable_dispute is not None:
-                expected = source_record(store, 'disputes.json', durable_dispute)
-                private = merged_dispute(session, [] if expected is None else [expected], durable_dispute, None, True)
-                if private.get('task_id') != task_id or private.get('publisher') != owner:
-                    raise ValueError('repair dispute identity mismatch')
-                count += repair_record(store, 'disputes.json', private, expected, DISPUTE_FIELDS)
-            return count
-        finally:
-            session.rollback()  # 只释放协调锁；不改变余额、流水或projected。
-
-
-def recover_snapshots(bind, store, after=None, limit=50, should_stop=lambda: False):
-    if type(limit) is not int or not 1 <= limit <= 500:
-        raise ValueError('invalid repair batch limit')
-    with Session(bind=bind, autoflush=False) as session:
-        query = session.query(MODELS[0].task_id).union(*(session.query(m.task_id) for m in MODELS[1:]))
-        selected = query.filter(MODELS[0].task_id > after) if after is not None else query
-        pending = selected.order_by(MODELS[0].task_id).limit(limit).all()
-        if not pending and after is not None:
-            pending = query.order_by(MODELS[0].task_id).limit(limit).all()
-    count, cursor = 0, after
-    for (task_id,) in pending:
-        if should_stop():
-            break
-        try:
-            count += repair_task(bind, store, task_id)
-        except Exception as error:
-            report_pending('task_snapshots', error, task_id)
-        cursor = task_id
-    return count, cursor if pending else None
+"""定期核对已确认任务投影：仅修复有SQL事实的字段，不重新执行资金操作。"""
+import copy
+from sqlalchemy.orm import Session
+from models_sql import TaskDisputeRecord, TaskDisputeResponseRecord, TaskResolution
+from task_read import MODELS, merge_task_source
+from task_dispute_read import merged_dispute
+from task_projection import report_pending
+from tip_transaction import lock_wallet_accounts
+
+
+TASK_FIELDS = ('title', 'description', 'reference_images', 'reference_link', 'total_price',
+               'frozen_amount', 'deposit_ratio', 'deposit_amount', 'deadline', 'created_at',
+               'status', 'assignee', 'assigned_at', 'deposit_paid', 'deliverables', 'submit_note',
+               'submitted_at', 'feedback', 'completed_at', 'dispute_id', 'refunded', 'refund_amount')
+DISPUTE_FIELDS = ('task_title', 'initiator', 'initiator_role', 'reason', 'evidence', 'created_at',
+                  'status', 'response', 'response_evidence', 'responded_at', 'resolution',
+                  'resolution_ratio', 'resolution_note', 'resolved_by', 'resolved_at')
+
+
+def source_record(store, filename, identifier):
+    source = store.load_data(filename, default_data=[], strict=True, skip_cache=True)
+    if (not isinstance(source, list) or any(not isinstance(r, dict) or not isinstance(r.get('id'), str)
+            or not r['id'] for r in source) or len({r['id'] for r in source}) != len(source)):
+        raise ValueError('invalid repair source')
+    return next((r for r in source if r['id'] == identifier), None)
+
+
+def repair_record(store, filename, canonical, expected, fields):
+    identifier = canonical['id']
+    patch = {k: copy.deepcopy(canonical[k]) for k in fields if k in canonical}
+    if expected is not None and all(k in expected and expected[k] == v for k, v in patch.items()):
+        return 0
+    def update(records):
+        if (not isinstance(records, list) or any(not isinstance(r, dict) or not isinstance(r.get('id'), str)
+                or not r['id'] for r in records) or len({r['id'] for r in records}) != len(records)):
+            raise ValueError('invalid atomic repair source')
+        current = next((r for r in records if r['id'] == identifier), None)
+        if expected is None:
+            if current is not None:
+                raise ValueError('record appeared during repair')
+            records.append(copy.deepcopy(canonical))
+        else:
+            if (current is None or any(current.get(k) != expected.get(k) or (k in current) != (k in expected)
+                    for k in ('publisher', 'assignee', 'task_id', *fields))):
+                raise ValueError('record changed during repair')
+            current.update(patch)  # 报名、互动和扩展字段保留回调中最新值。
+    store.atomic_update(filename, update, default_data=[])
+    return 1
+
+
+def repair_task(bind, store, task_id):
+    with Session(bind=bind, autoflush=False) as session:
+        try:
+            owner = None
+            for model in MODELS:
+                row = session.query(model).filter(model.task_id == task_id).first()
+                if row is not None:
+                    owner = row.publisher
+                    break
+            if owner is None:
+                raise ValueError('durable task missing')
+            # 与编辑/交付/资金提交串行，读取与文件写入之间不会发生新SQL版本。
+            lock_wallet_accounts(session, (owner,))
+            expected = source_record(store, 'tasks.json', task_id)
+            tasks = merge_task_source(session, [] if expected is None else [expected], task_id)
+            if len(tasks) != 1 or tasks[0].get('publisher') != owner:
+                raise ValueError('repair task identity mismatch')
+            canonical = tasks[0]
+            count = repair_record(store, 'tasks.json', canonical, expected, TASK_FIELDS)
+            durable_dispute = None
+            for model in (TaskDisputeRecord, TaskDisputeResponseRecord, TaskResolution):
+                row = session.query(model).filter(model.task_id == task_id).populate_existing().one_or_none()
+                if row is not None:
+                    if durable_dispute is not None and durable_dispute != row.dispute_id:
+                        raise ValueError('conflicting repair disputes')
+                    durable_dispute = row.dispute_id
+            if durable_dispute is not None:
+                expected = source_record(store, 'disputes.json', durable_dispute)
+                private = merged_dispute(session, [] if expected is None else [expected], durable_dispute, None, True)
+                if private.get('task_id') != task_id or private.get('publisher') != owner:
+                    raise ValueError('repair dispute identity mismatch')
+                count += repair_record(store, 'disputes.json', private, expected, DISPUTE_FIELDS)
+            return count
+        finally:
+            session.rollback()  # 只释放协调锁；不改变余额、流水或projected。
+
+
+def recover_snapshots(bind, store, after=None, limit=50, should_stop=lambda: False, *, on_uncertain=None):
+    if type(limit) is not int or not 1 <= limit <= 500:
+        raise ValueError('invalid repair batch limit')
+    with Session(bind=bind, autoflush=False) as session:
+        query = session.query(MODELS[0].task_id).union(*(session.query(m.task_id) for m in MODELS[1:]))
+        selected = query.filter(MODELS[0].task_id > after) if after is not None else query
+        pending = selected.order_by(MODELS[0].task_id).limit(limit).all()
+        if not pending and after is not None:
+            pending = query.order_by(MODELS[0].task_id).limit(limit).all()
+    count, cursor = 0, after
+    for (task_id,) in pending:
+        if should_stop():
+            break
+        try:
+            count += repair_task(bind, store, task_id)
+        except Exception as error:
+            # tasks写入后，disputes修复或释放SQL锁仍可能失败。
+            if on_uncertain is not None:
+                try:
+                    on_uncertain()
+                except Exception as notification_error:
+                    report_pending('sort_cache_notification', notification_error)
+            report_pending('task_snapshots', error, task_id)
+        cursor = task_id
+    return count, cursor if pending else None

user_store.py

@@ -77,6 +77,24 @@ def update_profile(account, updates):
     return _mutate(updater)
 
 
+def update_settings(account, updates, privacy):
+    # 联合设置必须完整提交四个严格布尔值，先验证再进入发布操作。
+    if not isinstance(privacy, dict) or set(privacy) != PRIVACY_FIELDS or any(type(value) is not bool for value in privacy.values()):
+        raise HTTPException(status_code=400, detail="隐私设置不正确")
+    settings = copy.deepcopy(privacy)
+    patch = copy.deepcopy({key: value for key, value in updates.items()
+                           if key in EDITABLE_FIELDS and value is not None})
+    if isinstance(patch.get("avatarDataUrl"), str) and patch["avatarDataUrl"].startswith("data:"):
+        del patch["avatarDataUrl"]
+
+    def updater(data):
+        record = _user(data, account)
+        record.update(patch)
+        record["privacy"] = settings
+        return project_profile(record, owner=True)
+    return _mutate(updater)
+
+
 def update_user_privacy(account, privacy):
     # 保持 PrivacySettings 的完整替换语义，而非隐式合并部分设置。
     if not isinstance(privacy, dict) or any(key not in PRIVACY_FIELDS or type(value) is not bool

zip_validation.py

@@ -105,7 +105,8 @@ def validate_zip_archive(path, size, *, max_directory_bytes=MAX_DIRECTORY_BYTES,
             expanded = 0
             for entry in entries:
                 check()
-                if (entry.flag_bits & 1 or entry.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED)
+                # 普通/强加密及压缩补丁均沿用ZipFile.open的拒绝契约。
+                if (entry.flag_bits & (1 | 32 | 64) or entry.compress_type not in (zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED)
                         or not 0 <= entry.file_size <= max_member_bytes or entry.compress_size < 0): _fail()
                 expanded += entry.file_size
                 if expanded > min(max_expanded_bytes, size * 10): _fail()

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

@@ -191,16 +191,18 @@ async def _delete_old_zip_cache(item_id: str, old_hash: str, entry=None):
         storage_repo = archive_repository(db.DATASET_REPO_ID)
         path = _stored_archive_path(item_id, old_hash, entry, repo_id=storage_repo)
         hf_api = HfApi(token=hf_token)
-        await asyncio.to_thread(hf_api.delete_file, path_in_repo=path, repo_id=storage_repo, repo_type="dataset")
+        await run_owned_thread(hf_api.delete_file, path_in_repo=path, repo_id=storage_repo, repo_type="dataset")
     except Exception as error:
         logger.warning("旧缓存清理失败: error_type=%s", type(error).__name__)
 
 
-async def cleanup_expired_cache(versions_db: dict):
+async def cleanup_expired_cache(versions_db: dict, stop=None):
     """清理过期归档；仅撤销仍对应本次快照的元数据，不覆盖并发新缓存。"""
     try:
         now = datetime.datetime.now(datetime.timezone.utc)
         for item_id, entry in list(versions_db.items()):
+            if stop is not None and stop.is_set():
+                return
             if not isinstance(entry, dict) or not entry.get('cached_at'):
                 continue
             try:
@@ -222,7 +224,7 @@ async def cleanup_expired_cache(versions_db: dict):
                 current[item_id] = {key: value for key, value in expected.items()
                     if key not in ('cached_at', 'zip_size') and not key.startswith('archive_')}
                 return True
-            changed = await asyncio.to_thread(db.atomic_update, 'versions.json', update, default_data={})
+            changed = await run_owned_thread(db.atomic_update, 'versions.json', update, default_data={})
             if changed:
                 versions_db[item_id] = {key: value for key, value in expected.items()
                     if key not in ('cached_at', 'zip_size') and not key.startswith('archive_')}
@@ -230,11 +232,24 @@ async def cleanup_expired_cache(versions_db: dict):
         logger.warning("缓存过期清理失败: error_type=%s", type(error).__name__)
 
 
-async def daily_version_check_task():
+async def _wait_periodic_stop(stop, seconds):
+    if stop is None:
+        await asyncio.sleep(seconds)
+        return False
+    if stop.is_set():
+        return True
+    try:
+        await asyncio.wait_for(stop.wait(), timeout=seconds)
+        return True
+    except asyncio.TimeoutError:
+        return stop.is_set()
+
+
+async def daily_version_check_task(stop=None):
     """每日 02:00 定时执行的守护进程"""
     cache_cursor = 0
-    while True:
-        now = datetime.datetime.utcnow() + datetime.timedelta(hours=8)
+    while stop is None or not stop.is_set():
+        now = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=8)
         # 计算距离今天凌晨 02:00 的秒数，如果过了就定在明天 02:00
         next_run = now.replace(hour=2, minute=0, second=0, microsecond=0)
         if now >= next_run:
@@ -242,32 +257,41 @@ async def daily_version_check_task():
         sleep_seconds = (next_run - now).total_seconds()
         
         print(f"🕒 版本探测引擎已挂载，距离下次 02:00 扫描还有 {sleep_seconds} 秒...")
-        await asyncio.sleep(sleep_seconds)
+        if await _wait_periodic_stop(stop, sleep_seconds):
+            return
         
         print("🚀 [02:00] 开始执行全站 GitHub 资源版本检测任务...")
-        items_db = await asyncio.to_thread(db.load_data, "items.json", default_data=[], strict=True, skip_cache=True)
-        versions_db = await asyncio.to_thread(db.load_data, "versions.json", default_data={}, strict=True, skip_cache=True)
+        items_db = await run_owned_thread(db.load_data, "items.json", default_data=[], strict=True, skip_cache=True)
+        if stop is not None and stop.is_set():
+            return
+        versions_db = await run_owned_thread(db.load_data, "versions.json", default_data={}, strict=True, skip_cache=True)
+        if stop is not None and stop.is_set():
+            return
         
         # 先清理过期缓存（cached_at 超过 30 天）
-        await cleanup_expired_cache(versions_db)
+        await cleanup_expired_cache(versions_db, stop=stop)
         
         updated_count = 0
         updated_items = []  # 记录有更新的插件信息
         
         for item in items_db:
+            if stop is not None and stop.is_set():
+                return
             link = item.get("link", "")
             if "github.com" in link:
                 try:
                     refreshed = await refresh_resource_version(db, item['id'])
                     if not refreshed:
                         continue
+                    if stop is not None and stop.is_set():
+                        return
                     current = refreshed['item']
                     token = refreshed['token']
                     latest_hash = refreshed['commit']
                     if refreshed['changed']:
                         updated_count += 1
                     # 每项刷新后再读最新缓存，不持有全站快照执行保存。
-                    current_versions = await asyncio.to_thread(db.load_data, "versions.json", default_data={}, strict=True, skip_cache=True)
+                    current_versions = await run_owned_thread(db.load_data, "versions.json", default_data={}, strict=True, skip_cache=True)
                     if cached_archive(current, token, current_versions.get(item['id']), repo_id=db.DATASET_REPO_ID) is None:
                         updated_items.append(dict(id=item['id'], link=current['link'], github_token=token, version_hash=latest_hash))
                 except Exception as error:
@@ -282,6 +306,8 @@ async def daily_version_check_task():
             batch = ordered[:PRECACHE_BATCH_SIZE]
             cache_cursor = (start + len(batch)) % len(updated_items)
             for idx, item in enumerate(batch):
+                if stop is not None and stop.is_set():
+                    return
                 success = await precache_github_zip(
                     item["link"],
                     item.get("github_token"),
@@ -292,24 +318,24 @@ async def daily_version_check_task():
                     precache_count += 1
                 # 间隔 10 秒（最后一个不需要等待）
                 if idx < len(batch) - 1:
-                    await asyncio.sleep(PRECACHE_INTERVAL)
+                    if await _wait_periodic_stop(stop, PRECACHE_INTERVAL):
+                        return
             logger.info(
                 f"[预缓存] 本次预缓存完成: {precache_count}/{len(batch)} 成功"
             )
 
 
-async def reset_daily_views_task():
+async def reset_daily_views_task(stop=None):
     """
     每日重置 daily_views 计数器的守护进程
-    - 每小时检查一次，在 UTC+8 00:00 执行重置
+    - 等待 UTC+8 下一次 00:00，关闭信号可唤醒等待
     - 使用 atomic_update 保证并发安全
     """
     # 需要处理的数据文件
     data_files = ["items.json", "tasks.json", "posts.json"]
     
-    while True:
-        now = datetime.datetime.utcnow() + datetime.timedelta(hours=8)
-        today = now.date().isoformat()  # 使用 UTC+8 日期而非系统本地日期
+    while stop is None or not stop.is_set():
+        now = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=8)
         
         # 计算距离今天 00:00 (UTC+8) 的秒数，如果过了就定在明天 00:00
         next_run = now.replace(hour=0, minute=0, second=0, microsecond=0)
@@ -318,13 +344,18 @@ async def reset_daily_views_task():
         sleep_seconds = (next_run - now).total_seconds()
         
         print(f"📊 daily_views 重置引擎已挂载，距离下次 00:00 执行还有 {sleep_seconds} 秒...")
-        await asyncio.sleep(sleep_seconds)
+        if await _wait_periodic_stop(stop, sleep_seconds):
+            return
         
+        # 等待跨过午夜后重新取日期，不能把入睡前的昨天写回。
+        today = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=8)).date().isoformat()
         print(f"🚀 [00:00] 开始执行 daily_views 每日重置任务 (日期: {today})...")
         
         total_reset_count = 0
         
         for file_name in data_files:
+            if stop is not None and stop.is_set():
+                return
             reset_count = 0
             
             def reset_updater(data):
@@ -344,7 +375,7 @@ async def reset_daily_views_task():
                         reset_count += 1
             
             try:
-                db.atomic_update(file_name, reset_updater, default_data=[])
+                await run_owned_thread(db.atomic_update, file_name, reset_updater, default_data=[])
                 total_reset_count += reset_count
                 if reset_count > 0:
                     print(f"  ✓ {file_name}: 重置了 {reset_count} 条记录的 daily_views")

数据库连接.py

@@ -121,7 +121,7 @@ CACHE_TTL = {
 }
 
 # 内存缓存存储
-_memory_cache = {}  # {file_name: {"data": ..., "time": ..., "mtime": ...}}
+_memory_cache = {}  # {file_name: {"data": ..., "time": ..., "identity": ...}}
 _cache_lock = threading.Lock()
 
 
@@ -130,6 +130,15 @@ def _get_cache_ttl(file_name: str) -> int:
     return CACHE_TTL.get(file_name, CACHE_TTL["_default"])
 
 
+def _file_cache_identity(stat):
+    """本机文件身份与变化提示；不是内容哈希或跨文件事务版本。"""
+    if not stat.st_ino:
+        return None  # 文件系统未提供可靠文件ID，不复用缓存。
+    # Windows的stat/fstat ctime语义存在差异，不跨两种接口比较该字段。
+    changed_ns = stat.st_ctime_ns if os.name != "nt" else None
+    return (stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, changed_ns)
+
+
 def _get_from_cache(file_name: str, local_path: str) -> Optional[Union[Dict, List]]:
     """
     从内存缓存获取数据
@@ -149,34 +158,50 @@ def _get_from_cache(file_name: str, local_path: str) -> Optional[Union[Dict, Lis
             del _memory_cache[file_name]
             return None
         
-        # 检查文件是否被修改（mtime 变化）
+        # 相同mtime的原子替换仍需识别；复制前后都确认路径仍指向同一来源。
         try:
-            current_mtime = os.path.getmtime(local_path)
-            if current_mtime != cache_entry.get("mtime"):
+            identity = _file_cache_identity(os.stat(local_path))
+            if identity is None or identity != cache_entry.get("identity"):
+                del _memory_cache[file_name]
+                return None
+            import copy
+            result = copy.deepcopy(cache_entry["data"])
+            if _file_cache_identity(os.stat(local_path)) != identity:
                 del _memory_cache[file_name]
                 return None
         except OSError:
+            _memory_cache.pop(file_name, None)
             return None
-        
         logger.debug(f"✨ 内存缓存命中: {file_name}")
-        # 返回深拷贝，防止外部修改影响缓存
-        import copy
-        return copy.deepcopy(cache_entry["data"])
+        return result
 
 
-def _set_to_cache(file_name: str, data: Union[Dict, List], local_path: str):
-    """将数据存入内存缓存"""
+def _set_to_cache(file_name: str, data: Union[Dict, List], local_path: str, *, source_stats=None):
+    """仅缓存已由真实读取句柄确认的数据；旧三参数写后调用只使缓存失效。"""
     with _cache_lock:
+        if source_stats is None:
+            _memory_cache.pop(file_name, None)
+            return
+        before, after = source_stats
+        if before is None or after is None:
+            return
+        identity = _file_cache_identity(before)
+        if (identity is None or identity != _file_cache_identity(after)
+                or before.st_ctime_ns != after.st_ctime_ns):
+            return
         try:
-            mtime = os.path.getmtime(local_path) if os.path.exists(local_path) else 0
+            if _file_cache_identity(os.stat(local_path)) != identity:
+                return
+            import copy
+            snapshot = copy.deepcopy(data)
+            if _file_cache_identity(os.stat(local_path)) != identity:
+                return
         except OSError:
-            mtime = 0
-        
-        import copy
+            return
         _memory_cache[file_name] = {
-            "data": copy.deepcopy(data),
+            "data": snapshot,
             "time": time.time(),
-            "mtime": mtime
+            "identity": identity
         }
         logger.debug(f"💾 内存缓存更新: {file_name}")
 
@@ -289,6 +314,40 @@ else:
 #   - 文件不存在时从 HF 下载
 #   - JSON 解析失败时返回默认值
 
+def _publish_read_snapshot(local_path: str, data):
+    """首次下载/非严格备份恢复的原子发布；调用方持有原稳定协调锁。"""
+    descriptor = None
+    staging_path = None
+    try:
+        descriptor, staging_path = tempfile.mkstemp(
+            prefix=f".{os.path.basename(local_path)}.read-", suffix=".tmp",
+            dir=os.path.dirname(local_path))
+        stream = os.fdopen(descriptor, "w", encoding="utf-8")
+        descriptor = None  # fdopen成功后由stream负责关闭。
+        with stream:
+            json.dump(data, stream, ensure_ascii=False, indent=2)
+            stream.flush()
+            os.fsync(stream.fileno())
+        with open(staging_path, "r", encoding="utf-8") as source:
+            verified = json.load(source)
+        if type(verified) is not type(data) or (hasattr(data, "__len__") and len(verified) != len(data)):
+            raise ValueError("Read snapshot verification failed")
+        os.replace(staging_path, local_path)
+    finally:
+        if descriptor is not None:
+            try:
+                os.close(descriptor)
+            except OSError:
+                logger.warning("JSON_READ_DESCRIPTOR_CLEANUP_FAILED")
+        if staging_path is not None:
+            try:
+                os.unlink(staging_path)
+            except FileNotFoundError:
+                pass
+            except OSError:
+                logger.warning("JSON_READ_STAGE_CLEANUP_FAILED")
+
+
 def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None, skip_cache: bool = False, strict: bool = False) -> Union[Dict, List]:
     """
     从 JSON 文件加载数据
@@ -330,9 +389,8 @@ def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None,
                     # 读取下载的文件
                     with open(downloaded_path, "r", encoding="utf-8") as f:
                         data = json.load(f)
-                    # 保存到本地缓存
-                    with open(local_path, "w", encoding="utf-8") as f:
-                        json.dump(data, f, ensure_ascii=False, indent=2)
+                    # 验证同目录暂存后原子发布，失败不留下半份主文件。
+                    _publish_read_snapshot(local_path, data)
                     logger.info(f"从 HF Dataset 下载 {file_name} 成功")
                     # 🚀 P1优化：存入内存缓存
                     _set_to_cache(file_name, data, local_path)
@@ -345,7 +403,7 @@ def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None,
                         logger.info(f"HF 中未找到 {file_name}（尚未创建），使用默认值")
                     else:
                         logger.warning(f"从 HF 下载 {file_name} 失败: {_safe_error_details(e)}")
-                    # 缓存 default_data，避免每次请求都重试下载
+                    # 无确认来源的默认值不入内存缓存，保持后续读取可恢复
                     _set_to_cache(file_name, default_data, local_path)
                     return default_data
             return default_data
@@ -356,12 +414,20 @@ def load_data(file_name: str, default_data: Optional[Union[Dict, List]] = None,
                 # 获取共享锁（读锁）
                 _lock_file(f, exclusive=False)
                 try:
+                    try:
+                        source_before = os.fstat(f.fileno())
+                    except OSError:
+                        source_before = None
                     data = json.load(f)
+                    try:
+                        source_after = os.fstat(f.fileno())
+                    except OSError:
+                        source_after = None
                 finally:
                     _unlock_file(f)
-            
-            # 🚀 P1优化：存入内存缓存
-            _set_to_cache(file_name, data, local_path)
+
+            # 句柄身份绑定实际解析的数据；迟到旧读取不得冒充新路径内容。
+            _set_to_cache(file_name, data, local_path, source_stats=(source_before, source_after))
             return data
         except json.JSONDecodeError as e:
             logger.error(f"JSON 解析错误 {file_name}: {_safe_error_details(e)}")
@@ -385,14 +451,13 @@ def _recover_from_backup(file_name: str, default_data: Union[Dict, List]) -> Uni
         try:
             with open(backup_path, "r", encoding="utf-8") as f:
                 data = json.load(f)
-            print(f"✅ 从备份恢复 {file_name} 成功")
-            # 恢复主文件
+            # 先完整发布，再记录成功；保留原备份，不写回默认值。
             local_path = os.path.join(LOCAL_DB_DIR, file_name)
-            with open(local_path, "w", encoding="utf-8") as f:
-                json.dump(data, f, ensure_ascii=False, indent=2)
+            _publish_read_snapshot(local_path, data)
+            logger.info("从备份恢复 %s 成功", file_name)
             return data
         except Exception as e:
-            print(f"🚨 备份恢复失败 {file_name}: {_safe_error_details(e)}")
+            logger.warning("备份恢复失败 %s: %s", file_name, _safe_error_details(e))
     
     return default_data
 
@@ -404,7 +469,7 @@ def _recover_from_backup(file_name: str, default_data: Union[Dict, List]) -> Uni
 #   1. 先写临时文件，验证成功后原子重命名
 #   2. 写入前备份上一版本
 #   3. 写入后验证数据完整性
-#   4. 🚀 P1优化：保存后更新内存缓存
+#   4. 保存后使旧内存缓存失效，确认读取后重建
 #   5. 异步同步到 HuggingFace
 
 def save_data(file_name: str, data: Union[Dict, List], force_sync: bool = False) -> bool:
@@ -473,7 +538,7 @@ def save_data(file_name: str, data: Union[Dict, List], force_sync: bool = False)
                 _mark_dirty(file_name)
             os.replace(temp_path, local_path)
             
-            # ========== 🚀 P1优化：更新内存缓存（在锁内，确保写后立即可见） ==========
+            # ========== 写后使旧内存缓存失效；下一次句柄确认读取才缓存 ==========
             _set_to_cache(file_name, data, local_path)
             
         except Exception as e:
@@ -902,7 +967,7 @@ def atomic_update(file_name: str, updater, default_data=None, force_sync: bool =
                 _mark_dirty(file_name)
             os.replace(temp_path, local_path)
             
-            # ========== 更新内存缓存（在锁内，确保写后立即可见） ==========
+            # ========== 写后使旧内存缓存失效；下一次句柄确认读取才缓存 ==========
             _set_to_cache(file_name, data, local_path)
             
         except Exception as e: