Select candidate in round-robin order for the given scenario.
(self, tokens: list[dict], scenario: str)
| 100 | return True |
| 101 | |
| 102 | async def _select_round_robin(self, tokens: list[dict], scenario: str) -> Optional[dict]: |
| 103 | """Select candidate in round-robin order for the given scenario.""" |
| 104 | if not tokens: |
| 105 | return None |
| 106 | |
| 107 | tokens_sorted = sorted(tokens, key=lambda item: item["token"].id or 0) |
| 108 | async with self._rr_lock: |
| 109 | last_id = self._round_robin_state.get(scenario) |
| 110 | start_idx = 0 |
| 111 | if last_id is not None: |
| 112 | for idx, item in enumerate(tokens_sorted): |
| 113 | if item["token"].id == last_id: |
| 114 | start_idx = (idx + 1) % len(tokens_sorted) |
| 115 | break |
| 116 | selected = tokens_sorted[start_idx] |
| 117 | self._round_robin_state[scenario] = selected["token"].id |
| 118 | return selected |
| 119 | |
| 120 | async def _check_extension_route(self, token: Token) -> tuple[bool, str]: |
| 121 | """Ensure extension captcha requests are routed to the selected account.""" |