| 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) |
|
|
| |
|
|
