scheduling-system by forever19735/garbage

When to use this skill

Use this skill when you need to:

Set up automated scheduled tasks and reminders
Implement cron-like job scheduling
Manage per-group independent schedules
Calculate rotation cycles based on natural weeks
Handle timezone-aware scheduling (Asia/Taipei)
Dynamically add/remove scheduled jobs
Implement weekly member rotation logic

How to use it

Core Components

1. APScheduler Setup

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import pytz

# Initialize scheduler
scheduler = BackgroundScheduler(timezone=pytz.timezone('Asia/Taipei'))
scheduler.start()

2. Data Structures

Group Schedule Configuration:

group_schedules = {
    "C群組ID1": {
        "days": "mon,wed,fri",  # or ["mon", "wed", "fri"]
        "hour": 17,
        "minute": 0
    },
    "C群組ID2": {
        "days": "tue,thu",
        "hour": 9,
        "minute": 30
    }
}

Member Rotation Structure:

groups = {
    "C群組ID1": {
        "1": ["Alice", "Bob"],
        "2": ["Charlie"],
        "3": ["David", "Eve"]
    }
}

3. Natural Week Calculation

Calculate current week based on Monday-Sunday cycles:

from datetime import date, timedelta

def get_current_week(group_id, base_date):
    """
    Calculate current week number based on natural weeks (Mon-Sun)
    
    Args:
        group_id: Group identifier
        base_date: Reference date for week 1
    
    Returns:
        int: Current week number (1-based)
    """
    today = date.today()
    
    # Get Monday of base_date's week
    base_monday = base_date - timedelta(days=base_date.weekday())
    
    # Get Monday of today's week
    today_monday = today - timedelta(days=today.weekday())
    
    # Calculate weeks difference
    weeks_diff = (today_monday - base_monday).days // 7
    
    # Calculate current week (循環)
    total_weeks = len(groups[group_id])
    current_week = (weeks_diff % total_weeks) + 1
    
    return current_week

4. Per-Day Member Assignment

Support different members for different weekdays within same week:

def get_current_day_member(group_id, target_date=None):
    """
    Get the member responsible for today based on broadcast days
    
    Args:
        group_id: Group identifier
        target_date: Target date (default: today)
    
    Returns:
        str: Member name or None if not a broadcast day
    """
    if target_date is None:
        target_date = date.today()
    
    # Get week's members
    current_members = get_current_group(group_id)
    if not current_members:
        return None
    
    # Get broadcast days for this group
    schedule = group_schedules.get(group_id, {})
    broadcast_days = schedule.get('days', '')
    
    if isinstance(broadcast_days, str):
        broadcast_days = [d.strip() for d in broadcast_days.split(',')]
    
    # Day mapping
    day_mapping = {
        'mon': 0, 'tue': 1, 'wed': 2, 'thu': 3,
        'fri': 4, 'sat': 5, 'sun': 6
    }
    
    # Get today's weekday
    today_weekday = target_date.weekday()
    today_day_name = [k for k, v in day_mapping.items() if v == today_weekday][0]
    
    if today_day_name not in broadcast_days:
        return None  # Not a broadcast day
    
    # Assign member based on broadcast day index
    day_index = broadcast_days.index(today_day_name)
    member_index = day_index % len(current_members)
    
    return current_members[member_index]

5. Dynamic Job Scheduling

Add/Update Group Schedule:

def update_group_schedule(group_id, days_str, hour, minute):
    """
    Update or create schedule for a specific group
    
    Args:
        group_id: Group identifier
        days_str: Comma-separated days (e.g., "mon,wed,fri")
        hour: Hour (0-23)
        minute: Minute (0-59)
    """
    # Save schedule to database
    group_schedules[group_id] = {
        'days': days_str,
        'hour': hour,
        'minute': minute
    }
    save_group_schedules(group_schedules)
    
    # Remove old job if exists
    job_id = f"reminder_{group_id}"
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)
    
    # Add new job
    days = [d.strip().lower() for d in days_str.split(',')]
    trigger = CronTrigger(
        day_of_week=','.join(days),
        hour=hour,
        minute=minute,
        timezone=pytz.timezone('Asia/Taipei')
    )
    
    scheduler.add_job(
        func=send_reminder,
        trigger=trigger,
        id=job_id,
        args=[group_id],
        replace_existing=True
    )

Remove Group Schedule:

def remove_group_schedule(group_id):
    """Remove scheduled job for a group"""
    job_id = f"reminder_{group_id}"
    if scheduler.get_job(job_id):
        scheduler.remove_job(job_id)
    
    if group_id in group_schedules:
        del group_schedules[group_id]
        save_group_schedules(group_schedules)

6. Reminder Function

def send_reminder(group_id):
    """
    Send reminder message to a specific group
    Called by scheduler at configured times
    """
    try:
        # Get today's responsible member(s)
        member = get_current_day_member(group_id)
        
        if not member:
            print(f"No broadcast for {group_id} today")
            return
        
        # Format message
        today = date.today()
        weekday_names = ['週一', '週二', '週三', '週四', '週五', '週六', '週日']
        weekday = weekday_names[today.weekday()]
        
        message = f"🗑️ 今天 {today.strftime('%m/%d')} ({weekday}) 輪到 {member} 收垃圾！"
        
        # Send via LINE Messaging API
        messaging_api.push_message(
            PushMessageRequest(
                to=group_id,
                messages=[TextMessage(text=message)]
            )
        )
        
        print(f"✅ Sent reminder to {group_id}: {member}")
    except Exception as e:
        print(f"❌ Failed to send reminder to {group_id}: {e}")

Best Practices

Timezone Awareness
- Always use Asia/Taipei timezone for Taiwan
- Use pytz for timezone handling
- Be consistent across all datetime operations
Job ID Naming
- Use descriptive, unique job IDs: reminder_{group_id}
- Makes it easy to find and remove specific jobs
Graceful Degradation
- Handle missing schedule configurations
- Don't crash if a group has no members
- Log errors but continue operation

Testing Schedules

# Get next run time for a job
job = scheduler.get_job(f"reminder_{group_id}")
if job:
    next_run = job.next_run_time
    print(f"Next reminder: {next_run}")

Scheduler Lifecycle

# Startup: Load all group schedules
def init_all_schedules():
    for group_id, config in group_schedules.items():
        update_group_schedule(
            group_id,
            config['days'],
            config['hour'],
            config['minute']
        )

# Shutdown: Clean up
def shutdown():
    scheduler.shutdown(wait=True)

Common Patterns

Weekly vs Daily Rotation

# Weekly: Same members all week
current_members = get_current_group(group_id)

# Daily: Different member each broadcast day
today_member = get_current_day_member(group_id)

Flexible Schedule Commands

# @time 18:00
# @day mon,wed,fri
# @cron mon,wed,fri 18 00

def handle_cron_command(group_id, days_str, hour, minute):
    if update_group_schedule(group_id, days_str, hour, minute):
        return f"✅ 已設定推播時間：{days_str} {hour:02d}:{minute:02d}"
    else:
        return "❌ 設定失敗"

scheduling-system