import os import re import sys server_code = open('server.py', encoding='utf-8').read() # 1. Update imports in server.py imports = """import hashlib import hmac import secrets from collections import OrderedDict import httpx from fastapi.responses import JSONResponse from fastapi import Request """ lines = server_code.split('\n') lines.insert(8, imports) server_code = '\n'.join(lines) # 2. Add config config = """ # --- Gateway Configuration & Caching --- SUPABASE_URL = os.getenv("SUPABASE_URL", "") SUPABASE_SERVICE_KEY = os.getenv("SUPABASE_SERVICE_KEY", "") LEMON_SQUEEZY_WEBHOOK_SECRET = os.getenv("LEMON_SQUEEZY_WEBHOOK_SECRET", "") PHISHVISION_BACKEND = os.getenv("PHISHVISION_BACKEND_URL", "https://opticparse-1opticparse-node-sg.onrender.com") _http_client = None async def get_http_client() -> httpx.AsyncClient: global _http_client if _http_client is None or _http_client.is_closed: _http_client = httpx.AsyncClient(timeout=30.0) return _http_client def supabase_headers() -> dict: return { "apikey": SUPABASE_SERVICE_KEY, "Authorization": f"Bearer {SUPABASE_SERVICE_KEY}", "Content-Type": "application/json", "Prefer": "return=representation", } async def supabase_query(method: str, table: str, params: str = "", body: dict = None) -> list: client = await get_http_client() url = f"{SUPABASE_URL}/rest/v1/{table}?{params}" resp = await client.request(method, url, headers=supabase_headers(), json=body) if resp.status_code >= 400: logger.error(f"Supabase {method} {table} failed: {resp.status_code} {resp.text}") raise HTTPException(status_code=502, detail="Database operation failed") try: return resp.json() if resp.text else [] except Exception: return [] def hash_key(raw_key: str) -> str: return hashlib.sha256(raw_key.encode()).hexdigest() def generate_api_key() -> tuple[str, str, str]: token = secrets.token_hex(24) raw_key = f"op_live_{token}" return raw_key, hash_key(raw_key), f"op_live_{token[:8]}" class LRUCache: def __init__(self, max_size=500, ttl=300): self._cache = OrderedDict() self._max_size = max_size self._ttl = ttl def get(self, key_hash): entry = self._cache.get(key_hash) if not entry: return None if time.time() - entry["ts"] > self._ttl: del self._cache[key_hash] return None self._cache.move_to_end(key_hash) return entry["data"] def set(self, key_hash, data): if key_hash in self._cache: self._cache.move_to_end(key_hash) self._cache[key_hash] = {"data": data, "ts": time.time()} if len(self._cache) > self._max_size: self._cache.popitem(last=False) def invalidate(self, key_hash): self._cache.pop(key_hash, None) key_cache = LRUCache() async def log_usage(user_context: dict, endpoint: str, service: str, status_code: int, response_time_ms: int): if user_context.get("user_id") in ("rapidapi", "dev"): return try: await supabase_query( "PATCH", "users", f"id=eq.{user_context['user_id']}", body={"current_usage": user_context["current_usage"] + 1}, ) await supabase_query("POST", "usage_logs", body={ "user_id": user_context["user_id"], "api_key_id": user_context["api_key_id"], "endpoint": endpoint, "service": service, "status_code": status_code, "response_time_ms": response_time_ms, }) except Exception as e: logger.warning(f"Failed to log usage: {e}") """ server_code = server_code.replace("app = FastAPI(", config + "\napp = FastAPI(") # 3. Rewrite get_api_key new_get_api_key = """ async def get_api_key( request: Request, api_key: str = Depends(api_key_header), # RapidAPI integration - reserved for future use x_rapidapi_key: str = Header(None, alias="X-RapidAPI-Key"), x_rapidapi_proxy_secret: str = Header(None, alias="X-RapidAPI-Proxy-Secret"), ): start_time = time.time() # RapidAPI integration - reserved for future use proxy_secret = os.environ.get("RAPIDAPI_PROXY_SECRET") # RapidAPI integration - reserved for future use rapidapi_key = os.environ.get("RAPIDAPI_KEY") if (proxy_secret and (x_rapidapi_proxy_secret == proxy_secret or x_rapidapi_key == proxy_secret)) or \\ (rapidapi_key and x_rapidapi_key == rapidapi_key): return {"user_id": "rapidapi", "tier": "enterprise"} if not api_key: if not proxy_secret and not rapidapi_key and not SUPABASE_URL: return {"user_id": "dev", "tier": "enterprise"} raise HTTPException(status_code=401, detail="Missing API Key") kh = hash_key(api_key) cached = key_cache.get(kh) if cached: context = cached else: rows = await supabase_query( "GET", "api_keys", f"key_hash=eq.{kh}&is_active=eq.true&select=id,user_id,users(id,email,tier,monthly_limit,current_usage)" ) if not rows: raise HTTPException(status_code=401, detail="Invalid API Key") row = rows[0] user = row.get("users", {}) context = { "user_id": user.get("id"), "email": user.get("email"), "api_key_id": row["id"], "tier": user.get("tier", "free"), "monthly_limit": user.get("monthly_limit", 100), "current_usage": user.get("current_usage", 0), } key_cache.set(kh, context) if context["current_usage"] >= context["monthly_limit"]: raise HTTPException( status_code=429, detail=f"Monthly quota exceeded ({context['current_usage']}/{context['monthly_limit']}). Upgrade your plan." ) request.state.user_ctx = context asyncio.create_task(log_usage(context, request.url.path, "opticparse", 200, 50)) return context """ import re old_get_api_key_pattern = r'async def get_api_key\(.*?\n return "dev-mode"\n.*?\n \)' server_code = re.sub(old_get_api_key_pattern, new_get_api_key.strip(), server_code, flags=re.DOTALL) # 4. Add Gateway Endpoints gateway_endpoints = """ class KeyGenerateRequest(BaseModel): user_id: str @app.post("/gateway/keys/generate") async def generate_key(req: KeyGenerateRequest): raw_key, kh, prefix = generate_api_key() await supabase_query("POST", "api_keys", body={ "user_id": req.user_id, "key_hash": kh, "key_prefix": prefix, "is_active": True, }) return {"api_key": raw_key, "prefix": prefix} @app.post("/gateway/keys/regenerate") async def regenerate_key(req: KeyGenerateRequest): await supabase_query("PATCH", "api_keys", f"user_id=eq.{req.user_id}", body={"is_active": False}) raw_key, kh, prefix = generate_api_key() await supabase_query("POST", "api_keys", body={ "user_id": req.user_id, "key_hash": kh, "key_prefix": prefix, "is_active": True, }) return {"api_key": raw_key, "prefix": prefix} @app.get("/gateway/usage/{user_id}") async def get_usage(user_id: str): rows = await supabase_query("GET", "users", f"id=eq.{user_id}&select=tier,monthly_limit,current_usage") if not rows: raise HTTPException(status_code=404, detail="User not found") return rows[0] def verify_lemon_signature(payload: bytes, signature: str) -> bool: if not LEMON_SQUEEZY_WEBHOOK_SECRET: return True expected = hmac.new(LEMON_SQUEEZY_WEBHOOK_SECRET.encode(), payload, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature) @app.post("/gateway/webhooks/lemonsqueezy") async def lemon_squeezy_webhook(request: Request): body = await request.body() signature = request.headers.get("X-Signature", "") if not verify_lemon_signature(body, signature): raise HTTPException(status_code=403, detail="Invalid signature") data = json.loads(body) event_name = data.get("meta", {}).get("event_name", "") user_id = data.get("meta", {}).get("custom_data", {}).get("user_id") if not user_id: return JSONResponse({"status": "ignored"}) if event_name in ("subscription_created", "subscription_payment_success", "subscription_resumed"): await supabase_query("PATCH", "users", f"id=eq.{user_id}", body={ "tier": "pro", "monthly_limit": 5000, "lemon_customer_id": str(data.get("data", {}).get("id", "")) }) elif event_name in ("subscription_cancelled", "subscription_expired", "subscription_paused"): await supabase_query("PATCH", "users", f"id=eq.{user_id}", body={"tier": "free", "monthly_limit": 100}) return JSONResponse({"status": "ok"}) @app.post("/api/vision-parse") async def vision_parse(request: Request, user_ctx: dict = Depends(get_api_key)): start_time = time.time() body = await request.json() hf_key = os.getenv("HUGGINGFACE_API_KEY") client = await get_http_client() resp = await client.post( "https://api-inference.huggingface.co/models/Qwen/Qwen2-VL-7B-Instruct", headers={"Authorization": f"Bearer {hf_key}", "Content-Type": "application/json"}, json={"inputs": body.get("prompt", ""), "image": body.get("image", "")}, timeout=30.0 ) if resp.status_code == 429: raise HTTPException(status_code=429, detail="Rate limit") resp.raise_for_status() asyncio.create_task(log_usage(user_ctx, "/api/vision-parse", "huggingface", 200, int((time.time() - start_time) * 1000))) return JSONResponse(content=resp.json()) # Proxy for PhishVision @app.api_route("/api/phish{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"]) async def proxy_phish(request: Request, path: str, user_ctx: dict = Depends(get_api_key)): start_time = time.time() body = await request.body() client = await get_http_client() resp = await client.request( method=request.method, url=f"{PHISHVISION_BACKEND}/api/phish{path}", headers={"Content-Type": request.headers.get("content-type", "application/json")}, content=body, params=dict(request.query_params) ) asyncio.create_task(log_usage(user_ctx, f"/api/phish{path}", "phishvision", resp.status_code, int((time.time() - start_time) * 1000))) return Response(content=resp.content, status_code=resp.status_code, media_type=resp.headers.get("content-type", "application/json")) @app.api_route("/api/monitor{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"]) async def proxy_monitor(request: Request, path: str, user_ctx: dict = Depends(get_api_key)): start_time = time.time() body = await request.body() client = await get_http_client() resp = await client.request( method=request.method, url=f"{PHISHVISION_BACKEND}/api/monitor{path}", headers={"Content-Type": request.headers.get("content-type", "application/json")}, content=body, params=dict(request.query_params) ) asyncio.create_task(log_usage(user_ctx, f"/api/monitor{path}", "phishvision", resp.status_code, int((time.time() - start_time) * 1000))) return Response(content=resp.content, status_code=resp.status_code, media_type=resp.headers.get("content-type", "application/json")) """ server_code = server_code.replace('if __name__ == "__main__":', gateway_endpoints + '\\nif __name__ == "__main__":') open('server.py', 'w', encoding='utf-8').write(server_code) print("Merge complete")