#!/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())
路边一条狗 09-30 02:461楼
syxing0270 09-30 04:202楼