Scheduler Documentation

Overview

The Scheduler system automates the complete content creation and publishing workflow for faceless YouTube videos. It orchestrates script generation, video assembly, and YouTube uploads on a flexible schedule.

Table of Contents

• Architecture
• Components
• Installation
• Quick Start
• API Reference
• Scheduling Patterns
• Best Practices
• Troubleshooting

---

Architecture

System Overview

┌─────────────────────────────────────────────────────────┐
│                   Scheduler System                      │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌─────────────────┐      ┌────────────────────────┐  │
│  │ RecurringScheduler│────▶│  ContentScheduler     │  │
│  │  (Patterns)      │      │  (Orchestration)      │  │
│  └─────────────────┘      └────────────────────────┘  │
│           │                         │                  │
│           │                         ▼                  │
│           │              ┌────────────────────────┐    │
│           │              │    JobExecutor         │    │
│           │              │  (Background Exec)     │    │
│           │              └────────────────────────┘    │
│           │                         │                  │
│           ▼                         ▼                  │
│  ┌─────────────────┐      ┌────────────────────────┐  │
│  │ CalendarManager │      │  Workflow Pipeline     │  │
│  │  (Planning)     │      │                        │  │
│  └─────────────────┘      │  1. Script Generator   │  │
│                            │  2. Video Assembler    │  │
│                            │  3. YouTube Uploader   │  │
│                            └────────────────────────┘  │
└─────────────────────────────────────────────────────────┘

Workflow Stages

1. Script Generation - AI generates video script
2. Asset Selection - Select background videos and audio
3. Video Assembly - TTS, timeline, rendering
4. Thumbnail Creation - Generate video thumbnail
5. Metadata Preparation - Prepare YouTube metadata
6. YouTube Upload - Upload with metadata
7. Analytics Tracking - Track performance

---

Components

ContentScheduler

Main orchestration service that manages the complete workflow.

Key Features:

• Schedule single videos
• Batch scheduling
• Automatic retry with exponential backoff
• Progress tracking
• Job cancellation/pause/resume
• Execution history

Usage:

from services.scheduler import ContentScheduler, ScheduleConfig

scheduler = ContentScheduler(config=ScheduleConfig())

# Schedule video
job_id = await scheduler.schedule_video(
    topic="Python Functions Tutorial",
    scheduled_at=datetime(2024, 12, 25, 10, 0),
    publish_at=datetime(2024, 12, 25, 12, 0),
    style="educational",
    duration_minutes=5,
    tags=["python", "tutorial"]
)

# Start scheduler
await scheduler.start()

# Monitor progress
job = await scheduler.get_job_status(job_id)
print(f"Progress: {job.progress_percent}%")

JobExecutor

Background execution engine with retry logic.

Key Features:

• Async job execution
• Configurable retry strategies (fixed, exponential, linear)
• Concurrent job limits
• Timeout handling
• Progress callbacks
• Execution history

Retry Strategies:

• NONE: No retries
• FIXED_DELAY: Fixed delay between retries
• EXPONENTIAL_BACKOFF: 2^n * base_delay (default)
• LINEAR_BACKOFF: n * base_delay

Usage:

from services.scheduler import JobExecutor, ExecutorConfig, RetryStrategy

executor = JobExecutor(config=ExecutorConfig(
    max_concurrent_jobs=3,
    max_retries=3,
    retry_strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
    base_retry_delay_seconds=60
))

result = await executor.execute(
    job_func=my_async_function,
    job_args={"param": "value"},
    job_id="my-job",
    timeout=3600
)

if result.status == ExecutionStatus.COMPLETED:
    print(f"Success! Result: {result.result_data}")

RecurringScheduler

Pattern-based scheduling using APScheduler.

Key Features:

• Daily schedules
• Weekly schedules (specific days)
• Monthly schedules (specific dates)
• Interval schedules
• Custom cron expressions
• Dynamic topic generation
• Pause/resume schedules

Usage:

from services.scheduler import RecurringScheduler, DayOfWeek

scheduler = RecurringScheduler(content_scheduler)

# Daily at 10 AM
await scheduler.create_daily_schedule(
    name="Daily Python Tips",
    topic_template="Python Tip - {date}",
    hour=10,
    minute=0
)

# Weekly: Mon/Wed/Fri at 2 PM
await scheduler.create_weekly_schedule(
    name="Weekly Tutorials",
    topic_template="Tutorial Week {week}",
    days=[DayOfWeek.MONDAY, DayOfWeek.WEDNESDAY, DayOfWeek.FRIDAY],
    hour=14,
    minute=0
)

# Monthly: 1st and 15th at 9 AM
await scheduler.create_monthly_schedule(
    name="Monthly Deep Dives",
    topic_template="Deep Dive {month} {year}",
    days_of_month=[1, 15],
    hour=9,
    minute=0
)

await scheduler.start()

Topic Template Variables:

• {date} - 2024-06-15
• {time} - 14:30
• {datetime} - 2024-06-15 14:30
• {year} - 2024
• {month} - June
• {month_num} - 6
• {day} - 15
• {weekday} - Saturday
• {week} - 24 (week number)
• {timestamp} - 1718456400 (Unix timestamp)

CalendarManager

Content calendar and planning system.

Key Features:

• Visual calendar representation
• Conflict detection (time and topic)
• Optimal slot suggestions
• Content gap analysis
• Day/week/month views
• Slot reservation
• Utilization tracking

Usage:

from services.scheduler import CalendarManager, CalendarConfig

calendar = CalendarManager(config=CalendarConfig(
    min_gap_hours=6,
    max_videos_per_day=3,
    preferred_hours=[10, 14, 18]
))

# Reserve slot
slot = await calendar.reserve_slot(
    scheduled_at=datetime(2024, 12, 25, 10, 0),
    topic="Python Tutorial",
    duration_minutes=5
)

# Get week view
week_view = await calendar.get_week_view(start_date=date.today())

for entry in week_view:
    print(f"{entry.date}: {entry.total_slots} slots")

# Suggest optimal slots
suggestions = await calendar.suggest_optimal_slots(
    count=5,
    start_date=date.today(),
    days=30
)

# Detect conflicts
conflicts = await calendar.detect_conflicts()

---

Installation

Prerequisites

pip install apscheduler pydantic asyncio

Dependencies

The scheduler integrates with:

• ScriptGenerator (Task #6) - AI script generation
• VideoAssembler (Task #7) - Video production
• YouTubeUploader (Task #8) - YouTube uploads
• CacheManager (Task #4) - State caching

---

Quick Start

Complete Automation Setup

import asyncio
from datetime import datetime, timedelta
from services.scheduler import (
    ContentScheduler,
    RecurringScheduler,
    CalendarManager,
    ScheduleConfig,
    DayOfWeek
)

async def setup_automation():
    # Initialize scheduler
    scheduler = ContentScheduler(config=ScheduleConfig(
        jobs_storage_path="scheduled_jobs",
        output_dir="output_videos",
        max_retries=3,
        youtube_account="main"
    ))

    # Create recurring schedules
    recurring = RecurringScheduler(scheduler)

    # Daily tips at 10 AM
    await recurring.create_daily_schedule(
        name="Daily Python Tips",
        topic_template="Python Tip #{date}",
        hour=10,
        tags=["python", "tips", "daily"]
    )

    # Weekly tutorials
    await recurring.create_weekly_schedule(
        name="Weekly Tutorials",
        topic_template="Python Tutorial - Week {week}",
        days=[DayOfWeek.MONDAY, DayOfWeek.THURSDAY],
        hour=14,
        tags=["python", "tutorial"]
    )

    # Start automation
    await scheduler.start()
    await recurring.start()

    print("Automation running! Videos will be created automatically.")

    # Monitor
    while True:
        stats = scheduler.get_statistics()
        print(f"Active jobs: {stats['active_jobs']}, "
              f"Completed: {stats['statistics']['total_completed']}")

        await asyncio.sleep(60)

if __name__ == "__main__":
    asyncio.run(setup_automation())

---

API Reference

ContentScheduler

Methods

schedule_video()

Schedule single video for creation and upload.

async def schedule_video(
    topic: str,
    scheduled_at: datetime,
    publish_at: Optional[datetime] = None,
    style: str = "educational",
    duration_minutes: int = 5,
    tags: Optional[List[str]] = None,
    category: Category = Category.EDUCATION,
    privacy_status: PrivacyStatus = PrivacyStatus.PRIVATE
) -> str

Parameters:

• topic - Video topic (required)
• scheduled_at - When to create video (required)
• publish_at - When to publish (None = immediately)
• style - Content style (default: "educational")
• duration_minutes - Target duration (default: 5)
• tags - YouTube tags
• category - YouTube category
• privacy_status - Privacy setting

Returns: Job ID (str)

schedule_batch()

Schedule multiple videos at once.

async def schedule_batch(
    videos: List[Dict[str, Any]]
) -> List[str]

Parameters:

• videos - List of video configs (each with schedule_video params)

Returns: List of job IDs

get_job_status()

Get job status and progress.

async def get_job_status(job_id: str) -> Optional[ScheduledJob]

Returns: ScheduledJob with status, progress, results

cancel_job()

Cancel scheduled or active job.

async def cancel_job(job_id: str)

pause_job() / resume_job()

Pause or resume pending job.

async def pause_job(job_id: str)
async def resume_job(job_id: str)

get_all_jobs()

Get all jobs with optional filter.

def get_all_jobs(
    status_filter: Optional[JobStatus] = None
) -> List[ScheduledJob]

get_statistics()

Get scheduler statistics.

def get_statistics() -> Dict[str, Any]

Returns:

{
    "total_jobs": 10,
    "active_jobs": 2,
    "status_counts": {
        "pending": 3,
        "completed": 5,
        "failed": 2
    },
    "statistics": {
        "total_scheduled": 10,
        "total_completed": 5,
        "total_failed": 2
    },
    "running": True
}

JobExecutor

Methods

execute()

Execute job with retry logic.

async def execute(
    job_func: Callable,
    job_args: Optional[Dict[str, Any]] = None,
    job_id: Optional[str] = None,
    max_retries: Optional[int] = None,
    retry_strategy: Optional[RetryStrategy] = None,
    timeout: Optional[int] = None,
    progress_callback: Optional[Callable] = None
) -> ExecutionResult

execute_batch()

Execute multiple jobs.

async def execute_batch(
    jobs: List[Dict[str, Any]],
    fail_fast: bool = False
) -> List[ExecutionResult]

cancel_execution()

Cancel active execution.

async def cancel_execution(execution_id: str)

RecurringScheduler

Methods

create_daily_schedule()

Create daily recurring schedule.

async def create_daily_schedule(
    name: str,
    topic_template: str,
    hour: int = 10,
    minute: int = 0,
    style: str = "educational",
    duration_minutes: int = 5,
    tags: Optional[List[str]] = None,
    start_date: Optional[datetime] = None,
    end_date: Optional[datetime] = None
) -> str

create_weekly_schedule()

Create weekly recurring schedule.

async def create_weekly_schedule(
    name: str,
    topic_template: str,
    days: List[DayOfWeek],
    hour: int = 10,
    minute: int = 0,
    ...
) -> str

create_monthly_schedule()

Create monthly recurring schedule.

async def create_monthly_schedule(
    name: str,
    topic_template: str,
    days_of_month: List[int],  # 1-31
    hour: int = 10,
    minute: int = 0,
    ...
) -> str

create_cron_schedule()

Create schedule from cron expression.

async def create_cron_schedule(
    name: str,
    topic_template: str,
    cron_expression: str,  # e.g., "0 10 * * *"
    ...
) -> str

CalendarManager

Methods

reserve_slot()

Reserve calendar slot.

async def reserve_slot(
    scheduled_at: datetime,
    topic: str,
    duration_minutes: Optional[int] = None,
    ...
) -> ContentSlot

get_day_view()

Get calendar for single day.

async def get_day_view(target_date: date) -> CalendarEntry

get_week_view()

Get calendar for week.

async def get_week_view(start_date: date) -> List[CalendarEntry]

get_month_view()

Get calendar for month.

async def get_month_view(year: int, month: int) -> List[CalendarEntry]

suggest_optimal_slots()

Suggest optimal time slots.

async def suggest_optimal_slots(
    count: int,
    start_date: Optional[date] = None,
    days: int = 30,
    preferred_hours: Optional[List[int]] = None
) -> List[datetime]

detect_conflicts()

Detect scheduling conflicts.

async def detect_conflicts() -> List[Dict[str, Any]]

---

Scheduling Patterns

Daily Schedules

Videos created every day at specific time.

# Every day at 10 AM
await scheduler.create_daily_schedule(
    name="Daily Videos",
    topic_template="Daily Video - {date}",
    hour=10,
    minute=0
)

Weekly Schedules

Videos on specific days of week.

# Monday, Wednesday, Friday at 2 PM
await scheduler.create_weekly_schedule(
    name="MWF Videos",
    topic_template="Video - {weekday}",
    days=[DayOfWeek.MONDAY, DayOfWeek.WEDNESDAY, DayOfWeek.FRIDAY],
    hour=14,
    minute=0
)

Monthly Schedules

Videos on specific dates each month.

# 1st and 15th of each month at 9 AM
await scheduler.create_monthly_schedule(
    name="Bi-Monthly Videos",
    topic_template="Video - {month} {day}",
    days_of_month=[1, 15],
    hour=9,
    minute=0
)

Interval Schedules

Videos at regular intervals.

# Every 6 hours
await scheduler.create_interval_schedule(
    name="6-Hour Videos",
    topic_template="Video #{timestamp}",
    interval_hours=6
)

Cron Schedules

Advanced patterns using cron expressions.

# Monday-Friday at 9 AM and 5 PM
await scheduler.create_cron_schedule(
    name="Business Hours",
    topic_template="Video {time}",
    cron_expression="0 9,17 * * 1-5"
)

Common Cron Patterns:

• 0 10 * * * - Daily at 10 AM
• 0 14 * * 1,3,5 - Mon/Wed/Fri at 2 PM
• 0 9 1,15 * * - 1st and 15th at 9 AM
• 0 */6 * * * - Every 6 hours
• 0 9-17 * * 1-5 - Every hour 9AM-5PM, Mon-Fri

---

Best Practices

1. Schedule Management

DO:

• Use recurring schedules for consistent content
• Set reasonable gaps between videos (6+ hours)
• Use calendar to visualize schedule
• Monitor job statistics regularly

DON'T:

• Schedule too many videos simultaneously
• Ignore conflicts and warnings
• Skip error handling
• Forget to clean old jobs

2. Error Handling

Always wrap scheduler operations in try/except:

try:
    job_id = await scheduler.schedule_video(
        topic="My Video",
        scheduled_at=tomorrow
    )
except Exception as e:
    logger.error(f"Scheduling failed: {e}")
    # Handle error

3. Resource Management

Limit concurrent jobs to avoid overload:

config = ScheduleConfig(
    max_concurrent_jobs=2,  # Process 2 videos at once
    max_retries=3,
    check_interval_seconds=60
)

4. Monitoring

Implement monitoring for production:

async def monitor_scheduler(scheduler):
    while True:
        stats = scheduler.get_statistics()

        # Alert if too many failures
        if stats['statistics']['total_failed'] > 10:
            send_alert("High failure rate!")

        # Log progress
        logger.info(f"Active: {stats['active_jobs']}, "
                    f"Completed: {stats['statistics']['total_completed']}")

        await asyncio.sleep(300)  # Check every 5 minutes

5. Job Persistence

Jobs are automatically saved to disk. Load on startup:

scheduler = ContentScheduler(config=ScheduleConfig())
await scheduler.load_jobs()  # Restore jobs from disk
await scheduler.start()

6. Calendar Planning

Use calendar to avoid conflicts:

# Check for conflicts before scheduling
conflicts = await calendar.detect_conflicts()

if conflicts:
    logger.warning(f"Found {len(conflicts)} conflicts")
    # Resolve conflicts...

# Get optimal suggestions
suggestions = await calendar.suggest_optimal_slots(count=5)

# Schedule at optimal times
for slot_time in suggestions:
    await scheduler.schedule_video(
        topic=f"Video at {slot_time}",
        scheduled_at=slot_time
    )

---

Troubleshooting

Issue: Jobs not executing

Symptoms: Jobs stay in PENDING status

Solutions:

1. Check scheduler is started: await scheduler.start()
2. Verify scheduled time is in future
3. Check concurrent job limit
4. Review logs for errors

Issue: High failure rate

Symptoms: Many jobs in FAILED status

Solutions:

1. Check external service credentials (YouTube, Ollama)
2. Increase retry count and delay
3. Verify network connectivity
4. Check resource availability (disk, memory)

Issue: Conflicts detected

Symptoms: Calendar shows CONFLICT status

Solutions:

1. Increase min_gap_hours in CalendarConfig
2. Reduce max_videos_per_day
3. Use calendar.suggest_optimal_slots() for better times
4. Review and adjust blackout days

Issue: Recurring schedules not triggering

Symptoms: Recurring jobs not creating videos

Solutions:

1. Verify recurring scheduler is started: await recurring.start()
2. Check job is enabled: job.enabled == True
3. Verify cron expression syntax
4. Check timezone settings
5. Review APScheduler logs

Issue: Out of disk space

Symptoms: Video assembly fails

Solutions:

1. Implement automatic cleanup:

   await scheduler.clean_old_jobs(days=30)
   

2. Monitor disk usage
3. Compress or delete old videos
4. Use external storage

Debug Mode

Enable detailed logging:

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger('services.scheduler')
logger.setLevel(logging.DEBUG)

---

Performance Optimization

1. Concurrent Job Limits

config = ScheduleConfig(
    max_concurrent_jobs=2  # Adjust based on system resources
)

2. Retry Strategy

executor_config = ExecutorConfig(
    retry_strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
    base_retry_delay_seconds=60,
    max_retry_delay_seconds=3600
)

3. Caching

Jobs and results are automatically cached. Clean periodically:

# Clean jobs older than 30 days
await scheduler.clean_old_jobs(days=30)

# Clean execution history
await executor.clean_history()

---

Contributing

See main project README for contribution guidelines.

---

License

Copyright (c) 2024 Faceless YouTube Automation Platform. All Rights Reserved.

---

Support

For issues or questions:

• Check troubleshooting section
• Review examples in examples/scheduler_usage.py
• Check logs for error details
• Review test suite for usage patterns

---

Last Updated: December 2024
Version: 1.0.0