server.py

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)