import os os.system("playwright install") import re import urllib.parse import asyncio from typing import Dict import gradio as gr from bs4 import BeautifulSoup, NavigableString from playwright.async_api import async_playwright # --- 1. GLOBAL RESOURCES & CONFIGURATION --- # This dictionary will hold the long-lived Playwright and Browser objects PLAYWRIGHT_STATE: Dict = {} # EXPANDED: A comprehensive list of search engines SEARCH_ENGINES = { # Original "DuckDuckGo": "https://duckduckgo.com/html/?q={query}", "Google": "https://www.google.com/search?q={query}", "Bing": "https://www.bing.com/search?q={query}", "Brave": "https://search.brave.com/search?q={query}", "Ecosia": "https://www.ecosia.org/search?q={query}", # 10 More Added "Yahoo": "https://search.yahoo.com/search?p={query}", "Startpage": "https://www.startpage.com/sp/search?q={query}", "Qwant": "https://www.qwant.com/?q={query}", "Swisscows": "https://swisscows.com/web?query={query}", "You.com": "https://you.com/search?q={query}", "SearXNG": "https://searx.be/search?q={query}", "MetaGer": "https://metager.org/meta/meta.ger-en?eingabe={query}", "Yandex": "https://yandex.com/search/?text={query}", "Baidu": "https://www.baidu.com/s?wd={query}", "Perplexity": "https://www.perplexity.ai/search?q={query}" } # --- 2. ADVANCED HTML-TO-MARKDOWN CONVERTER (Unchanged) --- class HTML_TO_MARKDOWN_CONVERTER: # ... [The class code is identical to the previous version and remains unchanged] ... def __init__(self, soup: BeautifulSoup, base_url: str): self.soup = soup self.base_url = base_url def _cleanup_html(self): selectors_to_remove = [ 'nav', 'footer', 'header', 'aside', 'form', 'script', 'style', 'svg', 'button', 'input', 'textarea', '[role="navigation"]', '[role="search"]', '[id*="comment"]', '[class*="comment-"]', '[id*="sidebar"]', '[class*="sidebar"]', '[id*="related"]', '[class*="related"]', '[id*="share"]', '[class*="share"]', '[id*="social"]', '[class*="social"]', '[id*="cookie"]', '[class*="cookie"]' ] for selector in selectors_to_remove: for element in self.soup.select(selector): element.decompose() def convert(self): self._cleanup_html() content_node = self.soup.find('main') or self.soup.find('article') or self.soup.find('body') if not content_node: return "Could not find main content." md = self._process_node(content_node) return re.sub(r'\n{3,}', '\n\n', md).strip() def _process_node(self, element): if isinstance(element, NavigableString): return re.sub(r'\s+', ' ', element.strip()) if element.name is None or not element.name: return '' inner_md = " ".join(self._process_node(child) for child in element.children).strip() if element.name in ['p', 'div', 'section']: return f"\n\n{inner_md}\n\n" if element.name == 'h1': return f"\n\n# {inner_md}\n\n" if element.name == 'h2': return f"\n\n## {inner_md}\n\n" if element.name == 'h3': return f"\n\n### {inner_md}\n\n" if element.name in ['h4', 'h5', 'h6']: return f"\n\n#### {inner_md}\n\n" if element.name == 'li': return f"* {inner_md}\n" if element.name in ['ul', 'ol']: return f"\n{inner_md}\n" if element.name == 'blockquote': return f"> {inner_md.replace(chr(10), chr(10) + '> ')}\n\n" if element.name == 'hr': return "\n\n---\n\n" if element.name == 'table': header = " | ".join(f"**{th.get_text(strip=True)}**" for th in element.select('thead th, tr th')) separator = " | ".join(['---'] * len(header.split('|'))) rows = [" | ".join(td.get_text(strip=True) for td in tr.find_all('td')) for tr in element.select('tbody tr')] return f"\n\n{header}\n{separator}\n" + "\n".join(rows) + "\n\n" if element.name == 'pre': return f"\n```\n{element.get_text(strip=True)}\n```\n\n" if element.name == 'code': return f"`{inner_md}`" if element.name in ['strong', 'b']: return f"**{inner_md}**" if element.name in ['em', 'i']: return f"*{inner_md}*" if element.name == 'a': href = element.get('href', '') full_href = urllib.parse.urljoin(self.base_url, href) return f"[{inner_md}]({full_href})" if element.name == 'img': src = element.get('src', '') alt = element.get('alt', 'Image').strip() full_src = urllib.parse.urljoin(self.base_url, src) return f"\n\n![{alt}]({full_src})\n\n" return inner_md # --- 3. CORE API FUNCTION --- async def initialize_playwright(): """Launches Playwright and browser instances if they don't already exist.""" if "playwright" not in PLAYWRIGHT_STATE: print("🚀 First request received, starting up Playwright...") p = await async_playwright().start() PLAYWRIGHT_STATE["playwright"] = p PLAYWRIGHT_STATE["chromium"] = await p.chromium.launch(headless=True) PLAYWRIGHT_STATE["firefox"] = await p.firefox.launch(headless=True) PLAYWRIGHT_STATE["webkit"] = await p.webkit.launch(headless=True) print("✅ Playwright and browsers are ready.") async def perform_web_browse(query: str, browser_name: str, search_engine: str): """ A stateless function that takes a query, browser, and search engine, then returns the parsed content of the resulting page. """ await initialize_playwright() # Determine if the query is a URL or a search term is_url = urllib.parse.urlparse(query).scheme in ['http', 'https'] if is_url: url = query else: search_url_template = SEARCH_ENGINES.get(search_engine) if not search_url_template: return {"error": f"Invalid search engine: '{search_engine}'. Please choose from the provided list."} url = search_url_template.format(query=urllib.parse.quote_plus(query)) browser_instance = PLAYWRIGHT_STATE.get(browser_name.lower()) if not browser_instance: return {"error": f"Invalid browser: '{browser_name}'. Use 'chromium', 'firefox', or 'webkit'."} context = await browser_instance.new_context(user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36') page = await context.new_page() try: print(f"Navigating to: {url} using {browser_name}...") await page.goto(url, wait_until='domcontentloaded', timeout=30000) final_url = page.url title = await page.title() or "No Title" print(f"Arrived at: {final_url}") html_content = await page.content() soup = BeautifulSoup(html_content, 'lxml') converter = HTML_TO_MARKDOWN_CONVERTER(soup, base_url=final_url) markdown_text = converter.convert() print("Content parsed successfully.") return { "status": "success", "query": query, "final_url": final_url, "page_title": title, "markdown_content": markdown_text, } except Exception as e: error_message = str(e).splitlines()[0] print(f"An error occurred: {error_message}") return {"status": "error", "query": query, "error_message": error_message} finally: if page: await page.close() if context: await context.close() print("Session context closed.") # --- 4. GRADIO INTERFACE & API LAUNCH --- with gr.Blocks(title="Web Browse API", theme=gr.themes.Soft()) as demo: gr.Markdown("# Web Browse API") gr.Markdown( "This interface exposes a stateless API endpoint (`/api/web_browse`) to fetch and parse web content." ) query_input = gr.Textbox( label="URL or Search Query", placeholder="e.g., https://openai.com or 'history of artificial intelligence'" ) with gr.Row(): browser_input = gr.Dropdown( label="Browser", choices=["firefox", "chromium", "webkit"], value="firefox", scale=1 ) search_engine_input = gr.Dropdown( label="Search Engine (for non-URL queries)", choices=sorted(list(SEARCH_ENGINES.keys())), value="DuckDuckGo", scale=2 ) submit_button = gr.Button("Browse", variant="primary") output_json = gr.JSON(label="API Result") submit_button.click( fn=perform_web_browse, inputs=[query_input, browser_input, search_engine_input], outputs=output_json, api_name="web_browse" # Creates the POST /api/web_browse endpoint ) if __name__ == "__main__": demo.launch()