business-assistant/scripts/test_deepseek_proxy.py

"""Read .env and diagnose the fixed DeepSeek gateway (POST by default).

Run on the deployment server: uv run --frozen python scripts/test_deepseek_proxy.py
No OpenAI SDK path joining, retries, or response cache is used.
"""

from __future__ import annotations

import argparse
from http.client import HTTPConnection, HTTPSConnection, IncompleteRead
import json
import os
import re
import socket
import sys
import time
from pathlib import Path
from urllib.parse import parse_qsl, quote, quote_plus, urlsplit, urlunsplit

def load_env_values(path: Path) -> dict[str, str]:
    """Read just the three diagnostic settings, with no third-party dependency."""
    values = {}
    if not path.is_file():
        return values
    for line in path.read_text(encoding="utf-8-sig").splitlines():
        line = line.strip()
        if line.startswith("export "):
            line = line[7:]
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        key, value = key.strip(), value.strip()
        if key not in {"DEEPSEEK_BASE_URL", "DEEPSEEK_API_KEY", "DEEPSEEK_MODEL"}:
            continue
        if value.startswith(("'", '"')):
            quote_char = value[0]
            end = value.find(quote_char, 1)
            if end < 0 or (value[end + 1:].strip() and not value[end + 1:].strip().startswith("#")):
                raise ValueError("invalid_env_quoting")
            value = value[1:end]
        else:
            value = value.split(" #", 1)[0].strip()
        values[key] = value
    return values


def _redact(value: str, secrets: list[str]) -> str:
    for secret in sorted(set(secrets), key=len, reverse=True):
        if secret:
            for variant in (secret, quote(secret, safe=""), quote_plus(secret)):
                value = value.replace(variant, "[REDACTED]")
    value = re.sub(r"(?i)(proxyToken[\"'\s]*[:=][\"'\s]*)[^&\s\"'<>]+", r"\1[REDACTED]", value)
    return re.sub(r"(?i)(Bearer\s+)\S+", r"\1[REDACTED]", value)


def probe(*, timeout: float = 30, method: str = "POST") -> dict:
    """Return safe diagnostics; environment variables override the root .env."""
    if method not in {"GET", "POST"}:
        return {"ok": False, "code": "invalid_method"}
    env_file = Path(__file__).resolve().parents[1] / ".env"
    try:
        values = load_env_values(env_file)
    except (OSError, ValueError):
        return {"ok": False, "code": "invalid_env_file", "hint": ".env无法读取或引号格式无效"}
    values.update({key: os.environ[key] for key in (
        "DEEPSEEK_BASE_URL", "DEEPSEEK_API_KEY", "DEEPSEEK_MODEL"
    ) if key in os.environ})
    url = values.get("DEEPSEEK_BASE_URL", "").strip()
    api_key = values.get("DEEPSEEK_API_KEY", "").strip()
    model = values.get("DEEPSEEK_MODEL", "deepseek-flash").strip()
    result = {
        "ok": False,
        "env_file_exists": env_file.is_file(),
        "api_key_configured": bool(api_key),
        "environment_proxy_configured": any(os.environ.get(k) for k in (
            "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"
        )),
        "use_environment_proxy": False,
        "transport": "http.client",
        "request_connection": "keep-alive",
    }
    secrets = [api_key]
    try:
        parts = urlsplit(url)
        params = parse_qsl(parts.query, keep_blank_values=True)
        secrets.extend(value for key, value in params if key.lower() == "proxytoken")
        if (parts.scheme not in ("http", "https") or not parts.hostname
                or parts.username or parts.password or parts.fragment
                or any(char.isspace() for char in url)):
            raise ValueError
        _ = parts.port
        if parts.path.rstrip("/").endswith("/chat/completions"):
            parts = parts._replace(path=parts.path.rstrip("/"))
            url = urlunsplit(parts)
    except ValueError:
        return {**result, "code": "invalid_base_url", "hint": "DEEPSEEK_BASE_URL须为单行完整HTTP(S)代理URL"}
    result.update(endpoint=f"{parts.scheme}://{parts.netloc}{parts.path}",
                  query_keys=sorted({key for key, _ in params}),
                  model=_redact(model, secrets), method=method,
                  chat_payload_sent=method == "POST")
    query = dict(params)
    if query.get("servertype") != "deepseek" or not query.get("proxyToken"):
        return {**result, "code": "missing_proxy_params", "hint": "URL须包含servertype=deepseek及非空proxyToken，注意参数大小写"}

    payload = {
        "model": model,
        "messages": [{"role": "user", "content": "1+1等于多少"}],
        "max_tokens": 128,
        "stream": False,
    }
    headers = {"Accept": "application/json", "Connection": "keep-alive"}
    if method == "POST":
        headers["Content-Type"] = "application/json"
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"
    # GET probes the documented gateway URL only. Do not guess how a custom
    # gateway encodes upstream chat requests in its query parameters.
    request_body = json.dumps(payload, ensure_ascii=False).encode("utf-8") if method == "POST" else None
    started = time.perf_counter()
    connection = None
    try:
        connection_type = HTTPSConnection if parts.scheme == "https" else HTTPConnection
        connection = connection_type(parts.hostname, parts.port, timeout=timeout)
        target = (parts.path or "/") + ("?" + parts.query if parts.query else "")
        # Direct connection: no environment proxy, redirects, or automatic retry.
        connection.request(method, target, body=request_body, headers=headers)
        response = connection.getresponse()
        with response:
            result["http_status"] = response.status
            result["http_ok"] = 200 <= response.status < 300
            result["content_type"] = _redact(response.headers.get("Content-Type", ""), secrets)
            result["transfer_encoding"] = _redact(response.headers.get("Transfer-Encoding", ""), secrets)[:120]
            length = response.headers.get("Content-Length", "")
            expected_length = int(length) if length.isascii() and length.isdecimal() else None
            result["content_length"] = expected_length
            result["response_complete"] = True
            try:
                raw = response.read(1024 * 1024 + 1)
            except IncompleteRead as exc:
                # Malformed/unfinished chunked responses can contain the useful
                # gateway error text in partial; do not discard that evidence.
                raw = exc.partial
                result.update(response_complete=False, error_type="IncompleteRead")
            if (expected_length is not None and not result["transfer_encoding"]
                    and len(raw) < expected_length):
                result.update(response_complete=False, error_type="ContentLengthMismatch")
            result["received_bytes"] = len(raw)
        if len(raw) > 1024 * 1024:
            result["code"] = "response_too_large"
            return result
        body = raw.decode("utf-8", errors="replace")
        result["response_preview"] = _redact(body, secrets)[:1200]
        if "authentication fails" in body.lower() and "governor" in body.lower():
            result.update(code="proxy_authentication_failed", hint="代理已响应但鉴权失败，请核对参数大小写、token及代理权限")
        elif not 200 <= result["http_status"] < 300:
            result["code"] = "http_error"
        elif not result["response_complete"]:
            result["code"] = "response_incomplete"
        else:
            try:
                data = json.loads(body)
                content = data["choices"][0]["message"]["content"]
                if not isinstance(content, str) or not content.strip():
                    raise ValueError
                result.update(ok=True, code="chat_success", assistant_text=_redact(content, secrets)[:1200])
            except (ValueError, KeyError, TypeError, IndexError):
                result.update(code="unexpected_response", hint="HTTP成功但不是含非空choices[0].message.content的聊天响应；需核对代理返回协议")
        if not result["response_complete"]:
            result["hint"] = (
                f"已收到HTTP {result['http_status']}响应头，但响应体传输不完整；"
                + ("未取得可解析的正文。" if not raw else "response_preview为已收到的脱敏片段。")
                + "HTTP 200不代表聊天成功，请代理同事检查重复Transfer-Encoding、分块编码及结束标记"
            )
    except (TimeoutError, socket.timeout):
        result["code"] = "timeout"
    except OSError as exc:
        result["code"] = "connection_failed"
        result["error_type"] = type(exc).__name__
    except Exception as exc:
        # Do not expose exceptions containing the complete URL or credentials.
        result.update(code="request_failed", error_type=type(exc).__name__)
    finally:
        if connection is not None:
            connection.close()
        result["elapsed_seconds"] = round(time.perf_counter() - started, 3)
    return result


def main() -> int:
    parser = argparse.ArgumentParser(description="读取.env，测试固定DeepSeek代理入口（默认POST，一次请求）")
    parser.add_argument("--timeout", type=float, default=30, help="请求超时秒数，默认30")
    parser.add_argument("--method", choices=("GET", "POST"), default="POST",
                        help="默认POST发送最小聊天JSON；GET仅请求代理入口")
    args = parser.parse_args()
    if not 0 < args.timeout <= 120:
        parser.error("--timeout须大于0且不超过120秒")
    result = probe(timeout=args.timeout, method=args.method)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result["ok"] else 1


if __name__ == "__main__":
    sys.exit(main())