Merge pull request #861 from elebumm/CordlessCoder-TOML
new TOML config checker
This commit is contained in:
23 files changed
+502
-468
No files matched your search
@@ -0,0 +1,43 @@
|
||||
[reddit.creds]
|
||||
client_id = { optional = false, nmin = 12, nmax = 30, explanation = "the ID of your Reddit app of SCRIPT type", example = "fFAGRNJru1FTz70BzhT3Zg", regex = "^[-a-zA-Z0-9._~+/]+=*$", input_error = "The client ID can only contain printable characters.", oob_error = "The ID should be over 12 and under 30 characters, double check your input." }
|
||||
client_secret = { optional = false, nmin = 20, nmax = 40, explanation = "the SECRET of your Reddit app of SCRIPT type", example = "fFAGRNJru1FTz70BzhT3Zg", regex = "^[-a-zA-Z0-9._~+/]+=*$", input_error = "The client ID can only contain printable characters.", oob_error = "The secret should be over 20 and under 40 characters, double check your input." }
|
||||
username = { optional = false, nmin = 3, nmax = 20, explanation = "the username of your reddit account", example = "asdfghjkl", regex = "^[-_0-9a-zA-Z]+$", oob_error = "A username HAS to be between 3 and 20 characters" }
|
||||
password = { optional = false, nmin = 8, explanation = "the password of your reddit account", example = "fFAGRNJru1FTz70BzhT3Zg", oob_error = "Password too short" }
|
||||
2fa = { optional = true, type = "bool", options = [
|
||||
true,
|
||||
false,
|
||||
], default = false, explanation = "Whether you have Reddit 2FA enabled, Valid options are True and False", example = true }
|
||||
|
||||
|
||||
[reddit.thread]
|
||||
random = { optional = true, options = [
|
||||
true,
|
||||
false,
|
||||
], default = false, type = "bool", explanation = "If set to no, it will ask you a thread link to extract the thread, if yes it will randomize it. Default: 'False'", example = "True" }
|
||||
subreddit = { optional = false, regex = "[_0-9a-zA-Z]+$", nmin = 3, nmax = 21, explanation = "what subreddit to pull posts from, the name of the sub, not the URL", example = "AskReddit", oob_error = "A subreddit name HAS to be between 3 and 20 characters" }
|
||||
post_id = { optional = true, default = "", regex = "^((?!://|://).)*$", explanation = "Used if you want to use a specific post.", example = "urdtfx" }
|
||||
max_comment_length = { default = 500, optional = false, nmin = 10, nmax = 10000, type = "int", explanation = "max number of characters a comment can have. default is 500", example = 500, oob_error = "the max comment length should be between 10 and 10000" }
|
||||
post_lang = { default = "", optional = true, explanation = "The language you would like to translate to.", example = "es-cr"}
|
||||
|
||||
[settings]
|
||||
allow_nsfw = { optional = false, type = "bool", default = false, example = false, options = [
|
||||
true,
|
||||
false,
|
||||
], explanation = "Whether to allow NSFW content, True or False" }
|
||||
theme = { optional = false, default = "light", example = "dark", options = [
|
||||
"dark",
|
||||
"light",
|
||||
], explanation = "sets the Reddit theme, either LIGHT or DARK" }
|
||||
times_to_run = { optional = false, default = 1, example = 2, explanation = "used if you want to run multiple times. set to an int e.g. 4 or 29 or 1", type = "int", nmin = 1, oob_error = "It's very hard to run something less than once." }
|
||||
opacity = { optional = false, default = 0.9, example = 0.8, explanation = "Sets the opacity of the comments when overlayed over the background", type = "float", nmin = 0, nmax = 1, oob_error = "The opacity HAS to be between 0 and 1", input_error = "The opacity HAS to be a decimal number between 0 and 1" }
|
||||
storymode = { optional = true, type = "bool", default = false, example = false, options = [
|
||||
true,
|
||||
false,
|
||||
] }
|
||||
|
||||
|
||||
[settings.tts]
|
||||
aws_polly_voice = { optional = false, default = "Matthew", example = "Matthew", explanation = "The voice used for AWS Polly" }
|
||||
streamlabs_polly_voice = { optional = false, default = "Matthew", example = "Matthew", explanation = "The voice used for Streamlabs Polly" }
|
||||
tiktok_voice = { optional = false, default = "en_us_006", example = "en_us_006", explanation = "The voice used for TikTok TTS" }
|
||||
choice = { optional = false, default = "", options = [ "streamlabspolly", "tiktok", "googletranslate", "awspolly", ""], example = "streamlabspolly", explanation = "The backend used for TTS generation. This can be left blank and you will be prompted to choose at runtime." }
|
||||
@@ -1,86 +0,0 @@
|
||||
|
||||
REDDIT_CLIENT_ID="" #fFAGRNJru1FTz70BzhT3Zg
|
||||
#EXPLANATION the ID of your Reddit app of SCRIPT type
|
||||
#RANGE 12:30
|
||||
#MATCH_REGEX [-a-zA-Z0-9._~+/]+=*$
|
||||
#OOB_ERROR The ID should be over 12 and under 30 characters, double check your input.
|
||||
|
||||
REDDIT_CLIENT_SECRET="" #fFAGRNJru1FTz70BzhT3Zg
|
||||
#EXPLANATION the SECRET of your Reddit app of SCRIPT type
|
||||
#RANGE 20:40
|
||||
#MATCH_REGEX [-a-zA-Z0-9._~+/]+=*$
|
||||
#OOB_ERROR The secret should be over 20 and under 40 characters, double check your input.
|
||||
|
||||
REDDIT_USERNAME="" #asdfghjkl
|
||||
#EXPLANATION the username of your reddit account
|
||||
#RANGE 3:20
|
||||
#MATCH_REGEX [-_0-9a-zA-Z]+$
|
||||
#OOB_ERROR A username HAS to be between 3 and 20 characters
|
||||
|
||||
REDDIT_PASSWORD="" #fFAGRNJru1FTz70BzhT3Zg
|
||||
#EXPLANATION the password of your reddit account
|
||||
#RANGE 8:None
|
||||
#OOB_ERROR Password too short
|
||||
|
||||
#OPTIONAL
|
||||
RANDOM_THREAD="no"
|
||||
# If set to no, it will ask you a thread link to extract the thread, if yes it will randomize it. Default: "no"
|
||||
|
||||
REDDIT_2FA="" #no
|
||||
#MATCH_REGEX ^(yes|no)
|
||||
#EXPLANATION Whether you have Reddit 2FA enabled, Valid options are "yes" and "no"
|
||||
|
||||
SUBREDDIT="AskReddit"
|
||||
#EXPLANATION what subreddit to pull posts from, the name of the sub, not the URL
|
||||
#RANGE 3:20
|
||||
#MATCH_REGEX [_0-9a-zA-Z]+$
|
||||
#OOB_ERROR A subreddit name HAS to be between 3 and 20 characters
|
||||
|
||||
ALLOW_NSFW="False"
|
||||
#EXPLANATION Whether to allow NSFW content, True or False
|
||||
#MATCH_REGEX ^(True|False)$
|
||||
|
||||
POST_ID=""
|
||||
#MATCH_REGEX ^((?!://|://).)*$
|
||||
#EXPLANATION Used if you want to use a specific post. example of one is urdtfx
|
||||
|
||||
THEME="LIGHT" #dark
|
||||
#EXPLANATION sets the Reddit theme, either LIGHT or DARK
|
||||
#MATCH_REGEX ^(dark|light|DARK|LIGHT)$
|
||||
|
||||
TIMES_TO_RUN="" #2
|
||||
#EXPLANATION used if you want to run multiple times. set to an int e.g. 4 or 29 and leave blank for 1
|
||||
|
||||
MAX_COMMENT_LENGTH="500" #500
|
||||
#EXPLANATION max number of characters a comment can have. default is 500
|
||||
#RANGE 0:10000
|
||||
#MATCH_TYPE int
|
||||
#OOB_ERROR the max comment length should be between 0 and 10000
|
||||
|
||||
OPACITY="1" #.8
|
||||
#EXPLANATION Sets the opacity of the comments when overlayed over the background
|
||||
#RANGE 0:1
|
||||
#MATCH_TYPE float
|
||||
#OOB_ERROR The opacity HAS to be between 0 and 1
|
||||
|
||||
# If you want to translate the comments to another language, set the language code here.
|
||||
# If empty, no translation will be done.
|
||||
POSTLANG=""
|
||||
#EXPLANATION Activates the translation feature, set the language code for translate or leave blank
|
||||
|
||||
TTSCHOICE="Polly"
|
||||
#EXPLANATION the backend used for TTS. Without anything specified, the user will be prompted to choose one.
|
||||
# IMPORTANT NOTE: if you use translate, you need to set this to googletranslate or tiktok and use custom voice in your language
|
||||
|
||||
STREAMLABS_VOICE="Joanna"
|
||||
#EXPLANATION Sets the voice for the Streamlabs Polly TTS Engine. Check the file for more information on different voices.
|
||||
|
||||
AWS_VOICE="Joanna"
|
||||
#EXPLANATION Sets the voice for the AWS Polly TTS Engine. Check the file for more information on different voices.
|
||||
|
||||
TIKTOK_VOICE="en_us_006"
|
||||
#EXPLANATION Sets the voice for the TikTok TTS Engine. Check the file for more information on different voices.
|
||||
|
||||
#OPTIONAL
|
||||
STORYMODE="False"
|
||||
# IN-PROGRESS - not yet implemented
|
||||
@@ -241,3 +241,5 @@ reddit-bot-351418-5560ebc49cac.json
|
||||
*.pyc
|
||||
video_creation/data/videos.json
|
||||
video_creation/data/envvars.txt
|
||||
|
||||
config.toml
|
||||
@@ -40,11 +40,13 @@ The only original thing being done is the editing and gathering of all materials
|
||||
1. Clone this repository
|
||||
2. Run `pip install -r requirements.txt`
|
||||
|
||||
3. Run `playwright install` and `playwright install-deps`. (if this fails try adding python -m to the front of the command)
|
||||
3. Run `python -m playwright install` and `python -m playwright install-deps`
|
||||
|
||||
4. Run `python main.py`
|
||||
required\*\*), visit [the Reddit Apps page.](https://www.reddit.com/prefs/apps) TL;DR set up an app that is a "script".
|
||||
5. Enjoy 😎
|
||||
5. Visit [the Reddit Apps page.](https://www.reddit.com/prefs/apps), and set up an app that is a "script".
|
||||
6. The bot will ask you to fill in your details to connect to the Reddit API, and configure the bot to your liking
|
||||
7. Enjoy 😎
|
||||
8. If you need to reconfigure the bot, simply open the `config.toml` file and delete the lines that need to be changed. On the next run of the bot, it will help you reconfigure those options.
|
||||
|
||||
(Note if you got an error installing or running the bot try first rerunning the command with a three after the name e.g. python3 or pip3)
|
||||
|
||||
|
||||
+6
-2
@@ -1,6 +1,6 @@
|
||||
#!/usr/bin/env python3
|
||||
import random
|
||||
import os
|
||||
from utils import settings
|
||||
from gtts import gTTS
|
||||
|
||||
max_chars = 0
|
||||
@@ -12,7 +12,11 @@ class GTTS:
|
||||
self.voices = []
|
||||
|
||||
def run(self, text, filepath):
|
||||
tts = gTTS(text=text, lang=os.getenv("POSTLANG") or "en", slow=False)
|
||||
tts = gTTS(
|
||||
text=text,
|
||||
lang=settings.config["reddit"]["thread"]["post_lang"] or "en",
|
||||
slow=False,
|
||||
)
|
||||
tts.save(filepath)
|
||||
|
||||
def randomvoice(self):
|
||||
|
||||
+12
-7
@@ -1,5 +1,5 @@
|
||||
import base64
|
||||
import os
|
||||
from utils import settings
|
||||
import random
|
||||
import requests
|
||||
from requests.adapters import HTTPAdapter, Retry
|
||||
@@ -62,9 +62,7 @@ noneng = [
|
||||
|
||||
class TikTok: # TikTok Text-to-Speech Wrapper
|
||||
def __init__(self):
|
||||
self.URI_BASE = (
|
||||
"https://api16-normal-useast5.us.tiktokv.com/media/api/text/speech/invoke/?text_speaker="
|
||||
)
|
||||
self.URI_BASE = "https://api16-normal-useast5.us.tiktokv.com/media/api/text/speech/invoke/?text_speaker="
|
||||
self.max_chars = 300
|
||||
self.voices = {"human": human, "nonhuman": nonhuman, "noneng": noneng}
|
||||
|
||||
@@ -75,10 +73,15 @@ class TikTok: # TikTok Text-to-Speech Wrapper
|
||||
voice = (
|
||||
self.randomvoice()
|
||||
if random_voice
|
||||
else (os.getenv("TIKTOK_VOICE") or random.choice(self.voices["human"]))
|
||||
else (
|
||||
settings.config["settings"]["tts"]["tiktok_voice"]
|
||||
or random.choice(self.voices["human"])
|
||||
)
|
||||
)
|
||||
try:
|
||||
r = requests.post(f"{self.URI_BASE}{voice}&req_text={text}&speaker_map_type=0")
|
||||
r = requests.post(
|
||||
f"{self.URI_BASE}{voice}&req_text={text}&speaker_map_type=0"
|
||||
)
|
||||
except requests.exceptions.SSLError:
|
||||
# https://stackoverflow.com/a/47475019/18516611
|
||||
session = requests.Session()
|
||||
@@ -86,7 +89,9 @@ class TikTok: # TikTok Text-to-Speech Wrapper
|
||||
adapter = HTTPAdapter(max_retries=retry)
|
||||
session.mount("http://", adapter)
|
||||
session.mount("https://", adapter)
|
||||
r = session.post(f"{self.URI_BASE}{voice}&req_text={text}&speaker_map_type=0")
|
||||
r = session.post(
|
||||
f"{self.URI_BASE}{voice}&req_text={text}&speaker_map_type=0"
|
||||
)
|
||||
# print(r.text)
|
||||
vstr = [r.json()["data"]["v_str"]][0]
|
||||
b64d = base64.b64decode(vstr)
|
||||
|
||||
+5
-3
@@ -2,7 +2,7 @@
|
||||
from boto3 import Session
|
||||
from botocore.exceptions import BotoCoreError, ClientError
|
||||
import sys
|
||||
import os
|
||||
from utils import settings
|
||||
import random
|
||||
|
||||
voices = [
|
||||
@@ -35,11 +35,13 @@ class AWSPolly:
|
||||
if random_voice:
|
||||
voice = self.randomvoice()
|
||||
else:
|
||||
if not os.getenv("AWS_VOICE"):
|
||||
if not settings.config["settings"]["tts"]["aws_polly_voice"]:
|
||||
return ValueError(
|
||||
f"Please set the environment variable AWS_VOICE to a valid voice. options are: {voices}"
|
||||
)
|
||||
voice = str(os.getenv("AWS_VOICE")).capitalize()
|
||||
voice = str(
|
||||
settings.config["settings"]["tts"]["aws_polly_voice"]
|
||||
).capitalize()
|
||||
try:
|
||||
# Request speech synthesis
|
||||
response = polly.synthesize_speech(
|
||||
|
||||
+18
-8
@@ -2,7 +2,6 @@
|
||||
from pathlib import Path
|
||||
from typing import Tuple
|
||||
import re
|
||||
from os import getenv
|
||||
|
||||
# import sox
|
||||
# from mutagen import MutagenError
|
||||
@@ -12,6 +11,7 @@ from rich.progress import track
|
||||
from moviepy.editor import AudioFileClip, CompositeAudioClip, concatenate_audioclips
|
||||
from utils.console import print_step, print_substep
|
||||
from utils.voice import sanitize_text
|
||||
from utils import settings
|
||||
|
||||
DEFUALT_MAX_LENGTH: int = 50 # video length variable
|
||||
|
||||
@@ -56,11 +56,16 @@ class TTSEngine:
|
||||
print_step("Saving Text to MP3 files...")
|
||||
|
||||
self.call_tts("title", self.reddit_object["thread_title"])
|
||||
if self.reddit_object["thread_post"] != "" and getenv("STORYMODE", "").casefold() == "true":
|
||||
if (
|
||||
self.reddit_object["thread_post"] != ""
|
||||
and settings.config["settings"]["storymode"] == True
|
||||
):
|
||||
self.call_tts("posttext", self.reddit_object["thread_post"])
|
||||
|
||||
idx = None
|
||||
for idx, comment in track(enumerate(self.reddit_object["comments"]), "Saving..."):
|
||||
for idx, comment in track(
|
||||
enumerate(self.reddit_object["comments"]), "Saving..."
|
||||
):
|
||||
# ! Stop creating mp3 files if the length is greater than max length.
|
||||
if self.length > self.max_length:
|
||||
break
|
||||
@@ -76,7 +81,9 @@ class TTSEngine:
|
||||
split_files = []
|
||||
split_text = [
|
||||
x.group().strip()
|
||||
for x in re.finditer(rf" *((.{{0,{self.tts_module.max_chars}}})(\.|.$))", text)
|
||||
for x in re.finditer(
|
||||
rf" *((.{{0,{self.tts_module.max_chars}}})(\.|.$))", text
|
||||
)
|
||||
]
|
||||
|
||||
idy = None
|
||||
@@ -94,12 +101,14 @@ class TTSEngine:
|
||||
Path(name).unlink()
|
||||
|
||||
# for i in range(0, idy + 1):
|
||||
# print(f"Cleaning up {self.path}/{idx}-{i}.part.mp3")
|
||||
# print(f"Cleaning up {self.path}/{idx}-{i}.part.mp3")
|
||||
|
||||
# Path(f"{self.path}/{idx}-{i}.part.mp3").unlink()
|
||||
# Path(f"{self.path}/{idx}-{i}.part.mp3").unlink()
|
||||
|
||||
def call_tts(self, filename: str, text: str):
|
||||
self.tts_module.run(text=process_text(text), filepath=f"{self.path}/{filename}.mp3")
|
||||
self.tts_module.run(
|
||||
text=process_text(text), filepath=f"{self.path}/{filename}.mp3"
|
||||
)
|
||||
# try:
|
||||
# self.length += MP3(f"{self.path}/{filename}.mp3").info.length
|
||||
# except (MutagenError, HeaderNotFoundError):
|
||||
@@ -108,8 +117,9 @@ class TTSEngine:
|
||||
self.length += clip.duration
|
||||
clip.close()
|
||||
|
||||
|
||||
def process_text(text: str):
|
||||
lang = getenv("POSTLANG", "")
|
||||
lang = settings.config["reddit"]["thread"]["post_lang"]
|
||||
new_text = sanitize_text(text)
|
||||
if lang:
|
||||
print_substep("Translating Text...")
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import random
|
||||
import os
|
||||
import requests
|
||||
from requests.exceptions import JSONDecodeError
|
||||
from utils import settings
|
||||
|
||||
voices = [
|
||||
"Brian",
|
||||
@@ -35,11 +35,13 @@ class StreamlabsPolly:
|
||||
if random_voice:
|
||||
voice = self.randomvoice()
|
||||
else:
|
||||
if not os.getenv("STREAMLABS_VOICE"):
|
||||
if not settings.config["settings"]["tts"]["streamlabs_polly_voice"]:
|
||||
return ValueError(
|
||||
f"Please set the environment variable STREAMLABS_VOICE to a valid voice. options are: {voices}"
|
||||
)
|
||||
voice = str(os.getenv("STREAMLABS_VOICE")).capitalize()
|
||||
voice = str(
|
||||
settings.config["settings"]["tts"]["streamlabs_polly_voice"]
|
||||
).capitalize()
|
||||
body = {"voice": voice, "text": text, "service": "polly"}
|
||||
response = requests.post(self.url, data=body)
|
||||
try:
|
||||
@@ -55,6 +57,3 @@ class StreamlabsPolly:
|
||||
|
||||
def randomvoice(self):
|
||||
return random.choice(self.voices)
|
||||
|
||||
|
||||
# StreamlabsPolly().run(text=str('hi hi ' * 92)[1:], filepath='hello.mp3', random_voice=True)
|
||||
@@ -1,12 +1,11 @@
|
||||
#!/usr/bin/env python
|
||||
import math
|
||||
from subprocess import Popen
|
||||
from os import getenv, name
|
||||
from dotenv import load_dotenv
|
||||
from os import name
|
||||
from reddit.subreddit import get_subreddit_threads
|
||||
from utils.cleanup import cleanup
|
||||
from utils.console import print_markdown, print_step
|
||||
from utils.checker import check_env
|
||||
from utils import settings
|
||||
|
||||
# from utils.checker import envUpdate
|
||||
from video_creation.background import download_background, chop_background_video
|
||||
@@ -31,6 +30,7 @@ print_markdown(
|
||||
)
|
||||
print_step(f"You are using V{VERSION} of the bot")
|
||||
|
||||
|
||||
def main(POST_ID=None):
|
||||
cleanup()
|
||||
reddit_object = get_subreddit_threads(POST_ID)
|
||||
@@ -38,32 +38,33 @@ def main(POST_ID=None):
|
||||
length = math.ceil(length)
|
||||
download_screenshots_of_reddit_posts(reddit_object, number_of_comments)
|
||||
download_background()
|
||||
chop_background_video(length)
|
||||
make_final_video(number_of_comments, length, reddit_object)
|
||||
credit = chop_background_video(length)
|
||||
make_final_video(number_of_comments, length, reddit_object, credit)
|
||||
|
||||
|
||||
def run_many(times):
|
||||
for x in range(1, times + 1):
|
||||
print_step(
|
||||
f'on the {x}{("st" if x == 1 else ("nd" if x == 2 else ("rd" if x == 3 else "th")))} iteration of {times}'
|
||||
f'on the {x}{("th", "st", "nd", "rd", "th", "th", "th", "th","th", "th")[x%10]} iteration of {times}'
|
||||
) # correct 1st 2nd 3rd 4th 5th....
|
||||
main()
|
||||
Popen("cls" if name == "nt" else "clear", shell=True).wait()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if check_env() is not True:
|
||||
exit()
|
||||
load_dotenv()
|
||||
config = settings.check_toml(".config.template.toml", "config.toml")
|
||||
config is False and exit()
|
||||
try:
|
||||
if getenv("TIMES_TO_RUN") and isinstance(int(getenv("TIMES_TO_RUN")), int):
|
||||
run_many(int(getenv("TIMES_TO_RUN")))
|
||||
if config["settings"]["times_to_run"]:
|
||||
run_many(config["settings"]["times_to_run"])
|
||||
|
||||
elif len(getenv("POST_ID", "").split("+")) > 1:
|
||||
for index, post_id in enumerate(getenv("POST_ID", "").split("+")):
|
||||
elif len(config["reddit"]["thread"]["post_id"].split("+")) > 1:
|
||||
for index, post_id in enumerate(
|
||||
config["reddit"]["thread"]["post_id"].split("+")
|
||||
):
|
||||
index += 1
|
||||
print_step(
|
||||
f'on the {index}{("st" if index == 1 else ("nd" if index == 2 else ("rd" if index == 3 else "th")))} post of {len(getenv("POST_ID", "").split("+"))}'
|
||||
f'on the {index}{("st" if index%10 == 1 else ("nd" if index%10 == 2 else ("rd" if index%10 == 3 else "th")))} post of {len(config["reddit"]["thread"]["post_id"].split("+"))}'
|
||||
)
|
||||
main(post_id)
|
||||
Popen("cls" if name == "nt" else "clear", shell=True).wait()
|
||||
|
||||
+32
-18
@@ -1,6 +1,6 @@
|
||||
import re
|
||||
from os import getenv
|
||||
|
||||
from utils import settings
|
||||
import praw
|
||||
from praw.models import MoreComments
|
||||
|
||||
@@ -17,20 +17,22 @@ def get_subreddit_threads(POST_ID: str):
|
||||
print_substep("Logging into Reddit.")
|
||||
|
||||
content = {}
|
||||
if str(getenv("REDDIT_2FA")).casefold() == "yes":
|
||||
print("\nEnter your two-factor authentication code from your authenticator app.\n")
|
||||
if settings.config["reddit"]["creds"]["2fa"] == True:
|
||||
print(
|
||||
"\nEnter your two-factor authentication code from your authenticator app.\n"
|
||||
)
|
||||
code = input("> ")
|
||||
print()
|
||||
pw = getenv("REDDIT_PASSWORD")
|
||||
pw = settings.config["reddit"]["creds"]["password"]
|
||||
passkey = f"{pw}:{code}"
|
||||
else:
|
||||
passkey = getenv("REDDIT_PASSWORD")
|
||||
username = getenv("REDDIT_USERNAME")
|
||||
passkey = settings.config["reddit"]["creds"]["password"]
|
||||
username = settings.config["reddit"]["creds"]["username"]
|
||||
if username.casefold().startswith("u/"):
|
||||
username = username[2:]
|
||||
reddit = praw.Reddit(
|
||||
client_id=getenv("REDDIT_CLIENT_ID"),
|
||||
client_secret=getenv("REDDIT_CLIENT_SECRET"),
|
||||
client_id=settings.config["reddit"]["creds"]["client_id"],
|
||||
client_secret=settings.config["reddit"]["creds"]["client_secret"],
|
||||
user_agent="Accessing Reddit threads",
|
||||
username=username,
|
||||
passkey=passkey,
|
||||
@@ -39,21 +41,26 @@ def get_subreddit_threads(POST_ID: str):
|
||||
|
||||
# Ask user for subreddit input
|
||||
print_step("Getting subreddit threads...")
|
||||
if not getenv(
|
||||
"SUBREDDIT"
|
||||
): # note to user. you can have multiple subreddits via reddit.subreddit("redditdev+learnpython")
|
||||
if not settings.config["reddit"]["thread"][
|
||||
"subreddit"
|
||||
]: # note to user. you can have multiple subreddits via reddit.subreddit("redditdev+learnpython")
|
||||
try:
|
||||
subreddit = reddit.subreddit(
|
||||
re.sub(r"r\/", "", input("What subreddit would you like to pull from? "))
|
||||
re.sub(
|
||||
r"r\/", "", input("What subreddit would you like to pull from? ")
|
||||
)
|
||||
# removes the r/ from the input
|
||||
)
|
||||
except ValueError:
|
||||
subreddit = reddit.subreddit("askreddit")
|
||||
print_substep("Subreddit not defined. Using AskReddit.")
|
||||
else:
|
||||
print_substep(f"Using subreddit: r/{getenv('SUBREDDIT')} from environment variable config")
|
||||
subreddit_choice = getenv("SUBREDDIT")
|
||||
if subreddit_choice.casefold().startswith("r/"): # removes the r/ from the input
|
||||
sub = settings.config["reddit"]["thread"]["subreddit"]
|
||||
print_substep(f"Using subreddit: r/{sub} from TOML config")
|
||||
subreddit_choice = sub
|
||||
if subreddit_choice.casefold().startswith(
|
||||
"r/"
|
||||
): # removes the r/ from the input
|
||||
subreddit_choice = subreddit_choice[2:]
|
||||
subreddit = reddit.subreddit(
|
||||
subreddit_choice
|
||||
@@ -61,8 +68,13 @@ def get_subreddit_threads(POST_ID: str):
|
||||
|
||||
if POST_ID: # would only be called if there are multiple queued posts
|
||||
submission = reddit.submission(id=POST_ID)
|
||||
elif getenv("POST_ID") and len(getenv("POST_ID").split("+")) == 1:
|
||||
submission = reddit.submission(id=getenv("POST_ID"))
|
||||
elif (
|
||||
settings.config["reddit"]["thread"]["post_id"]
|
||||
and len(settings.config["reddit"]["thread"]["post_id"].split("+")) == 1
|
||||
):
|
||||
submission = reddit.submission(
|
||||
id=settings.config["reddit"]["thread"]["post_id"]
|
||||
)
|
||||
else:
|
||||
|
||||
threads = subreddit.hot(limit=25)
|
||||
@@ -91,7 +103,9 @@ def get_subreddit_threads(POST_ID: str):
|
||||
if top_level_comment.body in ["[removed]", "[deleted]"]:
|
||||
continue # # see https://github.com/JasonLovesDoggo/RedditVideoMakerBot/issues/78
|
||||
if not top_level_comment.stickied:
|
||||
if len(top_level_comment.body) <= int(getenv("MAX_COMMENT_LENGTH", "500")):
|
||||
if len(top_level_comment.body) <= int(
|
||||
settings.config["reddit"]["thread"]["max_comment_length"]
|
||||
):
|
||||
if (
|
||||
top_level_comment.author is not None
|
||||
): # if errors occur with this change to if not.
|
||||
|
||||
+1
-2
@@ -2,11 +2,10 @@ boto3==1.24.12
|
||||
botocore==1.27.22
|
||||
gTTS==2.2.4
|
||||
moviepy==1.0.3
|
||||
mutagen==1.45.1
|
||||
playwright==1.23.0
|
||||
praw==7.6.0
|
||||
python-dotenv==0.20.0
|
||||
pytube==12.1.0
|
||||
requests==2.28.1
|
||||
rich==12.4.4
|
||||
toml==0.10.2
|
||||
translators==5.3.1
|
||||
@@ -1,193 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
import os
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
from rich import box
|
||||
import re
|
||||
import dotenv
|
||||
from utils.console import handle_input
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def check_env() -> bool:
|
||||
"""Checks to see what's been put in .env
|
||||
|
||||
Returns:
|
||||
bool: Whether or not everything was put in properly
|
||||
"""
|
||||
if not os.path.exists(".env.template"):
|
||||
console.print("[red]Couldn't find .env.template. Unable to check variables.")
|
||||
return True
|
||||
if not os.path.exists(".env"):
|
||||
console.print("[red]Couldn't find the .env file, creating one now.")
|
||||
with open(".env", "x", encoding="utf-8") as file:
|
||||
file.write("")
|
||||
success = True
|
||||
with open(".env.template", "r", encoding="utf-8") as template:
|
||||
# req_envs = [env.split("=")[0] for env in template.readlines() if "=" in env]
|
||||
matching = {}
|
||||
explanations = {}
|
||||
bounds = {}
|
||||
types = {}
|
||||
oob_errors = {}
|
||||
examples = {}
|
||||
req_envs = []
|
||||
var_optional = False
|
||||
for line in template.readlines():
|
||||
if line.startswith("#") is not True and "=" in line and var_optional is not True:
|
||||
req_envs.append(line.split("=")[0])
|
||||
if "#" in line:
|
||||
examples[line.split("=")[0]] = "#".join(line.split("#")[1:]).strip()
|
||||
elif "#OPTIONAL" in line:
|
||||
var_optional = True
|
||||
elif line.startswith("#MATCH_REGEX "):
|
||||
matching[req_envs[-1]] = line.removeprefix("#MATCH_REGEX ")[:-1]
|
||||
var_optional = False
|
||||
elif line.startswith("#OOB_ERROR "):
|
||||
oob_errors[req_envs[-1]] = line.removeprefix("#OOB_ERROR ")[:-1]
|
||||
var_optional = False
|
||||
elif line.startswith("#RANGE "):
|
||||
bounds[req_envs[-1]] = tuple(
|
||||
map(
|
||||
lambda x: float(x) if x != "None" else None,
|
||||
line.removeprefix("#RANGE ")[:-1].split(":"),
|
||||
)
|
||||
)
|
||||
var_optional = False
|
||||
elif line.startswith("#MATCH_TYPE "):
|
||||
types[req_envs[-1]] = eval(line.removeprefix("#MATCH_TYPE ")[:-1].split()[0])
|
||||
var_optional = False
|
||||
elif line.startswith("#EXPLANATION "):
|
||||
explanations[req_envs[-1]] = line.removeprefix("#EXPLANATION ")[:-1]
|
||||
var_optional = False
|
||||
else:
|
||||
var_optional = False
|
||||
missing = set()
|
||||
incorrect = set()
|
||||
dotenv.load_dotenv()
|
||||
for env in req_envs:
|
||||
value = os.getenv(env)
|
||||
if value is None:
|
||||
missing.add(env)
|
||||
continue
|
||||
if env in matching.keys():
|
||||
re.match(matching[env], value) is None and incorrect.add(env)
|
||||
if env in bounds.keys() and env not in types.keys():
|
||||
len(value) >= bounds[env][0] or (
|
||||
len(bounds[env]) > 1 and bounds[env][1] >= len(value)
|
||||
) or incorrect.add(env)
|
||||
continue
|
||||
if env in types.keys():
|
||||
try:
|
||||
temp = types[env](value)
|
||||
if env in bounds.keys():
|
||||
(bounds[env][0] <= temp or incorrect.add(env)) and len(bounds[env]) > 1 and (
|
||||
bounds[env][1] >= temp or incorrect.add(env)
|
||||
)
|
||||
except ValueError:
|
||||
incorrect.add(env)
|
||||
|
||||
if len(missing):
|
||||
table = Table(
|
||||
title="Missing variables",
|
||||
highlight=True,
|
||||
show_lines=True,
|
||||
box=box.ROUNDED,
|
||||
border_style="#414868",
|
||||
header_style="#C0CAF5 bold",
|
||||
title_justify="left",
|
||||
title_style="#C0CAF5 bold",
|
||||
)
|
||||
table.add_column("Variable", justify="left", style="#7AA2F7 bold", no_wrap=True)
|
||||
table.add_column("Explanation", justify="left", style="#BB9AF7", no_wrap=False)
|
||||
table.add_column("Example", justify="center", style="#F7768E", no_wrap=True)
|
||||
table.add_column("Min", justify="right", style="#F7768E", no_wrap=True)
|
||||
table.add_column("Max", justify="left", style="#F7768E", no_wrap=True)
|
||||
for env in missing:
|
||||
table.add_row(
|
||||
env,
|
||||
explanations[env] if env in explanations.keys() else "No explanation given",
|
||||
examples[env] if env in examples.keys() else "",
|
||||
str(bounds[env][0]) if env in bounds.keys() and bounds[env][1] is not None else "",
|
||||
str(bounds[env][1])
|
||||
if env in bounds.keys() and len(bounds[env]) > 1 and bounds[env][1] is not None
|
||||
else "",
|
||||
)
|
||||
console.print(table)
|
||||
success = False
|
||||
if len(incorrect):
|
||||
table = Table(
|
||||
title="Incorrect variables",
|
||||
highlight=True,
|
||||
show_lines=True,
|
||||
box=box.ROUNDED,
|
||||
border_style="#414868",
|
||||
header_style="#C0CAF5 bold",
|
||||
title_justify="left",
|
||||
title_style="#C0CAF5 bold",
|
||||
)
|
||||
table.add_column("Variable", justify="left", style="#7AA2F7 bold", no_wrap=True)
|
||||
table.add_column("Current value", justify="left", style="#F7768E", no_wrap=False)
|
||||
table.add_column("Explanation", justify="left", style="#BB9AF7", no_wrap=False)
|
||||
table.add_column("Example", justify="center", style="#F7768E", no_wrap=True)
|
||||
table.add_column("Min", justify="right", style="#F7768E", no_wrap=True)
|
||||
table.add_column("Max", justify="left", style="#F7768E", no_wrap=True)
|
||||
for env in incorrect:
|
||||
table.add_row(
|
||||
env,
|
||||
os.getenv(env),
|
||||
explanations[env] if env in explanations.keys() else "No explanation given",
|
||||
str(types[env].__name__) if env in types.keys() else "str",
|
||||
str(bounds[env][0]) if env in bounds.keys() else "None",
|
||||
str(bounds[env][1]) if env in bounds.keys() and len(bounds[env]) > 1 else "None",
|
||||
)
|
||||
missing.add(env)
|
||||
console.print(table)
|
||||
success = False
|
||||
if success is True:
|
||||
return True
|
||||
console.print(
|
||||
"[green]Do you want to automatically overwrite incorrect variables and add the missing variables? (y/n)"
|
||||
)
|
||||
if not input().casefold().startswith("y"):
|
||||
console.print("[red]Aborting: Unresolved missing variables")
|
||||
return False
|
||||
if len(incorrect):
|
||||
with open(".env", "r+", encoding="utf-8") as env_file:
|
||||
lines = []
|
||||
for line in env_file.readlines():
|
||||
line.split("=")[0].strip() not in incorrect and lines.append(line)
|
||||
env_file.seek(0)
|
||||
env_file.write("\n".join(lines))
|
||||
env_file.truncate()
|
||||
console.print("[green]Successfully removed incorrectly set variables from .env")
|
||||
with open(".env", "a", encoding="utf-8") as env_file:
|
||||
for env in missing:
|
||||
env_file.write(
|
||||
env
|
||||
+ "="
|
||||
+ ('"' if env not in types.keys() else "")
|
||||
+ str(
|
||||
handle_input(
|
||||
"[#F7768E bold]" + env + "[#C0CAF5 bold]=",
|
||||
types[env] if env in types.keys() else False,
|
||||
matching[env] if env in matching.keys() else ".*",
|
||||
explanations[env]
|
||||
if env in explanations.keys()
|
||||
else "Incorrect input. Try again.",
|
||||
bounds[env][0] if env in bounds.keys() else None,
|
||||
bounds[env][1] if env in bounds.keys() and len(bounds[env]) > 1 else None,
|
||||
oob_errors[env] if env in oob_errors.keys() else "Input too long/short.",
|
||||
extra_info="[#C0CAF5 bold]⮶ "
|
||||
+ (explanations[env] if env in explanations.keys() else "No info available"),
|
||||
)
|
||||
)
|
||||
+ ('"' if env not in types.keys() else "")
|
||||
+ "\n"
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
check_env()
|
||||
+3
-1
@@ -10,7 +10,9 @@ def cleanup() -> int:
|
||||
"""
|
||||
if exists("./assets/temp"):
|
||||
count = 0
|
||||
files = [f for f in os.listdir(".") if f.endswith(".mp4") and "temp" in f.lower()]
|
||||
files = [
|
||||
f for f in os.listdir(".") if f.endswith(".mp4") and "temp" in f.lower()
|
||||
]
|
||||
count += len(files)
|
||||
for f in files:
|
||||
os.remove(f)
|
||||
|
||||
@@ -1,46 +0,0 @@
|
||||
# write a class that takes .env file and parses it into a dictionary
|
||||
from dotenv import dotenv_values
|
||||
|
||||
DEFAULTS = {
|
||||
"SUBREDDIT": "AskReddit",
|
||||
"ALLOW_NSFW": "False",
|
||||
"POST_ID": "",
|
||||
"THEME": "DARK",
|
||||
"REDDIT_2FA": "no",
|
||||
"TIMES_TO_RUN": "",
|
||||
"MAX_COMMENT_LENGTH": "500",
|
||||
"OPACITY": "1",
|
||||
"VOICE": "en_us_001",
|
||||
"STORYMODE": "False",
|
||||
}
|
||||
|
||||
|
||||
class Config:
|
||||
def __init__(self):
|
||||
self.raw = dotenv_values("../.env")
|
||||
self.load_attrs()
|
||||
|
||||
def __getattr__(self, attr): # code completion for attributes fix.
|
||||
return getattr(self, attr)
|
||||
|
||||
def load_attrs(self):
|
||||
for key, value in self.raw.items():
|
||||
self.add_attr(key, value)
|
||||
|
||||
def add_attr(self, key, value):
|
||||
if value is None or value == "":
|
||||
setattr(self, key, DEFAULTS[key])
|
||||
else:
|
||||
setattr(self, key, str(value))
|
||||
|
||||
|
||||
config = Config()
|
||||
|
||||
print(config.SUBREDDIT)
|
||||
# def temp():
|
||||
# root = ''
|
||||
# if isinstance(root, praw.models.Submission):
|
||||
# root_type = 'submission'
|
||||
# elif isinstance(root, praw.models.Comment):
|
||||
# root_type = 'comment'
|
||||
#
|
||||
+84
-29
@@ -24,17 +24,17 @@ def print_step(text):
|
||||
console.print(panel)
|
||||
|
||||
|
||||
def print_substep(text, style=""):
|
||||
"""Prints a rich info message without the panelling."""
|
||||
console.print(text, style=style)
|
||||
|
||||
|
||||
def print_table(items):
|
||||
"""Prints items in a table."""
|
||||
|
||||
console.print(Columns([Panel(f"[yellow]{item}", expand=True) for item in items]))
|
||||
|
||||
|
||||
def print_substep(text, style=""):
|
||||
"""Prints a rich info message without the panelling."""
|
||||
console.print(text, style=style)
|
||||
|
||||
|
||||
def handle_input(
|
||||
message: str = "",
|
||||
check_type=False,
|
||||
@@ -44,33 +44,88 @@ def handle_input(
|
||||
nmax=None,
|
||||
oob_error="",
|
||||
extra_info="",
|
||||
options: list = None,
|
||||
default=NotImplemented,
|
||||
optional=False,
|
||||
):
|
||||
match = re.compile(match + "$")
|
||||
if optional:
|
||||
console.print(
|
||||
message
|
||||
+ "\n[green]This is an optional value. Do you want to skip it? (y/n)"
|
||||
)
|
||||
if input().casefold().startswith("y"):
|
||||
return default if default is not NotImplemented else ""
|
||||
if default is not NotImplemented:
|
||||
console.print(
|
||||
"[green]"
|
||||
+ message
|
||||
+ '\n[blue bold]The default value is "'
|
||||
+ str(default)
|
||||
+ '"\nDo you want to use it?(y/n)'
|
||||
)
|
||||
if input().casefold().startswith("y"):
|
||||
return default
|
||||
if options is None:
|
||||
match = re.compile(match)
|
||||
console.print("[green bold]" + extra_info, no_wrap=True)
|
||||
while True:
|
||||
console.print(message, end="")
|
||||
user_input = input("").strip()
|
||||
if check_type is not False:
|
||||
try:
|
||||
user_input = check_type(user_input)
|
||||
if (nmin is not None and user_input < nmin) or (
|
||||
nmax is not None and user_input > nmax
|
||||
):
|
||||
# FAILSTATE Input out of bounds
|
||||
console.print("[red]" + oob_error)
|
||||
continue
|
||||
break # Successful type conversion and number in bounds
|
||||
except ValueError:
|
||||
# Type conversion failed
|
||||
console.print("[red]" + err_message)
|
||||
continue
|
||||
elif match != "" and re.match(match, user_input) is None:
|
||||
console.print(
|
||||
"[red]"
|
||||
+ err_message
|
||||
+ "\nAre you absolutely sure it's correct?(y/n)"
|
||||
)
|
||||
if input().casefold().startswith("y"):
|
||||
break
|
||||
continue
|
||||
else:
|
||||
# FAILSTATE Input STRING out of bounds
|
||||
if (nmin is not None and len(user_input) < nmin) or (
|
||||
nmax is not None and len(user_input) > nmax
|
||||
):
|
||||
console.print("[red bold]" + oob_error)
|
||||
continue
|
||||
break # SUCCESS Input STRING in bounds
|
||||
return user_input
|
||||
console.print(extra_info, no_wrap=True)
|
||||
while True:
|
||||
console.print(message, end="")
|
||||
user_input = input("").strip()
|
||||
if re.match(match, user_input) is not None:
|
||||
if check_type is not False:
|
||||
try:
|
||||
user_input = check_type(user_input) # this line is fine
|
||||
if nmin is not None and user_input < nmin:
|
||||
console.print("[red]" + oob_error) # Input too low failstate
|
||||
continue
|
||||
if nmax is not None and user_input > nmax:
|
||||
console.print("[red]" + oob_error) # Input too high
|
||||
continue
|
||||
break # Successful type conversion and number in bounds
|
||||
except ValueError:
|
||||
console.print("[red]" + err_message) # Type conversion failed
|
||||
continue
|
||||
if nmin is not None and len(user_input) < nmin: # Check if string is long enough
|
||||
console.print("[red]" + oob_error)
|
||||
if check_type is not False:
|
||||
try:
|
||||
isinstance(eval(user_input), check_type)
|
||||
return check_type(user_input)
|
||||
except:
|
||||
console.print(
|
||||
"[red bold]"
|
||||
+ err_message
|
||||
+ "\nValid options are: "
|
||||
+ ", ".join(map(str, options))
|
||||
+ "."
|
||||
)
|
||||
continue
|
||||
if nmax is not None and len(user_input) > nmax: # Check if string is not too long
|
||||
console.print("[red]" + oob_error)
|
||||
continue
|
||||
break
|
||||
console.print("[red]" + err_message)
|
||||
|
||||
return user_input
|
||||
if user_input in options:
|
||||
return user_input
|
||||
console.print(
|
||||
"[red bold]"
|
||||
+ err_message
|
||||
+ "\nValid options are: "
|
||||
+ ", ".join(map(str, options))
|
||||
+ "."
|
||||
)
|
||||
Executable
+190
@@ -0,0 +1,190 @@
|
||||
#!/usr/bin/env python
|
||||
# import os
|
||||
import toml
|
||||
from rich.console import Console
|
||||
import re
|
||||
|
||||
from typing import Tuple, Dict
|
||||
|
||||
from utils.console import handle_input
|
||||
|
||||
# from console import handle_input
|
||||
|
||||
|
||||
console = Console()
|
||||
|
||||
|
||||
def crawl(obj: dict, func=lambda x, y: print(x, y, end="\n"), path: list = []):
|
||||
for key in obj.keys():
|
||||
if type(obj[key]) is dict:
|
||||
crawl(obj[key], func, path + [key])
|
||||
continue
|
||||
func(path + [key], obj[key])
|
||||
|
||||
|
||||
def check(value, checks, name):
|
||||
def get_check_value(key, default_result):
|
||||
return checks[key] if key in checks else default_result
|
||||
|
||||
incorrect = False
|
||||
if value == {}:
|
||||
incorrect = True
|
||||
if not incorrect and "type" in checks:
|
||||
try:
|
||||
value = eval(checks["type"])(value)
|
||||
except:
|
||||
incorrect = True
|
||||
|
||||
if (
|
||||
not incorrect and "options" in checks and value not in checks["options"]
|
||||
): # FAILSTATE Value is not one of the options
|
||||
incorrect = True
|
||||
if (
|
||||
not incorrect
|
||||
and "regex" in checks
|
||||
and (
|
||||
(isinstance(value, str) and re.match(checks["regex"], value) is None)
|
||||
or not isinstance(value, str)
|
||||
)
|
||||
): # FAILSTATE Value doesn't match regex, or has regex but is not a string.
|
||||
incorrect = True
|
||||
|
||||
if (
|
||||
not incorrect
|
||||
and not hasattr(value, "__iter__")
|
||||
and (
|
||||
("nmin" in checks and checks["nmin"] is not None and value < checks["nmin"])
|
||||
or (
|
||||
"nmax" in checks
|
||||
and checks["nmax"] is not None
|
||||
and value > checks["nmax"]
|
||||
)
|
||||
)
|
||||
):
|
||||
incorrect = True
|
||||
if (
|
||||
not incorrect
|
||||
and hasattr(value, "__iter__")
|
||||
and (
|
||||
(
|
||||
"nmin" in checks
|
||||
and checks["nmin"] is not None
|
||||
and len(value) < checks["nmin"]
|
||||
)
|
||||
or (
|
||||
"nmax" in checks
|
||||
and checks["nmax"] is not None
|
||||
and len(value) > checks["nmax"]
|
||||
)
|
||||
)
|
||||
):
|
||||
incorrect = True
|
||||
|
||||
if incorrect:
|
||||
value = handle_input(
|
||||
message=(
|
||||
(
|
||||
("[blue]Example: " + str(checks["example"]) + "\n")
|
||||
if "example" in checks
|
||||
else ""
|
||||
)
|
||||
+ "[red]"
|
||||
+ ("Non-optional ", "Optional ")[
|
||||
"optional" in checks and checks["optional"] is True
|
||||
]
|
||||
)
|
||||
+ "[#C0CAF5 bold]"
|
||||
+ str(name)
|
||||
+ "[#F7768E bold]=",
|
||||
extra_info=get_check_value("explanation", ""),
|
||||
check_type=eval(get_check_value("type", "False")),
|
||||
default=get_check_value("default", NotImplemented),
|
||||
match=get_check_value("regex", ""),
|
||||
err_message=get_check_value("input_error", "Incorrect input"),
|
||||
nmin=get_check_value("nmin", None),
|
||||
nmax=get_check_value("nmax", None),
|
||||
oob_error=get_check_value(
|
||||
"oob_error", "Input out of bounds(Value too high/low/long/short)"
|
||||
),
|
||||
options=get_check_value("options", None),
|
||||
optional=get_check_value("optional", False),
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def crawl_and_check(obj: dict, path: list, checks: dict = {}, name=""):
|
||||
if len(path) == 0:
|
||||
return check(obj, checks, name)
|
||||
if path[0] not in obj.keys():
|
||||
obj[path[0]] = {}
|
||||
obj[path[0]] = crawl_and_check(obj[path[0]], path[1:], checks, path[0])
|
||||
return obj
|
||||
|
||||
|
||||
def check_vars(path, checks):
|
||||
global config
|
||||
crawl_and_check(config, path, checks)
|
||||
|
||||
|
||||
def check_toml(template_file, config_file) -> Tuple[bool, Dict]:
|
||||
global config
|
||||
config = None
|
||||
try:
|
||||
template = toml.load(template_file)
|
||||
except Exception as error:
|
||||
console.print(
|
||||
f"[red bold]Encountered error when trying to to load {template_file}: {error}"
|
||||
)
|
||||
return False
|
||||
try:
|
||||
config = toml.load(config_file)
|
||||
except (toml.TomlDecodeError):
|
||||
console.print(
|
||||
f"""[blue]Couldn't read {config_file}.
|
||||
Overwrite it?(y/n)"""
|
||||
)
|
||||
if not input().startswith("y"):
|
||||
print("Unable to read config, and not allowed to overwrite it. Giving up.")
|
||||
return False
|
||||
else:
|
||||
try:
|
||||
with open(config_file, "w") as f:
|
||||
f.write("")
|
||||
except:
|
||||
console.print(
|
||||
f"[red bold]Failed to overwrite {config_file}. Giving up.\nSuggestion: check {config_file} permissions for the user."
|
||||
)
|
||||
return False
|
||||
except (FileNotFoundError):
|
||||
console.print(
|
||||
f"""[blue]Couldn't find {config_file}
|
||||
Creating it now."""
|
||||
)
|
||||
try:
|
||||
with open(config_file, "x") as f:
|
||||
f.write("")
|
||||
config = {}
|
||||
except:
|
||||
console.print(
|
||||
f"[red bold]Failed to write to {config_file}. Giving up.\nSuggestion: check the folder's permissions for the user."
|
||||
)
|
||||
return False
|
||||
|
||||
console.print(
|
||||
"""\
|
||||
[blue bold]###############################
|
||||
# #
|
||||
# Checking TOML configuration #
|
||||
# #
|
||||
###############################
|
||||
If you see any prompts, that means that you have unset/incorrectly set variables, please input the correct values.\
|
||||
"""
|
||||
)
|
||||
crawl(template, check_vars)
|
||||
with open(config_file, "w") as f:
|
||||
toml.dump(config, f)
|
||||
return config
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
check_toml(".config.template.toml", "config.toml")
|
||||
+5
-3
@@ -1,5 +1,5 @@
|
||||
import json
|
||||
from os import getenv
|
||||
from utils import settings
|
||||
from utils.console import print_substep
|
||||
|
||||
|
||||
@@ -15,14 +15,16 @@ def get_subreddit_undone(submissions: list, subreddit):
|
||||
"""
|
||||
# recursively checks if the top submission in the list was already done.
|
||||
|
||||
with open("./video_creation/data/videos.json", "r", encoding="utf-8") as done_vids_raw:
|
||||
with open(
|
||||
"./video_creation/data/videos.json", "r", encoding="utf-8"
|
||||
) as done_vids_raw:
|
||||
done_videos = json.load(done_vids_raw)
|
||||
for submission in submissions:
|
||||
if already_done(done_videos, submission):
|
||||
continue
|
||||
if submission.over_18:
|
||||
try:
|
||||
if getenv("ALLOW_NSFW").casefold() == "false":
|
||||
if settings.config["settings"]["allow_nsfw"] == False:
|
||||
print_substep("NSFW Post Detected. Skipping...")
|
||||
continue
|
||||
except AttributeError:
|
||||
|
||||
+7
-6
@@ -1,11 +1,10 @@
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
from os import getenv
|
||||
from typing import Dict
|
||||
|
||||
from praw.models import Submission
|
||||
|
||||
from utils import settings
|
||||
from utils.console import print_step
|
||||
|
||||
|
||||
@@ -21,11 +20,13 @@ def check_done(
|
||||
Returns:
|
||||
Dict[str]|None: Reddit object in args
|
||||
"""
|
||||
with open("./video_creation/data/videos.json", "r", encoding="utf-8") as done_vids_raw:
|
||||
with open(
|
||||
"./video_creation/data/videos.json", "r", encoding="utf-8"
|
||||
) as done_vids_raw:
|
||||
done_videos = json.load(done_vids_raw)
|
||||
for video in done_videos:
|
||||
if video["id"] == str(redditobj):
|
||||
if getenv("POST_ID"):
|
||||
if settings.config["reddit"]["thread"]["post_id"]:
|
||||
print_step(
|
||||
"You already have done this video but since it was declared specifically in the .env file the program will continue"
|
||||
)
|
||||
@@ -35,7 +36,7 @@ def check_done(
|
||||
return redditobj
|
||||
|
||||
|
||||
def save_data(filename: str, reddit_title: str, reddit_id: str):
|
||||
def save_data(filename: str, reddit_title: str, reddit_id: str, credit: str):
|
||||
"""Saves the videos that have already been generated to a JSON file in video_creation/data/videos.json
|
||||
|
||||
Args:
|
||||
@@ -51,7 +52,7 @@ def save_data(filename: str, reddit_title: str, reddit_id: str):
|
||||
payload = {
|
||||
"id": reddit_id,
|
||||
"time": str(int(time.time())),
|
||||
"background_credit": str(os.getenv("background_credit")),
|
||||
"background_credit": credit,
|
||||
"reddit_title": reddit_title,
|
||||
"filename": filename,
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import random
|
||||
from os import listdir, environ
|
||||
from os import listdir
|
||||
from pathlib import Path
|
||||
from random import randrange
|
||||
from typing import Tuple
|
||||
@@ -52,10 +52,12 @@ def download_background():
|
||||
"assets/backgrounds", filename=f"{credit}-{filename}"
|
||||
)
|
||||
|
||||
print_substep("Background videos downloaded successfully! 🎉", style="bold green")
|
||||
print_substep(
|
||||
"Background videos downloaded successfully! 🎉", style="bold green"
|
||||
)
|
||||
|
||||
|
||||
def chop_background_video(video_length: int):
|
||||
def chop_background_video(video_length: int) -> str:
|
||||
"""Generates the background footage to be used in the video and writes it to assets/temp/background.mp4
|
||||
|
||||
Args:
|
||||
@@ -63,7 +65,7 @@ def chop_background_video(video_length: int):
|
||||
"""
|
||||
print_step("Finding a spot in the backgrounds video to chop...✂️")
|
||||
choice = random.choice(listdir("assets/backgrounds"))
|
||||
environ["background_credit"] = choice.split("-")[0]
|
||||
credit = choice.split("-")[0]
|
||||
|
||||
background = VideoFileClip(f"assets/backgrounds/{choice}")
|
||||
|
||||
@@ -81,3 +83,4 @@ def chop_background_video(video_length: int):
|
||||
new = video.subclip(start_time, end_time)
|
||||
new.write_videofile("assets/temp/background.mp4")
|
||||
print_substep("Background video chopped successfully!", style="bold green")
|
||||
return credit
|
||||
@@ -20,25 +20,26 @@ from rich.console import Console
|
||||
from utils.cleanup import cleanup
|
||||
from utils.console import print_step, print_substep
|
||||
from utils.videos import save_data
|
||||
from utils import settings
|
||||
|
||||
console = Console()
|
||||
|
||||
W, H = 1080, 1920
|
||||
|
||||
|
||||
def name_normalize(
|
||||
name: str
|
||||
) -> str:
|
||||
name = re.sub(r'[?\\"%*:|<>]', '', name)
|
||||
name = re.sub(r'( [w,W]\s?\/\s?[o,O,0])', r' without', name)
|
||||
name = re.sub(r'( [w,W]\s?\/)', r' with', name)
|
||||
name = re.sub(r'([0-9]+)\s?\/\s?([0-9]+)', r'\1 of \2', name)
|
||||
name = re.sub(r'(\w+)\s?\/\s?(\w+)', r'\1 or \2', name)
|
||||
name = re.sub(r'\/', r'', name)
|
||||
def name_normalize(name: str) -> str:
|
||||
name = re.sub(r'[?\\"%*:|<>]', "", name)
|
||||
name = re.sub(r"( [w,W]\s?\/\s?[o,O,0])", r" without", name)
|
||||
name = re.sub(r"( [w,W]\s?\/)", r" with", name)
|
||||
name = re.sub(r"([0-9]+)\s?\/\s?([0-9]+)", r"\1 of \2", name)
|
||||
name = re.sub(r"(\w+)\s?\/\s?(\w+)", r"\1 or \2", name)
|
||||
name = re.sub(r"\/", r"", name)
|
||||
return name
|
||||
|
||||
|
||||
def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
def make_final_video(
|
||||
number_of_clips: int, length: int, reddit_obj: dict, background_credit: str
|
||||
):
|
||||
"""Gathers audio clips, gathers all screenshots, stitches them together and saves the final video to assets/temp
|
||||
|
||||
Args:
|
||||
@@ -49,7 +50,7 @@ def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
print_step("Creating the final video 🎥")
|
||||
VideoFileClip.reW = lambda clip: clip.resize(width=W)
|
||||
VideoFileClip.reH = lambda clip: clip.resize(width=H)
|
||||
opacity = os.getenv("OPACITY")
|
||||
opacity = settings.config["settings"]["opacity"]
|
||||
background_clip = (
|
||||
VideoFileClip("assets/temp/background.mp4")
|
||||
.without_audio()
|
||||
@@ -58,7 +59,9 @@ def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
)
|
||||
|
||||
# Gather all audio clips
|
||||
audio_clips = [AudioFileClip(f"assets/temp/mp3/{i}.mp3") for i in range(number_of_clips)]
|
||||
audio_clips = [
|
||||
AudioFileClip(f"assets/temp/mp3/{i}.mp3") for i in range(number_of_clips)
|
||||
]
|
||||
audio_clips.insert(0, AudioFileClip("assets/temp/mp3/title.mp3"))
|
||||
audio_concat = concatenate_audioclips(audio_clips)
|
||||
audio_composite = CompositeAudioClip([audio_concat])
|
||||
@@ -75,7 +78,7 @@ def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
.set_duration(audio_clips[0].duration)
|
||||
.set_position("center")
|
||||
.resize(width=W - 100)
|
||||
.set_opacity(new_opacity)
|
||||
.set_opacity(new_opacity),
|
||||
)
|
||||
|
||||
for i in range(0, number_of_clips):
|
||||
@@ -97,15 +100,18 @@ def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
# .set_opacity(float(opacity)),
|
||||
# )
|
||||
# else:
|
||||
image_concat = concatenate_videoclips(image_clips).set_position(("center", "center"))
|
||||
image_concat = concatenate_videoclips(image_clips).set_position(
|
||||
("center", "center")
|
||||
)
|
||||
image_concat.audio = audio_composite
|
||||
final = CompositeVideoClip([background_clip, image_concat])
|
||||
title = re.sub(r"[^\w\s-]", "", reddit_obj["thread_title"])
|
||||
idx = re.sub(r"[^\w\s-]", "", reddit_obj["thread_id"])
|
||||
filename = f"{name_normalize(title)}.mp4"
|
||||
subreddit = os.getenv("SUBREDDIT")
|
||||
|
||||
save_data(filename, title, idx)
|
||||
filename = f"{name_normalize(title)}.mp4"
|
||||
subreddit = settings.config["reddit"]["thread"]["subreddit"]
|
||||
|
||||
save_data(filename, title, idx, background_credit)
|
||||
|
||||
if not exists(f"./results/{subreddit}"):
|
||||
print_substep("The results folder didn't exist so I made it")
|
||||
@@ -120,7 +126,10 @@ def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
threads=multiprocessing.cpu_count(),
|
||||
)
|
||||
ffmpeg_tools.ffmpeg_extract_subclip(
|
||||
"assets/temp/temp.mp4", 0, final.duration, targetname=f"results/{subreddit}/{filename}"
|
||||
"assets/temp/temp.mp4",
|
||||
0,
|
||||
final.duration,
|
||||
targetname=f"results/{subreddit}/{filename}",
|
||||
)
|
||||
# os.remove("assets/temp/temp.mp4")
|
||||
|
||||
@@ -130,5 +139,5 @@ def make_final_video(number_of_clips: int, length: int, reddit_obj: dict):
|
||||
print_substep("See result in the results folder!")
|
||||
|
||||
print_step(
|
||||
f'Reddit title: {reddit_obj["thread_title"]} \n Background Credit: {os.getenv("background_credit")}'
|
||||
f'Reddit title: {reddit_obj["thread_title"]} \n Background Credit: {background_credit}'
|
||||
)
|
||||
@@ -1,9 +1,8 @@
|
||||
import json
|
||||
import os
|
||||
from os import getenv
|
||||
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from utils import settings
|
||||
from playwright.async_api import async_playwright # pylint: disable=unused-import
|
||||
|
||||
# do not remove the above line
|
||||
@@ -35,10 +34,14 @@ def download_screenshots_of_reddit_posts(reddit_object: dict, screenshot_num: in
|
||||
browser = p.chromium.launch()
|
||||
context = browser.new_context()
|
||||
|
||||
if getenv("THEME").upper() == "DARK":
|
||||
cookie_file = open("./video_creation/data/cookie-dark-mode.json", encoding="utf-8")
|
||||
if settings.config["settings"]["theme"] == "dark":
|
||||
cookie_file = open(
|
||||
"./video_creation/data/cookie-dark-mode.json", encoding="utf-8"
|
||||
)
|
||||
else:
|
||||
cookie_file = open("./video_creation/data/cookie-light-mode.json", encoding="utf-8")
|
||||
cookie_file = open(
|
||||
"./video_creation/data/cookie-light-mode.json", encoding="utf-8"
|
||||
)
|
||||
cookies = json.load(cookie_file)
|
||||
context.add_cookies(cookies) # load preference cookies
|
||||
# Get the thread screenshot
|
||||
@@ -56,9 +59,12 @@ def download_screenshots_of_reddit_posts(reddit_object: dict, screenshot_num: in
|
||||
|
||||
# translate code
|
||||
|
||||
if getenv("POSTLANG"):
|
||||
if settings.config["reddit"]["thread"]["post_lang"]:
|
||||
print_substep("Translating post...")
|
||||
texts_in_tl = ts.google(reddit_object["thread_title"], to_language=os.getenv("POSTLANG"))
|
||||
texts_in_tl = ts.google(
|
||||
reddit_object["thread_title"],
|
||||
to_language=settings.config["reddit"]["thread"]["post_lang"],
|
||||
)
|
||||
|
||||
page.evaluate(
|
||||
"tl_content => document.querySelector('[data-test-id=\"post-content\"] > div:nth-child(3) > div > div').textContent = tl_content",
|
||||
@@ -67,7 +73,9 @@ def download_screenshots_of_reddit_posts(reddit_object: dict, screenshot_num: in
|
||||
else:
|
||||
print_substep("Skipping translation...")
|
||||
|
||||
page.locator('[data-test-id="post-content"]').screenshot(path="assets/temp/png/title.png")
|
||||
page.locator('[data-test-id="post-content"]').screenshot(
|
||||
path="assets/temp/png/title.png"
|
||||
)
|
||||
|
||||
if storymode:
|
||||
page.locator('[data-click-id="text"]').screenshot(
|
||||
@@ -88,9 +96,10 @@ def download_screenshots_of_reddit_posts(reddit_object: dict, screenshot_num: in
|
||||
|
||||
# translate code
|
||||
|
||||
if getenv("POSTLANG"):
|
||||
if settings.config["reddit"]["thread"]["post_lang"]:
|
||||
comment_tl = ts.google(
|
||||
comment["comment_body"], to_language=os.getenv("POSTLANG")
|
||||
comment["comment_body"],
|
||||
to_language=settings.config["reddit"]["thread"]["post_lang"],
|
||||
)
|
||||
page.evaluate(
|
||||
'([tl_content, tl_id]) => document.querySelector(`#t1_${tl_id} > div:nth-child(2) > div > div[data-testid="comment"] > div`).textContent = tl_content',
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
import os
|
||||
from typing import Dict, Tuple
|
||||
|
||||
from rich.console import Console
|
||||
@@ -10,7 +9,7 @@ from TTS.GTTS import GTTS
|
||||
from TTS.streamlabs_polly import StreamlabsPolly
|
||||
from TTS.aws_polly import AWSPolly
|
||||
from TTS.TikTok import TikTok
|
||||
|
||||
from utils import settings
|
||||
from utils.console import print_table, print_step
|
||||
|
||||
|
||||
@@ -34,9 +33,11 @@ def save_text_to_mp3(reddit_obj) -> Tuple[int, int]:
|
||||
tuple[int,int]: (total length of the audio, the number of comments audio was generated for)
|
||||
"""
|
||||
|
||||
env = os.getenv("TTSCHOICE", "")
|
||||
if env.casefold() in map(lambda _: _.casefold(), TTSProviders):
|
||||
text_to_mp3 = TTSEngine(get_case_insensitive_key_value(TTSProviders, env), reddit_obj)
|
||||
voice = settings.config["settings"]["tts"]["choice"]
|
||||
if voice.casefold() in map(lambda _: _.casefold(), TTSProviders):
|
||||
text_to_mp3 = TTSEngine(
|
||||
get_case_insensitive_key_value(TTSProviders, voice), reddit_obj
|
||||
)
|
||||
else:
|
||||
while True:
|
||||
print_step("Please choose one of the following TTS providers: ")
|
||||
@@ -45,13 +46,19 @@ def save_text_to_mp3(reddit_obj) -> Tuple[int, int]:
|
||||
if choice.casefold() in map(lambda _: _.casefold(), TTSProviders):
|
||||
break
|
||||
print("Unknown Choice")
|
||||
text_to_mp3 = TTSEngine(get_case_insensitive_key_value(TTSProviders, choice), reddit_obj)
|
||||
text_to_mp3 = TTSEngine(
|
||||
get_case_insensitive_key_value(TTSProviders, choice), reddit_obj
|
||||
)
|
||||
|
||||
return text_to_mp3.run()
|
||||
|
||||
|
||||
def get_case_insensitive_key_value(input_dict, key):
|
||||
return next(
|
||||
(value for dict_key, value in input_dict.items() if dict_key.lower() == key.lower()),
|
||||
(
|
||||
value
|
||||
for dict_key, value in input_dict.items()
|
||||
if dict_key.lower() == key.lower()
|
||||
),
|
||||
None,
|
||||
)
|
||||
Reference in new issue
Block a user