removed python version
This commit is contained in:
		
							parent
							
								
									dce5a3c301
								
							
						
					
					
						commit
						9d13a08e89
					
				
							
								
								
									
										133
									
								
								dl.py
									
									
									
									
									
								
							
							
						
						
									
										133
									
								
								dl.py
									
									
									
									
									
								
							| 
						 | 
					@ -1,133 +0,0 @@
 | 
				
			||||||
import os, sys, json, subprocess, time;
 | 
					 | 
				
			||||||
from types import *;
 | 
					 | 
				
			||||||
from typing import *;
 | 
					 | 
				
			||||||
from dataclasses import dataclass
 | 
					 | 
				
			||||||
from argparse_dataclass import dataclass as argparser
 | 
					 | 
				
			||||||
from enum import IntEnum
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
@dataclass
 | 
					 | 
				
			||||||
class Process:
 | 
					 | 
				
			||||||
    proc: subprocess.Popen[bytes]
 | 
					 | 
				
			||||||
    name: str
 | 
					 | 
				
			||||||
    url: str
 | 
					 | 
				
			||||||
    file: str
 | 
					 | 
				
			||||||
    genre: str
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
PROCESSES: List[Process] =[]
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
class Level(IntEnum):
 | 
					 | 
				
			||||||
    ERROR = 0
 | 
					 | 
				
			||||||
    WARN = 1
 | 
					 | 
				
			||||||
    INFO = 2
 | 
					 | 
				
			||||||
    DEBUG = 3
 | 
					 | 
				
			||||||
        
 | 
					 | 
				
			||||||
def level_to_prefix(l: Level) -> str:
 | 
					 | 
				
			||||||
    match l:
 | 
					 | 
				
			||||||
        case Level.ERROR:
 | 
					 | 
				
			||||||
            return "\033[0;31m\033[1merror\033[0m"
 | 
					 | 
				
			||||||
        case Level.WARN:
 | 
					 | 
				
			||||||
            return "\033[1;33m\033[1mwarn\033[0m"
 | 
					 | 
				
			||||||
        case Level.INFO:
 | 
					 | 
				
			||||||
            return "\033[1;32m\033[1minfo\033[0m"
 | 
					 | 
				
			||||||
        case Level.DEBUG:
 | 
					 | 
				
			||||||
            return "\033[1;34m\033[1mdebug\033[0m"
 | 
					 | 
				
			||||||
            
 | 
					 | 
				
			||||||
        
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
class Logger:
 | 
					 | 
				
			||||||
    level: Level = Level.INFO
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
    def error(self, s: str):
 | 
					 | 
				
			||||||
        if self.level >= Level.ERROR:
 | 
					 | 
				
			||||||
            print(f"{level_to_prefix(Level.ERROR)}: {s}")
 | 
					 | 
				
			||||||
    def warn(self, s: str):
 | 
					 | 
				
			||||||
        if self.level >= Level.WARN:
 | 
					 | 
				
			||||||
            print(f"{level_to_prefix(Level.WARN)}: {s}")
 | 
					 | 
				
			||||||
    def info(self, s: str):
 | 
					 | 
				
			||||||
        if self.level >= Level.INFO:
 | 
					 | 
				
			||||||
            print(f"{level_to_prefix(Level.INFO)}: {s}")
 | 
					 | 
				
			||||||
    def debug(self, s: str):
 | 
					 | 
				
			||||||
        if self.level >= Level.DEBUG:
 | 
					 | 
				
			||||||
            print(f"{level_to_prefix(Level.DEBUG)}: {s}")
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
LOGGER = Logger()
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
if os.name == 'nt':
 | 
					 | 
				
			||||||
    YTDLP="yt-dlp.exe"
 | 
					 | 
				
			||||||
else:
 | 
					 | 
				
			||||||
    YTDLP="yt-dlp"
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
@dataclass
 | 
					 | 
				
			||||||
class ManifestSong:
 | 
					 | 
				
			||||||
    name: str
 | 
					 | 
				
			||||||
    url: str
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
type ManifestGenre = List[ManifestSong]
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
type Manifest = Dict[str, ManifestGenre]
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
@argparser
 | 
					 | 
				
			||||||
class CliArgs:
 | 
					 | 
				
			||||||
    manifest: str = "./manifest.json"
 | 
					 | 
				
			||||||
    output_dir: str = "./"
 | 
					 | 
				
			||||||
    debug: bool = False
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
def main(cliargs: CliArgs) -> int:
 | 
					 | 
				
			||||||
    if cliargs.debug:
 | 
					 | 
				
			||||||
        LOGGER.level = Level.DEBUG
 | 
					 | 
				
			||||||
    manifest: Manifest = get_manifest(cliargs)
 | 
					 | 
				
			||||||
    count = 0
 | 
					 | 
				
			||||||
    for (genre, v) in manifest.items():
 | 
					 | 
				
			||||||
        for song in v:
 | 
					 | 
				
			||||||
            if len(PROCESSES) > 10:
 | 
					 | 
				
			||||||
                wait_for_procs(10)
 | 
					 | 
				
			||||||
            if download_song(genre, song["name"], song["url"]):
 | 
					 | 
				
			||||||
                count += 1
 | 
					 | 
				
			||||||
    wait_for_procs()
 | 
					 | 
				
			||||||
    LOGGER.info(f"Downloaded {count} new songs")
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
def download_song(dir: str, name: str, url: str) -> bool:
 | 
					 | 
				
			||||||
    if not os.path.isdir(dir):
 | 
					 | 
				
			||||||
        os.mkdir(dir)
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
    outfile=f"{dir}/{name}.m4a"
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
    if os.path.isfile(outfile):
 | 
					 | 
				
			||||||
        LOGGER.debug(f"Already downloaded {outfile} ({url}), skipping")
 | 
					 | 
				
			||||||
        return False
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
    LOGGER.info(f"Downloading {outfile} ({url})")
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
    proc = subprocess.Popen(
 | 
					 | 
				
			||||||
        [YTDLP, "-x", "--audio-format", "m4a","-o", outfile, url],
 | 
					 | 
				
			||||||
        stdout=open(os.devnull, 'wb')
 | 
					 | 
				
			||||||
    )
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
    PROCESSES.append(Process(proc, name, url, outfile, dir))
 | 
					 | 
				
			||||||
    return True
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
def wait_for_procs(until: int = 0):
 | 
					 | 
				
			||||||
    while len(PROCESSES) > until:
 | 
					 | 
				
			||||||
        for proc in PROCESSES:
 | 
					 | 
				
			||||||
            retcode = proc.proc.poll()
 | 
					 | 
				
			||||||
            if retcode is not None: # Process finished.
 | 
					 | 
				
			||||||
                PROCESSES.remove(proc)
 | 
					 | 
				
			||||||
                LOGGER.info(f"Finished downloading {proc.file}")
 | 
					 | 
				
			||||||
                break
 | 
					 | 
				
			||||||
            else: # No process is done, wait a bit and check again.
 | 
					 | 
				
			||||||
                time.sleep(.1)
 | 
					 | 
				
			||||||
                continue
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
def get_manifest(cliargs: CliArgs) -> Manifest:
 | 
					 | 
				
			||||||
    with open(cliargs.manifest, "r") as f:
 | 
					 | 
				
			||||||
        data = f.read()
 | 
					 | 
				
			||||||
        return json.loads(data)
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
 | 
					 | 
				
			||||||
if __name__ == "__main__":
 | 
					 | 
				
			||||||
    ret = main(CliArgs.parse_args(sys.argv[1:]))
 | 
					 | 
				
			||||||
    sys.exit(ret)
 | 
					 | 
				
			||||||
| 
						 | 
					@ -1 +0,0 @@
 | 
				
			||||||
argparse_dataclass == 2.0.0
 | 
					 | 
				
			||||||
		Loading…
	
		Reference in New Issue
	
	Block a user