researchowl/src/scraper/exhaustive.py

"""
ResearchOwl Exhaustive Scraper
Core engine: discovers, expands, and evaluates sources recursively
"""
import asyncio
import random
import re
import time
from typing import Optional
from urllib.parse import urljoin, urlparse, parse_qs, quote, quote_plus

import aiohttp
import feedparser
import structlog
import trafilatura
from bs4 import BeautifulSoup
from duckduckgo_search import DDGS
from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound
from tenacity import retry, stop_after_attempt, wait_exponential

from src.config import settings, SAFE_ACCEPT_ENCODING
from src.db.database import ResearchDB

logger = structlog.get_logger()

HEADERS = {
    "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9,es;q=0.8",
    # Sin "br": el descompresor incremental de aiohttp 3.14 falla con streams
    # brotli válidos según el troceo de chunks (disclosure.org, todaywhy.com;
    # los mismos bytes descomprimen bien offline). Con gzip el camino zlib es
    # sólido. OJO: ya NO hay ningún backend brotli instalado (397546a quitó
    # brotlicffi) — si un servidor responde br sin que se haya anunciado,
    # aiohttp falla sin recuperación posible y esa fuente se pierde.
    "Accept-Encoding": SAFE_ACCEPT_ENCODING,
    "DNT": "1",
}

# Reddit requires its own headers — generic bots get 403
REDDIT_HEADERS = {
    "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept": "application/json, text/javascript, */*; q=0.01",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": SAFE_ACCEPT_ENCODING,  # sin "br", ver HEADERS
    "Referer": "https://www.reddit.com/",
    "X-Requested-With": "XMLHttpRequest",
}

# Domains to skip — not useful for research
BLACKLIST_DOMAINS = {
    "facebook.com", "twitter.com", "x.com", "instagram.com", "tiktok.com",
    "pinterest.com", "linkedin.com", "amazon.com", "ebay.com", "etsy.com",
    "ads.google.com", "doubleclick.net", "googleadservices.com",
}

# Source type patterns
YOUTUBE_RE = re.compile(r"(?:youtube\.com/watch\?v=|youtu\.be/)([a-zA-Z0-9_-]{11})")
PDF_RE = re.compile(r"\.pdf(\?|$)", re.IGNORECASE)
REDDIT_RE = re.compile(r"reddit\.com/(r/\w+/comments/\w+)")
WIKIPEDIA_RE = re.compile(r"wikipedia\.org/wiki/(.+)")
# Segmentos de path típicos de feeds, con límite de palabra: el substring puro
# clasificaba /feedback o /atomic como rss.
RSS_RE = re.compile(r"/(rss|feeds?|atom)(/|\.xml|\.rss|$)|\.rss$|format=rss",
                    re.IGNORECASE)


async def fetch_with_retry(fetch_fn, source_name: str, max_retries: int = 3):
    last_exc = None
    for attempt in range(max_retries):
        try:
            return await fetch_fn()
        except Exception as e:
            last_exc = e
            if attempt < max_retries - 1:
                wait = 2 ** attempt + random.random()
                logger.debug("fetch_with_retry backoff", source=source_name[:60],
                             attempt=attempt + 1, wait=round(wait, 1), error=str(e))
                await asyncio.sleep(wait)
    logger.warning("fetch_with_retry exhausted", source=source_name[:60], error=str(last_exc))
    raise last_exc


def detect_source_type(url: str) -> str:
    if YOUTUBE_RE.search(url):
        return "youtube"
    if PDF_RE.search(url):
        return "pdf"
    if REDDIT_RE.search(url):
        return "reddit"
    if WIKIPEDIA_RE.search(url):
        return "wikipedia"
    if "arxiv.org" in url:
        return "arxiv"
    if RSS_RE.search(url):
        return "rss"
    return "web"


def is_blacklisted(url: str) -> bool:
    try:
        domain = urlparse(url).netloc.lower().split(":")[0]
        if domain.startswith("www."):
            domain = domain[4:]
        # Exact domain or subdomain match — NOT substring (evitaba bloquear
        # netflix.com / phoenix.com por contener "x.com", etc.)
        return any(domain == bl or domain.endswith("." + bl)
                   for bl in BLACKLIST_DOMAINS)
    except Exception:
        return True


def _unwrap_news_link(link: str) -> str:
    """Bing News envuelve los links de entry en apiclick.aspx; la URL real del
    publisher viene en texto plano en el parámetro ?url=. Cualquier otro link
    pasa sin tocar. Si es un apiclick pero no trae URL usable, devuelve "" —
    el caller lo descarta por startswith("http") — en vez de dejar pasar el
    redirect de bing.com, que se rasparía como página de consent/basura."""
    try:
        parsed = urlparse(link)
        if (parsed.netloc.lower().endswith("bing.com")
                and parsed.path == "/news/apiclick.aspx"):
            real = parse_qs(parsed.query).get("url", [""])[0].strip()
            if real.startswith("//"):  # protocol-relative
                real = "https:" + real
            return real if real.startswith("http") else ""
    except Exception:
        return ""
    return link


def normalize_url(url: str) -> str:
    # Strip only the fragment. NO borrar el query string: rompía URLs de
    # YouTube (watch?v=...) y artículos que enrutan por query (?id=, ?p=).
    parsed = urlparse(url)
    clean = parsed._replace(fragment="")
    return clean.geturl().rstrip("/")


class ExhaustiveScraper:
    """
    Recursive source discoverer and content extractor.
    Keeps expanding until saturation or limits hit.
    """

    # Common stopwords to ignore when extracting topic keywords
    _STOPWORDS = {
        'the','a','an','and','or','of','in','on','at','to','for','is','are','was',
        'were','be','been','have','has','had','do','does','did','about','with','from',
        'el','la','los','las','de','del','en','un','una','y','o','que','se','por',
        'con','para','sobre','como','pero','más',
    }

    def __init__(self, db: ResearchDB, session_id: int, topic: str,
                 progress_callback=None):
        self.db = db
        self.session_id = session_id
        self.topic = topic
        self.progress_callback = progress_callback
        self.iteration = 0
        self.total_sources = 0
        self._stop = False
        self._http: Optional[aiohttp.ClientSession] = None
        # Pre-compute topic keywords for child-URL relevance filtering
        self._keywords = [
            w for w in re.findall(r'\b\w{3,}\b', topic.lower())
            if w not in self._STOPWORDS
        ]

    def _url_is_relevant(self, url: str, title: str = "") -> bool:
        """True if URL path or title contains at least one topic keyword."""
        if not self._keywords:
            return True
        text = (urlparse(url).path + " " + (title or "")).lower()
        return any(kw in text for kw in self._keywords)

    async def stop(self):
        self._stop = True

    async def _get_http(self) -> aiohttp.ClientSession:
        if not self._http or self._http.closed:
            timeout = aiohttp.ClientTimeout(total=settings.request_timeout)
            self._http = aiohttp.ClientSession(headers=HEADERS, timeout=timeout)
        return self._http

    async def close(self):
        if self._http and not self._http.closed:
            await self._http.close()

    # ─── Seed discovery ───────────────────────────────────────────────────────

    async def seed(self):
        """Initial broad search across multiple sources"""
        logger.info("Seeding research", topic=self.topic,
                    youtube=settings.enable_youtube, reddit=settings.enable_reddit)
        tasks = [
            self._seed_search(),
            self._seed_wikipedia(),
            self._seed_archive(),
        ]
        if settings.enable_reddit:
            tasks.append(self._seed_reddit())
        if settings.enable_youtube:
            tasks.append(self._seed_youtube())
        if settings.enable_news_seed:
            tasks.append(self._seed_news())
        await asyncio.gather(*tasks, return_exceptions=True)

    async def _generate_ddg_queries(self) -> list[str]:
        fallback = [
            self.topic,
            f"{self.topic} history facts",
            f"{self.topic} evidence analysis",
            f"{self.topic} official report",
            f"{self.topic} investigation",
            f"{self.topic} wikipedia",
            f"{self.topic} documentary",
            f"{self.topic} research study",
        ]

        if not settings.anthropic_api_key:
            return fallback

        try:
            from src.llm import get_anthropic_client
            logger.info("Generating DDG queries with Claude", topic=self.topic)
            client = get_anthropic_client()
            prompt = (
                f'Generate exactly 8 DuckDuckGo search queries to research: "{self.topic}"\n\n'
                f'Rules:\n'
                f'- Each query must be specific and distinct — no generic templates\n'
                f'- Cover different angles: facts, history, official sources, criticism, '
                f'technical details, recent developments, expert opinions, primary sources\n'
                f'- Use the most specific terminology for this topic\n'
                f'- Include the topic language naturally (if topic is in Spanish, '
                f'mix Spanish and English queries for broader coverage)\n'
                f'- Output ONLY the 8 queries, one per line, no numbering, '
                f'no explanations, no markdown\n'
            )
            msg = await client.messages.create(
                model=settings.claude_model,
                max_tokens=300,
                messages=[{"role": "user", "content": prompt}]
            )
            raw = msg.content[0].text.strip()
            queries = [q.strip() for q in raw.split('\n') if q.strip()]
            if self.topic not in queries:
                queries = [self.topic] + queries[:7]
            queries = queries[:8]
            logger.info("DDG queries generated by Claude", queries=queries)
            return queries
        except Exception as e:
            logger.warning("Claude query generation failed, using fallback",
                           error=str(e), error_type=type(e).__name__)
            return fallback

    async def _search_searxng(self, query: str) -> list[dict]:
        """Busca en SearXNG y retorna lista de {href, title}. Retorna [] si no disponible."""
        import aiohttp
        searxng_url = settings.searxng_url
        params = {
            "q": query,
            "format": "json",
            "engines": settings.searxng_engines,
            "language": "all",
        }
        headers = {
            "Accept": "application/json",
            "X-Forwarded-For": "127.0.0.1",
            "User-Agent": "ResearchOwl/1.0",
            # Sin esto la sesión hereda el default de aiohttp, que anunciaría
            # br si algún día se reinstala un backend brotli — y el decode roto
            # de 3.14 tumbaría TODAS las búsquedas SearXNG (el camino primario).
            "Accept-Encoding": SAFE_ACCEPT_ENCODING,
        }
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    searxng_url,
                    params=params,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=15)
                ) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        results = data.get("results", [])
                        logger.info("SearXNG query ok", query=query, results=len(results))
                        return [
                            {"href": r.get("url", ""), "title": r.get("title", "")}
                            for r in results
                            if r.get("url")
                        ]
                    else:
                        logger.warning("SearXNG non-200", status=resp.status, query=query)
                        return []
        except Exception as e:
            logger.warning("SearXNG failed", query=query, error=str(e))
            return []

    # DDGS es síncrono (requests bloqueante por dentro): llamarlo directamente
    # desde código async congela el event loop entero — bot de Telegram incluido —
    # mientras dura la búsqueda. Estos helpers deben ejecutarse SIEMPRE vía
    # run_in_executor.

    @staticmethod
    def _ddg_text_sync(query: str) -> list[dict]:
        with DDGS() as ddgs:
            return list(ddgs.text(query, max_results=settings.max_pages_per_search))

    @staticmethod
    def _ddg_videos_sync(query: str) -> list[dict]:
        with DDGS() as ddgs:
            return list(ddgs.videos(query, max_results=10))

    async def _seed_search(self):
        """SearXNG primary + DDG fallback per query"""
        queries = await self._generate_ddg_queries()
        for query in queries:
            if self._stop:
                break
            results = await self._search_searxng(query)
            if not results:
                logger.info("SearXNG vacío, usando DDG", query=query)
                try:
                    loop = asyncio.get_running_loop()
                    results = await loop.run_in_executor(
                        None, self._ddg_text_sync, query
                    )
                    logger.info("DDG fallback ok", query=query, results=len(results))
                except Exception as e:
                    logger.warning("DDG fallback failed", query=query, error=str(e))
                    results = []

            for r in results:
                url = normalize_url(r.get("href", ""))
                if url and not is_blacklisted(url):
                    await self.db.add_source(
                        self.session_id, url,
                        detect_source_type(url),
                        depth=0,
                        title=r.get("title")
                    )
            await asyncio.sleep(random.uniform(1, 3))

    async def _seed_wikipedia(self):
        """Siembra Wikipedia en AMBOS idiomas siempre, con búsqueda full-text
        (list=search). El opensearch anterior era prefix-de-título y devolvía 0
        para topics verbosos ("incidente ovni de Manises 1979") aunque el
        artículo exista; y el break tras en dejaba es solo como fallback.
        Cada idioma va aislado: un fallo no afecta al otro."""
        http = await self._get_http()

        for lang in ("en", "es"):
            try:
                api_url = (
                    f"https://{lang}.wikipedia.org/w/api.php?action=query"
                    f"&list=search&srsearch={quote_plus(self.topic)}"
                    f"&srlimit=8&format=json"
                )
                async with http.get(api_url) as resp:
                    data = await resp.json()
                hits = data.get("query", {}).get("search", [])
                added = 0
                for h in hits:
                    title = h.get("title")
                    if not title:
                        continue
                    url = (f"https://{lang}.wikipedia.org/wiki/"
                           f"{quote(title.replace(' ', '_'))}")
                    await self.db.add_source(self.session_id, url, "wikipedia",
                                             depth=0, title=title)
                    added += 1
                logger.info("Wikipedia seed", lang=lang, found=added)
            except Exception as e:
                logger.warning("Wikipedia API seed failed", lang=lang, error=str(e))

    async def _seed_archive(self):
        """Siembra Internet Archive (advancedsearch, mediatype:texts): documentos
        FOIA, informes y libros históricos. Se siembra la página /details/<id>;
        sus PDFs los descubre la recursión (la propia página los enlaza) y los
        extrae _extract_pdf. Sin fallback de query: el AND estricto de archive.org
        con el topic completo prima precisión — relajarlo mete ruido (verificado:
        'Manises' a secas devuelve loza valenciana). Best-effort: nunca rompe el seed."""
        try:
            http = await self._get_http()
            q = quote_plus(f"({self.topic}) AND mediatype:(texts)")
            url = (
                f"https://archive.org/advancedsearch.php?q={q}"
                "&fl[]=identifier&fl[]=title&rows=12&page=1&output=json"
            )
            async with http.get(url) as resp:
                if resp.status != 200:
                    logger.warning("Archive.org seed non-200", status=resp.status)
                    return
                data = await resp.json(content_type=None)
            docs = data.get("response", {}).get("docs", [])
            added = 0
            for d in docs:
                ident = d.get("identifier")
                if not ident:
                    continue
                item_url = f"https://archive.org/details/{ident}"
                title = d.get("title")
                await self.db.add_source(
                    self.session_id, item_url, "web", depth=0,
                    title=title if isinstance(title, str) else None
                )
                added += 1
            logger.info("Archive.org seed", found=len(docs), added=added)
        except Exception as e:
            logger.warning("Archive.org seed failed", error=str(e))

    async def _seed_news(self):
        """Siembra Bing News RSS para cobertura de actualidad. Solo registra
        el feed como fuente tipo rss: _extract_rss (dispatch del pipeline) lo
        parsea y siembra sus entries — sin duplicar lógica de parseo aquí.
        Bing y no Google News: los links de Google (news.google.com/rss/articles/
        CBMi…) van a un muro de consent + redirect cifrado solo resoluble vía su
        API interna batchexecute (verificado 2026-07-04); los de Bing llevan la
        URL real del publisher en ?url= (ver _unwrap_news_link).
        Best-effort: nunca rompe el seed."""
        try:
            url = f"https://www.bing.com/news/search?q={quote_plus(self.topic)}&format=rss"
            await self.db.add_source(
                self.session_id, url, "rss", depth=0,
                title=f"Bing News: {self.topic}"
            )
            logger.info("News seed added", topic=self.topic)
        except Exception as e:
            logger.warning("News seed failed", error=str(e))

    async def _seed_reddit(self):
        """Search Reddit — sequential to avoid rate limiting"""
        try:
            http = await self._get_http()
            url = f"https://www.reddit.com/search.json?q={quote_plus(self.topic)}&sort=top&limit=15&type=link"
            async with http.get(url, headers=REDDIT_HEADERS) as resp:
                if resp.status == 200:
                    data = await resp.json(content_type=None)
                    posts = data.get("data", {}).get("children", [])
                    for post in posts:
                        post_data = post.get("data", {})
                        permalink = post_data.get("permalink", "")
                        if permalink:
                            full_url = f"https://www.reddit.com{permalink}"
                            await self.db.add_source(
                                self.session_id, full_url, "reddit", depth=0,
                                title=post_data.get("title")
                            )
                    logger.info("Reddit seed", found=len(posts), status=resp.status)
                elif resp.status == 403:
                    logger.info("Reddit seed blocked (403) — server IP likely blocked by Reddit; skipping")
                else:
                    logger.warning("Reddit seed non-200", status=resp.status)
        except Exception as e:
            logger.warning("Reddit seed failed", error=str(e))

    async def _seed_youtube(self):
        """Search YouTube via DDG for video transcripts"""
        try:
            loop = asyncio.get_running_loop()
            results = await loop.run_in_executor(
                None, self._ddg_videos_sync,
                f"{self.topic} documentary explanation"
            )
            for r in results:
                url = r.get("content", "")
                if "youtube.com" in url or "youtu.be" in url:
                    await self.db.add_source(
                        self.session_id, url, "youtube", depth=0,
                        title=r.get("title")
                    )
        except Exception as e:
            logger.warning("YouTube seed failed", error=str(e))

    # ─── Main pipeline ────────────────────────────────────────────────────────

    async def run(self) -> dict:
        """
        Main exhaustive loop:
        1. Seed initial sources
        2. Process batch → extract content + new URLs
        3. Repeat until saturated or limits hit
        """
        await self.seed()

        while not self._stop:
            self.iteration += 1
            pending = await self.db.get_pending_sources(self.session_id, limit=20)

            if not pending:
                logger.info("No more pending sources — saturated", iteration=self.iteration)
                break

            if self.total_sources >= settings.max_sources:
                logger.info("Max sources reached", total=self.total_sources)
                break

            logger.info("Processing batch", iteration=self.iteration, batch_size=len(pending))

            # Reduced concurrency to 3 — avoids triggering Reddit/web rate limits
            semaphore = asyncio.Semaphore(3)
            tasks = [self._process_source(s, semaphore) for s in pending]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            new_sources = sum(1 for r in results if r and isinstance(r, int) and r > 0)
            self.total_sources += len(pending)

            stats = await self.db.get_session_stats(self.session_id)
            await self.db.update_session(
                self.session_id,
                iterations=self.iteration,
                total_sources=self.total_sources
            )

            if self.progress_callback:
                await self.progress_callback(self.iteration, self.total_sources)

            # Saturation check: if we found very few new URLs, we're done
            if new_sources < 3 and self.iteration > 2:
                logger.info("Saturation detected", new_sources=new_sources)
                break

            await asyncio.sleep(settings.request_delay)

        await self.close()
        final_stats = await self.db.get_session_stats(self.session_id)
        return final_stats

    async def _process_source(self, source: dict, semaphore: asyncio.Semaphore) -> int:
        """Extract content from a source and discover new URLs. Returns count of new URLs found."""
        async with semaphore:
            source_type = source["source_type"]
            url = source["url"]
            source_id = source["id"]

            # Saltar fuentes desactivadas (también las descubiertas dentro de
            # páginas web, no solo las del seed) sin gastar una petición de red.
            if ((source_type == "youtube" and not settings.enable_youtube) or
                    (source_type == "reddit" and not settings.enable_reddit)):
                await self.db.update_source(source_id, status="skipped",
                                            error=f"{source_type} disabled")
                return 0

            try:
                try:
                    cached = await self.db.get_cached_content(url)
                except Exception as cache_err:
                    logger.warning("Cache lookup failed", url=url, error=str(cache_err))
                    cached = None

                if cached:
                    logger.debug("Cache hit", url=url)
                    await self.db.save_source_content(source_id, cached)
                    await self.db.update_source(
                        source_id,
                        status="scraped",
                        scraped_at=time.time(),
                        word_count=len(cached.split()),
                    )
                    return 0

                if source_type == "youtube":
                    content, title = await fetch_with_retry(
                        lambda: self._extract_youtube(url), url
                    )
                elif source_type == "wikipedia":
                    content, title, new_urls = await fetch_with_retry(
                        lambda: self._extract_wikipedia(url), url
                    )
                    added = 0
                    for new_url in (new_urls or []):
                        if self._url_is_relevant(new_url):
                            await self.db.add_source(
                                self.session_id, new_url, "wikipedia",
                                depth=source["depth"] + 1
                            )
                            added += 1
                    await self._mark_scraped(source_id, content, title, url)
                    return added
                elif source_type == "reddit":
                    content, title = await fetch_with_retry(
                        lambda: self._extract_reddit(url), url
                    )
                    await asyncio.sleep(settings.request_delay)
                elif source_type == "rss":
                    # El feed no tiene contenido propio: es puro descubrimiento.
                    return await self._extract_rss(source_id, url, source["depth"])
                elif source_type == "pdf":
                    content, title = await fetch_with_retry(
                        lambda: self._extract_pdf(url), url
                    )
                else:
                    content, title, new_urls = await fetch_with_retry(
                        lambda: self._extract_web(url, source["depth"]), url
                    )
                    added = 0
                    for new_url in (new_urls or []):
                        if self._url_is_relevant(new_url):
                            await self.db.add_source(
                                self.session_id, new_url,
                                detect_source_type(new_url),
                                depth=source["depth"] + 1
                            )
                            added += 1
                    await self._mark_scraped(source_id, content, title, url)
                    return added

                await self._mark_scraped(source_id, content, title, url)
                return 0

            except Exception as e:
                logger.warning("Source extraction failed", url=url, error=str(e))
                await self.db.update_source(source_id, status="failed", error=str(e)[:200])
                return 0

    async def _mark_scraped(self, source_id: int, content: Optional[str],
                             title: Optional[str], url: str):
        if not content:
            logger.debug("No content returned", source_id=source_id, url=url[:60])
            await self.db.update_source(source_id, status="skipped",
                                        error="Content too short or empty")
            return
        if len(content) < settings.min_content_length:
            logger.debug("Content too short", source_id=source_id,
                         length=len(content), url=url[:60])
            await self.db.update_source(source_id, status="skipped",
                                        error="Content too short or empty")
            return

        word_count = len(content.split())

        await self.db.save_source_content(source_id, content)

        await self.db.update_source(
            source_id,
            status="scraped",
            title=title or url,
            word_count=word_count,
            scraped_at=time.time(),
            quality_score=min(1.0, word_count / 1000)
        )
        logger.info("Source scraped", source_id=source_id, words=word_count, url=url[:60])

    # ─── Extractors ───────────────────────────────────────────────────────────

    async def _extract_web(self, url: str, depth: int) -> tuple[Optional[str], Optional[str], list[str]]:
        """Extract text + discover internal/external links"""
        if is_blacklisted(url):
            return None, None, []

        http = await self._get_http()
        async with http.get(url) as resp:
            if resp.status != 200:
                return None, None, []
            html = await resp.text(errors="replace")

        # Extract main content with trafilatura (much better than BS4 for articles)
        content = trafilatura.extract(
            html,
            include_links=False,
            include_tables=True,
            favor_recall=True
        )

        # Extract title and new URLs with BS4
        soup = BeautifulSoup(html, "lxml")
        # .string es None si el <title> tiene tags anidados; get_text es robusto
        title = soup.title.get_text(strip=True) if soup.title else url

        new_urls = []
        if depth < settings.max_depth:
            base = f"{urlparse(url).scheme}://{urlparse(url).netloc}"
            for a in soup.find_all("a", href=True):
                href = a["href"]
                full_url = normalize_url(urljoin(base, href))
                if (full_url.startswith("http") and
                        not is_blacklisted(full_url) and
                        not await self.db.source_exists(self.session_id, full_url)):
                    new_urls.append(full_url)

        return content, title, new_urls[:30]  # cap links per page

    async def _extract_wikipedia(self, url: str) -> tuple[Optional[str], Optional[str], list[str]]:
        """Wikipedia: extract content + follow internal wiki links.
        Works for both en.wikipedia.org and es.wikipedia.org."""
        http = await self._get_http()
        async with http.get(url) as resp:
            if resp.status != 200:
                logger.debug("Wikipedia non-200", status=resp.status, url=url[:60])
                return None, None, []
            html = await resp.text(errors="replace")

        soup = BeautifulSoup(html, "lxml")
        title_tag = soup.find("h1", {"id": "firstHeading"})
        title = title_tag.text if title_tag else url

        # Get clean content
        content_div = soup.find("div", {"id": "mw-content-text"})
        if not content_div:
            return None, title, []

        # Remove navboxes, references, etc.
        for tag in content_div.find_all(["table", "sup", "style"]):
            tag.decompose()

        content = content_div.get_text(separator="\n", strip=True)

        # Extract Wikipedia internal links using the URL's actual domain
        parsed = urlparse(url)
        wiki_base = f"{parsed.scheme}://{parsed.netloc}"
        new_urls = []
        for a in content_div.find_all("a", href=True):
            href = a["href"]
            if href.startswith("/wiki/") and ":" not in href:
                full_url = normalize_url(f"{wiki_base}{href}")
                if not await self.db.source_exists(self.session_id, full_url):
                    new_urls.append(full_url)

        return content, title, new_urls[:20]

    async def _extract_youtube(self, url: str) -> tuple[Optional[str], Optional[str]]:
        """Extract YouTube transcript"""
        match = YOUTUBE_RE.search(url)
        if not match:
            return None, None

        video_id = match.group(1)
        loop = asyncio.get_running_loop()

        def _fetch():
            return YouTubeTranscriptApi.get_transcript(
                video_id, languages=["en", "es", "en-US", "en-GB"]
            )

        try:
            transcript_list = await asyncio.wait_for(
                loop.run_in_executor(None, _fetch),
                timeout=30.0
            )
            text = " ".join(t["text"] for t in transcript_list)
            return text, f"YouTube: {video_id}"
        except asyncio.TimeoutError:
            logger.warning("YouTube transcript timed out", video_id=video_id)
            return None, None
        except NoTranscriptFound:
            return None, None
        except Exception as e:
            logger.warning("YouTube transcript failed", video_id=video_id, error=str(e))
            return None, None

    async def _extract_reddit(self, url: str) -> tuple[Optional[str], Optional[str]]:
        """Extract Reddit post + top comments via JSON API"""
        json_url = url.rstrip("/") + ".json?limit=100&sort=top"
        http = await self._get_http()
        try:
            async with http.get(json_url, headers=REDDIT_HEADERS) as resp:
                if resp.status == 403:
                    logger.info("Reddit post blocked (403) — skipping", url=url[:60])
                    return None, None
                if resp.status != 200:
                    logger.debug("Reddit non-200", status=resp.status, url=url[:60])
                    return None, None
                data = await resp.json(content_type=None)

            post = data[0]["data"]["children"][0]["data"]
            title = post.get("title", "")
            selftext = post.get("selftext", "")

            comments = []
            if len(data) > 1:
                for child in data[1]["data"]["children"][:50]:
                    body = child.get("data", {}).get("body", "")
                    if body and body != "[deleted]" and len(body) > 50:
                        score = child.get("data", {}).get("score", 0)
                        if score > 5:  # only upvoted comments
                            comments.append(body)

            content = f"# {title}\n\n{selftext}\n\n## Top Comments\n\n" + "\n\n---\n\n".join(comments)
            return content, title

        except Exception as e:
            logger.warning("Reddit extraction failed", url=url, error=str(e))
            return None, None

    async def _extract_rss(self, source_id: int, url: str, depth: int) -> int:
        """RSS/Atom: siembra las entries del feed como fuentes nuevas (filtradas
        por relevancia de título/URL) y marca el feed como skipped — no tiene
        contenido propio que trocear. feedparser.parse es bloqueante: va en hilo,
        igual que en el monitor de noticias. Devuelve nº de fuentes añadidas."""
        parsed = await asyncio.to_thread(feedparser.parse, url,
                                         agent=HEADERS["User-Agent"])
        entries = parsed.entries or []
        added = 0
        if depth < settings.max_depth:
            # Se escanean TODAS las entries y el cap aplica a las añadidas:
            # capar antes de filtrar perdía las relevantes que no caen entre
            # las 20 más recientes (verificado con thedebrief: 0/20 vs 3/100).
            for e in entries:
                if added >= 20:  # cap, como los 30 links/página de web
                    break
                link = normalize_url(_unwrap_news_link(getattr(e, "link", "") or ""))
                title = getattr(e, "title", "") or ""
                if (not link.startswith("http") or is_blacklisted(link)
                        or not self._url_is_relevant(link, title)):
                    continue
                if await self.db.source_exists(self.session_id, link):
                    continue
                await self.db.add_source(
                    self.session_id, link, detect_source_type(link),
                    depth=depth + 1, title=title or None
                )
                added += 1
        await self.db.update_source(
            source_id, status="skipped",
            error=f"rss feed: {added} entries sembradas"
        )
        logger.info("RSS feed seeded", url=url[:60],
                    entries=len(entries), added=added)
        return added

    async def _extract_pdf(self, url: str) -> tuple[Optional[str], Optional[str]]:
        """Download and extract PDF text"""
        import pdfplumber
        import tempfile
        import os

        http = await self._get_http()
        try:
            async with http.get(url) as resp:
                if resp.status != 200:
                    return None, None
                content_length = int(resp.headers.get("content-length", 0))
                if content_length > 50 * 1024 * 1024:  # skip PDFs > 50MB
                    return None, None
                pdf_bytes = await resp.read()

            with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as f:
                f.write(pdf_bytes)
                tmp_path = f.name

            try:
                with pdfplumber.open(tmp_path) as pdf:
                    pages = [p.extract_text() or "" for p in pdf.pages[:50]]  # max 50 pages
                    text = "\n\n".join(pages)
                return text, url.split("/")[-1]
            finally:
                os.unlink(tmp_path)

        except Exception as e:
            logger.warning("PDF extraction failed", url=url, error=str(e))
            return None, None