Process the request data before sending to the server
(
self, request_data: Union[int, bytes], host: str, path: str
)
| 73 | return False |
| 74 | |
| 75 | async def process_request( |
| 76 | self, request_data: Union[int, bytes], host: str, path: str |
| 77 | ) -> Union[int, bytes]: |
| 78 | """ |
| 79 | Process the request data before sending to the server |
| 80 | """ |
| 81 | if not self.should_intercept(host, path): |
| 82 | return request_data |
| 83 | |
| 84 | # Log the request |
| 85 | self.logger.debug(f"[Network] Intercepted request: {host}{path}") |
| 86 | |
| 87 | # Check for Quota Exceeded errors in jserror requests |
| 88 | if "jserror" in path: |
| 89 | try: |
| 90 | decoded_path = unquote(path) |
| 91 | if any( |
| 92 | keyword in decoded_path |
| 93 | for keyword in [ |
| 94 | "exceeded quota", |
| 95 | "RESOURCE_EXHAUSTED", |
| 96 | "Failed to generate content", |
| 97 | ] |
| 98 | ): |
| 99 | self.logger.critical( |
| 100 | f"🚨 CRITICAL: Detected Quota Exceeded error in network traffic! URL: {path}" |
| 101 | ) |
| 102 | |
| 103 | from api_utils.server_state import state |
| 104 | |
| 105 | model_id = state.current_ai_studio_model_id |
| 106 | GlobalState.set_quota_exceeded( |
| 107 | message=decoded_path, model_id=model_id or "" |
| 108 | ) |
| 109 | except Exception as e: |
| 110 | self.logger.error(f"Error parsing jserror path: {e}") |
| 111 | |
| 112 | return request_data |
| 113 | |
| 114 | async def process_response( |
| 115 | self, |
no test coverage detected