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, 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
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",
    "Accept-Encoding": "gzip, deflate, br",
    "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": "gzip, deflate, br",
    "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/(.+)")


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 any(d in url for d in ["rss", "feed", "atom"]):
        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 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)
        tasks = [
            self._seed_search(),
            self._seed_wikipedia(),
            self._seed_reddit(),
            self._seed_youtube(),
        ]
        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": "duckduckgo,google,bing,brave",
            "language": "all",
        }
        headers = {
            "Accept": "application/json",
            "X-Forwarded-For": "127.0.0.1",
            "User-Agent": "ResearchOwl/1.0",
        }
        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 []

    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:
                    with DDGS() as ddgs:
                        ddg_results = list(ddgs.text(
                            query,
                            max_results=settings.max_pages_per_search
                        ))
                    results = ddg_results
                    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):
        """Search Wikipedia API for correct article URLs.
        Tries English first, falls back to Spanish if no results found."""
        http = await self._get_http()
        added = 0

        for lang in ("en", "es"):
            try:
                api_url = (
                    f"https://{lang}.wikipedia.org/w/api.php?action=opensearch"
                    f"&search={quote_plus(self.topic)}&limit=10&format=json"
                )
                async with http.get(api_url) as resp:
                    data = await resp.json()
                    urls = data[3] if len(data) > 3 else []
                    for url in urls:
                        if url:
                            await self.db.add_source(self.session_id, url, "wikipedia", depth=0)
                            added += 1
                logger.info("Wikipedia seed", lang=lang, found=len(urls))
                if added > 0:
                    break  # English results found — no need to try Spanish
            except Exception as e:
                logger.warning("Wikipedia API seed failed", lang=lang, 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:
            with DDGS() as ddgs:
                results = list(ddgs.videos(
                    f"{self.topic} documentary explanation",
                    max_results=10
                ))
                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"]

            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 == "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_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