diff --git a/demos/api_socials_providers.py b/demos/api_socials_providers.py new file mode 100644 index 0000000..aa4b14f --- /dev/null +++ b/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() diff --git a/src/app/api/core/social.py b/src/app/api/core/social.py index ca921a0..05953a3 100644 --- a/src/app/api/core/social.py +++ b/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): """ diff --git a/src/app/api/social/x.py b/src/app/api/social/x.py index a1b1bd4..c14f5b6 100644 --- a/src/app/api/social/x.py +++ b/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 tweets[:limit]: + for tweet in json_result.get('list', []): + time = datetime.fromisoformat(tweet['createdAt']) social_post = SocialPost() - social_post.time = tweet['createdAt'] - social_post.title = str(user) + " tweeted: " + social_post.set_timestamp(timestamp_s=int(time.timestamp())) + 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 diff --git a/tests/api/test_social_4chan.py b/tests/api/test_social_4chan.py index b39a36d..dcf42d2 100644 --- a/tests/api/test_social_4chan.py +++ b/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 re.match(r'\d{4}-\d{2}-\d{2}', post.time) + assert post.timestamp != "" + assert re.match(r'\d{4}-\d{2}-\d{2}', post.timestamp) assert isinstance(post.comments, list) diff --git a/tests/api/test_social_reddit.py b/tests/api/test_social_reddit.py index a83fe8a..adb4e13 100644 --- a/tests/api/test_social_reddit.py +++ b/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 diff --git a/tests/api/test_social_x_api.py b/tests/api/test_social_x_api.py index 15f39c3..39f75f9 100644 --- a/tests/api/test_social_x_api.py +++ b/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) diff --git a/tests/tools/test_socials_tool.py b/tests/tools/test_socials_tool.py index c021a90..3a481f7 100644 --- a/tests/tools/test_socials_tool.py +++ b/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