tabs/http_proxy.py

#!/usr/bin/python3
# -*- coding: utf-8 -*-
"""
HTTP 代理 Tab

提供一个通用的 HTTP 请求代理，既支持在 Gradio 页面上手动调试，
也可以通过 Gradio 的 API 从外部代码调用，例如:

    # 通过 gradio_client 调用
    from gradio_client import Client
    client = Client("http://127.0.0.1:7860/")
    status, headers, body = client.predict(
        url="https://httpbin.org/get",
        method="GET",
        headers="{}",
        params="{}",
        body="",
        timeout=30.0,
        follow_redirects=True,
        verify_ssl=True,
        api_name="/http_proxy",
    )

    # 通过 HTTP 调用 gradio 内置 API
    POST http://127.0.0.1:7860/gradio_api/call/http_proxy
    body: {"data": ["https://httpbin.org/get", "GET", "{}", "{}", "", 30.0, true, true]}

cookies 用法:
    没有单独的 cookies 字段, 直接在 headers (json) 里加一个 `Cookie` 头即可,
    多个 cookie 用 `;` 分隔, 例如:

        {
            "Cookie": "session_id=abc123; user_token=xyz789"
        }
"""
import json
import logging
from typing import Any, Optional

import gradio as gr
import httpx


logger = logging.getLogger("http_proxy")


SUPPORTED_METHODS = ["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS"]

MAX_BODY_PREVIEW = 2 * 1024 * 1024  # 2 MB


def _parse_json_field(value: Optional[str], field_name: str) -> Optional[Any]:
    if value is None:
        return None
    text = value.strip()
    if not text:
        return None
    try:
        return json.loads(text)
    except json.JSONDecodeError as e:
        raise ValueError(f"`{field_name}` JSON 解析失败: {e}") from e


def http_proxy_request(
    url: str,
    method: str = "GET",
    headers: Optional[str] = None,
    params: Optional[str] = None,
    body: Optional[str] = None,
    timeout: float = 30.0,
    follow_redirects: bool = True,
    verify_ssl: bool = True,
):
    """转发 HTTP 请求并返回 (status_code, response_headers_json, response_body)."""
    if not url or not url.strip():
        return 0, "{}", "url 不能为空"

    method_upper = (method or "GET").upper()
    if method_upper not in SUPPORTED_METHODS:
        return 0, "{}", f"不支持的 method: {method}"

    try:
        headers_dict = _parse_json_field(headers, "headers")
        params_dict = _parse_json_field(params, "params")
    except ValueError as e:
        return 0, "{}", str(e)

    request_kwargs: dict = {
        "method": method_upper,
        "url": url.strip(),
        "headers": headers_dict,
        "params": params_dict,
        "timeout": float(timeout) if timeout else 30.0,
    }

    if body and body.strip() and method_upper not in ("GET", "HEAD"):
        body_text = body
        try:
            request_kwargs["json"] = json.loads(body_text)
        except (json.JSONDecodeError, TypeError):
            request_kwargs["content"] = body_text.encode("utf-8")

    logger.info(f"proxy request: {method_upper} {url}")

    try:
        with httpx.Client(
            follow_redirects=bool(follow_redirects),
            verify=bool(verify_ssl),
            http2=True,
        ) as client:
            response = client.request(**request_kwargs)
    except httpx.HTTPError as e:
        logger.exception("http error")
        return 0, "{}", f"http 请求异常: {e!r}"
    except Exception as e:
        logger.exception("unexpected error")
        return 0, "{}", f"未知异常: {e!r}"

    response_headers = json.dumps(
        dict(response.headers), ensure_ascii=False, indent=2,
    )

    content_bytes = response.content or b""
    if len(content_bytes) > MAX_BODY_PREVIEW:
        truncated_note = (
            f"\n\n... [body 已截断, 原始长度 {len(content_bytes)} bytes, "
            f"仅展示前 {MAX_BODY_PREVIEW} bytes]"
        )
        content_bytes = content_bytes[:MAX_BODY_PREVIEW]
    else:
        truncated_note = ""

    try:
        encoding = response.encoding or "utf-8"
        response_body = content_bytes.decode(encoding, errors="replace")
    except Exception:
        response_body = repr(content_bytes)

    return response.status_code, response_headers, response_body + truncated_note


def get_http_proxy_tab():
    method_choices = SUPPORTED_METHODS

    with gr.TabItem("http_proxy"):
        gr.Markdown(
            "## HTTP 代理\n"
            "在页面上手动调试请求，也可以从外部通过 Gradio API 调用，"
            "把本服务当成统一的 HTTP 代理网关。\n\n"
            "- gradio_client: `client.predict(..., api_name=\"/http_proxy\")`\n"
            "- HTTP: `POST /gradio_api/call/http_proxy`，参数顺序见输入框"
        )
        with gr.Row():
            with gr.Column(scale=3):
                http_url = gr.Textbox(
                    label="url",
                    placeholder="https://example.com/api/xxx",
                    lines=1,
                )
                http_method = gr.Dropdown(
                    label="method",
                    choices=method_choices,
                    value=method_choices[0],
                )
                http_headers = gr.Textbox(
                    label="headers (json)",
                    value="{}",
                    lines=4,
                    max_lines=20,
                )
                http_params = gr.Textbox(
                    label="query params (json)",
                    value="{}",
                    lines=4,
                    max_lines=20,
                )
                http_body = gr.Textbox(
                    label="body (raw text 或 json)",
                    value="",
                    lines=6,
                    max_lines=30,
                )
                with gr.Row():
                    http_timeout = gr.Number(label="timeout", value=30.0)
                    http_follow_redirects = gr.Checkbox(
                        label="follow_redirects", value=True,
                    )
                    http_verify_ssl = gr.Checkbox(
                        label="verify_ssl", value=True,
                    )
                http_send = gr.Button("send", variant="primary")
            with gr.Column(scale=4):
                http_status_code = gr.Number(label="status_code", interactive=False)
                http_response_headers = gr.Textbox(
                    label="response headers",
                    lines=8,
                    max_lines=40,
                    interactive=False,
                    show_copy_button=True,
                )
                http_response_body = gr.Textbox(
                    label="response body",
                    lines=20,
                    max_lines=200,
                    interactive=False,
                    show_copy_button=True,
                )

        inputs = [
            http_url,
            http_method,
            http_headers,
            http_params,
            http_body,
            http_timeout,
            http_follow_redirects,
            http_verify_ssl,
        ]
        outputs = [
            http_status_code,
            http_response_headers,
            http_response_body,
        ]

        http_send.click(
            fn=http_proxy_request,
            inputs=inputs,
            outputs=outputs,
            api_name="http_proxy",
        )

        gr.Examples(
            examples=[
                [
                    "https://httpbin.org/get",
                    "GET",
                    "{}",
                    "{\"q\": \"hello\"}",
                    "",
                    30.0,
                    True,
                    True,
                ],
                [
                    "https://httpbin.org/post",
                    "POST",
                    "{\"Content-Type\": \"application/json\"}",
                    "{}",
                    "{\"hello\": \"world\"}",
                    30.0,
                    True,
                    True,
                ],
                [
                    "https://api.github.com/repos/gradio-app/gradio",
                    "GET",
                    "{\"User-Agent\": \"gradio-http-proxy\"}",
                    "{}",
                    "",
                    30.0,
                    True,
                    True,
                ],
                [
                    "https://httpbin.org/cookies",
                    "GET",
                    "{\"Cookie\": \"session_id=abc123; user_token=xyz789\"}",
                    "{}",
                    "",
                    30.0,
                    True,
                    True,
                ],
            ],
            inputs=inputs,
        )

    return locals()


if __name__ == "__main__":
    with gr.Blocks() as block:
        _ = get_http_proxy_tab()
    block.queue().launch()