owui-tools/reddit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updated Tool#get_subreddit_feed to be the more generic Tool#get_reddit_feed, which can take in both user and subreddit Reddit .json links.

Browse Source
This commit is contained in:
Nathan Windisch 2024-07-24 22:42:25 +01:00
parent 8e6c1549df
commit c656a67961
1 changed files with 20 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
src/reddit.py
View File

@ -170,84 +170,34 @@ class Tools:
description="The user agent to use when making requests to Reddit."
)
async def get_subreddit_feed(
async def get_reddit_feed(
self,
subreddit: str,
id: str,
page_type: RedditPageType,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {},
) -> str:
"""
Get the latest posts from a subreddit, as an array of JSON objects with the following properties: 'id', 'title', 'description', 'link', 'author_username', 'author_id', 'subreddit_name', 'subreddit_id', 'subreddit_subscribers', 'score', 'upvotes', 'downvotes', 'upvote_ratio', 'total_comments', 'total_crossposts', 'total_awards', 'domain', 'flair_text', 'media_embed', 'is_pinned', 'is_self', 'is_video', 'is_media_only', 'is_over_18', 'is_edited', 'is_hidden', 'is_archived', 'is_locked', 'is_quarantined', 'is_spoiler', 'is_stickied', 'is_send_replies', 'published_at'.
:param subreddit: The subreddit to get the latest posts from.
:return: A list of posts with the previously mentioned properties, or an error message.
Retrieves the popular posts from a specific feed, either a /u/username or /r/subreddit feed, or the comments on either.
:param id: The ID of the feed to retrieve, such as a username or a subreddit name. Additionally, a post ID can be appended to either to retrieve comments from a specific post.
:param page_type: The type of page to retrieve, must be 'USER', 'SUBREDDIT', 'USER_COMMENTS' or 'SUBREDDIT_COMMENTS'.
:return: An object containing a list of posts, comments, and the next ID to use under the 'after' key, or an error message.
Note: The 'USER' page_type will retrieve both posts and comments, while the 'SUBREDDIT' page_type will only retrieve posts (unless a post id is provided as well, and the page_type is 'SUBREDDIT_COMMENTS').
"""
headers = { "User-Agent": __user__["valves"].USER_AGENT }
await __event_emitter__({ "data": { "description": f"Starting retrieval for r/{subreddit}'s Reddit Feed...", "status": "in_progress", "done": False }, "type": "status" })
if subreddit == "":
await __event_emitter__({ "data": { "description": f"Error: No subreddit provided.", "status": "complete", "done": True }, "type": "status" })
return "Error: No subreddit provided"
subreddit = subreddit.replace("/r/", "").replace("r/", "")
if not re.match(r"^[A-Za-z0-9_]{2,21}$", subreddit):
await __event_emitter__({ "data": { "description": f"Error: Invalid subreddit name '{subreddit}' (either too long or two short).", "status": "complete", "done": True }, "type": "status" })
return "Error: Invalid subreddit name"
id = id.replace("/r/", "").replace("/u/", "").replace("u/", "").replace("r/", "") # Strip any /r/ or /u/ from the ID
# This accounts for the type being dropped by OpenWebUI
if not isinstance(page_type, RedditPageType):
try:
response = requests.get(f"https://reddit.com/r/{subreddit}.json", headers=headers)
if not response.ok:
await __event_emitter__({ "data": { "description": f"Error: Failed to retrieve r/{subreddit}'s Reddit Feed: {response.status_code}.", "status": "complete", "done": True }, "type": "status" })
return f"Error: {response.status_code}"
else:
output = parse_posts(parse_reddit_page(response))
await __event_emitter__({ "data": { "description": f"Retrieved {len(output)} posts from r/{subreddit}'s Reddit Feed.", "status": "complete", "done": True }, "type": "status" })
return json.dumps(output)
except Exception as e:
await __event_emitter__({ "data": { "description": f"Failed to retrieve any posts from r/{subreddit}'s Reddit Feed: {e}.", "status": "complete", "done": True }, "type": "status" })
return f"Error: {e}"
async def get_user_feed(
self,
username: str,
__event_emitter__: Callable[[dict], Awaitable[None]],
__user__: dict = {},
) -> str:
"""
Get the latest posts from a given user, as a JSON object with an array of 'post' objects with the following properties: 'id', 'title', 'description', 'link', 'author_username', 'author_id', 'subreddit_name', 'subreddit_id', 'subreddit_subscribers', 'score', 'upvotes', 'downvotes', 'upvote_ratio', 'total_comments', 'total_crossposts', 'total_awards', 'domain', 'flair_text', 'media_embed', 'is_pinned', 'is_self', 'is_video', 'is_media_only', 'is_over_18', 'is_edited', 'is_hidden', 'is_archived', 'is_locked', 'is_quarantined', 'is_spoiler', 'is_stickied', 'is_send_replies', 'published_at'.
Additionally, the resultant object will also contain an array of 'comment' objects with the following properties: 'id', 'body', 'link', 'post_id', 'post_title', 'post_link', 'author_id', 'post_author_username', 'subreddit_name', 'subreddit_id', 'subreddit_subscribers', 'score', 'upvotes', 'downvotes', 'total_comments', 'total_awards', 'is_edited', 'is_archived', 'is_locked', 'is_quarantined', 'is_stickied', 'is_send_replies', 'published_at'.
:param username: The username to get the latest posts from.
:return: A object with list of posts and a list of comments (both with the previously mentioned properties), or an error message.
"""
headers = { "User-Agent": __user__["valves"].USER_AGENT }
await __event_emitter__({ "data": { "description": f"Starting retrieval for u/{username}'s Reddit Feed...", "status": "in_progress", "done": False }, "type": "status" })
if username == "":
await __event_emitter__({ "data": { "description": f"Error: No username provided.", "status": "complete", "done": True }, "type": "status" })
return "Error: No username provided."
username = username.replace("/u/", "").replace("u/", "")
if not re.match(r"^[A-Za-z0-9_]{3,20}$", username):
await __event_emitter__({ "data": { "description": f"Error: Invalid username '{username}' (either too long or two short).", "status": "complete", "done": True }, "type": "status" })
return "Error: Invalid username."
try:
response = requests.get(f"https://reddit.com/u/{username}.json", headers=headers)
if not response.ok:
await __event_emitter__({ "data": { "description": f"Error: Failed to retrieve u/{username}'s Reddit Feed: {response.status_code}.", "status": "complete", "done": True }, "type": "status" })
return f"Error: {response.status_code}"
else:
page = parse_reddit_page(response) # user pages can have both posts and comments.
posts = parse_posts(page)
comments = parse_comments(page)
await __event_emitter__({ "data": { "description": f"Retrieved {len(posts)} posts and {len(comments)} comments from u/{username}'s Reddit Feed.", "status": "complete", "done": True }, "type": "status" })
return json.dumps({ "posts": posts, "comments": comments })
except Exception as e:
await __event_emitter__({ "data": { "description": f"Failed to retrieve any posts from u/{username}'s Reddit Feed: {e}.", "status": "complete", "done": True }, "type": "status" })
return f"Error: {e}"
page_type = RedditPageType[page_type]
except ValueError:
await __event_emitter__({ "data": { "description": f"Error: Invalid page type '{page_type}', try 'USER', 'SUBREDDIT', 'USER_COMMENTS' or 'SUBREDDIT_COMMENTS'.", "status": "complete", "done": True }, "type": "status" })
return f"Error: Invalid page type '{page_type}', try either 'USER', 'SUBREDDIT', 'USER_COMMENTS' or 'SUBREDDIT_COMMENTS'."
await __event_emitter__({ "data": { "description": f"Starting retrieval for {page_type.value}/{id}...", "status": "in_progress", "done": False }, "type": "status" })
page = RedditPage(id, page_type).get_page()
await __event_emitter__({ "data": { "description": f"Retrieved {len(page.posts)} posts and {len(page.comments)} comments from {page_type.value}/{id}.", "status": "complete", "done": True }, "type": "status" })
return str(page)
async def main():