import datetime import logging import feedparser from typing import List, Optional from bs4 import BeautifulSoup from domain.entities.article import Article from domain.interfaces import ScraperInterface import requests import time # Set a custom user agent for feedparser to bypass basic bot protections feedparser.USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36" logger = logging.getLogger(__name__) class RssScraper(ScraperInterface): def __init__(self, source_name: str, language: str, feed_url: str, category: str = "World"): self.source_name = source_name self.language = language self.feed_url = feed_url self.category = category def scrape(self) -> List[Article]: articles = [] try: feed = feedparser.parse(self.feed_url) # Check for network or parsing errors that feedparser hides if getattr(feed, 'bozo', 0) == 1 and hasattr(feed, 'bozo_exception'): logger.warning(f"Feed error for {self.feed_url}: {feed.bozo_exception}") if not feed.entries: return articles for entry in feed.entries[:10]: # Limit to top 10 per feed title = entry.get("title", "") link = entry.get("link", "") # Get summary, remove HTML tags summary_raw = entry.get("description", "") summary = BeautifulSoup(summary_raw, "html.parser").get_text()[:200] if not title or not link: continue # Try to extract image (from RSS or OpenGraph meta tags) image_url = self._extract_image(entry, link=link) # Parse date or default to now try: # feedparser parses date into struct_time published_at = datetime.datetime.fromtimestamp(time.mktime(entry.published_parsed)) if hasattr(entry, 'published_parsed') and entry.published_parsed else datetime.datetime.now() except Exception: published_at = datetime.datetime.now() article = Article( id=Article.generate_id_from_url(link), title=title, source=self.source_name, url=link, imageUrl=image_url, hasImage=True if image_url else False, summary=summary + "..." if len(summary) >= 200 else summary, publishedAt=published_at, language=self.language, category=self.category ) articles.append(article) except Exception as e: logger.error(f"Failed to scrape RSS {self.feed_url}: {e}") return articles def _extract_image(self, entry, link: str = None) -> Optional[str]: # 1. Check media_content if hasattr(entry, 'media_content') and entry.media_content: for media in entry.media_content: if media.get('url'): return media.get('url') # 2. Check media_thumbnail if hasattr(entry, 'media_thumbnail') and entry.media_thumbnail: return entry.media_thumbnail[0].get('url') # 3. Check enclosures if hasattr(entry, 'enclosures') and entry.enclosures: for enc in entry.enclosures: if 'image' in enc.get('type', '') and enc.get('href'): return enc.get('href') # 4. Check links for image type if hasattr(entry, 'links'): for l in entry.links: if 'image' in l.get('type', ''): return l.get('href') # 5. Check description/content HTML for tags for field in ['description', 'content']: if hasattr(entry, field): content_list = getattr(entry, field) raw_html = content_list[0].get('value', '') if isinstance(content_list, list) else content_list soup = BeautifulSoup(str(raw_html), "html.parser") img = soup.find('img') if img and img.get('src'): return img.get('src') # 6. Layer 2: OpenGraph Web Scraping Fallback (Fast 2s timeout) if link and link.startswith('http'): try: headers = {"User-Agent": feedparser.USER_AGENT} resp = requests.get(link, headers=headers, timeout=2.5) if resp.status_code == 200: soup = BeautifulSoup(resp.text, "html.parser") # Check og:image, twitter:image, image_src og_img = soup.find('meta', property='og:image') or soup.find('meta', attrs={'name': 'og:image'}) if og_img and og_img.get('content'): return og_img.get('content') tw_img = soup.find('meta', attrs={'name': 'twitter:image'}) or soup.find('meta', property='twitter:image') if tw_img and tw_img.get('content'): return tw_img.get('content') rel_img = soup.find('link', rel='image_src') if rel_img and rel_img.get('href'): return rel_img.get('href') except Exception: pass return None