commit 9f5c6a9647d277a874d6bda265b286faddfd5a5d Author: DylanDeFaoite Date: Thu Nov 6 14:15:50 2025 +0000 Initial Commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..42f7d5d --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.vscode +__pycache__/ +*.pyc \ No newline at end of file diff --git a/connectors/base_connector.py b/connectors/base_connector.py new file mode 100644 index 0000000..e7f4478 --- /dev/null +++ b/connectors/base_connector.py @@ -0,0 +1,18 @@ +from abc import abstractmethod +from abc import ABC +from dto.post import Post +from dto.user import User +from datetime import datetime + +class BaseConnector(ABC): + @abstractmethod + def get_top_posts(self, limit: int = 10, timeframe: str = 'day') -> list[Post]: + pass + + @abstractmethod + def search_posts(self, search: str, limit: int = 10, before = None, after = None) -> list[Post]: + pass + + @abstractmethod + def get_user(self, username: str) -> User: + pass \ No newline at end of file diff --git a/connectors/reddit_connector.py b/connectors/reddit_connector.py new file mode 100644 index 0000000..80f0244 --- /dev/null +++ b/connectors/reddit_connector.py @@ -0,0 +1,68 @@ +from connectors.base_connector import BaseConnector +from dto.post import Post +from dto.user import User +import requests + +class RedditConnector(BaseConnector): + def __init__(self): + self.url = "https://www.reddit.com/" + + def get_top_posts(self, limit: int = 10, timeframe: str = 'day') -> list[Post]: + params = { + 'limit': limit, + 't': timeframe + } + url = f"{self.url}top.json" + data = self._fetch_data(url, params) + return self._parse_posts(data) + + def search_posts(self, search: str, limit: int = 10, before = None, after = None) -> list[Post]: + params = { + 'q': search, + 'limit': limit, + 'before': before, + 'after': after, + 'sort': 'relevance', + 't': 'day' + } + url = f"{self.url}search.json" + data = self._fetch_data(url, params) + return self._parse_posts(data) + + def get_user(self, username: str) -> User: + data = self._fetch_data(f"user/{username}/about.json", {}) + return self._parse_user(data) + + def _parse_posts(self, data) -> list[Post]: + posts = [] + for item in data['data']['children']: + post_data = item['data'] + post = Post( + author=post_data['author'], + title=post_data['title'], + content=post_data.get('selftext', ''), + url=post_data['url'], + timestamp=post_data['created_utc']) + post.subreddit = post_data['subreddit'] + post.upvotes = post_data['ups'] + + posts.append(post) + return posts + + def _parse_user(self, data) -> User: + user_data = data['data'] + user = User( + username=user_data['name'], + created_utc=user_data['created_utc']) + user.karma = user_data['total_karma'] + return user + + def _fetch_data(self, endpoint: str, params: dict) -> dict: + url = f"{self.url}{endpoint}" + try: + response = requests.get(url, headers={'User-agent': 'your bot 0.1'}, params=params) + response.raise_for_status() + return response.json() + except requests.RequestException as e: + print(f"Error fetching data from Reddit API: {e}") + return {} \ No newline at end of file diff --git a/dto/post.py b/dto/post.py new file mode 100644 index 0000000..0e1eeda --- /dev/null +++ b/dto/post.py @@ -0,0 +1,11 @@ +class Post: + def __init__(self, author, title, content, url, timestamp): + self.author = author + self.title = title + self.content = content + self.url = url + self.timestamp = timestamp + + # Optionals + self.subreddit = None + self.upvotes = None \ No newline at end of file diff --git a/dto/user.py b/dto/user.py new file mode 100644 index 0000000..65a3984 --- /dev/null +++ b/dto/user.py @@ -0,0 +1,8 @@ +# Generic User Data Transfer Object for social media platforms +class User: + def __init__(self, username: str, created_utc: int, ): + self.username = username + self.created_utc = created_utc + + # Optionals + self.karma = None \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..190ce31 --- /dev/null +++ b/main.py @@ -0,0 +1,9 @@ +from connectors.reddit_connector import RedditConnector + +if __name__ == "__main__": + connector = RedditConnector() + + search_results = connector.search_posts(search="NVIDIA", limit=10) + for post in search_results: + print(f"Title: {post.title}\nAuthor: {post.author}\nSubreddit: {post.subreddit}\nUpvotes: {post.upvotes}") + print("---") \ No newline at end of file