app.py

import gradio as gr
import requests
from bs4 import BeautifulSoup
import re
from urllib.parse import urljoin, urlparse
import asyncio
import aiohttp
from collections import defaultdict
import unicodedata
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class WebsiteCrawler:
    def __init__(self, max_depth=3, max_pages=50):
        self.max_depth = max_depth
        self.max_pages = max_pages
        self.visited_urls = set()
        self.url_metadata = defaultdict(dict)
        self.homepage_metadata = None
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
        }

    def determine_category_importance(self, url, title, desc):
        url_lower = url.lower()
        path = urlparse(url).path.lower()

        if path == "/" or path == "":
            return "Main", 10

        if any(x in url_lower for x in ["/docs", "/faq", "/help"]):
            return "Documentation", 8

        elif any(x in url_lower for x in ["/api", "/developer"]):
            return "API", 8

        elif any(x in url_lower for x in ["/about", "/company", "/contact"]):
            return "About", 7

        elif any(x in url_lower for x in ["/news", "/blog", "/events"]):
            return "News", 5

        elif any(x in url_lower for x in ["/tools", "/pricing"]):
            return "Tools", 6

        return "Optional", 1

    def clean_text(self, text, is_title=False):
        if not text:
            return ""
        text = unicodedata.normalize("NFKD", text)
        text = re.sub(r"[^\x00-\x7F]+", "", text)
        text = " ".join(text.split()).strip()

        if is_title:
            text = re.sub(r"^\s*Welcome to\s+", "", text)
        return text

    async def crawl_page(self, url, depth, base_domain):
        if (
            depth > self.max_depth
            or url in self.visited_urls
            or len(self.visited_urls) >= self.max_pages
        ):
            return []

        try:
            async with aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=20)
            ) as session:
                async with session.get(
                    url, headers=self.headers, allow_redirects=True
                ) as response:
                    if response.status != 200:
                        return []
                    text = await response.text()
                    self.visited_urls.add(url)

                    soup = BeautifulSoup(text, "html.parser")
                    title_tag = soup.find("title")
                    title = (
                        self.clean_text(title_tag.text)
                        if title_tag
                        else url.split("/")[-1]
                    )

                    desc_tag = soup.find("meta", {"name": "description"})
                    desc = (
                        self.clean_text(desc_tag["content"])
                        if desc_tag and desc_tag.get("content")
                        else ""
                    )

                    category, importance = self.determine_category_importance(
                        url, title, desc
                    )

                    self.url_metadata[url] = {
                        "title": title,
                        "description": desc,
                        "category": category,
                        "importance": importance,
                    }

                    links = []
                    for a in soup.find_all("a", href=True):
                        next_url = urljoin(url, a["href"])
                        if urlparse(next_url).netloc == base_domain:
                            links.append(next_url)

                    return links
        except Exception as e:
            logger.error(f"Error crawling {url}: {str(e)}")
            return []

    async def process_homepage(self, url):
        try:
            async with aiohttp.ClientSession(
                timeout=aiohttp.ClientTimeout(total=20)
            ) as session:
                async with session.get(
                    url, headers=self.headers, allow_redirects=True
                ) as response:
                    if response.status != 200:
                        return
                    text = await response.text()
                    soup = BeautifulSoup(text, "html.parser")

                    site_name = (
                        soup.find("title").text.split("|")[0].strip()
                        if soup.find("title")
                        else urlparse(url).netloc
                    )
                    description = soup.find("meta", {"name": "description"})
                    description = (
                        description["content"].strip()
                        if description and description.get("content")
                        else None
                    )

                    self.homepage_metadata = {
                        "site_name": self.clean_text(site_name, is_title=True),
                        "description": (
                            self.clean_text(description) if description else None
                        ),
                    }
        except Exception as e:
            logger.error(f"Error processing homepage {url}: {str(e)}")

    async def crawl_website(self, start_url):
        try:
            await self.process_homepage(start_url)
            base_domain = urlparse(start_url).netloc
            queue = [(start_url, 0)]
            seen = {start_url}

            while queue and len(self.visited_urls) < self.max_pages:
                current_url, depth = queue.pop(0)
                if depth > self.max_depth:
                    continue
                links = await self.crawl_page(current_url, depth, base_domain)
                for link in links:
                    if link not in seen:
                        seen.add(link)
                        queue.append((link, depth + 1))

        except Exception as e:
            logger.error(f"Error during crawl: {str(e)}")
            raise

    def generate_llms_txt(self):
        if not self.url_metadata:
            return "No content available."

        content = []
        homepage_title = self.homepage_metadata.get("site_name", "Website")
        homepage_description = self.homepage_metadata.get(
            "description", "No description available."
        )
        content.append(f"# {homepage_title}\n\n> {homepage_description}\n")

        categories = defaultdict(list)
        for url, metadata in self.url_metadata.items():
            categories[metadata["category"]].append((url, metadata))

        category_order = [
            "Main",
            "Documentation",
            "API",
            "About",
            "News",
            "Tools",
            "Optional",
        ]
        for category in category_order:
            if category in categories:
                content.append(f"## {category}")
                for url, metadata in categories[category]:
                    content.append(
                        f"- [{metadata['title']}]({url}): {metadata['description']}"
                    )

        return "\n".join(content)


async def process_url(url, max_depth, max_pages):
    try:
        if not url.startswith(("http://", "https://")):
            url = "https://" + url
        result = urlparse(url)
        if not result.scheme or not result.netloc:
            return "", "Invalid URL format. Please enter a valid URL."

        crawler = WebsiteCrawler(max_depth=int(max_depth), max_pages=int(max_pages))
        await crawler.crawl_website(url)
        content = crawler.generate_llms_txt()

        return content, f"Successfully crawled {len(crawler.visited_urls)} pages."
    except Exception as e:
        logger.error(f"Error processing URL {url}: {str(e)}")
        return "", f"Error: {str(e)}"


# Gradio interface
theme = gr.themes.Soft(primary_hue="blue", font="Open Sans")

with gr.Blocks(theme=theme) as iface:
    with gr.Row():
        gr.Markdown("## Website Crawler - Generate llms.txt")

    with gr.Row():
        url_input = gr.Textbox(
            label="Website URL",
            placeholder="Enter the website URL (e.g., example.com)",
            info="The URL will be automatically prefixed with https:// if not provided",
            lines=1,
        )

    with gr.Row():
        depth_input = gr.Slider(
            minimum=1, maximum=5, value=3, step=1, label="Maximum Crawl Depth"
        )
        pages_input = gr.Slider(
            minimum=10, maximum=100, value=50, step=10, label="Maximum Pages"
        )

    with gr.Row():
        generate_btn = gr.Button("Generate llms.txt", variant="primary")

    with gr.Row():
        output = gr.Textbox(
            label="Generated llms.txt Content",
            lines=15,
            show_copy_button=True,
            container=True,
        )

    with gr.Row():
        status = gr.Textbox(label="Status", interactive=False)

    # Pass the asynchronous function directly
    generate_btn.click(
        fn=process_url,
        inputs=[url_input, depth_input, pages_input],
        outputs=[output, status],
    )

if __name__ == "__main__":
    iface.launch(asyncio_mode="auto")