feat: Provider 签到任务时间可配置

- 前端添加签到时间选择器（HH:MM 格式）
- 后端支持动态更新调度器执行时间
- 新增 reschedule_cron_job 方法支持运行时重调度

src/api/admin/system.py

@@ -614,6 +614,16 @@ class AdminSetSystemConfigAdapter(AdminApiAdapter):
             payload.get("description"),
         )
 
+        # 如果更新的是签到任务时间，动态更新调度器
+        if self.key == "provider_checkin_time" and value:
+            try:
+                from src.services.system.maintenance_scheduler import get_maintenance_scheduler
+
+                scheduler = get_maintenance_scheduler()
+                scheduler.update_checkin_time(value)
+            except Exception as e:
+                logger.warning(f"更新签到任务时间失败: {e}")
+
         # 返回时不暴露加密后的值
         display_value = "********" if self.key in self.ENCRYPTED_KEYS else config.value
 

src/services/system/config.py

@@ -117,7 +117,11 @@ class SystemConfigService:
         },
         "enable_provider_checkin": {
             "value": True,
-            "description": "是否启用 Provider 自动签到任务，每天凌晨 1:05 执行",
+            "description": "是否启用 Provider 自动签到任务",
+        },
+        "provider_checkin_time": {
+            "value": "01:05",
+            "description": "Provider 自动签到执行时间（HH:MM 格式，24小时制）",
         },
         "provider_priority_mode": {
             "value": "provider",

src/services/system/maintenance_scheduler.py

@@ -36,11 +36,82 @@ from src.utils.compression import compress_json
 class MaintenanceScheduler:
     """系统维护任务调度器"""
 
+    # 签到任务的 job_id
+    CHECKIN_JOB_ID = "provider_checkin"
+
     def __init__(self) -> None:
         self.running = False
         self._interval_tasks = []
         self._stats_aggregation_lock = asyncio.Lock()
 
+    def _get_checkin_time(self) -> tuple[int, int]:
+        """获取签到任务的执行时间
+
+        Returns:
+            (hour, minute) 元组
+        """
+        db = create_session()
+        try:
+            time_str = SystemConfigService.get_config(db, "provider_checkin_time", "01:05")
+            return self._parse_time_string(time_str)
+        finally:
+            db.close()
+
+    @staticmethod
+    def _parse_time_string(time_str: str) -> tuple[int, int]:
+        """解析时间字符串为 (hour, minute) 元组
+
+        Args:
+            time_str: HH:MM 格式的时间字符串
+
+        Returns:
+            (hour, minute) 元组，解析失败返回默认值 (1, 5)
+        """
+        try:
+            if not time_str or ":" not in time_str:
+                return (1, 5)
+            parts = time_str.split(":")
+            hour = int(parts[0])
+            minute = int(parts[1])
+            # 验证范围
+            if 0 <= hour <= 23 and 0 <= minute <= 59:
+                return (hour, minute)
+            return (1, 5)
+        except (ValueError, IndexError):
+            return (1, 5)
+
+    def update_checkin_time(self, time_str: str) -> bool:
+        """更新签到任务的执行时间
+
+        Args:
+            time_str: HH:MM 格式的时间字符串
+
+        Returns:
+            是否成功更新
+        """
+        hour, minute = self._parse_time_string(time_str)
+
+        scheduler = get_scheduler()
+        success = scheduler.reschedule_cron_job(
+            self.CHECKIN_JOB_ID,
+            hour=hour,
+            minute=minute,
+        )
+
+        if success:
+            logger.info(f"Provider 签到任务时间已更新为: {hour:02d}:{minute:02d}")
+
+        return success
+
+    def get_checkin_job_info(self) -> dict | None:
+        """获取签到任务的信息
+
+        Returns:
+            任务信息字典
+        """
+        scheduler = get_scheduler()
+        return scheduler.get_job_info(self.CHECKIN_JOB_ID)
+
     async def start(self) -> Any:
         """启动调度器"""
         if self.running:
@@ -112,12 +183,13 @@ class MaintenanceScheduler:
             name="Gemini文件映射清理",
         )
 
-        # Provider 签到任务 - 凌晨 1:05 执行
+        # Provider 签到任务 - 根据配置时间执行
+        checkin_hour, checkin_minute = self._get_checkin_time()
         scheduler.add_cron_job(
             self._scheduled_provider_checkin,
-            hour=1,
-            minute=5,  # 在统计聚合任务（1:00）之后执行
-            job_id="provider_checkin",
+            hour=checkin_hour,
+            minute=checkin_minute,
+            job_id=self.CHECKIN_JOB_ID,
             name="Provider签到",
         )
 

src/services/system/scheduler.py

@@ -190,6 +190,66 @@ class TaskScheduler:
         except Exception as e:
             logger.warning(f"移除定时任务失败 {job_id}: {e}")
 
+    def reschedule_cron_job(
+        self,
+        job_id: str,
+        hour: int,
+        minute: int = 0,
+    ) -> bool:
+        """
+        重新调度 cron 定时任务的执行时间
+
+        Args:
+            job_id: 任务ID
+            hour: 新的执行时间（小时），使用业务时区
+            minute: 新的执行时间（分钟）
+
+        Returns:
+            是否成功重新调度
+        """
+        try:
+            job = self.scheduler.get_job(job_id)
+            if not job:
+                logger.warning(f"任务不存在: {job_id}")
+                return False
+
+            trigger = CronTrigger(hour=hour, minute=minute, timezone=APP_TIMEZONE)
+            self.scheduler.reschedule_job(job_id, trigger=trigger)
+
+            logger.info(
+                f"已重新调度定时任务: {job.name}, "
+                f"新执行时间: {hour:02d}:{minute:02d} ({APP_TIMEZONE})"
+            )
+            return True
+        except Exception as e:
+            logger.exception(f"重新调度任务失败 {job_id}: {e}")
+            return False
+
+    def get_job_info(self, job_id: str) -> dict | None:
+        """
+        获取任务信息
+
+        Args:
+            job_id: 任务ID
+
+        Returns:
+            任务信息字典，包含 name, next_run_time 等
+        """
+        try:
+            job = self.scheduler.get_job(job_id)
+            if not job:
+                return None
+
+            next_run = job.next_run_time
+            return {
+                "id": job.id,
+                "name": job.name,
+                "next_run_time": next_run.isoformat() if next_run else None,
+            }
+        except Exception as e:
+            logger.warning(f"获取任务信息失败 {job_id}: {e}")
+            return None
+
     @property
     def is_running(self) -> bool:
         """调度器是否在运行"""