from __future__ import annotations
from aiohttp import ClientSession
import json
from Crypto.Cipher import AES
from Crypto.Util import Padding
import base64
import hashlib
import time
import math
from ..typing import AsyncResult, Messages
from .base_provider import AsyncGeneratorProvider
from .helper import format_prompt
class GptForLove(AsyncGeneratorProvider):
url = "https://ai18.gptforlove.com"
working = True
supports_gpt_35_turbo = True
@classmethod
async def create_async_generator(
cls,
model: str,
messages: Messages,
proxy: str = None,
**kwargs
) -> AsyncResult:
if not model:
model = "gpt-3.5-turbo"
headers = {
"authority": "api.gptplus.one",
"accept": "application/json, text/plain, */*",
"accept-language": "de-DE,de;q=0.9,en-DE;q=0.8,en;q=0.7,en-US;q=0.6,nl;q=0.5,zh-CN;q=0.4,zh-TW;q=0.3,zh;q=0.2",
"content-type": "application/json",
"origin": cls.url,
"referer": f"{cls.url}/",
"sec-ch-ua": "\"Google Chrome\";v=\"117\", \"Not;A=Brand\";v=\"8\", \"Chromium\";v=\"117\"",
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": "Linux",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "cross-site",
"user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/117.0.0.0 Safari/537.36"
}
async with ClientSession(headers=headers) as session:
prompt = format_prompt(messages)
data = {
"prompt": prompt,
"options": {},
"systemMessage": kwargs.get("system_message", "You are ChatGPT, the version is GPT3.5, a large language model trained by OpenAI. Follow the user's instructions carefully."),
"temperature": kwargs.get("temperature", 0.8),
"top_p": kwargs.get("top_p", 1),
"secret": get_secret(),
}
async with session.post("https://api.gptplus.one/chat-process", json=data, proxy=proxy) as response:
response.raise_for_status()
async for line in response.content:
try:
line = json.loads(line)
except:
raise RuntimeError(f"Broken line: {line}")
if "detail" in line:
content = line["detail"]["choices"][0]["delta"].get("content")
if content:
yield content
elif "10分钟内提问超过了5次" in line:
raise RuntimeError("Rate limit reached")
else:
raise RuntimeError(f"Response: {line}")
def get_secret() -> str:
k = '14487141bvirvvG'
e = math.floor(time.time())
plaintext = str(e).encode('utf-8')
key = hashlib.md5(k.encode('utf-8')).digest()
cipher = AES.new(key, AES.MODE_ECB)
ciphertext = cipher.encrypt(Padding.pad(plaintext, AES.block_size, style='pkcs7'))
return base64.b64encode(ciphertext).decode()
