fvps/loop

1import json
2import secrets
3from fastapi import FastAPI, HTTPException, status
4from fastapi.responses import JSONResponse, StreamingResponse
5from starlette.concurrency import run_in_threadpool
6import curl_cffi.requests as curl
7from curl_cffi import CurlOpt
8from schemas import ProxyRequest, BROWSER_TYPES, HTTPMethod
9
10
11def pick_impersonate(user_choice: str | None) -> str:
12    """校验或随机挑一个浏览器指纹"""
13    if user_choice:
14        if user_choice not in BROWSER_TYPES:
15            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
16                                detail=f"Unsupported impersonate={user_choice}")
17        return user_choice
18
19    return secrets.choice(BROWSER_TYPES)
20
21
22def build_curl_opts(timeout_ms: int):
23    """常用超时：总超时 + 连接超时（≤5s 或总超时的较小值）"""
24    return {
25        CurlOpt.TIMEOUT_MS: timeout_ms,
26        CurlOpt.CONNECTTIMEOUT_MS: min(timeout_ms, 5_000),
27    }
28
29
30async def perform_final_hop(req: ProxyRequest, impersonate: str) -> curl.Response:
31    """
32    真正把请求发到目标站点的最后一跳。
33    依旧放在线程池里执行，确保 uvloop 事件循环无阻塞
34    """
35    return await run_in_threadpool(
36        curl.request,
37        method=req.method.value,
38        url=str(req.url),
39        params=req.params,
40        headers=req.headers,
41        cookies=req.cookies,
42        json=req.data if req.method in {"POST", "PUT", "PATCH"} else None,
43        data=req.data if req.method not in {"POST", "PUT", "PATCH"} else None,
44        stream=req.stream,
45        impersonate=impersonate,
46        proxies=req.proxies,
47        curl_options=build_curl_opts(req.timeout_ms),
48    )
49
50
51async def forward_to_next_hop(
52        next_url: str, nested_req: ProxyRequest, impersonate: str
53) -> curl.Response:
54    """
55    把当前 JSON 负载 POST 给链路中的下一台代理服务。
56    使用 curl_cffi 直接 POST；对方再继续递归或最终落地。
57    """
58    return await run_in_threadpool(
59        curl.request,
60        method=HTTPMethod.POST,
61        url=next_url,
62        headers={"Content-Type": "application/json"},
63        json=json.loads(nested_req.json()),
64        stream=nested_req.stream,
65        impersonate=impersonate,
66        curl_options=build_curl_opts(nested_req.timeout_ms),
67    )
68
69
70def render_response(resp: curl.Response, impersonate: str, include_body: bool, stream: bool):
71    """统一把 curl.Response 转成 FastAPI Response"""
72    meta = {
73        "status_code": resp.status_code,
74        "url": resp.url,
75        "elapsed": resp.elapsed,
76        "headers": dict(resp.headers),
77        "cookies": resp.cookies.get_dict(),
78        "impersonate": impersonate,
79    }
80
81    if not include_body:
82        return JSONResponse(content=meta, status_code=resp.status_code)
83
84    if stream:
85        return StreamingResponse(resp.iter_content(), status_code=resp.status_code, media_type="application/octet-stream", headers=meta["headers"])
86
87    ctype = resp.headers.get("Content-Type", "")
88    if ctype.startswith("application/json"):
89        meta["data"] = resp.json()
90    elif ctype.startswith("text/"):
91        meta["data"] = resp.text
92    else:
93        meta["data"] = resp.content.hex()
94
95    return JSONResponse(content=meta, status_code=resp.status_code)
96