unified_timestamp

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')

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:
-        """
+        """ Use the unified_timestamp function to set the timestamp."""
-        Sets the timestamp from milliseconds or seconds.
+        self.timestamp = unified_timestamp(timestamp_ms, timestamp_s)
-        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')
 
     @staticmethod
     def aggregate(prices: dict[str, list['Price']]) -> list['Price']:

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:
     """

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:
+    def __time_str(self, timestamp: str) -> int:
-        """Converte una stringa da MM/GG/AA di timestamp nel formato GG/MM/AA"""
+        """Converte una stringa da MM/GG/AA(DAY)HH:MM:SS di 4chan a millisecondi"""
-        if len(timestamp) < 8: return ""
+        time = datetime.strptime(timestamp, "%m/%d/%y(%a)%H:%M:%S")
-
+        return int(time.timestamp() * 1000)
-        month = timestamp[:2]
-        day = timestamp[3:5]
-        year = timestamp[6:8]
-        return f"{day}/{month}/{year}"
 
     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]

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)
 

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)
+

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:

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)