修复默认入口忽略command参数的错误

Application/web.py

@@ -4,6 +4,7 @@ from fastapi.responses import JSONResponse
 from contextlib import asynccontextmanager
 from ..CoreModules.middleware import ConcurrencyLimitMiddleware
 from ..CoreModules.plugin_interface import ImportPlugins
+from ..CoreModules.clock_scheduler import get_clock_scheduler
 from ..CoreRouters import callback, health
 from ..Convention.Runtime.GlobalConfig import *
 from ..Convention.Runtime.Architecture import Architecture
@@ -16,6 +17,8 @@ APP_CONFIG = {
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """应用生命周期管理"""
+    scheduler = get_clock_scheduler()
+    await scheduler.start()
     # 启动
     logger.Log("Info", "应用启动中...")
     
@@ -25,17 +28,17 @@ async def lifespan(app: FastAPI):
     # 启动后台清理
     logger.Log("Info", "启动后台清理...")
 
+    try:
         yield
-    
+    finally:
         # 关闭
         try:
             logger.Log("Info", "关闭应用...")
-        # await cleanup_task
+            await scheduler.stop()
         except asyncio.CancelledError:
             pass
         finally:
             logger.Log("Info", "关闭应用完成...")
-        # db.close()
 
 def generate_app(kwargs: dict) -> FastAPI:
     '''

CoreModules/clock_scheduler.py

@@ -0,0 +1,183 @@
+import asyncio
+import importlib
+import inspect
+import json
+import time
+from typing import Any, Callable, Dict, Optional, Sequence
+
+from ..Convention.Runtime.Architecture import Architecture
+from ..Convention.Runtime.GlobalConfig import ConsoleFrontColor, ProjectConfig
+from ..CoreModules.database import (
+    STATUS_COMPLETED,
+    STATUS_FAILED,
+    STATUS_PENDING,
+    STATUS_RUNNING,
+    Database,
+    get_db,
+)
+
+
+class ClockScheduler:
+    def __init__(self) -> None:
+        self._db: Database = get_db()
+        self._config: ProjectConfig = Architecture.Get(ProjectConfig)
+        self._logger = self._config
+        self._tick_ms = int(self._config.FindItem("scheduler_tick_ms", 1000*60))
+        self._batch_size = int(self._config.FindItem("scheduler_max_batch", 1000))
+        self._loop_task: Optional[asyncio.Task] = None
+        self._running = False
+        self._lock = asyncio.Lock()
+        self._config.SaveProperties()
+
+    @classmethod
+    def instance(cls) -> "ClockScheduler":
+        if not Architecture.Contains(ClockScheduler):
+            Architecture.Register(ClockScheduler, ClockScheduler(), lambda: None)
+        return Architecture.Get(ClockScheduler)
+
+    async def start(self) -> None:
+        async with self._lock:
+            if self._running:
+                return
+            self._running = True
+            self._db.reset_running_tasks()
+            self._loop_task = asyncio.create_task(self._run_loop())
+            self._logger.Log(
+                "Info",
+                f"{ConsoleFrontColor.GREEN}ClockScheduler started with tick {self._tick_ms} ms{ConsoleFrontColor.RESET}",
+            )
+
+    async def stop(self) -> None:
+        async with self._lock:
+            if not self._running:
+                return
+            self._running = False
+            if self._loop_task:
+                self._loop_task.cancel()
+                try:
+                    await self._loop_task
+                except asyncio.CancelledError:
+                    pass
+                self._loop_task = None
+            self._logger.Log(
+                "Info",
+                f"{ConsoleFrontColor.GREEN}ClockScheduler stopped{ConsoleFrontColor.RESET}",
+            )
+
+    def register_task(
+        self,
+        plugin_module: str,
+        plugin_class: Optional[str],
+        callback_name: str,
+        delay_ms: int,
+        *,
+        args: Optional[Sequence[Any]] = None,
+        kwargs: Optional[Dict[str, Any]] = None,
+    ) -> int:
+        execute_at = int(time.time() * 1000) + max(delay_ms, 0)
+        payload = json.dumps({
+            "args": list(args) if args else [],
+            "kwargs": kwargs or {},
+        })
+        task_id = self._db.create_scheduled_task(
+            plugin_module,
+            plugin_class,
+            callback_name,
+            payload,
+            execute_at,
+        )
+        self._logger.Log(
+            "Info",
+            f"{ConsoleFrontColor.LIGHTCYAN_EX}Scheduled task {task_id} for {plugin_module}.{callback_name} at {execute_at}{ConsoleFrontColor.RESET}",
+        )
+        return task_id
+
+    async def _run_loop(self) -> None:
+        try:
+            while self._running:
+                await self._process_due_tasks()
+                await asyncio.sleep(self._tick_ms / 1000)
+        except asyncio.CancelledError:
+            pass
+        finally:
+            self._running = False
+
+    async def _process_due_tasks(self) -> None:
+        now_ms = int(time.time() * 1000)
+        tasks = self._db.get_due_tasks(now_ms, self._batch_size)
+        for task in tasks:
+            task_id = int(task["id"])
+            attempts = int(task["attempts"])
+            try:
+                self._db.update_task_status(task_id, STATUS_RUNNING, attempts=attempts)
+                await self._execute_task(task)
+            except Exception as exc:  # pylint: disable=broad-except
+                message = f"{type(exc).__name__}: {exc}"
+                self._logger.Log(
+                    "Error",
+                    f"{ConsoleFrontColor.RED}Task {task_id} failed: {message}{ConsoleFrontColor.RESET}",
+                )
+                self._db.update_task_status(
+                    task_id,
+                    STATUS_FAILED,
+                    attempts=attempts + 1,
+                    last_error=message,
+                )
+
+    async def _execute_task(self, task: Any) -> None:
+        task_id = int(task["id"])
+        attempts = int(task["attempts"])
+        plugin_module = task["plugin_module"]
+        plugin_class = task["plugin_class"]
+        callback_name = task["callback_name"]
+        payload_raw = task["payload"]
+        args, kwargs = self._decode_payload(payload_raw)
+        callback = self._resolve_callback(plugin_module, plugin_class, callback_name)
+        if not callback:
+            self._db.update_task_status(task_id, STATUS_FAILED, attempts=attempts+1)
+            return
+        result = callback(*args, **kwargs)
+        if inspect.isawaitable(result):
+            await result
+        self._db.update_task_status(task_id, STATUS_COMPLETED, attempts=attempts)
+
+    def _resolve_callback(
+        self,
+        plugin_module: str,
+        plugin_class: Optional[str],
+        callback_name: str,
+    ) -> Optional[Callable[..., Any]]:
+        callback: Optional[Callable[..., Any]] = None
+        module = importlib.import_module(plugin_module)
+        if plugin_class:
+            target_class = getattr(module, plugin_class)
+            if Architecture.Contains(target_class):
+                instance = Architecture.Get(target_class)
+                callback = getattr(instance, callback_name)
+        else:
+            callback = getattr(module, callback_name)
+        return callback
+
+    def _decode_payload(self, payload_raw: Optional[str]) -> tuple[list[Any], Dict[str, Any]]:
+        if not payload_raw:
+            return [], {}
+        try:
+            payload_obj = json.loads(payload_raw)
+            args = payload_obj.get("args", [])
+            kwargs = payload_obj.get("kwargs", {})
+            if not isinstance(args, list) or not isinstance(kwargs, dict):
+                raise ValueError("Invalid payload structure")
+            return args, kwargs
+        except Exception as exc:  # pylint: disable=broad-except
+            self._logger.Log(
+                "Warning",
+                f"{ConsoleFrontColor.YELLOW}Failed to decode payload: {exc}{ConsoleFrontColor.RESET}",
+            )
+            return [], {}
+
+
+def get_clock_scheduler() -> ClockScheduler:
+    return ClockScheduler.instance()
+
+
+__all__ = ["ClockScheduler", "get_clock_scheduler"]

CoreModules/database.py

@@ -1,5 +1,7 @@
 """SQLite数据库操作模块 - 使用标准库sqlite3"""
 import sqlite3
+import time
+from typing import Any, Optional
 from ..Convention.Runtime.Config import *
 from ..Convention.Runtime.GlobalConfig import ProjectConfig, ConsoleFrontColor
 from ..Convention.Runtime.Architecture import Architecture
@@ -9,6 +11,12 @@ logger: ProjectConfig = Architecture.Get(ProjectConfig)
 DATABASE_PATH = logger.GetFile(logger.FindItem("database_path", "db.db"), False).GetFullPath()
 logger.SaveProperties()
 
+SCHEDULED_TASK_TABLE = "scheduled_tasks"
+STATUS_PENDING = "pending"
+STATUS_RUNNING = "running"
+STATUS_COMPLETED = "completed"
+STATUS_FAILED = "failed"
+
 class Database:
     """数据库管理类"""
     
@@ -123,7 +131,112 @@ class Database:
 
     def init_tables(self):
         """初始化数据库表"""
-        pass
+        self._ensure_scheduled_tasks_table()
+    
+    def _ensure_scheduled_tasks_table(self) -> None:
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"""
+            CREATE TABLE IF NOT EXISTS {SCHEDULED_TASK_TABLE} (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                plugin_module TEXT NOT NULL,
+                plugin_class TEXT,
+                callback_name TEXT NOT NULL,
+                payload TEXT,
+                execute_at INTEGER NOT NULL,
+                status TEXT NOT NULL DEFAULT '{STATUS_PENDING}',
+                attempts INTEGER NOT NULL DEFAULT 0,
+                last_error TEXT,
+                created_at INTEGER NOT NULL,
+                updated_at INTEGER NOT NULL
+            )
+            """
+        )
+        cursor.execute(
+            f"CREATE INDEX IF NOT EXISTS idx_{SCHEDULED_TASK_TABLE}_status_execute_at ON {SCHEDULED_TASK_TABLE}(status, execute_at)"
+        )
+
+    def create_scheduled_task(
+        self,
+        plugin_module: str,
+        plugin_class: Optional[str],
+        callback_name: str,
+        payload: Optional[str],
+        execute_at_ms: int
+    ) -> int:
+        now_ms = int(time.time() * 1000)
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"""
+            INSERT INTO {SCHEDULED_TASK_TABLE} (
+                plugin_module, plugin_class, callback_name, payload,
+                execute_at, status, attempts, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (
+                plugin_module,
+                plugin_class,
+                callback_name,
+                payload,
+                execute_at_ms,
+                STATUS_PENDING,
+                0,
+                now_ms,
+                now_ms,
+            )
+        )
+        task_id = cursor.lastrowid
+        return int(task_id)
+
+    def get_due_tasks(self, now_ms: int, limit: int) -> list[sqlite3.Row]:
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"""
+            SELECT * FROM {SCHEDULED_TASK_TABLE}
+            WHERE status = ? AND execute_at <= ?
+            ORDER BY execute_at ASC
+            LIMIT ?
+            """,
+            (STATUS_PENDING, now_ms, limit)
+        )
+        return cursor.fetchall()
+
+    def update_task_status(
+        self,
+        task_id: int,
+        status: str,
+        *,
+        attempts: Optional[int] = None,
+        last_error: Optional[str] = None
+    ) -> None:
+        now_ms = int(time.time() * 1000)
+        sets = ["status = ?"]
+        params: list[Any] = [status]
+        if attempts is not None:
+            sets.append("attempts = ?")
+            params.append(attempts)
+        if last_error is not None:
+            sets.append("last_error = ?")
+            params.append(last_error)
+        params.append(task_id)
+        set_clause = ", ".join(sets)
+        cursor = self.conn.cursor()
+        cursor.execute(
+            f"UPDATE {SCHEDULED_TASK_TABLE} SET {set_clause} WHERE id = ?",
+            params
+        )
+
+    def reset_running_tasks(self) -> None:
+        cursor = self.conn.cursor()
+        now_ms = int(time.time() * 1000)
+        cursor.execute(
+            f"""
+            UPDATE {SCHEDULED_TASK_TABLE}
+            SET status = ?, updated_at = ?
+            WHERE status = ?
+            """,
+            (STATUS_PENDING, now_ms, STATUS_RUNNING)
+        )
     
     def close(self):
         """关闭数据库连接"""
@@ -132,10 +245,19 @@ class Database:
             self._conn = None
             logger.Log("Info", f"{ConsoleFrontColor.GREEN}数据库连接已关闭{ConsoleFrontColor.RESET}")
 
+
 def get_db() -> Database:
     """获取全局数据库实例（单例模式）"""
     if not Architecture.Contains(Database):
         return Database()
     return Architecture.Get(Database)
 
-__all__ = ["get_db"]
+__all__ = [
+    "get_db",
+    "Database",
+    "SCHEDULED_TASK_TABLE",
+    "STATUS_PENDING",
+    "STATUS_RUNNING",
+    "STATUS_COMPLETED",
+    "STATUS_FAILED",
+]

CoreModules/plugin_interface.py

@@ -5,6 +5,7 @@ from ..Convention.Runtime.GlobalConfig import ProjectConfig
 from ..Convention.Runtime.Architecture import Architecture
 from ..Convention.Runtime.File import ToolFile
 from ..CoreModules.database import get_db
+from ..CoreModules.clock_scheduler import get_clock_scheduler
 from fastapi import APIRouter, FastAPI
 from typing import *
 from pydantic import *
@@ -21,7 +22,7 @@ class DatabaseModel(BaseModel):
 class PluginInterface(ABC):
     plugin_instances: Dict[str, "PluginInterface"] = {}
 
-    async def callback(self, message: str, chat_id: int, user_id: int) -> str|None:
+    async def callback(self, message: str|None|Literal[""], chat_id: int, user_id: int) -> str|None:
         '''
         继承后重写该方法接受消息并返回消息
         返回空字符串代表不进行反馈
@@ -33,18 +34,9 @@ class PluginInterface(ABC):
         config.Log("Warning", f"{ConsoleFrontColor.YELLOW}插件{self.__class__.__name__}未实现callback方法{ConsoleFrontColor.RESET}")
         return ""
 
-    def execute(self, path:str) -> Optional[APIRouter]:
+    @final
-        '''
+    def register_table(self, db_model: DatabaseModel) -> None:
-        继承后是否返回路由决定是否启动该插件
-        若返回None, 则不启动该插件
-        '''
-        Architecture.Register(self.__class__, self, self.wake_up, *self.dependencies())
-        router = APIRouter()
-        router.post(path)(self.generate_router_callback())
-        # 在数据库保证必要的表和列存在
         db = get_db()
-        db_model = self.register_db_model()
-        if db_model:
         cursor = db.conn.cursor()
         sql = f"CREATE TABLE IF NOT EXISTS {db_model.table_name} ({', '.join([f'{name} {field_def}' for name, field_def in db_model.column_defs.items()])})"
         config.Log("Info", f"{ConsoleFrontColor.LIGHTMAGENTA_EX}为表 {db_model.table_name} 创建: {sql}{ConsoleFrontColor.RESET}")
@@ -53,15 +45,77 @@ class PluginInterface(ABC):
         except Exception as e:
             config.Log("Error", f"{ConsoleFrontColor.RED}为表 {db_model.table_name} 创建失败: {e}{ConsoleFrontColor.RESET}")
 
+        try:
+            cursor.execute(f"PRAGMA table_info({db_model.table_name})")
+            existing_columns = {row["name"] for row in cursor.fetchall()}
+        except Exception as e:
+            config.Log("Error", f"{ConsoleFrontColor.RED}查询表 {db_model.table_name} 列信息失败: {e}{ConsoleFrontColor.RESET}")
+            return
+
+        constraint_keywords = ("PRIMARY KEY", "FOREIGN KEY", "UNIQUE", "CHECK")
+        for column_name, column_def in db_model.column_defs.items():
+            column_name_upper = column_name.upper()
+            column_def_upper = column_def.upper()
+            if any(keyword in column_name_upper for keyword in constraint_keywords) or any(
+                column_def_upper.startswith(keyword) for keyword in constraint_keywords
+            ):
+                continue
+            if " " in column_name or "(" in column_name:
+                continue
+            if column_name in existing_columns:
+                continue
+            alter_sql = f"ALTER TABLE {db_model.table_name} ADD COLUMN {column_name} {column_def}"
+            config.Log(
+                "Info",
+                f"{ConsoleFrontColor.LIGHTMAGENTA_EX}为表 {db_model.table_name} 添加缺失列: {alter_sql}{ConsoleFrontColor.RESET}",
+            )
+            try:
+                cursor.execute(alter_sql)
+            except Exception as e:
+                config.Log("Error", f"{ConsoleFrontColor.RED}为表 {db_model.table_name} 添加列 {column_name} 失败: {e}{ConsoleFrontColor.RESET}")
+                continue
+        try:
+            db.conn.commit()
+        except Exception as e:
+            config.Log("Error", f"{ConsoleFrontColor.RED}提交表 {db_model.table_name} 列更新失败: {e}{ConsoleFrontColor.RESET}")
+
+    @final
+    def execute(self) -> Optional[APIRouter]:
+        '''
+        继承后是否返回路由决定是否启动该插件
+        若返回None, 则不启动该插件
+        '''
+
+        def setup() -> None:
+            # 在数据库保证必要的表和列存在
+            db_model = self.register_db_model()
+            if db_model is None:
+                pass
+            elif isinstance(db_model, DatabaseModel):
+                self.register_table(db_model)
+            else:
+                for model in db_model:
+                    self.register_table(model)
+            self.wake_up()
+
+        Architecture.Register(self.__class__, self, setup, *self.dependencies())
+        router = APIRouter()
+        router.post(f"/{self.__class__.__name__}/callback")(self.generate_router_callback())
+        if self.generate_router_illustrated_guide() is not None:
+            router.get(f"/{self.__class__.__name__}")(self.generate_router_illustrated_guide())
         return router
 
     def generate_router_callback(self) -> Callable|Coroutine:
         '''
         继承后重写该方法生成路由回调函数
         '''
-        async def callback(message: str, chat_id: int, user_id: int) -> Any:
+        return self.callback
-            return await self.callback(message, chat_id, user_id)
+
-        return callback
+    def generate_router_illustrated_guide(self) -> Callable|Coroutine|None:
+        '''
+        继承后重写该方法生成渲染图鉴与攻略网页的函数
+        '''
+        return None
 
     def dependencies(self) -> List[Type]:
         '''
@@ -76,6 +130,7 @@ class PluginInterface(ABC):
         '''
         pass
 
+    @final
     def register_plugin(self, command: str) -> None:
         '''
         将插件注册, 使其可以被命令匹配
@@ -86,12 +141,49 @@ class PluginInterface(ABC):
             config.Log("Info", f"插件{self.__class__.__name__}已注册命令{command}")
             PluginInterface.plugin_instances[command] = self
 
-    def register_db_model(self) -> Optional[DatabaseModel]:
+    def register_db_model(self) -> List[DatabaseModel]|DatabaseModel|None:
         '''
         继承后重写该方法注册数据库模型
         '''
         return None
 
+    def register_clock(
+        self,
+        callback: Callable[..., Any],
+        delay_ms: int,
+        *,
+        args: Optional[Sequence[Any]] = None,
+        kwargs: Optional[Dict[str, Any]] = None,
+    ) -> int:
+        '''
+        注册一次性延时任务
+        Args:
+            callback: 时间到期后调用的函数/方法
+            delay_ms: 延迟毫秒数
+            args: 传入回调的位置参数
+            kwargs: 传入回调的关键字参数
+        '''
+        if not callable(callback):
+            raise ValueError("callback must be callable")
+        scheduler = get_clock_scheduler()
+        plugin_module = callback.__module__
+        plugin_class: Optional[str] = None
+        if hasattr(callback, "__self__") and callback.__self__ is self:
+            plugin_module = self.__class__.__module__
+            plugin_class = self.__class__.__name__
+        callback_name = getattr(callback, "__name__", None)
+        if not callback_name:
+            raise ValueError("callback must have a __name__ attribute")
+        task_id = scheduler.register_task(
+            plugin_module,
+            plugin_class,
+            callback_name,
+            delay_ms,
+            args=args,
+            kwargs=kwargs,
+        )
+        return task_id
+
     def is_enable_plugin(self) -> bool:
         '''
         继承后重写该方法判断是否启用该插件
@@ -132,7 +224,7 @@ def ImportPlugins(app: FastAPI, plugin_dir:str = "Plugins") -> None:
                             plugin = plugin_class()
                             if plugin.is_enable_plugin() == False:
                                 continue
-                            router = plugin.execute(f"/{module_file.GetFullPath().replace(".py", '')}/{class_name}")
+                            router = plugin.execute()
                             if router:
                                 app.include_router(router, prefix=f"/api", tags=[plugin.get_plugin_tag()])
                 except Exception as e:

CoreRouters/callback.py

@@ -4,6 +4,7 @@ from ..Convention.Runtime.Architecture import Architecture
 
 from fastapi import APIRouter, Request
 from fastapi.responses import JSONResponse
+import re
 
 from ..CoreModules.models import CallbackRequest
 from ..CoreModules.plugin_interface import PluginInterface
@@ -21,13 +22,25 @@ async def callback_verify():
     logger.Log("Info", "收到Callback验证请求")
     return JSONResponse({"result": "ok"})
 
+# 机器人名称模式（用于从@消息中提取，兼容半角/全角空格）
+AT_PATTERN = re.compile(r'@[^\s]+[\s\u3000]+(.+)', re.DOTALL)
+
+def parse_message_after_at(message: str) -> str:
+    # 去除首尾空格
+    message = message.strip()
+    
+    # 尝试提取@后的内容
+    at_match = AT_PATTERN.search(message)
+    if at_match:
+        return at_match.group(1).strip()
+    return message
 
 @router.post("/callback/construct")
 async def callback_receive_construct(callback_data: CallbackRequest):
     """以构造好的Callback消息进行处理, 已知方式"""
     try:
         # 解析指令
-        content = callback_data.content
+        content = parse_message_after_at(callback_data.content)
         command = content.split(" ")[0]
         message = content[len(command):].strip()
         logger.Log("Info", f"识别指令: command={command}")
@@ -65,7 +78,7 @@ async def callback_receive(request: Request):
                 return JSONResponse({"result": "error", "message": str(e)})
         
         # 解析指令
-        content = callback_data.content
+        content = parse_message_after_at(callback_data.content)
         command = content.split(" ")[0]
         message = content[len(command):].strip()
         logger.Log("Info", f"识别指令: command={command}")
@@ -103,6 +116,9 @@ async def handle_command(command: str, message: str,
         if plugin:
             logger.Log("Info", f"已找到插件注册指令: {command}, class: {plugin.__class__.__name__}")
             return await plugin.callback(message, chat_id, user_id)
+        elif "default" in PluginInterface.plugin_instances:
+            logger.Log("Info", f"未找到插件注册指令: {command}, 使用默认插件: {PluginInterface.plugin_instances["default"].__class__.__name__}")
+            return await PluginInterface.plugin_instances["default"].callback(command+" "+message, chat_id, user_id)
         else:
             return f"❌ 未识别指令: {command}"
     except Exception as e: