"""Scrape news article text from URLs using trafilatura.""" from __future__ import annotations import hashlib import json import logging import time from pathlib import Path import trafilatura from src.llm.client import load_config logger = logging.getLogger(__name__) def scrape_article(url: str, timeout: int = 15) -> dict | None: """Scrape a single article and return its text content. Returns dict with url, title, text, or None if scraping fails. """ try: downloaded = trafilatura.fetch_url(url) if not downloaded: logger.warning(f" Failed to download: {url}") return None text = trafilatura.extract( downloaded, include_comments=False, include_tables=False, favor_precision=True, ) if not text or len(text) < 100: logger.warning(f" Extracted text too short from: {url}") return None metadata = trafilatura.extract( downloaded, output_format="json", include_comments=False, ) meta_dict = json.loads(metadata) if metadata else {} return { "url": url, "title": meta_dict.get("title", ""), "text": text, "date": meta_dict.get("date", ""), "source": meta_dict.get("sitename", ""), } except Exception as e: logger.warning(f" Error scraping {url}: {e}") return None def scrape_event_articles( news_data: dict, config: dict | None = None ) -> list[dict]: """Scrape all articles for a single event from GDELT results. Args: news_data: Output from gdelt_fetcher.fetch_news_for_event() config: Project config Returns: List of scraped article dicts with domain annotation. """ if config is None: config = load_config() timeout = config["data"]["scraper_timeout"] articles_dir = Path(config["paths"]["articles_dir"]) articles_dir.mkdir(parents=True, exist_ok=True) event_id = news_data["event_id"] all_articles = [] seen_urls = set() for domain, articles in news_data.get("articles_by_domain", {}).items(): for article_meta in articles: url = article_meta["url"] if url in seen_urls: continue seen_urls.add(url) logger.info(f" [{event_id}] Scraping: {url[:80]}...") result = scrape_article(url, timeout) if result: result["domain"] = domain result["event_id"] = event_id all_articles.append(result) # Save individual article url_hash = hashlib.md5(url.encode()).hexdigest()[:12] article_path = articles_dir / f"{event_id}_{url_hash}.json" article_path.write_text( json.dumps(result, indent=2, ensure_ascii=False) ) # Be polite to news sites time.sleep(0.5) logger.info(f" [{event_id}] Scraped {len(all_articles)}/{len(seen_urls)} articles successfully") return all_articles