GPT 逆向接口批量管理 | Convert reverse engineered LLM libs to OpenAI API format & channel management. | 使用 OpenAI 标准 API 访问 revChatGPT 等逆向工程库 | Supported ChatGPT, gpt4free, huggingchat, 通义千问, Bard, Claude | 免费 Chat GPT Free ChatGPT
from hashlib import md5

def md5_digest(s: str) -> str:
    # 创建md5对象
    new_md5 = md5()
    # 这里必须用encode()函数对字符串进行编码，不然会报 TypeError: Unicode-objects must be encoded before hashing
    new_md5.update(s.encode(encoding='utf-8'))
    # 加密
    return new_md5.hexdigest()
free_one_api/common/crypto.py
RockChinQ/free-one-api
"""Utility functions for key generation."""

import random
import string


def generate_api_key() -> str:
    """Generate API key which starts with `sk-foa`, 51 chars in total.
    """
    result = "sk-foa" + "".join(
        [random.choice(string.ascii_letters + string.digits) for _ in range(45)]
    )

    return result

free_one_api/common/key.py
RockChinQ/free-one-api
import random
import typing

ads = []

rate = 0.01

enabled = False

def generate_ad() -> typing.Generator[str, None, None]:
    """Generate random ad."""
    global ads
    global rate
    global enabled
    
    if not enabled:
        return
    
    if len(ads) == 0:
        return
    
    if random.random() < rate:
        ad_words = random.choice(ads).split(" ")
        
        for word in ad_words:
            yield word
            yield " "

free_one_api/common/randomad.py
RockChinQ/free-one-api
import datetime

from ..common import key


class FreeOneAPIKey:
    """API key."""
    id: int

    name: str
    """Name of this API key."""

    created_at: int
    """Created at."""

    raw: str
    """API key."""

    def __init__(self, id: str, name: str, created_at: int, raw: str):
        self.id = id
        self.name = name
        self.created_at = created_at
        self.raw = raw

    @classmethod
    def make_new(cls, name: str) -> 'FreeOneAPIKey':
        return cls(
            -1,
            name=name,
            created_at=datetime.datetime.now().timestamp(),
            raw=key.generate_api_key(),
        )

free_one_api/entities/apikey.py
RockChinQ/free-one-api
import asyncio
import json
import time

import tiktoken

from ..models.adapter import llm
from ..models import adapter


class Channel:
    """Entity for channel."""
    id: int

    name: str
    """Name of this channel."""

    adapter: llm.LLMLibAdapter

    model_mapping: dict
    """Mapping model name to another model name."""

    enabled: bool

    latency: int
    
    fail_count: int
    """Amount of sequential failures. Only in memory."""

    def __init__(self, id: int, name: str, adapter: llm.LLMLibAdapter, model_mapping: dict, enabled: bool, latency: int):
        self.id = id
        self.name = name
        self.adapter = adapter
        self.model_mapping = model_mapping
        self.enabled = enabled
        self.latency = latency
        self.fail_count = 0

    @classmethod
    def dump_channel(cls, chan: 'Channel') -> dict:
        return {
            "id": chan.id,
            "name": chan.name,
            "adapter": adapter.dump_adapter(chan.adapter),
            "model_mapping": chan.model_mapping,
            "enabled": chan.enabled,
            "latency": chan.latency,
        }

    @classmethod
    def load_channel(cls, data: dict) -> 'Channel':
        return cls(
            data["id"],
            data["name"],
            adapter.load_adapter(data["adapter"]),
            data["model_mapping"],
            data["enabled"],
            data["latency"],
        )

    def count_tokens(
        self,
        model: str,
        messages: list[str],
    ) -> int:
        """Count message tokens."""
        try:
            encoding = tiktoken.encoding_for_model(model)
        except KeyError:
            encoding = tiktoken.get_encoding("cl100k_base")
        num_tokens = 0
        for message in messages:
            for key, value in message.items():
                num_tokens += len(encoding.encode(value))
        num_tokens += 3  # every reply is primed with <|start|>assistant<|message|>
        return num_tokens
    
    async def heartbeat(self, timeout: int=300) -> int:
        """Call adapter test, returns fail count."""
        
        try:
            start = time.time()
            succ, err = await asyncio.wait_for(self.adapter.test(), timeout=timeout)
            if succ:
                latency = int((time.time() - start)*100)/100
                self.fail_count = 0
                self.latency = latency
                return 0
            else:
                self.fail_count += 1
                return self.fail_count
        finally:
            self.fail_count += 1
            return self.fail_count

free_one_api/entities/channel.py
RockChinQ/free-one-api


class QueryHandlingError(Exception):
    
    status_code: int
    code: str
    message: str
    type: str
    param: str
    
    def __init__(self, status_code: int, code: str, message: str, type: str=None, param: str=None):
        """Raise this exception when the query handling failed.
        
        Args:
            code (int): The http status code that should be returned to the client, please refer to the OpenAI API document.
            message (str): The error message that should be returned to the client.
            type (str): The error type that should be returned to the client.
            param (str): The error param that should be returned to the client.
        """
        self.status_code = status_code
        self.code = code
        self.message = message
        self.type = type
        self.param = param

free_one_api/entities/exceptions.py
RockChinQ/free-one-api

class Request:
    """Request from http interface.

    Send to LLM lib adapters.
    """
    model: str
    """LLM model name."""

    messages: list[dict[str, str]]
    """OpenAI GPT style message list."""

    functions: list[dict[str, str]]
    """OpenAI ChatCompletion format function list."""

    stream: bool
    """True if this is a streaming request, processed by http interface level."""

    def __init__(
        self,
        model: str,
        messages: list[dict[str, str]],
        functions: list[dict[str, str]],
        stream: bool=False,
    ):
        self.model = model
        self.messages = messages
        self.functions = functions
        self.stream = stream

free_one_api/entities/request.py
RockChinQ/free-one-api
"""Entity for both one-time and streaming response."""
import enum


class FinishReason(enum.Enum):
    """Finish reason type enum."""

    NULL = None
    """Null if this is a part of a streaming response."""

    FUNCTION_CALL = "function_call"
    """Function call requested in this response, the ending of a streaming response."""

    STOP = "stop"
    """Normal message response, the ending of a streaming response."""

    LENGTH = "length"
    """Reponse exceeded the length limit, the ending of a streaming response."""


class FunctionCall:
    """Function call."""

    function_name: str

    arguments: dict

    def __init__(self, function_name: str, arguments: dict):
        self.function_name = function_name
        self.arguments = arguments


class Response:
    """Entity for both one-time and streaming response.
    
    Be created by LLM lib adapters. Pass between LLM lib adapters and protocol wrapper(http interface).
    """
    id: str
    """Set by upstream lib, used to identify this response."""

    finish_reason: FinishReason
    """The reason why this response responded."""

    normal_message: str
    """Text of the normal message."""

    function_call: FunctionCall
    """Function call."""

    def __init__(self, id: str, finish_reason: FinishReason, normal_message: str = None, function_call: FunctionCall = None):
        self.id = id
        self.finish_reason = finish_reason
        self.normal_message = normal_message
        self.function_call = function_call
    
    def __str__(self) -> str:
        return f"Response({self.id}, {self.finish_reason}, {self.normal_message}, {self.function_call})"    
    
free_one_api/entities/response.py
RockChinQ/free-one-api
import typing
import traceback
import uuid
import random

import bardapi as bard

from free_one_api.entities import request, response

from ...models import adapter
from ...models.adapter import llm
from ...entities import request, response, exceptions


@adapter.llm_adapter
class BardAdapter(llm.LLMLibAdapter):
    
    @classmethod
    def name(cls) -> str:
        return "dsdanielpark/Bard-API"
    
    @classmethod
    def description(self) -> str:
        return "Use dsdanielpark/Bard-API to access Google Bard web edition."

    def supported_models(self) -> list[str]:
        return [
            "gpt-3.5-turbo",
            "gpt-4"
        ]

    def function_call_supported(self) -> bool:
        return False

    def stream_mode_supported(self) -> bool:
        return False

    def multi_round_supported(self) -> bool:
        return True
    
    @classmethod
    def config_comment(cls) -> str:
        return \
"""Currently supports non stream mode only.
You should provide __Secure-1PSID as token extracted from cookies of Bard site.

{
    "token": "bQhxxxxxxxxxxx"
}

Method of getting __Secure-1PSID string, please refer to https://github.com/dsdanielpark/Bard-API
"""

    @classmethod
    def supported_path(cls) -> str:
        return "/v1/chat/completions"
    
    _chatbot: bard.Bard = None
    
    @property
    def chatbot(self) -> bard.Bard:
        if self._chatbot == None:
            self._chatbot = bard.Bard(token=self.config['token'])
        return self._chatbot
    
    def __init__(self, config: dict):
        self.config = config
        
    async def test(self) -> (bool, str):
        try:
            self.chatbot.get_answer("hello, please reply 'hi' only.")
            return True, ""
        except Exception as e:
            traceback.print_exc()
            return False, str(e)

    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:
        prompt = ""
        
        for msg in req.messages:
            prompt += f"{msg['role']}: {msg['content']}\n"
        
        prompt += "assistant: "
        
        random_int = random.randint(0, 1000000000)
        
        resp_text = self.chatbot.get_answer(prompt)['content']
        
        yield response.Response(
            id=random_int,
            finish_reason=response.FinishReason.STOP,
            normal_message=resp_text,
            function_call=None
        )

free_one_api/impls/adapter/bard.py
RockChinQ/free-one-api
import typing
import traceback
import uuid
import random

import claude_api as claude

from free_one_api.entities import request, response

from ...models import adapter
from ...models.adapter import llm
from ...entities import request, response, exceptions


@adapter.llm_adapter
class ClaudeAdapter(llm.LLMLibAdapter):
    
    @classmethod
    def name(cls) -> str:
        return "KoushikNavuluri/Claude-API"
    
    @classmethod
    def description(self) -> str:
        return "Use KoushikNavuluri/Claude-API to access Claude web edition."

    def supported_models(self) -> list[str]:
        return [
            "gpt-3.5-turbo",
            "gpt-4"
        ]

    def function_call_supported(self) -> bool:
        return False

    def stream_mode_supported(self) -> bool:
        return False

    def multi_round_supported(self) -> bool:
        return True
    
    @classmethod
    def config_comment(cls) -> str:
        return \
"""Currently supports non stream mode only.
You should provide cookie string as `cookie` in config:
{
    "cookie": "your cookie string"
}

Method of getting cookie string, please refer to https://github.com/KoushikNavuluri/Claude-API
"""

    @classmethod
    def supported_path(cls) -> str:
        return "/v1/chat/completions"
    
    chatbot: claude.Client
    
    def __init__(self, config: dict):
        self.config = config
        self.chatbot = claude.Client(self.config["cookie"])
        
    async def test(self) -> (bool, str):
        try:
            conversation_id = self.chatbot.create_new_chat()['uuid']
            response = self.chatbot.send_message("Hello, Claude!", conversation_id)
            return True, ""
        except Exception as e:
            traceback.print_exc()
            return False, str(e)
        
    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:
        prompt = ""
        
        for msg in req.messages:
            prompt += f"{msg['role']}: {msg['content']}\n"
        
        prompt += "assistant: "
        
        random_int = random.randint(0, 1000000000)
        
        conversation_id = self.chatbot.create_new_chat()['uuid']
        resp_text = self.chatbot.send_message(prompt, conversation_id)
        
        self.chatbot.delete_conversation(conversation_id)
        
        yield response.Response(
            id=random_int,
            finish_reason=response.FinishReason.STOP,
            normal_message=resp_text,
            function_call=None
        )

free_one_api/impls/adapter/claude.py
RockChinQ/free-one-api
import typing
import traceback
import uuid
import logging
import random

import g4f

from free_one_api.entities import request, response

from ...models import adapter
from ...models.adapter import llm
from ...entities import request, response, exceptions


@adapter.llm_adapter
class GPT4FreeAdapter(llm.LLMLibAdapter):
    
    @classmethod
    def name(cls) -> str:
        return "xtekky/gpt4free"
    
    @classmethod
    def description(self) -> str:
        return "Use xtekky/gpt4free to access lots of GPT providers."

    def supported_models(self) -> list[str]:
        return [
            "gpt-3.5-turbo",
            "gpt-4"
        ]

    def function_call_supported(self) -> bool:
        return False

    def stream_mode_supported(self) -> bool:
        return True

    def multi_round_supported(self) -> bool:
        return True
    
    @classmethod
    def config_comment(cls) -> str:
        return \
"""GPT4Free is so unstable that it is not recommended to use.
You don't need to provide any authentification.

Please refer to https://github.com/xtekky/gpt4free
"""

    @classmethod
    def supported_path(cls) -> str:
        return "/v1/chat/completions"
    
    def __init__(self, config: dict):
        self.config = config
        
    _use_provider: g4f.Provider = None
    _use_stream_provider: g4f.Provider = None
    
    async def use_provider(self, stream: bool) -> g4f.Provider.BaseProvider:
        if self._use_provider is None:
            await self._select_provider()
        if stream and self._use_stream_provider is not None:
            return self._use_stream_provider
        return self._use_provider
    
    async def _select_provider(self):
        non_stream_tested = False
        if self._use_provider is not None:
            try:
                resp = await g4f.ChatCompletion.create_async(
                    model="gpt-3.5-turbo",
                    messages=[
                        {
                            "role": "user",
                            "content": "Hi, My name is Rock."
                        }
                    ],
                    provider=self._use_provider
                )
                non_stream_tested = True
            except Exception as e:
                self._use_provider = None
        if non_stream_tested and self._use_stream_provider is not None:
            try:
                resp = self._use_stream_provider.create_async_generator(
                    model="gpt-3.5-turbo",
                    messages=[
                        {
                            "role": "user",
                            "content": "Hi, My name is Rock."
                        }
                    ]
                )
                async for _ in resp:
                    return
            except Exception as e:
                self._use_stream_provider = None
        
        self._use_provider = None
        self._use_stream_provider = None

        from g4f.Provider import __all__ as providers

        exclude = [
            'Acytoo',
            'BaseProvider'
        ]

        for provider in providers:

            # print("Testing provider", provider)
            # logging.info("Testing provider %s", provider)

            if provider in exclude:
                continue

            provider = getattr(g4f.Provider, provider)
            
            try:
                assert hasattr(provider, 'supports_stream')
                resp = await g4f.ChatCompletion.create_async(
                    model="gpt-3.5-turbo",
                    messages=[
                        {
                            "role": "user",
                            "content": "Hi, My name is Rock."
                        }
                    ],
                    provider=provider
                )

                if 'Rock' in resp and '<' not in resp:
                    if self._use_provider is None:
                        self._use_provider = provider
                    
                    if provider.supports_stream:
                        try:
                            assert hasattr(provider, 'create_async_generator')
                            resp = provider.create_async_generator(
                                model="gpt-3.5-turbo",
                                messages=[
                                    {
                                        "role": "user",
                                        "content": "Hi, My name is Rock."
                                    }
                                ],
                                timeout=120
                            )
                            async for _ in resp:
                                pass
                            if self._use_stream_provider is None:
                                self._use_stream_provider = provider
                        except Exception as e:
                            traceback.print_exc()
                            print("provider", provider, "does not really support stream mode")
                        
                    
                    if self._use_provider is not None and self._use_stream_provider is not None:
                        print("selected provider", self._use_provider, self._use_stream_provider)
                        break
            except Exception as e:
                # traceback.print_exc()
                continue
            
        if self._use_provider is None:
            raise exceptions.QueryHandlingError(404, "no_provider_found", "No provider available.")

    async def test(self) -> (bool, str):
        try:
            await self._select_provider()
            resp = await g4f.ChatCompletion.create_async(
                model="gpt-3.5-turbo",
                messages=[{
                    "role": "user",
                    "content": "Hello, please reply 'hi' only."
                }],
                provider=await self.use_provider(stream=False)
            )
            return True, ""
        except Exception as e:
            traceback.print_exc()
            return False, str(e)
        
    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:
        provider = await self.use_provider(stream=True)
        
        if not req.stream:
            resp = await g4f.ChatCompletion.create_async(
                model=req.model,
                messages=req.messages,
                provider=provider
            )
        else:
            resp = provider.create_async_generator(
                model=req.model,
                messages=req.messages,
                timeout=120
            )
        
        if isinstance(resp, typing.Generator):
            for resp_text in resp:
                random_int = random.randint(0, 1000000000)
                yield response.Response(
                    id=random_int,
                    finish_reason=response.FinishReason.NULL,
                    normal_message=resp_text,
                    function_call=None
                )
            yield response.Response(
                id=random_int,
                finish_reason=response.FinishReason.STOP,
                normal_message="",
                function_call=None
            )
        elif isinstance(resp, typing.AsyncGenerator):
            async for resp_text in resp:
            
                random_int = random.randint(0, 1000000000)
                
                yield response.Response(
                    id=random_int,
                    finish_reason=response.FinishReason.NULL,
                    normal_message=resp_text,
                    function_call=None
                )
            yield response.Response(
                id=random_int,
                finish_reason=response.FinishReason.STOP,
                normal_message="",
                function_call=None
            )
        else:
            random_int = random.randint(0, 1000000000)
            
            yield response.Response(
                id=random_int,
                finish_reason=response.FinishReason.STOP,
                normal_message=resp,
                function_call=None
            )

free_one_api/impls/adapter/gpt4free.py
RockChinQ/free-one-api
import typing
import traceback
import uuid
import random

import requests

import hugchat.hugchat as hugchat
import hugchat.login as login

from ...models import adapter
from ...models.adapter import llm
from ...entities import request
from ...entities import response, exceptions


@adapter.llm_adapter
class HuggingChatAdapter(llm.LLMLibAdapter):
    
    @classmethod
    def name(cls) -> str:
        return "Soulter/hugging-chat-api"
    
    @classmethod
    def description(self) -> str:
        return "Use Soulter/hugging-chat-api to access reverse engineering huggingchat."

    def supported_models(self) -> list[str]:
        return [
            "gpt-3.5-turbo",
            "gpt-4"
        ]

    def function_call_supported(self) -> bool:
        return False

    def stream_mode_supported(self) -> bool:
        return True    

    def multi_round_supported(self) -> bool:
        return True
    
    @classmethod
    def config_comment(cls) -> str:
        return \
"""Please provide email and passwd to sign up for HuggingChat:

{
    "email": "your email",
    "passwd": "your password"
}

Please refer to https://github.com/Soulter/hugging-chat-api
"""
    
    @classmethod
    def supported_path(self) -> str:
        return "/v1/chat/completions"
    
    _chatbot: hugchat.ChatBot = None
    
    @property
    def chatbot(self) -> hugchat.ChatBot:
        if self._chatbot is None:
            sign = login.Login(self.config['email'], self.config['passwd'])
            cookie: requests.sessions.RequestsCookieJar = None
            try:
                cookie = sign.loadCookiesFromDir("data/hugchatCookies")
            except:
                cookie = sign.login()
                sign.saveCookiesToDir("data/hugchatCookies")
            
            self._chatbot = hugchat.ChatBot(cookies=cookie.get_dict())
        return self._chatbot
    
    def __init__(self, config: dict):
        self.config = config
    
    async def test(self) -> (bool, str):
        try:
            self.chatbot.change_conversation(self.chatbot.new_conversation())
            for data in self.chatbot.query(
                "Hi, respond 'Hello, world!' please.",
                stream=True
            ):
                pass
            
            self.chatbot.delete_conversation(self.chatbot.current_conversation)
            
            return True, ""
        except Exception as e:
            traceback.print_exc()
            return False, str(e)
    
    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:        
        prompt = ""
        
        for msg in req.messages:
            prompt += f"{msg['role']}: {msg['content']}\n"
        
        prompt += "assistant: "
        
        random_int = random.randint(0, 1000000000)
        self.chatbot.change_conversation(self.chatbot.new_conversation())
        
        for resp in self.chatbot.query(
            text=prompt,
            stream=True
        ):
            yield response.Response(
                id=random_int,
                finish_reason=response.FinishReason.NULL,
                normal_message=resp['token'],
                function_call=None
            )
        self.chatbot.delete_conversation(self.chatbot.current_conversation)

        yield response.Response(
            id=random_int,
            finish_reason=response.FinishReason.STOP,
            normal_message="",
            function_call=None
        )

free_one_api/impls/adapter/hugchat.py
RockChinQ/free-one-api
import typing
import traceback
import uuid
import random

import revTongYi.qianwen as qwen

from free_one_api.entities import request, response

from ...models import adapter
from ...models.adapter import llm
from ...entities import request, response, exceptions


@adapter.llm_adapter
class QianWenAdapter(llm.LLMLibAdapter):
    
    @classmethod
    def name(cls) -> str:
        return "xw5xr6/revTongYi"
    
    @classmethod
    def description(self) -> str:
        return "Use xw5xr6/revTongYi to access Aliyun TongYi QianWen."

    def supported_models(self) -> list[str]:
        return [
            "gpt-3.5-turbo",
            "gpt-4"
        ]

    def function_call_supported(self) -> bool:
        return False

    def stream_mode_supported(self) -> bool:
        return True

    def multi_round_supported(self) -> bool:
        return True
    
    @classmethod
    def config_comment(cls) -> str:
        return \
"""RevTongYi use cookies that can be extracted from https://qianwen.aliyun.com/
You should provide cookie string as `cookie` in config:
{
    "cookie": "your cookie string"
}

Method of getting cookie string, please refer to https://github.com/xw5xr6/revTongYi
"""

    @classmethod
    def supported_path(cls) -> str:
        return "/v1/chat/completions"
    
    chatbot: qwen.Chatbot
    
    def __init__(self, config: dict):
        self.config = config
        self.chatbot = qwen.Chatbot(
            cookies_str=config['cookie']
        )
        
    async def test(self) -> (bool, str):
        try:
            self.chatbot.create_session("Hello, reply 'hi' only.")
            resp = self.chatbot.ask(
                "Hello, reply 'hi' only.",
            )
            
            self.chatbot.delete_session(self.chatbot.sessionId)
            
            return True, ""
        except Exception as e:
            traceback.print_exc()
            return False, str(e)
        
    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:
        prompt = ""
        
        for msg in req.messages:
            prompt += f"{msg['role']}: {msg['content']}\n"
        
        prompt += "assistant: "
        
        random_int = random.randint(0, 1000000000)
        
        prev_text = ""
        self.chatbot.create_session(prompt)
        
        for resp in self.chatbot.ask(
            prompt=prompt,
            stream=True,
        ):
            
            yield response.Response(
                id=random_int,
                finish_reason=response.FinishReason.NULL,
                normal_message=resp['content'][0].replace(prev_text, ""),
                function_call=None
            )
            prev_text = resp['content'][0]
        
        self.chatbot.delete_session(self.chatbot.sessionId)
        
        yield response.Response(
            id=random_int,
            finish_reason=response.FinishReason.STOP,
            normal_message="",
            function_call=None
        )

free_one_api/impls/adapter/qianwen.py
RockChinQ/free-one-api
import typing
import traceback
import uuid
import random

import revChatGPT.V1 as chatgpt

from ...models import adapter
from ...models.adapter import llm
from ...entities import request
from ...entities import response, exceptions


@adapter.llm_adapter
class RevChatGPTAdapter(llm.LLMLibAdapter):
    
    @classmethod
    def name(cls) -> str:
        return "acheong08/ChatGPT"
    
    @classmethod
    def description(self) -> str:
        return "Use acheong08/ChatGPT to access reverse engineering OpenAI ChatGPT web edition."

    def supported_models(self) -> list[str]:
        return [
            "gpt-3.5-turbo",
            "gpt-4"
        ]

    def function_call_supported(self) -> bool:
        return False

    def stream_mode_supported(self) -> bool:
        return True    

    def multi_round_supported(self) -> bool:
        return True
    
    @classmethod
    def config_comment(cls) -> str:
        return \
"""You can provide `access_token` or `email/password` to config as:
{
    "access_token": "your access token",
}
or
{
    "email": "your email",
    "password": "your password"
}
and you can also provide other optional params supported by acheong08/ChatGPT:
{
  "conversation_id": "UUID...",
  "parent_id": "UUID...",
  "proxy": "...",
  "model": "gpt-4", // gpt-4-browsing, text-davinci-002-render-sha, gpt-4, gpt-4-plugins
  "plugin_ids": ["plugin-d1d6eb04-3375-40aa-940a-c2fc57ce0f51"], // Wolfram Alpha example
  "disable_history": true,
  "PUID": "<_puid cookie for plus accounts>", // Only if you have a plus account and use GPT-4
  "unverified_plugin_domains":["showme.redstarplugin.com"] // Unverfied plugins to install
}

Please refer to https://github.com/acheong08/ChatGPT
"""
    
    @classmethod
    def supported_path(self) -> str:
        return "/v1/chat/completions"
    
    chatbot: chatgpt.AsyncChatbot
    
    def __init__(self, config: dict):
        self.config = config
        self.chatbot = chatgpt.AsyncChatbot(
            config=config,
            base_url="https://chatproxy.rockchin.top/api/"
        )
    
    async def test(self) -> (bool, str):
        try:
            prev_text = ""
            conversation_id = ""
            self.chatbot.conversation_id = None
            async for data in self.chatbot.ask(
                "Hi, respond 'Hello, world!' please.",
            ):
                message = data["message"][len(prev_text):]
                prev_text = data["message"]
                conversation_id = data["conversation_id"]
                
            await self.chatbot.delete_conversation(conversation_id)
            return True, ""
        except Exception as e:
            traceback.print_exc()
            return False, str(e)
    
    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:        
        new_messages = []
        for i in range(len(req.messages)):
            new_messages.append({
                "id": str(uuid.uuid4()),
                "author": {"role": req.messages[i]['role']},
                "content": {
                    "content_type": "text",
                    "parts": [
                        req.messages[i]['content']
                    ]
                }
            })
        
        random_int = random.randint(0, 1000000000)
        
        prev_text = ""
        
        try:
            conversation_id = ""
            self.chatbot.conversation_id = None
            async for data in self.chatbot.post_messages(
                messages=new_messages,
            ):
                message = data["message"][len(prev_text):]
                prev_text = data["message"]
                conversation_id = data["conversation_id"]
                
                yield response.Response(
                    id=random_int,
                    finish_reason=response.FinishReason.NULL,
                    normal_message=message,
                    function_call=None
                )
                random_int += 1
                
            yield response.Response(
                id=random_int,
                finish_reason=response.FinishReason.STOP,
                normal_message="",
                function_call=None
            )
            
            await self.chatbot.delete_conversation(conversation_id)
        except chatgpt.t.Error as e:
            assert isinstance(e, chatgpt.t.ErrorType)
            code = e.code.name.lower()
            
            raise exceptions.QueryHandlingError(
                status_code=500,
                code=code,
                message=e.message,
            )

free_one_api/impls/adapter/revChatGPT.py
RockChinQ/free-one-api
import os
import sys
import asyncio
import logging
import colorlog

import yaml

from ..common import crypto
from .router import mgr as routermgr
from .router import api as apigroup

from ..models.database import db
from ..models.channel import mgr as chanmgr
from ..models.key import mgr as keymgr
from ..models.router import group as routergroup
from ..models.watchdog import wd as wdmgr

from .adapter import revChatGPT
from .adapter import claude
from .adapter import bard
from .adapter import gpt4free
from .adapter import hugchat
from .adapter import qianwen

from . import log


class Application:
    """Application instance."""

    dbmgr: db.DatabaseInterface
    """Database manager."""""

    router: routermgr.RouterManager
    """Router manager."""

    channel: chanmgr.AbsChannelManager
    """Channel manager."""
    
    key: keymgr.AbsAPIKeyManager
    """API Key manager."""
    
    watchdog: wdmgr.AbsWatchDog
    
    logging_level: int = logging.INFO
    
    def __init__(
        self,
        dbmgr: db.DatabaseInterface,
        router: routermgr.RouterManager,
        channel: chanmgr.AbsChannelManager,
        key: keymgr.AbsAPIKeyManager,
        watchdog: wdmgr.AbsWatchDog,
        logging_level: int = logging.INFO,
    ):
        self.dbmgr = dbmgr
        self.router = router
        self.channel = channel
        self.key = key
        self.watchdog = watchdog
        self.logging_level = logging_level
        
    async def run(self):
        """Run application."""
        loop = asyncio.get_running_loop()
        
        loop.create_task(self.watchdog.run())
        
        await self.router.serve(loop)

log_colors_config = {
    'DEBUG': 'green',  # cyan white
    'INFO': 'white',
    'WARNING': 'yellow',
    'ERROR': 'red',
    'CRITICAL': 'cyan',
}

default_config = {
    "database": {
        "type": "sqlite",
        "path": "./data/free_one_api.db",
    },
    "watchdog": {
        "heartbeat": {
            "interval": 1800,
            "timeout": 300,
            "fail_limit": 3,
        },
    },
    "router": {
        "port": 3000,
        "token": "12345678",
    },
    "web": {
        "frontend_path": "./web/dist/",
    },
    "logging": {
        "debug": False,
    },
    "random_ad": {
        "enabled": False,
        "rate": 0.05,
        "ad_list": [
            " (This response is sponsored by Free One API. Consider star the project on GitHub: https://github.com/RockChinQ/free-one-api )",
        ]
    }
}

async def make_application(config_path: str) -> Application:
    """Make application."""
    if not os.path.exists(config_path):
        with open(config_path, "w") as f:
            yaml.dump(default_config, f)
            print("Config file created at", config_path)
            # print("Please edit it and run again.")
            # sys.exit(0)
    config = {}
    with open(config_path, "r") as f:
        config = yaml.load(f, Loader=yaml.FullLoader)

    # logging
    logging_level = logging.INFO
    
    if 'logging' in config and 'debug' in config['logging'] and config['logging']['debug']:
        logging_level = logging.DEBUG
        
    if 'DEBUG' in os.environ and os.environ['DEBUG'] == 'true':
        logging_level = logging.DEBUG
    
    print("Logging level:", logging_level)
    logging.debug("Debug mode enabled.")
    
    terminal_out = logging.StreamHandler()
    
    terminal_out.setLevel(logging_level)
    terminal_out.setFormatter(colorlog.ColoredFormatter(
        "[%(asctime)s.%(msecs)03d] %(log_color)s%(pathname)s (%(lineno)d) - [%(levelname)s] :\n"
            "%(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
        log_colors=log_colors_config,
    ))
    
    for handler in logging.getLogger().handlers:
        logging.getLogger().removeHandler(handler)
    
    logging.getLogger().addHandler(terminal_out)
    
    # save ad to runtime
    if 'random_ad' in config and config['random_ad']['enabled']:
        from ..common import randomad
        
        randomad.enabled = config['random_ad']['enabled']
        randomad.rate = config['random_ad']['rate']
        randomad.ads = config['random_ad']['ad_list']
    
    from ..common import randomad

    # make database manager
    from .database import sqlite as sqlitedb
    
    dbmgr_cls_mapping = {
        "sqlite": sqlitedb.SQLiteDB,
    }
    
    dbmgr = dbmgr_cls_mapping[config['database']['type']](config['database'])
    await dbmgr.initialize()
    
    # database handler
    dblogger = log.SQLiteHandler(dbmgr)
    
    # failed to set debug level for db handler
    dblogger.setLevel(logging.INFO if logging_level <= logging.INFO else logging_level)
    dblogger.setFormatter(logging.Formatter("[%(asctime)s.%(msecs)03d] %(pathname)s (%(lineno)d) - [%(levelname)s] :\n%(message)s"))
    
    logging.getLogger().addHandler(dblogger)
    
    # make channel manager
    from .channel import mgr as chanmgr
    
    channelmgr = chanmgr.ChannelManager(dbmgr)
    await channelmgr.list_channels()
    
    # make key manager
    from .key import mgr as keymgr
    
    apikeymgr = keymgr.APIKeyManager(dbmgr)
    await apikeymgr.list_keys()
    
    # make forward manager
    from .forward import mgr as forwardmgr
    
    fwdmgr = forwardmgr.ForwardManager(channelmgr, apikeymgr)
    
    # make router manager
    from .router import mgr as routermgr
    
    #   import all api groups
    from .router import forward as forwardgroup
    from .router import api as apigroup
    from .router import web as webgroup
    
    # ========= API Groups =========
    group_forward = forwardgroup.ForwardAPIGroup(dbmgr, channelmgr, apikeymgr, fwdmgr)
    group_api = apigroup.WebAPIGroup(dbmgr, channelmgr, apikeymgr)
    group_api.tokens = [crypto.md5_digest(config['router']['token'])]
    group_web = webgroup.WebPageGroup(config['web'], config['router'])
    
    paths = []
    
    paths += group_forward.get_routers()
    paths += group_web.get_routers()
    paths += group_api.get_routers()
    
    # ========= API Groups =========
    
    routermgr = routermgr.RouterManager(
        routes=paths,
        config=config['router'],
    )
    
    # watchdog and tasks
    from .watchdog import wd as watchdog
    
    wdmgr = watchdog.WatchDog()
    
    # tasks
    from .watchdog.tasks import heartbeat
    
    hbtask = heartbeat.HeartBeatTask(
        channelmgr,
        config['watchdog']['heartbeat'],
    )
    
    wdmgr.add_task(hbtask)
    
    app = Application(
        dbmgr=dbmgr,
        router=routermgr,
        channel=channelmgr,
        key=apikeymgr,
        watchdog=wdmgr,
        logging_level=logging_level,
    )
    
    logging.info("Application initialized.")
    
    return app
    
free_one_api/impls/app.py
RockChinQ/free-one-api
"""Channel management."""
import time
import asyncio
import random

from ...entities import channel, request, exceptions
from ...models.database import db
from ...models.channel import mgr


class ChannelManager(mgr.AbsChannelManager):
    """Channel manager.
    
    Provides channel creation, deletion, updating and listing,
    or selecting channel according to load balance algorithm.
    """

    def __init__(
        self,
        dbmgr: db.DatabaseInterface,
    ):
        self.dbmgr = dbmgr
        self.channels = []
        
    async def has_channel(self, channel_id: int) -> bool:
        for chan in self.channels:
            if chan.id == channel_id:
                return True
        return False
    
    async def has_channel_in_db(self, channel_id: int) -> bool:
        for chan in await self.dbmgr.list_channels():
            if chan.id == channel_id:
                return True
        return False
    
    async def get_channel(self, channel_id: int) -> channel.Channel:
        """Get a channel."""
        for chan in self.channels:
            if chan.id == channel_id:
                return chan
        raise ValueError("Channel not found.")

    async def list_channels(self) -> list[channel.Channel]:
        """List all channels."""
        self.channels = await self.dbmgr.list_channels()
        
        return self.channels
    
    async def create_channel(self, chan: channel.Channel) -> None:
        """Create a channel."""
        assert not await self.has_channel(chan.id)
        
        await self.dbmgr.insert_channel(chan)
        self.channels.append(chan)

    async def delete_channel(self, channel_id: int) -> None:
        """Delete a channel."""
        assert await self.has_channel(channel_id)
        
        await self.dbmgr.delete_channel(channel_id)
        for i in range(len(self.channels)):
            if self.channels[i].id == channel_id:
                del self.channels[i]
                break

    async def update_channel(self, chan: channel.Channel) -> None:
        """Update a channel."""
        assert await self.has_channel(chan.id)
        
        await self.dbmgr.update_channel(chan)
        for i in range(len(self.channels)):
            if self.channels[i].id == chan.id:
                self.channels[i] = chan
                break
            
    async def enable_channel(self, channel_id: int) -> None:
        """Enable a channel."""
        assert await self.has_channel(channel_id)
        
        chan = await self.get_channel(channel_id)
        chan.enabled = True
        await self.update_channel(chan)
        
    async def disable_channel(self, channel_id: int) -> None:
        """Disable a channel."""
        assert await self.has_channel(channel_id)
        
        chan = await self.get_channel(channel_id)
        chan.enabled = False
        await self.update_channel(chan)
        
    async def test_channel(self, channel_id: int) -> int:
        assert await self.has_channel(channel_id)
        
        chan = await self.get_channel(channel_id)
        # 计时
        now = time.time()
        latency = -1
        try:
            res, error = await chan.adapter.test()
            if not res:
                raise ValueError(error)
            latency = int((time.time() - now)*100)/100
        except Exception as e:
            raise ValueError("Test failed.") from e
        finally:
            chan.latency = latency
            await self.update_channel(chan)
        return latency

    async def select_channel(
        self,
        path: str,
        req: request.Request,
    ) -> channel.Channel:
        """Select a channel.
        
        Method here will filter channels and select the best one.
        
        Hard filters, which channel not match these conditions will be excluded:
        1. disabled channels.
        2. path the client request.
        3. model name the client request.
        
        Soft filters, these filter give score to each channel,
        the channel with the highest score will be selected:
        1. support for stream mode matching the client request.
        2. support for multi-round matching the client request.
        3. support for function calling.
        4. usage times in lifetime.
        
        Args:
            path: path of this request.
            req: request object.
            
        """
        stream_mode = req.stream
        has_functions = req.functions is not None and len(req.functions) > 0
        is_multi_round = req.messages is not None and len(req.messages) > 0
        
        model_name = req.model
        
        channel_copy = self.channels.copy()
        
        # delete disabled channels
        channel_copy = list(filter(lambda chan: chan.enabled, channel_copy))
        
        # delete not matched path
        channel_copy = list(filter(lambda chan: chan.adapter.supported_path() == path, channel_copy))
        
        # delete not matched model name
        channel_copy_tmp = []
        
        for chan in channel_copy:
            models = []
            left_model_names = list(chan.model_mapping.keys())
            models.extend(left_model_names)
            models.extend(chan.adapter.supported_models())
            
            if model_name in models:
                channel_copy_tmp.append(chan)
                
        channel_copy = channel_copy_tmp
        
        if len(channel_copy) == 0:
            raise exceptions.QueryHandlingError(
                404,
                "channel_not_found",
                "No suitable channel found. You may need to contact your admin or check the documentation at https://github.com/RockChinQ/free-one-api",
            )
        
        # i just randomly select one channel now!
        random.seed(time.time())
        return random.choice(channel_copy)

free_one_api/impls/channel/mgr.py
RockChinQ/free-one-api
import os
import json

import aiosqlite

from ...models.database import db as dbmod
from ...models import adapter
from ...entities import channel, apikey

channel_table_sql = """
CREATE TABLE IF NOT EXISTS channel (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    adapter JSON NOT NULL,
    model_mapping JSON NOT NULL,
    enabled INTEGER NOT NULL,
    latency INTEGER NOT NULL
)
"""

key_table_sql = """
CREATE TABLE IF NOT EXISTS apikey (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    created_at INTEGER NOT NULL,
    raw TEXT NOT NULL
)
"""

log_table_sql = """
CREATE TABLE IF NOT EXISTS log (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp INTEGER NOT NULL,
    content TEXT NOT NULL
)
"""


class SQLiteDB(dbmod.DatabaseInterface):

    def __init__(self, config: dict):
        self.config = config
        self.db_path = config['path']

    async def initialize(self):
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute(channel_table_sql)
            await db.execute(key_table_sql)
            await db.execute(log_table_sql)
            await db.commit()

    async def get_channel(self, channel_id: int) -> channel.Channel:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT * FROM channel WHERE id = ?", (channel_id,)) as cursor:
                row = await cursor.fetchone()
                return channel.Channel(
                    id=row[0],
                    name=row[1],
                    adapter=adapter.load_adapter(json.loads(row[2])),
                    model_mapping=json.loads(row[3]),
                    enabled=bool(row[4]),
                    latency=row[5],
                )

    async def list_channels(self) -> list[channel.Channel]:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT * FROM channel") as cursor:
                rows = await cursor.fetchall()
                return [channel.Channel(
                    id=row[0],
                    name=row[1],
                    adapter=adapter.load_adapter(json.loads(row[2])),
                    model_mapping=json.loads(row[3]),
                    enabled=bool(row[4]),
                    latency=row[5],
                ) for row in rows]

    async def insert_channel(self, chan: channel.Channel) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("INSERT INTO channel (name, adapter, model_mapping, enabled, latency) VALUES (?, ?, ?, ?, ?)", (
                chan.name,
                json.dumps(adapter.dump_adapter(chan.adapter)),
                json.dumps(chan.model_mapping),
                int(chan.enabled),
                chan.latency,
            ))
            await db.commit()
            async with db.execute("SELECT last_insert_rowid()") as cursor:
                row = await cursor.fetchone()
                chan.id = row[0]
  
    async def update_channel(self, chan: channel.Channel) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("UPDATE channel SET name = ?, adapter = ?, model_mapping = ?, enabled = ?, latency = ? WHERE id = ?", (
                chan.name,
                json.dumps(adapter.dump_adapter(chan.adapter)),
                json.dumps(chan.model_mapping),
                int(chan.enabled),
                chan.latency,
                chan.id,
            ))
            await db.commit()

    async def delete_channel(self, channel_id: int) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("DELETE FROM channel WHERE id = ?", (channel_id,))
            await db.commit()

    async def list_keys(self) -> list[apikey.FreeOneAPIKey]:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT * FROM apikey") as cursor:
                rows = await cursor.fetchall()
                return [apikey.FreeOneAPIKey(
                    id=row[0],
                    name=row[1],
                    created_at=row[2],
                    raw=row[3],
                ) for row in rows]

    async def insert_key(self, key: apikey.FreeOneAPIKey) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("INSERT INTO apikey (name, created_at, raw) VALUES (?, ?, ?)", (
                key.name,
                key.created_at,
                key.raw,
            ))
            await db.commit()
            async with db.execute("SELECT last_insert_rowid()") as cursor:
                row = await cursor.fetchone()
                key.id = row[0]

    async def update_key(self, key: apikey.FreeOneAPIKey) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("UPDATE apikey SET name = ?, created_at = ?, raw = ? WHERE id = ?", (
                key.name,
                key.created_at,
                key.raw,
                key.id,
            ))
            await db.commit()

    async def delete_key(self, key_id: int) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("DELETE FROM apikey WHERE id = ?", (key_id,))
            await db.commit()

    async def insert_log(self, timestamp: int, content: str) -> None:
        async with aiosqlite.connect(self.db_path) as db:
            await db.execute("INSERT INTO log (timestamp, content) VALUES (?, ?)", (
                timestamp,
                content,
            ))
            await db.commit()

    async def select_logs(self, time_range: tuple[int, int]) -> list[tuple[int, int, str]]:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT * FROM log WHERE timestamp >= ? AND timestamp <= ?", time_range) as cursor:
                rows = await cursor.fetchall()
                return [(row[0], row[1], row[2]) for row in rows]

    async def select_logs_page(self, capacity: int, page: int) -> list[tuple[int, int, str]]:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT * FROM log ORDER BY id DESC LIMIT ? OFFSET ?", (capacity, capacity * page)) as cursor:
                rows = await cursor.fetchall()
                return [(row[0], row[1], row[2]) for row in rows]
    
    async def get_logs_amount(self) -> int:
        async with aiosqlite.connect(self.db_path) as db:
            async with db.execute("SELECT COUNT(*) FROM log") as cursor:
                row = await cursor.fetchone()
                return row[0]
    
free_one_api/impls/database/sqlite.py
RockChinQ/free-one-api
import time
import json
import string
import random
import logging
import typing

import quart

from ...models.forward import mgr as forwardmgr
from ...models.channel import mgr as channelmgr
from ...models.key import mgr as apikeymgr
from ...entities import channel, apikey, request, response, exceptions
from ...common import randomad


class ForwardManager(forwardmgr.AbsForwardManager):
    
    def __init__(self, chanmgr: channelmgr.AbsChannelManager, keymgr: apikeymgr.AbsAPIKeyManager):
        self.chanmgr = chanmgr
        self.keymgr = keymgr

    async def __stream_query(
        self,
        chan: channel.Channel,
        req: request.Request,
        id_suffix: str,
    ) -> quart.Response:
        before = time.time()
        
        t = int(time.time())
        async def _gen():
            try:
                async for resp in chan.adapter.query(req):
                    
                    if (resp.normal_message is None or len(resp.normal_message) == 0) and resp.finish_reason == response.FinishReason.NULL:
                        continue
                    
                    yield "data: {}\n\n".format(json.dumps({
                        "id": "chatcmpl-"+id_suffix,
                        "object": "chat.completion.chunk",
                        "created": t,
                        "model": req.model,
                        "choices": [{
                            "index": 0,
                            "delta": {
                                "content": resp.normal_message,
                            } if resp.normal_message else {},
                            "finish_reason": resp.finish_reason.value
                        }]
                    }))
                
                if randomad.enabled:
                    for word in randomad.generate_ad():
                        yield "data: {}\n\n".format(json.dumps({
                            "id": "chatcmpl-"+id_suffix,
                            "object": "chat.completion.chunk",
                            "created": t,
                            "model": req.model,
                            "choices": [{
                                "index": 0,
                                "delta": {
                                    "content": word,
                                },
                                "finish_reason": response.FinishReason.NULL.value
                            }]
                        }))
                
                yield "data: [DONE]\n\n"
            except exceptions.QueryHandlingError as e:
                yield "data: {}\n\ndata: [DONE]\n\n".format(json.dumps({
                    "error": {
                        "code": e.code,
                        "message": e.message,
                        "type": e.type,
                        "param": e.param,
                    }
                }))
            except Exception as e:
                logging.warning("Exception should be processed by adapter but caught by forward manager:")
                logging.error(e)
        
        spent_ms = int((time.time() - before)*1000)
        
        headers = {
            "Content-Type": "text/event-stream",
            "Transfer-Encoding": "chunked",
            "Connection": "keep-alive",
            "openai-processing-ms": str(spent_ms),
            "openai-version": "2020-10-01",
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        }
        
        return quart.Response(
            _gen(),
            mimetype="text/event-stream",
            headers=headers,
        )
    
    async def __non_stream_query(
        self,
        chan: channel.Channel,
        req: request.Request,
        id_suffix: str,
    ) -> quart.Response:
        before = time.time()
        
        normal_message = ""
        
        resp_tmp: response.Response = None
        
        try:
        
            async for resp in chan.adapter.query(req):
                if resp.normal_message is not None:
                    resp_tmp = resp
                    normal_message += resp.normal_message
                    
            if randomad.enabled:
                for word in randomad.generate_ad():
                    normal_message += word

        except exceptions.QueryHandlingError as e:
            # check for custom error raised by adapter
            return quart.jsonify({
                "error": {
                    "code": e.code,
                    "message": e.message,
                    "type": e.type,
                    "param": e.param,
                }
            }), e.status_code
                
        spent_ms = int((time.time() - before)*1000)
        
        prompt_tokens = chan.count_tokens(req.model, req.messages)
        completion_tokens = chan.count_tokens(
            req.model,
            [{
                "role": "assistant",
                "content": normal_message,
            }]
        )
        
        result = {
            "id": "chatcmpl-"+id_suffix,
            "object": "chat.completion",
            "created": int(time.time()),
            "model": req.model,
            "choices": [
                {
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": normal_message,
                    },
                    "finish_reason": resp_tmp.finish_reason.value
                }
            ],
            "usage": {
                "prompt_tokens": prompt_tokens,
                "completion_tokens": completion_tokens,
                "total_tokens": prompt_tokens + completion_tokens,
            }
        }
        
        return quart.jsonify(result)

    async def query(
        self,
        path: str,
        req: request.Request,
        raw_data: dict,
    ) -> quart.Response:
        
        chan: channel.Channel = await self.chanmgr.select_channel(
            path,
            req,
        )
        
        # find model replacement
        if len(chan.model_mapping.keys()) > 0:
            if req.model in chan.model_mapping.keys():
                req.model = chan.model_mapping[req.model]

        if chan is None:
            pass
        
        id_suffix = ""
        id_suffix += "{}".format(chan.id).zfill(3)
        id_suffix += chan.adapter.__class__.__name__[:10]
        id_suffix += "".join(random.choices(string.ascii_letters+string.digits, k=29-len(id_suffix)))
        
        auth = quart.request.headers.get("Authorization")
        if auth.startswith("Bearer "):
            auth = auth[7:]
        
        query_info_str = f"type=query, path={path}, model={req.model}, id_suffix={id_suffix}, channel_name={chan.name}, channel_adpater={chan.adapter.__class__.__name__}, key={auth}"
        
        logging.info(query_info_str)
        
        if req.stream:
            return await self.__stream_query(chan, req, id_suffix)
        else:
            return await self.__non_stream_query(chan, req, id_suffix)
free_one_api/impls/forward/mgr.py
RockChinQ/free-one-api
import asyncio

from ...entities import apikey
from ...models.database import db
from ...models.key import mgr as keymgr


class APIKeyManager(keymgr.AbsAPIKeyManager):
    """API Key manager."""
    
    def __init__(self, dbmgr: db.DatabaseInterface):
        self.dbmgr = dbmgr
        self.keys = []
        
    async def has_key(self, key_id: int) -> bool:
        for key in self.keys:
            if key.id == key_id:
                return True
        return False
    
    async def has_key_in_db(self, key_id: int) -> bool:
        for key in await self.dbmgr.list_keys():
            if key.id == key_id:
                return True
        return False
    
    async def has_key_name(self, key_name: str) -> bool:
        for key in self.keys:
            if key.name == key_name:
                return True
        return False
    
    async def has_key_name_in_db(self, key_name: str) -> bool:
        for key in await self.dbmgr.list_keys():
            if key.name == key_name:
                return True
        return False
    
    async def list_keys(self) -> list[apikey.FreeOneAPIKey]:
        """List all keys."""
        self.keys = await self.dbmgr.list_keys()
        
        return self.keys
    
    async def create_key(self, key: apikey.FreeOneAPIKey) -> None:
        # key already created by upper caller
        # only insert to db and save to memory here
        await self.dbmgr.insert_key(key)
        self.keys.append(key)
        
    async def revoke_key(self, key_id: int) -> None:
        assert await self.has_key(key_id)
        
        await self.dbmgr.delete_key(key_id)
        for i in range(len(self.keys)):
            if self.keys[i].id == key_id:
                del self.keys[i]
                break
            
    async def get_key(self, key_id: int) -> apikey.FreeOneAPIKey:
        """Get a key."""
        for key in self.keys:
            if key.id == key_id:
                return key
        raise ValueError("Key not found.")

free_one_api/impls/key/mgr.py
RockChinQ/free-one-api
import logging
import asyncio

from ..models.database import db


class SQLiteHandler(logging.Handler):
    """SQLite logging handler."""

    dbmgr: db.DatabaseInterface
    
    queue: asyncio.Queue = None

    def __init__(self, dbmgr: db.DatabaseInterface, level=logging.NOTSET):
        super().__init__(level)
        self.dbmgr = dbmgr
        
    async def _consumer(self):
        while True:
            record: logging.LogRecord = await self.queue.get()
            await self.dbmgr.insert_log(record.created, record.getMessage())
            self.queue.task_done()

    def emit(self, record: logging.LogRecord):
        """Emit a record."""
        loop = asyncio.get_running_loop()
        
        if self.queue is None:
            self.queue = asyncio.Queue()
            loop.create_task(self._consumer())
            
        loop.create_task(self.queue.put(record))

free_one_api/impls/log.py
RockChinQ/free-one-api
import json
import traceback

import quart

from ...models.router import group as routergroup
from ...models.database import db
from ...models.channel import mgr as channelmgr
from ...models.key import mgr as apikeymgr
from ...entities import channel, apikey
from ...models import adapter


class WebAPIGroup(routergroup.APIGroup):
    
    chanmgr: channelmgr.AbsChannelManager
    
    keymgr: apikeymgr.AbsAPIKeyManager
    
    def __init__(self, dbmgr: db.DatabaseInterface, chanmgr: channelmgr.AbsChannelManager, keymgr: apikeymgr.AbsAPIKeyManager):
        super().__init__(dbmgr)
        self.chanmgr = chanmgr
        self.keymgr = keymgr
        self.group_name = "/api"
        
        @self.api("/channel/list", ["GET"], auth=True)
        async def channel_list():
            # load channels from db to memory
            chan_list = await self.chanmgr.list_channels()
            
            chan_list_json = [channel.Channel.dump_channel(chan) for chan in chan_list]
            chan_list_json = [{
                "id": chan["id"],
                "name": chan["name"],
                "adapter": chan["adapter"]['type'],
                "enabled": chan["enabled"],
                "latency": chan["latency"],
            } for chan in chan_list_json]
            
            return quart.jsonify({
                "code": 0,
                "message": "ok",
                "data": chan_list_json,
            })
        
        @self.api("/channel/create", ["POST"], auth=True)
        async def channel_create():
            try:
                data = await quart.request.get_json()
                
                chan = channel.Channel.load_channel(data)
                
                await self.chanmgr.create_channel(chan)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                })
            except Exception as e:
                traceback.print_exc()
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/channel/delete/<int:chan_id>", ["DELETE"], auth=True)
        async def channel_delete(chan_id: int):
            try:
                await self.chanmgr.delete_channel(chan_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/channel/details/<int:chan_id>", ["GET"], auth=True)
        async def channel_details(chan_id: int):
            try:
                chan = await self.chanmgr.get_channel(chan_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": channel.Channel.dump_channel(chan),
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/channel/update/<int:chan_id>", ["PUT"], auth=True)
        async def channel_update(chan_id: int):
            try:
                chan = channel.Channel.load_channel(await quart.request.get_json())
                chan.id = chan_id
                await self.chanmgr.update_channel(chan)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/channel/enable/<int:chan_id>", ["POST"], auth=True)
        async def channel_enable(chan_id: int):
            try:
                await self.chanmgr.enable_channel(chan_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/channel/disable/<int:chan_id>", ["POST"], auth=True)
        async def channel_disable(chan_id: int):
            try:
                await self.chanmgr.disable_channel(chan_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/channel/test/<int:chan_id>", ["POST"], auth=True)
        async def channel_test(chan_id: int):
            try:
                latency = await self.chanmgr.test_channel(chan_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": {
                        "latency": latency,
                    },
                })
            except Exception as e:
                traceback.print_exc()
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/adapter/list", ["GET"], auth=True)
        async def adapter_list():
            res = {
                "code": 0,
                "message": "ok",
                "data": adapter.list_adapters(),
            }
            
            return quart.jsonify(res)
            
        @self.api("/key/list", ["GET"], auth=True)
        async def key_list():
            try:
                key_list = await self.keymgr.list_keys()
                
                key_list_json = []
                
                for key in key_list:
                    key_list_json.append({
                        "id": key.id,
                        "name": key.name,
                        "brief": key.raw[:10] + "..." + key.raw[-10:],
                        "created_at": key.created_at,
                    })
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": key_list_json,
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/key/raw/<int:key_id>", ["GET"], auth=True)
        async def key_raw(key_id: int):
            try:
                key = await self.keymgr.get_key(key_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": {
                        "key": key.raw,
                    },
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
            
        @self.api("/key/create", ["POST"], auth=True)
        async def key_create():
            try:
                data = await quart.request.get_json()
                
                key_name = data["name"]
                
                if await self.keymgr.has_key_name(key_name):
                    raise ValueError("key name already exists: "+key_name)
                
                key = apikey.FreeOneAPIKey.make_new(key_name)
                
                await self.keymgr.create_key(key)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": {
                        "id": key.id,
                        "raw": key.raw,
                    },
                })
            except Exception as e:
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })

        @self.api("/key/revoke/<int:key_id>", ["DELETE"], auth=True)
        async def key_revoke(key_id: int):
            try:
                await self.keymgr.revoke_key(key_id)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                })
            except Exception as e:
                import traceback
                traceback.print_exc()
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })

        @self.api("/log/list", ["GET"], auth=True)
        async def list_logs():
            try:
                data = quart.request.args
                
                capacity = int(data.get("capacity", 20))
                page = int(data.get("page", 0))
                
                amount = await self.dbmgr.get_logs_amount()
                page_count = (amount + capacity - 1) // capacity
                
                logs = await self.dbmgr.select_logs_page(capacity, page)
                
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": {
                        "page_count": page_count,
                        "logs": [
                            {
                                "id": log[0],
                                "timestamp": log[1],
                                "content": log[2],
                            } for log in logs
                        ],
                    },
                })
            except Exception as e:
                import traceback
                traceback.print_exc()
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })
        
        @self.api("/info/version", ["GET"], auth=False)
        async def info_version():
            try:
            
                from ...common import version
            
                return quart.jsonify({
                    "code": 0,
                    "message": "ok",
                    "data": "v"+version.__version__,
                })
            except Exception as e:
                import traceback
                traceback.print_exc()
                return quart.jsonify({
                    "code": 1,
                    "message": str(e),
                })

free_one_api/impls/router/api.py
RockChinQ/free-one-api
import traceback

import quart

from ...models.router import group as routergroup
from ...models.key import mgr as apikeymgr
from ...models.channel import mgr as channelmgr
from ...models.database import db
from ...models.forward import mgr as forwardmgr
from ...entities import channel, apikey, request, response, exceptions


class ForwardAPIGroup(routergroup.APIGroup):
    """Forward API group."""
    
    chanmgr: channelmgr.AbsChannelManager
    """Channel manager."""
    
    keymgr: apikeymgr.AbsAPIKeyManager
    """API key manager."""
    
    fwdmgr: forwardmgr.AbsForwardManager
    """Forward manager."""
    
    def __init__(
        self,
        dbmgr: db.DatabaseInterface,
        chanmgr: channelmgr.AbsChannelManager,
        keymgr: apikeymgr.AbsAPIKeyManager,
        fwdmgr: forwardmgr.AbsForwardManager,
    ):
        super().__init__(dbmgr)
        self.forwardmgr = forwardmgr
        self.group_name = ""
        self.chanmgr = chanmgr
        self.keymgr = keymgr
        self.fwdmgr = fwdmgr

        @self.api("/v1/chat/completions", ["POST"], auth=True)
        async def chat_completion():
            """Chat completion."""
            try:
            
                raw_data = await quart.request.get_json()
                
                req = request.Request(
                    raw_data["model"],
                    raw_data["messages"],
                    raw_data["functions"] if "functions" in raw_data else None,
                    'stream' in raw_data and raw_data["stream"],
                )
                
                return await self.fwdmgr.query(
                    "/v1/chat/completions",
                    req,
                    raw_data,
                )
            except exceptions.QueryHandlingError as e:
                return quart.jsonify(
                    {
                        "error": {
                            "message": e.message,
                            "type": e.type,
                            "param": e.param,
                            "code": e.code
                        }
                    }
                ), e.status_code
            except Exception as e:
                traceback.print_exc()
                return quart.jsonify(
                    {
                        "error": {
                            "message": "Error occurred while handling your request: {}. You can retry or contact your admin. See the documentation for details at https://github.com/RockChinQ/free-one-api".format(str(e)),
                            "type": "requests",
                            "param": None,
                            "code": None
                        }
                    }
                ), 500
        
    def get_tokens(self) -> list[str]:
        """Override to use keys in keymgr as tokens."""
        key_obj_list: apikey.FreeOneAPIKey = self.keymgr.get_key_list()
        
        key_list = [key_obj.raw for key_obj in key_obj_list]
        
        return key_list

    def check_auth(self, auth: str) -> quart.Response:
        if auth is None or not auth.startswith("Bearer "):
            return quart.jsonify(
                {
                    "error": {
                        "message": "You didn't provide an API key. You need to provide your API key in an Authorization header using Bearer auth (i.e. Authorization: Bearer YOUR_KEY), or as the password field (with blank username) if you're accessing the API from your browser and are prompted for a username and password. You can obtain an API key from the free-one-api admin page, please contact the admin. See documentation for details at https://github.com/RockChinQ/free-one-api",
                        "type": "invalid_request_error",
                        "param": None,
                        "code": None
                    }
                }
            ), 401
        token = auth[7:]
        
        if token not in self.get_tokens():
            prefix = token[:8]
            suffix = token[-4:]
            mid = "*" * ((51 - len(prefix) - len(suffix)) if (len(prefix) + len(suffix) < 51) else 0)
            return quart.jsonify(
                {
                    "error": {
                        "message": "Incorrect API key provided: "+prefix+mid+suffix+". You can find you API key on the admin page of free-one-api, please contact the admin. See documentation for details at https://github.com/RockChinQ/free-one-api",
                        "type": "invalid_request_error",
                        "param": None,
                        "code": "invalid_api_key"
                    }
                }
            ), 401
            
        return None

free_one_api/impls/router/forward.py
RockChinQ/free-one-api
import os

import quart


class RouterManager:
    """Router manager.
    
    Register all paths and serve.
    """

    port: int
    frontend_dir: str
    _app: quart.Quart

    def __init__(self, routes: list[tuple[str, list[str], callable, dict]], config: dict):
        self.port = config["port"] if "port" in config else 3000
        self._app = quart.Quart(__name__)

        for route, methods, handler, kwargs in routes:
            for method in methods:
                self._app.route(route, methods=[method], **kwargs)(handler)

    async def serve(self, loop):
        """Serve API."""
        return await self._app.run_task(host="0.0.0.0", port=self.port)


if __name__ == "__main__":
    # print workdir
    print(os.getcwd())

    frontend_dir = "../../../web/dist/"
    
    # register frontend apis
    async def ping():
        return "pong"
    
    # frontend page and public files
    async def index():
        return await quart.send_from_directory(frontend_dir, "index.html")
    
    async def static_proxy(path):
        if not os.path.exists(os.path.join(frontend_dir, path)):
            return await quart.send_from_directory(frontend_dir, "index.html")
        return await quart.send_from_directory(frontend_dir, path)

    api = RouterManager(
        routes=[
            ("/ping", ["GET"], ping, {}),
            ("/", ["GET"], index, {}),
            ("/<path:path>", ["GET"], static_proxy, {}),
        ],
        config={
            "port": 3000
        }
    )
    api.serve()

free_one_api/impls/router/mgr.py
RockChinQ/free-one-api
import os

import quart

from ...models.router import group as routergroup
from ...common import crypto


class WebPageGroup(routergroup.APIGroup):
    
    frontend_path: str
    
    def __init__(self, webcfg: dict, routercfg: dict):
        super().__init__(None)
        self.frontend_path = webcfg["frontend_path"] if "frontend_path" in webcfg else "./web/dist/"
        self.group_name = ""
        
        @self.api("/check_password", ["POST"])
        async def check_password():
            data = await quart.request.get_json()
            if "password" not in data:
                return quart.jsonify({
                    "code": 1,
                    "message": "No password provided."
                })
            if data["password"] != crypto.md5_digest(routercfg["token"]):
                return quart.jsonify({
                    "code": 2,
                    "message": "Wrong token."
                })
            return quart.jsonify({
                "code": 0,
                "message": "ok"
            })
        
        @self.api("/ping", ["GET"])
        async def ping():
            return "pong"
        
        @self.api("/", ["GET"])
        async def index():
            return await quart.send_from_directory(self.frontend_path, "index.html")
        
        @self.api("/<path:path>", ["GET"])
        async def static_proxy(path):
            # if not os.path.exists(os.path.join(self.frontend_path, path)):
            #     return await quart.send_from_directory(self.frontend_path, "index.html")
            return await quart.send_from_directory(self.frontend_path, path)

free_one_api/impls/router/web.py
RockChinQ/free-one-api
import asyncio
import traceback
import logging
import random

from ....models.watchdog import task
from ....models.channel import mgr as chanmgr
from ....entities import channel


class HeartBeatTask(task.AbsTask):
    """HeartBeat task."""

    channel: chanmgr.AbsChannelManager
    
    cfg: dict

    def __init__(self, chan: chanmgr.AbsChannelManager, cfg: dict):
        self.channel = chan
        self.cfg = cfg
        
        self.delay = 10
        self.interval = cfg['interval']

    async def trigger(self):
        """Trigger this task."""
        
        process_task = []
        
        for chan in self.channel.channels:
            if chan.enabled:
                async def process(ch: channel.Channel):
                    random_delay = random.randint(0, 10)
                    await asyncio.sleep(random_delay)
                    
                    fail_count = await ch.heartbeat(timeout=self.cfg["timeout"])
                    if fail_count > self.cfg["fail_limit"]:
                        try:
                            self.channel.disable_channel(ch.id)
                            logging.info(f"Disabled channel {ch.id} due to heartbeat failed {fail_count} times")
                        except Exception:
                            logging.warn(f"Failed to disable channel {ch.id}, traceback: {traceback.format_exc()}")
                    await self.channel.update_channel(ch)
                    
                process_task.append(process(chan))
                
        logging.info(f"Start heartbeat task, {len(process_task)} channels to process")
        await asyncio.gather(*process_task)
 
free_one_api/impls/watchdog/tasks/heartbeat.py
RockChinQ/free-one-api
import asyncio

from ...models.watchdog import wd
from ...models.watchdog import task


class WatchDog(wd.AbsWatchDog):
    """WatchDog implementation."""
    
    def __init__(self):
        self.tasks = []
        
    async def run(self):
        cor = []
        
        for task in self.tasks:
            cor.append(task.loop())
            
        await asyncio.gather(*cor)
        
    def add_task(self, task: task.AbsTask):
        """Add a task."""
        self.tasks.append(task)

free_one_api/impls/watchdog/wd.py
RockChinQ/free-one-api
import json

from . import llm


# Utility for adapter management

adapters: dict[str, llm.LLMLibAdapter] = {}
"""Registered adapters.

{
    "name": LLMLibAdapter
}
"""


def llm_adapter(cls: llm.LLMLibAdapter) -> llm.LLMLibAdapter:
    """Get adapter by name."""

    adapters[cls.name()] = cls

    return cls


def list_adapters() -> list[str]:
    """List all adapters."""
    return [
        {
            "name": name,
            "config_comment": cls.config_comment()
        } for name, cls in adapters.items()
    ]


def load_adapter(data: dict) -> llm.LLMLibAdapter:
    """Load adapter from dict."""
    obj = data

    return adapters[obj["type"]](obj["config"])


def dump_adapter(adapter: llm.LLMLibAdapter) -> dict:
    """Dump adapter to dict."""
    return {
        "type": adapter.name(),
        "config": adapter.get_config()
    }


__all__ = [
    'adapters',
    'llm_adapter',
    'load_adapter'
]

free_one_api/models/adapter/__init__.py
RockChinQ/free-one-api
import abc
import typing

from ...entities import response
from ...entities import request


class LLMLibAdapter(metaclass=abc.ABCMeta):
    """Base class for reverse engineering LLM Lib adapters."""
    
    config: dict

    @abc.abstractclassmethod
    def name(self) -> str:
        """Name of this adapter.
        
        Prefer the github path of LLM lib.
        e.g. acheong08/ChatGPT
        """
        return "not implemented"

    @abc.abstractclassmethod
    def description(self) -> str:
        """Description of this adapter."""
        return self.__class__.__doc__

    @abc.abstractmethod
    def supported_models(self) -> list[str]:
        """Return models supported by this reverse lib."""
        return []

    @abc.abstractmethod
    def function_call_supported(self) -> bool:
        """True if this adapter supports function call."""
        return False
    
    @abc.abstractmethod
    def stream_mode_supported(self) -> bool:
        """True if this adapter supports stream mode."""
        return False
    
    @abc.abstractmethod
    def multi_round_supported(self) -> bool:
        """True if this adapter supports multi round conversation."""
        return False

    @abc.abstractclassmethod
    def config_comment(self) -> str:
        """Comments of the config schema.
        
        Returns:
            str: comments, empty if no config needed.
        """
        return ""
    
    @abc.abstractclassmethod
    def supported_path(self) -> str:
        """Returns which path of OpenAI API should this adapter serve.
        
        e.g. /v1/chat/completions
        """
        return "/v1/chat/completions"

    @abc.abstractmethod
    def __init__(self, config: dict):
        """Init adapter with config.
        
        Args:
            config: config of this adapter.
        """
        self.config = config

    def get_config(self) -> dict:
        """Get set config.
        
        Returns:
            str: set config, empty if no config needed.
        """
        return self.config

    @abc.abstractmethod
    async def test(self) -> (bool, str):
        """Test the adapter.
        
        Returns:
            bool: True if adapter can be used.
            str: error message, empty if no error
        """
        return False, "not implemented"

    @abc.abstractmethod
    async def query(self, req: request.Request) -> typing.AsyncGenerator[response.Response, None]:
        """Query reply from LLM lib.
        
        Always in streaming mode. If upstream lib doesn't support streaming, just yield one time.
        """
        yield None

free_one_api/models/adapter/llm.py
RockChinQ/free-one-api
import abc

import tiktoken

from ...entities import channel
from ..database import db
from ...entities import request, response


class AbsChannelManager(metaclass=abc.ABCMeta):
    """Base class for channel manager."""
    
    dbmgr: db.DatabaseInterface
    """Database manager."""

    channels: list[channel.Channel]
    """Channel list in runtime."""

    @abc.abstractmethod
    async def list_channels(self) -> list[channel.Channel]:
        """List all channels."""
        pass

    @abc.abstractmethod
    async def create_channel(self, chan: channel.Channel) -> None:
        """Create a channel."""
        pass

    @abc.abstractmethod
    async def delete_channel(self, channel_id: int) -> None:
        """Delete a channel."""
        pass

    @abc.abstractmethod
    async def update_channel(self, chan: channel.Channel) -> None:
        """Update a channel."""
        pass

    @abc.abstractmethod
    async def enable_channel(self, channel_id: int) -> None:
        """Enable a channel."""
        pass

    @abc.abstractmethod
    async def disable_channel(self, channel_id: int) -> None:
        """Disable a channel."""
        pass

    @abc.abstractmethod
    async def test_channel(self, channel_id: int) -> int:
        """Test a channel.
        
        Return latency.
        """
        pass

    @abc.abstractmethod
    async def select_channel(
        self,
        path: str,
        req: request.Request,
    ) -> channel.Channel:
        """Select a channel.
        
        Implement load balance algorithm here.
        
        Args:
            path: path of this request.
            req: request object.
        """
        pass

free_one_api/models/channel/mgr.py
RockChinQ/free-one-api
import abc

from ...entities import channel, apikey


class DatabaseInterface(metaclass=abc.ABCMeta):
    """Base class for database interface."""

    @abc.abstractmethod
    async def get_channel(self, channel_id: int) -> channel.Channel:
        """Get a channel."""
        return

    @abc.abstractmethod
    async def list_channels(self) -> list[channel.Channel]:
        """Load all channels."""
        return

    @abc.abstractmethod
    async def insert_channel(self, chan: channel.Channel) -> None:
        """Insert a channel. Set the id of the channel."""
        return

    @abc.abstractmethod
    async def update_channel(self, chan: channel.Channel) -> None:
        """Update a channel."""
        return

    @abc.abstractmethod
    async def delete_channel(self, channel_id: int) -> None:
        """Delete a channel."""
        return

    @abc.abstractmethod
    async def list_keys(self) -> list[apikey.FreeOneAPIKey]:
        """Load all keys."""
        return

    @abc.abstractmethod
    async def insert_key(self, key: apikey.FreeOneAPIKey) -> None:
        """Insert a key. Set the id of the key."""
        return

    @abc.abstractmethod
    async def update_key(self, key: apikey.FreeOneAPIKey) -> None:
        """Update a key."""
        return

    @abc.abstractmethod
    async def delete_key(self, key_id: int) -> None:
        """Delete a key."""
        return

    @abc.abstractmethod
    async def insert_log(self, timestamp: int, content: str) -> None:
        """Insert a log."""
        return

    @abc.abstractmethod
    async def select_logs(self, time_range: tuple[int, int]) -> list[tuple[int, int, str]]:
        """Select logs.
        
        Args:
            time_range: (start, end)
        """
        return
    
    @abc.abstractmethod
    async def select_logs_page(self, capacity: int, page: int) -> list[tuple[int, int, str]]:
        """Select logs, sort descending by timestamp.
        
        Args:
            capacity: number of logs per page
            page: page number
        """
        return
    
    @abc.abstractmethod
    async def get_logs_amount(self) -> int:
        """Get the amount of logs."""
        return

free_one_api/models/database/db.py
RockChinQ/free-one-api
import abc

import quart

from ...models.channel import mgr as channelmgr
from ...models.key import mgr as apikeymgr
from ...entities import channel, apikey
from ...models import adapter
from ...entities import request, response


supported_paths = [
    "/v1/chat/completions",
]


class AbsForwardManager(metaclass=abc.ABCMeta):
    """Abstract forward manager.
    
    Receive request object, Select a channel from channel manager,
    call the channel adapter, receive response object, wrapper it 
    depends on the stream mode setting, and return structure.
    """
    
    chanmgr: channelmgr.AbsChannelManager
    """Channel manager."""
    
    keymgr: apikeymgr.AbsAPIKeyManager
    """API key manager."""
    
    @abc.abstractmethod
    async def query(
        self,
        path: str,
        req: request.Request,
        raw_data: dict,
    ) -> quart.Response:
        """Query.
        
        Args:
            path: path.
            req: request object.
            raw_data: raw structure data.
            stream_mode: stream mode.
        """
        pass

free_one_api/models/forward/mgr.py
RockChinQ/free-one-api
import abc

from ...entities import apikey
from ..database import db

class AbsAPIKeyManager(metaclass=abc.ABCMeta):
    
    dbmgr: db.DatabaseInterface
    """Database manager."""
    
    keys: list[apikey.FreeOneAPIKey]
    """Key list in runtime."""
    
    def get_key_list(self) -> list[apikey.FreeOneAPIKey]:
        """Get key list."""
        return self.keys
    
    @abc.abstractmethod
    async def has_key(self, key_id: int) -> bool:
        """Check if a key exists."""
        pass
    
    @abc.abstractmethod
    async def has_key_in_db(self, key_id: int) -> bool:
        """Check if a key exists in database."""
        pass
    
    @abc.abstractmethod
    async def has_key_name(self, key_name: str) -> bool:
        """Check if a key name exists."""
        pass
    
    @abc.abstractmethod
    async def has_key_name_in_db(self, key_name: str) -> bool:
        """Check if a key name exists in database."""
        pass
    
    @abc.abstractmethod
    async def list_keys(self) -> list[apikey.FreeOneAPIKey]:
        """List all keys."""
        pass
    
    @abc.abstractmethod
    async def create_key(self, key: apikey.FreeOneAPIKey) -> None:
        """Create a key."""
        pass
    
    @abc.abstractmethod
    async def revoke_key(self, key_id: int) -> None:
        """Revoke a key."""
        pass
    
    @abc.abstractmethod
    async def get_key(self, key_id: int) -> apikey.FreeOneAPIKey:
        """Get a key."""
        pass

free_one_api/models/key/mgr.py
RockChinQ/free-one-api
import abc

import quart

from ..database import db


class APIGroup(metaclass=abc.ABCMeta):
    
    group_name: str
    """Name of this group, also the path prefix of all APIs in this group.
    
    e.g. /api
    """
    
    tokens: list[str]
    """Tokens for this group."""
    
    routers: list[tuple[str, list[str], callable, dict]]
    
    dbmgr: db.DatabaseInterface
    
    def set_tokens(self, tokens: list[str]):
        """Set tokens for this group.
        
        Args:
            tokens: tokens.
        """
        self.tokens = tokens
    
    def api(self, path: str, methods: list[str], auth: bool=False, **kwargs):
        """Register an API.
        
        Args:
            path: path of this API, relative to this group.
            methods: HTTP methods.
            **kwargs: other arguments.
        """

        def decorator(handler):
            """Decorator."""
            
            new_handler = handler
            if auth:
                async def authenticated_handler(*args, **kwargs):
                    """Authenticated handler."""
                    auth = quart.request.headers.get("Authorization")

                    check_res = self.check_auth(auth)
                    
                    if check_res is not None:
                        return check_res
                    else:
                        return await handler(*args, **kwargs)
                new_handler = authenticated_handler
                new_handler.__name__ = handler.__name__
            
            self.routers.append((self.group_name+path, methods, new_handler, kwargs))
            return new_handler

        return decorator

    def __init__(self, dbmgr: db.DatabaseInterface):
        self.routers = []
        self.dbmgr = dbmgr
        
    def get_routers(self):
        """Get all APIs in this group.
        
        Returns:
            list of APIs.
        """
        return self.routers
    
    def get_tokens(self) -> list[str]:
        """Return set tokens for this group.
        
        For api auth only.
        This can be overrided if impl class has its own token management.
        """
        return self.tokens
    
    def check_auth(self, auth: str) -> quart.Response:
        """Check if auth is valid.
        
        Args:
            auth: auth string.
            
        Returns:
            Response if auth is invalid, None if auth is valid.
        """
        if auth is None:
            return quart.jsonify({
                "code": 1,
                "message": "No authorization provided."
            })
        if not auth.startswith("Bearer "):
            return quart.jsonify({
                "code": 2,
                "message": "Wrong authorization type."
            })
        token = auth[7:]
        if token not in self.get_tokens():
            return quart.jsonify({
                "code": 3,
                "message": "Wrong token, please re-login or check your token."
            })
        return None

free_one_api/models/router/group.py
RockChinQ/free-one-api
import abc
import asyncio
import traceback
import logging


class AbsTask(metaclass=abc.ABCMeta):
    """Task of WatchDog.
    
    A task may need instances of other modules to work. These dependencies should be set by outer constructor.
    Task's delay and interval may be set by outer constructor also, and be scheduled by WatchDog implementation.
    """
    
    delay: int = 0
    """Delay before first trigger."""
    
    interval: int = 60
    """Interval between two triggers."""
    
    @abc.abstractmethod
    async def trigger(self):
        """Trigger this task."""
        raise NotImplementedError
    
    async def loop(self):
        """Loop this task."""
        await asyncio.sleep(self.delay)
        while True:
            try:
                await self.trigger()
            except Exception:
                logging.warn(f"Failed to trigger task {self.__class__.__name__}, traceback: {traceback.format_exc()}")
            await asyncio.sleep(self.interval)

free_one_api/models/watchdog/task.py
RockChinQ/free-one-api
import abc

from . import task

class AbsWatchDog(metaclass=abc.ABCMeta):
    """Model of WatchDog."""
    
    tasks: list[task.AbsTask]
    """Added tasks."""
    
    @abc.abstractmethod
    async def run(self):
        """Run WatchDog system."""
        raise NotImplementedError

free_one_api/models/watchdog/wd.py
RockChinQ/free-one-api
import os
import sys
import asyncio
import logging

logging.basicConfig(level=logging.INFO)

if not os.path.exists('./data'):
    os.mkdir('./data')

from free_one_api.impls import app

def main():
    loop = asyncio.get_event_loop()
    
    application = loop.run_until_complete(app.make_application("./data/config.yaml"))

    logging.getLogger().setLevel(application.logging_level)
    loop.run_until_complete(application.run())

if __name__ == "__main__":
    main()

main.py
RockChinQ/free-one-api