Spaces:
Running
Running
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 | |
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}") | |
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") | |
) | |