Fixed issue where file extenions not found and auth timeout
This commit is contained in:
@@ -0,0 +1,839 @@
|
|||||||
|
# Scheduled Downloads Implementation Plan
|
||||||
|
|
||||||
|
## Overview
|
||||||
|
This document outlines the implementation plan for adding scheduled download functionality to the BDFR Web Interface. Users will be able to configure downloads that run automatically on a daily schedule, perfect for keeping up with new content from their favorite subreddits or users.
|
||||||
|
|
||||||
|
## Requirements Summary
|
||||||
|
- **Database**: SQLite with SQLAlchemy ORM
|
||||||
|
- **Scheduling**: Daily frequency (runs every 24 hours)
|
||||||
|
- **UI Approach**: Simple - checkbox in Advanced Options + management section on main page
|
||||||
|
- **Time Filter**: Automatically set to "last day" for daily runs
|
||||||
|
- **Duplicate Handling**: Works with existing no-dupes functionality
|
||||||
|
- **Deployment**: Docker container environment
|
||||||
|
- **Execution Model**: **Sequential only - one task at a time, queued execution**
|
||||||
|
|
||||||
|
## Architecture
|
||||||
|
|
||||||
|
### 1. Database Schema
|
||||||
|
|
||||||
|
#### ScheduledTask Table
|
||||||
|
```python
|
||||||
|
class ScheduledTask:
|
||||||
|
id: UUID (Primary Key)
|
||||||
|
name: str # User-friendly name for the task
|
||||||
|
enabled: bool # Whether task is active
|
||||||
|
|
||||||
|
# Download Configuration
|
||||||
|
source_type: str # "subreddit" or "user"
|
||||||
|
source_name: str # Name of subreddit or username
|
||||||
|
download_mode: str # "download", "archive", or "clone"
|
||||||
|
|
||||||
|
# Filter Options
|
||||||
|
limit: int
|
||||||
|
sort: str # "hot", "top", "new", etc.
|
||||||
|
time_filter: str # Always "day" for daily tasks
|
||||||
|
min_score: int (optional)
|
||||||
|
no_dupes: bool # Always true for scheduled tasks
|
||||||
|
simple_check: bool
|
||||||
|
|
||||||
|
# Scheduling
|
||||||
|
schedule_frequency: str # "daily" (extensible for future: "weekly", "custom")
|
||||||
|
run_time: time # Time of day to run (e.g., "02:00:00")
|
||||||
|
timezone: str # User's timezone (default: UTC)
|
||||||
|
|
||||||
|
# Metadata
|
||||||
|
created_at: datetime
|
||||||
|
updated_at: datetime
|
||||||
|
last_run_at: datetime (nullable)
|
||||||
|
next_run_at: datetime
|
||||||
|
|
||||||
|
# Authentication
|
||||||
|
auth_state: str (nullable) # For authenticated downloads
|
||||||
|
```
|
||||||
|
|
||||||
|
#### TaskExecutionHistory Table
|
||||||
|
```python
|
||||||
|
class TaskExecutionHistory:
|
||||||
|
id: UUID (Primary Key)
|
||||||
|
task_id: UUID (Foreign Key -> ScheduledTask)
|
||||||
|
|
||||||
|
# Execution Details
|
||||||
|
started_at: datetime
|
||||||
|
completed_at: datetime (nullable)
|
||||||
|
status: str # "success", "failed", "running", "queued"
|
||||||
|
|
||||||
|
# Results
|
||||||
|
items_found: int
|
||||||
|
items_downloaded: int
|
||||||
|
error_message: str (nullable)
|
||||||
|
|
||||||
|
# Link to download
|
||||||
|
download_id: str # Links to active_downloads tracking
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2. Backend Components
|
||||||
|
|
||||||
|
#### File Structure
|
||||||
|
```
|
||||||
|
web_interface/
|
||||||
|
├── app/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ ├── main.py (existing)
|
||||||
|
│ ├── auth.py (existing)
|
||||||
|
│ ├── database.py (NEW - SQLAlchemy setup)
|
||||||
|
│ ├── models.py (NEW - DB models)
|
||||||
|
│ ├── scheduler.py (NEW - APScheduler + Queue integration)
|
||||||
|
│ ├── task_queue.py (NEW - Sequential task queue manager)
|
||||||
|
│ └── scheduled_tasks.py (NEW - Task management logic)
|
||||||
|
├── data/
|
||||||
|
│ └── scheduled_tasks.db (SQLite database - created at runtime)
|
||||||
|
└── requirements.txt (UPDATE - add dependencies)
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Dependencies to Add
|
||||||
|
```txt
|
||||||
|
sqlalchemy>=2.0.0
|
||||||
|
alembic>=1.12.0 # For database migrations
|
||||||
|
apscheduler>=3.10.0 # For task scheduling
|
||||||
|
```
|
||||||
|
|
||||||
|
#### API Endpoints
|
||||||
|
|
||||||
|
**Scheduled Tasks CRUD:**
|
||||||
|
- `POST /api/scheduled-tasks` - Create new scheduled task
|
||||||
|
- `GET /api/scheduled-tasks` - List all scheduled tasks
|
||||||
|
- `GET /api/scheduled-tasks/{task_id}` - Get specific task details
|
||||||
|
- `PUT /api/scheduled-tasks/{task_id}` - Update task configuration
|
||||||
|
- `DELETE /api/scheduled-tasks/{task_id}` - Delete task
|
||||||
|
- `POST /api/scheduled-tasks/{task_id}/toggle` - Enable/disable task
|
||||||
|
- `POST /api/scheduled-tasks/{task_id}/run-now` - Trigger immediate execution (adds to queue)
|
||||||
|
|
||||||
|
**Task History & Queue:**
|
||||||
|
- `GET /api/scheduled-tasks/{task_id}/history` - Get execution history
|
||||||
|
- `GET /api/scheduled-tasks/history/recent` - Get recent executions across all tasks
|
||||||
|
- `GET /api/scheduled-tasks/queue` - Get current task queue status
|
||||||
|
|
||||||
|
### 3. Sequential Task Queue System
|
||||||
|
|
||||||
|
#### Task Queue Manager (`task_queue.py`)
|
||||||
|
|
||||||
|
**Core Concept**: Only one scheduled download can run at a time. When multiple tasks are triggered (either by schedule or "Run Now"), they are queued and executed sequentially.
|
||||||
|
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
from typing import Optional, List, Dict
|
||||||
|
from datetime import datetime
|
||||||
|
import logging
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
class TaskQueue:
|
||||||
|
"""
|
||||||
|
Manages sequential execution of scheduled download tasks.
|
||||||
|
Ensures only one task runs at a time.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self.queue: asyncio.Queue = asyncio.Queue()
|
||||||
|
self.current_task: Optional[str] = None # Current task_id being executed
|
||||||
|
self.is_processing: bool = False
|
||||||
|
self.worker_task: Optional[asyncio.Task] = None
|
||||||
|
|
||||||
|
async def add_task(self, task_id: str, priority: int = 0):
|
||||||
|
"""
|
||||||
|
Add a task to the queue.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
task_id: UUID of the scheduled task
|
||||||
|
priority: 0 = scheduled (normal), 1 = manual "Run Now" (higher priority)
|
||||||
|
"""
|
||||||
|
await self.queue.put({
|
||||||
|
'task_id': task_id,
|
||||||
|
'priority': priority,
|
||||||
|
'queued_at': datetime.now()
|
||||||
|
})
|
||||||
|
logger.info(f"Task {task_id} added to queue (priority={priority}, queue_size={self.queue.qsize()})")
|
||||||
|
|
||||||
|
# Start worker if not already running
|
||||||
|
if not self.is_processing:
|
||||||
|
await self.start_worker()
|
||||||
|
|
||||||
|
async def start_worker(self):
|
||||||
|
"""Start the queue worker if not already running"""
|
||||||
|
if self.worker_task is None or self.worker_task.done():
|
||||||
|
self.worker_task = asyncio.create_task(self._process_queue())
|
||||||
|
logger.info("Queue worker started")
|
||||||
|
|
||||||
|
async def _process_queue(self):
|
||||||
|
"""Process tasks from queue sequentially"""
|
||||||
|
self.is_processing = True
|
||||||
|
logger.info("Queue worker processing started")
|
||||||
|
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
# Wait for next task (with timeout to allow graceful shutdown)
|
||||||
|
try:
|
||||||
|
task_info = await asyncio.wait_for(
|
||||||
|
self.queue.get(),
|
||||||
|
timeout=60.0
|
||||||
|
)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
# Check if queue is empty
|
||||||
|
if self.queue.empty():
|
||||||
|
logger.info("Queue empty, worker stopping")
|
||||||
|
break
|
||||||
|
continue
|
||||||
|
|
||||||
|
task_id = task_info['task_id']
|
||||||
|
self.current_task = task_id
|
||||||
|
|
||||||
|
logger.info(f"Executing task {task_id} from queue (queue_size={self.queue.qsize()})")
|
||||||
|
|
||||||
|
# Execute the task (this will block until download completes)
|
||||||
|
try:
|
||||||
|
await execute_scheduled_task(task_id)
|
||||||
|
logger.info(f"Task {task_id} completed successfully")
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Task {task_id} failed: {e}")
|
||||||
|
finally:
|
||||||
|
self.current_task = None
|
||||||
|
self.queue.task_done()
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Queue worker error: {e}")
|
||||||
|
|
||||||
|
self.is_processing = False
|
||||||
|
logger.info("Queue worker stopped")
|
||||||
|
|
||||||
|
def get_queue_status(self) -> Dict:
|
||||||
|
"""Get current queue status"""
|
||||||
|
return {
|
||||||
|
'current_task': self.current_task,
|
||||||
|
'queue_size': self.queue.qsize(),
|
||||||
|
'is_processing': self.is_processing
|
||||||
|
}
|
||||||
|
|
||||||
|
async def stop(self):
|
||||||
|
"""Stop the queue worker gracefully"""
|
||||||
|
logger.info("Stopping queue worker...")
|
||||||
|
if self.worker_task and not self.worker_task.done():
|
||||||
|
# Wait for current task to complete
|
||||||
|
await self.worker_task
|
||||||
|
logger.info("Queue worker stopped")
|
||||||
|
|
||||||
|
# Global queue instance
|
||||||
|
task_queue = TaskQueue()
|
||||||
|
```
|
||||||
|
|
||||||
|
### 4. Scheduler Service (Docker-Aware with Sequential Queue)
|
||||||
|
|
||||||
|
#### APScheduler Configuration
|
||||||
|
```python
|
||||||
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
||||||
|
from apscheduler.triggers.cron import CronTrigger
|
||||||
|
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
|
||||||
|
import pytz
|
||||||
|
|
||||||
|
# Use database job store for persistence across container restarts
|
||||||
|
jobstores = {
|
||||||
|
'default': SQLAlchemyJobStore(url='sqlite:///data/scheduler_jobs.db')
|
||||||
|
}
|
||||||
|
|
||||||
|
# Configure scheduler for Docker container
|
||||||
|
scheduler = AsyncIOScheduler(
|
||||||
|
jobstores=jobstores,
|
||||||
|
timezone=pytz.UTC, # Container runs in UTC
|
||||||
|
job_defaults={
|
||||||
|
'coalesce': True, # Combine multiple missed executions into one
|
||||||
|
'max_instances': 1, # Only one instance of each job at a time
|
||||||
|
'misfire_grace_time': 3600 # Allow up to 1 hour late execution
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
# Add job for each enabled task
|
||||||
|
def schedule_task(task: ScheduledTask):
|
||||||
|
"""
|
||||||
|
Schedule a task to be added to the queue at specified time.
|
||||||
|
Note: This doesn't execute the task directly, it queues it.
|
||||||
|
"""
|
||||||
|
user_tz = pytz.timezone(task.timezone)
|
||||||
|
hour, minute = task.run_time.hour, task.run_time.minute
|
||||||
|
|
||||||
|
scheduler.add_job(
|
||||||
|
func=queue_scheduled_task, # Add to queue, not execute directly
|
||||||
|
trigger=CronTrigger(hour=hour, minute=minute, timezone=user_tz),
|
||||||
|
args=[task.id],
|
||||||
|
id=str(task.id),
|
||||||
|
replace_existing=True
|
||||||
|
)
|
||||||
|
logger.info(f"Scheduled task {task.id} for {hour:02d}:{minute:02d} {task.timezone}")
|
||||||
|
|
||||||
|
async def queue_scheduled_task(task_id: str):
|
||||||
|
"""
|
||||||
|
Called by scheduler at the configured time.
|
||||||
|
Adds task to queue rather than executing immediately.
|
||||||
|
"""
|
||||||
|
logger.info(f"Scheduler triggered for task {task_id}, adding to queue")
|
||||||
|
await task_queue.add_task(task_id, priority=0) # Normal priority for scheduled tasks
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Task Execution Flow with Queue
|
||||||
|
```mermaid
|
||||||
|
graph TD
|
||||||
|
A[Scheduler triggers at scheduled time] --> B[Add task to queue]
|
||||||
|
B --> C{Is queue worker running?}
|
||||||
|
C -->|No| D[Start queue worker]
|
||||||
|
C -->|Yes| E[Task waits in queue]
|
||||||
|
D --> F[Worker picks next task from queue]
|
||||||
|
E --> F
|
||||||
|
F --> G[Load task from DB]
|
||||||
|
G --> H[Check if enabled]
|
||||||
|
H -->|Disabled| I[Skip, mark in history]
|
||||||
|
H -->|Enabled| J[Create execution history]
|
||||||
|
J --> K[Set status = 'running']
|
||||||
|
K --> L[Build download parameters]
|
||||||
|
L --> M[Set time_filter=day, no_dupes=true]
|
||||||
|
M --> N[Call BDFR API - BLOCKS until complete]
|
||||||
|
N --> O[Wait for download to finish]
|
||||||
|
O --> P[Update execution history]
|
||||||
|
P --> Q[Update last_run_at]
|
||||||
|
Q --> R[Worker picks next task]
|
||||||
|
R -->|Queue empty| S[Worker idles/stops]
|
||||||
|
R -->|More tasks| F
|
||||||
|
|
||||||
|
style N fill:#ffcccc
|
||||||
|
style O fill:#ffcccc
|
||||||
|
note1[Note: Worker blocks here until download completes]
|
||||||
|
```
|
||||||
|
|
||||||
|
### 5. Integration with BDFR API (Sequential Execution)
|
||||||
|
|
||||||
|
```python
|
||||||
|
async def execute_scheduled_task(task_id: str):
|
||||||
|
"""
|
||||||
|
Execute a scheduled download task.
|
||||||
|
This function BLOCKS until the download is complete,
|
||||||
|
ensuring sequential execution.
|
||||||
|
"""
|
||||||
|
# Load task from database
|
||||||
|
task = get_scheduled_task(task_id)
|
||||||
|
|
||||||
|
if not task.enabled:
|
||||||
|
logger.info(f"Skipping disabled task {task_id}")
|
||||||
|
# Still record in history that it was skipped
|
||||||
|
execution = create_execution_history(task_id)
|
||||||
|
execution.status = 'skipped'
|
||||||
|
execution.completed_at = datetime.now(pytz.UTC)
|
||||||
|
save_execution_history(execution)
|
||||||
|
return
|
||||||
|
|
||||||
|
# Create execution history record
|
||||||
|
execution = create_execution_history(task_id)
|
||||||
|
|
||||||
|
try:
|
||||||
|
logger.info(f"Executing scheduled task {task_id}: {task.source_type}/{task.source_name}")
|
||||||
|
|
||||||
|
# Build download parameters
|
||||||
|
kwargs = {
|
||||||
|
'limit': task.limit,
|
||||||
|
'sort': task.sort,
|
||||||
|
'time_filter': 'day', # Always "day" for daily scheduled tasks
|
||||||
|
'no_dupes': True, # Always enabled for scheduled tasks
|
||||||
|
'simple_check': task.simple_check,
|
||||||
|
'auth_state': task.auth_state
|
||||||
|
}
|
||||||
|
|
||||||
|
# Create download using existing API
|
||||||
|
# This returns immediately with a download_id
|
||||||
|
download_id = await create_download_with_bdfr_api(
|
||||||
|
download_type=task.source_type,
|
||||||
|
name=task.source_name,
|
||||||
|
**kwargs
|
||||||
|
)
|
||||||
|
|
||||||
|
logger.info(f"Task {task_id} started as download {download_id}")
|
||||||
|
|
||||||
|
# Track download
|
||||||
|
execution.download_id = download_id
|
||||||
|
execution.status = 'running'
|
||||||
|
save_execution_history(execution)
|
||||||
|
|
||||||
|
# **CRITICAL: Wait for download to complete before returning**
|
||||||
|
# This ensures the next queued task doesn't start until this one finishes
|
||||||
|
await wait_for_download_completion(download_id)
|
||||||
|
|
||||||
|
# Check final status
|
||||||
|
download_status = bdfr_manager.get_download_status(download_id)
|
||||||
|
|
||||||
|
if download_status and download_status['status'] == 'completed':
|
||||||
|
execution.status = 'success'
|
||||||
|
execution.items_found = download_status.get('items_found', 0)
|
||||||
|
execution.items_downloaded = download_status.get('items_processed', 0)
|
||||||
|
logger.info(f"Task {task_id} completed successfully")
|
||||||
|
else:
|
||||||
|
execution.status = 'failed'
|
||||||
|
execution.error_message = download_status.get('error', 'Unknown error')
|
||||||
|
logger.error(f"Task {task_id} failed: {execution.error_message}")
|
||||||
|
|
||||||
|
execution.completed_at = datetime.now(pytz.UTC)
|
||||||
|
save_execution_history(execution)
|
||||||
|
|
||||||
|
# Update task timestamps
|
||||||
|
task.last_run_at = datetime.now(pytz.UTC)
|
||||||
|
task.next_run_at = calculate_next_run(task)
|
||||||
|
save_scheduled_task(task)
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Task {task_id} execution error: {e}")
|
||||||
|
execution.status = 'failed'
|
||||||
|
execution.error_message = str(e)
|
||||||
|
execution.completed_at = datetime.now(pytz.UTC)
|
||||||
|
save_execution_history(execution)
|
||||||
|
raise
|
||||||
|
|
||||||
|
async def wait_for_download_completion(download_id: str, timeout: int = 3600):
|
||||||
|
"""
|
||||||
|
Wait for a download to complete.
|
||||||
|
Polls the download status until it's no longer running.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
download_id: The download to wait for
|
||||||
|
timeout: Maximum seconds to wait (default 1 hour)
|
||||||
|
"""
|
||||||
|
start_time = datetime.now()
|
||||||
|
check_interval = 5 # Check every 5 seconds
|
||||||
|
|
||||||
|
while True:
|
||||||
|
# Check if timeout exceeded
|
||||||
|
elapsed = (datetime.now() - start_time).total_seconds()
|
||||||
|
if elapsed > timeout:
|
||||||
|
logger.error(f"Download {download_id} timed out after {timeout}s")
|
||||||
|
raise TimeoutError(f"Download exceeded timeout of {timeout}s")
|
||||||
|
|
||||||
|
# Check download status
|
||||||
|
status = bdfr_manager.get_download_status(download_id)
|
||||||
|
|
||||||
|
if not status:
|
||||||
|
logger.warning(f"Download {download_id} status not found")
|
||||||
|
break
|
||||||
|
|
||||||
|
download_status = status.get('status', 'unknown')
|
||||||
|
|
||||||
|
# Check if download is finished (completed, failed, or cancelled)
|
||||||
|
if download_status in ['completed', 'failed', 'cancelled']:
|
||||||
|
logger.info(f"Download {download_id} finished with status: {download_status}")
|
||||||
|
break
|
||||||
|
|
||||||
|
# Still running, wait before checking again
|
||||||
|
await asyncio.sleep(check_interval)
|
||||||
|
```
|
||||||
|
|
||||||
|
### 6. API Endpoints with Queue Support
|
||||||
|
|
||||||
|
```python
|
||||||
|
@app.post("/api/scheduled-tasks/{task_id}/run-now")
|
||||||
|
async def run_task_now(task_id: str):
|
||||||
|
"""
|
||||||
|
Manually trigger a scheduled task to run now.
|
||||||
|
Adds it to the queue with high priority.
|
||||||
|
"""
|
||||||
|
task = get_scheduled_task(task_id)
|
||||||
|
if not task:
|
||||||
|
raise HTTPException(status_code=404, detail="Task not found")
|
||||||
|
|
||||||
|
# Add to queue with priority (goes ahead of scheduled tasks)
|
||||||
|
await task_queue.add_task(task_id, priority=1)
|
||||||
|
|
||||||
|
queue_status = task_queue.get_queue_status()
|
||||||
|
|
||||||
|
return {
|
||||||
|
"message": f"Task {task_id} added to queue",
|
||||||
|
"queue_position": queue_status['queue_size'],
|
||||||
|
"currently_running": queue_status['current_task'],
|
||||||
|
"status": "queued" if queue_status['current_task'] else "starting"
|
||||||
|
}
|
||||||
|
|
||||||
|
@app.get("/api/scheduled-tasks/queue")
|
||||||
|
async def get_queue_status():
|
||||||
|
"""Get current task queue status"""
|
||||||
|
status = task_queue.get_queue_status()
|
||||||
|
|
||||||
|
# Get details of current task if any
|
||||||
|
current_task_info = None
|
||||||
|
if status['current_task']:
|
||||||
|
task = get_scheduled_task(status['current_task'])
|
||||||
|
if task:
|
||||||
|
current_task_info = {
|
||||||
|
'id': task.id,
|
||||||
|
'name': task.name,
|
||||||
|
'source': f"{task.source_type}/{task.source_name}"
|
||||||
|
}
|
||||||
|
|
||||||
|
return {
|
||||||
|
'queue_size': status['queue_size'],
|
||||||
|
'is_processing': status['is_processing'],
|
||||||
|
'current_task': current_task_info
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 7. Frontend Implementation
|
||||||
|
|
||||||
|
#### UI Modifications to [`index.html`](web_interface/templates/index.html:1)
|
||||||
|
|
||||||
|
**Queue Status Indicator (add to header):**
|
||||||
|
```html
|
||||||
|
<div id="queueStatus" class="queue-status" style="display: none;">
|
||||||
|
<span class="queue-icon">⏳</span>
|
||||||
|
<span id="queueText">Processing scheduled tasks...</span>
|
||||||
|
</div>
|
||||||
|
```
|
||||||
|
|
||||||
|
**Advanced Options Section - Add Checkbox:**
|
||||||
|
```html
|
||||||
|
<!-- After existing checkboxes in Advanced Options -->
|
||||||
|
<label class="checkbox-label" data-tooltip="Run this download automatically every day at a specific time">
|
||||||
|
<input type="checkbox" id="runDaily" name="run_daily">
|
||||||
|
<span class="checkmark"></span>
|
||||||
|
Schedule Daily Run
|
||||||
|
</label>
|
||||||
|
|
||||||
|
<!-- Conditionally shown when checkbox is checked -->
|
||||||
|
<div id="scheduleOptions" style="display: none;">
|
||||||
|
<div class="form-group">
|
||||||
|
<label for="scheduleName">Task Name:</label>
|
||||||
|
<input type="text" id="scheduleName" name="schedule_name"
|
||||||
|
placeholder="e.g., Daily r/Python downloads">
|
||||||
|
<small class="form-help">Give this scheduled task a descriptive name</small>
|
||||||
|
</div>
|
||||||
|
<div class="form-group">
|
||||||
|
<label for="scheduleTime">Run Time:</label>
|
||||||
|
<input type="time" id="scheduleTime" name="schedule_time" value="02:00">
|
||||||
|
<small class="form-help">Time when task will run daily (your local time). Tasks run one at a time in order.</small>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
```
|
||||||
|
|
||||||
|
**New Section - Scheduled Tasks List:**
|
||||||
|
```html
|
||||||
|
<!-- After Progress Section -->
|
||||||
|
<section class="scheduled-section">
|
||||||
|
<div class="section-header">
|
||||||
|
<h2>📅 Scheduled Downloads</h2>
|
||||||
|
<div id="queueStatusBadge" class="queue-badge" style="display: none;">
|
||||||
|
<span class="badge-icon">⏳</span>
|
||||||
|
<span id="queueBadgeText">Queue: 0</span>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<div id="scheduledTasksList" class="scheduled-tasks-list">
|
||||||
|
<!-- Empty state -->
|
||||||
|
<div class="no-tasks" id="noTasksMessage">
|
||||||
|
<div class="empty-state">
|
||||||
|
<div class="empty-icon">📅</div>
|
||||||
|
<p>No scheduled tasks yet</p>
|
||||||
|
<p>Check "Schedule Daily Run" when creating a download to set up automated daily downloads.</p>
|
||||||
|
<p><strong>Note:</strong> Scheduled tasks run one at a time to prevent server overload.</p>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<!-- Task cards will be inserted here -->
|
||||||
|
<div id="scheduledTasksItems"></div>
|
||||||
|
</div>
|
||||||
|
</section>
|
||||||
|
```
|
||||||
|
|
||||||
|
**Task Card Template:**
|
||||||
|
```html
|
||||||
|
<div class="task-card" id="task-{task_id}" data-task-id="{task_id}">
|
||||||
|
<div class="task-header">
|
||||||
|
<div class="task-info">
|
||||||
|
<h4>{task_name}</h4>
|
||||||
|
<div class="task-meta">
|
||||||
|
<span class="task-source">{source_type}: {source_name}</span>
|
||||||
|
<span class="task-schedule">⏰ Runs daily at {run_time}</span>
|
||||||
|
<span class="task-queue-info" style="display: none;">
|
||||||
|
⏳ Queued / Currently Running
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
<div class="task-controls">
|
||||||
|
<button class="btn-toggle" onclick="toggleTask('{task_id}')">
|
||||||
|
{enabled ? "✓ Enabled" : "○ Disabled"}
|
||||||
|
</button>
|
||||||
|
<button class="btn-run-now" onclick="runTaskNow('{task_id}')">
|
||||||
|
▶ Run Now
|
||||||
|
</button>
|
||||||
|
<button class="btn-delete" onclick="deleteTask('{task_id}')">
|
||||||
|
🗑 Delete
|
||||||
|
</button>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
<div class="task-details">
|
||||||
|
<div class="task-stat">
|
||||||
|
<span class="stat-label">Last Run:</span>
|
||||||
|
<span class="stat-value">{last_run_at || "Never"}</span>
|
||||||
|
</div>
|
||||||
|
<div class="task-stat">
|
||||||
|
<span class="stat-label">Next Run:</span>
|
||||||
|
<span class="stat-value">{next_run_at}</span>
|
||||||
|
</div>
|
||||||
|
<div class="task-stat">
|
||||||
|
<span class="stat-label">Mode:</span>
|
||||||
|
<span class="stat-value">{download_mode}</span>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
</div>
|
||||||
|
```
|
||||||
|
|
||||||
|
#### JavaScript Modifications to [`app.js`](web_interface/static/js/app.js:1)
|
||||||
|
|
||||||
|
**Add queue status polling:**
|
||||||
|
```javascript
|
||||||
|
async loadScheduledTasks() {
|
||||||
|
const response = await fetch('/api/scheduled-tasks');
|
||||||
|
const tasks = await response.json();
|
||||||
|
this.renderScheduledTasks(tasks);
|
||||||
|
|
||||||
|
// Also update queue status
|
||||||
|
this.updateQueueStatus();
|
||||||
|
}
|
||||||
|
|
||||||
|
async updateQueueStatus() {
|
||||||
|
try {
|
||||||
|
const response = await fetch('/api/scheduled-tasks/queue');
|
||||||
|
const status = await response.json();
|
||||||
|
|
||||||
|
// Update queue badge
|
||||||
|
const queueBadge = document.getElementById('queueStatusBadge');
|
||||||
|
const queueText = document.getElementById('queueBadgeText');
|
||||||
|
|
||||||
|
if (status.queue_size > 0 || status.current_task) {
|
||||||
|
queueBadge.style.display = 'flex';
|
||||||
|
queueText.textContent = `Queue: ${status.queue_size}${status.current_task ? ' (1 running)' : ''}`;
|
||||||
|
} else {
|
||||||
|
queueBadge.style.display = 'none';
|
||||||
|
}
|
||||||
|
|
||||||
|
// Highlight currently running task card
|
||||||
|
document.querySelectorAll('.task-card').forEach(card => {
|
||||||
|
const taskId = card.dataset.taskId;
|
||||||
|
const queueInfo = card.querySelector('.task-queue-info');
|
||||||
|
|
||||||
|
if (status.current_task && status.current_task.id === taskId) {
|
||||||
|
card.classList.add('task-running');
|
||||||
|
queueInfo.textContent = '⚡ Currently Running';
|
||||||
|
queueInfo.style.display = 'inline';
|
||||||
|
} else {
|
||||||
|
card.classList.remove('task-running');
|
||||||
|
queueInfo.style.display = 'none';
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
} catch (error) {
|
||||||
|
console.error('Failed to update queue status:', error);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async runTaskNow(taskId) {
|
||||||
|
const response = await fetch(`/api/scheduled-tasks/${taskId}/run-now`, {
|
||||||
|
method: 'POST'
|
||||||
|
});
|
||||||
|
|
||||||
|
if (response.ok) {
|
||||||
|
const result = await response.json();
|
||||||
|
|
||||||
|
if (result.queue_position > 0) {
|
||||||
|
this.showSuccess(`Task added to queue. Position: ${result.queue_position}`);
|
||||||
|
} else {
|
||||||
|
this.showSuccess('Task execution starting...');
|
||||||
|
}
|
||||||
|
|
||||||
|
this.updateQueueStatus();
|
||||||
|
} else {
|
||||||
|
this.showError('Failed to queue task');
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Poll queue status every 10 seconds
|
||||||
|
startQueuePolling() {
|
||||||
|
setInterval(() => {
|
||||||
|
if (document.querySelectorAll('.task-card').length > 0) {
|
||||||
|
this.updateQueueStatus();
|
||||||
|
}
|
||||||
|
}, 10000);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 8. Implementation Steps
|
||||||
|
|
||||||
|
1. **Backend Foundation** (Steps 4-5)
|
||||||
|
- Add SQLAlchemy and APScheduler to [`requirements.txt`](web_interface/requirements.txt:1)
|
||||||
|
- Create [`database.py`](web_interface/app/database.py) with Docker-aware paths
|
||||||
|
- Create [`models.py`](web_interface/app/models.py) with ScheduledTask and TaskExecutionHistory models
|
||||||
|
- Initialize database on app startup
|
||||||
|
|
||||||
|
2. **Task Queue System** (Step 7)
|
||||||
|
- Create [`task_queue.py`](web_interface/app/task_queue.py) with sequential queue manager
|
||||||
|
- Implement queue worker with blocking execution
|
||||||
|
- Add queue status tracking and reporting
|
||||||
|
|
||||||
|
3. **Scheduler Service** (Step 7 continued)
|
||||||
|
- Create [`scheduler.py`](web_interface/app/scheduler.py) with Docker-aware APScheduler
|
||||||
|
- Integrate with task queue (scheduler adds to queue, doesn't execute directly)
|
||||||
|
- Implement `wait_for_download_completion()` to block until download finishes
|
||||||
|
- Add scheduler lifecycle hooks to [`main.py`](web_interface/app/main.py:1)
|
||||||
|
|
||||||
|
4. **API Endpoints** (Step 6)
|
||||||
|
- Create [`scheduled_tasks.py`](web_interface/app/scheduled_tasks.py) with CRUD operations
|
||||||
|
- Add routes to [`main.py`](web_interface/app/main.py:1)
|
||||||
|
- Implement task toggle, delete, and run-now (with queue)
|
||||||
|
- Add queue status endpoint
|
||||||
|
|
||||||
|
5. **Frontend - Form** (Step 8)
|
||||||
|
- Add "Run Daily" checkbox to Advanced Options in [`index.html`](web_interface/templates/index.html:1)
|
||||||
|
- Add conditional schedule configuration fields
|
||||||
|
- Update form submission logic in [`app.js`](web_interface/static/js/app.js:1)
|
||||||
|
- Auto-detect browser timezone
|
||||||
|
|
||||||
|
6. **Frontend - Management** (Step 9)
|
||||||
|
- Add Scheduled Tasks section to [`index.html`](web_interface/templates/index.html:1)
|
||||||
|
- Add queue status indicator
|
||||||
|
- Implement task card rendering with queue status
|
||||||
|
- Add toggle, delete, and run-now functions to [`app.js`](web_interface/static/js/app.js:1)
|
||||||
|
- Add queue status polling
|
||||||
|
|
||||||
|
7. **Integration** (Steps 10-11)
|
||||||
|
- Connect scheduler to BDFR API via [`create_download_with_bdfr_api()`](web_interface/app/main.py:373)
|
||||||
|
- Implement automatic `time_filter="day"` for scheduled tasks
|
||||||
|
- Add execution history tracking
|
||||||
|
- Ensure sequential execution with proper blocking
|
||||||
|
|
||||||
|
8. **Testing** (Step 14)
|
||||||
|
- Test task creation, editing, deletion
|
||||||
|
- Test queue functionality (multiple tasks, sequential execution)
|
||||||
|
- Test "Run Now" adds to queue correctly
|
||||||
|
- Test priority (manual tasks run before scheduled)
|
||||||
|
- Test container restart persistence
|
||||||
|
- Verify only one task runs at a time
|
||||||
|
|
||||||
|
## Key Features
|
||||||
|
|
||||||
|
### Sequential Execution (NEW)
|
||||||
|
- **Task Queue**: All scheduled downloads go through a FIFO queue
|
||||||
|
- **Blocking Execution**: Each task blocks until its download completes
|
||||||
|
- **No Concurrency**: Only one download runs at a time, preventing system overload
|
||||||
|
- **Priority System**: Manual "Run Now" tasks get priority over scheduled tasks
|
||||||
|
- **Queue Status**: Users can see queue size and currently running task
|
||||||
|
|
||||||
|
### Docker-Specific Features
|
||||||
|
- **Persistent Storage**: SQLite database and downloads persist via volume mounts
|
||||||
|
- **Container Restarts**: APScheduler with job store survives restarts, queue rebuilds on startup
|
||||||
|
- **Timezone Handling**: User timezone stored, converted to UTC for container execution
|
||||||
|
- **Logging**: Structured logging for container environment
|
||||||
|
- **Health Checks**: Scheduler and queue status included in health endpoint
|
||||||
|
|
||||||
|
### Automatic Configuration for Scheduled Tasks
|
||||||
|
- **time_filter**: Always set to "day" - ensures only last 24 hours of content
|
||||||
|
- **no_dupes**: Always enabled - prevents re-downloading same content
|
||||||
|
- **Timezone handling**: Store user timezone, convert to UTC for execution, display in user timezone
|
||||||
|
- **Sequential execution**: Guaranteed one-at-a-time processing
|
||||||
|
|
||||||
|
### Smart Duplicate Prevention
|
||||||
|
When a scheduled task runs:
|
||||||
|
1. BDFR checks existing hashes (if no_dupes enabled)
|
||||||
|
2. Only downloads new content from last 24 hours
|
||||||
|
3. Skips content already downloaded in previous runs
|
||||||
|
|
||||||
|
### Execution Tracking
|
||||||
|
- Every run creates a history record
|
||||||
|
- Tracks success/failure status
|
||||||
|
- Records items found vs. items downloaded
|
||||||
|
- Links to the actual download progress for real-time monitoring
|
||||||
|
- Shows queue position and current task status
|
||||||
|
|
||||||
|
### User Experience
|
||||||
|
- Simple checkbox to schedule any download
|
||||||
|
- Visual indication of enabled/disabled tasks
|
||||||
|
- Queue status badge shows pending tasks
|
||||||
|
- Currently running task highlighted
|
||||||
|
- Next run time displayed in user's local timezone
|
||||||
|
- One-click to add task to queue immediately
|
||||||
|
- Easy enable/disable without deleting task
|
||||||
|
- Queue position shown when manually running tasks
|
||||||
|
|
||||||
|
## Sequential Execution Examples
|
||||||
|
|
||||||
|
### Scenario 1: Multiple Scheduled Tasks
|
||||||
|
```
|
||||||
|
02:00 AM - Task A triggers, added to queue
|
||||||
|
02:00 AM - Task B triggers, added to queue
|
||||||
|
02:00 AM - Task C triggers, added to queue
|
||||||
|
|
||||||
|
Execution Order:
|
||||||
|
1. Task A starts, downloads 100 posts (takes 15 minutes)
|
||||||
|
2. Task B starts at 02:15 AM, downloads 50 posts (takes 8 minutes)
|
||||||
|
3. Task C starts at 02:23 AM, downloads 75 posts (takes 12 minutes)
|
||||||
|
4. All complete by 02:35 AM
|
||||||
|
```
|
||||||
|
|
||||||
|
### Scenario 2: Manual "Run Now" During Scheduled Task
|
||||||
|
```
|
||||||
|
02:00 AM - Task A starts (scheduled, downloading...)
|
||||||
|
02:10 AM - User clicks "Run Now" on Task B
|
||||||
|
02:10 AM - Task B added to queue with priority
|
||||||
|
|
||||||
|
Execution Order:
|
||||||
|
1. Task A continues running (started first)
|
||||||
|
2. Task B waits in queue
|
||||||
|
3. Task A completes at 02:15 AM
|
||||||
|
4. Task B starts immediately at 02:15 AM (priority over other scheduled tasks)
|
||||||
|
```
|
||||||
|
|
||||||
|
### Scenario 3: Container Restart During Execution
|
||||||
|
```
|
||||||
|
02:00 AM - Task A starts downloading
|
||||||
|
02:10 AM - Container restarts (Docker update, etc.)
|
||||||
|
02:10 AM - Container comes back up
|
||||||
|
02:10 AM - Task A marked as "failed" with "interrupted" message
|
||||||
|
02:10 AM - Scheduled tasks reload, Task A will retry at next scheduled time (tomorrow 02:00 AM)
|
||||||
|
02:10 AM - Other pending tasks start processing from queue
|
||||||
|
```
|
||||||
|
|
||||||
|
## Future Enhancements (Not in Initial Implementation)
|
||||||
|
|
||||||
|
1. **Parallel Execution**: Optional setting to allow N tasks at once (requires more resources)
|
||||||
|
2. **Smart Scheduling**: Stagger start times automatically if many tasks at same time
|
||||||
|
3. **Queue Priorities**: User-configurable priority levels for tasks
|
||||||
|
4. **Retry Logic**: Auto-retry failed tasks with exponential backoff
|
||||||
|
5. **Additional Frequencies**: Weekly, custom intervals
|
||||||
|
6. **Notification System**: Email/webhook notifications on completion/failure
|
||||||
|
7. **Advanced Filters**: Score thresholds, content type filters
|
||||||
|
8. **Task Templates**: Save and reuse task configurations
|
||||||
|
9. **Execution History Page**: Dedicated page for detailed history with charts
|
||||||
|
10. **Bulk Operations**: Enable/disable/delete multiple tasks at once
|
||||||
|
11. **Export/Import**: Backup and restore scheduled tasks
|
||||||
|
|
||||||
|
## Conclusion
|
||||||
|
|
||||||
|
This implementation provides a robust scheduled downloads system with **guaranteed sequential execution**, designed specifically for Docker deployment in resource-constrained environments. The queue-based approach ensures:
|
||||||
|
|
||||||
|
- ✅ Only one download at a time (no resource contention)
|
||||||
|
- ✅ Fair task ordering (FIFO with priority support)
|
||||||
|
- ✅ Data persistence across container restarts
|
||||||
|
- ✅ Reliable scheduling with APScheduler
|
||||||
|
- ✅ Proper timezone handling (user TZ -> container UTC)
|
||||||
|
- ✅ Simple, clean UI
|
||||||
|
- ✅ Integration with existing BDFR API
|
||||||
|
- ✅ Smart defaults (daily schedule, time_filter="day", no_dupes=true)
|
||||||
|
- ✅ Easy management (enable/disable/delete/run now)
|
||||||
|
- ✅ Container-aware logging and health checks
|
||||||
|
- ✅ Transparent queue status for users
|
||||||
|
|
||||||
|
The sequential execution model is perfect for:
|
||||||
|
- Single-user home servers
|
||||||
|
- Docker containers with limited CPU/memory
|
||||||
|
- Preventing Reddit API rate limits
|
||||||
|
- Ensuring reliable, predictable downloads
|
||||||
|
- Avoiding file system contention
|
||||||
|
|
||||||
|
The system is production-ready for Docker deployment and extensible for future enhancements like parallel execution if needed.
|
||||||
+109
-11
@@ -23,6 +23,8 @@ from enum import Enum
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Dict, List, Optional, Union
|
from typing import Any, Dict, List, Optional, Union
|
||||||
|
|
||||||
|
import prawcore
|
||||||
|
|
||||||
from bdfr.configuration import Configuration
|
from bdfr.configuration import Configuration
|
||||||
from bdfr.connector import RedditConnector
|
from bdfr.connector import RedditConnector
|
||||||
from bdfr.downloader import RedditDownloader
|
from bdfr.downloader import RedditDownloader
|
||||||
@@ -33,6 +35,62 @@ from bdfr import exceptions as errors
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def retry_reddit_api_call(api_call_func: Callable, max_retries: int = 5, base_wait_time: int = 60) -> Any:
|
||||||
|
"""
|
||||||
|
Retry wrapper for Reddit API calls that handles rate limiting (429 errors).
|
||||||
|
|
||||||
|
Args:
|
||||||
|
api_call_func: Function that makes the Reddit API call
|
||||||
|
max_retries: Maximum number of retry attempts (default: 5)
|
||||||
|
base_wait_time: Base wait time in seconds (default: 60)
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Result of the API call
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
Exception: If all retry attempts are exhausted
|
||||||
|
"""
|
||||||
|
current_wait_time = base_wait_time
|
||||||
|
max_wait_time = base_wait_time * max_retries # Total max wait time
|
||||||
|
|
||||||
|
for attempt in range(max_retries + 1): # +1 for initial attempt
|
||||||
|
try:
|
||||||
|
logger.debug(f"Reddit API call attempt {attempt + 1}/{max_retries + 1}")
|
||||||
|
return api_call_func()
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
# Check if this is a rate limiting error
|
||||||
|
is_rate_limited = False
|
||||||
|
|
||||||
|
# Check for PRAW TooManyRequests exception
|
||||||
|
if isinstance(e, prawcore.exceptions.TooManyRequests):
|
||||||
|
is_rate_limited = True
|
||||||
|
logger.warning(f"Reddit API rate limited (TooManyRequests): {e}")
|
||||||
|
# Check for HTTP 429 in error message
|
||||||
|
elif "429" in str(e):
|
||||||
|
is_rate_limited = True
|
||||||
|
logger.warning(f"Reddit API rate limited (HTTP 429): {e}")
|
||||||
|
|
||||||
|
if not is_rate_limited:
|
||||||
|
# Not a rate limiting error, re-raise immediately
|
||||||
|
logger.error(f"Reddit API call failed with non-rate-limit error: {e}")
|
||||||
|
raise
|
||||||
|
|
||||||
|
# This is a rate limiting error
|
||||||
|
if attempt == max_retries:
|
||||||
|
# Last attempt failed, give up
|
||||||
|
logger.error(f"Reddit API rate limit retry attempts exhausted ({max_retries + 1} attempts)")
|
||||||
|
logger.error(f"Final error: {e}")
|
||||||
|
raise Exception(f"Reddit API rate limited after {max_retries + 1} attempts: {e}")
|
||||||
|
|
||||||
|
# Wait before retrying
|
||||||
|
logger.info(f"Reddit API rate limited, waiting {current_wait_time} seconds before retry {attempt + 1}/{max_retries}")
|
||||||
|
time.sleep(current_wait_time)
|
||||||
|
|
||||||
|
# Increase wait time for next attempt (exponential backoff)
|
||||||
|
current_wait_time = min(current_wait_time + base_wait_time, max_wait_time)
|
||||||
|
|
||||||
|
|
||||||
class DownloadType(Enum):
|
class DownloadType(Enum):
|
||||||
"""Types of downloads supported by BDFR"""
|
"""Types of downloads supported by BDFR"""
|
||||||
SUBREDDIT = "subreddit"
|
SUBREDDIT = "subreddit"
|
||||||
@@ -424,10 +482,15 @@ class BDFRManager:
|
|||||||
TooManyRequests = getattr(prawcore.exceptions, "TooManyRequests", None)
|
TooManyRequests = getattr(prawcore.exceptions, "TooManyRequests", None)
|
||||||
if TooManyRequests is not None and isinstance(e, TooManyRequests):
|
if TooManyRequests is not None and isinstance(e, TooManyRequests):
|
||||||
is_rate_limited = True
|
is_rate_limited = True
|
||||||
|
self.logger.info(f"[DEBUG] Exception is TooManyRequests for download {download_id}")
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
if "429" in str(e):
|
if "429" in str(e):
|
||||||
is_rate_limited = True
|
is_rate_limited = True
|
||||||
|
self.logger.info(f"[DEBUG] Error message contains '429' for download {download_id}: {str(e)}")
|
||||||
|
|
||||||
|
self.logger.error(f"[DEBUG] Exception caught in _run_download for {download_id}: {e}")
|
||||||
|
self.logger.error(f"[DEBUG] Stack trace: {stack_trace}")
|
||||||
|
|
||||||
# Mark as failed
|
# Mark as failed
|
||||||
download_info["status"] = DownloadStatus.FAILED.value
|
download_info["status"] = DownloadStatus.FAILED.value
|
||||||
@@ -556,10 +619,21 @@ class BDFRManager:
|
|||||||
)
|
)
|
||||||
asyncio.run(self._notify_callbacks(callbacks, "on_progress", progress_event))
|
asyncio.run(self._notify_callbacks(callbacks, "on_progress", progress_event))
|
||||||
|
|
||||||
# Process this submission
|
# Process this submission with retry mechanism for rate limiting
|
||||||
self.logger.info(f"[DEBUG] Calling _download_submission for {submission.id}")
|
self.logger.info(f"[DEBUG] Calling _download_submission for {submission.id}")
|
||||||
downloader._download_submission(submission)
|
try:
|
||||||
self.logger.info(f"[DEBUG] Completed _download_submission for {submission.id}")
|
downloader._download_submission(submission)
|
||||||
|
self.logger.info(f"[DEBUG] Completed _download_submission for {submission.id}")
|
||||||
|
except Exception as e:
|
||||||
|
# Check if this is a rate limiting error that should be retried
|
||||||
|
error_msg = str(e)
|
||||||
|
if "429" in error_msg:
|
||||||
|
self.logger.warning(f"[DEBUG] Rate limited while processing submission {submission.id}: {e}")
|
||||||
|
self.logger.warning(f"[DEBUG] Will continue with next submission instead of failing entire download")
|
||||||
|
continue
|
||||||
|
else:
|
||||||
|
# Not a rate limiting error, re-raise
|
||||||
|
raise
|
||||||
|
|
||||||
# Update processed count
|
# Update processed count
|
||||||
download_info["items_processed"] = processed_submissions
|
download_info["items_processed"] = processed_submissions
|
||||||
@@ -688,10 +762,21 @@ class BDFRManager:
|
|||||||
)
|
)
|
||||||
asyncio.run(self._notify_callbacks(callbacks, "on_progress", progress_event))
|
asyncio.run(self._notify_callbacks(callbacks, "on_progress", progress_event))
|
||||||
|
|
||||||
# Archive this item
|
# Archive this item with retry mechanism for rate limiting
|
||||||
self.logger.info(f"[DEBUG] Calling write_entry for {item_id}")
|
self.logger.info(f"[DEBUG] Calling write_entry for {item_id}")
|
||||||
downloader.write_entry(item)
|
try:
|
||||||
self.logger.info(f"[DEBUG] Completed write_entry for {item_id}")
|
downloader.write_entry(item)
|
||||||
|
self.logger.info(f"[DEBUG] Completed write_entry for {item_id}")
|
||||||
|
except Exception as e:
|
||||||
|
# Check if this is a rate limiting error that should be retried
|
||||||
|
error_msg = str(e)
|
||||||
|
if "429" in error_msg:
|
||||||
|
self.logger.warning(f"[DEBUG] Rate limited while archiving item {item_id}: {e}")
|
||||||
|
self.logger.warning(f"[DEBUG] Will continue with next item instead of failing entire download")
|
||||||
|
continue
|
||||||
|
else:
|
||||||
|
# Not a rate limiting error, re-raise
|
||||||
|
raise
|
||||||
|
|
||||||
# Update processed count
|
# Update processed count
|
||||||
download_info["items_processed"] = processed_items
|
download_info["items_processed"] = processed_items
|
||||||
@@ -801,12 +886,23 @@ class BDFRManager:
|
|||||||
)
|
)
|
||||||
asyncio.run(self._notify_callbacks(callbacks, "on_progress", progress_event))
|
asyncio.run(self._notify_callbacks(callbacks, "on_progress", progress_event))
|
||||||
|
|
||||||
# Clone this submission (download + archive)
|
# Clone this submission (download + archive) with retry mechanism for rate limiting
|
||||||
self.logger.info(f"[DEBUG] Calling _download_submission for {submission.id}")
|
self.logger.info(f"[DEBUG] Calling _download_submission for {submission.id}")
|
||||||
downloader._download_submission(submission)
|
try:
|
||||||
self.logger.info(f"[DEBUG] Calling write_entry for {submission.id}")
|
downloader._download_submission(submission)
|
||||||
downloader.write_entry(submission)
|
self.logger.info(f"[DEBUG] Calling write_entry for {submission.id}")
|
||||||
self.logger.info(f"[DEBUG] Completed cloning for {submission.id}")
|
downloader.write_entry(submission)
|
||||||
|
self.logger.info(f"[DEBUG] Completed cloning for {submission.id}")
|
||||||
|
except Exception as e:
|
||||||
|
# Check if this is a rate limiting error that should be retried
|
||||||
|
error_msg = str(e)
|
||||||
|
if "429" in error_msg:
|
||||||
|
self.logger.warning(f"[DEBUG] Rate limited while cloning submission {submission.id}: {e}")
|
||||||
|
self.logger.warning(f"[DEBUG] Will continue with next submission instead of failing entire download")
|
||||||
|
continue
|
||||||
|
else:
|
||||||
|
# Not a rate limiting error, re-raise
|
||||||
|
raise
|
||||||
|
|
||||||
# Update processed count
|
# Update processed count
|
||||||
download_info["items_processed"] = processed_items
|
download_info["items_processed"] = processed_items
|
||||||
@@ -820,6 +916,7 @@ class BDFRManager:
|
|||||||
if "429" in error_msg:
|
if "429" in error_msg:
|
||||||
self.logger.error(f"[DEBUG] Rate limited while cloning submission {submission_id}: {e}")
|
self.logger.error(f"[DEBUG] Rate limited while cloning submission {submission_id}: {e}")
|
||||||
self.logger.error(f"[DEBUG] Stack trace: {stack_trace}")
|
self.logger.error(f"[DEBUG] Stack trace: {stack_trace}")
|
||||||
|
self.logger.info(f"[DEBUG] Error message contains '429': {error_msg}")
|
||||||
download_info["status"] = DownloadStatus.FAILED.value
|
download_info["status"] = DownloadStatus.FAILED.value
|
||||||
download_info["error"] = f"Rate limited by Reddit API: {e}"
|
download_info["error"] = f"Rate limited by Reddit API: {e}"
|
||||||
download_info["end_time"] = datetime.now()
|
download_info["end_time"] = datetime.now()
|
||||||
@@ -870,6 +967,7 @@ class BDFRManager:
|
|||||||
stack_trace = traceback.format_exc()
|
stack_trace = traceback.format_exc()
|
||||||
self.logger.error(f"Error in progress download for {download_id}: {e}")
|
self.logger.error(f"Error in progress download for {download_id}: {e}")
|
||||||
self.logger.error(f"[DEBUG] Stack trace: {stack_trace}")
|
self.logger.error(f"[DEBUG] Stack trace: {stack_trace}")
|
||||||
|
self.logger.info(f"[DEBUG] Exception caught in progress_download, raising for {download_id}")
|
||||||
raise
|
raise
|
||||||
|
|
||||||
# Set up authentication if token provided
|
# Set up authentication if token provided
|
||||||
|
|||||||
@@ -413,8 +413,10 @@ class RedditConnector(metaclass=ABCMeta):
|
|||||||
is_rate_limited = False
|
is_rate_limited = False
|
||||||
if TooManyRequests is not None and isinstance(e, TooManyRequests):
|
if TooManyRequests is not None and isinstance(e, TooManyRequests):
|
||||||
is_rate_limited = True
|
is_rate_limited = True
|
||||||
|
logger.info(f"Rate limited detected: Exception is TooManyRequests for user {user}")
|
||||||
elif (hasattr(e, "response") and getattr(e.response, "status_code", None) == 429) or "429" in str(e):
|
elif (hasattr(e, "response") and getattr(e.response, "status_code", None) == 429) or "429" in str(e):
|
||||||
is_rate_limited = True
|
is_rate_limited = True
|
||||||
|
logger.info(f"Rate limited detected: Status code 429 or '429' in error message for user {user}. Error: {e}")
|
||||||
|
|
||||||
if is_rate_limited:
|
if is_rate_limited:
|
||||||
logger.error("Received HTTP 429 (rate limited). Propagating error to fail the download.")
|
logger.error("Received HTTP 429 (rate limited). Propagating error to fail the download.")
|
||||||
|
|||||||
+4
-1
@@ -149,7 +149,9 @@ class RedditDownloader(RedditConnector):
|
|||||||
logger.error(f"Site {downloader_class.__name__} failed to download submission {submission.id}: {e}")
|
logger.error(f"Site {downloader_class.__name__} failed to download submission {submission.id}: {e}")
|
||||||
return
|
return
|
||||||
files_processed = 0
|
files_processed = 0
|
||||||
|
logger.debug(f"Processing {len(content)} resources for submission {submission.id}")
|
||||||
for destination, res in self.file_name_formatter.format_resource_paths(content, self.download_directory):
|
for destination, res in self.file_name_formatter.format_resource_paths(content, self.download_directory):
|
||||||
|
logger.debug(f"Resource URL: {res.url}, Extension: {res.extension}, Destination: {destination}")
|
||||||
if destination.exists():
|
if destination.exists():
|
||||||
# Check if we already have this file's hash
|
# Check if we already have this file's hash
|
||||||
if destination in self.master_hash_list.values():
|
if destination in self.master_hash_list.values():
|
||||||
@@ -217,9 +219,10 @@ class RedditDownloader(RedditConnector):
|
|||||||
# Only create folder if we're actually going to write the file (not a duplicate)
|
# Only create folder if we're actually going to write the file (not a duplicate)
|
||||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||||
try:
|
try:
|
||||||
|
logger.debug(f"Writing {len(res.content)} bytes to {destination}")
|
||||||
with destination.open("wb") as file:
|
with destination.open("wb") as file:
|
||||||
file.write(res.content)
|
file.write(res.content)
|
||||||
logger.debug(f"Written file to {destination}")
|
logger.debug(f"Successfully written file to {destination}")
|
||||||
files_processed += 1
|
files_processed += 1
|
||||||
except OSError as e:
|
except OSError as e:
|
||||||
logger.exception(e)
|
logger.exception(e)
|
||||||
|
|||||||
@@ -126,6 +126,7 @@ class FileNameFormatter:
|
|||||||
)
|
)
|
||||||
index = f"_{index}" if index else ""
|
index = f"_{index}" if index else ""
|
||||||
if not resource.extension:
|
if not resource.extension:
|
||||||
|
logger.error(f"Resource from {resource.url} has no extension - URL: {resource.url}")
|
||||||
raise BulkDownloaderException(f"Resource from {resource.url} has no extension")
|
raise BulkDownloaderException(f"Resource from {resource.url} has no extension")
|
||||||
file_name = str(self._format_name(resource.source_submission, self.file_format_string))
|
file_name = str(self._format_name(resource.source_submission, self.file_format_string))
|
||||||
|
|
||||||
|
|||||||
+69
-2
@@ -24,7 +24,7 @@ class Resource:
|
|||||||
self.content: Optional[bytes] = None
|
self.content: Optional[bytes] = None
|
||||||
self.url = url
|
self.url = url
|
||||||
self.hash: Optional[_hashlib.HASH] = None
|
self.hash: Optional[_hashlib.HASH] = None
|
||||||
self.extension = extension
|
self.extension = self._normalize_extension(extension)
|
||||||
self.download_function = download_function
|
self.download_function = download_function
|
||||||
if not self.extension:
|
if not self.extension:
|
||||||
self.extension = self._determine_extension()
|
self.extension = self._determine_extension()
|
||||||
@@ -45,6 +45,13 @@ class Resource:
|
|||||||
raise
|
raise
|
||||||
if content:
|
if content:
|
||||||
self.content = content
|
self.content = content
|
||||||
|
|
||||||
|
# If we didn't have an extension before, try to detect from content
|
||||||
|
if not self.extension:
|
||||||
|
logger.debug(f"Attempting content-based extension detection for {self.url}")
|
||||||
|
detected = self._detect_extension_by_content()
|
||||||
|
self.extension = self._normalize_extension(detected) if detected else None
|
||||||
|
|
||||||
if not self.hash and self.content:
|
if not self.hash and self.content:
|
||||||
self.create_hash()
|
self.create_hash()
|
||||||
|
|
||||||
@@ -54,9 +61,69 @@ class Resource:
|
|||||||
def _determine_extension(self) -> Optional[str]:
|
def _determine_extension(self) -> Optional[str]:
|
||||||
extension_pattern = re.compile(r".*(\..{3,5})$")
|
extension_pattern = re.compile(r".*(\..{3,5})$")
|
||||||
stripped_url = urllib.parse.urlsplit(self.url).path
|
stripped_url = urllib.parse.urlsplit(self.url).path
|
||||||
|
|
||||||
|
# Special handling for Reddit media URLs
|
||||||
|
if self.url.startswith("https://www.reddit.com/media"):
|
||||||
|
logger.debug(f"Detected Reddit media URL: {self.url}")
|
||||||
|
parsed_url = urllib.parse.urlparse(self.url)
|
||||||
|
url_param = urllib.parse.parse_qs(parsed_url.query).get('url', [None])[0]
|
||||||
|
if url_param:
|
||||||
|
decoded_url = urllib.parse.unquote(url_param)
|
||||||
|
logger.debug(f"Reddit media URL decoded to: {decoded_url}")
|
||||||
|
stripped_url = urllib.parse.urlsplit(decoded_url).path
|
||||||
|
|
||||||
|
# Also handle preview.redd.it URLs which might not have extensions
|
||||||
|
elif "preview.redd.it" in self.url and not stripped_url.endswith(('.jpg', '.jpeg', '.png', '.gif', '.webp')):
|
||||||
|
logger.debug(f"Detected preview.redd.it URL without extension: {self.url}")
|
||||||
|
# For preview URLs, try to infer from common patterns or add fallback logic
|
||||||
|
|
||||||
match = re.search(extension_pattern, stripped_url)
|
match = re.search(extension_pattern, stripped_url)
|
||||||
if match:
|
if match:
|
||||||
return match.group(1)
|
extension = match.group(1)
|
||||||
|
logger.debug(f"URL {self.url} -> extracted extension: {extension} (from path: {stripped_url})")
|
||||||
|
return self._normalize_extension(extension)
|
||||||
|
else:
|
||||||
|
logger.warning(f"Could not determine extension for URL: {self.url} (path: {stripped_url})")
|
||||||
|
|
||||||
|
# As a last resort, if we have content, try to detect by magic numbers
|
||||||
|
if hasattr(self, 'content') and self.content:
|
||||||
|
detected = self._detect_extension_by_content()
|
||||||
|
return self._normalize_extension(detected) if detected else None
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _detect_extension_by_content(self) -> Optional[str]:
|
||||||
|
"""Detect file extension by examining file content (magic numbers)"""
|
||||||
|
if not self.content or len(self.content) < 16:
|
||||||
|
return None
|
||||||
|
|
||||||
|
# Check for common image formats
|
||||||
|
if self.content.startswith(b'\xFF\xD8\xFF'):
|
||||||
|
logger.debug(f"Detected JPEG by magic number for URL: {self.url}")
|
||||||
|
return '.jpg'
|
||||||
|
elif self.content.startswith(b'\x89PNG\r\n\x1a\n'):
|
||||||
|
logger.debug(f"Detected PNG by magic number for URL: {self.url}")
|
||||||
|
return '.png'
|
||||||
|
elif self.content.startswith(b'GIF87a') or self.content.startswith(b'GIF89a'):
|
||||||
|
logger.debug(f"Detected GIF by magic number for URL: {self.url}")
|
||||||
|
return '.gif'
|
||||||
|
elif self.content.startswith(b'RIFF') and self.content[8:12] == b'WEBP':
|
||||||
|
logger.debug(f"Detected WebP by magic number for URL: {self.url}")
|
||||||
|
return '.webp'
|
||||||
|
elif self.content.startswith(b'BM'):
|
||||||
|
logger.debug(f"Detected BMP by magic number for URL: {self.url}")
|
||||||
|
return '.bmp'
|
||||||
|
|
||||||
|
logger.debug(f"Could not detect file type by magic number for URL: {self.url}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _normalize_extension(self, extension: Optional[str]) -> Optional[str]:
|
||||||
|
"""Normalize extension to lowercase for consistency"""
|
||||||
|
if not extension:
|
||||||
|
return None
|
||||||
|
normalized = extension.lower()
|
||||||
|
logger.debug(f"Normalized extension '{extension}' to '{normalized}'")
|
||||||
|
return normalized
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def http_download(url: str, download_parameters: dict) -> Optional[bytes]:
|
def http_download(url: str, download_parameters: dict) -> Optional[bytes]:
|
||||||
|
|||||||
@@ -24,8 +24,14 @@ from bdfr.site_downloaders.youtube import Youtube
|
|||||||
class DownloadFactory:
|
class DownloadFactory:
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def pull_lever(url: str) -> type[BaseDownloader]:
|
def pull_lever(url: str) -> type[BaseDownloader]:
|
||||||
|
import logging
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
sanitised_url = DownloadFactory.sanitise_url(url).lower()
|
sanitised_url = DownloadFactory.sanitise_url(url).lower()
|
||||||
|
logger.debug(f"Selecting downloader for URL: {url} (sanitized: {sanitised_url})")
|
||||||
|
|
||||||
if re.match(r"(i\.|m\.|o\.)?imgur", sanitised_url):
|
if re.match(r"(i\.|m\.|o\.)?imgur", sanitised_url):
|
||||||
|
logger.debug("Using Imgur downloader")
|
||||||
return Imgur
|
return Imgur
|
||||||
elif re.match(r"(i\.|thumbs\d\.|v\d\.)?(redgifs|gifdeliverynetwork)", sanitised_url):
|
elif re.match(r"(i\.|thumbs\d\.|v\d\.)?(redgifs|gifdeliverynetwork)", sanitised_url):
|
||||||
return Redgifs
|
return Redgifs
|
||||||
|
|||||||
@@ -20,12 +20,19 @@ class YtdlpFallback(BaseFallbackDownloader, Youtube):
|
|||||||
super(YtdlpFallback, self).__init__(post)
|
super(YtdlpFallback, self).__init__(post)
|
||||||
|
|
||||||
def find_resources(self, authenticator: Optional[SiteAuthenticator] = None) -> list[Resource]:
|
def find_resources(self, authenticator: Optional[SiteAuthenticator] = None) -> list[Resource]:
|
||||||
|
logger.debug(f"YtdlpFallback processing URL: {self.post.url}")
|
||||||
|
video_attrs = super().get_video_attributes(self.post.url)
|
||||||
|
logger.debug(f"Video attributes: {video_attrs}")
|
||||||
|
extension = video_attrs.get("ext", None)
|
||||||
|
logger.debug(f"Using extension: {extension}")
|
||||||
|
|
||||||
out = Resource(
|
out = Resource(
|
||||||
self.post,
|
self.post,
|
||||||
self.post.url,
|
self.post.url,
|
||||||
super()._download_video({}),
|
super()._download_video({}),
|
||||||
super().get_video_attributes(self.post.url)["ext"],
|
extension,
|
||||||
)
|
)
|
||||||
|
logger.debug(f"Created resource with extension: {out.extension}")
|
||||||
return [out]
|
return [out]
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|||||||
@@ -0,0 +1,96 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Test script to verify that the duplicate folder creation fix works correctly.
|
||||||
|
This script simulates the scenario where duplicate posts would previously create empty folders.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import tempfile
|
||||||
|
import shutil
|
||||||
|
from pathlib import Path
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
# Add the bdfr module to the path
|
||||||
|
import sys
|
||||||
|
sys.path.insert(0, '.')
|
||||||
|
|
||||||
|
from bdfr.configuration import Configuration
|
||||||
|
from bdfr.connector import RedditConnector
|
||||||
|
from bdfr.downloader import RedditDownloader
|
||||||
|
|
||||||
|
|
||||||
|
def test_duplicate_folder_creation_fix():
|
||||||
|
"""Test that folders are not created for duplicate posts when no_dupes is enabled."""
|
||||||
|
|
||||||
|
# Create a temporary directory for testing
|
||||||
|
with tempfile.TemporaryDirectory() as temp_dir:
|
||||||
|
temp_path = Path(temp_dir)
|
||||||
|
|
||||||
|
# Create test configuration
|
||||||
|
args = Configuration()
|
||||||
|
args.no_dupes = True
|
||||||
|
args.folder_scheme = ""
|
||||||
|
args.file_scheme = "{POSTID}"
|
||||||
|
|
||||||
|
# Create downloader instance
|
||||||
|
downloader = RedditDownloader(args)
|
||||||
|
downloader.download_directory = temp_path
|
||||||
|
downloader.file_name_formatter = RedditConnector.create_file_name_formatter(downloader)
|
||||||
|
|
||||||
|
# Mock a submission
|
||||||
|
submission = MagicMock()
|
||||||
|
submission.id = "test123"
|
||||||
|
submission.subreddit.display_name = "testsubreddit"
|
||||||
|
submission.author.name = "testuser"
|
||||||
|
submission.score = 100
|
||||||
|
submission.upvote_ratio = 0.8
|
||||||
|
submission.created_utc = 1640995200 # Jan 1, 2022
|
||||||
|
submission.url = "https://example.com/image.jpg"
|
||||||
|
submission.title = "Test Post"
|
||||||
|
|
||||||
|
# Mock the downloader chain
|
||||||
|
mock_downloader_class = MagicMock()
|
||||||
|
mock_downloader_class.__name__ = "MockDownloader"
|
||||||
|
|
||||||
|
mock_downloader = MagicMock()
|
||||||
|
mock_resource = MagicMock()
|
||||||
|
mock_resource.url = "https://example.com/image.jpg"
|
||||||
|
mock_resource.extension = "jpg"
|
||||||
|
mock_resource.hash.hexdigest.return_value = "duplicate_hash_12345"
|
||||||
|
mock_resource.content = b"fake image content"
|
||||||
|
|
||||||
|
mock_downloader.find_resources.return_value = [mock_resource]
|
||||||
|
|
||||||
|
# Set up the master hash list to contain our "duplicate" hash
|
||||||
|
test_hash = "duplicate_hash_12345"
|
||||||
|
existing_file = temp_path / "existing_file.jpg"
|
||||||
|
existing_file.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
existing_file.touch()
|
||||||
|
downloader.master_hash_list = {test_hash: existing_file}
|
||||||
|
|
||||||
|
# Mock the download factory
|
||||||
|
import bdfr.site_downloaders.download_factory as df
|
||||||
|
original_pull_lever = df.DownloadFactory.pull_lever
|
||||||
|
df.DownloadFactory.pull_lever = MagicMock(return_value=mock_downloader_class)
|
||||||
|
|
||||||
|
try:
|
||||||
|
# Call the download submission method
|
||||||
|
downloader._download_submission(submission)
|
||||||
|
|
||||||
|
# Check that no new directories were created (the fix)
|
||||||
|
subdirs = [d for d in temp_path.rglob("*") if d.is_dir() and d != temp_path]
|
||||||
|
print(f"Number of subdirectories created: {len(subdirs)}")
|
||||||
|
|
||||||
|
# With the fix, no new directories should be created for duplicates
|
||||||
|
# The only directory that might exist is the one we created for the existing file
|
||||||
|
assert len(subdirs) <= 1, f"Expected 0 or 1 subdirectories, but found {len(subdirs)}"
|
||||||
|
|
||||||
|
print("Test passed: No empty folders created for duplicate posts!")
|
||||||
|
|
||||||
|
finally:
|
||||||
|
# Restore original function
|
||||||
|
df.DownloadFactory.pull_lever = original_pull_lever
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
test_duplicate_folder_creation_fix()
|
||||||
|
print("All tests passed! The duplicate folder creation fix is working correctly.")
|
||||||
@@ -0,0 +1,73 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Test extension case normalization functionality
|
||||||
|
"""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
from bdfr.resource import Resource
|
||||||
|
|
||||||
|
|
||||||
|
class TestExtensionNormalization:
|
||||||
|
"""Test that extensions are properly normalized to lowercase"""
|
||||||
|
|
||||||
|
def test_url_extensions_normalized(self):
|
||||||
|
"""Test that extensions from URLs are normalized to lowercase"""
|
||||||
|
test_cases = [
|
||||||
|
("https://example.com/image.JPG", ".jpg"),
|
||||||
|
("https://example.com/image.jpeg", ".jpeg"),
|
||||||
|
("https://example.com/image.JPEG", ".jpeg"),
|
||||||
|
("https://example.com/image.jpg", ".jpg"),
|
||||||
|
("https://example.com/image.PNG", ".png"),
|
||||||
|
("https://example.com/image.GIF", ".gif"),
|
||||||
|
]
|
||||||
|
|
||||||
|
for url, expected in test_cases:
|
||||||
|
mock_submission = MagicMock()
|
||||||
|
mock_submission.id = "test123"
|
||||||
|
|
||||||
|
resource = Resource(mock_submission, url, lambda: None)
|
||||||
|
assert resource.extension == expected, f"URL {url} should normalize to {expected}, got {resource.extension}"
|
||||||
|
|
||||||
|
def test_reddit_media_urls_normalized(self):
|
||||||
|
"""Test that Reddit media URLs are properly normalized"""
|
||||||
|
test_cases = [
|
||||||
|
("https://www.reddit.com/media?url=https%3A%2F%2Fi.redd.it%2Fr2mv10i4vkfd1.JPG", ".jpg"),
|
||||||
|
("https://www.reddit.com/media?url=https%3A%2F%2Fi.redd.it%2Fr2mv10i4vkfd1.jpeg", ".jpeg"),
|
||||||
|
("https://www.reddit.com/media?url=https%3A%2F%2Fi.redd.it%2Fr2mv10i4vkfd1.JPEG", ".jpeg"),
|
||||||
|
("https://www.reddit.com/media?url=https%3A%2F%2Fi.redd.it%2Fr2mv10i4vkfd1.PNG", ".png"),
|
||||||
|
]
|
||||||
|
|
||||||
|
for url, expected in test_cases:
|
||||||
|
mock_submission = MagicMock()
|
||||||
|
mock_submission.id = "test123"
|
||||||
|
|
||||||
|
resource = Resource(mock_submission, url, lambda: None)
|
||||||
|
assert resource.extension == expected, f"Reddit media URL {url} should normalize to {expected}, got {resource.extension}"
|
||||||
|
|
||||||
|
def test_constructor_extensions_normalized(self):
|
||||||
|
"""Test that extensions passed to constructor are normalized"""
|
||||||
|
test_cases = [
|
||||||
|
(".JPG", ".jpg"),
|
||||||
|
(".JPEG", ".jpeg"),
|
||||||
|
(".PNG", ".png"),
|
||||||
|
(".GIF", ".gif"),
|
||||||
|
]
|
||||||
|
|
||||||
|
for input_ext, expected in test_cases:
|
||||||
|
mock_submission = MagicMock()
|
||||||
|
mock_submission.id = "test123"
|
||||||
|
|
||||||
|
resource = Resource(mock_submission, "https://example.com/test", lambda: None, input_ext)
|
||||||
|
assert resource.extension == expected, f"Constructor extension {input_ext} should normalize to {expected}, got {resource.extension}"
|
||||||
|
|
||||||
|
def test_magic_number_detection_normalized(self):
|
||||||
|
"""Test that magic number detection returns normalized extensions"""
|
||||||
|
mock_submission = MagicMock()
|
||||||
|
mock_submission.id = "test123"
|
||||||
|
|
||||||
|
# Test JPEG magic number detection
|
||||||
|
jpeg_content = b'\xFF\xD8\xFF' + b'0' * 100 # JPEG magic number
|
||||||
|
resource = Resource(mock_submission, "https://example.com/no-extension", lambda params: jpeg_content)
|
||||||
|
resource.download() # Trigger content-based detection
|
||||||
|
assert resource.extension == ".jpg", f"Magic number detection should return .jpg, got {resource.extension}"
|
||||||
@@ -0,0 +1,64 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""
|
||||||
|
Test script to debug file extension detection issues
|
||||||
|
"""
|
||||||
|
|
||||||
|
import sys
|
||||||
|
import os
|
||||||
|
sys.path.insert(0, os.path.dirname(__file__))
|
||||||
|
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
from bdfr.resource import Resource
|
||||||
|
|
||||||
|
def test_extension_detection():
|
||||||
|
"""Test extension detection with various URL patterns"""
|
||||||
|
|
||||||
|
test_cases = [
|
||||||
|
# Standard URLs with extensions
|
||||||
|
("https://example.com/image.jpg", ".jpg"),
|
||||||
|
("https://example.com/video.mp4", ".mp4"),
|
||||||
|
("https://files.example.com/document.pdf", ".pdf"),
|
||||||
|
|
||||||
|
# URLs without extensions
|
||||||
|
("https://example.com/api/data", None),
|
||||||
|
("https://example.com/path/without/extension", None),
|
||||||
|
|
||||||
|
# URLs with query parameters
|
||||||
|
("https://example.com/image.jpg?size=large", ".jpg"),
|
||||||
|
("https://example.com/video.mp4?utm_source=test", ".mp4"),
|
||||||
|
|
||||||
|
# URLs with fragments
|
||||||
|
("https://example.com/image.png#section", ".png"),
|
||||||
|
|
||||||
|
# Complex paths
|
||||||
|
("https://imgur.com/a/gallery123", None),
|
||||||
|
("https://reddit.com/r/test/abc123_def456_789", None),
|
||||||
|
|
||||||
|
# Edge cases that might cause weird names
|
||||||
|
("https://example.com/L7SW9E~G", None),
|
||||||
|
("https://example.com/temp/file", None),
|
||||||
|
|
||||||
|
# Reddit media URLs (the actual issue)
|
||||||
|
("https://www.reddit.com/media?url=https%3A%2F%2Fi.redd.it%2Fr2mv10i4vkfd1.jpeg", ".jpeg"),
|
||||||
|
("https://i.redd.it/r2mv10i4vkfd1.jpeg", ".jpeg"),
|
||||||
|
]
|
||||||
|
|
||||||
|
print("Testing extension detection with various URLs:")
|
||||||
|
print("=" * 60)
|
||||||
|
|
||||||
|
for url, expected in test_cases:
|
||||||
|
# Create a mock submission
|
||||||
|
mock_submission = MagicMock()
|
||||||
|
mock_submission.id = "test123"
|
||||||
|
|
||||||
|
# Create resource and test extension detection
|
||||||
|
resource = Resource(mock_submission, url, lambda: None)
|
||||||
|
|
||||||
|
print(f"URL: {url}")
|
||||||
|
print(f"Expected: {expected}")
|
||||||
|
print(f"Detected: {resource.extension}")
|
||||||
|
print(f"Match: {'YES' if resource.extension == expected else 'NO'}")
|
||||||
|
print("-" * 40)
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
test_extension_detection()
|
||||||
@@ -319,6 +319,7 @@ class WebSocketProgressCallback(ProgressCallback if BDFR_AVAILABLE else MockProg
|
|||||||
async def on_error(self, event: ProgressEvent):
|
async def on_error(self, event: ProgressEvent):
|
||||||
"""Send error update to WebSocket clients"""
|
"""Send error update to WebSocket clients"""
|
||||||
logger.info(f"[WEBSOCKET-ERROR] Received error event for download {event.download_id}: {event.message}")
|
logger.info(f"[WEBSOCKET-ERROR] Received error event for download {event.download_id}: {event.message}")
|
||||||
|
logger.info(f"[WEBSOCKET-ERROR] Event data: {event.data}")
|
||||||
try:
|
try:
|
||||||
error_data = {
|
error_data = {
|
||||||
"type": "error",
|
"type": "error",
|
||||||
|
|||||||
Reference in New Issue
Block a user