1160 lines
44 KiB
Python
1160 lines
44 KiB
Python
from fastapi import FastAPI, HTTPException, Header, Request
|
|
from playwright.async_api import async_playwright
|
|
import os
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import time
|
|
import signal
|
|
import psutil
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, List, Any
|
|
from urllib.parse import unquote
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
from fastapi.middleware.base import BaseHTTPMiddleware
|
|
from app.config import BROWSER_INSTANCE_TIMEOUT_MINUTES
|
|
|
|
# Add imports for browser pool
|
|
from asyncio import Queue, Lock, Semaphore
|
|
from contextlib import asynccontextmanager
|
|
from app.utils.browser_utils import force_cleanup_old_pages
|
|
from app.utils.cloudflare_bypass import CloudflareBypass
|
|
|
|
# Add memory monitoring
|
|
import gc
|
|
|
|
app = FastAPI()
|
|
|
|
# Get API key from environment variable
|
|
API_KEY = os.getenv('API_KEY')
|
|
if not API_KEY:
|
|
raise ValueError("API_KEY environment variable must be set")
|
|
|
|
# Get cache expiry time from environment variable (default: 36 hours)
|
|
CACHE_EXPIRY_HOURS = int(os.getenv('CACHE_EXPIRY_HOURS', '36'))
|
|
|
|
# Get cleanup cron schedule from environment variable (default: every day at 3 AM)
|
|
CLEANUP_CRON = os.getenv('CLEANUP_CRON', '0 3 * * *')
|
|
|
|
# Define custom user agent
|
|
CUSTOM_USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36'
|
|
|
|
# Browser pool configuration - Reduced for better resource management
|
|
MAX_BROWSERS = int(os.getenv('MAX_BROWSERS', '2')) # Reduced from 3 to 2 for better resource management
|
|
BROWSER_TTL = int(os.getenv('BROWSER_TTL', '900')) # Reduced from 1800 to 900 seconds (15 minutes)
|
|
MAX_CONCURRENT_OPERATIONS = int(os.getenv('MAX_CONCURRENT_OPERATIONS', '3')) # Reduced from 5 to 3
|
|
BROWSER_OPERATION_TIMEOUT = int(os.getenv('BROWSER_OPERATION_TIMEOUT', '60')) # New: timeout for browser operations
|
|
PAGE_TIMEOUT = int(os.getenv('PAGE_TIMEOUT', '30')) # New: timeout for page operations
|
|
|
|
# Browser pool management
|
|
browser_pool = Queue(maxsize=MAX_BROWSERS) # Add maxsize to prevent unbounded growth
|
|
browser_lock = Lock()
|
|
browser_creation_times = {}
|
|
page_creation_times = {} # Track page creation times for force cleanup
|
|
active_browsers = set() # Track active browsers
|
|
active_pages = set() # Track active pages for force cleanup
|
|
operation_semaphore = Semaphore(MAX_CONCURRENT_OPERATIONS) # Limit concurrent operations
|
|
playwright_instance = None # Global playwright instance
|
|
browser_health_check_task = None # Track health check task
|
|
|
|
# Rate limiting configuration
|
|
RATE_LIMIT_MINUTE = int(os.getenv('RATE_LIMIT_MINUTE', '60')) # requests per minute
|
|
RATE_LIMIT_WINDOW = 60 # window size in seconds
|
|
|
|
class RateLimitMiddleware(BaseHTTPMiddleware):
|
|
def __init__(self, app):
|
|
super().__init__(app)
|
|
self.requests = {}
|
|
self.lock = asyncio.Lock()
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
# Skip rate limiting for health check
|
|
if request.url.path == "/" and request.method == "HEAD":
|
|
return await call_next(request)
|
|
|
|
api_key = request.headers.get("x-api-key")
|
|
if not api_key:
|
|
raise HTTPException(status_code=401, detail="API key required")
|
|
|
|
async with self.lock:
|
|
now = time.time()
|
|
# Clean old requests
|
|
self.requests = {k: v for k, v in self.requests.items()
|
|
if now - v[-1] < RATE_LIMIT_WINDOW}
|
|
|
|
# Get request times for this API key
|
|
requests = self.requests.get(api_key, [])
|
|
# Remove old requests outside the window
|
|
requests = [t for t in requests if now - t < RATE_LIMIT_WINDOW]
|
|
|
|
if len(requests) >= RATE_LIMIT_MINUTE:
|
|
oldest = requests[0]
|
|
wait_time = RATE_LIMIT_WINDOW - (now - oldest)
|
|
raise HTTPException(
|
|
status_code=429,
|
|
detail=f"Rate limit exceeded. Try again in {int(wait_time)} seconds"
|
|
)
|
|
|
|
requests.append(now)
|
|
self.requests[api_key] = requests
|
|
|
|
return await call_next(request)
|
|
|
|
# Add rate limiting middleware
|
|
app.add_middleware(RateLimitMiddleware)
|
|
|
|
async def create_browser():
|
|
"""Create a new browser instance with improved resource management"""
|
|
global playwright_instance
|
|
|
|
try:
|
|
if playwright_instance is None:
|
|
playwright_instance = await async_playwright().start()
|
|
|
|
# Get proxy configuration from environment
|
|
proxy_url = os.getenv('PROXY_URL')
|
|
proxy_config = CloudflareBypass.get_proxy_config(proxy_url)
|
|
|
|
# Add memory and performance optimizations
|
|
browser_args = CloudflareBypass.get_stealth_args() + [
|
|
'--memory-pressure-off',
|
|
'--max_old_space_size=256', # Reduced memory usage
|
|
'--disable-dev-shm-usage',
|
|
'--disable-gpu-sandbox',
|
|
'--disable-software-rasterizer',
|
|
'--disable-background-timer-throttling',
|
|
'--disable-backgrounding-occluded-windows',
|
|
'--disable-renderer-backgrounding',
|
|
'--disable-features=TranslateUI',
|
|
'--disable-ipc-flooding-protection',
|
|
'--disable-hang-monitor',
|
|
'--disable-prompt-on-repost',
|
|
'--disable-sync',
|
|
'--no-first-run',
|
|
'--no-default-browser-check',
|
|
'--no-pings',
|
|
'--no-zygote',
|
|
'--single-process',
|
|
'--disable-background-networking',
|
|
'--disable-background-timer-throttling',
|
|
'--disable-backgrounding-occluded-windows',
|
|
'--disable-breakpad',
|
|
'--disable-component-extensions-with-background-pages',
|
|
'--disable-dev-shm-usage',
|
|
'--disable-features=TranslateUI',
|
|
'--disable-ipc-flooding-protection',
|
|
'--disable-renderer-backgrounding',
|
|
'--disable-sync',
|
|
'--force-color-profile=srgb',
|
|
'--metrics-recording-only',
|
|
'--no-first-run',
|
|
'--safebrowsing-disable-auto-update',
|
|
'--enable-automation',
|
|
'--password-store=basic',
|
|
'--use-mock-keychain',
|
|
]
|
|
|
|
browser = await playwright_instance.chromium.launch(
|
|
headless=True,
|
|
args=browser_args,
|
|
ignore_default_args=['--enable-automation'],
|
|
**proxy_config
|
|
)
|
|
|
|
browser_creation_times[browser] = time.time()
|
|
active_browsers.add(browser)
|
|
print(f"Created new browser instance. Total active browsers: {len(active_browsers)}")
|
|
return browser
|
|
except Exception as e:
|
|
print(f"Error creating browser: {e}")
|
|
# Clean up playwright instance if browser creation fails
|
|
if playwright_instance:
|
|
try:
|
|
await playwright_instance.stop()
|
|
playwright_instance = None
|
|
except:
|
|
pass
|
|
raise
|
|
|
|
async def cleanup_browser(browser):
|
|
"""Safely cleanup a browser instance with improved error handling"""
|
|
try:
|
|
if browser in active_browsers:
|
|
active_browsers.remove(browser)
|
|
|
|
if browser in browser_creation_times:
|
|
del browser_creation_times[browser]
|
|
|
|
# Close all pages first and clean up page tracking
|
|
try:
|
|
contexts = browser.contexts
|
|
for context in contexts:
|
|
pages = context.pages
|
|
for page in pages:
|
|
try:
|
|
if page in active_pages:
|
|
active_pages.remove(page)
|
|
if page in page_creation_times:
|
|
del page_creation_times[page]
|
|
await asyncio.wait_for(page.close(), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error closing page: {e}")
|
|
# Force close if normal close fails
|
|
try:
|
|
await page.close(force=True)
|
|
except:
|
|
pass
|
|
|
|
# Close context
|
|
try:
|
|
await asyncio.wait_for(context.close(), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error closing context: {e}")
|
|
except Exception as e:
|
|
print(f"Error during context/page cleanup: {e}")
|
|
|
|
# Close browser with timeout
|
|
try:
|
|
await asyncio.wait_for(browser.close(), timeout=10.0)
|
|
except asyncio.TimeoutError:
|
|
print("Browser close timeout, forcing close")
|
|
try:
|
|
await browser.close(force=True)
|
|
except:
|
|
pass
|
|
except Exception as e:
|
|
print(f"Error during browser cleanup: {e}")
|
|
|
|
print(f"Browser cleaned up. Total active browsers: {len(active_browsers)}")
|
|
except Exception as e:
|
|
print(f"Error during browser cleanup: {e}")
|
|
|
|
async def check_browser_health():
|
|
"""Check browser health and recycle if needed with improved efficiency"""
|
|
global browser_health_check_task
|
|
|
|
while True:
|
|
try:
|
|
# Sleep for 1 minute between checks (reduced from 2 minutes)
|
|
await asyncio.sleep(60)
|
|
|
|
# Monitor memory usage
|
|
await monitor_memory_usage()
|
|
|
|
# First, force cleanup old instances
|
|
await force_cleanup_old_instances()
|
|
|
|
# Use a timeout for the entire health check operation
|
|
async with asyncio.timeout(30): # 30 second timeout for health check
|
|
async with browser_lock:
|
|
# Get all browsers from the pool
|
|
browsers = []
|
|
while not browser_pool.empty():
|
|
try:
|
|
browsers.append(await browser_pool.get_nowait())
|
|
except asyncio.QueueEmpty:
|
|
break
|
|
|
|
# Check each browser
|
|
for browser in browsers:
|
|
try:
|
|
# Check if browser is too old
|
|
if time.time() - browser_creation_times.get(browser, 0) > BROWSER_TTL:
|
|
print(f"Recycling old browser (age: {time.time() - browser_creation_times.get(browser, 0):.0f}s)")
|
|
await cleanup_browser(browser)
|
|
browser = await create_browser()
|
|
else:
|
|
# Quick health check - just verify browser is still responsive
|
|
try:
|
|
contexts = browser.contexts
|
|
if not contexts:
|
|
# Browser has no contexts, might be dead
|
|
print("Browser has no contexts, recycling")
|
|
await cleanup_browser(browser)
|
|
browser = await create_browser()
|
|
except Exception as e:
|
|
print(f"Browser health check failed: {e}")
|
|
await cleanup_browser(browser)
|
|
browser = await create_browser()
|
|
|
|
# Put back in pool if healthy
|
|
if not browser_pool.full():
|
|
await browser_pool.put(browser)
|
|
else:
|
|
# Pool is full, cleanup this browser
|
|
await cleanup_browser(browser)
|
|
except Exception as e:
|
|
print(f"Error checking browser health: {e}")
|
|
# Cleanup the problematic browser
|
|
try:
|
|
await cleanup_browser(browser)
|
|
except:
|
|
pass
|
|
|
|
except asyncio.TimeoutError:
|
|
print("Browser health check timed out, continuing...")
|
|
except Exception as e:
|
|
print(f"Error in browser health check: {e}")
|
|
# Continue the loop even if there's an error
|
|
await asyncio.sleep(30) # Wait a bit before retrying
|
|
|
|
async def force_cleanup_old_instances():
|
|
"""Force cleanup old browser and page instances based on timeout"""
|
|
print(f"Checking for old browser/page instances (timeout: {BROWSER_INSTANCE_TIMEOUT_MINUTES} minutes)...")
|
|
|
|
current_time = time.time()
|
|
timeout_seconds = BROWSER_INSTANCE_TIMEOUT_MINUTES * 60
|
|
cleaned_browsers = 0
|
|
cleaned_pages = 0
|
|
|
|
async with browser_lock:
|
|
# Clean up old browsers
|
|
browsers_to_cleanup = []
|
|
for browser in list(active_browsers):
|
|
creation_time = browser_creation_times.get(browser, 0)
|
|
if current_time - creation_time > timeout_seconds:
|
|
browsers_to_cleanup.append(browser)
|
|
print(f"Marking browser for cleanup (age: {(current_time - creation_time)/60:.1f} minutes)")
|
|
|
|
for browser in browsers_to_cleanup:
|
|
try:
|
|
await cleanup_browser(browser)
|
|
cleaned_browsers += 1
|
|
except Exception as e:
|
|
print(f"Error cleaning up old browser: {e}")
|
|
|
|
# Clean up old pages (this is a fallback for pages that might not be properly tracked)
|
|
for browser in list(active_browsers):
|
|
try:
|
|
if browser.contexts:
|
|
for context in browser.contexts:
|
|
for page in context.pages:
|
|
if page in page_creation_times:
|
|
creation_time = page_creation_times[page]
|
|
if current_time - creation_time > timeout_seconds:
|
|
try:
|
|
if page in active_pages:
|
|
active_pages.remove(page)
|
|
if page in page_creation_times:
|
|
del page_creation_times[page]
|
|
await page.close()
|
|
cleaned_pages += 1
|
|
print(f"Force closed old page (age: {(current_time - creation_time)/60:.1f} minutes)")
|
|
except Exception as e:
|
|
print(f"Error closing old page: {e}")
|
|
except Exception as e:
|
|
print(f"Error checking pages in browser: {e}")
|
|
|
|
# Also cleanup pages from browser_utils module
|
|
await force_cleanup_old_pages()
|
|
|
|
print(f"Force cleanup completed: {cleaned_browsers} browsers, {cleaned_pages} pages cleaned")
|
|
|
|
async def force_cleanup_all_browsers():
|
|
"""Force cleanup all browsers - useful for emergency situations with improved efficiency"""
|
|
print("Force cleaning up all browsers...")
|
|
|
|
try:
|
|
async with asyncio.timeout(20): # 20 second timeout for force cleanup
|
|
async with browser_lock:
|
|
# Clean up browsers in pool
|
|
while not browser_pool.empty():
|
|
try:
|
|
browser = await asyncio.wait_for(browser_pool.get_nowait(), timeout=1.0)
|
|
await asyncio.wait_for(cleanup_browser(browser), timeout=5.0)
|
|
except (asyncio.TimeoutError, asyncio.QueueEmpty):
|
|
break
|
|
except Exception as e:
|
|
print(f"Error cleaning up browser from pool: {e}")
|
|
|
|
# Clean up any remaining active browsers
|
|
for browser in list(active_browsers):
|
|
try:
|
|
await asyncio.wait_for(cleanup_browser(browser), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error force cleaning browser: {e}")
|
|
|
|
print("Force cleanup completed")
|
|
except asyncio.TimeoutError:
|
|
print("Force cleanup timed out")
|
|
except Exception as e:
|
|
print(f"Error during force cleanup: {e}")
|
|
|
|
@app.on_event("startup")
|
|
async def init_browser_pool():
|
|
"""Initialize the browser pool on startup with improved error handling"""
|
|
print("Initializing browser pool...")
|
|
|
|
try:
|
|
# Start browser health check task
|
|
global browser_health_check_task
|
|
browser_health_check_task = asyncio.create_task(check_browser_health())
|
|
|
|
# Pre-populate pool with initial browsers (reduced from 2 to 1)
|
|
for _ in range(min(1, MAX_BROWSERS)):
|
|
try:
|
|
browser = await asyncio.wait_for(create_browser(), timeout=30.0)
|
|
await asyncio.wait_for(browser_pool.put(browser), timeout=5.0)
|
|
print(f"Created initial browser {_ + 1}")
|
|
except (asyncio.TimeoutError, Exception) as e:
|
|
print(f"Error creating initial browser {_ + 1}: {e}")
|
|
break
|
|
|
|
print(f"Browser pool initialized with {browser_pool.qsize()} browsers")
|
|
except Exception as e:
|
|
print(f"Error during browser pool initialization: {e}")
|
|
# Don't fail startup, but log the error
|
|
|
|
@app.on_event("shutdown")
|
|
async def cleanup_browser_pool():
|
|
"""Cleanup browser pool on shutdown with improved error handling"""
|
|
print("Cleaning up browser pool...")
|
|
|
|
try:
|
|
# Cancel health check task
|
|
global browser_health_check_task
|
|
if browser_health_check_task and not browser_health_check_task.done():
|
|
browser_health_check_task.cancel()
|
|
try:
|
|
await browser_health_check_task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
# Force cleanup all browsers with timeout
|
|
await asyncio.wait_for(force_cleanup_all_browsers(), timeout=30.0)
|
|
|
|
# Stop playwright instance
|
|
global playwright_instance
|
|
if playwright_instance:
|
|
try:
|
|
await asyncio.wait_for(playwright_instance.stop(), timeout=10.0)
|
|
except asyncio.TimeoutError:
|
|
print("Playwright stop timeout")
|
|
except Exception as e:
|
|
print(f"Error stopping playwright: {e}")
|
|
finally:
|
|
playwright_instance = None
|
|
|
|
print("Browser pool cleanup completed")
|
|
except asyncio.TimeoutError:
|
|
print("Browser pool cleanup timed out")
|
|
except Exception as e:
|
|
print(f"Error during browser pool cleanup: {e}")
|
|
|
|
def signal_handler(signum, frame):
|
|
"""Handle shutdown signals"""
|
|
print(f"Received signal {signum}, shutting down gracefully...")
|
|
asyncio.create_task(cleanup_browser_pool())
|
|
exit(0)
|
|
|
|
# Register signal handlers
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
signal.signal(signal.SIGTERM, signal_handler)
|
|
|
|
# Add memory monitoring
|
|
async def monitor_memory_usage():
|
|
"""Monitor memory usage and trigger cleanup if needed"""
|
|
try:
|
|
memory = psutil.virtual_memory()
|
|
if memory.percent > 85: # If memory usage is above 85%
|
|
print(f"High memory usage detected: {memory.percent}%")
|
|
# Force garbage collection
|
|
gc.collect()
|
|
# Force cleanup old instances
|
|
await force_cleanup_old_instances()
|
|
print("Memory cleanup completed")
|
|
except Exception as e:
|
|
print(f"Error monitoring memory: {e}")
|
|
|
|
def init_db():
|
|
"""Initialize the SQLite database"""
|
|
conn = sqlite3.connect('/db/cache.db')
|
|
cursor = conn.cursor()
|
|
|
|
# Create cache table
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS cache (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
url TEXT NOT NULL,
|
|
route TEXT NOT NULL,
|
|
data TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(url, route)
|
|
)
|
|
''')
|
|
|
|
# Create index for faster lookups
|
|
cursor.execute('''
|
|
CREATE INDEX IF NOT EXISTS idx_cache_url_route
|
|
ON cache(url, route)
|
|
''')
|
|
|
|
# Create index for cleanup operations
|
|
cursor.execute('''
|
|
CREATE INDEX IF NOT EXISTS idx_cache_created_at
|
|
ON cache(created_at)
|
|
''')
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
print("Database initialized")
|
|
|
|
def get_cached_data(url, route):
|
|
"""Get cached data for a URL and route with improved error handling"""
|
|
try:
|
|
conn = sqlite3.connect('/db/cache.db', timeout=10.0) # Add timeout
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute('''
|
|
SELECT data, created_at FROM cache
|
|
WHERE url = ? AND route = ?
|
|
''', (url, route))
|
|
|
|
result = cursor.fetchone()
|
|
conn.close()
|
|
|
|
if result:
|
|
data, created_at = result
|
|
created_time = datetime.fromisoformat(created_at)
|
|
|
|
# Check if cache is still valid
|
|
if datetime.now() - created_time < timedelta(hours=CACHE_EXPIRY_HOURS):
|
|
return json.loads(data)
|
|
|
|
return None
|
|
except sqlite3.OperationalError as e:
|
|
print(f"Database operational error getting cached data: {e}")
|
|
return None
|
|
except Exception as e:
|
|
print(f"Error getting cached data: {e}")
|
|
return None
|
|
|
|
def save_to_cache(url, route, data):
|
|
"""Save data to cache with improved error handling"""
|
|
try:
|
|
conn = sqlite3.connect('/db/cache.db', timeout=10.0) # Add timeout
|
|
cursor = conn.cursor()
|
|
|
|
# Limit data size to prevent memory issues
|
|
data_str = json.dumps(data)
|
|
if len(data_str) > 10 * 1024 * 1024: # 10MB limit
|
|
print(f"Data too large for cache: {len(data_str)} bytes")
|
|
conn.close()
|
|
return
|
|
|
|
cursor.execute('''
|
|
INSERT OR REPLACE INTO cache (url, route, data, created_at)
|
|
VALUES (?, ?, ?, CURRENT_TIMESTAMP)
|
|
''', (url, route, data_str))
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
except sqlite3.OperationalError as e:
|
|
print(f"Database operational error saving to cache: {e}")
|
|
except Exception as e:
|
|
print(f"Error saving to cache: {e}")
|
|
|
|
def cleanup_old_cache_entries():
|
|
"""Clean up old cache entries"""
|
|
try:
|
|
conn = sqlite3.connect('/db/cache.db')
|
|
cursor = conn.cursor()
|
|
|
|
# Delete entries older than CACHE_EXPIRY_HOURS
|
|
cutoff_time = datetime.now() - timedelta(hours=CACHE_EXPIRY_HOURS)
|
|
|
|
cursor.execute('''
|
|
DELETE FROM cache
|
|
WHERE created_at < ?
|
|
''', (cutoff_time.isoformat(),))
|
|
|
|
deleted_count = cursor.rowcount
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
print(f"Cleaned up {deleted_count} old cache entries")
|
|
except Exception as e:
|
|
print(f"Error cleaning up cache: {e}")
|
|
|
|
# Initialize database
|
|
init_db()
|
|
|
|
# Initialize scheduler for periodic cache cleanup
|
|
scheduler = BackgroundScheduler()
|
|
scheduler.add_job(
|
|
cleanup_old_cache_entries,
|
|
CronTrigger.from_crontab(CLEANUP_CRON),
|
|
id='cache_cleanup_job',
|
|
replace_existing=True
|
|
)
|
|
|
|
@app.on_event("startup")
|
|
def start_scheduler():
|
|
scheduler.start()
|
|
print(f"Cache cleanup scheduler started with cron: {CLEANUP_CRON}")
|
|
|
|
@app.on_event("shutdown")
|
|
def shutdown_scheduler():
|
|
scheduler.shutdown(wait=False)
|
|
print("Cache cleanup scheduler stopped")
|
|
|
|
async def wait_for_network_idle(page):
|
|
"""Wait for network to be idle"""
|
|
await page.wait_for_load_state('networkidle')
|
|
|
|
@app.head("/")
|
|
async def health_check():
|
|
return {"status": "healthy"}
|
|
|
|
async def safe_browser_operation(url, operation_func):
|
|
"""Safely perform a browser operation with proper resource management and timeouts"""
|
|
async with operation_semaphore:
|
|
browser = None
|
|
context = None
|
|
page = None
|
|
|
|
try:
|
|
# Get browser from pool or create new one with timeout
|
|
try:
|
|
browser = await asyncio.wait_for(browser_pool.get(), timeout=5.0)
|
|
except asyncio.TimeoutError:
|
|
print("Timeout getting browser from pool, creating new one")
|
|
browser = await asyncio.wait_for(create_browser(), timeout=30.0)
|
|
|
|
# Create stealth context with Cloudflare bypass
|
|
context = await asyncio.wait_for(
|
|
CloudflareBypass.setup_stealth_context(browser),
|
|
timeout=10.0
|
|
)
|
|
|
|
page = await asyncio.wait_for(context.new_page(), timeout=10.0)
|
|
page.set_default_timeout(PAGE_TIMEOUT * 1000) # Convert to milliseconds
|
|
|
|
# Track page creation time for force cleanup
|
|
page_creation_times[page] = time.time()
|
|
active_pages.add(page)
|
|
|
|
# Setup stealth page with additional measures
|
|
await asyncio.wait_for(
|
|
CloudflareBypass.setup_stealth_page(page),
|
|
timeout=5.0
|
|
)
|
|
|
|
# Set up optimized request interception for better performance
|
|
await page.route("**/*", lambda route: route.abort()
|
|
if route.request.resource_type in ['image', 'stylesheet', 'font', 'media', 'script']
|
|
and not route.request.url.startswith('data:')
|
|
else route.continue_())
|
|
|
|
# Perform the operation with timeout
|
|
result = await asyncio.wait_for(
|
|
operation_func(page),
|
|
timeout=BROWSER_OPERATION_TIMEOUT
|
|
)
|
|
return result
|
|
|
|
except asyncio.TimeoutError as e:
|
|
print(f"Browser operation timeout for {url}: {e}")
|
|
raise HTTPException(status_code=408, detail=f"Operation timeout: {str(e)}")
|
|
except Exception as e:
|
|
print(f"Error in browser operation for {url}: {e}")
|
|
raise
|
|
finally:
|
|
# Cleanup with timeouts
|
|
if page:
|
|
try:
|
|
# Remove from tracking
|
|
if page in active_pages:
|
|
active_pages.remove(page)
|
|
if page in page_creation_times:
|
|
del page_creation_times[page]
|
|
await asyncio.wait_for(page.close(), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception):
|
|
try:
|
|
await page.close(force=True)
|
|
except:
|
|
pass
|
|
if context:
|
|
try:
|
|
await asyncio.wait_for(context.close(), timeout=5.0)
|
|
except (asyncio.TimeoutError, Exception):
|
|
pass
|
|
if browser:
|
|
try:
|
|
# Return browser to pool if it's still healthy and pool isn't full
|
|
if not browser_pool.full():
|
|
# Quick health check before returning to pool
|
|
try:
|
|
contexts = browser.contexts
|
|
if contexts:
|
|
await asyncio.wait_for(browser_pool.put(browser), timeout=2.0)
|
|
else:
|
|
await cleanup_browser(browser)
|
|
except (asyncio.TimeoutError, Exception):
|
|
await cleanup_browser(browser)
|
|
else:
|
|
await cleanup_browser(browser)
|
|
except Exception as e:
|
|
print(f"Error returning browser to pool: {e}")
|
|
try:
|
|
await cleanup_browser(browser)
|
|
except:
|
|
pass
|
|
|
|
@app.get("/")
|
|
async def visit_url(url: str, x_api_key: Optional[str] = Header(None)):
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
# Decode URL if it's encoded
|
|
decoded_url = unquote(url)
|
|
|
|
# Check cache first
|
|
cached_result = get_cached_data(decoded_url, "visit")
|
|
if cached_result:
|
|
return cached_result
|
|
|
|
try:
|
|
async def visit_operation(page):
|
|
try:
|
|
# Use Cloudflare bypass navigation
|
|
success = await CloudflareBypass.navigate_with_stealth(page, decoded_url)
|
|
|
|
if not success:
|
|
return {"status": "error", "error": "Failed to bypass Cloudflare protection", "url": decoded_url}
|
|
|
|
# Get page content
|
|
content = await page.content()
|
|
return {"status": "success", "content": content}
|
|
except Exception as e:
|
|
print(f"Error during page navigation: {e}")
|
|
# Try to get content anyway
|
|
try:
|
|
content = await page.content()
|
|
return {"status": "partial", "content": content, "error": str(e)}
|
|
except:
|
|
raise Exception(f"Failed to get page content: {str(e)}")
|
|
|
|
result = await safe_browser_operation(decoded_url, visit_operation)
|
|
save_to_cache(decoded_url, "visit", result)
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error visiting URL {decoded_url}: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/seo")
|
|
async def extract_seo(url: str, x_api_key: Optional[str] = Header(None)):
|
|
"""Extract SEO information from a website"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
# Decode URL if it's encoded
|
|
decoded_url = unquote(url)
|
|
|
|
# Check cache first
|
|
cached_result = get_cached_data(decoded_url, "seo")
|
|
if cached_result:
|
|
return cached_result
|
|
|
|
try:
|
|
async def seo_operation(page):
|
|
try:
|
|
# Use Cloudflare bypass navigation
|
|
success = await CloudflareBypass.navigate_with_stealth(page, decoded_url)
|
|
|
|
if not success:
|
|
return {"status": "error", "error": "Failed to bypass Cloudflare protection", "url": decoded_url}
|
|
|
|
# Extract SEO information
|
|
seo_data = await page.evaluate('''() => {
|
|
const data = {
|
|
title: document.title || '',
|
|
description: '',
|
|
canonical: '',
|
|
h1: [],
|
|
h2: [],
|
|
images: 0,
|
|
links: 0
|
|
};
|
|
|
|
// Get meta description
|
|
const metaDescription = document.querySelector('meta[name="description"]');
|
|
if (metaDescription) {
|
|
data.description = metaDescription.getAttribute('content') || '';
|
|
}
|
|
|
|
// Get canonical link
|
|
const canonicalLink = document.querySelector('link[rel="canonical"]');
|
|
if (canonicalLink) {
|
|
data.canonical = canonicalLink.getAttribute('href') || '';
|
|
}
|
|
|
|
// Get h1 tags
|
|
document.querySelectorAll('h1').forEach(h1 => {
|
|
const text = h1.innerText.trim();
|
|
if (text) data.h1.push(text);
|
|
});
|
|
|
|
// Get h2 tags
|
|
document.querySelectorAll('h2').forEach(h2 => {
|
|
const text = h2.innerText.trim();
|
|
if (text) data.h2.push(text);
|
|
});
|
|
|
|
// Count images
|
|
data.images = document.querySelectorAll('img').length;
|
|
|
|
// Count links
|
|
data.links = document.querySelectorAll('a').length;
|
|
|
|
return data;
|
|
}''')
|
|
|
|
result = {
|
|
"status": "success",
|
|
"url": decoded_url,
|
|
"seo": seo_data
|
|
}
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error during SEO extraction: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
result = await safe_browser_operation(decoded_url, seo_operation)
|
|
save_to_cache(decoded_url, "seo", result)
|
|
return result
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/meta")
|
|
async def extract_meta_tags(url: str, x_api_key: Optional[str] = Header(None)):
|
|
"""Extract meta tags from a website"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
# Decode URL if it's encoded
|
|
decoded_url = unquote(url)
|
|
|
|
# Check cache first
|
|
cached_result = get_cached_data(decoded_url, "meta")
|
|
if cached_result:
|
|
return cached_result
|
|
|
|
try:
|
|
async def meta_operation(page):
|
|
try:
|
|
# Use Cloudflare bypass navigation
|
|
success = await CloudflareBypass.navigate_with_stealth(page, decoded_url)
|
|
|
|
if not success:
|
|
return {"status": "error", "error": "Failed to bypass Cloudflare protection", "url": decoded_url}
|
|
|
|
# Extract meta tags using Playwright
|
|
meta_data = await page.evaluate('''() => {
|
|
const data = {
|
|
meta_tags: [],
|
|
open_graph: {},
|
|
twitter_card: {},
|
|
title: document.title || ''
|
|
};
|
|
|
|
// Extract all meta tags
|
|
document.querySelectorAll('meta').forEach(meta => {
|
|
const attributes = {};
|
|
for (let attr of meta.attributes) {
|
|
attributes[attr.name] = attr.value;
|
|
}
|
|
data.meta_tags.push(attributes);
|
|
});
|
|
|
|
// Extract Open Graph tags
|
|
document.querySelectorAll('meta[property^="og:"]').forEach(meta => {
|
|
data.open_graph[meta.getAttribute('property')] = meta.getAttribute('content');
|
|
});
|
|
|
|
// Extract Twitter card tags
|
|
document.querySelectorAll('meta[name^="twitter:"]').forEach(meta => {
|
|
data.twitter_card[meta.getAttribute('name')] = meta.getAttribute('content');
|
|
});
|
|
|
|
return data;
|
|
}''')
|
|
|
|
result = {
|
|
"status": "success",
|
|
"url": decoded_url,
|
|
"meta_tags": meta_data['meta_tags'],
|
|
"open_graph": meta_data['open_graph'],
|
|
"twitter_card": meta_data['twitter_card'],
|
|
"title": meta_data['title']
|
|
}
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error during meta tag extraction: {e}")
|
|
return {"status": "error", "url": decoded_url, "error": str(e)}
|
|
|
|
result = await safe_browser_operation(decoded_url, meta_operation)
|
|
save_to_cache(decoded_url, "meta", result)
|
|
return result
|
|
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/cache/clear")
|
|
async def clear_cache(x_api_key: Optional[str] = Header(None)):
|
|
"""Clear all cached data"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
try:
|
|
conn = sqlite3.connect('/db/cache.db')
|
|
cursor = conn.cursor()
|
|
cursor.execute('DELETE FROM cache')
|
|
deleted_count = cursor.rowcount
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
return {"status": "success", "message": f"Cleared {deleted_count} cache entries"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/cache/stats")
|
|
async def cache_stats(x_api_key: Optional[str] = Header(None)):
|
|
"""Get cache statistics"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
try:
|
|
conn = sqlite3.connect('/db/cache.db')
|
|
cursor = conn.cursor()
|
|
|
|
# Get total count
|
|
cursor.execute('SELECT COUNT(*) FROM cache')
|
|
total_count = cursor.fetchone()[0]
|
|
|
|
# Get count by route
|
|
cursor.execute('''
|
|
SELECT route, COUNT(*) as count
|
|
FROM cache
|
|
GROUP BY route
|
|
''')
|
|
route_counts = dict(cursor.fetchall())
|
|
|
|
# Get oldest and newest entries
|
|
cursor.execute('''
|
|
SELECT MIN(created_at), MAX(created_at)
|
|
FROM cache
|
|
''')
|
|
oldest, newest = cursor.fetchone()
|
|
|
|
# Get database size
|
|
cursor.execute('PRAGMA page_count')
|
|
page_count = cursor.fetchone()[0]
|
|
cursor.execute('PRAGMA page_size')
|
|
page_size = cursor.fetchone()[0]
|
|
db_size = page_count * page_size
|
|
|
|
conn.close()
|
|
|
|
return {
|
|
"status": "success",
|
|
"total_entries": total_count,
|
|
"route_counts": route_counts,
|
|
"oldest_entry": oldest,
|
|
"newest_entry": newest,
|
|
"database_size_bytes": db_size,
|
|
"cache_expiry_hours": CACHE_EXPIRY_HOURS
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/status")
|
|
async def system_status(x_api_key: Optional[str] = Header(None)):
|
|
"""Get system status and health information"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
try:
|
|
# Get system information
|
|
cpu_percent = psutil.cpu_percent(interval=1)
|
|
memory = psutil.virtual_memory()
|
|
disk = psutil.disk_usage('/')
|
|
|
|
# Get browser pool information
|
|
pool_size = browser_pool.qsize()
|
|
active_browser_count = len(active_browsers)
|
|
active_page_count = len(active_pages)
|
|
|
|
# Get cache statistics
|
|
conn = sqlite3.connect('/db/cache.db')
|
|
cursor = conn.cursor()
|
|
cursor.execute('SELECT COUNT(*) FROM cache')
|
|
cache_count = cursor.fetchone()[0]
|
|
conn.close()
|
|
|
|
return {
|
|
"status": "success",
|
|
"system": {
|
|
"cpu_percent": cpu_percent,
|
|
"memory_percent": memory.percent,
|
|
"memory_available_gb": round(memory.available / (1024**3), 2),
|
|
"memory_used_gb": round(memory.used / (1024**3), 2),
|
|
"memory_free_gb": round(memory.free / (1024**3), 2),
|
|
"memory_total_gb": round(memory.total / (1024**3), 2),
|
|
"disk_percent": disk.percent,
|
|
"disk_free_gb": round(disk.free / (1024**3), 2),
|
|
"disk_used_gb": round(disk.used / (1024**3), 2),
|
|
"disk_total_gb": round(disk.total / (1024**3), 2)
|
|
},
|
|
"browser_pool": {
|
|
"pool_size": pool_size,
|
|
"max_browsers": MAX_BROWSERS,
|
|
"active_browsers": active_browser_count,
|
|
"active_pages": active_page_count,
|
|
"browser_ttl_seconds": BROWSER_TTL,
|
|
"instance_timeout_minutes": BROWSER_INSTANCE_TIMEOUT_MINUTES,
|
|
"max_concurrent_operations": MAX_CONCURRENT_OPERATIONS,
|
|
"browser_operation_timeout_seconds": BROWSER_OPERATION_TIMEOUT,
|
|
"page_timeout_seconds": PAGE_TIMEOUT,
|
|
"pool_health": "healthy" if pool_size > 0 and active_browser_count <= MAX_BROWSERS else "warning"
|
|
},
|
|
"cache": {
|
|
"total_entries": cache_count,
|
|
"expiry_hours": CACHE_EXPIRY_HOURS
|
|
},
|
|
"rate_limiting": {
|
|
"requests_per_minute": RATE_LIMIT_MINUTE,
|
|
"window_seconds": RATE_LIMIT_WINDOW
|
|
},
|
|
"performance": {
|
|
"memory_pressure": "high" if memory.percent > 85 else "normal" if memory.percent > 70 else "low",
|
|
"cpu_pressure": "high" if cpu_percent > 80 else "normal" if cpu_percent > 50 else "low"
|
|
}
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.post("/emergency-cleanup")
|
|
async def emergency_cleanup(x_api_key: Optional[str] = Header(None)):
|
|
"""Emergency cleanup endpoint to force cleanup all browsers"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
try:
|
|
await force_cleanup_all_browsers()
|
|
return {"status": "success", "message": "Emergency cleanup completed"}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.post("/force-cleanup-old")
|
|
async def force_cleanup_old(x_api_key: Optional[str] = Header(None)):
|
|
"""Force cleanup old browser and page instances based on timeout"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
try:
|
|
await force_cleanup_old_instances()
|
|
return {
|
|
"status": "success",
|
|
"message": f"Force cleanup of old instances completed (timeout: {BROWSER_INSTANCE_TIMEOUT_MINUTES} minutes)"
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.get("/test-cloudflare")
|
|
async def test_cloudflare_bypass(url: str, x_api_key: Optional[str] = Header(None)):
|
|
"""Test Cloudflare bypass functionality on a specific URL"""
|
|
# Validate API key
|
|
if not x_api_key or x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
# Decode URL if it's encoded
|
|
decoded_url = unquote(url)
|
|
|
|
try:
|
|
async def test_operation(page):
|
|
try:
|
|
# Use Cloudflare bypass navigation
|
|
success = await CloudflareBypass.navigate_with_stealth(page, decoded_url)
|
|
|
|
if not success:
|
|
return {
|
|
"status": "error",
|
|
"error": "Failed to bypass Cloudflare protection",
|
|
"url": decoded_url,
|
|
"cloudflare_detected": True
|
|
}
|
|
|
|
# Get page information
|
|
title = await page.title()
|
|
url_after_navigation = page.url
|
|
|
|
# Check for Cloudflare indicators
|
|
cloudflare_indicators = await page.evaluate('''() => {
|
|
const indicators = {
|
|
has_cloudflare_title: document.title.toLowerCase().includes('cloudflare'),
|
|
has_challenge_form: !!document.querySelector('#challenge-form'),
|
|
has_cf_wrapper: !!document.querySelector('#cf-wrapper'),
|
|
has_please_wait: !!document.querySelector('#cf-please-wait'),
|
|
has_browser_verification: !!document.querySelector('.cf-browser-verification')
|
|
};
|
|
return indicators;
|
|
}''')
|
|
|
|
return {
|
|
"status": "success",
|
|
"url": decoded_url,
|
|
"final_url": url_after_navigation,
|
|
"title": title,
|
|
"cloudflare_bypassed": True,
|
|
"cloudflare_indicators": cloudflare_indicators,
|
|
"content_length": len(await page.content())
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error during Cloudflare test: {e}")
|
|
return {
|
|
"status": "error",
|
|
"error": str(e),
|
|
"url": decoded_url,
|
|
"cloudflare_detected": False
|
|
}
|
|
|
|
result = await safe_browser_operation(decoded_url, test_operation)
|
|
return result
|
|
|
|
except Exception as e:
|
|
print(f"Error testing Cloudflare bypass for {decoded_url}: {e}")
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@asynccontextmanager
|
|
async def get_browser():
|
|
"""Context manager for getting a browser from the pool"""
|
|
browser = None
|
|
try:
|
|
browser = await browser_pool.get()
|
|
yield browser
|
|
finally:
|
|
if browser:
|
|
await browser_pool.put(browser)
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|