lgcosta
7af7ba530d
The image context was backend/, so COPY frontend/ failed. Build from the repo root, place Python modules on the import path, and pin packages that exist on PyPI. The web process no longer loads the LLM at startup.
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
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)
|
|
|
|
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
|