TypeGPTProxy / main.py
AIMaster7's picture
Update main.py
9ebd969 verified
raw
history blame
2.6 kB
from fastapi import FastAPI, Request, Header, HTTPException
from fastapi.responses import JSONResponse, Response
import httpx
import socket
app = FastAPI()
# Your internal real API key (DO NOT expose this to users)
REAL_API_KEY = "sk-qO9N6kQEEULMWtF4YGVlTTSjIPllEm1h1wfEBzSmnSbxiXwe"
BASE_URL = "https://fast.typegpt.net"
# Public token that users must use
PUBLIC_AUTH_TOKEN = "TypeGPT-Free4ALL"
# Print server IP address on startup
@app.on_event("startup")
async def startup_event():
try:
hostname = socket.gethostname()
server_ip = socket.gethostbyname(hostname)
print(f"===== Server Started =====")
print(f"📡 Server IP: {server_ip}")
except Exception as e:
print(f"⚠️ Could not get server IP: {e}")
@app.api_route("/{path:path}", methods=["GET", "POST"])
async def proxy(request: Request, path: str, authorization: str = Header(None)):
# Validate Authorization token
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or malformed Authorization header.")
token = authorization.replace("Bearer ", "").strip()
if token != PUBLIC_AUTH_TOKEN:
raise HTTPException(status_code=401, detail="Invalid Authorization token. Use 'TypeGPT-Free4ALL'.")
# Prepare the backend target URL
target_url = f"{BASE_URL}/{path}"
# Clone headers, replace Authorization
headers = dict(request.headers)
headers["Authorization"] = f"Bearer {REAL_API_KEY}"
headers.pop("host", None)
# Read request body
body = await request.body()
# Forward request to backend
async with httpx.AsyncClient() as client:
try:
response = await client.request(
method=request.method,
url=target_url,
content=body,
headers=headers,
timeout=60
)
except httpx.RequestError as e:
raise HTTPException(status_code=502, detail=f"Request to backend failed: {e}")
# Debug log
print(f"🔄 Forwarded to: {target_url}")
print(f"↩️ TypeGPT Status: {response.status_code}")
print(f"🧾 TypeGPT Response (first 200 chars): {response.text[:200]}")
# Try to return JSON or fallback
try:
return JSONResponse(content=response.json(), status_code=response.status_code)
except Exception:
return Response(
content=response.content,
status_code=response.status_code,
media_type=response.headers.get("content-type", "text/plain")
)