Merge branch '2-news-api' into 3-market-api

src/app/news/__init__.py

@@ -0,0 +1,32 @@
+from app.utils.wrapper_handler import WrapperHandler
+from .base import NewsWrapper, Article
+from .news_api import NewsApiWrapper
+from .gnews_api import GoogleNewsWrapper
+from .cryptopanic_api import CryptoPanicWrapper
+from .duckduckgo import DuckDuckGoWrapper
+
+__all__ = ["NewsApiWrapper", "GoogleNewsWrapper", "CryptoPanicWrapper", "DuckDuckGoWrapper"]
+
+
+class NewsAPIs(NewsWrapper):
+    """
+    A wrapper class that aggregates multiple news API wrappers and tries them in order until one succeeds.
+    This class uses the WrapperHandler to manage multiple NewsWrapper instances.
+    It includes, and tries, the following news API wrappers in this order:
+    - GoogleNewsWrapper
+    - DuckDuckGoWrapper
+    - NewsApiWrapper
+    - CryptoPanicWrapper
+
+    It provides methods to get top headlines and latest news by delegating the calls to the first successful wrapper.
+    If all wrappers fail, it raises an exception.
+    """
+
+    def __init__(self):
+        wrappers = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
+        self.wrapper_handler: WrapperHandler[NewsWrapper] = WrapperHandler.build_wrappers(wrappers)
+
+    def get_top_headlines(self, total: int = 100) -> list[Article]:
+        return self.wrapper_handler.try_call(lambda w: w.get_top_headlines(total))
+    def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
+        return self.wrapper_handler.try_call(lambda w: w.get_latest_news(query, total))

src/app/news/base.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel
+
+class Article(BaseModel):
+    source: str = ""
+    time: str = ""
+    title: str = ""
+    description: str = ""
+
+class NewsWrapper:
+    """
+    Base class for news API wrappers.
+    All news API wrappers should inherit from this class and implement the methods.
+    """
+
+    def get_top_headlines(self, total: int = 100) -> list[Article]:
+        """
+        Get top headlines, optionally limited by total.
+        Args:
+            total (int): The maximum number of articles to return.
+        Returns:
+            list[Article]: A list of Article objects.
+        """
+        raise NotImplementedError("This method should be overridden by subclasses")
+
+    def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
+        """
+        Get latest news based on a query.
+        Args:
+            query (str): The search query.
+            total (int): The maximum number of articles to return.
+        Returns:
+            list[Article]: A list of Article objects.
+        """
+        raise NotImplementedError("This method should be overridden by subclasses")
+

src/app/news/cryptopanic_api.py

@@ -0,0 +1,77 @@
+import os
+import requests
+from enum import Enum
+from .base import NewsWrapper, Article
+
+class CryptoPanicFilter(Enum):
+    RISING = "rising"
+    HOT = "hot"
+    BULLISH = "bullish"
+    BEARISH = "bearish"
+    IMPORTANT = "important"
+    SAVED = "saved"
+    LOL = "lol"
+    ANY = ""
+
+class CryptoPanicKind(Enum):
+    NEWS = "news"
+    MEDIA = "media"
+    ALL = "all"
+
+def get_articles(response: dict) -> list[Article]:
+    articles = []
+    if 'results' in response:
+        for item in response['results']:
+            article = Article()
+            article.source = item.get('source', {}).get('title', '')
+            article.time = item.get('published_at', '')
+            article.title = item.get('title', '')
+            article.description = item.get('description', '')
+            articles.append(article)
+    return articles
+
+class CryptoPanicWrapper(NewsWrapper):
+    """
+    A wrapper for the CryptoPanic API (Documentation: https://cryptopanic.com/developers/api/)
+    Requires an API key set in the environment variable CRYPTOPANIC_API_KEY.
+    It is free to use, but has rate limits and restrictions based on the plan type (the free plan is 'developer' with 100 req/month).
+    Supports different plan types via the CRYPTOPANIC_API_PLAN environment variable (developer, growth, enterprise).
+    """
+
+    def __init__(self):
+        self.api_key = os.getenv("CRYPTOPANIC_API_KEY", "")
+        assert self.api_key, "CRYPTOPANIC_API_KEY environment variable not set"
+
+        # Set here for the future, but currently not needed
+        plan_type = os.getenv("CRYPTOPANIC_API_PLAN", "developer").lower()
+        assert plan_type in ["developer", "growth", "enterprise"], "Invalid CRYPTOPANIC_API_PLAN value"
+
+        self.base_url = f"https://cryptopanic.com/api/{plan_type}/v2"
+        self.filter = CryptoPanicFilter.ANY
+        self.kind = CryptoPanicKind.NEWS
+
+    def get_base_params(self) -> dict[str, str]:
+        params = {}
+        params['public'] = 'true' # recommended for app and bots
+        params['auth_token'] = self.api_key
+        params['kind'] = self.kind.value
+        if self.filter != CryptoPanicFilter.ANY:
+            params['filter'] = self.filter.value
+        return params
+
+    def set_filter(self, filter: CryptoPanicFilter):
+        self.filter = filter
+
+    def get_top_headlines(self, total: int = 100) -> list[Article]:
+        return self.get_latest_news("", total) # same endpoint so just call the other method
+
+    def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
+        params = self.get_base_params()
+        params['currencies'] = query
+
+        response = requests.get(f"{self.base_url}/posts/", params=params)
+        assert response.status_code == 200, f"Error fetching data: {response}"
+
+        json_response = response.json()
+        articles = get_articles(json_response)
+        return articles[:total]

src/app/news/duckduckgo.py

@@ -0,0 +1,32 @@
+import json
+from .base import Article, NewsWrapper
+from agno.tools.duckduckgo import DuckDuckGoTools
+
+def create_article(result: dict) -> Article:
+    article = Article()
+    article.source = result.get("source", "")
+    article.time = result.get("date", "")
+    article.title = result.get("title", "")
+    article.description = result.get("body", "")
+    return article
+
+class DuckDuckGoWrapper(NewsWrapper):
+    """
+    A wrapper for DuckDuckGo News search using the Tool from agno.tools.duckduckgo.
+    It can be rewritten to use direct API calls if needed in the future, but currently is easy to write and use.
+    """
+
+    def __init__(self):
+        self.tool = DuckDuckGoTools()
+        self.query = "crypto"
+
+    def get_top_headlines(self, total: int = 100) -> list[Article]:
+        results = self.tool.duckduckgo_news(self.query, max_results=total)
+        json_results = json.loads(results)
+        return [create_article(result) for result in json_results]
+
+    def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
+        results = self.tool.duckduckgo_news(query or self.query, max_results=total)
+        json_results = json.loads(results)
+        return [create_article(result) for result in json_results]
+

src/app/news/gnews_api.py

@@ -0,0 +1,36 @@
+from gnews import GNews
+from .base import Article, NewsWrapper
+
+def result_to_article(result: dict) -> Article:
+    article = Article()
+    article.source = result.get("source", "")
+    article.time = result.get("publishedAt", "")
+    article.title = result.get("title", "")
+    article.description = result.get("description", "")
+    return article
+
+class GoogleNewsWrapper(NewsWrapper):
+    """
+    A wrapper for the Google News RSS Feed (Documentation: https://github.com/ranahaani/GNews/?tab=readme-ov-file#about-gnews)
+    It does not require an API key and is free to use.
+    """
+
+    def get_top_headlines(self, total: int = 100) -> list[Article]:
+        gnews = GNews(language='en', max_results=total, period='7d')
+        results = gnews.get_top_news()
+
+        articles = []
+        for result in results:
+            article = result_to_article(result)
+            articles.append(article)
+        return articles
+
+    def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
+        gnews = GNews(language='en', max_results=total, period='7d')
+        results = gnews.get_news(query)
+
+        articles = []
+        for result in results:
+            article = result_to_article(result)
+            articles.append(article)
+        return articles

src/app/news/news_api.py

@@ -0,0 +1,50 @@
+import os
+import newsapi
+from .base import Article, NewsWrapper
+
+def result_to_article(result: dict) -> Article:
+    article = Article()
+    article.source = result.get("source", {}).get("name", "")
+    article.time = result.get("publishedAt", "")
+    article.title = result.get("title", "")
+    article.description = result.get("description", "")
+    return article
+
+class NewsApiWrapper(NewsWrapper):
+    """
+    A wrapper for the NewsAPI (Documentation: https://newsapi.org/docs/get-started)
+    Requires an API key set in the environment variable NEWS_API_KEY.
+    It is free to use, but has rate limits and restrictions based on the plan type (the free plan is 'developer' with 100 req/day).
+    """
+
+    def __init__(self):
+        api_key = os.getenv("NEWS_API_KEY")
+        assert api_key is not None, "NEWS_API_KEY environment variable not set"
+
+        self.client = newsapi.NewsApiClient(api_key=api_key)
+        self.category = "business" # Cryptocurrency is under business
+        self.language = "en" # TODO Only English articles for now?
+        self.max_page_size = 100
+
+    def get_top_headlines(self, total: int = 100) -> list[Article]:
+        page_size = min(self.max_page_size, total)
+        pages = (total // page_size) + (1 if total % page_size > 0 else 0)
+
+        articles = []
+        for page in range(1, pages + 1):
+            headlines = self.client.get_top_headlines(q="", category=self.category, language=self.language, page_size=page_size, page=page)
+            results = [result_to_article(article) for article in headlines.get("articles", [])]
+            articles.extend(results)
+        return articles
+
+    def get_latest_news(self, query: str, total: int = 100) -> list[Article]:
+        page_size = min(self.max_page_size, total)
+        pages = (total // page_size) + (1 if total % page_size > 0 else 0)
+
+        articles = []
+        for page in range(1, pages + 1):
+            everything = self.client.get_everything(q=query, language=self.language, sort_by="publishedAt", page_size=page_size, page=page)
+            results = [result_to_article(article) for article in everything.get("articles", [])]
+            articles.extend(results)
+        return articles
+

src/app/social/__init.py

@@ -0,0 +1 @@
+from .base import SocialWrapper

src/app/social/base.py

@@ -0,0 +1,22 @@
+from pydantic import BaseModel
+
+
+class SocialPost(BaseModel):
+    time: str = ""
+    title: str = ""
+    description: str = ""
+    comments: list["SocialComment"] = []
+
+    def __str__(self):
+        return f"Title: {self.title}\nDescription: {self.description}\nComments: {len(self.comments)}\n[{" | ".join(str(c) for c in self.comments)}]"
+
+class SocialComment(BaseModel):
+    time: str = ""
+    description: str = ""
+
+    def __str__(self):
+        return f"Time: {self.time}\nDescription: {self.description}"
+
+# TODO IMPLEMENTARLO SE SI USANO PIU' WRAPPER (E QUINDI PIU' SOCIAL)
+class SocialWrapper:
+    pass

src/app/social/reddit.py

@@ -0,0 +1,53 @@
+import os
+from praw import Reddit
+from praw.models import Submission, MoreComments
+from .base import SocialWrapper, SocialPost, SocialComment
+
+MAX_COMMENTS = 5
+
+
+def create_social_post(post: Submission) -> SocialPost:
+    social = SocialPost()
+    social.time = str(post.created)
+    social.title = post.title
+    social.description = post.selftext
+
+    for i, top_comment in enumerate(post.comments):
+        if i >= MAX_COMMENTS:
+            break
+        if isinstance(top_comment, MoreComments): #skip MoreComments objects
+            continue
+
+        comment = SocialComment()
+        comment.time = str(top_comment.created)
+        comment.description = top_comment.body
+        social.comments.append(comment)
+    return social
+
+class RedditWrapper(SocialWrapper):
+    """
+    A wrapper for the Reddit API using PRAW (Python Reddit API Wrapper).
+    Requires the following environment variables to be set:
+    - REDDIT_API_CLIENT_ID
+    - REDDIT_API_CLIENT_SECRET
+    You can get them by creating an app at https://www.reddit.com/prefs/apps
+    """
+
+    def __init__(self):
+        self.client_id = os.getenv("REDDIT_API_CLIENT_ID")
+        assert self.client_id is not None, "REDDIT_API_CLIENT_ID environment variable is not set"
+
+        self.client_secret = os.getenv("REDDIT_API_CLIENT_SECRET")
+        assert self.client_secret is not None, "REDDIT_API_CLIENT_SECRET environment variable is not set"
+
+        self.tool = Reddit(
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            user_agent="upo-appAI",
+        )
+
+    def get_top_crypto_posts(self, limit=5) -> list[SocialPost]:
+        subreddit = self.tool.subreddit("CryptoCurrency")
+        top_posts = subreddit.top(limit=limit, time_filter="week")
+        return [create_social_post(post) for post in top_posts]
+