demos/api_socials_providers.py

@@ -0,0 +1,17 @@
+from dotenv import load_dotenv
+from app.api.tools import SocialAPIsTool
+
+def main():
+    api = SocialAPIsTool()
+    articles_aggregated = api.get_top_crypto_posts_aggregated(limit_per_wrapper=2)
+    for provider, posts in articles_aggregated.items():
+        print("===================================")
+        print(f"Provider: {provider}")
+        for post in posts:
+            print(f"== [{post.timestamp}] - {post.title} ==")
+            print(f"   {post.description}")
+            print(f"   {len(post.comments)}")
+
+if __name__ == "__main__":
+    load_dotenv()
+    main()

src/app/api/core/social.py

@@ -9,14 +9,14 @@ class SocialPost(BaseModel):
     """
     Represents a social media post with time, title, description, and comments.
     """
-    time: str = ""
+    timestamp: str = ""
     title: str = ""
     description: str = ""
     comments: list["SocialComment"] = []
 
     def set_timestamp(self, timestamp_ms: int | None = None, timestamp_s: int | None = None) -> None:
         """ Use the unified_timestamp function to set the time."""
-        self.time = unified_timestamp(timestamp_ms, timestamp_s)
+        self.timestamp = unified_timestamp(timestamp_ms, timestamp_s)
 
 class SocialComment(BaseModel):
     """

src/app/api/social/x.py

@@ -2,6 +2,7 @@ import os
 import json
 import subprocess
 from shutil import which
+from datetime import datetime
 from app.api.core.social import SocialWrapper, SocialPost
 
 
@@ -28,19 +29,20 @@ class XWrapper(SocialWrapper):
 
 
     def get_top_crypto_posts(self, limit:int = 5) -> list[SocialPost]:
-        social_posts: list[SocialPost] = []
+        posts: list[SocialPost] = []
 
         for user in X_USERS:
-            process = subprocess.run(f"rettiwt -k {self.api_key} tweet search -f {str(user)}", capture_output=True)
+            cmd = f"rettiwt -k {self.api_key} tweet search {limit} -f {str(user)}"
+            process = subprocess.run(cmd, capture_output=True, shell=True)
             results = process.stdout.decode()
             json_result = json.loads(results)
 
-            tweets = json_result['list']
+            for tweet in json_result.get('list', []):
-            for tweet in tweets[:limit]:
+                time = datetime.fromisoformat(tweet['createdAt'])
                 social_post = SocialPost()
-                social_post.time = tweet['createdAt']
+                social_post.set_timestamp(timestamp_s=int(time.timestamp()))
-                social_post.title = str(user) + " tweeted: "
+                social_post.title = f"{user} tweeted: "
                 social_post.description = tweet['fullText']
-                social_posts.append(social_post)
+                posts.append(social_post)
 
-        return social_posts
+        return posts

tests/api/test_social_4chan.py

@@ -16,7 +16,7 @@ class TestChanWrapper:
         assert len(posts) == 2
         for post in posts:
             assert post.title != ""
-            assert post.time != ""
+            assert post.timestamp != ""
-            assert re.match(r'\d{4}-\d{2}-\d{2}', post.time)
+            assert re.match(r'\d{4}-\d{2}-\d{2}', post.timestamp)
             assert isinstance(post.comments, list)
 

tests/api/test_social_reddit.py

@@ -19,7 +19,7 @@ class TestRedditWrapper:
         assert len(posts) == 2
         for post in posts:
             assert post.title != ""
-            assert re.match(r'\d{4}-\d{2}-\d{2}', post.time)
+            assert re.match(r'\d{4}-\d{2}-\d{2}', post.timestamp)
 
             assert isinstance(post.comments, list)
             assert len(post.comments) <= MAX_COMMENTS

tests/api/test_social_x_api.py

@@ -1,11 +1,13 @@
 import os
 import re
 import pytest
+from shutil import which
 from app.api.social.x import XWrapper
 
 @pytest.mark.social
 @pytest.mark.api
 @pytest.mark.skipif(not os.getenv("X_API_KEY"), reason="X_API_KEY not set in environment variables")
+@pytest.mark.skipif(which('rettiwt') is None, reason="rettiwt not installed")
 class TestXWrapper:
     def test_initialization(self):
         wrapper = XWrapper()
@@ -18,5 +20,5 @@ class TestXWrapper:
         assert len(posts) == 2
         for post in posts:
             assert post.title != ""
-            assert re.match(r'\d{4}-\d{2}-\d{2}', post.time)
+            assert re.match(r'\d{4}-\d{2}-\d{2}', post.timestamp)
             assert isinstance(post.comments, list)

tests/tools/test_socials_tool.py

@@ -17,7 +17,7 @@ class TestSocialAPIsTool:
         assert len(result) > 0
         for post in result:
             assert post.title is not None
-            assert post.time is not None
+            assert post.timestamp is not None
 
     def test_social_api_tool_get_top__all_results(self):
         tool = SocialAPIsTool()
@@ -27,4 +27,4 @@ class TestSocialAPIsTool:
         for _provider, posts in result.items():
             for post in posts:
                 assert post.title is not None
-                assert post.time is not None
+                assert post.timestamp is not None