""" Scheduler - 定时扫描并自动修复 Bug """ import signal import threading import time from datetime import datetime from typing import Optional import httpx from loguru import logger from ..config import settings from .core import RepairEngine from .task_manager import TaskManager class RepairScheduler: """定时扫描 Bug 并自动触发修复""" def __init__( self, interval: int = 3600, run_tests: bool = True, auto_commit: bool = False, projects: Optional[list[str]] = None, ): """ Args: interval: 扫描间隔(秒),默认 3600(1 小时) run_tests: 修复后是否运行测试 auto_commit: 是否自动提交代码 projects: 要监控的项目列表,None 表示从 API 动态获取 """ self.interval = interval self.run_tests = run_tests self.auto_commit = auto_commit self.projects = projects # None 表示每次扫描时动态获取 self._stop_event = threading.Event() self._repairing = False @staticmethod def _fetch_projects_from_api() -> list[str]: """从 Log Center API 动态获取所有已注册项目""" try: resp = httpx.get(f"{settings.log_center_url}/api/v1/projects", timeout=10) resp.raise_for_status() projects = resp.json().get("projects", []) project_ids = [p["project_id"] for p in projects] logger.info(f"从 API 获取到 {len(project_ids)} 个项目: {', '.join(project_ids)}") return project_ids except Exception as e: logger.warning(f"从 API 获取项目列表失败: {e},回退到本地配置") return RepairScheduler._get_local_projects() @staticmethod def _get_local_projects() -> list[str]: """回退:从本地 .env 配置获取有路径的项目""" candidates = ["rtc_backend", "rtc_web", "airhub_app"] return [p for p in candidates if settings.get_project_path(p)] def start(self): """启动定时任务(阻塞式,Ctrl+C 退出)""" signal.signal(signal.SIGINT, self._handle_signal) signal.signal(signal.SIGTERM, self._handle_signal) logger.info("=" * 60) logger.info("Repair Scheduler 启动") logger.info(f" 扫描间隔: {self.interval}s ({self.interval // 60} min)") if self.projects: logger.info(f" 监控项目: {', '.join(self.projects)} (指定)") else: logger.info(" 监控项目: 从 API 动态获取") logger.info(f" 运行测试: {self.run_tests}") logger.info(f" 自动提交: {self.auto_commit}") logger.info("=" * 60) # 启动后立即执行一次 self._tick() while not self._stop_event.is_set(): self._stop_event.wait(self.interval) if not self._stop_event.is_set(): self._tick() logger.info("Repair Scheduler 已停止") def stop(self): """停止定时任务""" logger.info("正在停止 Scheduler...") self._stop_event.set() def _handle_signal(self, signum, frame): logger.info(f"收到信号 {signum},准备停止") self.stop() def _tick(self): """执行一次扫描和修复""" ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") logger.info(f"[{ts}] 开始扫描...") # 每次扫描时动态获取项目列表(除非启动时指定了固定列表) projects = self.projects or self._fetch_projects_from_api() task_manager = TaskManager() total_found = 0 total_fixed = 0 try: for project_id in projects: bugs = task_manager.fetch_pending_bugs(project_id) if not bugs: logger.info(f" [{project_id}] 无待修复 Bug") continue count = len(bugs) total_found += count logger.info(f" [{project_id}] 发现 {count} 个待修复 Bug") # 项目路径由 fix_project() 从 API 动态获取,此处不再阻断 fixed = self._run_repair(project_id) total_fixed += fixed logger.info( f"[{ts}] 扫描完成: " f"发现 {total_found} 个 Bug, 成功修复 {total_fixed} 个" ) logger.info(f"下次扫描: {self.interval}s 后") except Exception as e: logger.error(f"扫描过程出错: {e}") finally: task_manager.close() def _run_repair(self, project_id: str) -> int: """对指定项目执行修复,返回成功修复的数量""" if self._repairing: logger.warning(f" [{project_id}] 上一次修复尚未完成,跳过") return 0 self._repairing = True try: engine = RepairEngine() result = engine.fix_project( project_id=project_id, run_tests=self.run_tests, auto_commit=self.auto_commit, ) engine.close() if result.total == 0: return 0 logger.info( f" [{project_id}] 修复结果: " f"{result.success_count}/{result.total} 成功" ) for r in result.results: icon = "✓" if r.success else "✗" logger.info(f" {icon} Bug #{r.bug_id}: {r.message}") return result.success_count except Exception as e: logger.error(f" [{project_id}] 修复过程出错: {e}") return 0 finally: self._repairing = False