Spaces:
Runtime error
Runtime error
| import requests | |
| from bs4 import BeautifulSoup | |
| from urllib.parse import urlparse, urljoin | |
| import logging | |
| from src.document_processing.chunking import chunk_text | |
| from typing import List, Optional, Set, Dict, Callable | |
| from langchain_community.embeddings import HuggingFaceEmbeddings | |
| from langchain_community.vectorstores import FAISS | |
| from langchain_core.documents.base import Document | |
| logger = logging.getLogger(__name__) | |
| class WebScraper: | |
| def __init__(self, headers: Optional[dict] = None): | |
| self.headers = headers or { | |
| "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" | |
| } | |
| self.visited_urls = set() | |
| def scrape_url(self, url: str) -> str: | |
| try: | |
| response = requests.get(url, headers=self.headers) | |
| response.raise_for_status() | |
| soup = BeautifulSoup(response.content, "html.parser") | |
| for script in soup(["script", "style", "nav", "footer", "header"]): | |
| script.extract() | |
| text = soup.get_text(separator='\n') | |
| lines = (line.strip() for line in text.splitlines()) | |
| chunks = (phrase.strip() for line in lines for phrase in line.split(" ")) | |
| text = '\n'.join(chunk for chunk in chunks if chunk) | |
| return text, soup | |
| except Exception as e: | |
| logger.error(f"Error scraping {url}: {str(e)}") | |
| return f"Error scraping {url}: {str(e)}", None | |
| def extract_links(self, url: str, soup: BeautifulSoup) -> List[str]: | |
| """Extract all hyperlinks from a webpage""" | |
| links = [] | |
| base_url = urlparse(url) | |
| domain = base_url.netloc | |
| for link in soup.find_all('a', href=True): | |
| href = link['href'] | |
| # Create absolute URL | |
| if href.startswith('/'): | |
| full_url = f"{base_url.scheme}://{domain}{href}" | |
| elif href.startswith(('http://', 'https://')): | |
| full_url = href | |
| else: | |
| full_url = urljoin(url, href) | |
| # Only include links from same domain | |
| if urlparse(full_url).netloc == domain: | |
| links.append(full_url) | |
| return links | |
| def get_domain(self, url: str) -> str: | |
| return urlparse(url).netloc | |
| def crawl_website(self, start_url: str, max_depth: int = 2, max_pages: int = 20, | |
| callback: Optional[Callable[[str], None]] = None) -> Dict[str, str]: | |
| self.visited_urls = set() | |
| results = {} | |
| def _crawl(url: str, depth: int): | |
| if depth > max_depth or len(results) >= max_pages or url in self.visited_urls: | |
| return | |
| self.visited_urls.add(url) | |
| if callback: | |
| callback(url) | |
| logger.info(f"Scraping {url} (depth {depth})") | |
| content, soup = self.scrape_url(url) | |
| if soup is not None: | |
| results[url] = content | |
| links = self.extract_links(url, soup) | |
| for link in links: | |
| _crawl(link, depth + 1) | |
| _crawl(start_url, 1) | |
| return results | |
| def process_web_content(urls: List[str], vector_store_path: str, | |
| max_depth: int = 1, max_pages: int = 20, | |
| ui_callback: Optional[Callable[[str], None]] = None) -> FAISS: | |
| scraper = WebScraper() | |
| documents = [] | |
| for start_url in urls: | |
| if ui_callback: | |
| ui_callback(f"Starting to crawl from: {start_url}") | |
| page_contents = scraper.crawl_website( | |
| start_url, | |
| max_depth=max_depth, | |
| max_pages=max_pages, | |
| callback=ui_callback | |
| ) | |
| for url, content in page_contents.items(): | |
| domain = scraper.get_domain(url) | |
| chunks = chunk_text(content) | |
| for i, chunk in enumerate(chunks): | |
| metadata = { | |
| "source": url, | |
| "domain": domain, | |
| "chunk": i | |
| } | |
| documents.append(Document(page_content=chunk, metadata=metadata)) | |
| if ui_callback: | |
| ui_callback(f"Creating vector embeddings for {len(documents)} chunks...") | |
| embeddings = HuggingFaceEmbeddings( | |
| model_name="sentence-transformers/all-MiniLM-L6-v2", | |
| model_kwargs={'device': 'cpu'} | |
| ) | |
| vector_store = FAISS.from_documents(documents, embeddings) | |
| vector_store.save_local(vector_store_path) | |
| return vector_store |