fixed a error(self-made), added default values

pull/1085/head
Supreme-hub 2 years ago
parent 6ca3223ca3
commit ae41127a91

@ -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}
@ -81,7 +79,9 @@ class TikTok: # TikTok Text-to-Speech Wrapper
)
)
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()
@ -89,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)

@ -40,7 +40,9 @@ class AWSPolly:
raise ValueError(
f"Please set the TOML variable AWS_VOICE to a valid voice. options are: {voices}"
)
voice = str(settings.config["settings"]["tts"]["aws_polly_voice"]).capitalize()
voice = str(
settings.config["settings"]["tts"]["aws_polly_voice"]
).capitalize()
try:
# Request speech synthesis
response = polly.synthesize_speech(

@ -59,10 +59,7 @@ class TTSEngine:
self.call_tts("title", process_text(self.reddit_object["thread_title"]))
processed_text = process_text(self.reddit_object["thread_post"])
if (
processed_text != ""
and settings.config["settings"]["storymode"] == True
):
if processed_text != "" and settings.config["settings"]["storymode"] == True:
self.call_tts("posttext", processed_text)
idx = None
@ -95,7 +92,7 @@ class TTSEngine:
offset = 0
for idy, text_cut in enumerate(split_text):
# print(f"{idx}-{idy}: {text_cut}\n")
new_text = process_text(text_cut)
new_text = process_text(text_cut)
if not new_text or new_text.isspace():
offset += 1
continue
@ -120,9 +117,7 @@ class TTSEngine:
# Path(f"{self.path}/{idx}-{i}.part.mp3").unlink()
def call_tts(self, filename: str, text: str):
self.tts_module.run(
text, filepath=f"{self.path}/{filename}.mp3"
)
self.tts_module.run(text, filepath=f"{self.path}/{filename}.mp3")
# try:
# self.length += MP3(f"{self.path}/{filename}.mp3").info.length
# except (MutagenError, HeaderNotFoundError):

@ -1,13 +1,12 @@
import random
import pyttsx3
from utils import settings
import random
max_chars = 0
# Uses the system voices, significantly faster than other tts
#Uses the system voices, significantly faster than other tts
class pyttsx:
class Pyttsx:
def __init__(self):
self.max_chars = 0
self.voices = []
@ -17,22 +16,27 @@ class pyttsx:
text: str = "Python Text to Speech",
filepath: str = "assets/temp/mp3",
random_voice=False,
censor=False,
):
voice_id = int(settings.config["settings"]["tts"]["python_voice"])
voice_num = int(settings.config["settings"]["tts"]["python_voice_number"])
for i in range(voice_num-1):
if (settings.config["settings"]["tts"]["python_voice"] == '' or settings.config["settings"]["tts"]["python_voice"] == ""):
voice_id = 0
voice_num = 3
else:
voice_id = int(settings.config["settings"]["tts"]["python_voice"])
voice_num = int(settings.config["settings"]["tts"]["python_voice"])
for i in range(voice_num - 1):
self.voices.append(i)
i=+1
i = +1
if random_voice:
voice_id = self.randomvoice()
else:
voice = voice_id
engine = pyttsx3.init()
voices = engine.getProperty('voices')
engine.setProperty('voice', voices[voice].id) #changing index changes voices but ony 0 and 1 are working here
voices = engine.getProperty("voices")
engine.setProperty(
"voice", voices[voice].id
) # changing index changes voices but ony 0 and 1 are working here
engine.save_to_file(text, f"{filepath}")
engine.runAndWait()
def randomvoice(self):
return random.choice(self.voices)
return random.choice(self.voices)

@ -40,7 +40,9 @@ class StreamlabsPolly:
raise ValueError(
f"Please set the config variable STREAMLABS_POLLY_VOICE to a valid voice. options are: {voices}"
)
voice = str(settings.config["settings"]["tts"]["streamlabs_polly_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)
if not check_ratelimit(response):

@ -75,7 +75,9 @@ if __name__ == "__main__":
run_many(config["settings"]["times_to_run"])
elif len(config["reddit"]["thread"]["post_id"].split("+")) > 1:
for index, post_id in enumerate(config["reddit"]["thread"]["post_id"].split("+")):
for index, post_id in enumerate(
config["reddit"]["thread"]["post_id"].split("+")
):
index += 1
print_step(
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("+"))}'

@ -1,9 +1,10 @@
import pyttsx3
engine = pyttsx3.init()
voices = engine.getProperty('voices')
voices = engine.getProperty("voices")
for voice in voices:
print(voice, voice.id)
engine.setProperty('voice', voice.id)
engine.setProperty("voice", voice.id)
engine.say("Hello World!")
engine.runAndWait()
engine.stop()
engine.stop()

@ -19,7 +19,9 @@ def get_subreddit_threads(POST_ID: str):
content = {}
if settings.config["reddit"]["creds"]["2fa"]:
print("\nEnter your two-factor authentication code from your authenticator app.\n")
print(
"\nEnter your two-factor authentication code from your authenticator app.\n"
)
code = input("> ")
print()
pw = settings.config["reddit"]["creds"]["password"]
@ -45,7 +47,9 @@ def get_subreddit_threads(POST_ID: str):
]: # 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:
@ -55,7 +59,9 @@ def get_subreddit_threads(POST_ID: str):
sub = settings.config["reddit"]["thread"]["subreddit"]
print_substep(f"Using subreddit: r/{sub} from TOML config")
subreddit_choice = sub
if str(subreddit_choice).casefold().startswith("r/"): # removes the r/ from the input
if (
str(subreddit_choice).casefold().startswith("r/")
): # removes the r/ from the input
subreddit_choice = subreddit_choice[2:]
subreddit = reddit.subreddit(
subreddit_choice
@ -67,7 +73,9 @@ def get_subreddit_threads(POST_ID: str):
settings.config["reddit"]["thread"]["post_id"]
and len(str(settings.config["reddit"]["thread"]["post_id"]).split("+")) == 1
):
submission = reddit.submission(id=settings.config["reddit"]["thread"]["post_id"])
submission = reddit.submission(
id=settings.config["reddit"]["thread"]["post_id"]
)
else:
threads = subreddit.hot(limit=25)
submission = get_subreddit_undone(threads, subreddit)

Binary file not shown.

@ -14,7 +14,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)

@ -49,7 +49,10 @@ def handle_input(
optional=False,
):
if optional:
console.print(message + "\n[green]This is an optional value. Do you want to skip it? (y/n)")
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:
@ -83,7 +86,11 @@ def handle_input(
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)")
console.print(
"[red]"
+ err_message
+ "\nAre you absolutely sure it's correct?(y/n)"
)
if input().casefold().startswith("y"):
break
continue
@ -116,5 +123,9 @@ def handle_input(
if user_input in options:
return user_input
console.print(
"[red bold]" + err_message + "\nValid options are: " + ", ".join(map(str, options)) + "."
"[red bold]"
+ err_message
+ "\nValid options are: "
+ ", ".join(map(str, options))
+ "."
)

@ -54,7 +54,11 @@ def check(value, checks, name):
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"])
or (
"nmax" in checks
and checks["nmax"] is not None
and value > checks["nmax"]
)
)
):
incorrect = True
@ -62,8 +66,16 @@ def check(value, checks, name):
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"])
(
"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
@ -71,9 +83,15 @@ def check(value, checks, name):
if incorrect:
value = handle_input(
message=(
(("[blue]Example: " + str(checks["example"]) + "\n") if "example" in checks else "")
(
("[blue]Example: " + str(checks["example"]) + "\n")
if "example" in checks
else ""
)
+ "[red]"
+ ("Non-optional ", "Optional ")["optional" in checks and checks["optional"] is True]
+ ("Non-optional ", "Optional ")[
"optional" in checks and checks["optional"] is True
]
)
+ "[#C0CAF5 bold]"
+ str(name)
@ -114,7 +132,9 @@ def check_toml(template_file, config_file) -> Tuple[bool, Dict]:
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}")
console.print(
f"[red bold]Encountered error when trying to to load {template_file}: {error}"
)
return False
try:
config = toml.load(config_file)

@ -19,7 +19,9 @@ def get_subreddit_undone(submissions: list, subreddit, times_checked=0):
if not exists("./video_creation/data/videos.json"):
with open("./video_creation/data/videos.json", "w+") as f:
json.dump([], f)
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):
@ -34,7 +36,9 @@ def get_subreddit_undone(submissions: list, subreddit, times_checked=0):
if submission.stickied:
print_substep("This post was pinned by moderators. Skipping...")
continue
if submission.num_comments <= int(settings.config["reddit"]["thread"]["min_comments"]):
if submission.num_comments <= int(
settings.config["reddit"]["thread"]["min_comments"]
):
print_substep(
f'This post has under the specified minimum of comments ({settings.config["reddit"]["thread"]["min_comments"]}). Skipping...'
)
@ -55,7 +59,8 @@ def get_subreddit_undone(submissions: list, subreddit, times_checked=0):
return get_subreddit_undone(
subreddit.top(
time_filter=VALID_TIME_FILTERS[index], limit=(50 if int(index) == 0 else index + 1 * 50)
time_filter=VALID_TIME_FILTERS[index],
limit=(50 if int(index) == 0 else index + 1 * 50),
),
subreddit,
times_checked=index,

@ -35,10 +35,18 @@ class Video:
return ImageClip(path)
def add_watermark(
self, text, opacity=0.5, duration: int | float = 5, position: Tuple = (0.7, 0.9), fontsize=15
self,
text,
opacity=0.5,
duration: int | float = 5,
position: Tuple = (0.7, 0.9),
fontsize=15,
):
compensation = round(
(position[0] / ((len(text) * (fontsize / 5) / 1.5) / 100 + position[0] * position[0])),
(
position[0]
/ ((len(text) * (fontsize / 5) / 1.5) / 100 + position[0] * position[0])
),
ndigits=2,
)
position = (compensation, position[1])

@ -20,7 +20,9 @@ def check_done(
Returns:
Submission|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):
@ -34,7 +36,9 @@ def check_done(
return redditobj
def save_data(subreddit: str, filename: str, reddit_title: str, reddit_id: str, credit: str):
def save_data(
subreddit: str, 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:

@ -40,7 +40,9 @@ def sleep_until(time):
if sys.version_info[0] >= 3 and time.tzinfo:
end = time.astimezone(timezone.utc).timestamp()
else:
zoneDiff = pytime.time() - (datetime.now() - datetime(1970, 1, 1)).total_seconds()
zoneDiff = (
pytime.time() - (datetime.now() - datetime(1970, 1, 1)).total_seconds()
)
end = (time - datetime(1970, 1, 1)).total_seconds() + zoneDiff
# Type check

@ -31,7 +31,9 @@ def get_start_and_end_times(video_length: int, length_of_clip: int) -> Tuple[int
def get_background_config():
"""Fetch the background/s configuration"""
try:
choice = str(settings.config["settings"]["background"]["background_choice"]).casefold()
choice = str(
settings.config["settings"]["background"]["background_choice"]
).casefold()
except AttributeError:
print_substep("No background selected. Picking random background'")
choice = None
@ -56,13 +58,15 @@ def download_background(background_config: Tuple[str, str, str, Any]):
)
print_substep("Downloading the backgrounds videos... please be patient 🙏 ")
print_substep(f"Downloading {filename} from {uri}")
YouTube(uri, on_progress_callback=on_progress).streams.filter(res="1080p").first().download(
"assets/backgrounds", filename=f"{credit}-{filename}"
)
YouTube(uri, on_progress_callback=on_progress).streams.filter(
res="1080p"
).first().download("assets/backgrounds", filename=f"{credit}-{filename}")
print_substep("Background video downloaded successfully! 🎉", style="bold green")
def chop_background_video(background_config: Tuple[str, str, str, Any], video_length: int):
def chop_background_video(
background_config: Tuple[str, str, str, Any], video_length: int
):
"""Generates the background footage to be used in the video and writes it to assets/temp/background.mp4
Args:

@ -75,7 +75,9 @@ def make_final_video(
)
# 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])
@ -85,7 +87,9 @@ def make_final_video(
image_clips = []
# Gather all images
new_opacity = 1 if opacity is None or float(opacity) >= 1 else float(opacity)
new_transition = 0 if transition is None or float(transition) > 2 else float(transition)
new_transition = (
0 if transition is None or float(transition) > 2 else float(transition)
)
image_clips.insert(
0,
ImageClip("assets/temp/png/title.png")

@ -36,9 +36,13 @@ def download_screenshots_of_reddit_posts(reddit_object: dict, screenshot_num: in
context = browser.new_context()
if settings.config["settings"]["theme"] == "dark":
cookie_file = open("./video_creation/data/cookie-dark-mode.json", encoding="utf-8")
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
@ -50,7 +54,7 @@ def download_screenshots_of_reddit_posts(reddit_object: dict, screenshot_num: in
print_substep("Post is NSFW. You are spicy...")
page.locator('[data-testid="content-gate"] button').click()
page.wait_for_load_state() # Wait for page to fully load
page.wait_for_load_state() # Wait for page to fully load
if page.locator('[data-click-id="text"] button').is_visible():
page.locator(
@ -73,7 +77,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(

@ -9,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 TTS.pyttsx import pyttsx
from TTS.pyttsx import Pyttsx
from utils import settings
from utils.console import print_table, print_step
@ -21,7 +21,7 @@ TTSProviders = {
"AWSPolly": AWSPolly,
"StreamlabsPolly": StreamlabsPolly,
"TikTok": TikTok,
"pyttsx" : pyttsx,
"pyttsx": Pyttsx,
}
@ -37,7 +37,9 @@ def save_text_to_mp3(reddit_obj) -> Tuple[int, int]:
voice = settings.config["settings"]["tts"]["voice_choice"]
if str(voice).casefold() in map(lambda _: _.casefold(), TTSProviders):
text_to_mp3 = TTSEngine(get_case_insensitive_key_value(TTSProviders, voice), reddit_obj)
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: ")
@ -46,12 +48,18 @@ 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,
)

Loading…
Cancel
Save