1  from typing import List, Tuple
   2  import requests
   3  from bs4 import BeautifulSoup
   4  from duckduckgo_search import DDGS
   5  from googlesearch import search as google_search
   6  import arxiv
   7  import logging
   8  from pydantic import ValidationError
   9  import json
  10  import wikipedia
  11  import yfinance as yf
  12  from ..tools.schemas import (
  13      FetchNewsParams,
  14      FetchArxivParams,
  15      FetchDDGParams,
  16      FetchTorrentsParams,
  17      FetchWikiParams,
  18      FetchIMDBParams,
  19      FetchYouTubeParams,
  20      FetchFinanceParams
  21  )
  22  
  23  logging.basicConfig(level=logging.INFO)
  24  logger = logging.getLogger(__name__)
  25  
  26  
  27  # ——— Utility: scrape a page for title + snippet ———
  28  def scrape_page(url: str) -> Tuple[str, str]:
  29      try:
  30          global scrape_counter
  31          resp = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=5)
  32          resp.raise_for_status()
  33          soup = BeautifulSoup(resp.text, "html.parser")
  34          title = soup.title.string.strip() if soup.title else url
  35          paragraphs = soup.find_all("p")
  36          snippet = "\n".join(p.get_text().strip() for p in paragraphs)
  37          scrape_id = f"scrape_{scrape_counter}"
  38          scrape_counter += 1
  39          scraped_store[scrape_id] = {"href": url, "text": snippet}
  40          return title, snippet
  41      except Exception:
  42          return url, ""
  43  
  44  # ——— Tool: Google N search ———
  45  def fetch_news(**kwargs) -> List[str]:
  46      try:
  47          params = FetchNewsParams(**kwargs)
  48          results = []
  49          for url in google_search(params.query, num_results=params.max_results, unique=True):
  50              title, snippet = scrape_page(url)
  51              results.append(f"{title}\n{snippet}\n{url}")
  52              # logger.info("Google: %s", f"{title}\n{snippet}\n{url}")
  53              if len(results) >= params.max_results:
  54                  break
  55          return results
  56      except ValidationError as e:
  57          logger.error(f"Invalid parameters: {e}")
  58          return []
  59  
  60  # ——— Tool: arXiv search ———
  61  def fetch_arxiv(**kwargs) -> List[str]:
  62      try:
  63          params = FetchArxivParams(**kwargs)
  64          papers = []
  65          search = arxiv.Search(query=params.query, max_results=params.max_results)
  66          for r in search.results():
  67              summary = r.summary.replace("\n", " ").strip()
  68              papers.append(f"{r.title}\nAbstract: {summary}\nPDF: {r.pdf_url}")
  69          return papers
  70      except ValidationError as e:
  71          logger.error(f"Invalid parameters: {e}")
  72          return []
  73  
  74  # ——— Tool: DuckDuckGo text search ———
  75  def search_google(**kwargs) -> List[str]:
  76      try:
  77          return fetch_news(**kwargs)
  78      except ValidationError as e:
  79          logger.error(f"Invalid parameters: {e}")
  80          return []
  81  
  82  # ——— Tool: Torrent search ———
  83  def fetch_torrents(**kwargs) -> List[str]:
  84      try:
  85          params = FetchTorrentsParams(**kwargs)
  86          return search_google(query=f"""{params.query} proxy "torrent" """, max_results=params.max_results)
  87      except ValidationError as e:
  88          logger.error(f"Invalid parameters: {e}")
  89          return []
  90  
  91  
  92  scraped_store = {}
  93  scrape_counter = 0
  94  
  95  def fetch_wikipedia(**kwargs) -> List[str]:
  96      try:
  97          params = FetchWikiParams(**kwargs)
  98          wikipedia.set_lang("en")
  99          results = []
 100          global scrape_counter
 101          for title in wikipedia.search(params.query, results=params.max_results):
 102              try:
 103                  page = wikipedia.page(title, auto_suggest=False)
 104                  scrape_id = f"scrape_{scrape_counter}"
 105                  scrape_counter += 1
 106                  scraped_store[scrape_id] = {"href": page.url, "text": page.summary}
 107                  results.append(f"{page.title}\n{page.summary}\n{page.url}")
 108              except Exception as e:
 109                  logger.warning(f"Wikipedia subpage error: {e}")
 110          return results
 111      except Exception as e:
 112          logger.error(f"Wikipedia fetch error: {e}")
 113          return []
 114  
 115  def fetch_imdb(**kwargs) -> List[str]:
 116      try:
 117          params = FetchIMDBParams(**kwargs)
 118          return search_google(query=f"{params.query} site:imdb.com", max_results=params.max_results)
 119      except Exception as e:
 120          logger.error(f"IMDB fetch error: {e}")
 121          return []
 122  
 123  def fetch_youtube(**kwargs) -> List[str]:
 124      try:
 125          params = FetchYouTubeParams(**kwargs)
 126          return search_google(query=f"{params.query} site:youtube.com", max_results=params.max_results)
 127      except Exception as e:
 128          logger.error(f"YouTube fetch error: {e}")
 129          return []
 130  
 131  def fetch_finance(**kwargs) -> List[str]:
 132      try:
 133          params = FetchFinanceParams(**kwargs)
 134          results = []
 135          global scrape_counter
 136          tickers = params.query.strip().split()
 137          for symbol in tickers:
 138              ticker = yf.Ticker(symbol)
 139              info = ticker.info
 140              text = info.get("longBusinessSummary") or info.get("summary") or ""
 141              url = f"https://finance.yahoo.com/quote/{symbol}"
 142              scrape_id = f"scrape_{scrape_counter}"
 143              scrape_counter += 1
 144              scraped_store[scrape_id] = {"href": url, "text": text}
 145              results.append(f"{symbol.upper()}\n{text}\n{url}")
 146          return results
 147      except Exception as e:
 148          logger.error(f"Finance fetch error: {e}")
 149          return []
 150