driver.py

# 修改历史 (Revision History)
# ==================================
# 版本: v1.9
# 日期: 2026-06-25
# 修改说明: 在 create_completion 的代理自动检测逻辑中增加对全局代理环境变量 ALL_PROXY / all_proxy 的读取支持，防止用户仅设置 ALL_PROXY 时程序因无法识别而退化为直连，解决直连被微软重定向到主页的 HTML 报错。
#
# 版本: v1.8
# 日期: 2026-06-24
# 修改说明: 在 create_completion 中增加对 conversation 创建以及 attachments 上传接口 JSON 解析失败 (json.JSONDecodeError) 的捕获与详细错误报告，打印响应状态码、Content-Type 及前 500 字符，用于辅助诊断 Cloudflare 验证码拦截和网络劫持。
#
# 版本: v1.7
# 日期: 2026-06-24
# 修改说明: 在 create_completion 入口加入自动解析并清洗代理环境变量逻辑，将 socks5h:// 替换为 socks5://，彻底解决 Windows 全局代理下 WebSocket 握手时 libcurl 发生 TLS connection failed / invalid library 的缺陷。
#
# 版本: v1.6
# 日期: 2026-06-24
# 修改说明: 在 ws_connect 连接及流读取处增加 try-except 容错回退机制，若携带 Token 握手失败或报异常，自动退回到不带 accessToken 的匿名模式重试连接一次，提升过期 Token 下的可用性。
#
# 版本: v1.5
# 日期: 2026-06-24
# 修改说明: 在 challengeResponse 载荷中，当 challenge method 为 None/null 时，过滤掉 method 字段，防止微软服务器返回 invalid-event。
#
# 版本: v1.4
# 日期: 2026-06-24
# 修改说明: 将 print 中的 emoji 警告字符替换为 ASCII 文本，防止在 Windows GBK 终端下抛出 UnicodeEncodeError。


"""Pure-HTTP Copilot driver.

Speaks Microsoft Copilot's consumer chat protocol directly over a
Cloudflare-impersonating ``curl_cffi`` session — no browser required. This is the
low-level engine; most callers should use :class:`copilot.client.CopilotClient`.
See :mod:`copilot.browser` for the Playwright-backed fallback.
"""

import sys
if hasattr(sys.stdout, "reconfigure"):
    try:
        sys.stdout.reconfigure(errors="replace")
    except Exception:
        pass
if hasattr(sys.stderr, "reconfigure"):
    try:
        sys.stderr.reconfigure(errors="replace")
    except Exception:
        pass

import json
import time
from select import select
from typing import Dict, Optional
from urllib.parse import quote

from curl_cffi.const import CurlECode, CurlInfo
from curl_cffi.curl import CurlError
from curl_cffi.requests import Session, CurlWsFlag

# curl_cffi's WebSocket.recv() loops on CURLE_AGAIN forever (select() then retry)
# and never returns on an idle socket, so we drive the fragment loop ourselves to
# honour a deadline. CURL_SOCKET_BAD is libcurl's "no active socket" sentinel.
_CURL_SOCKET_BAD = -1

from .challenges import solve_copilot_challenge, solve_hashcash
from .models import AbstractProvider, Conversation, ImageResponse, ImageType
from .utils import drain_json, is_accepted_format, raise_for_status, to_bytes


class Copilot(AbstractProvider):
    label = "Microsoft Copilot"
    url = "https://copilot.microsoft.com"
    working = True
    supports_stream = True
    default_model = "Copilot"
    needs_auth = False  # consumer chat works anonymously (cookies only)
    websocket_url = "wss://copilot.microsoft.com/c/api/chat?api-version=2"
    conversation_url = f"{url}/c/api/conversations"

    def create_completion(
            self,
            prompt: str,
            stream: bool = False,
            proxy: str = None,
            timeout: int = 900,
            image: ImageType = None,
            conversation: Optional[Conversation] = None,
            conversation_id: str = None,
            return_conversation: bool = False,
            cookies: Dict[str, str] = None,
            access_token: str = None,
            **kwargs
        ):
        """Stream a Copilot reply to ``prompt``.

        Runs Copilot's own chat protocol over a Cloudflare-impersonating
        ``curl_cffi`` session: ``POST /c/api/conversations`` then a chat
        WebSocket (``send`` -> proof-of-work ``challenge`` -> ``appendText``* ->
        ``done``). The challenge is solved in-process (see
        :mod:`copilot.challenges`); no browser is required.

        ``prompt`` is the user message sent straight to the chat socket (the
        protocol has no separate system/role channel). Anonymous by default;
        pass ``cookies`` and/or ``access_token`` (e.g. exported from a signed-in
        browser session) to run as a logged-in user — required where anonymous
        consumer chat is region-restricted.

        Conversation targeting (first match wins):
          * ``conversation`` — reuse an existing :class:`Conversation` object;
          * ``conversation_id`` — resume a conversation by its id string (no
            create call), e.g. one saved from a previous run;
          * neither — create a fresh conversation. With ``return_conversation``
            the new :class:`Conversation` is yielded first.
        """
        # Resolve proxy: auto-detect environment variables and convert socks5h to socks5
        if proxy is None:
            import os
            proxy = (
                os.environ.get("HTTPS_PROXY")
                or os.environ.get("HTTP_PROXY")
                or os.environ.get("ALL_PROXY")
                or os.environ.get("https_proxy")
                or os.environ.get("http_proxy")
                or os.environ.get("all_proxy")
            )
        if proxy and "socks5h://" in proxy:
            proxy = proxy.replace("socks5h://", "socks5://")

        # Resolve auth: explicit args win, else fall back to the conversation's.
        if cookies is None and conversation is not None:
            cookies = conversation.cookies
        if access_token is None and conversation is not None:
            access_token = conversation.access_token

        # Auth model mirrors the browser:
        #   * REST calls (conversation create, attachment upload) authenticate by
        #     COOKIE only. Sending the token as an Authorization: Bearer header
        #     there gets a 401 (browsers never do it), so we don't.
        #   * the chat WebSocket carries the signed-in identity via its
        #     ?accessToken= param. This must be the Copilot chat token (MSAL scope
        #     ChatAI.ReadWrite, selected in browser._FIND_TOKEN_JS): a
        #     wrong-audience token 401s the WS upgrade, while *no* token makes the
        #     chat backend treat the session as anonymous -> chat-service-
        #     unavailable in geo-restricted regions (e.g. India).
        websocket_url = self.websocket_url
        if access_token:
            websocket_url = f"{websocket_url}&accessToken={quote(access_token)}"

        max_attempts = 3
        for attempt in range(1, max_attempts + 1):
            try:
                with Session(
                    timeout=timeout,
                    proxy=proxy,
                    impersonate="chrome",
                    cookies=cookies,
                ) as session:
                    # Establish cookies + Cloudflare clearance (anonymous is fine).
                    session.get(f"{self.url}/")

                    if conversation is not None:
                        conversation_id = conversation.conversation_id
                    elif conversation_id is not None:
                        pass  # resume an existing conversation by id; skip create
                    else:
                        response = session.post(self.conversation_url)
                        raise_for_status(response)
                        try:
                            conversation_id = response.json().get("id")
                        except json.JSONDecodeError as jde:
                            err_msg = (
                                f"Failed to parse conversation creation response as JSON. "
                                f"HTTP Status: {response.status_code}, "
                                f"Content-Type: {response.headers.get('content-type', 'unknown')}. "
                                f"Response Snippet: {response.text[:500]}"
                            )
                            print(f"[ERROR] {err_msg}", flush=True)
                            raise RuntimeError(err_msg) from jde
                        if return_conversation:
                            yield Conversation(conversation_id, session.cookies.jar)

                    images = []
                    if image is not None:
                        data = to_bytes(image)
                        response = session.post(
                            f"{self.url}/c/api/attachments",
                            headers={"content-type": is_accepted_format(data)},
                            data=data,
                        )
                        raise_for_status(response)
                        try:
                            img_url = response.json().get("url")
                        except json.JSONDecodeError as jde:
                            err_msg = (
                                f"Failed to parse image attachment response as JSON. "
                                f"HTTP Status: {response.status_code}, "
                                f"Content-Type: {response.headers.get('content-type', 'unknown')}. "
                                f"Response Snippet: {response.text[:500]}"
                            )
                            print(f"[ERROR] {err_msg}", flush=True)
                            raise RuntimeError(err_msg) from jde
                        images.append({"type": "image", "url": img_url})

                    send_frame = json.dumps({
                        "event": "send",
                        "conversationId": conversation_id,
                        "content": [*images, {"type": "text", "text": prompt}],
                        "mode": "chat",
                    }).encode()

                    try:
                        wss = session.ws_connect(websocket_url)
                        print(f"[WS OUTGOING (send_frame)] {send_frame.decode('utf-8', errors='ignore')}", flush=True)
                        wss.send(send_frame, CurlWsFlag.TEXT)
                        yield from self._read_stream(wss, send_frame, timeout)
                    except Exception as ws_err:
                        if access_token:
                            print(f"[Warning] 携带 Token 连接 WebSocket 失败 ({ws_err})，正在尝试退回到匿名模式连接...", flush=True)
                            anon_url = self.websocket_url
                            try:
                                wss = session.ws_connect(anon_url)
                                print(f"[WS OUTGOING (send_frame, anonymous)] {send_frame.decode('utf-8', errors='ignore')}", flush=True)
                                wss.send(send_frame, CurlWsFlag.TEXT)
                                yield from self._read_stream(wss, send_frame, timeout)
                            except Exception as anon_err:
                                raise RuntimeError(f"携带 Token 连接与匿名回退连接均失败。Token 错误: {ws_err}，匿名错误: {anon_err}") from anon_err
                        else:
                            raise
                    break
            except Exception as e:
                # Handle Windows-specific TLS initialization flakiness by retrying
                if attempt < max_attempts and ("TLS connect error" in str(e) or "invalid library" in str(e)):
                    print(f"[Warning] Windows TLS connection failed ({e}), retrying {attempt + 1}/{max_attempts}...", flush=True)
                    time.sleep(0.5)
                    continue
                raise

    def _read_stream(self, wss, send_frame: bytes, timeout: int, idle_timeout: int = 60):
        """Consume chat-socket frames, solving challenges, yielding text/images.

        ``idle_timeout`` bounds how long we wait for the *next* frame: the chat
        backend normally answers within a second, so prolonged silence means a
        stalled socket (or a challenge we failed to answer) — we raise rather
        than block for the full ``timeout``.
        """
        buffer = b""
        is_started = False
        answered_ids = set()
        image_prompt = None
        last_msg = None

        overall_deadline = time.time() + timeout
        while True:
            idle_deadline = time.time() + idle_timeout
            try:
                chunk = self._recv_frame(wss, min(overall_deadline, idle_deadline))
            except Exception:
                break  # socket closed/errored -> end of stream
            if chunk is None:  # deadline passed with no frame
                if time.time() >= overall_deadline:
                    raise TimeoutError(f"Copilot stream exceeded {timeout}s")
                raise TimeoutError(
                    f"Copilot chat socket went silent for {idle_timeout}s; "
                    f"last frame was {last_msg!r}."
                )

            buffer += chunk if isinstance(chunk, (bytes, bytearray)) else chunk.encode()
            messages, buffer = drain_json(buffer)
            for msg in messages:
                print(f"[WS INCOMING] {json.dumps(msg, ensure_ascii=False)}", flush=True)
                last_msg = msg
                event = msg.get("event")
                if event == "challenge" and msg.get("id") not in answered_ids:
                    token = self._solve_challenge(msg)
                    if token is None:
                        raise RuntimeError(
                            f"Unsolvable Copilot challenge (method={msg.get('method')!r}). "
                            "Microsoft may have escalated to a browser-only challenge; "
                            "fall back to copilot.browser.BrowserCopilot."
                        )
                    resp_data = {
                        "event": "challengeResponse",
                        "token": token,
                        "id": msg.get("id"),
                    }
                    if msg.get("method"):
                        resp_data["method"] = msg.get("method")
                    resp_bytes = json.dumps(resp_data).encode()
                    print(f"[WS OUTGOING (challengeResponse)] {resp_bytes.decode('utf-8')}", flush=True)
                    wss.send(resp_bytes, CurlWsFlag.TEXT)
                    answered_ids.add(msg.get("id"))
                    # The client re-sends the held message after a challenge.
                    # ONLY re-send if the challenge method is truthy (e.g. hashcash or copilot).
                    # Empty/no-op challenge (method=None) doesn't need message re-send.
                    if msg.get("method"):
                        print(f"[WS OUTGOING (re-send)] {send_frame.decode('utf-8', errors='ignore')}", flush=True)
                        wss.send(send_frame, CurlWsFlag.TEXT)
                elif event == "appendText":
                    is_started = True
                    yield msg.get("text")
                elif event == "generatingImage":
                    image_prompt = msg.get("prompt")
                elif event == "imageGenerated":
                    yield ImageResponse(msg.get("url"), image_prompt, {"preview": msg.get("thumbnailUrl")})
                elif event == "done":
                    return
                elif event == "error":
                    code = msg.get("errorCode") or msg
                    if code == "chat-service-unavailable":
                        raise RuntimeError(
                            "Copilot error: chat-service-unavailable. The chat backend is "
                            "typically geo-restricted; if you are outside a supported region, "
                            "retry via a proxy in a supported region, e.g. "
                            "create_completion(..., proxy='http://user:pass@host:port')."
                        )
                    raise RuntimeError(f"Copilot error: {code}")

        if not is_started:
            raise RuntimeError(f"Invalid response: {last_msg}")

    @staticmethod
    def _recv_frame(wss, deadline: float):
        """Block for one complete WS frame, or return ``None`` past ``deadline``.

        Reassembles libcurl's fragments like ``curl_cffi``'s own ``recv()`` but
        breaks out of the ``CURLE_AGAIN`` wait once ``deadline`` (epoch seconds)
        is reached, so an idle socket can't hang us indefinitely. Non-AGAIN curl
        errors (e.g. a closed connection) propagate to the caller.
        """
        sock_fd = wss.curl.getinfo(CurlInfo.ACTIVESOCKET)
        if sock_fd == _CURL_SOCKET_BAD:
            raise ConnectionError("WebSocket has no active socket")
        chunks = []
        while True:
            try:
                chunk, frame = wss.recv_fragment()
                chunks.append(chunk)
                if frame.bytesleft == 0 and frame.flags & CurlWsFlag.CONT == 0:
                    return b"".join(chunks)
            except CurlError as e:
                if e.code != CurlECode.AGAIN:
                    raise
                remaining = deadline - time.time()
                if remaining <= 0:
                    return None
                select([sock_fd], [], [], min(0.5, remaining))

    @staticmethod
    def _solve_challenge(msg: dict):
        """Return the challenge-response token, or ``None`` if we can't solve it.

        Copilot's chat socket precedes the answer with a challenge frame that the
        client must acknowledge. An *empty* challenge (no ``method``/``parameter``)
        only needs an acknowledging response, so we return an empty token; the
        proof-of-work variants are computed in :mod:`copilot.challenges`. A
        ``None`` return means the challenge needs a browser-solved token (e.g. a
        Cloudflare Turnstile) and the caller should surface that.
        """
        method = msg.get("method")
        parameter = msg.get("parameter")
        if not method and not parameter:
            return ""  # empty/no-op challenge: just acknowledge it
        if method == "hashcash" and parameter:
            return solve_hashcash(parameter)
        if method == "copilot" and parameter:
            return solve_copilot_challenge(parameter)
        # 'cloudflare' (Turnstile) / unknown PoW needs a browser-solved token.
        return None