|
|
from __future__ import annotations |
|
|
|
|
|
from urllib.parse import unquote |
|
|
|
|
|
from ...typing import AsyncResult, Messages |
|
|
from ..base_provider import AbstractProvider |
|
|
from ...webdriver import WebDriver |
|
|
from ...requests import Session, get_session_from_browser |
|
|
|
|
|
class AiChatting(AbstractProvider): |
|
|
url = "https://www.aichatting.net" |
|
|
supports_gpt_35_turbo = True |
|
|
_session: Session = None |
|
|
|
|
|
@classmethod |
|
|
def create_completion( |
|
|
cls, |
|
|
model: str, |
|
|
messages: Messages, |
|
|
stream: bool, |
|
|
proxy: str = None, |
|
|
timeout: int = 120, |
|
|
webdriver: WebDriver = None, |
|
|
**kwargs |
|
|
) -> AsyncResult: |
|
|
if not cls._session: |
|
|
cls._session = get_session_from_browser(cls.url, webdriver, proxy, timeout) |
|
|
visitorId = unquote(cls._session.cookies.get("aichatting.website.visitorId")) |
|
|
|
|
|
headers = { |
|
|
"accept": "application/json, text/plain, */*", |
|
|
"lang": "en", |
|
|
"source": "web" |
|
|
} |
|
|
data = { |
|
|
"roleId": 0, |
|
|
} |
|
|
try: |
|
|
response = cls._session.post("https://aga-api.aichatting.net/aigc/chat/record/conversation/create", json=data, headers=headers) |
|
|
response.raise_for_status() |
|
|
conversation_id = response.json()["data"]["conversationId"] |
|
|
except Exception as e: |
|
|
cls.reset() |
|
|
raise e |
|
|
headers = { |
|
|
"authority": "aga-api.aichatting.net", |
|
|
"accept": "text/event-stream,application/json, text/event-stream", |
|
|
"lang": "en", |
|
|
"source": "web", |
|
|
"vtoken": visitorId, |
|
|
} |
|
|
data = { |
|
|
"spaceHandle": True, |
|
|
"roleId": 0, |
|
|
"messages": messages, |
|
|
"conversationId": conversation_id, |
|
|
} |
|
|
response = cls._session.post("https://aga-api.aichatting.net/aigc/chat/v2/stream", json=data, headers=headers, stream=True) |
|
|
response.raise_for_status() |
|
|
for chunk in response.iter_lines(): |
|
|
if chunk.startswith(b"data:"): |
|
|
yield chunk[5:].decode().replace("-=- --", " ").replace("-=-n--", "\n").replace("--@DONE@--", "") |
|
|
|
|
|
@classmethod |
|
|
def reset(cls): |
|
|
cls._session = None |
