5c9b13ed19
- Marked shrimp_data subproject as dirty - Added pre-tool hook configuration to settings.local.json
237 lines
8.6 KiB
Python
237 lines
8.6 KiB
Python
"""
|
||
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()
|