import importlib import importlib.metadata import os import subprocess import sys from pathlib import Path from app.library.log import get_logger from .httpx_client import sync_client LOG = get_logger() def parse_version(v: str) -> tuple[int, ...]: return tuple(int("".join(filter(str.isdigit, part)) or "0") for part in v.strip().split(".")) class Packages: def __init__(self, env: str | None, file: str | Path | None, upgrade: bool = False): from_env: list[str] = env.split() if env else [] from_file = [] if file: file = Path(file) if file.exists() and os.access(str(file), os.R_OK): with open(file) as f: from_file: list[str] = [pkg.strip() for pkg in f if pkg.strip()] self.packages: list[str] = list(set(from_env + from_file)) self.upgrade = bool(upgrade) def has_packages(self) -> bool: return len(self.packages) > 0 def allow_upgrade(self) -> bool: return self.upgrade class PackageInstaller: def __init__(self, pkg_path: Path | None = None): self.user_site: Path | None = None "Where to install user packages." if pkg_path: self.user_site = pkg_path if not self.user_site and (base_dir := os.environ.get("YTP_CONFIG_PATH")): base_dir = Path(base_dir) self.user_site = base_dir / f"python{sys.version_info.major}.{sys.version_info.minor}-packages" self.user_site.mkdir(parents=True, exist_ok=True) if not self.user_site: LOG.error("No valid package path provided or found in environment. Aborting.") return if str(self.user_site) not in sys.path: sys.path.insert(0, str(self.user_site)) def action(self, pkg: str, upgrade: bool = False) -> None: version: str | None = None if "==" in pkg: pkg, version = pkg.split("==", 1) current_version: str | None = self._get_installed_version(pkg) if current_version and version and self.compare_versions(current_version, version): LOG.info( "Package '%s' is already installed at version '%s'; skipping installation.", pkg, version, extra={"package": pkg, "installed_version": current_version, "requested_version": version}, ) return if upgrade and current_version and not version: latest_version: str | None = self._get_latest_version(pkg) if latest_version and parse_version(current_version) >= parse_version(latest_version): LOG.info( "Package '%s' is already at the latest version '%s'; skipping upgrade.", pkg, current_version, extra={"package": pkg, "installed_version": current_version, "latest_version": latest_version}, ) return if current_version: LOG.info( "Package '%s' is installed at '%s'; proceeding with upgrade.", pkg, current_version, extra={"package": pkg, "installed_version": current_version}, ) else: LOG.info("Package '%s' is not installed; installing it.", pkg, extra={"package": pkg}) self._install_pkg(pkg, version=version) def _get_installed_version(self, pkg: str) -> str | None: try: return importlib.metadata.version(pkg) except importlib.metadata.PackageNotFoundError: return None def _get_latest_version(self, pkg: str) -> str | None: url: str = f"https://pypi.org/pypi/{pkg}/json" try: with sync_client(timeout=5.0) as client: resp = client.get(url) if 200 == resp.status_code: return resp.json()["info"]["version"] LOG.warning( "Failed to fetch PyPI metadata for '%s': HTTP %s.", pkg, resp.status_code, extra={"package": pkg, "status_code": resp.status_code, "url": url}, ) except Exception as e: LOG.warning( "Failed to query PyPI for '%s': %s", pkg, e, extra={"package": pkg, "url": url, "exception_type": type(e).__name__}, exc_info=True, ) return None def _install_pkg(self, pkg: str, version: str | None = None) -> bool: cmd: list[str] = [ sys.executable, "-m", "pip", "install", "--no-warn-script-location", "--upgrade", "--target", str(self.user_site), ] if "yt_dlp" == pkg: pkg = "yt-dlp" pkg_name: str = "yt-dlp[default]" if pkg == "yt-dlp" else pkg if version: if "nightly" == version and "yt-dlp" == pkg: cmd.extend(["--pre", pkg_name]) elif "master" == version and "yt-dlp" == pkg: cmd.append("git+https://github.com/yt-dlp/yt-dlp.git@master") else: cmd.append(version if str(version).startswith("git+") else f"{pkg_name}=={version}") else: cmd.extend(["--disable-pip-version-check", pkg_name]) try: proc: subprocess.CompletedProcess[bytes] = subprocess.run( cmd, check=True, capture_output=True, creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0, ) self.out(out=proc.stdout, err=proc.stderr) return 0 == proc.returncode except subprocess.CalledProcessError as e: LOG.exception( "Failed to install package '%s' (exit %s).", pkg, e.returncode, extra={"package": pkg, "returncode": e.returncode, "exception_type": type(e).__name__}, ) self.out(out=e.stdout, err=e.stderr) raise def check(self, pkgs: Packages): """ Checks for user supplied pip packages and installs them if they are not already installed. Args: pkgs (GetPackages): the class with packages and their settings. """ if not pkgs.has_packages() or not self.user_site: return LOG.info("Checking configured user pip packages.", extra={"packages": pkgs.packages}) for package in pkgs.packages: try: self.action(package, upgrade=pkgs.allow_upgrade()) except Exception as e: LOG.exception( "Failed to install or upgrade package '%s'.", package, extra={ "package": package, "allow_upgrade": pkgs.allow_upgrade(), "exception_type": type(e).__name__, }, ) def compare_versions(self, current: str, target: str) -> bool: """ Compare versions, handling yt-dlp format where pip uses 2025.7.21 but actual is 2025.07.21 Returns True if versions match, False otherwise """ if current == target: return True # Handle yt-dlp version format differences current_parts: list[str] = current.split(".") target_parts: list[str] = target.split(".") if len(current_parts) == len(target_parts): # Normalize parts by zero-padding single digits current_normalized: list[str] = [ part.zfill(2) if part.isdigit() and len(part) == 1 else part for part in current_parts ] target_normalized: list[str] = [ part.zfill(2) if part.isdigit() and len(part) == 1 else part for part in target_parts ] if current_normalized == target_normalized: return True return False def out(self, out: bytes | str | None = None, err: bytes | str | None = None) -> None: if out: for line in (line.strip() for line in out.strip().splitlines() if line.strip()): LOG.info(line.decode() if isinstance(line, bytes) else line) if err: for line in (line.strip() for line in err.strip().splitlines() if line.strip()): LOG.warning(line.decode() if isinstance(line, bytes) else line)