fix downloader
This commit is contained in:
@@ -442,24 +442,30 @@ async def download_media(
|
|||||||
original_stderr = sys.stderr
|
original_stderr = sys.stderr
|
||||||
error_filter = None
|
error_filter = None
|
||||||
|
|
||||||
|
logger.info(f"Starting yt-dlp download for URL: {url}")
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# Redirect stderr to filter non-critical errors
|
# Redirect stderr to filter non-critical errors
|
||||||
error_filter = YtDlpErrorFilter(original_stderr)
|
error_filter = YtDlpErrorFilter(original_stderr)
|
||||||
sys.stderr = error_filter
|
sys.stderr = error_filter
|
||||||
|
|
||||||
|
logger.info(f"Extracting info for URL: {url}")
|
||||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||||
# Check for cancellation before start
|
# Check for cancellation before start
|
||||||
if cancel_event and cancel_event.is_set():
|
if cancel_event and cancel_event.is_set():
|
||||||
raise KeyboardInterrupt("Download cancelled")
|
raise KeyboardInterrupt("Download cancelled")
|
||||||
|
|
||||||
# Get video information
|
# Get video information
|
||||||
|
logger.info(f"Extracting video info for: {url}")
|
||||||
info = ydl.extract_info(url, download=False)
|
info = ydl.extract_info(url, download=False)
|
||||||
|
logger.info(f"Video info extracted: title={info.get('title', 'N/A')[:50]}, duration={info.get('duration', 'N/A')}")
|
||||||
|
|
||||||
# Check for cancellation after getting info
|
# Check for cancellation after getting info
|
||||||
if cancel_event and cancel_event.is_set():
|
if cancel_event and cancel_event.is_set():
|
||||||
raise KeyboardInterrupt("Download cancelled")
|
raise KeyboardInterrupt("Download cancelled")
|
||||||
|
|
||||||
# Download (progress hook will be called from this thread)
|
# Download (progress hook will be called from this thread)
|
||||||
|
logger.info(f"Starting download for: {url}")
|
||||||
# Note: Some postprocessors may show errors (like FixupM3u8 with aspect ratio),
|
# Note: Some postprocessors may show errors (like FixupM3u8 with aspect ratio),
|
||||||
# but the video file is still downloaded correctly
|
# but the video file is still downloaded correctly
|
||||||
try:
|
try:
|
||||||
|
|||||||
@@ -726,8 +726,10 @@ async def url_handler(client: Client, message: Message):
|
|||||||
return
|
return
|
||||||
|
|
||||||
# Add to queue (with duplicate URL check) AFTER saving to database
|
# Add to queue (with duplicate URL check) AFTER saving to database
|
||||||
|
logger.info(f"Adding task {task_id} to queue (URL: {url[:50]}...)")
|
||||||
success = await task_queue.add_task(task, check_duplicate_url=True)
|
success = await task_queue.add_task(task, check_duplicate_url=True)
|
||||||
if not success:
|
if not success:
|
||||||
|
logger.warning(f"Failed to add task {task_id} to queue (duplicate or error)")
|
||||||
# If failed to add to queue, remove from database
|
# If failed to add to queue, remove from database
|
||||||
try:
|
try:
|
||||||
async with get_async_session_local()() as session:
|
async with get_async_session_local()() as session:
|
||||||
@@ -745,7 +747,11 @@ async def url_handler(client: Client, message: Message):
|
|||||||
|
|
||||||
# Start executor if not already started
|
# Start executor if not already started
|
||||||
if not task_executor._running:
|
if not task_executor._running:
|
||||||
|
logger.info(f"Starting task executor for task {task_id}")
|
||||||
await task_executor.start()
|
await task_executor.start()
|
||||||
|
logger.info(f"Task executor started, workers should begin processing tasks")
|
||||||
|
else:
|
||||||
|
logger.debug(f"Task executor already running, task {task_id} will be processed by existing workers")
|
||||||
|
|
||||||
# Send initial status message and save message_id for updates
|
# Send initial status message and save message_id for updates
|
||||||
from bot.modules.task_scheduler.executor import set_task_message
|
from bot.modules.task_scheduler.executor import set_task_message
|
||||||
|
|||||||
@@ -196,6 +196,8 @@ class TaskExecutor:
|
|||||||
await asyncio.sleep(0.5)
|
await asyncio.sleep(0.5)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
|
logger.info(f"Worker {name} got task {task.id} from queue (URL: {task.url[:50] if task.url else 'N/A'}...)")
|
||||||
|
|
||||||
# Check for cancellation before starting processing
|
# Check for cancellation before starting processing
|
||||||
current_task = await task_queue.get_task_by_id(task.id)
|
current_task = await task_queue.get_task_by_id(task.id)
|
||||||
if current_task and current_task.status == TaskStatus.CANCELLED:
|
if current_task and current_task.status == TaskStatus.CANCELLED:
|
||||||
@@ -204,13 +206,14 @@ class TaskExecutor:
|
|||||||
|
|
||||||
# Update status
|
# Update status
|
||||||
await task_queue.update_task_status(task.id, TaskStatus.PROCESSING)
|
await task_queue.update_task_status(task.id, TaskStatus.PROCESSING)
|
||||||
|
logger.info(f"Worker {name} processing task {task.id} (status: PROCESSING)")
|
||||||
logger.info(f"Worker {name} processing task {task.id}")
|
|
||||||
|
|
||||||
# Execute task (doesn't block other workers and message processing)
|
# Execute task (doesn't block other workers and message processing)
|
||||||
# Each task executes independently
|
# Each task executes independently
|
||||||
await self._execute_task(task)
|
await self._execute_task(task)
|
||||||
|
|
||||||
|
logger.info(f"Worker {name} completed task {task.id}")
|
||||||
|
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
logger.info(f"Worker {name} stopped")
|
logger.info(f"Worker {name} stopped")
|
||||||
break
|
break
|
||||||
@@ -281,6 +284,8 @@ class TaskExecutor:
|
|||||||
|
|
||||||
async def _download_with_ytdlp(self, task: Task):
|
async def _download_with_ytdlp(self, task: Task):
|
||||||
"""Download via yt-dlp"""
|
"""Download via yt-dlp"""
|
||||||
|
logger.info(f"Starting download for task {task.id}: {task.url}")
|
||||||
|
|
||||||
# Get cancellation event for this task
|
# Get cancellation event for this task
|
||||||
cancel_event = task_queue.get_cancel_event(task.id)
|
cancel_event = task_queue.get_cancel_event(task.id)
|
||||||
|
|
||||||
@@ -296,7 +301,9 @@ class TaskExecutor:
|
|||||||
from bot.modules.media_loader.ytdlp import get_media_info
|
from bot.modules.media_loader.ytdlp import get_media_info
|
||||||
from shared.config import settings
|
from shared.config import settings
|
||||||
|
|
||||||
|
logger.info(f"Getting media info for task {task.id}...")
|
||||||
media_info = await get_media_info(task.url, cookies_file=settings.COOKIES_FILE)
|
media_info = await get_media_info(task.url, cookies_file=settings.COOKIES_FILE)
|
||||||
|
logger.info(f"Media info retrieved for task {task.id}: title={media_info.get('title') if media_info else 'N/A'}")
|
||||||
if media_info:
|
if media_info:
|
||||||
# Check duration
|
# Check duration
|
||||||
max_duration = settings.max_duration_minutes_int
|
max_duration = settings.max_duration_minutes_int
|
||||||
|
|||||||
Reference in New Issue
Block a user