I am America 美国白宫最新力作 政府聊天机器人

#!/usr/bin/env python3
"""america_proxy.py — america.gov 聊天机器人 → OpenAI 兼容反向代理（单文件，只依赖 curl_cffi）。

运行：python america_proxy.py [--once "问题"]
端点：POST /v1/chat/completions（stream 可选）、GET /v1/models、GET /health
实测两个坑：① Cloudflare 只看 TLS/HTTP2 指纹 —— curl_cffi 伪装 safari 零 cookie 即 200，
curl/httpx/urllib 一律 403，且不能强制 HTTP/1.1；② 上游校验历史签名（historyInvalid），
不接受自造的 assistant 消息，所以多轮一律压平成一条 prompt。
"""
import argparse, json, sys, time, uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

try:
    from curl_cffi import requests as cr
except ImportError:
    sys.exit("缺少依赖：pip install curl_cffi")

URL, MODEL = "https://america.gov/api/chat", "america-gov"
HDRS = {"content-type": "application/json", "accept": "*/*",
        "origin": "https://america.gov", "referer": "https://america.gov/"}
FPS = ["safari", "safari_17_2_1", "safari_17_0", "chrome", "edge101"]  # 仅 safari 实测通过
LANG, TIMEOUT, FP, SESS = "en", 120.0, None, {}
class Gone(Exception): pass  # 客户端断开哨兵（curl_cffi 的异常也是 OSError，不能直接 except OSError）

def err(m, c): e = Exception(m); e.code = c; raise e  # 带 HTTP 状态码的异常

def upstream(b):
    """指纹轮换，首次成功后固定；返回上游 SSE 行迭代器。"""
    global FP
    tried = []
    for fp in [FP] if FP else FPS:
        r = SESS.setdefault(fp, cr.Session(impersonate=fp)).post(
            URL, json=b, headers=HDRS, stream=True, timeout=TIMEOUT, allow_redirects=False)
        if r.status_code == 200 and "text/event-stream" in (r.headers.get("content-type") or ""):
            FP = fp
            return r.iter_lines()
        tried.append(f"{fp}->{r.status_code}")
        if r.status_code != 403 or r.headers.get("x-america-chat-error"):
            err(f"上游 {r.status_code} ({r.headers.get('x-america-chat-error')}): {r.text[:200]}", r.status_code)
    err("Cloudflare 拦截，已试过 " + ", ".join(tried) + "；可能出口 IP 被限流", 502)

def parse(lines, cb=None):
    """UIMessageStream：正文取 text-delta，官方来源取 data-search，其余事件内容已加密无用。"""
    txt, src = [], {}
    for raw in lines:
        line = raw.decode("utf-8", "replace") if isinstance(raw, bytes) else raw
        if line[:5] != "data:": continue
        try:
            ev = json.loads(line[5:])
        except Exception:
            continue
        t = ev.get("type") if isinstance(ev, dict) else None
        if t == "text-delta":
            d = ev.get("delta") or ""
            txt.append(d)
            if cb: cb(d)
        elif t == "data-search":
            for g in (ev.get("data") or {}).get("groundings") or []:
                if g.get("url"): src[g["url"]] = g
    return "".join(txt), list(src.values())

def body(ms, lang):
    """OpenAI messages → 上游单条 user 消息（多轮压平成 transcript）。"""
    def t(m):
        c = m.get("content")
        return c if isinstance(c, str) else "".join((p or {}).get("text", "") for p in c or [])
    ts = [("User" if m["role"] == "user" else "Assistant") + ": " + t(m)
          for m in ms if m.get("role") in ("user", "assistant") and t(m)]
    if not ts or not ts[-1].startswith("User: "): err("最后一条必须是用户提问", 400)
    text = ts[-1][6:] if len(ts) == 1 else "Answer using this transcript.\n" + "\n".join(ts)
    return {"messages": [{"id": uuid.uuid4().hex[:16], "role": "user",
                          "parts": [{"type": "text", "text": text}]}],
            "trigger": "submit", "language": lang}

class H(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(s, *a): pass

    def out(s, c, o):
        b = json.dumps(o, ensure_ascii=False).encode()
        s.send_response(c)
        s.send_header("content-type", "application/json; charset=utf-8")
        s.send_header("content-length", str(len(b)))
        s.end_headers()
        s.wfile.write(b)

    def do_GET(s):
        if s.path == "/v1/models":
            return s.out(200, {"object": "list", "data": [{"id": MODEL, "object": "model"}]})
        return s.out(200, {"status": "ok", "upstream": URL, "fingerprint": FP})

    def do_POST(s):
        try:
            p = json.loads(s.rfile.read(int(s.headers.get("content-length") or 0)) or b"{}")
        except Exception as e:
            return s.out(400, {"error": {"message": f"请求体无效: {e}"}})
        try:
            lines = upstream(body(p.get("messages") or [], LANG))
        except Exception as e:
            return s.out(getattr(e, "code", 502), {"error": {"message": str(e)}})
        mid, ts, mdl = "chatcmpl-" + uuid.uuid4().hex[:20], int(time.time()), p.get("model") or MODEL
        if p.get("stream") is not True:
            txt, src = parse(lines)
            return s.out(200, {"id": mid, "object": "chat.completion", "created": ts, "model": mdl,
                               "choices": [{"index": 0, "message": {"role": "assistant", "content": txt},
                                            "finish_reason": "stop"}],
                               "usage": dict(prompt_tokens=0, completion_tokens=0, total_tokens=0),
                               "x_sources": src})
        s.send_response(200)
        s.send_header("content-type", "text/event-stream; charset=utf-8")
        s.send_header("cache-control", "no-store")
        s.send_header("connection", "close")
        s.end_headers()

        def w(o):
            try:
                s.wfile.write(f"data: {json.dumps(o, ensure_ascii=False)}\n\n".encode())
                s.wfile.flush()
            except OSError:
                raise Gone

        def pk(d, fr=None):
            w({"id": mid, "object": "chat.completion.chunk", "created": ts, "model": mdl,
               "choices": [{"index": 0, "delta": d, "finish_reason": fr}]})
        try:
            pk({"role": "assistant"})
            _, src = parse(lines, lambda d: pk({"content": d}))
            pk({}, "stop")
            if src: w({"id": mid, "created": ts, "x_sources": src})
            s.wfile.write(b"data: [DONE]\n\n")
        except Gone:
            pass

def main():
    ap = argparse.ArgumentParser(description="america.gov 聊天机器人反向代理（OpenAI 兼容）")
    for flag, tp, dflt, hlp in [("--host", str, "127.0.0.1", "监听地址"), ("--port", int, 8787, "监听端口"),
                                ("--language", str, "en", "en | es | fr"), ("--timeout", float, 120.0, "上游超时秒"),
                                ("--once", str, None, "只问一次，不启动服务")]:
        ap.add_argument(flag, type=tp, default=dflt, help=hlp)
    a = ap.parse_args()
    global LANG, TIMEOUT
    LANG, TIMEOUT = a.language, a.timeout
    if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8")  # Windows 控制台
    if a.once:
        try:
            _, src = parse(upstream(body([{"role": "user", "content": a.once}], LANG)),
                           lambda d: (sys.stdout.write(d), sys.stdout.flush()))
        except Exception as e:
            return print(f"错误：{e}", file=sys.stderr) or 1
        print()
        for g in src[:10]: print(f"  - {g.get('title')}\n    {g.get('url')}", file=sys.stderr)
        return 0
    srv = ThreadingHTTPServer((a.host, a.port), H)
    print(f'america-proxy: http://{a.host}:{a.port}  model="{MODEL}"', file=sys.stderr)
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        return 0

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