""" Base Crawler - Recursive web crawler with polite crawling. Handles rate-limiting, retries, and content extraction. """ import os import re import json import time import hashlib import logging from urllib.parse import urljoin, urlparse from typing import Optional import requests from bs4 import BeautifulSoup import trafilatura from scraper.config import ( USER_AGENT, CRAWL_DELAY, MAX_PAGES_PER_DOMAIN, REQUEST_TIMEOUT, MAX_RETRIES, ) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s" ) logger = logging.getLogger("BaseCrawler") class BaseCrawler: """ Recursive web crawler that: - Respects crawl delays - Extracts clean text via trafilatura - Follows internal links recursively - Saves results as JSON files """ def __init__(self, name: str, base_url: str, start_urls: list, allowed_domains: list, output_dir: str, max_pages: int = MAX_PAGES_PER_DOMAIN, crawl_delay: float = CRAWL_DELAY): self.name = name self.base_url = base_url self.start_urls = start_urls self.allowed_domains = allowed_domains self.output_dir = output_dir self.max_pages = max_pages self.crawl_delay = crawl_delay self.visited_urls: set = set() self.to_visit: list = list(start_urls) self.results: list = [] self.session = self._create_session() os.makedirs(output_dir, exist_ok=True) def _create_session(self) -> requests.Session: session = requests.Session() session.headers.update({ "User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "sq,en;q=0.5", "Accept-Encoding": "gzip, deflate", "Connection": "keep-alive", }) return session def _is_allowed(self, url: str) -> bool: """Check if URL belongs to allowed domains.""" parsed = urlparse(url) domain = parsed.netloc.lower().replace("www.", "") return any( domain == d.lower().replace("www.", "") or domain.endswith("." + d.lower().replace("www.", "")) for d in self.allowed_domains ) def _is_valid_page(self, url: str) -> bool: """Filter out non-content URLs.""" skip_extensions = { '.jpg', '.jpeg', '.png', '.gif', '.svg', '.ico', '.bmp', '.webp', '.mp3', '.mp4', '.avi', '.mov', '.wmv', '.wav', '.zip', '.rar', '.tar', '.gz', '.css', '.js', '.json', '.xml', '.exe', '.msi', '.dll', } parsed = urlparse(url) path = parsed.path.lower() # Allow PDFs (we handle them separately) if path.endswith('.pdf'): return True return not any(path.endswith(ext) for ext in skip_extensions) def _normalize_url(self, url: str) -> str: """Normalize URL for deduplication.""" parsed = urlparse(url) # Remove fragment and trailing slash normalized = f"{parsed.scheme}://{parsed.netloc}{parsed.path.rstrip('/')}" if parsed.query: normalized += f"?{parsed.query}" return normalized def _url_to_filename(self, url: str) -> str: """Convert URL to safe filename.""" url_hash = hashlib.md5(url.encode()).hexdigest()[:10] parsed = urlparse(url) path = parsed.path.strip("/").replace("/", "_")[:80] if not path: path = "index" return f"{path}_{url_hash}.json" def fetch_page(self, url: str) -> Optional[str]: """Fetch a page with retries.""" for attempt in range(MAX_RETRIES): try: response = self.session.get( url, timeout=REQUEST_TIMEOUT, allow_redirects=True ) response.raise_for_status() # Check content type content_type = response.headers.get("Content-Type", "") if "text/html" in content_type or "application/xhtml" in content_type: response.encoding = response.apparent_encoding or "utf-8" return response.text elif "application/pdf" in content_type: # Save PDF separately self._save_pdf(url, response.content) return None return response.text except requests.RequestException as e: logger.warning(f"Attempt {attempt + 1}/{MAX_RETRIES} failed for {url}: {e}") if attempt < MAX_RETRIES - 1: time.sleep(self.crawl_delay * (attempt + 1)) logger.error(f"Failed to fetch: {url}") return None def _save_pdf(self, url: str, content: bytes): """Save downloaded PDF to pdfs directory.""" from scraper.config import RAW_SUBDIRS pdf_dir = RAW_SUBDIRS.get("pdfs", os.path.join(self.output_dir, "pdfs")) os.makedirs(pdf_dir, exist_ok=True) filename = self._url_to_filename(url).replace(".json", ".pdf") filepath = os.path.join(pdf_dir, filename) with open(filepath, "wb") as f: f.write(content) logger.info(f"Saved PDF: {filepath}") def extract_text(self, html: str, url: str) -> dict: """Extract clean text from HTML using trafilatura.""" # Primary: trafilatura (best quality) text = trafilatura.extract( html, include_comments=False, include_tables=True, include_links=False, include_images=False, favor_precision=False, favor_recall=True, url=url, ) # Fallback: BeautifulSoup if not text or len(text.strip()) < 50: soup = BeautifulSoup(html, "lxml") # Remove script, style, nav, footer for tag in soup(["script", "style", "nav", "footer", "header", "aside"]): tag.decompose() text = soup.get_text(separator="\n", strip=True) # Extract title soup = BeautifulSoup(html, "lxml") title_tag = soup.find("title") title = title_tag.get_text(strip=True) if title_tag else "" # Extract meta description meta_desc = "" meta = soup.find("meta", attrs={"name": "description"}) if meta: meta_desc = meta.get("content", "") return { "url": url, "title": title, "meta_description": meta_desc, "text": text or "", "text_length": len(text) if text else 0, "source": self.name, "crawled_at": time.strftime("%Y-%m-%dT%H:%M:%S"), } def extract_links(self, html: str, current_url: str) -> list: """Extract all internal links from page.""" soup = BeautifulSoup(html, "lxml") links = [] for a_tag in soup.find_all("a", href=True): href = a_tag["href"].strip() # Skip anchors, javascript, mailto if href.startswith(("#", "javascript:", "mailto:", "tel:")): continue # Make absolute absolute_url = urljoin(current_url, href) normalized = self._normalize_url(absolute_url) if (self._is_allowed(normalized) and self._is_valid_page(normalized) and normalized not in self.visited_urls): links.append(normalized) return list(set(links)) def save_result(self, data: dict): """Save extracted data to JSON file.""" filename = self._url_to_filename(data["url"]) filepath = os.path.join(self.output_dir, filename) with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def crawl(self): """Main crawling loop.""" logger.info(f"🚀 Starting crawl: {self.name}") logger.info(f" Start URLs: {len(self.start_urls)}") logger.info(f" Max pages: {self.max_pages}") page_count = 0 while self.to_visit and page_count < self.max_pages: # Sort to_visit to prioritize news/recency if they appear in URL # High priority keywords: lajme, news, njoftime, press-release, 2025, 2026 self.to_visit.sort(key=lambda u: any(k in u.lower() for k in ["lajme", "news", "njoftime", "press", "2025", "2026"]), reverse=True) url = self.to_visit.pop(0) normalized = self._normalize_url(url) if normalized in self.visited_urls: continue self.visited_urls.add(normalized) logger.info(f"[{page_count + 1}/{self.max_pages}] Crawling: {normalized}") # Fetch html = self.fetch_page(normalized) if not html: continue # Extract text data = self.extract_text(html, normalized) # Only save if there's meaningful content if data["text_length"] > 100: self.save_result(data) self.results.append(data) page_count += 1 logger.info(f" ✅ Saved: {data['title'][:60]} ({data['text_length']} chars)") else: logger.info(f" ⏭️ Skipped (too short): {normalized}") # Extract links for recursive crawling new_links = self.extract_links(html, normalized) self.to_visit.extend(new_links) # Polite delay time.sleep(self.crawl_delay) logger.info(f"✅ Crawl complete: {self.name}") logger.info(f" Pages crawled: {page_count}") logger.info(f" Total results: {len(self.results)}") return self.results def get_stats(self) -> dict: """Return crawling statistics.""" return { "source": self.name, "pages_crawled": len(self.results), "urls_visited": len(self.visited_urls), "total_chars": sum(r["text_length"] for r in self.results), }