from __future__ import annotations import time import hashlib import uuid from ...typing import AsyncResult, Messages from ...requests import StreamSession, raise_for_status from ...errors import RateLimitError from ..base_provider import AsyncGeneratorProvider, ProviderModelMixin class ChatForAi(AsyncGeneratorProvider, ProviderModelMixin): url = "https://chatforai.store" working = False default_model = "gpt-3.5-turbo" supports_message_history = True supports_gpt_35_turbo = True @classmethod async def create_async_generator( cls, model: str, messages: Messages, proxy: str = None, timeout: int = 120, temperature: float = 0.7, top_p: float = 1, **kwargs ) -> AsyncResult: model = cls.get_model(model) headers = { "Content-Type": "text/plain;charset=UTF-8", "Origin": cls.url, "Referer": f"{cls.url}/?r=b", } async with StreamSession(impersonate="chrome", headers=headers, proxies={"https": proxy}, timeout=timeout) as session: timestamp = int(time.time() * 1e3) conversation_id = str(uuid.uuid4()) data = { "conversationId": conversation_id, "conversationType": "chat_continuous", "botId": "chat_continuous", "globalSettings":{ "baseUrl": "https://api.openai.com", "model": model, "messageHistorySize": 5, "temperature": temperature, "top_p": top_p, **kwargs }, "prompt": "", "messages": messages, "timestamp": timestamp, "sign": generate_signature(timestamp, "", conversation_id) } async with session.post(f"{cls.url}/api/handle/provider-openai", json=data) as response: await raise_for_status(response) async for chunk in response.iter_content(): if b"https://chatforai.store" in chunk: raise RuntimeError(f"Response: {chunk.decode(errors='ignore')}") yield chunk.decode(errors="ignore") def generate_signature(timestamp: int, message: str, id: str): buffer = f"{id}:{timestamp}:{message}:h496Jd6b" return hashlib.sha256(buffer.encode()).hexdigest()