Delete replicate.py

replicate.py

@@ -1,199 +0,0 @@
-import json
-import uuid
-import asyncio
-from typing import Dict, Optional, Tuple, List, AsyncGenerator, Any
-import httpx
-
-from utils import get_proxies, load_module, create_proxy_mounts
-from config import AMAZONQ_API_URL, DEFAULT_HEADERS
-
-try:
-    _parser = load_module("v2_claude_parser", "claude_parser.py")
-    EventStreamParser = _parser.EventStreamParser
-    extract_event_info = _parser.extract_event_info
-except Exception as e:
-    print(f"Warning: Failed to load claude_parser: {e}")
-    EventStreamParser = None
-    extract_event_info = None
-
-class StreamTracker:
-    def __init__(self):
-        self.has_content = False
-
-    async def track(self, gen: AsyncGenerator[str, None]) -> AsyncGenerator[str, None]:
-        async for item in gen:
-            if item:
-                self.has_content = True
-            yield item
-
-def load_template() -> Tuple[str, Dict[str, str]]:
-    """
-    加载 Amazon Q API 请求模板
-
-    Returns:
-        (url, headers): API 端点 URL 和默认请求头
-    """
-    return AMAZONQ_API_URL, DEFAULT_HEADERS.copy()
-
-def _merge_headers(as_log: Dict[str, str], bearer_token: str) -> Dict[str, str]:
-    headers = dict(as_log)
-    for k in list(headers.keys()):
-        kl = k.lower()
-        if kl in ("content-length","host","connection","transfer-encoding"):
-            headers.pop(k, None)
-    def set_header(name: str, value: str):
-        for key in list(headers.keys()):
-            if key.lower() == name.lower():
-                del headers[key]
-        headers[name] = value
-    set_header("Authorization", f"Bearer {bearer_token}")
-    set_header("amz-sdk-invocation-id", str(uuid.uuid4()))
-    return headers
-
-async def send_chat_request(
-    access_token: str,
-    messages: List[Dict[str, Any]],
-    model: Optional[str] = None,
-    stream: bool = False,
-    timeout: Tuple[int,int] = (30,300),
-    client: Optional[httpx.AsyncClient] = None,
-    raw_payload: Dict[str, Any] = None
-) -> Tuple[Optional[str], Optional[AsyncGenerator[str, None]], StreamTracker, Optional[AsyncGenerator[Any, None]]]:
-    """
-    发送聊天请求到 Amazon Q API
-
-    Args:
-        access_token: Amazon Q access token
-        messages: 消息列表(已废弃,使用 raw_payload)
-        model: 模型名称(已废弃,使用 raw_payload)
-        stream: 是否流式响应
-        timeout: 超时配置
-        client: HTTP 客户端
-        raw_payload: Claude API 转换后的请求体(必需)
-    """
-    if raw_payload is None:
-        raise ValueError("raw_payload is required")
-
-    url, headers_from_log = load_template()
-    headers_from_log["amz-sdk-invocation-id"] = str(uuid.uuid4())
-
-    # Use raw payload (for Claude API)
-    body_json = raw_payload
-    # Ensure conversationId is set if missing
-    if "conversationState" in body_json and "conversationId" not in body_json["conversationState"]:
-         body_json["conversationState"]["conversationId"] = str(uuid.uuid4())
-
-    payload_str = json.dumps(body_json, ensure_ascii=False)
-    headers = _merge_headers(headers_from_log, access_token)
-
-    local_client = False
-    if client is None:
-        local_client = True
-        mounts = create_proxy_mounts()
-        # 增加连接超时时间，避免 TLS 握手超时
-        timeout_config = httpx.Timeout(connect=60.0, read=timeout[1], write=timeout[0], pool=10.0)
-        # 只在有代理时才传递 mounts 参数
-        if mounts:
-            client = httpx.AsyncClient(mounts=mounts, timeout=timeout_config)
-        else:
-            client = httpx.AsyncClient(timeout=timeout_config)
-
-    # Use manual request sending to control stream lifetime
-    req = client.build_request("POST", url, headers=headers, content=payload_str)
-
-    resp = None
-    try:
-        resp = await client.send(req, stream=True)
-
-        if resp.status_code >= 400:
-            try:
-                await resp.read()
-                err = resp.text
-            except Exception:
-                err = f"HTTP {resp.status_code}"
-            await resp.aclose()
-            if local_client:
-                await client.aclose()
-            raise httpx.HTTPError(f"Upstream error {resp.status_code}: {err}")
-
-        tracker = StreamTracker()
-
-        # Track if the response has been consumed to avoid double-close
-        response_consumed = False
-
-        async def _iter_events() -> AsyncGenerator[Any, None]:
-            nonlocal response_consumed
-            try:
-                # Use EventStreamParser from claude_parser.py
-                async def byte_gen():
-                    async for chunk in resp.aiter_bytes():
-                        if chunk:
-                            yield chunk
-
-                async for message in EventStreamParser.parse_stream(byte_gen()):
-                    event_info = extract_event_info(message)
-                    if event_info:
-                        event_type = event_info.get('event_type')
-                        payload = event_info.get('payload')
-                        if event_type and payload:
-                            yield (event_type, payload)
-            except Exception:
-                if not tracker.has_content:
-                    raise
-            finally:
-                response_consumed = True
-                await resp.aclose()
-                if local_client:
-                    await client.aclose()
-
-        if stream:
-            # Wrap generator to ensure cleanup on early termination
-            async def _safe_iter_events():
-                try:
-                    # 托底方案: 300秒强制超时
-                    async with asyncio.timeout(300):
-                        async for item in _iter_events():
-                            yield item
-                except asyncio.TimeoutError:
-                    # 超时强制关闭
-                    if resp and not resp.is_closed:
-                        await resp.aclose()
-                    if local_client and client:
-                        await client.aclose()
-                    raise
-                except GeneratorExit:
-                    # Generator was closed without being fully consumed
-                    # Ensure cleanup happens even if finally block wasn't reached
-                    if resp and not resp.is_closed:
-                        await resp.aclose()
-                    if local_client and client:
-                        await client.aclose()
-                    raise
-                except Exception:
-                    # Any exception should also trigger cleanup
-                    if resp and not resp.is_closed:
-                        await resp.aclose()
-                    if local_client and client:
-                        await client.aclose()
-                    raise
-            return None, None, tracker, _safe_iter_events()
-        else:
-            # Non-streaming: consume all events
-            try:
-                async for _ in _iter_events():
-                    pass
-            finally:
-                # Ensure response is closed even if iteration is incomplete
-                if not response_consumed and resp:
-                    await resp.aclose()
-                    if local_client:
-                        await client.aclose()
-            return None, None, tracker, None
-
-    except Exception:
-        # Critical: close response on any exception before generators are created
-        if resp and not resp.is_closed:
-            await resp.aclose()
-        if local_client and client:
-            await client.aclose()
-        raise