"""
title: Reddit
author: @nathanwindisch
author_url: https://git.wnd.sh/owui-tools/reddit
funding_url: https://patreon.com/NathanWindisch
version: 0.0.1
changelog:
- 0.0.1 - Initial upload to openwebui community.
- 0.0.2 - Renamed from "Reddit Feeds" to just "Reddit".
- 0.0.3 - Updated author_url in docstring to point to
git repo.
"""
import re
import json
import requests
from typing import Awaitable, Callable
from pydantic import BaseModel, Field
from requests.models import Response
def parse_reddit_page(response: Response):
data = json.loads(response.content)
output = []
if "data" not in data: return output
if "children" not in data["data"]: return output
for item in data["data"]["children"]: output.append(item)
return output
def parse_posts(data: list):
posts = []
for item in data:
if item["kind"] != "t3": continue
item = item["data"]
posts.append({
"id": item["name"],
"title": item["title"],
"description": item["selftext"],
"link": item["url"],
"author_username": item["author"],
"author_id": item["author_fullname"],
"subreddit_name": item["subreddit"],
"subreddit_id": item["subreddit_id"],
"subreddit_subscribers": item["subreddit_subscribers"],
"score": item["score"],
"upvotes": item["ups"],
"downvotes": item["downs"],
"upvote_ratio": item["upvote_ratio"],
"total_comments": item["num_comments"],
"total_crossposts": item["num_crossposts"],
"total_awards": item["total_awards_received"],
"domain": item["domain"],
"flair_text": item["link_flair_text"],
"media_embed": item["media_embed"],
"is_pinned": item["pinned"],
"is_self": item["is_self"],
"is_video": item["is_video"],
"is_media_only": item["media_only"],
"is_over_18": item["over_18"],
"is_edited": item["edited"],
"is_hidden": item["hidden"],
"is_archived": item["archived"],
"is_locked": item["locked"],
"is_quarantined": item["quarantine"],
"is_spoiler": item["spoiler"],
"is_stickied": item["stickied"],
"is_send_replies": item["send_replies"],
"published_at": item["created_utc"],
})
return posts
def parse_comments(data: list):
comments = []
for item in data:
if item["kind"] != "t1": continue
item = item["data"]
comments.append({
"id": item["name"],
"body": item["body"],
"link": item["permalink"],
"post_id": item["link_id"],
"post_title": item["link_title"],
"post_link": item["link_permalink"],
"author_username": item["author"],
"author_id": item["author_fullname"],
"subreddit_name": item["subreddit"],
"subreddit_id": item["subreddit_id"],
"score": item["score"],
"upvotes": item["ups"],
"downvotes": item["downs"],
"total_comments": item["num_comments"],
"total_awards": item["total_awards_received"],
"is_edited": item["edited"],
"is_archived": item["archived"],
"is_locked": item["locked"],
"is_quarantined": item["quarantine"],
"is_stickied": item["stickied"],
"is_send_replies": item["send_replies"],
"published_at": item["created_utc"],
})
return comments
class Tools:
def __init__(self): pass
class UserValves(BaseModel):
USER_AGENT: str = Field(
default="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
description="The user agent to use when making requests to Reddit."
)
async def get_subreddit_feed(
self,
subreddit: str,
__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.
"""
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"
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}"