355 lines
14 KiB
Python
355 lines
14 KiB
Python
import os
|
|
import praw
|
|
import requests
|
|
from redvid import Downloader
|
|
import yt_dlp
|
|
import re
|
|
from datetime import datetime
|
|
|
|
try:
|
|
from logindata import REDDIT_USERNAME, REDDIT_PASSWORD
|
|
from logindata import REDDIT_CLIENT_ID, REDDIT_SECRET
|
|
except ImportError:
|
|
REDDIT_USERNAME = os.getenv("REDDIT_USERNAME")
|
|
REDDIT_PASSWORD = os.getenv("REDDIT_PASSWORD")
|
|
REDDIT_CLIENT_ID = os.getenv("REDDIT_CLIENT_ID")
|
|
REDDIT_SECRET = os.getenv("REDDIT_SECRET")
|
|
|
|
IMAGE_EXTENSIONS = ["gif", "gifv", "jpg", "jpeg", "png"]
|
|
VIDEO_EXTENSIONS = ["mp4"]
|
|
PLATFORMS = ["redgifs.com", "gfycat.com", "imgur.com", "youtube.com"]
|
|
|
|
|
|
def make_client():
|
|
"""Creates a PRAW client with the details in the secrets.py file."""
|
|
|
|
print(REDDIT_USERNAME)
|
|
|
|
return praw.Reddit(
|
|
username=REDDIT_USERNAME,
|
|
password=REDDIT_PASSWORD,
|
|
client_id=REDDIT_CLIENT_ID,
|
|
client_secret=REDDIT_SECRET,
|
|
user_agent="reddit-save",
|
|
)
|
|
|
|
|
|
def get_previous(location, html_file):
|
|
html_files = [f for f in os.listdir(location) if f.endswith(".html")]
|
|
pattern = html_file.replace(".html", r"\.(\d+)?\.html")
|
|
matches = [re.match(pattern, f) for f in html_files]
|
|
matches = [m[0] for m in matches if m]
|
|
matches.sort(key=lambda x: int(x.split(".")[1]))
|
|
existing_ids = []
|
|
existing_posts_html = []
|
|
existing_comments_html = []
|
|
if html_file in html_files: matches.append(html_file)
|
|
for match in matches:
|
|
with open(os.path.join(location, match), encoding="utf-8") as f:
|
|
current_html = f.read()
|
|
for id in re.findall(r'id="(.+?)"', current_html):
|
|
if id not in existing_ids:
|
|
existing_ids.append(id)
|
|
posts = re.findall(
|
|
r'(<div class="post"[\S\n\t\v ]+?<!--postend--><\/div>)',
|
|
current_html
|
|
)
|
|
comments = re.findall(
|
|
r'(<div class="comment"[\S\n\t\v ]+?<!--commentend--><\/div>)',
|
|
current_html
|
|
)
|
|
for post in posts:
|
|
if post not in existing_posts_html:
|
|
existing_posts_html.append(post)
|
|
for comment in comments:
|
|
if comment not in existing_comments_html:
|
|
existing_comments_html.append(comment)
|
|
return existing_ids, existing_posts_html, existing_comments_html
|
|
|
|
|
|
def get_saved_posts(client):
|
|
"""Gets a list of posts that the user has saved."""
|
|
|
|
return [
|
|
saved for saved in client.user.me().saved(limit=None)
|
|
if saved.__class__.__name__ == "Submission"
|
|
]
|
|
|
|
|
|
def get_upvoted_posts(client):
|
|
"""Gets a list of posts that the user has upvoted."""
|
|
|
|
return [
|
|
upvoted for upvoted in client.user.me().upvoted(limit=None)
|
|
if upvoted.__class__.__name__ == "Submission"
|
|
]
|
|
|
|
|
|
def get_saved_comments(client):
|
|
"""Gets a list of comments that the user has saved."""
|
|
|
|
return [
|
|
saved for saved in client.user.me().saved(limit=None)
|
|
if saved.__class__.__name__ != "Submission"
|
|
]
|
|
|
|
|
|
def get_user_posts(client, username):
|
|
"""Gets a list of posts that the user has made."""
|
|
|
|
return [
|
|
post for post in client.redditor(username).submissions.new(limit=None)
|
|
]
|
|
|
|
|
|
def get_user_comments(client, username):
|
|
"""Gets a list of comments that the user has made."""
|
|
|
|
return [
|
|
comment for comment in client.redditor(username).comments.new(limit=None)
|
|
]
|
|
|
|
|
|
def get_post_html(post):
|
|
"""Takes a post object and creates a HTML for it - but not including the
|
|
preview HTML."""
|
|
|
|
with open(os.path.join("html", "post-div.html"), encoding="utf-8") as f:
|
|
html = f.read()
|
|
dt = datetime.utcfromtimestamp(post.created_utc)
|
|
html = html.replace("<!--title-->", post.title)
|
|
html = html.replace("<!--subreddit-->", f"/r/{str(post.subreddit)}")
|
|
html = html.replace("<!--user-->", f"/u/{post.author.name}" if post.author else "[deleted]")
|
|
html = html.replace("<!--link-->", f"posts/{post.id}.html")
|
|
html = html.replace("<!--reddit-link-->", f"https://reddit.com{post.permalink}")
|
|
html = html.replace("<!--content-link-->", post.url)
|
|
html = html.replace("<!--id-->", post.id)
|
|
html = html.replace("<!--body-->", (post.selftext_html or "").replace(
|
|
'<a href="/r/', '<a href="https://reddit.com/r/'
|
|
))
|
|
html = html.replace("<!--timestamp-->", str(dt))
|
|
html = html.replace("<!--date-->", dt.strftime("%d %B, %Y"))
|
|
return html
|
|
|
|
|
|
def save_media(post, location):
|
|
"""Takes a post object and tries to download any image/video it might be
|
|
associated with. If it can, it will return the filename."""
|
|
url = post.url
|
|
stripped_url = url.split("?")[0]
|
|
if url.endswith(post.permalink): return None
|
|
|
|
# What is the key information?
|
|
extension = stripped_url.split(".")[-1].lower()
|
|
domain = ".".join(post.url.split("/")[2].split(".")[-2:])
|
|
readable_name = list(filter(bool, post.permalink.split("/")))[-1]
|
|
|
|
# Handle galleries
|
|
# When we saved a cross_post, we don't know if it is a gallery.
|
|
if hasattr(post, "gallery_data") or ("gallery" in url and hasattr(post, "crosspost_parent_list")):
|
|
|
|
if not hasattr(post, "gallery_data") and hasattr(post, "crosspost_parent_list"):
|
|
for crosspost in post.crosspost_parent_list:
|
|
if crosspost["gallery_data"] is not None:
|
|
# hard hack
|
|
post.gallery_data = crosspost["gallery_data"]
|
|
post.media_metadata = crosspost["media_metadata"]
|
|
break
|
|
if not hasattr(post, "gallery_data"): return None
|
|
|
|
images = [ ]
|
|
for item in sorted(post.gallery_data['items'], key=lambda x: x['id']):
|
|
media_id = item['media_id']
|
|
meta = post.media_metadata[media_id]
|
|
source = meta['s']
|
|
if meta['e'] == 'Image':
|
|
url = source['u']
|
|
elif meta['e'] == 'AnimatedImage':
|
|
url = source['gif']
|
|
else:
|
|
return None
|
|
stripped_url = url.split("?")[0]
|
|
extension = stripped_url.split(".")[-1].lower()
|
|
filename = f"{readable_name}_{post.id}_{media_id}.{extension}"
|
|
try:
|
|
response = requests.get(url)
|
|
with open(os.path.join(location, "media", filename), "wb") as f:
|
|
f.write(response.content)
|
|
images.append(filename)
|
|
except:
|
|
print(f"Failed to download {url}")
|
|
return images if len(images) > 0 else None
|
|
|
|
# If it's an imgur gallery, forget it
|
|
if domain == "imgur.com" and "gallery" in url: return None
|
|
|
|
# Can the media be obtained directly?
|
|
if extension in IMAGE_EXTENSIONS + VIDEO_EXTENSIONS:
|
|
filename = f"{readable_name}_{post.id}.{extension}"
|
|
try:
|
|
response = requests.get(post.url)
|
|
except:
|
|
return
|
|
media_type = response.headers.get("Content-Type", "")
|
|
if media_type.startswith("image") or media_type.startswith("video"):
|
|
with open(os.path.join(location, "media", filename), "wb") as f:
|
|
f.write(response.content)
|
|
return [ filename ]
|
|
|
|
# Is this a v.redd.it link?
|
|
if domain == "redd.it":
|
|
downloader = Downloader(max_q=True, log=False)
|
|
downloader.url = url
|
|
current = os.getcwd()
|
|
try:
|
|
name = downloader.download()
|
|
extension = name.split(".")[-1]
|
|
filename = f"{readable_name}_{post.id}.{extension}"
|
|
os.rename(name, os.path.join(location, "media", filename))
|
|
return [ filename ]
|
|
except:
|
|
os.chdir(current)
|
|
return None
|
|
|
|
# Is it a gfycat link that redirects? Update the URL if possible
|
|
if domain == "gfycat.com":
|
|
html = requests.get(post.url).content
|
|
if len(html) < 50000:
|
|
match = re.search(r"http([\dA-Za-z\+\:\/\.]+)\.mp4", html.decode())
|
|
if match:
|
|
url = match.group()
|
|
else:
|
|
return None
|
|
|
|
# Is this an imgur image?
|
|
if domain == "imgur.com" and extension != "gifv":
|
|
for extension in IMAGE_EXTENSIONS:
|
|
direct_url = f'https://i.{url[url.find("//") + 2:]}.{extension}'
|
|
direct_url = direct_url.replace("i.imgur.com", "imgur.com")
|
|
direct_url = direct_url.replace("m.imgur.com", "imgur.com")
|
|
try:
|
|
response = requests.get(direct_url)
|
|
except: continue
|
|
if response.status_code == 200:
|
|
filename = f"{readable_name}_{post.id}.{extension}"
|
|
with open(os.path.join(location, "media", filename), "wb") as f:
|
|
f.write(response.content)
|
|
return [ filename ]
|
|
|
|
# Try to use youtube_dl if it's one of the possible domains
|
|
if domain in PLATFORMS:
|
|
options = {
|
|
"nocheckcertificate": True, "quiet": True, "no_warnings": True,
|
|
"ignoreerrors": True, "no-progress": True,
|
|
"outtmpl": os.path.join(
|
|
location, "media", f"{readable_name}_{post.id}" + ".%(ext)s"
|
|
)
|
|
}
|
|
with yt_dlp.YoutubeDL(options) as ydl:
|
|
try:
|
|
ydl.download([url])
|
|
except:
|
|
os.chdir(current)
|
|
return
|
|
for f in os.listdir(os.path.join(location, "media")):
|
|
if f.startswith(f"{readable_name}_{post.id}"):
|
|
return [ f ]
|
|
|
|
|
|
def add_media_preview_to_html(post_html, media):
|
|
"""Takes post HTML and returns a modified version with the preview
|
|
inserted."""
|
|
|
|
media_html_list = []
|
|
|
|
for m in media:
|
|
extension = m.split(".")[-1]
|
|
location = "/".join(["media", m])
|
|
if extension in IMAGE_EXTENSIONS:
|
|
media_html_list.append(f'<img src="{location}">')
|
|
if extension in VIDEO_EXTENSIONS:
|
|
media_html_list.append(f'<video controls><source src="{location}"></video>')
|
|
return post_html.replace('<!--preview-->', ''.join([ f"<div class=\"preview\">{item}</div>" for item in media_html_list]))
|
|
|
|
|
|
def create_post_page_html(post, post_html):
|
|
"""Creates the HTML for a post's own page."""
|
|
|
|
with open(os.path.join("html", "post.html"), encoding="utf-8") as f:
|
|
html = f.read()
|
|
html = html.replace("<!--title-->", post.title)
|
|
html = html.replace("<!--post-->", post_html.replace("h2>", "h1>").replace(
|
|
'<img src="media/', '<img src="../media/'
|
|
).replace(
|
|
'<source src="media/', '<source src="../media/'
|
|
))
|
|
html = re.sub(r'<a href="posts(.+?)</a>', "", html)
|
|
with open(os.path.join("html", "style.css"), encoding="utf-8") as f:
|
|
html = html.replace("<style></style>", f"<style>\n{f.read()}\n</style>")
|
|
with open(os.path.join("html", "main.js"), encoding="utf-8") as f:
|
|
html = html.replace("<script></script>", f"<script>\n{f.read()}\n</script>")
|
|
comments_html = []
|
|
post.comments.replace_more(limit=0)
|
|
for comment in post.comments:
|
|
comments_html.append(get_comment_html(
|
|
comment, op=post.author.name if post.author else None
|
|
))
|
|
html = html.replace("<!--comments-->", "\n".join(comments_html))
|
|
return html
|
|
|
|
|
|
def get_comment_html(comment, children=True, op=None):
|
|
"""Takes a post object and creates a HTML for it - it will get its children
|
|
too unless you specify otherwise."""
|
|
|
|
with open(os.path.join("html", "comment-div.html"), encoding="utf-8") as f:
|
|
html = f.read()
|
|
dt = datetime.utcfromtimestamp(comment.created_utc)
|
|
author = "[deleted]"
|
|
if comment.author:
|
|
if comment.author == op:
|
|
author = f'<span class="op">/u/{comment.author.name}</span>'
|
|
else:
|
|
author = f"/u/{comment.author.name}"
|
|
html = html.replace("<!--user-->", author)
|
|
html = html.replace("<!--body-->", (comment.body_html or "").replace(
|
|
'<a href="/r/', '<a href="https://reddit.com/r/'
|
|
))
|
|
html = html.replace("<!--score-->", str(comment.score))
|
|
html = html.replace("<!--link-->", f"https://reddit.com{comment.permalink}")
|
|
html = html.replace("<!--timestamp-->", str(dt))
|
|
html = html.replace("<!--id-->", comment.id)
|
|
html = html.replace("<!--date-->", dt.strftime("%H:%M - %d %B, %Y"))
|
|
if children:
|
|
children_html = []
|
|
for child in comment.replies:
|
|
children_html.append(get_comment_html(child, children=False, op=op))
|
|
html = html.replace("<!--children-->", "\n".join(children_html))
|
|
return html
|
|
|
|
|
|
def save_html(posts, comments, location, html_file, page, has_next, username=None):
|
|
if username:
|
|
with open(os.path.join("html", "username.html"), encoding="utf-8") as f:
|
|
html = f.read().replace("[username]", username)
|
|
else:
|
|
with open(os.path.join("html", html_file), encoding="utf-8") as f:
|
|
html = f.read()
|
|
with open(os.path.join("html", "style.css"), encoding="utf-8") as f:
|
|
html = html.replace("<style></style>", f"<style>\n{f.read()}\n</style>")
|
|
with open(os.path.join("html", "main.js"), encoding="utf-8") as f:
|
|
html = html.replace("<script></script>", f"<script>\n{f.read()}\n</script>")
|
|
if page == 0 or page is None:
|
|
html = html.replace("Previous</a>", "</a>")
|
|
else:
|
|
html = html.replace(".p.html", f".{page-1}.html")
|
|
if not has_next or page is None:
|
|
html = html.replace("Next</a>", "</a>")
|
|
else:
|
|
html = html.replace(".n.html", f".{page+1}.html")
|
|
html = html.replace("<!--posts-->", "\n".join(posts))
|
|
html = html.replace("<!--comments-->", "\n".join(comments))
|
|
file_name = html_file if page is None else html_file.replace(".html", f".{page}.html")
|
|
with open(os.path.join(location, file_name), "w", encoding="utf-8") as f:
|
|
f.write(html)
|