diff --git a/src/app/api/core/__init__.py b/src/app/api/core/__init__.py index e69de29..3cddea7 100644 --- a/src/app/api/core/__init__.py +++ b/src/app/api/core/__init__.py @@ -0,0 +1,22 @@ +from datetime import datetime + + +def unified_timestamp(timestamp_ms: int | None = None, timestamp_s: int | None = None) -> str: + """ + Transform the timestamp from milliseconds or seconds to a unified string format. + The resulting string is a formatted string 'YYYY-MM-DD HH:MM'. + Args: + timestamp_ms: Timestamp in milliseconds. + timestamp_s: Timestamp in seconds. + Raises: + ValueError: If neither timestamp_ms nor timestamp_s is provided. + """ + if timestamp_ms is not None: + timestamp = timestamp_ms // 1000 + elif timestamp_s is not None: + timestamp = timestamp_s + else: + raise ValueError("Either timestamp_ms or timestamp_s must be provided") + assert timestamp > 0, "Invalid timestamp data received" + + return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M') \ No newline at end of file diff --git a/src/app/api/core/markets.py b/src/app/api/core/markets.py index 8b6c754..6b53f61 100644 --- a/src/app/api/core/markets.py +++ b/src/app/api/core/markets.py @@ -1,6 +1,6 @@ import statistics -from datetime import datetime from pydantic import BaseModel +from app.api.core import unified_timestamp class ProductInfo(BaseModel): @@ -64,24 +64,8 @@ class Price(BaseModel): """Timestamp in format YYYY-MM-DD HH:MM""" def set_timestamp(self, timestamp_ms: int | None = None, timestamp_s: int | None = None) -> None: - """ - Sets the timestamp from milliseconds or seconds. - The timestamp is saved as a formatted string 'YYYY-MM-DD HH:MM'. - Args: - timestamp_ms: Timestamp in milliseconds. - timestamp_s: Timestamp in seconds. - Raises: - ValueError: If neither timestamp_ms nor timestamp_s is provided. - """ - if timestamp_ms is not None: - timestamp = timestamp_ms // 1000 - elif timestamp_s is not None: - timestamp = timestamp_s - else: - raise ValueError("Either timestamp_ms or timestamp_s must be provided") - assert timestamp > 0, "Invalid timestamp data received" - - self.timestamp = datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M') + """ Use the unified_timestamp function to set the timestamp.""" + self.timestamp = unified_timestamp(timestamp_ms, timestamp_s) @staticmethod def aggregate(prices: dict[str, list['Price']]) -> list['Price']: diff --git a/src/app/api/core/social.py b/src/app/api/core/social.py index deca505..fe4d5bf 100644 --- a/src/app/api/core/social.py +++ b/src/app/api/core/social.py @@ -1,4 +1,5 @@ from pydantic import BaseModel +from app.api.core import unified_timestamp @@ -13,6 +14,10 @@ class SocialPost(BaseModel): 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) + class SocialComment(BaseModel): """ Represents a comment on a social media post. @@ -20,6 +25,10 @@ class SocialComment(BaseModel): time: str = "" description: str = "" + 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) + class SocialWrapper: """ diff --git a/src/app/api/social/chan.py b/src/app/api/social/chan.py index bea039d..df89f31 100644 --- a/src/app/api/social/chan.py +++ b/src/app/api/social/chan.py @@ -5,6 +5,7 @@ import re import html import requests from bs4 import BeautifulSoup +from datetime import datetime from app.api.core.social import * @@ -12,14 +13,10 @@ class ChanWrapper(SocialWrapper): def __init__(self): super().__init__() - def __time_str(self, timestamp: str) -> str: - """Converte una stringa da MM/GG/AA di timestamp nel formato GG/MM/AA""" - if len(timestamp) < 8: return "" - - month = timestamp[:2] - day = timestamp[3:5] - year = timestamp[6:8] - return f"{day}/{month}/{year}" + def __time_str(self, timestamp: str) -> int: + """Converte una stringa da MM/GG/AA(DAY)HH:MM:SS di 4chan a millisecondi""" + time = datetime.strptime(timestamp, "%m/%d/%y(%a)%H:%M:%S") + return int(time.timestamp() * 1000) def __unformat_html_str(self, html_element: str) -> str: """Pulisce il commento rimuovendo HTML e formattazioni inutili""" @@ -78,15 +75,16 @@ class ChanWrapper(SocialWrapper): if not comment: continue - social_comment = SocialComment(time=time, description=comment) + social_comment = SocialComment(description=comment) + social_comment.set_timestamp(timestamp_ms=time) comments_list.append(social_comment) social_post: SocialPost = SocialPost( - time=time, title=title, description=thread_description, comments=comments_list ) + social_post.set_timestamp(timestamp_ms=time) social_posts.append(social_post) return social_posts[:limit] diff --git a/src/app/api/social/reddit.py b/src/app/api/social/reddit.py index c247cac..201166c 100644 --- a/src/app/api/social/reddit.py +++ b/src/app/api/social/reddit.py @@ -23,13 +23,13 @@ SUBREDDITS = [ def extract_post(post: Submission) -> SocialPost: social = SocialPost() - social.time = str(post.created) + social.set_timestamp(timestamp_ms=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.set_timestamp(timestamp_ms=top_comment.created) comment.description = top_comment.body social.comments.append(comment) diff --git a/tests/api/test_social_4chan.py b/tests/api/test_social_4chan.py new file mode 100644 index 0000000..b39a36d --- /dev/null +++ b/tests/api/test_social_4chan.py @@ -0,0 +1,22 @@ +import re +import pytest +from app.api.social.chan import ChanWrapper + +@pytest.mark.social +@pytest.mark.api +class TestChanWrapper: + def test_initialization(self): + wrapper = ChanWrapper() + assert wrapper is not None + + def test_get_top_crypto_posts(self): + wrapper = ChanWrapper() + posts = wrapper.get_top_crypto_posts(limit=2) + assert isinstance(posts, list) + 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 isinstance(post.comments, list) + diff --git a/tests/api/test_reddit.py b/tests/api/test_social_reddit.py similarity index 92% rename from tests/api/test_reddit.py rename to tests/api/test_social_reddit.py index d4533a5..a83fe8a 100644 --- a/tests/api/test_reddit.py +++ b/tests/api/test_social_reddit.py @@ -1,4 +1,5 @@ import os +import re import pytest from app.api.social.reddit import MAX_COMMENTS, RedditWrapper @@ -18,6 +19,8 @@ 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 isinstance(post.comments, list) assert len(post.comments) <= MAX_COMMENTS for comment in post.comments: diff --git a/tests/api/test_social_x_api.py b/tests/api/test_social_x_api.py new file mode 100644 index 0000000..15f39c3 --- /dev/null +++ b/tests/api/test_social_x_api.py @@ -0,0 +1,22 @@ +import os +import re +import pytest +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") +class TestXWrapper: + def test_initialization(self): + wrapper = XWrapper() + assert wrapper is not None + + def test_get_top_crypto_posts(self): + wrapper = XWrapper() + posts = wrapper.get_top_crypto_posts(limit=2) + assert isinstance(posts, list) + assert len(posts) == 2 + for post in posts: + assert post.title != "" + assert re.match(r'\d{4}-\d{2}-\d{2}', post.time) + assert isinstance(post.comments, list)