init reinitializing
This commit is contained in:
Danilo Reyes
@@ -11,6 +11,7 @@ Also following in line more posix and python rules.
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
import argparse
|
||||
import yaml
|
||||
from typing import Dict
|
||||
from functions import LOG
|
||||
@@ -38,6 +39,20 @@ def init_globals() -> None:
|
||||
ARGS = argparser(USERS)
|
||||
|
||||
|
||||
def get_args() -> argparse.Namespace:
|
||||
"""Return initialized CLI args."""
|
||||
init_globals()
|
||||
assert ARGS is not None
|
||||
return ARGS
|
||||
|
||||
|
||||
def get_configs() -> dict:
|
||||
"""Return initialized config."""
|
||||
init_globals()
|
||||
assert CONFIGS is not None
|
||||
return CONFIGS
|
||||
|
||||
|
||||
class Video:
|
||||
"""Just a simple class to unify the Video parameters into a single one."""
|
||||
|
||||
@@ -50,29 +65,32 @@ class Video:
|
||||
|
||||
def get_index(name: str) -> int:
|
||||
"""Find the index in the config file"""
|
||||
return next((i for i, d in enumerate(CONFIGS["users"]) if d["name"] == name), -1)
|
||||
configs = get_configs()
|
||||
return next((i for i, d in enumerate(configs["users"]) if d["name"] == name), -1)
|
||||
|
||||
|
||||
def parse_gallery(gdl_list: str, user: User) -> None:
|
||||
"""Processes the gallery-dl command based on the selected gallery"""
|
||||
args = get_args()
|
||||
gallery = Gallery()
|
||||
gallery.archive = ARGS.flag_archive
|
||||
gallery.skip_arg = " -o skip=true" if not ARGS.flag_skip else ""
|
||||
gallery.archive = args.flag_archive
|
||||
gallery.skip_arg = " -o skip=true" if not args.flag_skip else ""
|
||||
gallery.dest = "download"
|
||||
gallery.list = gdl_list
|
||||
gallery.opt_args = parse_instagram(gdl_list)
|
||||
|
||||
gallery.generate_command(user)
|
||||
gallery.run_command(ARGS.flag_verbose)
|
||||
gallery.run_command(args.flag_verbose)
|
||||
|
||||
|
||||
def parse_instagram(link: str) -> list[str]:
|
||||
"""Fix instagram links"""
|
||||
args = get_args()
|
||||
if "instagram" not in link:
|
||||
return []
|
||||
if isinstance(ARGS.post_type, list):
|
||||
return ["-o", f"include={','.join(ARGS.post_type)}"]
|
||||
return ["-o", f"include={ARGS.post_type}"]
|
||||
if isinstance(args.post_type, list):
|
||||
return ["-o", f"include={','.join(args.post_type)}"]
|
||||
return ["-o", f"include={args.post_type}"]
|
||||
|
||||
|
||||
def video_command(video: Video):
|
||||
@@ -119,17 +137,19 @@ def video_command(video: Video):
|
||||
|
||||
def comic_manager(skip_arg: str, category: str) -> None:
|
||||
"""Process the information to download manga"""
|
||||
args = get_args()
|
||||
configs = get_configs()
|
||||
re_cat = "manga|webtoon" if category == "manga" else "readcomiconline"
|
||||
with open(CONFIGS["comic"]["comic-list"], "r", encoding="utf-8") as r_file:
|
||||
with open(configs["comic"]["comic-list"], "r", encoding="utf-8") as r_file:
|
||||
links = list(filter(lambda x: re.search(re_cat, x), r_file))
|
||||
|
||||
for link in links:
|
||||
gallery = Gallery()
|
||||
gallery.archive = ARGS.flag_archive
|
||||
gallery.archive = args.flag_archive
|
||||
gallery.skip_arg = skip_arg
|
||||
gallery.link = link
|
||||
gallery.generate_command(is_comic=True)
|
||||
gallery.run_command(ARGS.flag_verbose)
|
||||
gallery.run_command(args.flag_verbose)
|
||||
|
||||
|
||||
def print_webcomics(webcomics: Dict[str, Dict]) -> int:
|
||||
@@ -151,7 +171,9 @@ def print_webcomics(webcomics: Dict[str, Dict]) -> int:
|
||||
|
||||
def webcomic_manager():
|
||||
"""Process the information to download webcomics"""
|
||||
with open(CONFIGS["comic"]["webcomic-list"], "r", encoding="utf-8") as r_file:
|
||||
args = get_args()
|
||||
configs = get_configs()
|
||||
with open(configs["comic"]["webcomic-list"], "r", encoding="utf-8") as r_file:
|
||||
webcomics = yaml.safe_load(r_file)
|
||||
|
||||
usr_input = print_webcomics(webcomics)
|
||||
@@ -178,12 +200,13 @@ def webcomic_manager():
|
||||
"--cbz",
|
||||
]
|
||||
|
||||
run(command, ARGS.flag_verbose, cwd=Path(dest))
|
||||
run(command, args.flag_verbose, cwd=Path(dest))
|
||||
|
||||
|
||||
def save_comic(link: str) -> None:
|
||||
"""Add comic/manga link to the list"""
|
||||
list_comic = CONFIGS["comic"]["comic-list"]
|
||||
configs = get_configs()
|
||||
list_comic = configs["comic"]["comic-list"]
|
||||
with open(list_comic, "r", encoding="utf-8") as r_file:
|
||||
links = r_file.read().lower()
|
||||
if parse_link(link).lower() in links:
|
||||
@@ -197,6 +220,7 @@ def save_comic(link: str) -> None:
|
||||
|
||||
def push_manager(user: User):
|
||||
"""Filters out the URL to use the appropiate downloader"""
|
||||
args = get_args()
|
||||
# Creates an array which will store any links that should use youtube-dl
|
||||
rgx_gallery = re.compile(
|
||||
r"(x\.com\/\w+((?=.*media)|(?!.*status)))"
|
||||
@@ -239,38 +263,38 @@ def push_manager(user: User):
|
||||
|
||||
for link in links_galleries:
|
||||
gallery = Gallery()
|
||||
gallery.archive = ARGS.flag_archive
|
||||
gallery.skip_arg = " -o skip=true" if not ARGS.flag_skip else ""
|
||||
gallery.archive = args.flag_archive
|
||||
gallery.skip_arg = " -o skip=true" if not args.flag_skip else ""
|
||||
gallery.link = parse_link(link)
|
||||
gallery.dest = "download"
|
||||
gallery.opt_args = parse_instagram(link)
|
||||
gallery.generate_command(user)
|
||||
gallery.run_command(ARGS.flag_verbose)
|
||||
gallery.run_command(args.flag_verbose)
|
||||
user.save_link(link)
|
||||
|
||||
for link in links_comics:
|
||||
if ARGS.flag_skip and re.search(r"readcomiconline", link):
|
||||
if args.flag_skip and re.search(r"readcomiconline", link):
|
||||
skip_arg = " --chapter-range 1"
|
||||
elif ARGS.flag_skip and re.search(r"manganato|mangahere|webtoons", link):
|
||||
elif args.flag_skip and re.search(r"manganato|mangahere|webtoons", link):
|
||||
skip_arg = " --chapter-range 1-5"
|
||||
else:
|
||||
skip_arg = ""
|
||||
|
||||
gallery = Gallery()
|
||||
gallery.archive = ARGS.flag_archive
|
||||
gallery.archive = args.flag_archive
|
||||
gallery.skip_arg = skip_arg
|
||||
gallery.link = link
|
||||
gallery.generate_command(is_comic=True)
|
||||
gallery.run_command(ARGS.flag_verbose)
|
||||
gallery.run_command(args.flag_verbose)
|
||||
save_comic(link)
|
||||
|
||||
for link in links_videos:
|
||||
video = Video()
|
||||
video.use_archive = ARGS.flag_archive
|
||||
video.use_archive = args.flag_archive
|
||||
video.link = link
|
||||
video.dest = str(user.directories["media"])
|
||||
video.database = str(user.dbs["media"])
|
||||
run(video_command(video), ARGS.flag_verbose)
|
||||
run(video_command(video), args.flag_verbose)
|
||||
|
||||
for link in links_other:
|
||||
LOG.info("Other type of download %s", link)
|
||||
@@ -280,7 +304,7 @@ def push_manager(user: User):
|
||||
gallery.link = link
|
||||
gallery.dest = "push"
|
||||
gallery.generate_command(user)
|
||||
gallery.run_command(ARGS.flag_verbose)
|
||||
gallery.run_command(args.flag_verbose)
|
||||
|
||||
# Flush the push list, cleans all the contents
|
||||
with open(user.lists["push"], "w", encoding="utf-8") as w_file:
|
||||
@@ -289,41 +313,43 @@ def push_manager(user: User):
|
||||
|
||||
def scrapper_manager(user: User) -> None:
|
||||
"""Analyze the user arguments and call in functions"""
|
||||
args = get_args()
|
||||
user.list_manager()
|
||||
if re.search(r"main|instagram|kemono", ARGS.scrapper):
|
||||
skip_arg = "" if ARGS.flag_skip else " -o skip=true"
|
||||
parse_gallery(ARGS.scrapper, user)
|
||||
elif ARGS.scrapper == "push":
|
||||
if re.search(r"main|instagram|kemono", args.scrapper):
|
||||
parse_gallery(args.scrapper, user)
|
||||
elif args.scrapper == "push":
|
||||
push_manager(user)
|
||||
elif re.search("^comic|manga", ARGS.scrapper):
|
||||
skip_arg = " --chapter-range 1" if ARGS.flag_skip else ""
|
||||
skip_arg += "-5" if ARGS.scrapper == "manga" else ""
|
||||
comic_manager(skip_arg, ARGS.scrapper)
|
||||
elif re.search("webcomic", ARGS.scrapper):
|
||||
elif re.search("^comic|manga", args.scrapper):
|
||||
skip_arg = " --chapter-range 1" if args.flag_skip else ""
|
||||
skip_arg += "-5" if args.scrapper == "manga" else ""
|
||||
comic_manager(skip_arg, args.scrapper)
|
||||
elif re.search("webcomic", args.scrapper):
|
||||
webcomic_manager()
|
||||
|
||||
|
||||
def scrap_everyone() -> None:
|
||||
"""Iterates over every user of my scrapper"""
|
||||
for current_user in CONFIGS["users"]:
|
||||
args = get_args()
|
||||
configs = get_configs()
|
||||
for current_user in configs["users"]:
|
||||
user = User(get_index(current_user["name"]))
|
||||
LOG.info("Scrapping %s for %s", ARGS.scrapper, current_user["name"])
|
||||
LOG.info("Scrapping %s for %s", args.scrapper, current_user["name"])
|
||||
scrapper_manager(user)
|
||||
|
||||
|
||||
def main():
|
||||
"""Main module to decide what to do based on the parsed arguments"""
|
||||
init_globals()
|
||||
if ARGS.scrapper:
|
||||
args = get_args()
|
||||
if args.scrapper:
|
||||
rgx_shared = re.compile("push|main|instagram|kemono")
|
||||
if (ARGS.user == "everyone") and (rgx_shared.search(ARGS.scrapper)):
|
||||
if (args.user == "everyone") and (rgx_shared.search(args.scrapper)):
|
||||
scrap_everyone()
|
||||
else:
|
||||
scrapper_manager(User(get_index(ARGS.user)))
|
||||
elif ARGS.link:
|
||||
is_admin = ARGS.user in ("everyone", "jawz")
|
||||
user = User(get_index("jawz" if is_admin else ARGS.user))
|
||||
for arg_link in [lnk for grp in ARGS.link for lnk in grp]:
|
||||
scrapper_manager(User(get_index(args.user)))
|
||||
elif args.link:
|
||||
is_admin = args.user in ("everyone", "jawz")
|
||||
user = User(get_index("jawz" if is_admin else args.user))
|
||||
for arg_link in [lnk for grp in args.link for lnk in grp]:
|
||||
user.append_list("push", parse_link(arg_link))
|
||||
|
||||
push_manager(user)
|
||||
|
||||
Reference in New Issue
Block a user