批量Push管理器：收集多个文件后一次性push，减少git操作频率 触发push的条件： 1. 队列达到batch_size 2. 队列不满但超过timeout_seconds（默认2分钟）- 通过后台定时器主动触发
| 38 | # 批量Push队列管理器 |
| 39 | # ========================================== |
| 40 | class BatchPushManager: |
| 41 | """ |
| 42 | 批量Push管理器：收集多个文件后一次性push，减少git操作频率 |
| 43 | |
| 44 | 触发push的条件： |
| 45 | 1. 队列达到batch_size |
| 46 | 2. 队列不满但超过timeout_seconds（默认2分钟）- 通过后台定时器主动触发 |
| 47 | """ |
| 48 | def __init__(self, batch_size: int = 5, timeout_seconds: float = 120.0, idle_timeout: float = 15.0): |
| 49 | self.batch_size = batch_size |
| 50 | self.timeout_seconds = timeout_seconds # 硬超时时间（秒） |
| 51 | self.idle_timeout = idle_timeout # 空闲超时（队列有文件但没有新文件加入的时间） |
| 52 | self._pending_files: List[Dict[str, Any]] = [] # [{"filename": str, "code": str, "future": asyncio.Future}] |
| 53 | self._lock = asyncio.Lock() |
| 54 | self._submitter = None |
| 55 | self._first_file_time: Optional[float] = None # 队列中第一个文件加入的时间 |
| 56 | self._last_add_time: Optional[float] = None # 最后一个文件加入的时间 |
| 57 | self._timeout_task: Optional[asyncio.Task] = None # 超时定时器任务 |
| 58 | self._running = False # 是否正在运行 |
| 59 | self._expected_total: Optional[int] = None # 预期的总文件数 |
| 60 | self._added_count: int = 0 # 已加入队列的文件总数（包括已push的） |
| 61 | |
| 62 | def set_submitter(self, submitter: CodeSubmitter): |
| 63 | """设置submitter实例""" |
| 64 | self._submitter = submitter |
| 65 | |
| 66 | def set_expected_total(self, total: int): |
| 67 | """ |
| 68 | 设置预期的总文件数。当队列达到这个数时会立即push，不需要等待batch_size或超时。 |
| 69 | 这用于处理任务数少于batch_size的情况。 |
| 70 | """ |
| 71 | self._expected_total = total |
| 72 | self._added_count = 0 |
| 73 | logger.info(f"| 📊 Expected total files: {total}") |
| 74 | |
| 75 | def start(self): |
| 76 | """启动管理器""" |
| 77 | self._running = True |
| 78 | |
| 79 | async def stop(self): |
| 80 | """停止管理器，取消定时器""" |
| 81 | self._running = False |
| 82 | if self._timeout_task and not self._timeout_task.done(): |
| 83 | self._timeout_task.cancel() |
| 84 | try: |
| 85 | await self._timeout_task |
| 86 | except asyncio.CancelledError: |
| 87 | pass |
| 88 | self._timeout_task = None |
| 89 | |
| 90 | def _check_timeout(self) -> tuple[bool, str]: |
| 91 | """ |
| 92 | 检查是否超时（调用前需要持有锁） |
| 93 | |
| 94 | Returns: |
| 95 | (is_timeout, reason): 是否超时及原因 |
| 96 | """ |
| 97 | if not self._pending_files: |
no outgoing calls
no test coverage detected