reasoning/scraper.py

import requests
from bs4 import BeautifulSoup
import re
import time
import json
import os
from typing import List, Dict

HEADERS = {
    "User-Agent": "Mozilla/5.0 (MVI-AI Engine)"
}

CACHE_FILE = "knowledge_cache.json"

WIKI_PAGE = "https://en.wikipedia.org/wiki/{query}"
WIKI_SEARCH = "https://en.wikipedia.org/w/index.php?search={query}"
DDG_SEARCH = "https://duckduckgo.com/html/?q={query}"


# -------------------------
# CACHE
# -------------------------

def load_cache():
    if os.path.exists(CACHE_FILE):
        with open(CACHE_FILE, "r") as f:
            return json.load(f)
    return {}

def save_cache(cache):
    with open(CACHE_FILE, "w") as f:
        json.dump(cache, f)

CACHE = load_cache()


# -------------------------
# UTIL
# -------------------------

def normalize_query(query: str) -> str:
    query = query.lower()

    for phrase in ["what is", "who is", "define", "explain"]:
        query = query.replace(phrase, "")

    query = re.sub(r"[^\w\s]", "", query)
    return query.strip()


def clean_text(text: str) -> str:
    text = re.sub(r"\[\d+\]", "", text)
    text = re.sub(r"\s+", " ", text)
    return text.strip()


# -------------------------
# FETCH
# -------------------------

def fetch(url: str) -> str:
    try:
        r = requests.get(url, headers=HEADERS, timeout=8)
        if r.status_code != 200:
            return ""
        return r.text
    except:
        return ""


# -------------------------
# PARSE
# -------------------------

def extract_paragraphs(html: str) -> List[str]:
    soup = BeautifulSoup(html, "html.parser")
    paragraphs = soup.find_all("p")

    results = []
    for p in paragraphs:
        text = clean_text(p.get_text())
        if len(text) > 60:
            results.append(text)

    return results


def extract_wiki(html: str) -> List[str]:
    soup = BeautifulSoup(html, "html.parser")
    content = soup.find("div", {"id": "mw-content-text"})
    if not content:
        return []
    return extract_paragraphs(str(content))


# -------------------------
# SEARCH FALLBACK
# -------------------------

def wikipedia_search(query: str) -> str:
    html = fetch(WIKI_SEARCH.format(query=query.replace(" ", "+")))
    soup = BeautifulSoup(html, "html.parser")

    result = soup.select_one(".mw-search-result-heading a")
    if result:
        return "https://en.wikipedia.org" + result.get("href")

    return ""


def duckduckgo_search(query: str) -> List[str]:
    html = fetch(DDG_SEARCH.format(query=query.replace(" ", "+")))
    soup = BeautifulSoup(html, "html.parser")

    links = []
    for a in soup.select(".result__a"):
        href = a.get("href")
        if href and href.startswith("http"):
            links.append(href)

    return links[:3]


# -------------------------
# SCRAPERS
# -------------------------

def scrape_wikipedia(query: str) -> List[str]:
    url = WIKI_PAGE.format(query=query.replace(" ", "_"))
    html = fetch(url)

    if "Wikipedia does not have an article" in html:
        url = wikipedia_search(query)
        if not url:
            return []
        html = fetch(url)

    return extract_wiki(html)


def scrape_generic(url: str) -> List[str]:
    html = fetch(url)
    return extract_paragraphs(html)


# -------------------------
# RANKING
# -------------------------

def rank_results(paragraphs: List[str], query: str) -> List[str]:
    q_words = set(query.lower().split())

    def score(p):
        return sum(word in p.lower() for word in q_words)

    return sorted(paragraphs, key=score, reverse=True)


# -------------------------
# MAIN
# -------------------------

def scrape_knowledge(query: str, limit: int = 5) -> List[Dict]:
    if query in CACHE:
        return CACHE[query]

    clean_query = normalize_query(query)
    if not clean_query:
        return []

    paragraphs = scrape_wikipedia(clean_query)

    if not paragraphs:
        links = duckduckgo_search(clean_query)

        for link in links:
            paragraphs.extend(scrape_generic(link))

    if not paragraphs:
        return []

    ranked = rank_results(paragraphs, clean_query)

    knowledge = []
    for p in ranked[:limit]:
        knowledge.append({
            "query": query,
            "text": p,
            "timestamp": time.time()
        })

    CACHE[query] = knowledge
    save_cache(CACHE)

    return knowledge