jawz/scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

migrate os.run to subprocess

Browse Source
This commit is contained in:
Danilo Reyes
2026-02-28 19:21:36 -06:00
parent 7b6e411ebc
commit 9149a03898
3 changed files with 108 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
src/download/classes/gallery.py
View File

@@ -1,8 +1,10 @@
#!/usr/bin/env python3
import os
import shlex
from typing import Sequence
from classes.user import User
from functions import LOG
from functions import load_config_variables
from functions import quote
from functions import run
@@ -13,43 +15,54 @@ class Gallery:
self.link: str = ""
self.dest: str = ""
self.list: str = ""
self.opt_args: str = ""
self.command: str = ""
self.opt_args: str | Sequence[str] = ""
self.command: list[str] = []
def generate_command(self, user: User = User(1), is_comic: bool = False) -> None:
"""Generates a command string."""
if is_comic:
configs = load_config_variables()
directory = quote(configs["comic"]["download-dir"])
database = quote(configs["comic"]["database"])
queue = quote(configs["comic"][f"{self.list}-list"]) if self.list else ""
directory = str(configs["comic"]["download-dir"])
database = str(configs["comic"]["database"])
queue = str(configs["comic"][f"{self.list}-list"]) if self.list else ""
else:
directory = quote(str(user.directories[self.dest]))
database = quote(str(user.dbs["gallery"]))
queue = quote(str(user.lists[self.list])) if self.list else ""
directory = str(user.directories[self.dest])
database = str(user.dbs["gallery"])
queue = str(user.lists[self.list]) if self.list else ""
command = f"gallery-dl --sleep {str(user.sleep)}"
command += self.skip_arg if self.skip_arg else ""
command += f" --dest {directory}" if self.dest or is_comic else ""
command += f" --download-archive {database}" if self.archive else ""
command += self.opt_args if self.opt_args else ""
command = ["gallery-dl", "--sleep", str(user.sleep)]
if self.skip_arg:
command += shlex.split(self.skip_arg)
if self.dest or is_comic:
command += ["--dest", directory]
if self.archive:
command += ["--download-archive", database]
if self.opt_args:
if isinstance(self.opt_args, str):
command += shlex.split(self.opt_args)
else:
command += list(self.opt_args)
# Add authentication options from environment variables
command += ' -o "extractor.pixiv.refresh-token=${GALLERY_DL_PIXIV_REFRESH_TOKEN}"'
command += ' -o "extractor.bluesky.password=${GALLERY_DL_BLUESKY_PASSWORD}"'
command += ' -o "extractor.deviantart.refresh-token=${GALLERY_DL_DEVIANTART_REFRESH_TOKEN}"'
command += ' -o "extractor.flickr.access-token=${GALLERY_DL_FLICKR_ACCESS_TOKEN}"'
command += ' -o "extractor.flickr.access-token-secret=${GALLERY_DL_FLICKR_ACCESS_TOKEN_SECRET}"'
command += ' -o "extractor.reddit.refresh-token=${GALLERY_DL_REDDIT_REFRESH_TOKEN}"'
command += ' -o "extractor.tumblr.access-token=${GALLERY_DL_TUMBLR_ACCESS_TOKEN}"'
command += ' -o "extractor.tumblr.access-token-secret=${GALLERY_DL_TUMBLR_ACCESS_TOKEN_SECRET}"'
command += ' -o "extractor.tumblr.api-key=${GALLERY_DL_TUMBLR_API_KEY}"'
command += ' -o "extractor.tumblr.api-secret=${GALLERY_DL_TUMBLR_API_SECRET}"'
auth_env = {
"extractor.pixiv.refresh-token": "GALLERY_DL_PIXIV_REFRESH_TOKEN",
"extractor.bluesky.password": "GALLERY_DL_BLUESKY_PASSWORD",
"extractor.deviantart.refresh-token": "GALLERY_DL_DEVIANTART_REFRESH_TOKEN",
"extractor.flickr.access-token": "GALLERY_DL_FLICKR_ACCESS_TOKEN",
"extractor.flickr.access-token-secret": "GALLERY_DL_FLICKR_ACCESS_TOKEN_SECRET",
"extractor.reddit.refresh-token": "GALLERY_DL_REDDIT_REFRESH_TOKEN",
"extractor.tumblr.access-token": "GALLERY_DL_TUMBLR_ACCESS_TOKEN",
"extractor.tumblr.access-token-secret": "GALLERY_DL_TUMBLR_ACCESS_TOKEN_SECRET",
"extractor.tumblr.api-key": "GALLERY_DL_TUMBLR_API_KEY",
"extractor.tumblr.api-secret": "GALLERY_DL_TUMBLR_API_SECRET",
}
for key, env_var in auth_env.items():
command += ["-o", f"{key}={os.environ.get(env_var, '')}"]
if self.link and not self.list:
command += f" {quote(self.link)}"
command.append(self.link)
if self.list and not self.link:
command += f" -i {queue}"
command += ["-i", queue]
LOG.debug(command)
self.command = command

75
src/download/download.py
View File

@@ -10,11 +10,11 @@ Also following in line more posix and python rules.
"""
import re
from pathlib import Path
import yaml
from typing import Dict
from functions import LOG
from functions import run
from functions import quote
from functions import list_lines
from functions import load_config_variables
from functions import parse_link
@@ -57,40 +57,55 @@ def parse_gallery(gdl_list: str, user: User) -> None:
gallery.run_command(ARGS.flag_verbose)
def parse_instagram(link: str) -> str:
def parse_instagram(link: str) -> list[str]:
"""Fix instagram links"""
if "instagram" not in link:
return ""
return []
if isinstance(ARGS.post_type, list):
return f" -o include={quote(','.join(ARGS.post_type))}"
return f" -o include={quote(ARGS.post_type)}"
return ["-o", f"include={','.join(ARGS.post_type)}"]
return ["-o", f"include={ARGS.post_type}"]
def video_command(video: Video) -> str:
def video_command(video: Video):
"""Filters and processes the required command to download videos"""
command = "yt-dlp"
command = ["yt-dlp"]
rgx_yt = re.compile(r"(https:\/\/youtube|https:\/\/www.youtube|https:\/\/youtu.be)")
rgx_music = re.compile(r"(https:\/\/music.youtube.*)")
if re.search(r"chaturbate", video.link):
return f"stream-dl {video.link.rstrip("/").split("/")[-1]}"
return ["stream-dl", video.link.rstrip("/").split("/")[-1]]
if rgx_yt.search(video.link):
command += " --embed-subs --embed-thumbnail"
command += " --embed-metadata --embed-chapters"
command += f" -o {quote(video.dest + '/%(title)s.%(ext)s')}"
command += [
"--embed-subs",
"--embed-thumbnail",
"--embed-metadata",
"--embed-chapters",
"-o",
f"{video.dest}/%(title)s.%(ext)s",
]
elif rgx_music.search(video.link):
command += f" --download-archive {video.database}" if video.use_archive else ""
command += " --no-playlist --newline -x"
command += " --audio-format best --add-metadata --audio-quality 0 -o"
command += f" {quote(video.dest + '/%(title)s.%(ext)s')}"
if video.use_archive:
command += ["--download-archive", video.database]
command += [
"--no-playlist",
"--newline",
"-x",
"--audio-format",
"best",
"--add-metadata",
"--audio-quality",
"0",
"-o",
f"{video.dest}/%(title)s.%(ext)s",
]
else: # Any other video link, just do it generic
command += f" -f mp4 -o {quote(video.dest + '/%(title)s.%(ext)s')}"
command += ["-f", "mp4", "-o", f"{video.dest}/%(title)s.%(ext)s"]
LOG.info("%s %s", command, video.link)
return f"{command} {quote(video.link)}"
LOG.info("%s %s", " ".join(command), video.link)
return command + [video.link]
def comic_manager(skip_arg: str, category: str) -> None:
@@ -133,15 +148,19 @@ def webcomic_manager():
LOG.info("The webcomic is %s", dest)
command = f"cd {quote(dest)} && webcomix custom"
command += f" {quote(name)}"
command += " --start-url"
command += f" {quote(link)}"
command += f" --next-page-xpath={quote(nxt_code)}"
command += f" --image-xpath={quote(img_code)}"
command += " -y --cbz"
command = [
"webcomix",
"custom",
name,
"--start-url",
link,
f"--next-page-xpath={nxt_code}",
f"--image-xpath={img_code}",
"-y",
"--cbz",
]
run(command, ARGS.flag_verbose)
run(command, ARGS.flag_verbose, cwd=Path(dest))
def save_comic(link: str) -> None:
@@ -231,8 +250,8 @@ def push_manager(user: User):
video = Video()
video.use_archive = ARGS.flag_archive
video.link = link
video.dest = f"{user.directories['media']}"
video.database = quote(f"{user.dbs['media']}")
video.dest = str(user.directories["media"])
video.database = str(user.dbs["media"])
run(video_command(video), ARGS.flag_verbose)
for link in links_other:

29
src/download/functions.py
View File

@@ -6,6 +6,9 @@ import fileinput
import re
import os
import logging
import shlex
import subprocess
from typing import Sequence
from pathlib import Path
import yaml
@@ -60,14 +63,24 @@ def clean_cache(directory: Path):
directory.rmdir()
def run(command: str, verbose: bool):
def run(command: str | Sequence[str], verbose: bool, cwd: Path | None = None) -> None:
"""Run command in a subprocess"""
# pylint: disable=subprocess-run-check
# This toggle allows for a really wasy debug when using -v
if verbose:
print(command)
if isinstance(command, (list, tuple)):
print(shlex.join(command))
else:
print(command)
return
if isinstance(command, str):
expanded = os.path.expandvars(command)
args = shlex.split(expanded)
else:
os.system(command)
args = list(command)
subprocess.run(args, check=False, cwd=cwd)
def list_lines(i: int, line: str) -> str:
@@ -84,19 +97,19 @@ def sort_txt_file(file_path: Path):
"""Sort every line alphabetically
remove duplicated and empty lines"""
file = str(file_path.resolve())
run(f"sort -u {quote(file)} -o {quote(file)}", VERBOSE_G)
run(f"sed -i '/^$/d' {quote(file)}", VERBOSE_G)
run(f'sed -i -e "s,http:,https:," {quote(file)}', VERBOSE_G)
run(["sort", "-u", file, "-o", file], VERBOSE_G)
run(["sed", "-i", "/^$/d", file], VERBOSE_G)
run(["sed", "-i", "-e", "s,http:,https:,", file], VERBOSE_G)
# fix this using strip on python
# line.strip("/")
run(f'sed -i -e "s,/$,," {quote(file)}', VERBOSE_G) # trailing /
run(["sed", "-i", "-e", "s,/$,,", file], VERBOSE_G) # trailing /
def randomize_txt_file(file_path: Path):
"""Randomize the order of the
lines of the txt file"""
file = str(file_path.resolve())
run(f"sort -R {quote(file)} -o {quote(file)}", VERBOSE_G)
run(["sort", "-R", file, "-o", file], VERBOSE_G)
def parse_list(file):