| import http.client |
| import json |
| import os |
| import requests |
| SERPER_API_KEYS = [ |
| os.getenv("SERPER_API_KEY_1", ""), |
| os.getenv("SERPER_API_KEY_2", ""), |
| os.getenv("SERPER_API_KEY_3", ""), |
| ] |
|
|
| CURRENT_SERPER_INDEX = 0 |
|
|
|
|
| def get_serper_credits(api_key: str) -> int: |
| try: |
| conn = http.client.HTTPSConnection("google.serper.dev") |
| headers = {"X-API-KEY": api_key} |
| conn.request("GET", "/credits", headers=headers) |
| res = conn.getresponse() |
| data = json.loads(res.read().decode("utf-8")) |
| return int(data.get("credits", 0)) |
| except: |
| return 0 |
|
|
|
|
| def get_valid_serper_key(min_credits=100): |
| global CURRENT_SERPER_INDEX |
|
|
| for idx, key in enumerate(SERPER_API_KEYS): |
| if not key: |
| continue |
| credits = get_serper_credits(key) |
| if credits >= min_credits: |
| CURRENT_SERPER_INDEX = idx |
| return key, credits, CURRENT_SERPER_INDEX |
|
|
| raise Exception( |
| "❌ SERPER credits exhausted on all APIs. Please contact the developer. " |
| "❌ يجب التواصل مع المطور لإضافة API Keys جديدة لSERPER" |
| ) |
|
|
