""" Telegram 봇 - FE AI 표준화 프로젝트 팀 업무 지원 봇 기능: - 기본 에코 - 스탠드업 리마인더 자동 발송 (매일 09:30 KST) - 로드맵 태스크 상태 조회 필요 패키지: pip install python-telegram-bot[job-queue] 환경변수: TELEGRAM_BOT_TOKEN - BotFather에서 발급받은 토큰 (필수) STANDUP_CHAT_IDS - 스탠드업 리마인더를 받을 채팅 ID 목록 (쉼표 구분, 선택) 예) STANDUP_CHAT_IDS=123456789,987654321 """ import json import logging import os from datetime import time from pathlib import Path from typing import Final from telegram import Update from telegram.ext import ( Application, CommandHandler, ContextTypes, MessageHandler, filters, ) # ── 로깅 설정 ───────────────────────────────────────────── logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO, ) logger = logging.getLogger(__name__) # ── 설정 ───────────────────────────────────────────────── BOT_TOKEN: Final[str] = os.environ.get("TELEGRAM_BOT_TOKEN", "") TASKS_FILE: Final[Path] = Path(__file__).parent / "shrimp_data" / "tasks.json" # 스탠드업 리마인더 발송 시간 (KST = UTC+9) STANDUP_HOUR: Final[int] = 9 STANDUP_MINUTE: Final[int] = 30 # 환경변수로 사전 등록할 채팅 ID 목록 STANDUP_CHAT_IDS: Final[list[int]] = [ int(cid) for cid in os.environ.get("STANDUP_CHAT_IDS", "").split(",") if cid.strip() ] STATUS_LABELS: Final[dict[str, str]] = { "pending": "⏳ 대기", "in-progress": "🔄 진행중", "completed": "✅ 완료", "deferred": "⏸ 보류", "cancelled": "❌ 취소", } # ── 유틸리티 ────────────────────────────────────────────── def load_tasks() -> list[dict]: """shrimp_data/tasks.json에서 태스크 목록을 로드한다.""" if not TASKS_FILE.exists(): return [] with TASKS_FILE.open(encoding="utf-8") as f: data = json.load(f) return data.get("tasks", []) def format_task_line(task: dict) -> str: """태스크 한 줄 요약을 반환한다.""" status = STATUS_LABELS.get(task.get("status", ""), task.get("status", "")) name = task.get("name", "(이름 없음)") return f"{status} {name}" # ── 커맨드 핸들러 ───────────────────────────────────────── async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """/start — 봇 소개 및 명령어 안내""" text = ( "안녕하세요! FE AI 표준화 프로젝트 봇입니다. 👋\n\n" "사용 가능한 명령어\n" "/tasks — 전체 태스크 목록\n" "/tasks\\_pending — 대기 중인 태스크\n" "/tasks\\_done — 완료된 태스크\n" "/standup — 스탠드업 리마인더 구독\n" "/help — 도움말" ) await update.message.reply_text(text, parse_mode="Markdown") async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """/help — 도움말 (start와 동일)""" await start(update, context) async def tasks_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """/tasks — 전체 태스크를 상태별로 보여준다.""" all_tasks = load_tasks() if not all_tasks: await update.message.reply_text("태스크 데이터를 불러올 수 없습니다.") return total = len(all_tasks) done = sum(1 for t in all_tasks if t.get("status") == "completed") lines = [f"📋 *전체 태스크 ({done}/{total} 완료)*\n"] for task in all_tasks: lines.append(format_task_line(task)) await update.message.reply_text("\n".join(lines), parse_mode="Markdown") async def tasks_pending_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """/tasks_pending — 대기 중인 태스크만 조회한다.""" pending = [t for t in load_tasks() if t.get("status") == "pending"] if not pending: await update.message.reply_text("✅ 대기 중인 태스크가 없습니다!") return lines = [f"⏳ *대기 중인 태스크 ({len(pending)}개)*\n"] for task in pending: lines.append(format_task_line(task)) await update.message.reply_text("\n".join(lines), parse_mode="Markdown") async def tasks_done_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """/tasks_done — 완료된 태스크만 조회한다.""" done = [t for t in load_tasks() if t.get("status") == "completed"] if not done: await update.message.reply_text("아직 완료된 태스크가 없습니다.") return lines = [f"✅ *완료된 태스크 ({len(done)}개)*\n"] for task in done: lines.append(format_task_line(task)) await update.message.reply_text("\n".join(lines), parse_mode="Markdown") async def standup_subscribe(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """/standup — 이 채팅을 스탠드업 리마인더 수신 대상으로 등록한다.""" chat_id = update.effective_chat.id registered: set[int] = context.bot_data.setdefault("standup_chats", set()) if chat_id in registered: await update.message.reply_text("이미 등록된 채팅입니다.") return registered.add(chat_id) await update.message.reply_text( f"✅ 스탠드업 리마인더에 등록되었습니다.\n" f"매일 *{STANDUP_HOUR:02d}:{STANDUP_MINUTE:02d} KST*에 알림을 보내드립니다.", parse_mode="Markdown", ) # ── 에코 핸들러 ─────────────────────────────────────────── async def echo(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """명령어가 아닌 텍스트 메시지를 그대로 반환한다.""" await update.message.reply_text(update.message.text) # ── 스케줄 작업 ─────────────────────────────────────────── async def send_standup_reminder(context: ContextTypes.DEFAULT_TYPE) -> None: """등록된 모든 채팅에 스탠드업 리마인더를 발송한다.""" message = ( "🌅 *스탠드업 리마인더*\n\n" "오늘의 스탠드업을 시작할 시간입니다!\n\n" "1️⃣ *어제 한 일*:\n" "2️⃣ *오늘 할 일*:\n" "3️⃣ *블로커*:" ) # 환경변수 등록 + 런타임 /standup 구독 채팅 합산 target_chats: set[int] = set(STANDUP_CHAT_IDS) | context.bot_data.get( "standup_chats", set() ) for chat_id in target_chats: try: await context.bot.send_message( chat_id=chat_id, text=message, parse_mode="Markdown", ) logger.info("스탠드업 리마인더 발송 완료 (chat_id=%s)", chat_id) except Exception as e: logger.error("스탠드업 리마인더 발송 실패 (chat_id=%s): %s", chat_id, e) # ── 앱 빌드 & 실행 ──────────────────────────────────────── def main() -> None: if not BOT_TOKEN: raise ValueError( "TELEGRAM_BOT_TOKEN 환경변수를 설정해주세요.\n" "예) export TELEGRAM_BOT_TOKEN=your_token_here" ) app = Application.builder().token(BOT_TOKEN).build() # 커맨드 핸들러 등록 app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("help", help_command)) app.add_handler(CommandHandler("tasks", tasks_command)) app.add_handler(CommandHandler("tasks_pending", tasks_pending_command)) app.add_handler(CommandHandler("tasks_done", tasks_done_command)) app.add_handler(CommandHandler("standup", standup_subscribe)) # 에코 핸들러 (명령어가 아닌 텍스트) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo)) # 매일 09:30 KST (UTC 00:30)에 스탠드업 리마인더 발송 app.job_queue.run_daily( send_standup_reminder, time=time(hour=0, minute=30), # UTC 기준 (KST -9h) ) logger.info("봇 시작 — polling 모드") app.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()