lgcosta
8eeb6b89f4
Usa o modelo multilingual, ignora trechos curtos e o template do Llama 3.2 para a resposta sair em uma frase completa. O chat deixa a API livre durante a geração, o upload indexa o PDF e os agendamentos não são gravados duas vezes.
79 lines
2.8 KiB
Python
79 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)
|
|
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
|