mirror of https://github.com/xtekky/gpt4free
H Lohaus
9 months ago
committed by
GitHub
commit
b1b8ed40a4
@ -0,0 +1,54 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from .typing import Messages, CreateResult, Union
|
||||
|
||||
class BaseProvider(ABC):
|
||||
url: str
|
||||
working: bool = False
|
||||
needs_auth: bool = False
|
||||
supports_stream: bool = False
|
||||
supports_gpt_35_turbo: bool = False
|
||||
supports_gpt_4: bool = False
|
||||
supports_message_history: bool = False
|
||||
params: str
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def create_completion(
|
||||
cls,
|
||||
model: str,
|
||||
messages: Messages,
|
||||
stream: bool,
|
||||
**kwargs
|
||||
) -> CreateResult:
|
||||
raise NotImplementedError()
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
async def create_async(
|
||||
cls,
|
||||
model: str,
|
||||
messages: Messages,
|
||||
**kwargs
|
||||
) -> str:
|
||||
raise NotImplementedError()
|
||||
|
||||
@classmethod
|
||||
def get_dict(cls):
|
||||
return {'name': cls.__name__, 'url': cls.url}
|
||||
|
||||
class BaseRetryProvider(BaseProvider):
|
||||
__name__: str = "RetryProvider"
|
||||
supports_stream: bool = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
providers: list[type[BaseProvider]],
|
||||
shuffle: bool = True
|
||||
) -> None:
|
||||
self.providers: list[type[BaseProvider]] = providers
|
||||
self.shuffle: bool = shuffle
|
||||
self.working: bool = True
|
||||
self.exceptions: dict[str, Exception] = {}
|
||||
self.last_provider: type[BaseProvider] = None
|
||||
|
||||
ProviderType = Union[type[BaseProvider], BaseRetryProvider]
|
||||
@ -1,45 +1,5 @@
|
||||
from os import environ
|
||||
import requests
|
||||
from importlib.metadata import version as get_package_version, PackageNotFoundError
|
||||
from subprocess import check_output, CalledProcessError, PIPE
|
||||
from .errors import VersionNotFoundError
|
||||
from .base_provider import ProviderType
|
||||
|
||||
logging = False
|
||||
version_check = True
|
||||
|
||||
def get_version() -> str:
|
||||
# Read from package manager
|
||||
try:
|
||||
return get_package_version("g4f")
|
||||
except PackageNotFoundError:
|
||||
pass
|
||||
# Read from docker environment
|
||||
current_version = environ.get("G4F_VERSION")
|
||||
if current_version:
|
||||
return current_version
|
||||
# Read from git repository
|
||||
try:
|
||||
command = ["git", "describe", "--tags", "--abbrev=0"]
|
||||
return check_output(command, text=True, stderr=PIPE).strip()
|
||||
except CalledProcessError:
|
||||
pass
|
||||
raise VersionNotFoundError("Version not found")
|
||||
|
||||
def get_latest_version() -> str:
|
||||
if environ.get("G4F_VERSION"):
|
||||
url = "https://registry.hub.docker.com/v2/repositories/"
|
||||
url += "hlohaus789/g4f"
|
||||
url += "/tags?page_size=2&ordering=last_updated"
|
||||
response = requests.get(url).json()
|
||||
return response["results"][1]["name"]
|
||||
response = requests.get("https://pypi.org/pypi/g4f/json").json()
|
||||
return response["info"]["version"]
|
||||
|
||||
def check_pypi_version() -> None:
|
||||
try:
|
||||
version = get_version()
|
||||
latest_version = get_latest_version()
|
||||
if version != latest_version:
|
||||
print(f'New pypi version: {latest_version} (current: {version}) | pip install -U g4f')
|
||||
except Exception as e:
|
||||
print(f'Failed to check g4f pypi version: {e}')
|
||||
logging: bool = False
|
||||
version_check: bool = True
|
||||
last_provider: ProviderType = None
|
||||
@ -0,0 +1,47 @@
|
||||
from os import environ
|
||||
import requests
|
||||
from functools import cached_property
|
||||
from importlib.metadata import version as get_package_version, PackageNotFoundError
|
||||
from subprocess import check_output, CalledProcessError, PIPE
|
||||
from .errors import VersionNotFoundError
|
||||
|
||||
|
||||
class VersionUtils():
|
||||
@cached_property
|
||||
def current_version(self) -> str:
|
||||
# Read from package manager
|
||||
try:
|
||||
return get_package_version("g4f")
|
||||
except PackageNotFoundError:
|
||||
pass
|
||||
# Read from docker environment
|
||||
version = environ.get("G4F_VERSION")
|
||||
if version:
|
||||
return version
|
||||
# Read from git repository
|
||||
try:
|
||||
command = ["git", "describe", "--tags", "--abbrev=0"]
|
||||
return check_output(command, text=True, stderr=PIPE).strip()
|
||||
except CalledProcessError:
|
||||
pass
|
||||
raise VersionNotFoundError("Version not found")
|
||||
|
||||
@cached_property
|
||||
def latest_version(self) -> str:
|
||||
try:
|
||||
get_package_version("g4f")
|
||||
response = requests.get("https://pypi.org/pypi/g4f/json").json()
|
||||
return response["info"]["version"]
|
||||
except PackageNotFoundError:
|
||||
url = "https://api.github.com/repos/xtekky/gpt4free/releases/latest"
|
||||
response = requests.get(url).json()
|
||||
return response["tag_name"]
|
||||
|
||||
def check_pypi_version(self) -> None:
|
||||
try:
|
||||
if self.current_version != self.latest_version:
|
||||
print(f'New pypi version: {self.latest_version} (current: {self.version}) | pip install -U g4f')
|
||||
except Exception as e:
|
||||
print(f'Failed to check g4f pypi version: {e}')
|
||||
|
||||
utils = VersionUtils()
|
||||
Loading…
Reference in New Issue