shinalok
0b50444e43
- IDEA-2 스마트 알림: td_reminders 테이블, set_reminder/list_reminders 도구,
SchedulerService(asyncio 60초 루프, D-7/D-1/D-0 Telegram push),
FastAPI lifespan 연동, GET /reminders/{user_id} 엔드포인트
- IDEA-1 대화 기반 RAG: IngestionService.store_text() 추가,
AgentService._maybe_index_conversation() — 응답 후 LLM 판단 → Qdrant 저장
(CONV_RAG_ENABLED=true 활성화, background task로 응답 속도 무관)
- IDEA-5 CRAG: AgentState에 crag_fallback_used 플래그 추가,
crag_check LangGraph 노드 — search_documents 결과 없으면 web_search 자동 주입,
route_after_crag으로 fallback 1회 루프 제어 (CRAG_ENABLED=true 활성화)
- IDEA-7 RAG Auto-Eval: eval/auto_tune.py — API 서버 없이 파라미터 조합별
context_precision/recall 비교, 최적 설정 추천
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import date
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_NOTIFY_PREFIX = {
|
|
"d0": "🔔 오늘 일정",
|
|
"d1": "📅 내일 일정",
|
|
"d7": "📆 7일 후 일정",
|
|
}
|
|
|
|
|
|
class SchedulerService:
|
|
"""asyncio 태스크 기반 알림 스케줄러.
|
|
|
|
매 60초마다 td_reminders를 확인해 D-7/D-1/D-0 Telegram 알림을 발송한다.
|
|
TELEGRAM_BOT_TOKEN이 비어 있으면 발송 없이 로그만 출력한다.
|
|
"""
|
|
|
|
def __init__(self, reminder_repo, bot_token: str, user_map_json: str):
|
|
self._repo = reminder_repo
|
|
self._token = bot_token
|
|
try:
|
|
self._user_map: dict[str, str] = json.loads(user_map_json) if user_map_json else {}
|
|
except Exception:
|
|
self._user_map = {}
|
|
self._task: asyncio.Task | None = None
|
|
|
|
def start(self) -> None:
|
|
self._task = asyncio.create_task(self._loop())
|
|
logger.info("[Scheduler] 알림 스케줄러 시작 (60초 간격)")
|
|
|
|
def shutdown(self) -> None:
|
|
if self._task:
|
|
self._task.cancel()
|
|
logger.info("[Scheduler] 알림 스케줄러 종료")
|
|
|
|
async def _loop(self) -> None:
|
|
while True:
|
|
try:
|
|
await self._check_reminders()
|
|
except Exception as e:
|
|
logger.error(f"[Scheduler] 알림 확인 중 오류: {e}")
|
|
await asyncio.sleep(60)
|
|
|
|
async def _check_reminders(self) -> None:
|
|
today = date.today()
|
|
reminders = self._repo.get_due(today)
|
|
for r in reminders:
|
|
notify_type = r.get("notify_type")
|
|
if not notify_type:
|
|
continue
|
|
text = self._format_message(r["message"], notify_type, r["remind_date"])
|
|
chat_id = self._user_map.get(r["user_id"])
|
|
if self._token and chat_id:
|
|
try:
|
|
self._send_telegram(chat_id, text)
|
|
except Exception as e:
|
|
logger.error(f"[Scheduler] Telegram 발송 실패 (user={r['user_id']}): {e}")
|
|
continue
|
|
else:
|
|
logger.info(f"[Scheduler] 알림(Telegram 미설정): {text}")
|
|
try:
|
|
self._repo.mark_sent(r["id"], notify_type)
|
|
except Exception as e:
|
|
logger.error(f"[Scheduler] mark_sent 실패: {e}")
|
|
|
|
def _format_message(self, message: str, notify_type: str, remind_date) -> str:
|
|
prefix = _NOTIFY_PREFIX.get(notify_type, "알림")
|
|
return f"[율봇 알림] {prefix}\n날짜: {remind_date}\n{message}"
|
|
|
|
def _send_telegram(self, chat_id: str, text: str) -> None:
|
|
url = f"https://api.telegram.org/bot{self._token}/sendMessage"
|
|
data = urllib.parse.urlencode({"chat_id": chat_id, "text": text}).encode()
|
|
req = urllib.request.Request(url, data=data, method="POST")
|
|
with urllib.request.urlopen(req, timeout=10):
|
|
pass
|