318 lines
12 KiB
Python
318 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Tool to convert yt-dlp command-line flags to a JSON config using go-ytdlp.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Dict, List
|
|
|
|
logger = logging.getLogger('config_tool')
|
|
|
|
|
|
def get_go_ytdlp_path(user_path: str = None) -> str:
|
|
"""
|
|
Get the path to the go-ytdlp binary.
|
|
Checks in order:
|
|
1. User-provided path
|
|
2. 'go-ytdlp' in PATH
|
|
3. Local binary in ytops_client/go_ytdlp_cli/go-ytdlp
|
|
4. Binary in go-ytdlp/go-ytdlp (the library's built binary)
|
|
5. Binary in /usr/local/bin/go-ytdlp
|
|
"""
|
|
def is_exe(fpath):
|
|
return os.path.isfile(fpath) and os.access(fpath, os.X_OK)
|
|
|
|
if user_path:
|
|
if is_exe(user_path):
|
|
return user_path
|
|
# If user provided a path, we return it even if check fails,
|
|
# so subprocess can raise the appropriate error for that specific path.
|
|
return user_path
|
|
|
|
# Check in PATH
|
|
import shutil
|
|
path_exe = shutil.which('go-ytdlp')
|
|
if path_exe:
|
|
return path_exe
|
|
|
|
# Check local build directory
|
|
local_path = Path(__file__).parent / 'go_ytdlp_cli' / 'go-ytdlp'
|
|
if is_exe(str(local_path)):
|
|
return str(local_path)
|
|
|
|
# Check the go-ytdlp library directory
|
|
project_root = Path(__file__).parent.parent
|
|
library_binary = project_root / 'go-ytdlp' / 'go-ytdlp'
|
|
if is_exe(str(library_binary)):
|
|
return str(library_binary)
|
|
|
|
# Check /usr/local/bin
|
|
if is_exe('/usr/local/bin/go-ytdlp'):
|
|
return '/usr/local/bin/go-ytdlp'
|
|
|
|
# Default to 'go-ytdlp' which will raise FileNotFoundError if not in PATH
|
|
return 'go-ytdlp'
|
|
|
|
def convert_flags_to_json(flags: List[str], go_ytdlp_path: str = None) -> Dict:
|
|
"""
|
|
Converts a list of yt-dlp command-line flags to a JSON config dictionary.
|
|
|
|
Args:
|
|
flags: A list of strings representing the command-line flags.
|
|
go_ytdlp_path: Path to the go-ytdlp executable. If None, will try to find it.
|
|
|
|
Returns:
|
|
A dictionary representing the JSON config.
|
|
|
|
Raises:
|
|
ValueError: If no flags are provided.
|
|
FileNotFoundError: If the go-ytdlp executable is not found.
|
|
subprocess.CalledProcessError: If go-ytdlp returns a non-zero exit code.
|
|
json.JSONDecodeError: If the output from go-ytdlp is not valid JSON.
|
|
"""
|
|
if not flags:
|
|
raise ValueError("No flags provided to convert.")
|
|
|
|
# Get the actual binary path
|
|
actual_path = get_go_ytdlp_path(go_ytdlp_path)
|
|
|
|
# Use '--' to separate the subcommand flags from the flags to be converted.
|
|
# This prevents go-ytdlp from trying to parse the input flags as its own flags.
|
|
cmd = [actual_path, 'flags-to-json', '--'] + flags
|
|
|
|
logger.debug(f"Executing command: {' '.join(shlex.quote(s) for s in cmd)}")
|
|
try:
|
|
process = subprocess.run(cmd, capture_output=True, check=True, encoding='utf-8')
|
|
|
|
if process.stderr:
|
|
logger.info(f"go-ytdlp output on stderr:\n{process.stderr.strip()}")
|
|
|
|
return json.loads(process.stdout)
|
|
except json.JSONDecodeError:
|
|
logger.error("Failed to parse JSON from go-ytdlp stdout.")
|
|
logger.error(f"Stdout was: {process.stdout.strip()}")
|
|
raise
|
|
except FileNotFoundError:
|
|
logger.error(f"Executable '{actual_path}' not found.")
|
|
logger.error("Please ensure go-ytdlp is installed and in your PATH.")
|
|
logger.error("You can run the 'bin/install-goytdlp.sh' script to install it.")
|
|
raise
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"go-ytdlp exited with error code {e.returncode}.")
|
|
logger.error(f"Stderr:\n{e.stderr.strip()}")
|
|
if "not supported" in e.stderr:
|
|
logger.error("NOTE: The installed version of go-ytdlp does not support converting flags to JSON.")
|
|
raise
|
|
except PermissionError:
|
|
logger.error(f"Permission denied executing '{actual_path}'.")
|
|
logger.error("Please ensure the file is executable (chmod +x).")
|
|
raise
|
|
|
|
|
|
def add_flags_to_json_parser(subparsers):
|
|
"""Add the parser for the 'flags-to-json' command."""
|
|
parser = subparsers.add_parser(
|
|
'flags-to-json',
|
|
description='Convert yt-dlp command-line flags to a JSON config using go-ytdlp.',
|
|
formatter_class=argparse.RawTextHelpFormatter,
|
|
help='Convert yt-dlp flags to a JSON config.',
|
|
epilog="""
|
|
Examples:
|
|
|
|
# Convert flags from a string
|
|
ytops-client flags-to-json --from-string "-f best --no-playlist"
|
|
|
|
# Convert flags from a file (like cli.config)
|
|
ytops-client flags-to-json --from-file cli.config
|
|
|
|
# Convert flags passed directly as arguments
|
|
ytops-client flags-to-json -- --retries 5 --fragment-retries 5
|
|
|
|
# Combine sources (direct arguments override file/string)
|
|
ytops-client flags-to-json --from-file cli.config -- --retries 20
|
|
|
|
The go-ytdlp executable must be in your PATH.
|
|
You can install it by running the 'bin/install-goytdlp.sh' script.
|
|
"""
|
|
)
|
|
source_group = parser.add_mutually_exclusive_group()
|
|
source_group.add_argument('--from-file', type=argparse.FileType('r', encoding='utf-8'), help='Read flags from a file (e.g., a yt-dlp config file).')
|
|
source_group.add_argument('--from-string', help='Read flags from a single string.')
|
|
|
|
parser.add_argument('flags', nargs=argparse.REMAINDER, help='yt-dlp flags to convert. Use "--" to separate them from this script\'s own flags.')
|
|
parser.add_argument('--go-ytdlp-path', default='go-ytdlp', help='Path to the go-ytdlp executable. Defaults to "go-ytdlp" in PATH.')
|
|
parser.add_argument('--verbose', action='store_true', help='Enable verbose output for this script.')
|
|
return parser
|
|
|
|
def main_flags_to_json(args):
|
|
"""Main logic for the 'flags-to-json' command."""
|
|
if args.verbose:
|
|
# Reconfigure root logger for verbose output to stderr
|
|
for handler in logging.root.handlers[:]:
|
|
logging.root.removeHandler(handler)
|
|
logging.basicConfig(level=logging.DEBUG, format='%(name)s - %(levelname)s - %(message)s', stream=sys.stderr)
|
|
else:
|
|
# Default to INFO level, also to stderr
|
|
for handler in logging.root.handlers[:]:
|
|
logging.root.removeHandler(handler)
|
|
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s', stream=sys.stderr)
|
|
|
|
flags = []
|
|
if args.from_file:
|
|
logger.info(f"Reading flags from file: {args.from_file.name}")
|
|
content = args.from_file.read()
|
|
# A config file can have comments and one arg per line, or be a single line of args.
|
|
# shlex.split is good for single lines, but for multi-line we should split by line and filter.
|
|
lines = content.splitlines()
|
|
for line in lines:
|
|
line = line.strip()
|
|
if line and not line.startswith('#'):
|
|
# shlex.split can handle quoted arguments within the line
|
|
flags.extend(shlex.split(line))
|
|
elif args.from_string:
|
|
logger.info("Reading flags from string.")
|
|
flags.extend(shlex.split(args.from_string))
|
|
|
|
if args.flags:
|
|
# The 'flags' remainder might contain '--' which we should remove if it's the first element.
|
|
remainder_flags = args.flags
|
|
if remainder_flags and remainder_flags[0] == '--':
|
|
remainder_flags = remainder_flags[1:]
|
|
|
|
if remainder_flags:
|
|
logger.info("Appending flags from command-line arguments.")
|
|
flags.extend(remainder_flags)
|
|
|
|
if not flags:
|
|
logger.error("No flags provided to convert.")
|
|
return 1
|
|
|
|
try:
|
|
json_output = convert_flags_to_json(flags, args.go_ytdlp_path)
|
|
# Print to actual stdout for piping.
|
|
print(json.dumps(json_output, indent=2))
|
|
return 0
|
|
except (ValueError, FileNotFoundError, subprocess.CalledProcessError, json.JSONDecodeError, PermissionError):
|
|
# Specific error is already logged by the helper function.
|
|
return 1
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {e}", exc_info=args.verbose)
|
|
return 1
|
|
|
|
|
|
def convert_json_to_flags(json_input: str, go_ytdlp_path: str = None) -> str:
|
|
"""
|
|
Converts a JSON config string to yt-dlp command-line flags.
|
|
|
|
Args:
|
|
json_input: A string containing the JSON config.
|
|
go_ytdlp_path: Path to the go-ytdlp executable. If None, will try to find it.
|
|
|
|
Returns:
|
|
A string of command-line flags.
|
|
|
|
Raises:
|
|
ValueError: If the json_input is empty.
|
|
FileNotFoundError: If the go-ytdlp executable is not found.
|
|
subprocess.CalledProcessError: If go-ytdlp returns a non-zero exit code.
|
|
"""
|
|
if not json_input:
|
|
raise ValueError("No JSON input provided to convert.")
|
|
|
|
# Get the actual binary path
|
|
actual_path = get_go_ytdlp_path(go_ytdlp_path)
|
|
|
|
cmd = [actual_path, 'json-to-flags']
|
|
|
|
logger.debug(f"Executing command: {' '.join(shlex.quote(s) for s in cmd)}")
|
|
try:
|
|
process = subprocess.run(cmd, input=json_input, capture_output=True, check=True, encoding='utf-8')
|
|
|
|
if process.stderr:
|
|
logger.info(f"go-ytdlp output on stderr:\n{process.stderr.strip()}")
|
|
|
|
return process.stdout.strip()
|
|
except FileNotFoundError:
|
|
logger.error(f"Executable '{actual_path}' not found.")
|
|
logger.error("Please ensure go-ytdlp is installed and in your PATH.")
|
|
logger.error("You can run the 'bin/install-goytdlp.sh' script to install it.")
|
|
raise
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"go-ytdlp exited with error code {e.returncode}.")
|
|
logger.error(f"Stderr:\n{e.stderr.strip()}")
|
|
raise
|
|
except PermissionError:
|
|
logger.error(f"Permission denied executing '{actual_path}'.")
|
|
logger.error("Please ensure the file is executable (chmod +x).")
|
|
raise
|
|
|
|
|
|
def add_json_to_flags_parser(subparsers):
|
|
"""Add the parser for the 'json-to-flags' command."""
|
|
parser = subparsers.add_parser(
|
|
'json-to-flags',
|
|
description='Convert a JSON config to yt-dlp command-line flags using go-ytdlp.',
|
|
formatter_class=argparse.RawTextHelpFormatter,
|
|
help='Convert a JSON config to yt-dlp flags.',
|
|
epilog="""
|
|
Examples:
|
|
|
|
# Convert JSON from a string
|
|
ytops-client json-to-flags --from-string '{"postprocessor": {"ffmpeg": {"ppa": "SponsorBlock"}}}'
|
|
|
|
# Convert JSON from a file
|
|
ytops-client json-to-flags --from-file config.json
|
|
|
|
The go-ytdlp executable must be in your PATH.
|
|
You can install it by running the 'bin/install-goytdlp.sh' script.
|
|
"""
|
|
)
|
|
source_group = parser.add_mutually_exclusive_group(required=True)
|
|
source_group.add_argument('--from-file', type=argparse.FileType('r', encoding='utf-8'), help='Read JSON from a file.')
|
|
source_group.add_argument('--from-string', help='Read JSON from a single string.')
|
|
|
|
parser.add_argument('--go-ytdlp-path', default='go-ytdlp', help='Path to the go-ytdlp executable. Defaults to "go-ytdlp" in PATH.')
|
|
parser.add_argument('--verbose', action='store_true', help='Enable verbose output for this script.')
|
|
return parser
|
|
|
|
|
|
def main_json_to_flags(args):
|
|
"""Main logic for the 'json-to-flags' command."""
|
|
if args.verbose:
|
|
# Reconfigure root logger for verbose output to stderr
|
|
for handler in logging.root.handlers[:]:
|
|
logging.root.removeHandler(handler)
|
|
logging.basicConfig(level=logging.DEBUG, format='%(name)s - %(levelname)s - %(message)s', stream=sys.stderr)
|
|
else:
|
|
# Default to INFO level, also to stderr
|
|
for handler in logging.root.handlers[:]:
|
|
logging.root.removeHandler(handler)
|
|
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s', stream=sys.stderr)
|
|
|
|
json_input = ""
|
|
if args.from_file:
|
|
logger.info(f"Reading JSON from file: {args.from_file.name}")
|
|
json_input = args.from_file.read()
|
|
elif args.from_string:
|
|
logger.info("Reading JSON from string.")
|
|
json_input = args.from_string
|
|
|
|
try:
|
|
flags_output = convert_json_to_flags(json_input, args.go_ytdlp_path)
|
|
# Print to actual stdout for piping.
|
|
print(flags_output)
|
|
return 0
|
|
except (ValueError, FileNotFoundError, subprocess.CalledProcessError, PermissionError):
|
|
# Specific error is already logged by the helper function.
|
|
return 1
|
|
except Exception as e:
|
|
logger.error(f"An unexpected error occurred: {e}", exc_info=args.verbose)
|
|
return 1
|