import logging import os import re import requests from src.engines.core.base_downloader import EngineDownloader from src.engines.ffmpeg.ffmpeg_engine import FFMPEG from src.utilities.misc_helper import current_system_cpu, current_system_os logger = logging.getLogger() class FFMPEGDownloader(EngineDownloader): engine = FFMPEG # macOS FFMPEG mirror maintained by Evermeet - https://evermeet.cx/ffmpeg/ macos_url = "https://evermeet.cx/pub/ffmpeg/" # Linux FFMPEG mirror maintained by John van Sickle - https://johnvansickle.com/ffmpeg/ linux_url = "https://johnvansickle.com/ffmpeg/" # macOS FFMPEG mirror maintained by GyanD - https://www.gyan.dev/ffmpeg/builds/ windows_download_url = "https://github.com/GyanD/codexffmpeg/releases/download/" windows_api_url = "https://api.github.com/repos/GyanD/codexffmpeg/releases" # used to cache renderer versions in case they need to be accessed frequently version_cache = {} @classmethod def __get_macos_versions(cls, use_cache=True): # cache the versions locally version_cache = cls.version_cache.get('macos') if version_cache and use_cache: return version_cache response = requests.get(cls.macos_url, timeout=5) response.raise_for_status() link_pattern = r'>(.*\.zip)[^\.]' link_matches = re.findall(link_pattern, response.text) releases = [link.split('-')[-1].split('.zip')[0] for link in link_matches] cls.version_cache['macos'] = releases return releases @classmethod def __get_linux_versions(cls, use_cache=True): # cache the versions locally version_cache = cls.version_cache.get('linux') if version_cache and use_cache: return version_cache # Link 1 / 2 - Current Version response = requests.get(cls.linux_url, timeout=5) response.raise_for_status() current_release = re.findall(r'release: ([\w\.]+)', response.text)[0] # Link 2 / 2 - Previous Versions response = requests.get(os.path.join(cls.linux_url, 'old-releases'), timeout=5) response.raise_for_status() releases = list(set(re.findall(r'href="ffmpeg-([\w\.]+)-.*">ffmpeg', response.text))) releases.sort(reverse=True) releases.insert(0, current_release) # Add to cache cls.version_cache['linux'] = releases return releases @classmethod def __get_windows_versions(cls, use_cache=True): version_cache = cls.version_cache.get('windows') if version_cache and use_cache: return version_cache response = requests.get(cls.windows_api_url, timeout=5) response.raise_for_status() releases = [] all_git_releases = response.json() for item in all_git_releases: if re.match(r'^[0-9.]+$', item['tag_name']): releases.append(item['tag_name']) cls.version_cache['linux'] = releases return releases @classmethod def all_versions(cls, system_os=None, cpu=None): system_os = system_os or current_system_os() cpu = cpu or current_system_cpu() versions_per_os = {'linux': cls.__get_linux_versions, 'macos': cls.__get_macos_versions, 'windows': cls.__get_windows_versions} if not versions_per_os.get(system_os): logger.error(f"Cannot find version list for {system_os}") return results = [] all_versions = versions_per_os[system_os]() for version in all_versions: remote_url = cls.__get_remote_url_for_version(version=version, system_os=system_os, cpu=cpu) results.append({'cpu': cpu, 'file': os.path.basename(remote_url), 'system_os': system_os, 'url': remote_url, 'version': version}) return results @classmethod def __get_remote_url_for_version(cls, version, system_os, cpu): # Platform specific naming cleanup remote_url = None if system_os == 'macos': remote_url = os.path.join(cls.macos_url, f"ffmpeg-{version}.zip") elif system_os == 'linux': cpu = cpu.replace('x64', 'amd64') # change cpu to match repo naming convention latest_release = (version == cls.__get_linux_versions(use_cache=True)[0]) release_dir = 'releases' if latest_release else 'old-releases' release_filename = f'ffmpeg-release-{cpu}-static.tar.xz' if latest_release else \ f'ffmpeg-{version}-{cpu}-static.tar.xz' remote_url = os.path.join(cls.linux_url, release_dir, release_filename) elif system_os == 'windows': remote_url = f"{cls.windows_download_url.strip('/')}/{version}/ffmpeg-{version}-full_build.zip" else: logger.error("Unknown system os") return remote_url @classmethod def find_most_recent_version(cls, system_os=None, cpu=None, lts_only=False): try: system_os = system_os or current_system_os() cpu = cpu or current_system_cpu() return cls.all_versions(system_os, cpu)[0] except (IndexError, requests.exceptions.RequestException) as e: logger.error(f"Cannot get most recent version of ffmpeg: {e}") return {} @classmethod def version_is_available_to_download(cls, version, system_os=None, cpu=None): for ver in cls.all_versions(system_os, cpu): if ver['version'] == version: return ver return None @classmethod def download_engine(cls, version, download_location, system_os=None, cpu=None, timeout=120): system_os = system_os or current_system_os() cpu = cpu or current_system_cpu() # Verify requested version is available found_version = [item for item in cls.all_versions(system_os, cpu) if item['version'] == version] if not found_version: logger.error(f"Cannot find FFMPEG version {version} for {system_os} and {cpu}") return # Platform specific naming cleanup remote_url = cls.__get_remote_url_for_version(version=version, system_os=system_os, cpu=cpu) if system_os == 'macos': # override location to match linux download_location = os.path.join(download_location, f'ffmpeg-{version}-{system_os}-{cpu}') # Download and extract try: logger.info(f"Requesting download of ffmpeg-{version}-{system_os}-{cpu}") cls.download_and_extract_app(remote_url=remote_url, download_location=download_location, timeout=timeout) # naming cleanup to match existing naming convention output_path = os.path.join(download_location, f'ffmpeg-{version}-{system_os}-{cpu}') if system_os == 'linux': initial_cpu = cpu.replace('x64', 'amd64') # change cpu to match repo naming convention os.rename(os.path.join(download_location, f'ffmpeg-{version}-{initial_cpu}-static'), output_path) elif system_os == 'windows': os.rename(os.path.join(download_location, f'ffmpeg-{version}-full_build'), output_path) return output_path except (IndexError, FileNotFoundError) as e: logger.error(f"Cannot download requested engine: {e}") except OSError as e: logger.error(f"OS error while processing engine download: {e}") if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') # print(FFMPEGDownloader.download_engine('6.0', '/Users/brett/zordon-uploads/engines/')) # print(FFMPEGDownloader.find_most_recent_version(system_os='linux')) print(FFMPEGDownloader.download_engine(version='6.0', download_location='/Users/brett/zordon-uploads/engines/', system_os='linux', cpu='x64'))