Merge branch 'main' into 6-telegram-interface

src/app/api/social/__init__.py

@@ -0,0 +1,53 @@
+from agno.tools import Toolkit
+from app.api.wrapper_handler import WrapperHandler
+from app.api.base.social import SocialPost, SocialWrapper
+from app.api.social.reddit import RedditWrapper
+
+__all__ = ["SocialAPIsTool", "RedditWrapper", "SocialPost"]
+
+
+class SocialAPIsTool(SocialWrapper, Toolkit):
+    """
+    Aggregates multiple social media API wrappers and manages them using WrapperHandler.
+    This class supports retrieving top crypto-related posts by querying multiple sources:
+    - RedditWrapper
+
+    By default, it returns results from the first successful wrapper. 
+    Optionally, it can be configured to collect posts from all wrappers.
+    If no wrapper succeeds, an exception is raised.
+    """
+
+    def __init__(self):
+        """
+        Initialize the SocialAPIsTool with multiple social media API wrappers.
+        The tool uses WrapperHandler to manage and invoke the different social media API wrappers.
+        The following wrappers are included in this order:
+        - RedditWrapper.
+        """
+
+        wrappers: list[type[SocialWrapper]] = [RedditWrapper]
+        self.handler = WrapperHandler.build_wrappers(wrappers)
+
+        Toolkit.__init__( # type: ignore
+            self,
+            name="Socials Toolkit",
+            tools=[
+                self.get_top_crypto_posts,
+                self.get_top_crypto_posts_aggregated,
+            ],
+        )
+
+    def get_top_crypto_posts(self, limit: int = 5) -> list[SocialPost]:
+        return self.handler.try_call(lambda w: w.get_top_crypto_posts(limit))
+
+    def get_top_crypto_posts_aggregated(self, limit_per_wrapper: int = 5) -> dict[str, list[SocialPost]]:
+        """
+        Calls get_top_crypto_posts on all wrappers/providers and returns a dictionary mapping their names to their posts.
+        Args:
+            limit_per_wrapper (int): Maximum number of posts to retrieve from each provider.
+        Returns:
+            dict[str, list[SocialPost]]: A dictionary where keys are wrapper names and values are lists of SocialPost objects.
+        Raises:
+            Exception: If all wrappers fail to provide results.
+        """
+        return self.handler.try_call_all(lambda w: w.get_top_crypto_posts(limit_per_wrapper))

src/app/api/social/reddit.py

@@ -0,0 +1,68 @@
+import os
+from praw import Reddit # type: ignore
+from praw.models import Submission # type: ignore
+from app.api.base.social import SocialWrapper, SocialPost, SocialComment
+
+
+MAX_COMMENTS = 5
+# metterne altri se necessario.
+# fonti: https://lkiconsulting.io/marketing/best-crypto-subreddits/
+SUBREDDITS = [
+    "CryptoCurrency", 
+    "Bitcoin",
+    "Ethereum",
+    "CryptoMarkets",
+    "Dogecoin",
+    "Altcoin",
+    "DeFi",
+    "NFT",
+    "BitcoinBeginners",
+    "CryptoTechnology",
+    "btc" # alt subs of Bitcoin
+]
+
+
+def extract_post(post: Submission) -> SocialPost:
+    social = SocialPost()
+    social.time = str(post.created)
+    social.title = post.title
+    social.description = post.selftext
+
+    for top_comment in post.comments:
+        comment = SocialComment()
+        comment.time = str(top_comment.created)
+        comment.description = top_comment.body
+        social.comments.append(comment)
+
+        if len(social.comments) >= MAX_COMMENTS:
+            break
+    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):
+        client_id = os.getenv("REDDIT_API_CLIENT_ID")
+        assert client_id, "REDDIT_API_CLIENT_ID environment variable is not set"
+
+        client_secret = os.getenv("REDDIT_API_CLIENT_SECRET")
+        assert client_secret, "REDDIT_API_CLIENT_SECRET environment variable is not set"
+
+        self.tool = Reddit(
+            client_id=client_id,
+            client_secret=client_secret,
+            user_agent="upo-appAI",
+            check_for_async=False,
+        )
+        self.subreddits = self.tool.subreddit("+".join(SUBREDDITS))
+
+    def get_top_crypto_posts(self, limit: int = 5) -> list[SocialPost]:
+        top_posts = self.subreddits.top(limit=limit, time_filter="week")
+        return [extract_post(post) for post in top_posts]