import json import time import hashlib import base64 import asyncio import aiohttp from fastapi import FastAPI, Request, Response from typing import Optional, Dict, List, Any import uvicorn from pydantic import BaseModel # 配置参数 API_CHALLENGE_URL = 'https://api.eqing.tech/api/altcaptcha/challenge' # 验证码挑战接口 NEXTWAY_CHAT_URL = 'https://origin.eqing.tech/api/openai/v1/chat/completions' # 聊天完成接口 CREDENTIAL_EXPIRY_MARGIN = 60 * 1000 # 凭证过期边界时间（60秒，毫秒单位） PORT = 7860 # 服务器端口 API_ENDPOINT = '/v1/chat/completions' # API端点 MODEL_NAME = "gpt-4o-free" # 默认模型名称 REQUEST_TIMEOUT = 480 # 请求超时时间（秒） MAX_RETRIES = 3 # 最大重试次数 RETRY_DELAY = 1 # 重试延迟（秒） # 全局变量 current_credential = None # 当前凭证 credential_expiry = None # 凭证过期时间 is_refreshing_credential = False # 是否正在刷新凭证 # 模型映射字典 MODEL_MAPPING = { 'gpt-4o-all-lite': 'gpt-4o-mini', 'gpt-4o-image': 'gpt-4o-mini-image-free', 'grok-3-re': 'grok-3-free', 'gemini-2.0-flash': 'gemini-2.0-flash-free' } app = FastAPI() class ChatRequest(BaseModel): """聊天请求的数据模型""" messages: List[Dict[str, str]] model: Optional[str] stream: Optional[bool] = True temperature: Optional[float] = 0.5 presence_penalty: Optional[float] = 0 frequency_penalty: Optional[float] = 0 top_p: Optional[float] = 1 max_tokens: Optional[int] = 4000 async def extract_content(text: str) -> str: """ 从响应文本中提取内容 Args: text: 响应文本 Returns: 提取的AI响应内容 """ lines = text.split('\n') ai_response = '' ignored_id = 'chatcmpl-Em8sqOsIIDLYPNvC65WfFvqv' created = 1687070102 for line in lines: line = line.strip() if line.startswith('data:'): data_str = line[5:].strip() if not data_str or data_str in ['[ORIGIN]', '[DONE]']: continue try: json_data = json.loads(data_str) # 跳过特定的响应 if json_data.get('id') == ignored_id or json_data.get('created') == created: continue # 提取内容 if (json_data.get('choices') and json_data['choices'][0].get('delta') and 'content' in json_data['choices'][0]['delta']): content = json_data['choices'][0]['delta']['content'] ai_response += content except json.JSONDecodeError: print(f'跳过非JSON数据') return ai_response async def solve_challenge(challenge: str, salt: str, algorithm: str = "SHA-512", max_number: int = 1000000): """ 解决验证码挑战 Args: challenge: 挑战字符串 salt: 盐值 algorithm: 哈希算法 max_number: 最大尝试次数 Returns: 解决方案字典 """ start_time = time.time() for number in range(max_number): hash_value = await verify_hash(salt, number, algorithm) if hash_value == challenge: return { "number": number, "took": int((time.time() - start_time) * 1000) } return None async def verify_hash(salt: str, number: int, algorithm: str) -> str: """ 验证哈希值 Args: salt: 盐值 number: 数字 algorithm: 哈希算法 Returns: 哈希字符串 """ input_str = f"{salt}{number}" if algorithm == "SHA-512": hash_obj = hashlib.sha512(input_str.encode()) return hash_obj.hexdigest() elif algorithm == "SHA-256": hash_obj = hashlib.sha256(input_str.encode()) return hash_obj.hexdigest() else: raise ValueError(f"不支持的算法: {algorithm}") async def generate_credential(): """ 生成新的凭证 Returns: 凭证信息字典 """ global current_credential, credential_expiry async with aiohttp.ClientSession() as session: try: async with session.get(API_CHALLENGE_URL) as response: if response.status != 200: print(f"验证码请求失败，状态码: {response.status}") return None data = await response.json() solution = await solve_challenge( data['challenge'], data['salt'], data['algorithm'], data['maxnumber'] ) if not solution: print("解决验证码挑战失败") return None credential_obj = { "algorithm": data['algorithm'], "challenge": data['challenge'], "number": solution['number'], "salt": data['salt'], "signature": data['signature'], "took": solution['took'] } credential = base64.b64encode(json.dumps(credential_obj).encode()).decode() expiry = int(data['salt'].split('?expires=')[1]) * 1000 return {"credential": credential, "expiry": expiry} except Exception as e: print(f"生成凭证时出错: {e}") return None async def get_credential(): """ 获取有效的凭证 Returns: 当前有效的凭证 """ global current_credential, credential_expiry, is_refreshing_credential if (not current_credential or not credential_expiry or credential_expiry <= time.time() * 1000 + CREDENTIAL_EXPIRY_MARGIN): if not is_refreshing_credential: is_refreshing_credential = True try: cred_data = await generate_credential() if cred_data: current_credential = cred_data['credential'] credential_expiry = cred_data['expiry'] finally: is_refreshing_credential = False else: await asyncio.sleep(2) # 等待其他进程完成凭证刷新 return current_credential @app.post(API_ENDPOINT) async def chat_endpoint(request: ChatRequest): """ 聊天API端点 Args: request: 聊天请求对象 Returns: 聊天响应 """ try: model = MODEL_MAPPING.get(request.model, request.model or MODEL_NAME) response_content = await handle_chat_request(model, request.messages) if response_content is None: return Response( content="从API获取响应失败", status_code=500 ) return { "choices": [{ "message": { "role": "assistant", "content": response_content.strip() }, "finish_reason": "stop", "index": 0 }], "model": model, "object": "chat.completion" } except Exception as e: print(f"处理聊天请求时出错: {e}") return Response(content="内部服务器错误", status_code=500) async def handle_chat_request(model: str, messages: List[Dict[str, str]]): """ 处理聊天请求 Args: model: 模型名称 messages: 消息列表 Returns: 聊天响应内容 """ captcha_token = await get_credential() if not captcha_token: return None body = { "messages": messages, "stream": True, "model": model, "temperature": 0.5, "presence_penalty": 0, "frequency_penalty": 0, "top_p": 1, "max_tokens": 4000, "captchaToken": captcha_token } timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout) as session: try: async with session.post(NEXTWAY_CHAT_URL, json=body) as response: if response.status != 200: print(f"请求失败，状态码: {response.status}") return None complete_response = "" buffer = "" # 用于存储未完成的数据块 async for chunk in response.content.iter_chunks(): if chunk[0]: # chunk[0] 是数据，chunk[1] 是布尔值表示是否是最后一块 try: chunk_text = (buffer + chunk[0].decode()).strip() buffer = "" # 清空缓冲区 # 处理可能的不完整JSON if chunk_text.endswith('}'): content = await extract_content(chunk_text) if content: complete_response += content else: buffer = chunk_text # 存储不完整的数据到缓冲区 except UnicodeDecodeError as e: print(f"解码错误: {e}") buffer = "" # 清空缓冲区，跳过损坏的数据 continue # 检查最终响应 if not complete_response: print("警告: 收到空响应") return None return complete_response.strip() except asyncio.TimeoutError: print("请求超时") return None except Exception as e: print(f"处理请求时出错: {e}") await get_credential() # 刷新凭证 return None if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=PORT)