119 lines
3.1 KiB
Python
119 lines
3.1 KiB
Python
import argparse
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from ..core.config import BASE_DIR, YT_DIR
|
|
|
|
COOKIES_PATH = Path(f"{BASE_DIR}/utils/cookies.txt")
|
|
|
|
|
|
GLOBAL_ARGS = [
|
|
"--socket-timeout", "20",
|
|
"--retries", "10",
|
|
"--fragment-retries", "10",
|
|
"--js-runtimes", "deno",
|
|
"--remote-components", "ejs:github",
|
|
"--embed-thumbnail",
|
|
"--convert-thumbnails", "jpg",
|
|
"--merge-output-format", "mp4",
|
|
"--sleep-interval", "5",
|
|
"--max-sleep-interval", "15",
|
|
"--sleep-requests", "3",
|
|
"--limit-rate", "3M",
|
|
]
|
|
|
|
DEFAULT_FORMAT = (
|
|
"bestvideo[height<=720][height>=480]+bestaudio/"
|
|
"bestvideo[height<=1080][height>=480]+bestaudio/"
|
|
"best[height>=480]"
|
|
)
|
|
|
|
def load_channels(config_path: Path, yt_dir: Path) -> list[dict]:
|
|
if not config_path.exists():
|
|
print(f"not found: {config_path}")
|
|
sys.exit(1)
|
|
|
|
data = json.loads(config_path.read_text())
|
|
channels = []
|
|
for raw in data:
|
|
name = raw["name"]
|
|
print(raw)
|
|
folder = raw.get("dir", name)
|
|
channels.append({
|
|
"name": name,
|
|
"link": raw["link"],
|
|
"path": yt_dir / folder,
|
|
"enabled_sync": raw.get("enabled_sync", True),
|
|
"extra_args": raw.get("extra_args", []),
|
|
"format": DEFAULT_FORMAT,
|
|
})
|
|
|
|
return channels
|
|
|
|
def build_cmd(ch: dict, dry_run: bool) -> list[str]:
|
|
channel_dir: Path = ch["path"]
|
|
archive = channel_dir / "archive.txt"
|
|
|
|
return [
|
|
"yt-dlp",
|
|
"-P", str(channel_dir),
|
|
"-o", "%(upload_date)s - %(title)s.%(ext)s",
|
|
"--download-archive", str(archive),
|
|
"-f", ch["format"],
|
|
"--cookies", str(COOKIES_PATH),
|
|
*GLOBAL_ARGS,
|
|
*ch["extra_args"],
|
|
ch["link"],
|
|
]
|
|
|
|
def sync_channel(ch: dict, dry_run: bool) -> int:
|
|
if not ch["enabled_sync"]:
|
|
print (f"skip disabled: {ch["name"]}")
|
|
return 0
|
|
|
|
ch["path"].mkdir(parents=True, exist_ok=True)
|
|
cmd = build_cmd(ch, dry_run)
|
|
|
|
print(f"\n=== {ch['name']} ===")
|
|
print(" ".join(cmd))
|
|
|
|
if dry_run:
|
|
return 0
|
|
return subprocess.run(cmd, check=False).returncode
|
|
|
|
def parse_args():
|
|
p = argparse.ArgumentParser(description="Sync channels via yt-dlp")
|
|
p.add_argument("--channels", "-c", type=Path, default=YT_DIR / "channels.json")
|
|
p.add_argument("--dry-run", action="store_true")
|
|
p.add_argument("ytdlp_args", nargs=argparse.REMAINDER,
|
|
help="extra yt-dlp args (after --)")
|
|
return p.parse_args()
|
|
|
|
def main():
|
|
args = parse_args()
|
|
channels = load_channels(args.channels, YT_DIR)
|
|
|
|
global_extra = args.ytdlp_args
|
|
if global_extra:
|
|
print(f"global extra args: {global_extra}")
|
|
|
|
enabled = sum(1 for c in channels if c["enabled_sync"])
|
|
print(f"channels: {len(channels)}, enabled: {enabled}")
|
|
|
|
failed = []
|
|
for ch in channels:
|
|
if global_extra:
|
|
ch["extra_args"] = global_extra + ch["extra_args"]
|
|
code = sync_channel(ch, args.dry_run)
|
|
if code != 0:
|
|
failed.append(ch["name"])
|
|
|
|
if failed:
|
|
print(f"\nFAILED: {failed}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |