File size: 3,376 Bytes
2ee3bb2 2280bd6 cb2432b b1049e3 2ee3bb2 3b27764 2ee3bb2 cb2432b 2ee3bb2 cb2432b 2ee3bb2 cb2432b 2ee3bb2 cb2432b 3b27764 2ee3bb2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import httpx
import asyncio
from typing import Optional
app = FastAPI()
# 添加 CORS 支持
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 目标网站
TARGET_URL = "http://beibeioo.top"
# 创建一个全局的 httpx 客户端
async def get_http_client():
return httpx.AsyncClient(
timeout=30.0, # 30秒超时
follow_redirects=True,
http2=True
)
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "HEAD", "PATCH"])
async def proxy(path: str, request: Request):
try:
# 构建目标URL
url = f"{TARGET_URL}/{path}"
# 获取并处理请求头
headers = dict(request.headers)
# 删除可能导致问题的头部
headers.pop('host', None)
headers.pop('connection', None)
headers.pop('content-length', None)
headers.pop('transfer-encoding', None)
# 如果有 referer，修改它
if 'referer' in headers:
headers['referer'] = headers['referer'].replace(str(request.base_url), TARGET_URL)
# 获取查询参数
params = dict(request.query_params)
# 获取请求体
body = await request.body()
async with await get_http_client() as client:
response = await client.request(
method=request.method,
url=url,
params=params,
headers=headers,
content=body if body else None,
)
# 处理响应头
response_headers = dict(response.headers)
# 删除可能导致问题的响应头
response_headers.pop('transfer-encoding', None)
response_headers.pop('content-encoding', None)
# 修改所有涉及原始域名的响应头
for key, value in response_headers.items():
if isinstance(value, str) and TARGET_URL in value:
response_headers[key] = value.replace(TARGET_URL, str(request.base_url).rstrip('/'))
# 返回响应
return StreamingResponse(
response.iter_bytes(),
status_code=response.status_code,
headers=response_headers,
media_type=response.headers.get('content-type')
)
except httpx.TimeoutException:
return Response(
content="请求超时",
status_code=504,
media_type="text/plain"
)
except httpx.RequestError:
return Response(
content="无法连接到目标服务器",
status_code=502,
media_type="text/plain"
)
except Exception as e:
return Response(
content=f"服务器错误: {str(e)}",
status_code=500,
media_type="text/plain"
)
@app.get("/healthcheck")
async def healthcheck():
"""健康检查端点"""
return {"status": "healthy"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=7860) |