import json import os import shutil import subprocess import tempfile import time from typing import Optional from curl_cffi import AsyncSession, Response from fastapi import FastAPI, Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from loguru import logger from starlette.middleware.cors import CORSMiddleware from app.config import SCRIPT_URL, FP, API_KEY, MODELS, SYSTEM_PROMPT_INJECT, TIMEOUT from app.errors import CursorWebError from app.models import ChatCompletionRequest, Message, ModelsResponse, Model, Usage from app.utils import error_wrapper, to_async, generate_random_string, non_stream_chat_completion, \ stream_chat_completion, safe_stream_wrapper main_code = open('./jscode/main.js', 'r', encoding='utf-8').read() env_code = open('./jscode/env.js', 'r', encoding='utf-8').read() app = FastAPI() security = HTTPBearer() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.post("/v1/chat/completions") async def chat_completions( request: ChatCompletionRequest, credentials: HTTPAuthorizationCredentials = Depends(security), ): """处理聊天完成请求""" if credentials.credentials != API_KEY: raise HTTPException(401, 'api key 错误') chat_generator = cursor_chat(request) # async for c in chat_generator: # logger.debug(c) if request.stream: return await error_wrapper(safe_stream_wrapper, stream_chat_completion, request, chat_generator) else: return await error_wrapper(non_stream_chat_completion, request, chat_generator) @app.get("/v1/models") async def list_models(credentials: HTTPAuthorizationCredentials = Depends(security)): models = MODELS.split(',') model_list = [] for model_id in models: model_list.append( Model( id=model_id, # 使用model name作为对外的id object="model", created=int(time.time()), owned_by='', ) ) return ModelsResponse(object="list", data=model_list) def to_cursor_messages(list_openai_message: list[Message]): if list_openai_message is None: list_openai_message = [] result = [] if len(list_openai_message) > 0: if list_openai_message[0].role == 'system': if isinstance(list_openai_message[0].content, str): list_openai_message[0].content += f'\n{SYSTEM_PROMPT_INJECT}' else: list_openai_message.insert(0, Message(role='system', content=f'\n{SYSTEM_PROMPT_INJECT}', tool_call_id=None, tool_calls=None)) for m in list_openai_message: if not m: continue text = '' if isinstance(m.content, str): text = m.content else: for content in m.content: if not content.text: continue text = text + content.text message = { 'role': m.role, 'parts': [{ 'type': 'text', 'text': text }] } result.append(message) return result def parse_sse_line(line: str) -> Optional[str]: """解析SSE数据行""" line = line.strip() if line.startswith("data: "): return line[6:] # 去掉 'data: ' 前缀 return None async def cursor_chat(request: ChatCompletionRequest): json_data = { "context": [ ], "model": request.model, "id": generate_random_string(16), "messages": to_cursor_messages(request.messages), "trigger": "submit-message" } async with AsyncSession(impersonate='chrome', timeout=TIMEOUT) as session: x_is_human = await get_x_is_human(session) logger.debug(x_is_human) headers = { 'User-Agent': FP.get("userAgent"), # 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Content-Type': 'application/json', 'sec-ch-ua-platform': '"Windows"', 'x-path': '/api/chat', 'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"', 'x-method': 'POST', 'sec-ch-ua-bitness': '"64"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-arch': '"x86"', 'x-is-human': x_is_human, 'sec-ch-ua-platform-version': '"19.0.0"', 'origin': 'https://cursor.com', 'sec-fetch-site': 'same-origin', 'sec-fetch-mode': 'cors', 'sec-fetch-dest': 'empty', 'referer': 'https://cursor.com/en-US/learn/how-ai-models-work', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'priority': 'u=1, i', } # logger.debug(json_data) async with session.stream("POST", 'https://cursor.com/api/chat', headers=headers, json=json_data, impersonate='chrome') as response: response: Response if response.status_code != 200: text = await response.atext() if 'Attention Required! | Cloudflare' in text: text = 'Cloudflare 403' raise CursorWebError(response.status_code, text) async for line in response.aiter_lines(): line = line.decode("utf-8") data = parse_sse_line(line) if not data: continue if data and data.strip(): try: event_data = json.loads(data) if event_data.get('type') == 'error': raise CursorWebError(response.status_code, event_data.get('errorText', 'errorText为空')) if event_data.get('type') == 'finish': usage = event_data.get('messageMetadata', {}).get('usage') if not usage: continue yield Usage(prompt_tokens=usage.get('inputTokens'), completion_tokens=usage.get('outputTokens'), total_tokens=usage.get('totalTokens')) return delta = event_data.get('delta') # logger.debug(delta) if not delta: continue yield delta except json.JSONDecodeError: continue async def get_x_is_human(session: AsyncSession): headers = { 'User-Agent': FP.get("userAgent"), # 'Accept-Encoding': 'gzip, deflate, br, zstd', 'sec-ch-ua-arch': '"x86"', 'sec-ch-ua-platform': '"Windows"', 'sec-ch-ua': '"Chromium";v="140", "Not=A?Brand";v="24", "Google Chrome";v="140"', 'sec-ch-ua-bitness': '"64"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform-version': '"19.0.0"', 'sec-fetch-site': 'same-origin', 'sec-fetch-mode': 'no-cors', 'sec-fetch-dest': 'script', 'referer': 'https://cursor.com/en-US/learn/how-ai-models-work', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', } response = await session.get(SCRIPT_URL, headers=headers, impersonate='chrome') cursor_js = response.text # 替换指纹 main = (main_code.replace("$$currentScriptSrc$$", SCRIPT_URL) .replace("$$UNMASKED_VENDOR_WEBGL$$", FP.get("UNMASKED_VENDOR_WEBGL")) .replace("$$UNMASKED_RENDERER_WEBGL$$", FP.get("UNMASKED_RENDERER_WEBGL")) .replace("$$userAgent$$", FP.get("userAgent"))) # 替换代码 main = main.replace('$$env_jscode$$', env_code) main = main.replace("$$cursor_jscode$$", cursor_js) return await runjs(main) @to_async def runjs(jscode: str) -> str: """ 执行 JavaScript 代码并返回标准输出内容。 Args: jscode: 要执行的 JavaScript 代码字符串 Returns: Node.js 程序的标准输出内容 Raises: FileNotFoundError: Node.js 未安装或不在系统 PATH 中 subprocess.CalledProcessError: Node.js 程序执行失败，异常信息包含 stdout 和 stderr """ temp_dir = tempfile.mkdtemp() try: js_file_path = os.path.join(temp_dir, "script.js") with open(js_file_path, "w", encoding="utf-8") as f: f.write(jscode) result = subprocess.run( ['node', js_file_path], capture_output=True, text=True, encoding="utf-8" ) if result.returncode != 0: error_msg = f"Node.js 执行失败 (退出码: {result.returncode})\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" logger.error(error_msg) raise subprocess.CalledProcessError(result.returncode, ['node', js_file_path], result.stdout, result.stderr) return result.stdout.strip() finally: shutil.rmtree(temp_dir) if __name__ == "__main__": import uvicorn uvicorn.run( "main:app", host="0.0.0.0", port=8000, reload=False, log_level="info", )