from fastapi import FastAPI, HTTPException, Header from pyppeteer import launch import os import asyncio import json import re from typing import Optional, Dict, List, Any from urllib.parse import unquote 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") # 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' async def wait_for_network_idle(page): """Wait until no network requests are in flight""" await page.waitForNetworkIdle(idleTime=500, timeout=30000) @app.head("/") async def health_check(): return {"status": "ok"} @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") browser = None try: # Decode URL if it's encoded decoded_url = unquote(url) print(decoded_url) # Launch browser using installed Chrome browser = await launch( headless=True, executablePath='/usr/bin/google-chrome', args=['--no-sandbox', '--disable-setuid-sandbox'] ) # Create new page page = await browser.newPage() # Set custom user agent await page.setUserAgent(CUSTOM_USER_AGENT) # Navigate to URL and wait for network idle await page.goto(decoded_url, waitUntil='networkidle2') # Get page content content = await page.content() # Close page await page.close() return {"status": "success", "content": content} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: # Ensure browser is closed even if an error occurs if browser: await browser.close() @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") browser = None try: # Decode URL if it's encoded decoded_url = unquote(url) print(f"Extracting SEO from: {decoded_url}") # Launch browser using installed Chrome browser = await launch( headless=True, executablePath='/usr/bin/google-chrome', args=['--no-sandbox', '--disable-setuid-sandbox'] ) # Create new page page = await browser.newPage() # Set custom user agent await page.setUserAgent(CUSTOM_USER_AGENT) # Navigate to URL and wait for network idle await page.goto(decoded_url, waitUntil='networkidle2') # Extract JSON-LD data json_ld_data = await page.evaluate('''() => { const jsonLdScripts = Array.from(document.querySelectorAll('script[type="application/ld+json"]')); return jsonLdScripts.map(script => { try { return JSON.parse(script.textContent); } catch (e) { return { error: "Failed to parse JSON-LD", content: script.textContent }; } }); }''') # Extract meta tags meta_tags = await page.evaluate('''() => { const metaTags = Array.from(document.querySelectorAll('meta')); return metaTags.map(tag => { const attributes = Array.from(tag.attributes); const result = {}; attributes.forEach(attr => { result[attr.name] = attr.value; }); return result; }); }''') # Extract title title = await page.evaluate('document.title') # Extract canonical URL canonical_url = await page.evaluate('''() => { const link = document.querySelector('link[rel="canonical"]'); return link ? link.href : null; }''') # Extract Open Graph data og_data = await page.evaluate('''() => { const ogTags = Array.from(document.querySelectorAll('meta[property^="og:"]')); const result = {}; ogTags.forEach(tag => { const property = tag.getAttribute('property'); const content = tag.getAttribute('content'); result[property] = content; }); return result; }''') # Extract Twitter Card data twitter_data = await page.evaluate('''() => { const twitterTags = Array.from(document.querySelectorAll('meta[name^="twitter:"]')); const result = {}; twitterTags.forEach(tag => { const name = tag.getAttribute('name'); const content = tag.getAttribute('content'); result[name] = content; }); return result; }''') # Close page await page.close() # Compile all SEO data seo_data = { "title": title, "canonical_url": canonical_url, "json_ld": json_ld_data, "meta_tags": meta_tags, "open_graph": og_data, "twitter_card": twitter_data } return {"status": "success", "seo_data": seo_data} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: # Ensure browser is closed even if an error occurs if browser: await browser.close() @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") browser = None try: # Decode URL if it's encoded decoded_url = unquote(url) print(f"Extracting meta tags from: {decoded_url}") # Launch browser using installed Chrome browser = await launch( headless=True, executablePath='/usr/bin/google-chrome', args=['--no-sandbox', '--disable-setuid-sandbox'] ) # Create new page page = await browser.newPage() # Set custom user agent await page.setUserAgent(CUSTOM_USER_AGENT) # Navigate to URL and wait for network idle await page.goto(decoded_url, waitUntil='networkidle2') # Extract meta tags meta_tags = await page.evaluate('''() => { const metaTags = Array.from(document.querySelectorAll('meta')); return metaTags.map(tag => { const attributes = Array.from(tag.attributes); const result = {}; attributes.forEach(attr => { result[attr.name] = attr.value; }); return result; }); }''') # Extract title title = await page.evaluate('document.title') # Close page await page.close() return { "status": "success", "url": decoded_url, "title": title, "meta_tags": meta_tags } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) finally: # Ensure browser is closed even if an error occurs if browser: await browser.close() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)