|
import copy |
|
import json |
|
import re |
|
import uuid |
|
|
|
from pathlib import Path |
|
|
|
from curl_cffi import requests |
|
from tclogger import logger, OSEnver |
|
from constants.envs import PROXIES |
|
|
|
|
|
class OpenaiAPI: |
|
def __init__(self): |
|
self.init_requests_params() |
|
|
|
def init_requests_params(self): |
|
self.api_base = "https://chat.openai.com/backend-anon" |
|
self.api_me = f"{self.api_base}/me" |
|
self.api_models = f"{self.api_base}/models" |
|
self.api_chat_requirements = f"{self.api_base}/sentinel/chat-requirements" |
|
self.api_conversation = f"{self.api_base}/conversation" |
|
self.uuid = str(uuid.uuid4()) |
|
self.requests_headers = { |
|
|
|
"Accept-Encoding": "gzip, deflate, br, zstd", |
|
"Accept-Language": "en-US,en;q=0.9", |
|
"Cache-Control": "no-cache", |
|
"Content-Type": "application/json", |
|
"Oai-Device-Id": self.uuid, |
|
"Oai-Language": "en-US", |
|
"Pragma": "no-cache", |
|
"Referer": "https://chat.openai.com/", |
|
"Sec-Ch-Ua": 'Google Chrome";v="123", "Not:A-Brand";v="8", "Chromium";v="123"', |
|
"Sec-Ch-Ua-Mobile": "?0", |
|
"Sec-Ch-Ua-Platform": '"Windows"', |
|
"Sec-Fetch-Dest": "empty", |
|
"Sec-Fetch-Mode": "cors", |
|
"Sec-Fetch-Site": "same-origin", |
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36", |
|
} |
|
|
|
def log_request(self, url, method="GET"): |
|
logger.note(f"> {method}:", end=" ") |
|
logger.mesg(f"{url}", end=" ") |
|
|
|
def log_response(self, res: requests.Response, stream=False, verbose=False): |
|
status_code = res.status_code |
|
status_code_str = f"[{status_code}]" |
|
|
|
if status_code == 200: |
|
logger_func = logger.success |
|
else: |
|
logger_func = logger.warn |
|
|
|
logger_func(status_code_str) |
|
|
|
if verbose: |
|
if stream: |
|
if not hasattr(self, "content_offset"): |
|
self.content_offset = 0 |
|
|
|
for line in res.iter_lines(): |
|
line = line.decode("utf-8") |
|
line = re.sub(r"^data:\s*", "", line) |
|
if re.match(r"^\[DONE\]", line): |
|
logger.success("\n[Finished]") |
|
break |
|
line = line.strip() |
|
if line: |
|
try: |
|
data = json.loads(line, strict=False) |
|
message_role = data["message"]["author"]["role"] |
|
message_status = data["message"]["status"] |
|
if ( |
|
message_role == "assistant" |
|
and message_status == "in_progress" |
|
): |
|
content = data["message"]["content"]["parts"][0] |
|
delta_content = content[self.content_offset :] |
|
self.content_offset = len(content) |
|
logger_func(delta_content, end="") |
|
except Exception as e: |
|
logger.warn(e) |
|
else: |
|
logger_func(res.json()) |
|
|
|
def get_models(self): |
|
self.log_request(self.api_models) |
|
res = requests.get( |
|
self.api_models, |
|
headers=self.requests_headers, |
|
proxies=PROXIES, |
|
timeout=10, |
|
impersonate="chrome120", |
|
) |
|
self.log_response(res) |
|
|
|
def auth(self): |
|
self.log_request(self.api_chat_requirements, method="POST") |
|
res = requests.post( |
|
self.api_chat_requirements, |
|
headers=self.requests_headers, |
|
proxies=PROXIES, |
|
timeout=10, |
|
impersonate="chrome120", |
|
) |
|
self.chat_requirements_token = res.json()["token"] |
|
self.log_response(res) |
|
|
|
def transform_messages(self, messages: list[dict]): |
|
def get_role(role): |
|
if role in ["system", "user", "assistant"]: |
|
return role |
|
else: |
|
return "system" |
|
|
|
new_messages = [ |
|
{ |
|
"author": {"role": get_role(message["role"])}, |
|
"content": {"content_type": "text", "parts": [message["content"]]}, |
|
"metadata": {}, |
|
} |
|
for message in messages |
|
] |
|
return new_messages |
|
|
|
def chat_completions(self, messages: list[dict]): |
|
new_headers = { |
|
"Accept": "text/event-stream", |
|
"Openai-Sentinel-Chat-Requirements-Token": self.chat_requirements_token, |
|
} |
|
requests_headers = copy.deepcopy(self.requests_headers) |
|
requests_headers.update(new_headers) |
|
post_data = { |
|
"action": "next", |
|
"messages": self.transform_messages(messages), |
|
"parent_message_id": "", |
|
"model": "text-davinci-002-render-sha", |
|
"timezone_offset_min": -480, |
|
"suggestions": [], |
|
"history_and_training_disabled": False, |
|
"conversation_mode": {"kind": "primary_assistant"}, |
|
"force_paragen": False, |
|
"force_paragen_model_slug": "", |
|
"force_nulligen": False, |
|
"force_rate_limit": False, |
|
"websocket_request_id": str(uuid.uuid4()), |
|
} |
|
self.log_request(self.api_conversation, method="POST") |
|
s = requests.Session() |
|
res = s.post( |
|
self.api_conversation, |
|
headers=requests_headers, |
|
json=post_data, |
|
proxies=PROXIES, |
|
timeout=10, |
|
impersonate="chrome120", |
|
stream=True, |
|
) |
|
self.log_response(res, stream=True, verbose=True) |
|
|
|
|
|
if __name__ == "__main__": |
|
api = OpenaiAPI() |
|
|
|
api.auth() |
|
messages = [ |
|
{"role": "system", "content": "I am Niansuh"}, |
|
{"role": "system", "content": "I have a cat named Lucky"}, |
|
{"role": "user", "content": "Repeat my name and my cat's name"}, |
|
{ |
|
"role": "assistant", |
|
"content": "Your name is Niansuh and your cat's name is Lucky.", |
|
}, |
|
{"role": "user", "content": "summarize our conversation"}, |
|
] |
|
api.chat_completions(messages) |
|
|
|
|
|
|
