File size: 1,690 Bytes
f6928fa d184447 f6928fa d184447 f6928fa d184447 f6928fa d184447 f6928fa d184447 f6928fa d184447 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | import logging
import aiohttp
from abc import ABC, abstractmethod
from typing import Dict, List, Any
from .. import config
logger = logging.getLogger("free_api.providers")
PROVIDER_PRIORITY = {"kilo": 0, "openrouter": 1, "zen": 2}
class Provider(ABC):
name: str
base_url: str
models_path: str = "/models"
@abstractmethod
def is_available(self) -> bool:
...
@abstractmethod
def is_free_model(self, model: Dict[str, Any]) -> bool:
...
@abstractmethod
def get_chat_url(self) -> str:
...
@abstractmethod
def get_headers(self, model_id: str) -> Dict[str, str]:
...
def priority(self) -> int:
return PROVIDER_PRIORITY.get(self.name, 99)
async def fetch_models(self) -> List[Dict[str, Any]]:
url = f"{self.base_url}{self.models_path}"
headers = self._fetch_headers()
timeout = aiohttp.ClientTimeout(total=config.get_request_timeout())
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, timeout=timeout) as response:
if response.status == 200:
data = await response.json()
return data.get("data", [])
logger.warning("[%s] fetch_models HTTP %d", self.name, response.status)
return []
except Exception as e:
logger.error("[%s] fetch_models error: %s", self.name, e)
return []
def _fetch_headers(self) -> Dict[str, str]:
return {}
def get_provider_priority(provider_name: str) -> int:
return PROVIDER_PRIORITY.get(provider_name, 99)
|