import json, requests import gradio as gr import uuid import time import random import os # 创建新的聊天频道 def generate_uuid(): random_uuid = uuid.uuid4() random_uuid_str = str(random_uuid) formatted_uuid = f"{random_uuid_str[0:8]}-{random_uuid_str[9:13]}-{random_uuid_str[14:18]}-{random_uuid_str[19:23]}-{random_uuid_str[24:]}" return formatted_uuid # Deletes the conversation def delete_conversation(user_id,channel_id,cookie): url = f"https://claude.ai/api/organizations/{user_id}/chat_conversations/{channel_id}" payload = json.dumps(f"{channel_id}") headers = { 'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82', 'Accept-Language': 'en-US,en;q=0.5', 'Content-Type': 'application/json', 'Content-Length': '38', 'Referer': 'https://claude.ai/chats', 'Origin': 'https://claude.ai', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Connection': 'keep-alive', 'Cookie': f'{cookie}', 'TE': 'trailers' } response = requests.delete(url, headers=headers, data=payload) # Returns True if deleted or False if any error in deleting if response.status_code == 204: return True else: return False def create_new_chat(user_id,cookie): url = f"https://claude.ai/api/organizations/{user_id}/chat_conversations" uuid = generate_uuid() payload = json.dumps({"uuid": uuid, "name": ""}) headers = { 'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6', 'Referer': 'https://claude.ai/chats', 'Content-Type': 'application/json', 'Origin': 'https://claude.ai', 'Connection': 'keep-alive', 'Cookie': cookie, 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', } # response = requests.request("POST", url, headers=headers, data=payload) # 使用代理,解决国内网络不能访问问题 # response = request("POST", url, headers=headers, data=payload, proxies=proxies) response = requests.post(url, headers=headers, data=payload) print(f"新的对话页面:{response.json()}") # Returns JSON of the newly created conversation information return response.json() def get_content_type(file_path): # Function to determine content type based on file extension extension = os.path.splitext(file_path)[-1].lower() if extension == '.pdf': return 'application/pdf' elif extension == '.txt': return 'text/plain' elif extension == '.csv': return 'text/csv' # Add more content types as needed for other file types else: return 'application/octet-stream' def upload_attachment(cookie ,organization_id,file_path): url = 'https://claude.ai/api/convert_document' headers = { 'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.82', 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6', 'Referer': 'https://claude.ai/chats', 'Origin': 'https://claude.ai', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Connection': 'keep-alive', 'Cookie': f'{cookie}', 'TE': 'trailers' } file_name = os.path.basename(file_path) print(file_name) content_type = get_content_type(file_path) print(content_type) files = { 'file': (file_name, open(file_path, 'rb'), content_type), 'orgUuid': (None, organization_id) # 'orgUuid': organization_id } print(files) response = requests.post(url, headers=headers, files=files) print(f"response:{response.text}") if response.status_code == 200: return response.json() else: return False def request_to_v2(message, cookie, user_id,attachment=None,context=[]): context = [message] ## 是否加入文件 # Upload attachment if provided attachment = attachment.name attachments = [] if attachment: attachment_response = upload_attachment(cookie,user_id,attachment) if attachment_response: attachments = [attachment_response] print(f"attachment_response:{attachment_response}") else: bots = "\nError: Invalid file format. Please try again." print(bots) context += [bots] responses = [(u, b) for u, b in zip(context[::2], context[1::2])] return responses, context # Ensure attachments is an empty list when no attachment is provided if not attachment: attachments = [] response_json = create_new_chat(user_id,cookie) channel_id = response_json.get('uuid',"") print(channel_id) timeout = 5*60 #5分钟不回复,显示超时 context = [message] headers = { # ':Authority': 'claude.ai', # ':Method': 'POST', # ':Path': '/api/append_message', # ':Scheme': 'https', 'Content-Type': 'application/json', 'Origin': 'https://claude.ai', 'Referer': f'https://claude.ai/chat/{channel_id}', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36', 'Cookie': cookie, 'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8', # 'Accept-Encoding': 'gzip, deflate, br', 'Accept': 'text/event-stream, text/event-stream', 'Sec-Ch-Ua': '"Not.A/Brand";v="8", "Chromium";v="114", "Google Chrome";v="114"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"Windows"', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Connection': 'keep-alive', } post_msg_url = 'https://claude.ai/api/append_message' post_msg_data = { "completion":{ "incremental":True, "prompt": message, "timezone":"Asia/Shanghai", "model":"claude-2" }, "organization_uuid": user_id, "conversation_uuid": channel_id, "text": message, "attachments": attachments } post_msg_data = json.dumps(post_msg_data) # print(post_msg_data) response = { "status_code":-1, "text":"报错啦" } try: print(f"输入的问题:{message}") response = requests.post(post_msg_url, headers=headers, data=post_msg_data, verify=False, stream=False,timeout = timeout) bots = "" print("a"*100) for data in response.iter_lines(): if data: # 将字节字符串转换为字符串 data_str = data.decode('utf-8') # 删除开头的 "data: " 部分 data_str = data_str.replace("data: ", "") data_json = json.loads(data_str) # print(data_json) bots += data_json["completion"] print(bots) context += [bots] responses = [(u, b) for u, b in zip(context[::2], context[1::2])] return responses, context except Exception as e: print(">>>>>> 查询失败") print(e) print(response.text) response =">>>>>> 查询失败\n报错信息为:"+ str(e) + f"\n response信息:\n错误代码:{response.status_code}\n错误信息{response.text}" context += [response] responses = [(u, b) for u, b in zip(context[::2], context[1::2])] return responses, context finally: # time.sleep(random.uniform(1,2)) # ds = delete_conversation(user_id,channel_id,cookie) # if ds: # print(f"成功删除") # else: # print(f"删除失败") print("*-"*100) with gr.Blocks() as dialog_app: gr.HTML("""