import json import os from datetime import datetime from typing import List, Dict, Optional class ScheduleManager: def __init__(self, schedules_path: str = "/app/schedules.json"): self.schedules_path = schedules_path self.schedules = self._load_schedules() def _load_schedules(self) -> List[Dict]: """Load schedules from JSON file.""" if os.path.exists(self.schedules_path): try: with open(self.schedules_path, 'r') as f: return json.load(f) except json.JSONDecodeError: return [] return [] def save_schedules(self): """Save schedules to JSON file.""" os.makedirs(os.path.dirname(self.schedules_path), exist_ok=True) with open(self.schedules_path, 'w') as f: json.dump(self.schedules, f, indent=2) f.write("\n") def add_schedule(self, schedule: Dict) -> Dict: """Add a new schedule.""" schedule_id = max([s['id'] for s in self.schedules], default=0) + 1 new_schedule = { 'id': schedule_id, 'time': schedule['time'], 'book': schedule.get('book'), 'enabled': schedule.get('enabled', True), } self.schedules.append(new_schedule) self.save_schedules() return new_schedule def update_schedule(self, schedule_id: int, updates: Dict) -> Optional[Dict]: """Update an existing schedule.""" for i, schedule in enumerate(self.schedules): if schedule['id'] == schedule_id: self.schedules[i].update(updates) self.save_schedules() return self.schedules[i] return None def delete_schedule(self, schedule_id: int) -> bool: """Delete a schedule.""" for i, schedule in enumerate(self.schedules): if schedule['id'] == schedule_id: self.schedules.pop(i) self.save_schedules() return True return False def get_schedule(self, schedule_id: int) -> Optional[Dict]: """Get a specific schedule.""" for schedule in self.schedules: if schedule['id'] == schedule_id: return schedule return None def list_schedules(self) -> List[Dict]: """List all schedules.""" return self.schedules def get_enabled_schedules(self) -> List[Dict]: """Get all enabled schedules.""" return [s for s in self.schedules if s.get('enabled', True)] def is_schedule_ready(self, schedule: Dict) -> bool: """Check if a schedule is ready to be sent.""" schedule_time = datetime.fromisoformat(schedule['time'].replace('Z', '+00:00')) current_time = datetime.now() return current_time >= schedule_time