🔧 chore: Update shrimp_data subproject status and add pre-tool hook configuration
- Marked shrimp_data subproject as dirty - Added pre-tool hook configuration to settings.local.json
This commit is contained in:
hyeonggil
236
telegram_bot.py
Normal file
236
telegram_bot.py
Normal file
@@ -0,0 +1,236 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user