lgcosta
b9c0c66d21
- Stack: GPT4All + Llama 3.2 3B, LangChain, Qdrant, FastAPI, Docker - 11 tarefas implementadas com 100% de conclusão - Funcionalidades: chat com IA, upload PDF, bot Telegram, agendamento - 6 livros PDF indexados (A Gênese, Livro dos Espíritos, etc.) - Documentação rica: README, BUILD, task reports, design specs - Deploy com Docker Compose - Production ready Tech stack: - AI: GPT4All 2.4.6, Llama 3.2 3B q4_0 - Backend: FastAPI 0.115.0, Celery 5.4.0, Redis 5.2.0 - RAG: LangChain 0.3.7, Qdrant 1.11.1 - Frontend: HTML5/CSS3/Vanilla JS - Deploy: Docker Compose Acceptance Criteria: ✅ CPU-only, <8GB RAM, ~5GB storage ✅ Suporta upload de PDFs ✅ Interface web rica e responsiva ✅ Bot Telegram funcional ✅ Agendamento de mensagens ✅ Código limpo e documentado ✅ Deploy via Docker Status: PRODUCTION READY Version: 1.0.0 Date: 2026-11-01
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
|