import yt_dlp
import random
import logging
import os # Added for cookie rotation
import random # Ensure random is imported
from collections import deque
from yt_dlp.utils import DownloadError
from yt_dlp.networking.impersonate import ImpersonateTarget # Added for impersonation
# Configure basic logging for this utility
log = logging.getLogger(__name__)
PROXY_FILE_PATH = 'proxies.txt' # Relative to project root
PROXY_FORMAT = 'socks5://{}:1080'
PROXY_SIGN_IN_WARNING = 'Sign in to confirm you’re not a bot'
PROXY_TIMEOUT_STRINGS = ['timed out. Retrying', 'Unable to download webpage: timed out']
class ProxyRotationError(Exception):
"""Custom exception for when all proxies fail."""
pass
class YtdlpProxyLogger:
"""Custom logger to detect proxy-related warnings."""
def __init__(self):
self.proxy_error_detected = False
def debug(self, msg):
# You can optionally log debug messages if needed
# log.debug(f"YTDLP_DEBUG: {msg}")
pass
def warning(self, msg):
log.warning(f"YTDLP_WARN: {msg}")
if PROXY_SIGN_IN_WARNING in msg or any(timeout_str in msg for timeout_str in PROXY_TIMEOUT_STRINGS):
self.proxy_error_detected = True
def error(self, msg):
log.error(f"YTDLP_ERROR: {msg}")
# Optionally, consider certain errors as proxy failures too
# if "some other error text" in msg:
# self.proxy_error_detected = True
class ProxyRotator:
"""Manages a list of proxies, rotates them, and handles banning/recycling."""
def __init__(self, proxy_file=PROXY_FILE_PATH, proxy_format=PROXY_FORMAT):
self.proxy_format = proxy_format
self.active_proxies = deque()
self.banned_proxies = set()
self._load_proxies(proxy_file)
log.info(f"ProxyRotator initialized with {len(self.active_proxies)} active proxies.")
def _load_proxies(self, proxy_file):
"""Loads proxies from the file."""
try:
with open(proxy_file, 'r') as f:
hostnames = [line.strip() for line in f if line.strip()]
if not hostnames:
log.warning(f"Proxy file '{proxy_file}' is empty or contains no valid hostnames.")
return
formatted_proxies = [self.proxy_format.format(host) for host in hostnames]
random.shuffle(formatted_proxies) # Shuffle initially
self.active_proxies.extend(formatted_proxies)
except FileNotFoundError:
log.error(f"Proxy file '{proxy_file}' not found. Proxy rotation disabled.")
except Exception as e:
log.error(f"Error loading proxy file '{proxy_file}': {e}")
def get_next_proxy(self):
"""Gets the next active proxy, recycling if necessary."""
if not self.active_proxies:
log.warning("Active proxy list is empty. Attempting to recycle banned proxies.")
self.recycle_proxies()
if not self.active_proxies:
log.error("No active proxies available, even after recycling.")
return None
# Rotate the deque: move the first element to the end
self.active_proxies.rotate(-1)
next_proxy = self.active_proxies[0]
log.debug(f"Next proxy: {next_proxy}")
return next_proxy
def ban_proxy(self, proxy):
"""Moves a proxy from the active list to the banned set."""
if proxy in self.active_proxies:
try:
# Deques don't have a direct remove, so convert, remove, convert back
temp_list = list(self.active_proxies)
temp_list.remove(proxy)
self.active_proxies = deque(temp_list)
self.banned_proxies.add(proxy)
log.warning(f"Banned proxy: {proxy}. Active: {len(self.active_proxies)}, Banned: {len(self.banned_proxies)}")
except ValueError:
# Should not happen if proxy was in active_proxies, but handle defensively
log.warning(f"Proxy {proxy} already removed from active list?")
elif proxy in self.banned_proxies:
log.debug(f"Proxy {proxy} was already banned.")
else:
log.warning(f"Attempted to ban unknown proxy: {proxy}")
def recycle_proxies(self):
"""Moves all banned proxies back to the active list and shuffles."""
if not self.banned_proxies:
log.info("No proxies in the banned list to recycle.")
return
log.info(f"Recycling {len(self.banned_proxies)} banned proxies.")
recycled_list = list(self.banned_proxies)
random.shuffle(recycled_list)
self.active_proxies.extend(recycled_list)
self.banned_proxies.clear()
log.info(f"Recycling complete. Active proxies: {len(self.active_proxies)}")
def get_active_proxy_count(self):
"""Returns the number of currently active proxies."""
return len(self.active_proxies)
# --- Impersonation Targets ---
# List of available targets (use underscores as required by yt-dlp option)
# List of simple browser identifiers expected by ImpersonateTarget.from_str()
IMPERSONATE_TARGET_STRINGS = [
'chrome', 'firefox', 'edge', 'safari'
# Add more simple identifiers if needed and supported
]
# --- Global Proxy Rotator Instance ---
# Instantiate it once when the module is loaded
proxy_rotator_instance = ProxyRotator()
# --- Cookie File Rotation ---
COOKIE_DIR_PATH = '/cookies' # Directory inside the container where cookie files are mounted
class CookieRotator:
"""Manages a list of cookie files found in a directory and rotates them."""
def __init__(self, cookie_dir=COOKIE_DIR_PATH):
self.cookie_dir = cookie_dir
self.cookie_files = deque()
self._load_cookie_files()
log.info(f"CookieRotator initialized with {len(self.cookie_files)} cookie files from '{self.cookie_dir}'.")
def _load_cookie_files(self):
"""Scans the directory for .txt cookie files."""
self.cookie_files.clear()
try:
if not os.path.isdir(self.cookie_dir):
log.warning(f"Cookie directory '{self.cookie_dir}' not found or not a directory. Cookie rotation disabled.")
return
found_files = [
os.path.join(self.cookie_dir, f)
for f in os.listdir(self.cookie_dir)
if os.path.isfile(os.path.join(self.cookie_dir, f)) and f.lower().endswith('.txt')
]
if not found_files:
log.warning(f"No .txt cookie files found in '{self.cookie_dir}'. Cookie rotation disabled.")
return
random.shuffle(found_files) # Shuffle initially
self.cookie_files.extend(found_files)
log.info(f"Loaded {len(self.cookie_files)} cookie files: {', '.join(os.path.basename(f) for f in self.cookie_files)}")
except Exception as e:
log.error(f"Error loading cookie files from '{self.cookie_dir}': {e}")
def get_next_cookie_file(self):
"""Gets the next cookie file path, reloading if the list is empty."""
if not self.cookie_files:
log.warning("Cookie file list is empty. Attempting to reload.")
self._load_cookie_files() # Reload the list if empty
if not self.cookie_files:
log.error("No cookie files available, even after reload.")
return None
# Rotate the deque: move the first element to the end
self.cookie_files.rotate(-1)
next_cookie_file = self.cookie_files[0]
log.debug(f"Next cookie file: {os.path.basename(next_cookie_file)}")
return next_cookie_file
def get_cookie_file_count(self):
"""Returns the number of currently loaded cookie files."""
return len(self.cookie_files)
# --- Global Cookie Rotator Instance ---
# Instantiate it once when the module is loaded
cookie_rotator_instance = CookieRotator()
# --- End Cookie File Rotation ---
# --- Wrapper Function ---
def execute_ytdl_with_retry(base_options, url, action='extract_info'):
"""
Executes a yt-dlp action (extract_info or download) with optional proxy rotation,
cookie rotation.
Proxy rotation is OFF by default, enable with ENABLE_PROXY_ROTATION=true env var.
Args:
base_options (dict): The base yt-dlp options (without proxy or logger).
url (str or list): The URL(s) to process.
action (str): 'extract_info' or 'download'.
Returns:
The result of the yt-dlp action.
Raises:
ProxyRotationError: If all proxies fail during proxy rotation.
DownloadError: If a non-proxy-related yt-dlp error occurs.
ValueError: If an invalid action is specified.
"""
global proxy_rotator_instance, cookie_rotator_instance # Use global instances
# --- Prepare Base Options for this execution ---
# Start with a copy of the provided base options
options = base_options.copy()
# --- Add Impersonation Target for this call ---
try:
selected_target_str = random.choice(IMPERSONATE_TARGET_STRINGS)
impersonate_target_obj = ImpersonateTarget.from_str(selected_target_str)
options['impersonate'] = impersonate_target_obj
log.debug(f"Using impersonate target for this call: {selected_target_str}")
except NameError: # Handle case where IMPERSONATE_TARGET_STRINGS might not be defined
log.error("IMPERSONATE_TARGET_STRINGS not defined. Impersonation disabled.")
except ValueError:
log.error(f"Failed to create ImpersonateTarget from string: {selected_target_str}. Impersonation disabled for this call.")
# Optionally remove the key if it might exist from base_options
options.pop('impersonate', None)
except ImportError:
log.error("ImpersonateTarget could not be imported. Is 'yt-dlp[curl-cffi]' installed correctly? Impersonation disabled.")
options.pop('impersonate', None)
# --- Proxy Enable Switch ---
# Check environment variable to enable proxy rotation; default is disabled.
enable_proxy = os.environ.get('ENABLE_PROXY_ROTATION', 'false').lower() == 'true'
# --- Proxy Rotation Path (Only if enabled and proxies exist) ---
if enable_proxy and proxy_rotator_instance and proxy_rotator_instance.get_active_proxy_count() > 0:
log.info("ENABLE_PROXY_ROTATION is true and proxies are available. Proceeding with proxy rotation.")
max_attempts = proxy_rotator_instance.get_active_proxy_count() + len(proxy_rotator_instance.banned_proxies)
if max_attempts == 0: # Should not happen due to check above, but safety first
raise ProxyRotationError("No proxies configured or loaded despite enable flag.")
log.info(f"Attempting yt-dlp action '{action}' for '{url}' with proxy rotation (max attempts: {max_attempts})")
last_error = None
tried_proxies = set()
while True:
current_proxy = proxy_rotator_instance.get_next_proxy()
if current_proxy is None:
log.error(f"Failed to get any proxy during rotation loop. Aborting.")
break
if current_proxy in tried_proxies:
log.warning(f"Already tried proxy {current_proxy} in this cycle. All active proxies failed.")
break
tried_proxies.add(current_proxy)
log.info(f"Attempting with proxy {current_proxy} ({len(tried_proxies)}/{max_attempts} unique proxies tried this cycle)")
# Add proxy for this attempt
options['proxy'] = current_proxy
# Add Cookie Rotation for this attempt
if cookie_rotator_instance and cookie_rotator_instance.get_cookie_file_count() > 0:
selected_cookie_file = cookie_rotator_instance.get_next_cookie_file()
if selected_cookie_file:
options['cookies'] = selected_cookie_file
log.debug(f"Using Cookie File: {os.path.basename(selected_cookie_file)}")
else:
log.warning("Failed to get a cookie file from rotator during proxy loop.")
# Remove cookies key if no file was selected for this attempt (prevents carrying over old value)
elif 'cookies' in options:
options.pop('cookies')
log.debug("No cookie file selected for this attempt.")
logger = YtdlpProxyLogger()
options['logger'] = logger
options['quiet'] = True
options['socket_timeout'] = 10
log.debug(f"Final options for yt-dlp (attempt with proxy): {options}")
logger.proxy_error_detected = False
try:
with yt_dlp.YoutubeDL(options) as ydl:
if action == 'extract_info':
url_to_extract = url[0] if isinstance(url, list) else url
result = ydl.extract_info(url_to_extract, download=False)
if result:
log.debug(f"[Proxy Path] Extracted Info - ID: {result.get('id')}, Webpage URL: {result.get('webpage_url')}")
elif action == 'download':
result = ydl.download(url)
else:
raise ValueError(f"Invalid action specified: {action}")
# Check logger *after* potential error but before returning success
if logger.proxy_error_detected:
log.warning(f"Proxy error detected with {current_proxy} even on success? Banning.")
proxy_rotator_instance.ban_proxy(current_proxy)
last_error = ProxyRotationError(f"Proxy error detected post-action with {current_proxy}")
options.pop('proxy', None) # Remove failed proxy before continuing
if 'cookies' in options: options.pop('cookies') # Remove cookies too for next attempt
continue # Try next proxy
log.info(f"Action '{action}' succeeded with proxy {current_proxy}")
return result # Success!
except DownloadError as e:
last_error = e
log.warning(f"DownloadError occurred with proxy {current_proxy}: {e}")
error_str = str(e)
# Check logger flag OR error message content for sign-in OR timeout strings
is_proxy_related_error = (
logger.proxy_error_detected or
PROXY_SIGN_IN_WARNING in error_str or
any(timeout_str in error_str for timeout_str in PROXY_TIMEOUT_STRINGS)
)
if is_proxy_related_error:
log.warning(f"Proxy failure confirmed for {current_proxy}. Banning and retrying.")
proxy_rotator_instance.ban_proxy(current_proxy)
options.pop('proxy', None) # Remove failed proxy before continuing
if 'cookies' in options: options.pop('cookies') # Remove cookies too for next attempt
continue # Try next proxy
else:
log.error(f"Non-proxy related DownloadError with {current_proxy}. Aborting rotation.")
raise e # Re-raise error if it wasn't the specific proxy warning
except Exception as e:
# Catch other potential errors during instantiation or download
last_error = e
log.error(f"Unexpected error with proxy {current_proxy}: {e}. Banning and retrying.")
proxy_rotator_instance.ban_proxy(current_proxy)
options.pop('proxy', None) # Remove failed proxy before continuing
if 'cookies' in options: options.pop('cookies') # Remove cookies too for next attempt
continue # Try next proxy
# If loop finishes without success (meaning all active proxies failed in this cycle)
log.error(f"All available proxy attempts failed for action '{action}' on '{url}'. Last error: {last_error}")
raise ProxyRotationError("Proxy error occurred. Please try again.") from last_error
# --- Direct Execution Path (Default or if proxy enabled but none available) ---
else:
if not enable_proxy:
log.debug("Proxy rotation is disabled (default). Running directly.")
else: # This means enable_proxy was true, but no proxies were found/loaded
log.warning("ENABLE_PROXY_ROTATION is true, but no proxies available. Running directly.")
# Add Cookie Rotation for direct execution
if cookie_rotator_instance and cookie_rotator_instance.get_cookie_file_count() > 0:
selected_cookie_file = cookie_rotator_instance.get_next_cookie_file()
if selected_cookie_file:
options['cookies'] = selected_cookie_file # Modify the global 'options'
log.debug(f"[Direct] Using Cookie File: {os.path.basename(selected_cookie_file)}")
else:
log.warning("[Direct] Failed to get a cookie file from rotator.")
else:
log.debug("[Direct] Cookie rotation disabled or no cookie files found.")
# Execute directly using the prepared 'options'
try:
# Set logger/quiet options just before execution
logger = YtdlpProxyLogger() # Still useful for potential non-proxy warnings
options['logger'] = logger
options['quiet'] = True
# No socket timeout needed here usually, rely on base_options or yt-dlp defaults
log.debug(f"[Direct] Final options for yt-dlp: {options}")
with yt_dlp.YoutubeDL(options) as ydl:
if action == 'extract_info':
url_to_extract = url[0] if isinstance(url, list) else url
result = ydl.extract_info(url_to_extract, download=False)
if result:
log.debug(f"[Direct Path] Extracted Info - ID: {result.get('id')}, Webpage URL: {result.get('webpage_url')}")
return result
elif action == 'download':
return ydl.download(url)
else:
raise ValueError(f"Invalid action specified: {action}")
except DownloadError as e:
log.error(f"yt-dlp failed during direct execution (no proxy): {e}")
raise # Re-raise the original error
except Exception as e:
log.exception(f"Unexpected error during direct execution: {e}")
raise
Mythic Mobs Mod (1.21) | New Unique Mythical Creatures
Mythic Mobs is a mod that adds mythical creatures to Minecraft. These mobs can be found in all sorts of biomes, from the forests to the deserts and they will give you a new challenge to conquer.
Creatures:
Automaton: Step into villages protected by Automatons, intricate and sentient beings crafted with precision and purpose. These guardians stand even mightier than Iron Golems, so be careful to not punch one of those annoying villagers. When they take damage they can be restored to their former glory using Bronze Ingots.
Chupacabra: The Chupacabra, a creature of the night with luminous red eyes and sharp fangs. This legendary being lurks in the darkness, targeting livestock as it emerges from obscure corners and remote locations.
Drake: Discover the Drake, a small yet loyal creature that can be tamed. Devotion to its master courses through its veins and ready to give its life if necessary. Exercise caution, however, as Drakes and livestock, especially chickens, do not harmonize well. Can be tamed using Chupacabra Skewers.
Kobold: Encounter the Kobolds, passive entities that refrain from harming players as long as you don’t start it. They offer the possibility of trade.
Kobold Warrior: The Kobold Warriors are protectors of their kin. Armed with swift legs and keen-edged spears, they will attack those who disturb their people. Defeat a Kobold Warrior, and you stand a chance to get the Kobold Spear.
🎬 Want to download videos & audio for free? 🎧
OneForAll Downloader makes it easy — no ads, no paywalls, no redirects!
✅ Free, safe, and secure
✅ Supports 1,800+ platforms including YouTube, TikTok, Instagram & more
Just paste the URL, pick your quality, and download! 👉 Try OneForAll Downloader Now!
