modify: diskcache 缓存替换

src/chongming/__init__.py

@@ -8,7 +8,6 @@ def serve():
     env = config["default"]["env"]
     server_config = config[env]["server"]
     log_config = config[env]["logging"]
-    setup_logging(log_config)
     logget = get_logger("app")
     logget.info(f"启动服务: {server_config}")
     logget.info(f"启动日志: {log_config}")

src/chongming/app/__init__.py

@@ -1,15 +1,26 @@
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request, Depends
+from fastapi.responses import JSONResponse
 from fastapi.middleware.cors import CORSMiddleware
 from sqlmodel import SQLModel, text
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.exc import OperationalError
+from pydantic import BaseModel
 
 from .core.config import get_config
-from .core.jwt_cache import get_jwt_cache
+from .core.cache import get_cache
 from .core.logger import get_logger
+from .core.scheduler import get_task_service, get_task_service_instance, TaskService
 from .api import api_router
+from .task.execute_background import execute_background_task
+from .task import init_tasks_callback
+
+
+class TaskRequest(BaseModel):
+    task_name: str
+    interval: int
+
 
 config = get_config()
 logger = get_logger("app")
@@ -32,6 +43,12 @@ async def lifespan(app: FastAPI):
     if database_config is None:
         raise ValueError("配置不存在")
 
+    task_service = get_task_service_instance()
+
+    await task_service.start(init_tasks_callback)
+
+    app.state.task_service = task_service
+
     # 启动时：创建引擎和会话工厂，初始化数据库表
     engine = create_async_engine(database["url"], **database_config)
     async_session_maker = async_sessionmaker(
@@ -61,8 +78,9 @@ async def lifespan(app: FastAPI):
     yield  # 应用运行期间
 
     # 关闭时：释放引擎
+    await task_service.shutdown(wait=True)
     await engine.dispose()
-    get_jwt_cache().close()
+    get_cache().close()
 
 
 cors = config[env].get("cors", {})
@@ -87,19 +105,36 @@ app.include_router(api_router, prefix="/api/v1")
 async def root():
     """根路径，返回应用信息"""
     return {
-        "message": f"欢迎使用 {config["default"]["app"]["name"]}",
+        "message": f"欢迎使用 {config['default']['app']['name']}",
         "version": config["default"]["app"]["version"],
         "docs": "/docs" if config[env]["debug"] else None,
         "health": "/api/v1/health",
     }
 
 
+@app.post("/schedule")
+async def add_task(
+    request: TaskRequest, task_service: TaskService = Depends(get_task_service)
+):
+    """添加定时任务接口"""
+    job = await task_service.add_interval_job(
+        execute_background_task,
+        seconds=request.interval,
+        args=[request.task_name],
+        job_id=request.task_name,
+    )
+    return {"status": "success", "job_id": job.id}
+
+
 # 全局异常处理器
 @app.exception_handler(Exception)
-async def global_exception_handler(request, exc):
+async def global_exception_handler(_: Request, exc: Exception):
     """全局异常处理器"""
     logger.error(f"全局异常: {exc}")
-    return {
-        "error": "内部服务器错误",
-        "detail": str(exc) if config[env]["debug"] else "请查看服务器日志",
-    }
+    return JSONResponse(
+        status_code=500,
+        content={
+            "error": "内部服务器错误",
+            "detail": str(exc) if config[env]["debug"] else "请查看服务器日志",
+        },
+    )

src/chongming/app/api/deps.py

@@ -5,11 +5,11 @@ from ..service.auth import get_auth_service
 from ..core.jwt_cache import TokenData
 
 
-security = HTTPBearer()
+access_scheme = HTTPBearer(scheme_name="AccessToken")
 
 
 async def get_current_user(
-    credentials: HTTPAuthorizationCredentials = Depends(security),
+    credentials: HTTPAuthorizationCredentials = Depends(access_scheme),
 ) -> TokenData:
     """
     获取当前用户依赖项

src/chongming/app/api/routers/auth.py

@@ -3,7 +3,7 @@ from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ...core.database import get_session
-from ...core.jwt_cache import get_jwt_cache, TokenData
+from ...core.jwt_cache import TokenData
 from ...core.security import verify_password
 from ...service.user import UserService
 from ...service.auth import get_auth_service
@@ -13,7 +13,8 @@ from ..deps import get_current_user
 
 router = APIRouter(tags=["authentication"])
 
-security = HTTPBearer()
+access_scheme = HTTPBearer(scheme_name="AccessToken")
+refresh_scheme = HTTPBearer(scheme_name="RefreshToken")
 
 
 @router.post("/register", response_model=UserRead)
@@ -50,29 +51,14 @@ async def login(
         raise HTTPException(status_code=400, detail="邮箱或密码错误")
     if not user.is_active:
         raise HTTPException(status_code=400, detail="用户被禁用")
-
-    cache = get_jwt_cache()
-    user_data_dict = {
-        "sub": str(user.id),
-        "email": user.email,
-        "username": user.username,
-    }
-    token = cache.create_token(
-        user_id=str(user.id),
-        user_data=user_data_dict,
-        token_type="access",
-        device_id=None,
-        user_agent=request.headers.get("user-agent") if request else None,
-        ip_address=request.client.host if request else None,  # type: ignore
-    )
-
-    return {"access_token": token, "token_type": "bearer"}
+    auth_service = get_auth_service()
+    return await auth_service.create_tokens(user, request, device_id=None)
 
 
 @router.post("/refresh", response_model=dict, summary="刷新访问令牌")
 async def refresh_token(
     request: Request,
-    credentials: HTTPAuthorizationCredentials = Depends(security),
+    credentials: HTTPAuthorizationCredentials = Depends(refresh_scheme),
     session: AsyncSession = Depends(get_session),
 ):
     auth_service = get_auth_service()
@@ -92,7 +78,7 @@ async def refresh_token(
 
 @router.post("/logout")
 async def logout(
-    credentials: HTTPAuthorizationCredentials = Depends(security),
+    credentials: HTTPAuthorizationCredentials = Depends(access_scheme),
 ):
     auth_service = get_auth_service()
     token = credentials.credentials
@@ -100,7 +86,7 @@ async def logout(
     return {"message": "登出成功"}
 
 
-@router.post("/logout-all", summary="等出所有设备")
+@router.post("/logout-all", summary="登出所有设备")
 async def logout_all(current_user: TokenData = Depends(get_current_user)):
     auth_service = get_auth_service()
     await auth_service.logout_all(current_user.user_id)

src/chongming/app/api/routers/todo.py

@@ -10,11 +10,11 @@ from ...model.user import User
 from ..deps import get_current_user
 
 
-router = APIRouter()
+router = APIRouter(dependencies=[Depends(get_current_user)])
 
 
 # -------------------- 路由 --------------------
-@router.post("/", response_model=TodoRead, dependencies=[Depends(get_current_user)])
+@router.post("/", response_model=TodoRead)
 async def create_todo(
     todo: TodoCreate,
     session: AsyncSession = Depends(get_session),
@@ -22,18 +22,14 @@ async def create_todo(
     return await TodoService.create_todo(todo, session)
 
 
-@router.get(
-    "/", response_model=List[TodoRead], dependencies=[Depends(get_current_user)]
-)
+@router.get("/", response_model=List[TodoRead])
 async def read_todos(
     offset: int = 0, limit: int = 100, session: AsyncSession = Depends(get_session)
 ):
     return await TodoService.get_todos(offset, limit, session)
 
 
-@router.get(
-    "/{todo_id}", response_model=TodoRead, dependencies=[Depends(get_current_user)]
-)
+@router.get("/{todo_id}", response_model=TodoRead)
 async def read_todo(todo_id: int, session: AsyncSession = Depends(get_session)):
     todo = await TodoService.get_todo_by_id(todo_id, session)
     if not todo:
@@ -41,9 +37,7 @@ async def read_todo(todo_id: int, session: AsyncSession = Depends(get_session)):
     return todo
 
 
-@router.patch(
-    "/{todo_id}", response_model=TodoRead, dependencies=[Depends(get_current_user)]
-)
+@router.patch("/{todo_id}", response_model=TodoRead)
 async def update_todo(
     todo_id: int, todo_update: TodoUpdate, session: AsyncSession = Depends(get_session)
 ):
@@ -51,7 +45,7 @@ async def update_todo(
     return todo
 
 
-@router.delete("/{todo_id}", dependencies=[Depends(get_current_user)])
+@router.delete("/{todo_id}")
 async def delete_todo(todo_id: int, session: AsyncSession = Depends(get_session)):
     is_deleted = await TodoService.delete_todo(todo_id, session)
     if not is_deleted:

src/chongming/app/core/cache.py

@@ -0,0 +1,248 @@
+import time
+from typing import Any, Optional
+
+import diskcache
+
+from ..core.config import get_config
+
+
+class Cache:
+    def __init__(self):
+        config = get_config()
+        cache_dir = config["cache"]["cache_store_path"]
+        self._cache = diskcache.Cache(cache_dir)
+
+    def set(
+        self,
+        key: str,
+        value: Any,
+        ex: Optional[int] = None,
+        px: Optional[int] = None,
+        raw: bool = False,
+    ) -> bool:
+        """
+        设置缓存
+
+        Args:
+            key: 键
+            value: 值
+            ex: 过期时间（秒）
+            px: 过期时间（毫秒）
+
+        Returns:
+            bool: 是否设置成功
+        """
+        try:
+            expire_at = None
+            expire_seconds = None
+
+            if ex is not None:
+                expire_seconds = ex
+                expire_at = time.time() + ex
+            elif px is not None:
+                expire_seconds = px / 1000.0
+                expire_at = time.time() + expire_seconds
+
+            stored_value = (value, expire_at)
+
+            self._cache.set(key, stored_value, expire=expire_seconds)
+            return True
+
+        except Exception as e:
+            print(f"[缓存] set 错误: {e}")
+            return False
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """
+        获取缓存
+
+        Args:
+            key: 键
+            default: 默认值
+
+        Returns:
+            缓存的值，如果key不存在或已过期返回default
+        """
+        try:
+            stored = self._cache.get(key)
+            if stored is None:
+                return default
+            value, expire_at = stored  # type: ignore
+
+            if expire_at is not None and time.time() > expire_at:  # type: ignore
+                self._cache.delete(key)
+                return default
+
+            return value
+
+        except Exception as e:
+            print(f"[缓存] get 错误: {e}")
+            return default
+
+    def delete(self, *keys: str) -> int:
+        """
+        删除一个或多个key
+
+        Args:
+            keys: 要删除的key
+
+        Returns:
+            成功删除的数量
+        """
+        deleted = 0
+        for key in keys:
+            if self._cache.delete(key):
+                deleted += 1
+        return deleted
+
+    def exists(self, key: str) -> bool:
+        """
+        检查key是否存在且未过期
+
+        Args:
+            key: 键
+
+        Returns:
+            bool: 是否存在
+        """
+        return self.get(key, None) is not None
+
+    def expire(self, key: str, _time: int) -> bool:
+        """
+        设置过期时间
+
+        Args:
+            key: 键
+            time: 过期时间（秒）
+
+        Returns:
+            bool: 是否设置成功
+        """
+        try:
+            stored = self._cache.get(key)
+            if stored is None:
+                return False
+            value, _ = stored  # type: ignore
+            expire_at = time.time() + _time
+            new_stored = (value, expire_at)
+            self._cache.set(key, new_stored, expire=_time)
+            return True
+        except Exception as e:
+            print(f"[缓存] expire 错误: {e}")
+            return False
+
+    def ttl(self, key: str) -> int:
+        """
+        获取key的剩余存活时间
+
+        Args:
+            key: 键
+
+        Returns:
+            int: 剩余秒数，-1表示永不过期，-2表示key不存在
+        """
+        try:
+            stored = self._cache.get(key)
+            if stored is None:
+                return -2
+
+            _, expire_at = stored  # type: ignore
+            if expire_at is None:
+                return -1
+
+            remaining = expire_at - time.time()  # type: ignore
+            return max(0, int(remaining))
+        except Exception as e:
+            print(f"[缓存] ttl 错误: {e}")
+            return -2
+
+    def keys(self, pattern: Optional[str] = None) -> list:
+        """
+        获取所有key，支持简单的通配符匹配（*）
+
+        Args:
+            pattern: 匹配模式，如 "user:*"
+
+        Returns:
+            list: key列表
+        """
+        all_keys = list(self._cache.iterkeys())
+
+        if pattern and "*" in pattern:
+            prefix = pattern.replace("*", "")
+            all_keys = [k for k in all_keys if k.startswith(prefix)]  # type: ignore
+
+        return all_keys
+
+    def clear(self) -> bool:
+        """清空所有缓存。"""
+        try:
+            self._cache.clear()
+            return True
+        except Exception as e:
+            print(f"[缓存] clear 错误: {e}")
+            return False
+
+    def size(self) -> int:
+        """获取当前缓存数量（不含过期但尚未清理的项）。"""
+        # diskcache 的 __len__ 会返回当前有效条目数
+        return len(list(self._cache.iterkeys()))
+
+    def get_all(self) -> dict:
+        """获取所有未过期的缓存数据。"""
+        result = {}
+        for key in self._cache.iterkeys():
+            value = self.get(str(key))  # get 内部会检查并清理过期
+            if value is not None:
+                result[key] = value
+        return result
+
+    def close(self):
+        """关闭缓存。"""
+        self._cache.close()
+
+
+_cache: Optional[Cache] = None
+
+
+def get_cache() -> Cache:
+    global _cache
+    if _cache is None:
+        _cache = Cache()
+    return _cache
+
+
+# 装饰器：缓存函数结果
+def cached(ttl: Optional[int] = None):
+    """
+    函数结果缓存装饰器
+
+    Args:
+        ttl: 缓存时间（秒），None表示永不过期
+
+    Example:
+        @cached(ttl=60)
+        def get_user(user_id: int):
+            return {"id": user_id, "name": "test"}
+    """
+
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            # 生成缓存key
+            key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
+            cache = get_cache()
+            # 尝试从缓存获取
+            result = cache.get(key)
+            if result is not None:
+                return result
+
+            # 执行函数
+            result = func(*args, **kwargs)
+
+            # 存入缓存
+            cache.set(key, result, ex=ttl)
+            return result
+
+        return wrapper
+
+    return decorator

src/chongming/app/core/config.py

@@ -1,6 +1,6 @@
 import sys
 from pathlib import Path
-from typing import TypedDict, Literal, Optional, cast
+from typing import TypedDict, Literal, Optional, cast, List
 
 
 if sys.version_info >= (3, 11):
@@ -53,7 +53,6 @@ class Security(TypedDict):
     secret_key: str
     algorithm: Literal["HS256"]
     access_token_expire_minutes: int
-    jwt_cache_db_path: str
     refresh_token_expire_days: int
     max_sessions_per_user: int
 
@@ -74,20 +73,20 @@ class Logging(TypedDict):
 
 
 class Cors(TypedDict, total=False):
-    allow_origins: list[str]
+    allow_origins: List[str]
     allow_credentials: bool
-    allow_methods: list[str]
-    allow_headers: list[str]
+    allow_methods: List[str]
+    allow_headers: List[str]
 
 
 class ModuleSystem(TypedDict, total=False):
     type: str
-    path: list[str]
+    path: List[str]
 
 
 class FileSystem(TypedDict, total=False):
     type: str
-    path: list[str]
+    path: List[str]
 
 
 class Env(TypedDict, total=False):
@@ -99,9 +98,19 @@ class Env(TypedDict, total=False):
     file_system: FileSystem
 
 
+class Scheduler(TypedDict):
+    job_store_path: str
+
+
+class Cache(TypedDict):
+    cache_store_path: str
+
+
 class Config(TypedDict):
     default: Default
     database: Database
+    scheduler: Scheduler
+    cache: Cache
     security: Security
     development: Env
     production: Env

src/chongming/app/core/jwt_cache.py

@@ -1,17 +1,15 @@
-import jwt
 import json
 from datetime import datetime, timedelta
-from typing import Optional, Dict, Any, List, TypedDict
-from sqlitedict import SqliteDict
+from typing import Optional, Dict, Any, List
 from pydantic import BaseModel
 from dataclasses import dataclass, asdict
-
-# from pydantic import BaseModel
 import hashlib
 import threading
-from pathlib import Path
+
+import jwt
 
 from .config import get_config
+from .cache import get_cache
 
 
 @dataclass
@@ -43,43 +41,9 @@ class JWTCache:
     def __init__(self, db_path: Optional[str] = None):
         self.config = get_config()
 
-        if db_path is None:
-            db_path = self.config["security"]["jwt_cache_db_path"]
-
-        db_file_path = Path(db_path)
-        db_file_path.mkdir(parents=True, exist_ok=True)
-
         self.lock = threading.RLock()
 
-        # 主 Token 缓存
-        self.token_db = SqliteDict(
-            rf"{db_path}/tokens.cache",
-            tablename="tokens",
-            autocommit=True,
-            encode=self._json_encode,
-            decode=self._json_decode,
-            journal_mode="WAL",
-        )
-
-        # 黑名单缓存
-        self.blacklist_db = SqliteDict(
-            rf"{db_path}/blacklist.cache",
-            tablename="blacklist",
-            autocommit=True,
-            encode=json.dumps,
-            decode=json.loads,
-            journal_mode="WAL",
-        )
-
-        # 用户会话索引
-        self.user_sessions_db = SqliteDict(
-            rf"{db_path}/user_sessions.cache",
-            tablename="user_sessions",
-            autocommit=True,
-            encode=json.dumps,
-            decode=json.loads,
-            journal_mode="WAL",
-        )
+        self.cache = get_cache()
 
         self.secret_key = self.config["security"]["secret_key"]
         self.algorithm = self.config["security"]["algorithm"]
@@ -168,7 +132,7 @@ class JWTCache:
             )
 
             # 存储到缓存
-            self.token_db[token_hash] = asdict(cache_data)
+            self.cache.set(f"token:{token_hash}", asdict(cache_data))
 
             # 更新用户会话索引
             self._update_user_session_index(user_id, token_hash, token_type)
@@ -189,8 +153,8 @@ class JWTCache:
     ):
         """更新用户会话索引"""
         with self.lock:
-            key = f"{user_id}:{token_type}"
-            user_tokens = self.user_sessions_db.get(key, [])
+            key = f"user:{user_id}:{token_type}"
+            user_tokens = self.cache.get(key, [])
 
             # 添加新 token
             user_tokens.append(token_hash)
@@ -200,36 +164,36 @@ class JWTCache:
             if len(user_tokens) > max_tokens:
                 user_tokens = user_tokens[-max_tokens:]
 
-            self.user_sessions_db[key] = user_tokens
+            self.cache.set(key, user_tokens)
 
     def _enforce_session_limit(self, user_id: str):
         """强制用户会话限制"""
         max_sessions = self.config["security"]["max_sessions_per_user"]
 
-        access_key = f"{user_id}:access"
-        refresh_key = f"{user_id}:refresh"
+        access_key = f"user:{user_id}:access"
+        refresh_key = f"user:{user_id}:refresh"
 
         # 检查 access token 数量
-        access_tokens = self.user_sessions_db.get(access_key, [])
+        access_tokens = self.cache.get(access_key, [])
         if len(access_tokens) > max_sessions:
             # 删除最早的 token
             tokens_to_remove = access_tokens[:-max_sessions]
             for token_hash in tokens_to_remove:
-                if token_hash in self.token_db:
-                    del self.token_db[token_hash]
+                if self.cache.exists(f"token:{token_hash}"):
+                    self.cache.delete(f"token:{token_hash}")
 
             # 更新索引
-            self.user_sessions_db[access_key] = access_tokens[-max_sessions:]
+            self.cache.set(access_key, access_tokens[-max_sessions:])
 
         # 检查 refresh token 数量
-        refresh_tokens = self.user_sessions_db.get(refresh_key, [])
+        refresh_tokens = self.cache.get(refresh_key, [])
         if len(refresh_tokens) > max_sessions:
             tokens_to_remove = refresh_tokens[:-max_sessions]
             for token_hash in tokens_to_remove:
-                if token_hash in self.token_db:
-                    del self.token_db[token_hash]
+                if self.cache.exists(f"token:{token_hash}"):
+                    self.cache.delete(f"token:{token_hash}")
 
-            self.user_sessions_db[refresh_key] = refresh_tokens[-max_sessions:]
+            self.cache.set(refresh_key, refresh_tokens[-max_sessions:])
 
     def validate_token(
         self, token: str, token_type: str = "access"
@@ -259,14 +223,11 @@ class JWTCache:
             if not user_id or not jti:
                 return None
 
-            # 检查黑名单
             token_hash = self._hash_token(token)
-            if self._is_blacklisted(token_hash):
-                return None
 
             # 检查缓存
             with self.lock:
-                cache_data = self.token_db.get(token_hash)
+                cache_data = self.cache.get(f"token:{token_hash}")
                 if not cache_data:
                     return None
 
@@ -279,12 +240,12 @@ class JWTCache:
                 if datetime.utcnow() > expires_at:
                     # 标记为不活跃
                     cache_data["is_active"] = False
-                    self.token_db[token_hash] = cache_data
+                    self.cache.set(f"token:{token_hash}", cache_data)
                     return None
 
                 # 更新最后使用时间
                 cache_data["last_used"] = datetime.utcnow().isoformat()
-                self.token_db[token_hash] = cache_data
+                self.cache.set(f"token:{token_hash}", cache_data)
 
             return {
                 "user_id": user_id,
@@ -300,68 +261,38 @@ class JWTCache:
         except Exception:
             return None
 
-    def add_to_blacklist(self, token: str, reason: str = "logout"):
-        """添加 Token 到黑名单"""
-        token_hash = self._hash_token(token)
-        expire_time = datetime.utcnow() + timedelta(days=7)  # 黑名单保留7天
-
-        self.blacklist_db[token_hash] = {
-            "reason": reason,
-            "blacklisted_at": datetime.utcnow().isoformat(),
-            "expires_at": expire_time.isoformat(),
-        }
-
-    def _is_blacklisted(self, token_hash: str) -> bool:
-        """检查 Token 是否在黑名单中"""
-        with self.lock:
-            if token_hash not in self.blacklist_db:
-                return False
-
-            blacklist_data = self.blacklist_db[token_hash]
-            expires_at = datetime.fromisoformat(blacklist_data["expires_at"])
-
-            # 如果黑名单条目已过期，删除它
-            if datetime.utcnow() > expires_at:
-                del self.blacklist_db[token_hash]
-                return False
-
-            return True
-
     def invalidate_token(self, token: str):
         """使单个 Token 失效"""
         with self.lock:
             token_hash = self._hash_token(token)
 
-            # 添加到黑名单
-            self.add_to_blacklist(token)
-
             # 标记为不活跃
-            if token_hash in self.token_db:
-                cache_data = self.token_db[token_hash]
+            if self.cache.exists(f"token:{token_hash}"):
+                cache_data = self.cache.get(f"token:{token_hash}")
                 cache_data["is_active"] = False
-                self.token_db[token_hash] = cache_data
+                self.cache.set(f"token:{token_hash}", cache_data)
 
     def invalidate_user_tokens(self, user_id: str, token_type: Optional[str] = None):
         """使用户的所有 Token 失效"""
         with self.lock:
             if token_type:
-                keys = [f"{user_id}:{token_type}"]
+                keys = [f"user:{user_id}:{token_type}"]
             else:
-                keys = [f"{user_id}:access", f"{user_id}:refresh"]
+                keys = [f"user:{user_id}:access", f"{user_id}:refresh"]
 
             for key in keys:
-                token_hashes = self.user_sessions_db.get(key, [])
+                token_hashes = self.cache.get(key, [])
                 for token_hash in token_hashes:
                     # 添加到黑名单
                     # 注意：这里我们只有 token_hash，没有原始 token
                     # 所以只能标记为不活跃
-                    if token_hash in self.token_db:
-                        cache_data = self.token_db[token_hash]
+                    if self.cache.exists(f"token:{token_hash}"):
+                        cache_data = self.cache.get(f"token:{token_hash}")
                         cache_data["is_active"] = False
-                        self.token_db[token_hash] = cache_data
+                        self.cache.set(f"token:{token_hash}", cache_data)
 
                 # 清空用户会话索引
-                self.user_sessions_db[key] = []
+                self.cache.set(key, [])
 
     def get_user_sessions(self, user_id: str) -> List[Dict[str, Any]]:
         """获取用户的所有活跃会话"""
@@ -369,12 +300,12 @@ class JWTCache:
             sessions = []
 
             for token_type in ["access", "refresh"]:
-                key = f"{user_id}:{token_type}"
-                token_hashes = self.user_sessions_db.get(key, [])
+                key = f"user:{user_id}:{token_type}"
+                token_hashes = self.cache.get(key, [])
 
                 for token_hash in token_hashes:
-                    if token_hash in self.token_db:
-                        cache_data = self.token_db[token_hash]
+                    if self.cache.exists(f"token:{token_hash}"):
+                        cache_data = self.cache.get(f"token:{token_hash}")
 
                         # 检查是否活跃且未过期
                         if cache_data.get("is_active", True):
@@ -403,28 +334,14 @@ class JWTCache:
             current_time = datetime.utcnow()
 
             # 清理过期 Token
-            for token_hash, data in list(self.token_db.items()):
-                expires_at = datetime.fromisoformat(data["expires_at"])
+            for token_hash in self.cache.keys("token:*"):
+                expires_at = datetime.fromisoformat(self.cache.get(token_hash))
                 if current_time > expires_at:
-                    del self.token_db[token_hash]
-                    cleaned_count += 1
-
-            # 清理过期黑名单
-            for token_hash, data in list(self.blacklist_db.items()):
-                expires_at = datetime.fromisoformat(data["expires_at"])
-                if current_time > expires_at:
-                    del self.blacklist_db[token_hash]
+                    self.cache.delete(token_hash)
                     cleaned_count += 1
 
             return cleaned_count
 
-    def close(self):
-        with self.lock:
-            """关闭所有数据库连接"""
-            self.token_db.close()
-            self.blacklist_db.close()
-            self.user_sessions_db.close()
-
 
 # 全局缓存实例
 _jwt_cache = None

src/chongming/app/core/logger.py

@@ -1,30 +1,43 @@
 import logging
+import datetime
 from pathlib import Path
-from .config import Logging
 from logging.handlers import RotatingFileHandler
 
+from ..core.config import get_config
+
+
+config = get_config()
+env = config["default"]["env"]
+server_config = config[env]["server"]
+log_config = config[env]["logging"]
+
 
 class ExcludeWatchFilesFilter(logging.Filter):
+    def __init__(self):
+        super().__init__()
+        self._filter_starts_with_tulpe = (
+            "watchfiles",
+            "aiosqlite",
+            "apscheduler",
+            "sqlalchemy",
+            "sqlitedict",
+        )
+
+    # 重写过滤方法
     def filter(self, record):
-        # 如果日志记录来自 watchfiles 模块，则不记录（返回 False）
-        return not record.name.startswith("watchfiles")
+        return not record.name.startswith(self._filter_starts_with_tulpe)
 
 
-def setup_logging(log_config: Logging):
+def setup_logging(logger: logging.Logger, log_name: str = "app"):
     log_level = getattr(logging, log_config["level"], logging.INFO)
     log_format = log_config["format"]
     # 创建文件夹
     log_file_path = Path(log_config["file"])
-    log_file_dir = log_file_path.parent
-    if log_file_dir != Path("."):
-        log_file_dir.mkdir(parents=True, exist_ok=True)
-
-    # 设置特定日志级别的过滤
-    logging.getLogger("watchfiles").setLevel(logging.WARNING)
+    if log_file_path != Path("."):
+        log_file_path.mkdir(parents=True, exist_ok=True)
 
     # 配置根日志记录器
-    root_logger = logging.getLogger()
-    root_logger.setLevel(log_level)
+    logger.setLevel(log_level)
 
     # 创建格式化器
     formatter = logging.Formatter(log_format)
@@ -34,7 +47,7 @@ def setup_logging(log_config: Logging):
     backup_count = log_config.get("backup_count", 5)  # 默认保留 5 个备份
 
     file_handler = RotatingFileHandler(
-        log_config["file"],
+        f"{log_config['file']}/{log_name}-{datetime.date.today()}.log",
         encoding="utf-8",
         maxBytes=max_bytes,
         backupCount=backup_count,
@@ -48,8 +61,8 @@ def setup_logging(log_config: Logging):
     console_handler.addFilter(ExcludeWatchFilesFilter())
 
     # 添加处理器到根日志记录器
-    root_logger.addHandler(file_handler)
-    root_logger.addHandler(console_handler)
+    logger.addHandler(file_handler)
+    logger.addHandler(console_handler)
 
     print(
         f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} "
@@ -59,6 +72,5 @@ def setup_logging(log_config: Logging):
 
 def get_logger(name: str) -> logging.Logger:
     logger = logging.getLogger(name)
-    # 为每个logger添加过滤器
-    logger.addFilter(ExcludeWatchFilesFilter())
+    setup_logging(logger, name)
     return logger

src/chongming/app/core/scheduler.py

@@ -0,0 +1,281 @@
+"""
+APScheduler 集成模块
+提供持久化任务调度服务
+"""
+
+import os
+import sys
+import tempfile
+from contextlib import suppress
+from typing import Dict, List, Optional, Callable, AsyncGenerator, TypedDict, Any
+from datetime import datetime
+
+from fastapi import Request
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
+from apscheduler.executors.asyncio import AsyncIOExecutor
+from apscheduler.job import Job
+from apscheduler.triggers.interval import IntervalTrigger
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.date import DateTrigger
+
+from ..core.config import get_config
+from ..core.logger import get_logger
+
+
+config = get_config()
+logger = get_logger("scheduler")
+
+
+def acquire_scheduler_lock():
+    lock_path = os.path.join(tempfile.gettempdir(), "chongming_scheduler.lock")
+    lock_file = open(lock_path, "w")
+    try:
+        if sys.platform == "win32":
+            import msvcrt
+
+            msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_NB | fcntl.LOCK_EX)
+        return lock_file, True
+    except (IOError, OSError):
+        lock_file.close()
+        return None, False
+
+
+class TaskService:
+    """任务调度服务"""
+
+    def __init__(self):
+        # 使用独立的 SQLite 数据库存储任务（与业务数据库分开）
+        self.job_store_db_path = config.get("scheduler", {}).get(
+            "job_store_path", "scheduler_jobs.db"
+        )
+        job_store_url = f"sqlite:///{self.job_store_db_path}"
+
+        # 配置 job stores 和 executors
+        jobstores = {"default": SQLAlchemyJobStore(url=job_store_url)}
+        executors = {"default": AsyncIOExecutor()}
+        job_defaults = {
+            "coalesce": True,  # 合并错过的任务
+            "max_instances": 3,  # 最大并发实例数
+            "misfire_grace_time": 30,  # 错过任务的宽限时间（秒）
+        }
+
+        self.scheduler = AsyncIOScheduler(
+            jobstores=jobstores,
+            executors=executors,
+            job_defaults=job_defaults,
+            timezone="Asia/Shanghai",
+        )
+        self._started = False
+        self.lock_file, self.is_scheduler_worker = acquire_scheduler_lock()
+
+    async def start(self, init_tasks_callback: Optional[Callable] = None):
+        """启动调度器"""
+        if self.is_scheduler_worker:
+            if not self._started:
+                self.scheduler.start()
+                self._started = True
+                logger.info("任务调度器已启动")
+            if init_tasks_callback:
+                await init_tasks_callback(self)
+
+    async def shutdown(self, wait: bool = True):
+        """关闭调度器"""
+        if self.lock_file:
+            with suppress(Exception):
+                for job in await self.get_jobs():
+                    await self.pause_job(job.get("id", ""))
+
+                if self._started:
+                    self.scheduler.shutdown(wait=wait)
+                    self._started = False
+                    logger.info("任务调度器已关闭")
+                if sys.platform == "win32":
+                    import msvcrt
+
+                    msvcrt.locking(self.lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+                else:
+                    import fcntl
+
+                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+            self.lock_file.close()
+
+    async def add_interval_job(
+        self,
+        func: Callable,
+        seconds: int = 0,
+        minutes: int = 0,
+        hours: int = 0,
+        days: int = 0,
+        job_id: Optional[str] = None,
+        args: Optional[List] = None,
+        kwargs: Optional[Dict] = None,
+        replace_existing: bool = True,
+    ) -> Job:
+        """添加间隔执行任务"""
+        trigger = IntervalTrigger(
+            seconds=seconds, minutes=minutes, hours=hours, days=days
+        )
+        job = self.scheduler.add_job(
+            func,
+            trigger=trigger,
+            id=job_id,
+            args=args or [],
+            kwargs=kwargs or {},
+            replace_existing=replace_existing,
+        )
+        logger.info(f"添加间隔任务: {job.id}")
+        return job
+
+    async def add_cron_job(
+        self,
+        func: Callable,
+        job_id: Optional[str] = None,
+        year: Optional[int] = None,
+        month: Optional[int] = None,
+        day: Optional[int] = None,
+        week: Optional[int] = None,
+        day_of_week: Optional[int] = None,
+        hour: Optional[int] = None,
+        minute: Optional[int] = None,
+        second: Optional[int] = None,
+        args: Optional[List] = None,
+        kwargs: Optional[Dict] = None,
+        replace_existing: bool = True,
+    ) -> Job:
+        """添加 cron 表达式任务"""
+        trigger = CronTrigger(
+            year=year,
+            month=month,
+            day=day,
+            week=week,
+            day_of_week=day_of_week,
+            hour=hour,
+            minute=minute,
+            second=second,
+        )
+        job = self.scheduler.add_job(
+            func,
+            trigger=trigger,
+            id=job_id,
+            args=args or [],
+            kwargs=kwargs or {},
+            replace_existing=replace_existing,
+        )
+        logger.info(f"添加 cron 任务: {job.id}")
+        return job
+
+    async def add_date_job(
+        self,
+        func: Callable,
+        run_date: datetime,
+        job_id: Optional[str] = None,
+        args: Optional[List] = None,
+        kwargs: Optional[Dict] = None,
+    ) -> Job:
+        """添加一次性任务"""
+        trigger = DateTrigger(run_date=run_date)
+        job = self.scheduler.add_job(
+            func, trigger=trigger, id=job_id, args=args or [], kwargs=kwargs or {}
+        )
+        logger.info(f"添加一次性任务: {job.id}")
+        return job
+
+    async def remove_job(self, job_id: str) -> bool:
+        """删除任务"""
+        try:
+            self.scheduler.remove_job(job_id)
+            logger.info(f"删除任务: {job_id}")
+            return True
+        except Exception as e:
+            logger.error(f"删除任务失败 {job_id}: {e}")
+            return False
+
+    async def pause_job(self, job_id: str) -> bool:
+        """暂停任务"""
+        try:
+            self.scheduler.pause_job(job_id)
+            logger.info(f"暂停任务: {job_id}")
+            return True
+        except Exception as e:
+            logger.error(f"暂停任务失败 {job_id}: {e}")
+            return False
+
+    async def resume_job(self, job_id: str) -> bool:
+        """恢复任务"""
+        try:
+            self.scheduler.resume_job(job_id)
+            logger.info(f"恢复任务: {job_id}")
+            return True
+        except Exception as e:
+            logger.error(f"恢复任务失败 {job_id}: {e}")
+            return False
+
+    async def get_job(self, job_id: str) -> Optional[Dict]:
+        """获取任务信息"""
+        job = self.scheduler.get_job(job_id)
+        if job:
+            return self._job_to_dict(job)
+        return None
+
+    async def get_jobs(self) -> List[Dict]:
+        """获取所有任务"""
+        jobs = self.scheduler.get_jobs()
+        return [self._job_to_dict(job) for job in jobs]
+
+    async def reschedule_job(
+        self, job_id: str, trigger_type: str, **trigger_args
+    ) -> Optional[Job]:
+        """重新调度任务（修改触发时间）"""
+        try:
+            job = self.scheduler.get_job(job_id)
+            if not job:
+                return None
+
+            if trigger_type == "interval":
+                trigger = IntervalTrigger(**trigger_args)
+            elif trigger_type == "cron":
+                trigger = CronTrigger(**trigger_args)
+            elif trigger_type == "date":
+                trigger = DateTrigger(**trigger_args)
+            else:
+                raise ValueError(f"不支持的 trigger 类型: {trigger_type}")
+
+            job.reschedule(trigger=trigger)
+            logger.info(f"重新调度任务: {job_id}")
+            return job
+        except Exception as e:
+            logger.error(f"重新调度任务失败 {job_id}: {e}")
+            return None
+
+    def _job_to_dict(self, job: Job) -> Dict:
+        """将 Job 对象转换为字典"""
+        return {
+            "id": job.id,
+            "name": job.name,
+            "next_run_time": (
+                job.next_run_time.isoformat() if job.next_run_time else None
+            ),
+            "trigger": str(job.trigger),
+            "pending": job.pending,
+            "args": job.args,
+            "kwargs": job.kwargs,
+        }
+
+
+_task_service: Optional[TaskService] = None
+
+
+def get_task_service_instance():
+    global _task_service
+    if _task_service is None:
+        _task_service = TaskService()
+    return _task_service
+
+
+async def get_task_service(request: Request) -> AsyncGenerator:
+    return request.app.state.task_service

src/chongming/app/model/user.py

@@ -1,4 +1,5 @@
 from typing import Optional
+
 from sqlmodel import SQLModel, Field
 from pydantic import BaseModel
 

src/chongming/app/service/auth.py

@@ -8,9 +8,9 @@ from passlib.context import CryptContext
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..model.user import User
-from .user import UserService
 from ..core.jwt_cache import get_jwt_cache, TokenResponse, TokenData
 from ..core.config import get_config
+from .user import UserService
 
 
 class AuthService:
@@ -105,7 +105,7 @@ class AuthService:
         request: Optional[Request] = None,
     ) -> Optional[str]:
         """使用刷新令牌获取新的访问令牌"""
-        validation = self.jwt_cache.validate_token(refresh_token, token_type="access")
+        validation = self.jwt_cache.validate_token(refresh_token, token_type="refresh")
         if not validation:
             return None
 

src/chongming/app/service/user.py

@@ -1,6 +1,7 @@
+from typing import Optional
+
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
-from typing import Optional
 
 from ..model.user import User
 from ..core.security import get_password_hash

src/chongming/app/task/__init__.py

@@ -0,0 +1,20 @@
+from typing import Callable, Optional
+
+from ..core.scheduler import TaskService
+from .execute_background import execute_background_task
+
+
+task_dict = {"execute_background_task": execute_background_task}
+
+
+def get_task(task_name: str) -> Optional[Callable]:
+    return task_dict.get(task_name, None)
+
+
+async def init_tasks_callback(task_service: TaskService):
+    await task_service.add_interval_job(
+        execute_background_task,
+        seconds=1,
+        args=["task_name"],
+        job_id="task_name",
+    )

src/chongming/app/task/execute_background.py

@@ -0,0 +1,8 @@
+from ..core.scheduler import logger
+
+
+async def execute_background_task(task_name: str):
+    from datetime import datetime
+
+    """模拟后台任务执行"""
+    logger.info(f"Executing task: {task_name} at {datetime.utcnow()}")

src/chongming/scripts/build.py

@@ -8,46 +8,6 @@ import sys
 import asyncio
 
 
-def create_admin():
-    try:
-        cmd = [
-            sys.executable,
-            "../src/chongming/scripts/create_admin.py",
-        ]
-        print(f"正在运行 create_admin.py: {' '.join(cmd)}")
-
-        # 创建包含 UTF-8 编码的环境变量
-        env = os.environ.copy()
-        env["PYTHONIOENCODING"] = "utf-8"
-        env["PYTHONUTF8"] = "1"
-
-        result = subprocess.run(
-            cmd,
-            cwd="./build",
-            capture_output=True,
-            text=True,
-            encoding="utf-8",  # 明确指定编码
-            errors="replace",  # 替换无法解码的字符
-            env=env,  # 传递修改后的环境变量
-        )
-
-        if result.returncode == 0:
-            print("创建管理员用户成功")
-            print(result.stdout)
-        else:
-            print("创建管理员用户失败")
-            # 安全地打印错误信息，避免编码问题
-            try:
-                print("错误信息:", result.stderr)
-            except UnicodeEncodeError:
-                print(
-                    "错误信息:", result.stderr.encode("ascii", "ignore").decode("ascii")
-                )
-    except Exception as e:
-        print("创建管理员用户失败")
-        print(f"异常: {e}")
-
-
 def run_pyarmor_obfuscate():
     # pyarmor obfuscate --output=obfuscated_script.py original_script.py
     obfuscate_files = [
@@ -113,6 +73,8 @@ def run_pyinstaller():
             "--hidden-import",
             "aiosqlite",
             "--hidden-import",
+            "apscheduler",
+            "--hidden-import",
             "pyjwt",
             "--hidden-import",
             "sqlitedict",