import asyncio import base64 import hashlib import json import os os.environ["PLAYWRIGHT_BROWSERS_PATH"] = "0" import sys import time from dotenv import load_dotenv import hashlib import hmac import secrets from collections import OrderedDict import httpx from fastapi.responses import JSONResponse from fastapi import Request if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) from fastapi import FastAPI, Depends, HTTPException, Header, status, BackgroundTasks from fastapi.responses import Response from fastapi.security import APIKeyHeader from fastapi.middleware.cors import CORSMiddleware # Load environment variables from a .env file if present load_dotenv() import logging from pydantic import BaseModel, validator from typing import Optional from datetime import datetime import uuid as uuid_module _async_jobs = {} from playwright.async_api import async_playwright from openai import OpenAI, AsyncOpenAI # Setup standard logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s" ) logger = logging.getLogger("vision-scrape") # --------------------------------------------------------------------------- # AI Provider Rotation — Groq → Gemini → GitHub Models → OpenRouter → DeepSeek # --------------------------------------------------------------------------- AI_PROVIDERS = [ { "name": "Groq", "key": os.getenv("GROQ_API_KEY"), "base": "https://api.groq.com/openai/v1", "model": "llama-3.2-11b-vision-preview", "vision": True, }, { "name": "Gemini", "key": os.getenv("GEMINI_API_KEY"), "base": "https://generativelanguage.googleapis.com/v1beta/openai/", "model": "gemini-1.5-flash", "vision": True, }, { "name": "GitHub Models", "key": os.getenv("GITHUB_TOKEN"), "base": "https://models.inference.ai.azure.com", "model": "gpt-4o", "vision": True, }, { "name": "OpenRouter", "key": os.getenv("OPENROUTER_KEY") or os.getenv("FREE_AI_KEY"), "base": "https://openrouter.ai/api/v1", "model": "openai/gpt-4o-mini", "vision": True, }, { "name": "DeepSeek", "key": os.getenv("DEEPSEEK_API_KEY"), "base": "https://api.deepseek.com/v1", "model": "deepseek-chat", "vision": False, # DeepSeek-chat uses text; falls back gracefully }, ] async def call_ai_with_rotation(screenshot_bytes: bytes, query: str, response_schema: dict = None) -> str: """Tries AI providers in order. Returns extracted JSON string.""" import base64 img_b64 = base64.b64encode(screenshot_bytes).decode() system_prompt = ( "You are an automated JSON data extractor. " "Analyze the provided image and extract information based on the user's query. " "Return ONLY valid raw JSON. No markdown, no extra text." ) if response_schema: system_prompt += ( f"\n\nCRITICAL: Your output MUST strictly match and validate against this JSON Schema:\n" f"{json.dumps(response_schema)}\n" f"Ensure all keys and types match the schema definitions exactly." ) errors = [] for provider in AI_PROVIDERS: if not provider["key"]: errors.append(f"{provider['name']}: no API key") continue try: client = AsyncOpenAI(api_key=provider["key"], base_url=provider["base"]) content_parts = [ {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{img_b64}", "detail": "high"}}, {"type": "text", "text": query}, ] if provider["vision"] else [{"type": "text", "text": f"Analyze this page for: {query}"}] extra_args = {} # Use JSON mode if supported and schema is provided if provider["name"] in ("GitHub Models", "OpenRouter") or response_schema: extra_args["response_format"] = {"type": "json_object"} response = await client.chat.completions.create( model=provider["model"], messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": content_parts}, ], max_tokens=2048, temperature=0, **extra_args ) result = response.choices[0].message.content or "{}" logger.info(f"AI extraction succeeded via {provider['name']}") return result except Exception as e: logger.warning(f"Provider {provider['name']} failed: {e}") errors.append(f"{provider['name']}: {e}") raise HTTPException(status_code=503, detail=f"All AI providers failed: {'; '.join(errors)}") # --------------------------------------------------------------------------- # Simple 5-minute in-memory response cache # --------------------------------------------------------------------------- REDIS_URL = os.getenv("REDIS_URL") if REDIS_URL: import redis.asyncio as redis redis_client = redis.from_url(REDIS_URL, decode_responses=True) else: redis_client = None _cache: dict = {} CACHE_TTL = 86400 # 24 hours async def get_cache(url: str, query: str, response_schema: dict = None): schema_str = json.dumps(response_schema, sort_keys=True) if response_schema else "" key = hashlib.md5(f"{url}|{query}|{schema_str}".encode()).hexdigest() if redis_client: try: res = await redis_client.get(key) if res: logger.info(f"Redis Cache HIT for {url}") return res return None except Exception as e: logger.warning(f"Redis get failed: {e}") return None entry = _cache.get(key) if entry and time.time() - entry["ts"] < CACHE_TTL: logger.info(f"Memory Cache HIT for {url}") return entry["data"] return None async def set_cache(url: str, query: str, data: str, response_schema: dict = None): schema_str = json.dumps(response_schema, sort_keys=True) if response_schema else "" key = hashlib.md5(f"{url}|{query}|{schema_str}".encode()).hexdigest() if redis_client: try: await redis_client.setex(key, CACHE_TTL, data) return except Exception as e: logger.warning(f"Redis set failed: {e}") _cache[key] = {"data": data, "ts": time.time()} # --- 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") IS_PRODUCTION = os.getenv("RENDER") == "true" if not IS_PRODUCTION and not SUPABASE_URL: logger.warning("Local dev bypass is ACTIVE. Unauthenticated requests will be granted enterprise access.") BROWSER_SEMAPHORE = asyncio.Semaphore(2) _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=90.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}") app = FastAPI( title="Vision-Scrape API", description="Extracts data from webpages using Playwright screenshotting and an AI Agent.", version="1.0.0" ) from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware( CORSMiddleware, allow_origins=[ 'https://opticparse.com', 'https://dashboard.opticparse.com', 'http://localhost:5173', ], allow_credentials=True, allow_methods=['GET', 'POST', 'DELETE', 'PUT'], allow_headers=['*'], ) # --------------------------------------------------------------------------- # Health Check — used by Render and automated verification agents # --------------------------------------------------------------------------- @app.get("/health") async def health_check(): """Returns service status for uptime monitoring and deploy verification.""" try: return { "status": "ok", "service": "opticparse", "version": "1.0.0", } except Exception as e: return JSONResponse(status_code=500, content={"status": "error", "detail": str(e)}) # --------------------------------------------------------------------------- # Database initialization (Supports SQLite or PostgreSQL via DATABASE_URL) # --------------------------------------------------------------------------- import sqlite3 import uuid DATABASE_URL = os.getenv("DATABASE_URL") _pg_pool = None def get_db_placeholder(): return "%s" if DATABASE_URL else "?" def get_pg_pool(): global _pg_pool if _pg_pool is None and DATABASE_URL: from psycopg2.pool import ThreadedConnectionPool import psycopg2 try: _pg_pool = ThreadedConnectionPool(1, 20, DATABASE_URL) except psycopg2.OperationalError as e: if "6543" in str(e) or "pooler" in DATABASE_URL: fallback_url = DATABASE_URL.replace(":6543", ":5432").replace(".pooler.", ".") import logging logging.warning("Pooler connection failed, falling back to direct port 5432") _pg_pool = ThreadedConnectionPool(1, 20, fallback_url) else: raise return _pg_pool def run_db_query(func, *args, **kwargs): """Synchronous helper to run queries safely with pooling.""" conn = None try: if DATABASE_URL: pool = get_pg_pool() conn = pool.getconn() else: conn = sqlite3.connect("opticparse.db") cursor = conn.cursor() res = func(cursor, *args, **kwargs) conn.commit() return res except Exception as e: if conn: conn.rollback() raise e finally: if conn: if DATABASE_URL: pool = get_pg_pool() pool.putconn(conn) else: conn.close() def init_db(): def _do_init(cursor): if DATABASE_URL: cursor.execute(""" CREATE TABLE IF NOT EXISTS watches ( id VARCHAR(255) PRIMARY KEY, user_id VARCHAR(255) NOT NULL, url TEXT NOT NULL, query TEXT NOT NULL, schema_text TEXT, last_result TEXT, created_at DOUBLE PRECISION NOT NULL ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS user_settings ( user_id VARCHAR(255) PRIMARY KEY, webhook_url TEXT ) """) else: cursor.execute(""" CREATE TABLE IF NOT EXISTS watches ( id TEXT PRIMARY KEY, user_id TEXT NOT NULL, url TEXT NOT NULL, query TEXT NOT NULL, schema_text TEXT, last_result TEXT, created_at REAL NOT NULL ) """) cursor.execute(""" CREATE TABLE IF NOT EXISTS user_settings ( user_id TEXT PRIMARY KEY, webhook_url TEXT ) """) try: run_db_query(_do_init) logger.info("Database initialized successfully") except Exception as e: logger.error(f"Failed to initialize database: {e}") try: init_db() logger.info("Database initialized successfully") except Exception as e: logger.warning( f"Database init failed — watch feature " f"unavailable until DB is fixed: {e}" ) class LoginInfo(BaseModel): login_url: str username_field: str password_field: str username: str password: str submit_button: str = None class ActionInfo(BaseModel): type: str selector: Optional[str] = None value: Optional[str] = None ms: Optional[int] = None key: Optional[str] = None class ScrapeRequest(BaseModel): target_url: str extraction_query: str viewport_width: int = 1280 viewport_height: int = 800 wait_until: str = "load" timeout: int = 30000 response_schema: dict = None login: LoginInfo = None actions: list[ActionInfo] = None webhook_url: Optional[str] = None @validator('target_url') def validate_url(cls, v): if not v.startswith(('http://', 'https://')): raise ValueError('URL must start with http:// or https://') blocked = ['localhost', '127.0.0.1', '0.0.0.0', '169.254.', '10.0.', '192.168.', '172.16.'] for b in blocked: if b in v: raise ValueError('Internal network URLs not allowed') if len(v) > 2048: raise ValueError('URL too long (max 2048 chars)') return v @validator('extraction_query') def validate_query(cls, v): if len(v) < 3: raise ValueError('Query too short (min 3 chars)') if len(v) > 1000: raise ValueError('Query too long (max 1000 chars)') return v @validator('timeout') def validate_timeout(cls, v): return max(5000, min(60000, v)) @validator('wait_until') def validate_wait_until(cls, v): allowed = ['load', 'domcontentloaded', 'networkidle'] return v if v in allowed else 'load' class DirectScrapeRequest(BaseModel): image_base64: str extraction_query: str response_schema: dict = None @validator('extraction_query') def validate_query(cls, v): if len(v) < 3: raise ValueError('Query too short (min 3 chars)') if len(v) > 1000: raise ValueError('Query too long (max 1000 chars)') return v class CrawlRequest(BaseModel): start_url: str extraction_query: str follow_selector: str max_pages: int = 5 viewport_width: int = 1280 viewport_height: int = 800 wait_until: str = "load" timeout: int = 30000 response_schema: dict = None class WatchRequest(BaseModel): target_url: str extraction_query: str viewport_width: int = 1280 viewport_height: int = 800 wait_until: str = "load" timeout: int = 30000 response_schema: dict = None class BatchItem(BaseModel): target_url: str extraction_query: str viewport_width: int = 1280 viewport_height: int = 800 wait_until: str = "load" timeout: int = 30000 response_schema: dict = None login: LoginInfo = None class BatchRequest(BaseModel): requests: list[BatchItem] # --------------------------------------------------------------------------- # Unified API key authentication dependency # Accepts: X-API-Key (direct clients) # --------------------------------------------------------------------------- API_KEY_NAME = "X-API-Key" api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False) async def get_api_key( request: Request, api_key: str = Depends(api_key_header), ): if api_key and not api_key.startswith('op_live_'): raise HTTPException( status_code=401, detail="Invalid API Key format" ) start_time = time.time() if not api_key: if not IS_PRODUCTION 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: try: 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)" ) except Exception as e: logger.error(f"API key lookup failed: {e}") raise HTTPException(status_code=401, detail="Invalid API Key") 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={ 'error': 'Monthly request limit exceeded', 'current_usage': context["current_usage"], 'monthly_limit': context["monthly_limit"], 'tier': context["tier"], 'upgrade_url': 'https://opticparse.com' } ) request.state.user_ctx = context asyncio.create_task(log_usage(context, request.url.path, "opticparse", 200, 50)) return context def clean_json_response(text: str) -> str: """ Cleans markdown formatting and extracts the first JSON object or array found in the text. """ text = text.strip() if text.startswith("```json"): text = text[7:].strip() elif text.startswith("```"): text = text[3:].strip() if text.endswith("```"): text = text[:-3].strip() start_idx = -1 end_idx = -1 for idx, char in enumerate(text): if char in ('{', '['): start_idx = idx break for idx in range(len(text) - 1, -1, -1): if text[idx] in ('}', ']'): end_idx = idx break if start_idx != -1 and end_idx != -1 and end_idx >= start_idx: return text[start_idx:end_idx + 1] return text async def run_vision_extraction( target_url: str, extraction_query: str, wait_until: str = "load", timeout: int = 30000, viewport_width: int = 1280, viewport_height: int = 800, response_schema: dict = None, login: LoginInfo = None, actions: list = None ) -> str: # 1. Check cache first cached = await get_cache(target_url, extraction_query, response_schema) if cached: return cached # Stealth mode setup — bypasses basic bot detection (Cloudflare JS challenges) STEALTH_UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/125.0.0.0 Safari/537.36" ) screenshot_bytes = None try: async with BROWSER_SEMAPHORE: async with async_playwright() as p: for attempt in range(2): try: logger.info(f"Launching headless Chromium browser (stealth mode) - Attempt {attempt+1}") browserless_key = os.getenv('BROWSERLESS_API_KEY') if browserless_key: browser = await p.chromium.connect_over_cdp( f"wss://chrome.browserless.io?token={browserless_key}" ) else: browser = await p.chromium.launch(headless=True, executable_path=os.getenv("CHROMIUM_PATH", None)) try: context = await browser.new_context( user_agent=STEALTH_UA, extra_http_headers={"Accept-Language": "en-US,en;q=0.9"}, java_script_enabled=True, ) # Remove webdriver flag — prevents Cloudflare detection await context.add_init_script( "Object.defineProperty(navigator, 'webdriver', {get: () => undefined});" ) page = await context.new_page() await page.set_viewport_size({"width": viewport_width, "height": viewport_height}) # Block bandwidth-heavy assets (saves ~60% per request) async def block_heavy_assets(route): if route.request.resource_type in ("media", "font", "websocket", "other"): await route.abort() else: await route.continue_() await page.route("**/*", block_heavy_assets) # Execute login if requested if login: logger.info(f"Executing login on: {login.login_url}") await page.goto(login.login_url, wait_until="load", timeout=timeout) await page.fill(login.username_field, login.username) await page.fill(login.password_field, login.password) if login.submit_button: await page.click(login.submit_button) else: await page.keyboard.press("Enter") # Wait for redirects/cookies/session setup await page.wait_for_load_state(state="networkidle", timeout=timeout) logger.info("Login action executed and network is idle") logger.info(f"Navigating to {target_url}") try: await page.goto(target_url, wait_until=wait_until, timeout=timeout) except Exception as goto_err: if "Timeout" in str(goto_err): logger.warning("Page navigation timed out, attempting screenshot of current state.") else: raise goto_err if actions: logger.info(f"Executing {len(actions)} agentic actions") for action in actions: # action can be a dict (if passed from api) or ActionInfo act_type = action.type if hasattr(action, 'type') else action.get("type") act_sel = action.selector if hasattr(action, 'selector') else action.get("selector") act_val = action.value if hasattr(action, 'value') else action.get("value") act_ms = action.ms if hasattr(action, 'ms') else action.get("ms") act_key = action.key if hasattr(action, 'key') else action.get("key") try: if act_type == "click" and act_sel: await page.click(act_sel) elif act_type == "fill" and act_sel and act_val is not None: await page.fill(act_sel, act_val) elif act_type == "wait" and act_ms: await page.wait_for_timeout(act_ms) elif act_type == "press" and act_key: await page.keyboard.press(act_key) except Exception as act_err: logger.warning(f"Action {act_type} failed: {act_err}") logger.info("Taking screenshot") screenshot_bytes = await page.screenshot(full_page=True, type="png") break # Success, break out of retry loop finally: logger.info("Closing browser") await browser.close() except Exception as loop_err: if attempt == 1: raise loop_err logger.warning(f"Playwright attempt {attempt+1} failed: {str(loop_err)}. Retrying...") except Exception as e: logger.error(f"Playwright error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"Playwright error: {str(e)}") if not screenshot_bytes: logger.error("Failed to capture page screenshot.") raise HTTPException(status_code=500, detail="Failed to capture page screenshot.") # 3. Use AI provider rotation for extraction try: raw_response = await call_ai_with_rotation(screenshot_bytes, extraction_query, response_schema) cleaned_json = clean_json_response(raw_response) try: json.loads(cleaned_json) logger.info("Successfully extracted and parsed valid JSON response") except json.JSONDecodeError: logger.warning("Response could not be parsed as JSON, returning raw text") await set_cache(target_url, extraction_query, cleaned_json, response_schema) return cleaned_json except HTTPException: raise except Exception as e: logger.error(f"AI extraction error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"AI extraction error: {str(e)}") @app.post("/api/vision-scrape") @limiter.limit("10/minute") async def vision_scrape(request: Request, body: ScrapeRequest, api_key: str = Depends(get_api_key)): logger.info(f"Received scraping request for target_url: {body.target_url}") if body.wait_until not in ("networkidle", "load", "domcontentloaded"): logger.warning(f"Invalid wait_until option: {body.wait_until}") raise HTTPException( status_code=400, detail="wait_until must be one of 'networkidle', 'load', 'domcontentloaded'" ) result = await run_vision_extraction( target_url=body.target_url, extraction_query=body.extraction_query, wait_until=body.wait_until, timeout=body.timeout, viewport_width=body.viewport_width, viewport_height=body.viewport_height, response_schema=body.response_schema, login=body.login, actions=body.actions ) return Response(content=result, media_type="application/json") @app.post("/api/vision-scrape/direct") @limiter.limit("20/minute") async def vision_scrape_direct(request: Request, body: DirectScrapeRequest, api_key: str = Depends(get_api_key)): logger.info(f"Received direct scraping request with base64 image") try: img_str = body.image_base64 if "base64," in img_str: img_str = img_str.split("base64,")[1] screenshot_bytes = base64.b64decode(img_str) except Exception as e: logger.error(f"Failed to decode base64 image: {e}") raise HTTPException(status_code=400, detail="Invalid image_base64 format") try: raw_response = await call_ai_with_rotation(screenshot_bytes, body.extraction_query, body.response_schema) cleaned_json = clean_json_response(raw_response) try: json.loads(cleaned_json) except json.JSONDecodeError: logger.warning("Response could not be parsed as JSON, returning raw text") return Response(content=cleaned_json, media_type="application/json") except HTTPException: raise except Exception as e: logger.error(f"AI extraction error: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"AI extraction error: {str(e)}") @app.post("/api/vision-scrape/async") async def vision_scrape_async( request: ScrapeRequest, background_tasks: BackgroundTasks, api_key_data: dict = Depends(get_api_key) ): """Submit a scraping job and get a job_id back immediately""" job_id = str(uuid_module.uuid4()) _async_jobs[job_id] = { "status": "queued", "created_at": datetime.utcnow().isoformat(), "result": None, "error": None, "webhook_url": request.webhook_url if hasattr(request, 'webhook_url') else None } async def run_job(): try: _async_jobs[job_id]["status"] = "processing" result = await run_vision_extraction( target_url=request.target_url, extraction_query=request.extraction_query, wait_until=request.wait_until, timeout=request.timeout, viewport_width=request.viewport_width, viewport_height=request.viewport_height, response_schema=request.response_schema, login=request.login ) _async_jobs[job_id]["status"] = "completed" _async_jobs[job_id]["result"] = json.loads(result) if isinstance(result, str) else result webhook_url = _async_jobs[job_id].get("webhook_url") if webhook_url: try: async with httpx.AsyncClient() as client: await client.post( webhook_url, json={ "job_id": job_id, "status": "completed", "result": _async_jobs[job_id]["result"] }, timeout=10.0 ) except Exception as webhook_err: logger.warning(f"Webhook delivery failed: {webhook_err}") except Exception as e: _async_jobs[job_id]["status"] = "failed" _async_jobs[job_id]["error"] = str(e) logger.error(f"Async job {job_id} failed: {e}") background_tasks.add_task(run_job) return { "job_id": job_id, "status": "queued", "poll_url": f"/api/vision-scrape/jobs/{job_id}", "message": "Job queued. Poll the poll_url for results." } @app.get("/api/vision-scrape/jobs/{job_id}") async def get_job_status( job_id: str, api_key_data: dict = Depends(get_api_key) ): """Check the status of an async scraping job""" if job_id not in _async_jobs: raise HTTPException( status_code=404, detail="Job not found" ) job = _async_jobs[job_id] return { "job_id": job_id, "status": job["status"], "created_at": job["created_at"], "result": job["result"] if job["status"] == "completed" else None, "error": job["error"] if job["status"] == "failed" else None } @app.post("/api/crawl") @limiter.limit("5/minute") async def api_crawl(request: Request, body: CrawlRequest, api_key: str = Depends(get_api_key)): logger.info(f"Received crawling request starting at: {body.start_url}") if body.wait_until not in ("networkidle", "load", "domcontentloaded"): logger.warning(f"Invalid wait_until option: {body.wait_until}") raise HTTPException( status_code=400, detail="wait_until must be one of 'networkidle', 'load', 'domcontentloaded'" ) STEALTH_UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/125.0.0.0 Safari/537.36" ) results = [] try: async with BROWSER_SEMAPHORE: async with async_playwright() as p: logger.info("Launching headless Chromium browser for crawl") browserless_key = os.getenv('BROWSERLESS_API_KEY') if browserless_key: browser = await p.chromium.connect_over_cdp( f"wss://chrome.browserless.io?token={browserless_key}" ) else: browser = await p.chromium.launch(headless=True, executable_path=os.getenv("CHROMIUM_PATH", None)) try: context = await browser.new_context( user_agent=STEALTH_UA, extra_http_headers={"Accept-Language": "en-US,en;q=0.9"}, java_script_enabled=True, ) await context.add_init_script( "Object.defineProperty(navigator, 'webdriver', {get: () => undefined});" ) page = await context.new_page() await page.set_viewport_size({"width": body.viewport_width, "height": body.viewport_height}) async def block_heavy_assets(route): if route.request.resource_type in ("media", "font", "websocket", "other"): await route.abort() else: await route.continue_() await page.route("**/*", block_heavy_assets) current_url = request.start_url logger.info(f"Navigating to start URL: {current_url}") await page.goto(current_url, wait_until=request.wait_until, timeout=request.timeout) for page_num in range(1, request.max_pages + 1): logger.info(f"Scraping page {page_num} (URL: {page.url})") # Check cache first for this specific URL + query + schema cached = await get_cache(page.url, request.extraction_query, request.response_schema) if cached: try: page_json = json.loads(cached) if isinstance(page_json, list): results.extend(page_json) else: results.append(page_json) except Exception: results.append(cached) else: screenshot_bytes = await page.screenshot(full_page=True, type="png") raw_response = await call_ai_with_rotation( screenshot_bytes, request.extraction_query, request.response_schema ) cleaned_json = clean_json_response(raw_response) await set_cache(page.url, request.extraction_query, cleaned_json, request.response_schema) try: page_json = json.loads(cleaned_json) if isinstance(page_json, list): results.extend(page_json) else: results.append(page_json) except Exception: results.append(cleaned_json) if page_num == request.max_pages: break # Look for next button next_btn = None try: next_btn = page.locator(request.follow_selector) if await next_btn.count() > 0 and await next_btn.first.is_visible(): next_btn = next_btn.first else: next_btn = None except Exception: next_btn = None if not next_btn: try: next_btn = page.get_by_text(request.follow_selector, exact=False) if await next_btn.count() > 0 and await next_btn.first.is_visible(): next_btn = next_btn.first else: next_btn = None except Exception: next_btn = None if not next_btn: logger.info(f"Next button not found/visible after page {page_num}. Ending crawl.") break logger.info(f"Clicking next button to proceed to page {page_num + 1}") await next_btn.click() await page.wait_for_load_state(state=request.wait_until, timeout=request.timeout) finally: logger.info("Closing browser") await browser.close() except Exception as e: logger.error(f"Playwright error during crawl: {str(e)}", exc_info=True) raise HTTPException(status_code=500, detail=f"Playwright error during crawl: {str(e)}") return results def compute_json_diff(prev_val, curr_val): """Computes a structured diff between two JSON structures (lists, dicts, or primitives)""" if isinstance(prev_val, list) and isinstance(curr_val, list): prev_strs = [json.dumps(item, sort_keys=True) for item in prev_val] curr_strs = [json.dumps(item, sort_keys=True) for item in curr_val] added = [curr_val[i] for i, s in enumerate(curr_strs) if s not in prev_strs] removed = [prev_val[i] for i, s in enumerate(prev_strs) if s not in curr_strs] return { "changed": len(added) > 0 or len(removed) > 0, "type": "list", "added": added, "removed": removed } elif isinstance(prev_val, dict) and isinstance(curr_val, dict): added = {} removed = {} modified = {} for k, v in curr_val.items(): if k not in prev_val: added[k] = v elif prev_val[k] != v: modified[k] = {"from": prev_val[k], "to": v} for k, v in prev_val.items(): if k not in curr_val: removed[k] = v changed = len(added) > 0 or len(removed) > 0 or len(modified) > 0 return { "changed": changed, "type": "dict", "added": added, "removed": removed, "modified": modified } else: return { "changed": prev_val != curr_val, "type": "primitive", "previous": prev_val, "current": curr_val } @app.post("/api/watch") @limiter.limit("20/minute") async def create_watch(request: Request, body: WatchRequest, api_key: str = Depends(get_api_key)): logger.info(f"Creating watch for target_url: {body.target_url}") # 1. Run the initial scrape initial_result = await run_vision_extraction( target_url=body.target_url, extraction_query=body.extraction_query, wait_until=body.wait_until, timeout=body.timeout, viewport_width=body.viewport_width, viewport_height=body.viewport_height, response_schema=body.response_schema ) # 2. Store watch inside SQLite watch_id = str(uuid.uuid4()) schema_str = json.dumps(body.response_schema) if body.response_schema else None user_id = request.state.user_ctx.get("user_id") try: def _do_insert(cursor): placeholder = get_db_placeholder() cursor.execute( f"INSERT INTO watches (id, user_id, url, query, schema_text, last_result, created_at) VALUES ({placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder}, {placeholder})", (watch_id, user_id, body.target_url, body.extraction_query, schema_str, initial_result, time.time()) ) await asyncio.to_thread(run_db_query, _do_insert) logger.info(f"Watch created successfully with ID: {watch_id}") except Exception as e: logger.error(f"Failed to store watch in database: {e}") raise HTTPException(status_code=500, detail="Database write error.") try: parsed_result = json.loads(initial_result) except Exception: parsed_result = initial_result return { "watch_id": watch_id, "url": body.target_url, "query": body.extraction_query, "initial_result": parsed_result } @app.get("/api/watch/{watch_id}/diff") async def get_watch_diff(watch_id: str, api_key: str = Depends(get_api_key)): logger.info(f"Fetching diff for watch ID: {watch_id}") user_id = request.state.user_ctx.get("user_id") # 1. Retrieve watch details from SQLite try: def _do_select(cursor): placeholder = get_db_placeholder() cursor.execute(f"SELECT url, query, schema_text, last_result FROM watches WHERE id = {placeholder} AND user_id = {placeholder}", (watch_id, user_id)) return cursor.fetchone() row = await asyncio.to_thread(run_db_query, _do_select) except Exception as e: logger.error(f"Database error while querying watch {watch_id}: {e}") raise HTTPException(status_code=500, detail="Database query error.") if not row: raise HTTPException(status_code=404, detail="Watch not found.") url, query, schema_text, last_result_str = row response_schema = json.loads(schema_text) if schema_text else None # 2. Re-scrape the page new_result_str = await run_vision_extraction( target_url=url, extraction_query=query, response_schema=response_schema ) # 3. Compare JSONs try: prev_json = json.loads(last_result_str) curr_json = json.loads(new_result_str) diff = compute_json_diff(prev_json, curr_json) except Exception as e: logger.warning(f"Failed to parse results as JSON, falling back to raw diff: {e}") diff = { "changed": last_result_str != new_result_str, "type": "raw", "previous": last_result_str, "current": new_result_str } # 4. If changed, update the last_result in database if diff["changed"]: try: def _do_update(cursor): placeholder = get_db_placeholder() cursor.execute(f"UPDATE watches SET last_result = {placeholder} WHERE id = {placeholder} AND user_id = {placeholder}", (new_result_str, watch_id, user_id)) await asyncio.to_thread(run_db_query, _do_update) logger.info(f"Watch {watch_id} updated with new result") except Exception as e: logger.error(f"Failed to update watch in database: {e}") return { "watch_id": watch_id, "url": url, "query": query, "diff": diff } @app.delete("/api/watch/{watch_id}") async def delete_watch(watch_id: str, api_key: str = Depends(get_api_key)): logger.info(f"Deleting watch ID: {watch_id}") user_id = request.state.user_ctx.get("user_id") try: def _do_delete(cursor): placeholder = get_db_placeholder() cursor.execute(f"SELECT id FROM watches WHERE id = {placeholder} AND user_id = {placeholder}", (watch_id, user_id)) if not cursor.fetchone(): return False cursor.execute(f"DELETE FROM watches WHERE id = {placeholder} AND user_id = {placeholder}", (watch_id, user_id)) return True found = await asyncio.to_thread(run_db_query, _do_delete) if not found: raise HTTPException(status_code=404, detail="Watch not found.") logger.info(f"Watch ID: {watch_id} deleted successfully") return {"status": "deleted", "watch_id": watch_id} except HTTPException: raise except Exception as e: logger.error(f"Failed to delete watch in database: {e}") raise HTTPException(status_code=500, detail="Database error.") @app.get("/gateway/watches") async def list_all_watches(): try: def _do_select(cursor): cursor.execute("SELECT id, url, query, created_at, last_result FROM watches ORDER BY created_at DESC") # Fetch column names columns = [desc[0] for desc in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()] watches = await asyncio.to_thread(run_db_query, _do_select) return {"watches": watches} except Exception as e: logger.error(f"Failed to list watches: {e}") raise HTTPException(status_code=500, detail="Database error") from pydantic import BaseModel class WatchCreateRequest(BaseModel): url: str query: str @app.post("/gateway/watches") async def create_watch(req: WatchCreateRequest): import uuid import time watch_id = str(uuid.uuid4()) try: def _do_insert(cursor): if DATABASE_URL: cursor.execute( "INSERT INTO watches (id, url, query, created_at) VALUES (%s, %s, %s, %s)", (watch_id, req.url, req.query, time.time()) ) else: cursor.execute( "INSERT INTO watches (id, url, query, created_at) VALUES (?, ?, ?, ?)", (watch_id, req.url, req.query, time.time()) ) await asyncio.to_thread(run_db_query, _do_insert) return {"status": "success", "id": watch_id} except Exception as e: logger.error(f"Failed to create watch: {e}") raise HTTPException(status_code=500, detail="Database error") @app.post("/api/batch") @limiter.limit("3/minute") async def api_batch(request: Request, body: BatchRequest, api_key: str = Depends(get_api_key)): logger.info(f"Received batch scraping request with size: {len(body.requests)}") if len(body.requests) > 20: raise HTTPException(status_code=400, detail="Maximum batch size is 20 requests.") tasks = [] for req in body.requests: tasks.append( run_vision_extraction( target_url=req.target_url, extraction_query=req.extraction_query, wait_until=req.wait_until, timeout=req.timeout, viewport_width=req.viewport_width, viewport_height=req.viewport_height, response_schema=req.response_schema, login=req.login ) ) raw_results = await asyncio.gather(*tasks, return_exceptions=True) formatted_results = [] for i, res in enumerate(raw_results): req = body.requests[i] if isinstance(res, Exception): formatted_results.append({ "url": req.target_url, "status": "error", "error": str(res) }) else: try: parsed = json.loads(res) except Exception: parsed = res formatted_results.append({ "url": req.target_url, "status": "success", "data": parsed }) return {"results": formatted_results} class KeyGenerateRequest(BaseModel): user_id: str email: str = None @app.post("/gateway/keys/generate") @limiter.limit("5/minute") async def generate_key(request: Request, req: KeyGenerateRequest): user_check = await supabase_query("GET", "users", f"id=eq.{req.user_id}") if not user_check: logger.info(f"User {req.user_id} not found in public.users. Creating them now.") await supabase_query("POST", "users", body={ "id": req.user_id, "email": req.email, "tier": "free", "monthly_limit": 100, "current_usage": 0 }) existing_keys = await supabase_query( "GET", "api_keys", f"user_id=eq.{req.user_id}&is_active=eq.true&select=id,key_prefix" ) if len(existing_keys) >= 3: raise HTTPException( status_code=400, detail="Maximum of 3 active API keys per account. Regenerate or delete an existing key." ) 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/keys/{user_id}") async def list_keys(user_id: str): keys = await supabase_query( "GET", "api_keys", f"user_id=eq.{user_id}&is_active=eq.true&select=id,key_prefix,created_at" ) return {"keys": keys} @app.delete("/gateway/keys/{user_id}/{prefix}") async def revoke_key(user_id: str, prefix: str): await supabase_query("PATCH", "api_keys", f"user_id=eq.{user_id}&key_prefix=eq.{prefix}", body={"is_active": False}) return {"status": "success"} @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] @app.get("/gateway/usage/{user_id}/history") async def get_usage_history( user_id: str, api_key_data: dict = Depends(get_api_key) ): """Returns daily usage breakdown for last 30 days""" try: # Query usage_logs table grouped by date logs = await supabase_query( "GET", "usage_logs", f"user_id=eq.{user_id}&select=created_at&order=created_at.desc&limit=1000" ) # Group by date from collections import defaultdict from datetime import datetime as dt, timedelta daily_counts = defaultdict(int) for log in logs: if log.get('created_at'): date = log['created_at'][:10] daily_counts[date] += 1 # Fill in last 30 days including zeros today = dt.utcnow().date() history = [] for i in range(29, -1, -1): date = today - timedelta(days=i) date_str = str(date) history.append({ "date": date_str, "count": daily_counts.get(date_str, 0) }) return { "user_id": user_id, "history": history, "total_days": 30 } except Exception as e: logger.error(f"Usage history error: {e}") raise HTTPException( status_code=500, detail="Failed to fetch usage history" ) @app.get("/gateway/usage/{user_id}/logs") async def get_usage_logs_raw(user_id: str): """Returns raw API logs for the audit trail table""" try: logs = await supabase_query( "GET", "usage_logs", f"user_id=eq.{user_id}&select=created_at,endpoint,service,status_code,response_time_ms&order=created_at.desc&limit=50" ) return {"logs": logs} except Exception as e: logger.error(f"Failed to fetch raw usage logs: {e}") raise HTTPException(status_code=500, detail="Failed to fetch logs") 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) VARIANT_MAPPING = { # Replace these IDs with your actual Lemon Squeezy variant IDs when you create them "variant_pro_123": {"tier": "pro", "monthly_limit": 2000}, "variant_bus_456": {"tier": "business", "monthly_limit": 10000}, "variant_ent_789": {"tier": "enterprise", "monthly_limit": 50000}, } @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"}) variant_id = str(data.get("data", {}).get("attributes", {}).get("variant_id", "")) if event_name in ("subscription_created", "subscription_payment_success", "subscription_resumed"): # Default to Pro if we can't find the variant in the mapping plan = VARIANT_MAPPING.get(variant_id, {"tier": "pro", "monthly_limit": 2000}) await supabase_query("PATCH", "users", f"id=eq.{user_id}", body={ "tier": plan["tier"], "monthly_limit": plan["monthly_limit"], "lemon_customer_id": str(data.get("data", {}).get("attributes", {}).get("customer_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": 50}) 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")) LEMON_SQUEEZY_WEBHOOK_SECRET = os.environ.get("LEMON_SQUEEZY_WEBHOOK_SECRET") @app.post("/gateway/webhooks/lemonsqueezy") async def lemonsqueezy_webhook(request: Request): # Get raw body BEFORE parsing body = await request.body() # Get signature from header signature = request.headers.get('X-Signature', '') # Verify signature using HMAC-SHA256 if LEMON_SQUEEZY_WEBHOOK_SECRET: secret = LEMON_SQUEEZY_WEBHOOK_SECRET.encode() expected = hmac.new( secret, body, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(expected, signature): logger.warning("Invalid webhook signature received") raise HTTPException( status_code=401, detail="Invalid webhook signature" ) try: data = json.loads(body) except Exception: raise HTTPException(status_code=400, detail="Invalid JSON body") # Logic for webhook handling... logger.info(f"Webhook received: {data.get('meta', {}).get('event_name')}") return {"status": "success"} class WebhookRequest(BaseModel): url: str @app.post("/api/settings/webhook") async def save_webhook(req: WebhookRequest, user_ctx: dict = Depends(get_api_key)): user_id = user_ctx["user_id"] try: def _do_upsert(cursor): if DATABASE_URL: cursor.execute( "INSERT INTO user_settings (user_id, webhook_url) VALUES (%s, %s) ON CONFLICT (user_id) DO UPDATE SET webhook_url = EXCLUDED.webhook_url", (user_id, req.url) ) else: cursor.execute( "INSERT OR REPLACE INTO user_settings (user_id, webhook_url) VALUES (?, ?)", (user_id, req.url) ) await asyncio.to_thread(run_db_query, _do_upsert) return {"status": "success"} except Exception as e: logger.error(f"Failed to save webhook: {e}") raise HTTPException(status_code=500, detail="Database error") @app.get("/api/settings/webhook") async def get_webhook(user_ctx: dict = Depends(get_api_key)): user_id = user_ctx["user_id"] try: def _do_select(cursor): if DATABASE_URL: cursor.execute("SELECT webhook_url FROM user_settings WHERE user_id = %s", (user_id,)) else: cursor.execute("SELECT webhook_url FROM user_settings WHERE user_id = ?", (user_id,)) res = cursor.fetchone() return res[0] if res else None url = await asyncio.to_thread(run_db_query, _do_select) return {"webhook_url": url or ""} except Exception as e: logger.error(f"Failed to get webhook: {e}") raise HTTPException(status_code=500, detail="Database error") import httpx async def background_watch_worker(): logger.info("Starting background watch worker...") while True: try: # 1. Fetch all watches def _get_watches(cursor): cursor.execute("SELECT id, url, query, last_result FROM watches") columns = [desc[0] for desc in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()] watches = await asyncio.to_thread(run_db_query, _get_watches) if watches: # 2. Get the global webhook URL (assuming single admin for now) def _get_webhook(cursor): cursor.execute("SELECT webhook_url FROM user_settings LIMIT 1") res = cursor.fetchone() return res[0] if res else None webhook_url = await asyncio.to_thread(run_db_query, _get_webhook) for watch in watches: logger.info(f"Processing watch: {watch['id']} for {watch['url']}") try: # Perform extraction result = await run_vision_extraction(target_url=watch['url'], extraction_query=watch['query']) new_result_str = json.dumps(result) if watch['last_result'] != new_result_str: logger.info(f"Change detected for {watch['id']}! Dispatching webhook...") # Update DB def _update_watch(cursor): cursor.execute("UPDATE watches SET last_result = %s WHERE id = %s" if DATABASE_URL else "UPDATE watches SET last_result = ? WHERE id = ?", (new_result_str, watch['id'])) await asyncio.to_thread(run_db_query, _update_watch) # Dispatch webhook if webhook_url: payload = { "event": "watch_change", "watch_id": watch['id'], "url": watch['url'], "new_data": result } async with httpx.AsyncClient() as client: await client.post(webhook_url, json=payload, timeout=10.0) except Exception as ex: logger.error(f"Failed to process watch {watch['id']}: {ex}") except Exception as e: logger.error(f"Error in background watch worker: {e}") await asyncio.sleep(300) # Run every 5 minutes @app.on_event("startup") async def startup_event(): asyncio.create_task(background_watch_worker()) if __name__ == "__main__": import uvicorn uvicorn.run("server:app", host="0.0.0.0", port=8000, reload=False)