mirasai/media-profiling

1"""2article_cache.py3Caches scraped articles per media outlet domain for benchmark fairness.4Ensures all models and modes use the same scraped articles.5 6Storage format:7  article_cache/8    <domain>/9      articles.json   <-- List of serialized Article objects10      metadata.json   <-- Scrape timestamp, count, source_url11"""12 13import json14import logging15import os16from datetime import datetime17from pathlib import Path18from typing import Optional, List19from urllib.parse import urlparse20 21from scraper import Article, MediaScraper22 23logger = logging.getLogger(__name__)24 25ARTICLE_CACHE_DIR = Path("article_cache")26 27 28class ArticleCache:29    """30    Scrape-once, reuse-everywhere article cache for benchmark fairness.31 32    Keyed by outlet domain. On first request for a domain, scrapes articles33    and saves them. Subsequent requests return the cached articles.34    """35 36    def __init__(self, cache_dir: Path = ARTICLE_CACHE_DIR):37        self.cache_dir = cache_dir38        self.cache_dir.mkdir(exist_ok=True)39        self._memory_cache: dict[str, list[Article]] = {}40 41    @staticmethod42    def _domain_from_url(url: str) -> str:43        parsed = urlparse(url if url.startswith("http") else f"https://{url}")44        domain = parsed.netloc or parsed.path45        domain = domain.replace("www.", "").strip("/")46        return domain47 48    def _domain_dir(self, domain: str) -> Path:49        return self.cache_dir / domain.replace("/", "_")50 51    def get_articles(self, source_url: str, max_articles: int = 20) -> list[Article]:52        """53        Get articles for a source URL. Returns cached if available,54        otherwise scrapes and caches.55        """56        domain = self._domain_from_url(source_url)57 58        # 1. Check in-memory cache59        if domain in self._memory_cache:60            logger.info(f"[ArticleCache] HIT (memory): {domain}, {len(self._memory_cache[domain])} articles")61            return self._memory_cache[domain][:max_articles]62 63        # 2. Check disk cache64        domain_dir = self._domain_dir(domain)65        articles_path = domain_dir / "articles.json"66        if articles_path.exists():67            try:68                articles = self._load_from_disk(articles_path)69                self._memory_cache[domain] = articles70                logger.info(f"[ArticleCache] HIT (disk): {domain}, {len(articles)} articles")71                return articles[:max_articles]72            except Exception as e:73                logger.warning(f"[ArticleCache] Disk cache corrupt for {domain}: {e}")74 75        # 3. Scrape and cache76        logger.info(f"[ArticleCache] MISS: {domain}, scraping up to {max_articles} articles...")77        scraper = MediaScraper(source_url, max_articles=max_articles)78        articles = scraper.scrape_feed()79 80        self._save_to_disk(domain, source_url, articles)81        self._memory_cache[domain] = articles82        logger.info(f"[ArticleCache] Cached {len(articles)} articles for {domain}")83 84        return articles[:max_articles]85 86    def _save_to_disk(self, domain: str, source_url: str, articles: list[Article]):87        domain_dir = self._domain_dir(domain)88        domain_dir.mkdir(parents=True, exist_ok=True)89 90        articles_data = [91            {92                "url": a.url,93                "title": a.title,94                "text": a.text,95                "author": a.author,96                "date": a.date,97                "category": a.category,98                "has_sources": a.has_sources,99                "source_links": a.source_links,100                "is_opinion": a.is_opinion,101            }102            for a in articles103        ]104        with open(domain_dir / "articles.json", "w", encoding="utf-8") as f:105            json.dump(articles_data, f, indent=2, ensure_ascii=False)106 107        metadata = {108            "domain": domain,109            "source_url": source_url,110            "scrape_date": datetime.now().isoformat(),111            "article_count": len(articles),112        }113        with open(domain_dir / "metadata.json", "w", encoding="utf-8") as f:114            json.dump(metadata, f, indent=2)115 116    def _load_from_disk(self, articles_path: Path) -> list[Article]:117        with open(articles_path, "r", encoding="utf-8") as f:118            articles_data = json.load(f)119        return [120            Article(121                url=a["url"],122                title=a["title"],123                text=a["text"],124                author=a.get("author"),125                date=a.get("date"),126                category=a.get("category"),127                has_sources=a.get("has_sources", False),128                source_links=a.get("source_links", []),129                is_opinion=a.get("is_opinion", False),130            )131            for a in articles_data132        ]133 134    def clear(self):135        import shutil136        if self.cache_dir.exists():137            shutil.rmtree(self.cache_dir)138        self.cache_dir.mkdir(exist_ok=True)139        self._memory_cache.clear()140        logger.info("[ArticleCache] Cache cleared")141