Check for blocking quota errors immediately.
(page: AsyncPage, req_id: str)
| 37 | |
| 38 | |
| 39 | async def check_quota_limit(page: AsyncPage, req_id: str) -> None: |
| 40 | """Check for blocking quota errors immediately.""" |
| 41 | # 1. Check Global State first |
| 42 | if GlobalState.IS_QUOTA_EXCEEDED: |
| 43 | raise QuotaExceededError("Global Quota Exceeded Flag is Active.") |
| 44 | |
| 45 | try: |
| 46 | # 2. Check UI for Quota Error (New Selector) |
| 47 | if await page.locator(QUOTA_EXCEEDED_SELECTOR).count() > 0: |
| 48 | element = page.locator(QUOTA_EXCEEDED_SELECTOR).first |
| 49 | if await element.is_visible(timeout=500): |
| 50 | text = await element.text_content() |
| 51 | if text and "user has exceeded quota" in text.lower(): |
| 52 | logger.critical( |
| 53 | f"[{req_id}] ❌ Quota Limit Detected via UI! Text: {text}" |
| 54 | ) |
| 55 | GlobalState.set_quota_exceeded(message=text) |
| 56 | raise QuotaExceededError(f"Quota exceeded detected via UI: {text}") |
| 57 | |
| 58 | # 3. Check UI for Quota Error (Old Selector - Legacy Fallback) |
| 59 | quota_selector = ( |
| 60 | 'ms-callout.warning-callout:has-text("You are out of free generations")' |
| 61 | ) |
| 62 | if await page.locator(quota_selector).count() > 0: |
| 63 | if await page.locator(quota_selector).first.is_visible(timeout=500): |
| 64 | logger.critical( |
| 65 | f"[{req_id}] ❌ Quota Limit Detected (Legacy)! Account is out of free generations." |
| 66 | ) |
| 67 | GlobalState.set_quota_exceeded( |
| 68 | message="AI Studio Account is out of free generations" |
| 69 | ) |
| 70 | raise QuotaExceededError( |
| 71 | "AI Studio Account is out of free generations." |
| 72 | ) |
| 73 | |
| 74 | except QuotaExceededError: |
| 75 | raise |
| 76 | except Exception as e: |
| 77 | # Don't let check errors block the main flow, unless it's the quota error itself |
| 78 | logger.warning(f"[{req_id}] Error checking for quota limit: {e}") |
| 79 | |
| 80 | |
| 81 | async def get_raw_text_content( |
no test coverage detected