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
72 lines
2.1 KiB
Python
72 lines
2.1 KiB
Python
from celery import Celery
|
|
from celery.schedules import crontab
|
|
from app.services.bot_service import BotService
|
|
import os
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from typing import List, Dict, Optional
|
|
|
|
celery = Celery(
|
|
'tasks',
|
|
broker=os.getenv("CELERY_BROKER_URL", "redis://localhost:6379/0"),
|
|
backend=os.getenv("CELERY_RESULT_BACKEND", "redis://localhost:6379/0"),
|
|
)
|
|
|
|
bot_service = BotService(
|
|
os.getenv("TELEGRAM_BOT_TOKEN", ""),
|
|
os.getenv("TELEGRAM_CHAT_ID", "")
|
|
)
|
|
|
|
# Load schedules from JSON
|
|
def load_schedules() -> List[Dict]:
|
|
schedules_path = "/app/schedules.json"
|
|
if os.path.exists(schedules_path):
|
|
with open(schedules_path, "r") as f:
|
|
return json.load(f)
|
|
return []
|
|
|
|
# Save schedules to JSON
|
|
def save_schedules(schedules: List[Dict]):
|
|
with open("/app/schedules.json", "w") as f:
|
|
json.dump(schedules, f, indent=2)
|
|
|
|
@celery.task
|
|
def send_scheduled_message(schedule_id: int):
|
|
"""Send scheduled message to Telegram."""
|
|
schedules = load_schedules()
|
|
schedule = next((s for s in schedules if s['id'] == schedule_id), None)
|
|
|
|
if not schedule:
|
|
print(f"Schedule {schedule_id} not found")
|
|
return
|
|
|
|
if schedule.get('enabled', True):
|
|
# Send to bot service
|
|
try:
|
|
bot_service.send_scheduled_message(schedule)
|
|
print(f"Sent scheduled message: {schedule['time']}")
|
|
except Exception as e:
|
|
print(f"Error sending scheduled message: {e}")
|
|
|
|
@celery.task
|
|
def generate_daily_content():
|
|
"""Generate daily content for all books."""
|
|
try:
|
|
bot_service.generate_daily_content()
|
|
print("Sent daily content")
|
|
except Exception as e:
|
|
print(f"Error generating daily content: {e}")
|
|
|
|
# Periodic tasks
|
|
celery.conf.beat_schedule = {
|
|
'send-scheduled-messages': {
|
|
'task': 'app.tasks.celery_tasks.send_scheduled_message',
|
|
'schedule': crontab(minute=0), # Every hour
|
|
'args': (None,), # Process all enabled schedules
|
|
},
|
|
'generate-daily-content': {
|
|
'task': 'app.tasks.celery_tasks.generate_daily_content',
|
|
'schedule': crontab(hour=8, minute=0), # Daily at 8 AM
|
|
},
|
|
}
|