# 맨 위 import에 추가 import re from html import escape import os import asyncio import logging from typing import Dict from telegram import Update, InlineQueryResultArticle, InputTextMessageContent, ReplyKeyboardMarkup, KeyboardButton from telegram.constants import ParseMode from telegram.ext import ( Application, ApplicationBuilder, CommandHandler, ContextTypes, MessageHandler, InlineQueryHandler, filters, ) from app.services.supabase_service import logger from app.services.subscription_service import toggle_subscription, is_subscribed from app.services.search_service import search_people # 기존 검색 서비스 사용 # ----- /sync 에서 호출할 크롤러들 ----- from app.crawler.motie_org_sync_n8n import run_once as motie_org_sync from app.crawler.moef_org_sync_n8n import run_once as moef_org_sync from app.crawler.motie_n8n import run_once as motie_id_sync from app.crawler.moef_n8n import run_once as moef_id_sync from typing import Optional # ----------------------------------- # 환경변수 로딩은 supabase_service에서 처리된다고 가정 # TELEGRAM_BOT_TOKEN 도 systemd 환경파일에 설정되어 있음 # 전역 락: /sync 중복실행 방지 SYNC_LOCK = asyncio.Lock() # ---------- 기본 핸들러들 ---------- async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): text = ( "안녕하세요! 대정부·전력그룹사 봇입니다.\n\n" "명령어 안내:\n" "/start - 도움말\n" "/subscribe - 인사 알림 구독/해지\n" "/search - 직원 검색(산업부·기재부 통합)\n" "팁: 채팅창에서 @봇아이디 키워드 로 인라인 검색이 가능합니다." ) kb = ReplyKeyboardMarkup( [[KeyboardButton("검색"), KeyboardButton("구독")]], resize_keyboard=True, one_time_keyboard=False, selective=False, ) await update.message.reply_text(text, reply_markup=kb) async def subscribe(update: Update, context: ContextTypes.DEFAULT_TYPE): chat_id = update.effective_chat.id now_on = await toggle_subscription(chat_id) if now_on: await update.message.reply_text( "구독이 설정되었습니다.\n" "- 산업부/기재부 인사발령\n" "- 전력그룹사 인사동정\n" "새 글이 올라오면 알림을 보내드릴게요." ) else: await update.message.reply_text("구독이 해지되었습니다. 다시 /subscribe 하면 구독돼요.") # /search 흐름: /search 치면 키워드 요청 -> 평문 메시지에서 검색 실행 SEARCH_WAITING = "waiting_keyword" async def search_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE): context.user_data[SEARCH_WAITING] = True await update.message.reply_text("검색할 키워드를 입력해주세요. (이름/부서/직위/업무)") async def text_router(update: Update, context: ContextTypes.DEFAULT_TYPE): msg = update.message.text.strip() # 키보드 단축 버튼 라우팅 if msg in ("검색",): return await search_cmd(update, context) if msg in ("구독",): return await subscribe(update, context) if context.user_data.get(SEARCH_WAITING): context.user_data[SEARCH_WAITING] = False # 검색 실행 rows = await asyncio.get_event_loop().run_in_executor( None, lambda: search_people(msg) ) if not rows: await update.message.reply_text("검색 결과가 없습니다.") return msg_text = format_staff_results(rows[:10]) await update.message.reply_text( msg_text, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ) return # ---------- 여기부터 /sync (권한 체크 제거 버전) ---------- async def sync_all(update: Update, context: ContextTypes.DEFAULT_TYPE): # 누구나 실행 가능 if SYNC_LOCK.locked(): await update.message.reply_text("이미 동기화가 진행 중입니다. 잠시만 기다려주세요.") return await update.message.reply_text("동기화를 시작합니다.\n- MOTIE/MOEF 조직도(org)\n- MOTIE/MOEF 인사발령(id)") async with SYNC_LOCK: loop = asyncio.get_event_loop() # 블로킹 작업은 스레드 풀에서 실행 motie_org = await loop.run_in_executor(None, motie_org_sync) moef_org = await loop.run_in_executor(None, moef_org_sync) motie_id = await loop.run_in_executor(None, motie_id_sync) moef_id = await loop.run_in_executor(None, moef_id_sync) # 요약 메세지 생성 # org는 int 리턴(삽입건수), id는 dict 리턴({"inserted": x, "sent": y}) motie_id_ins = motie_id.get("inserted", 0) if isinstance(motie_id, dict) else motie_id motie_id_sent = motie_id.get("sent", 0) if isinstance(motie_id, dict) else 0 moef_id_ins = moef_id.get("inserted", 0) if isinstance(moef_id, dict) else moef_id moef_id_sent = moef_id.get("sent", 0) if isinstance(moef_id, dict) else 0 summary = ( "동기화 완료 ✅\n\n" f"- MOTIE 조직도: {motie_org}건 갱신\n" f"- MOEF 조직도: {moef_org}건 갱신\n" f"- MOTIE 인사: {motie_id_ins}건 신규, {motie_id_sent}건 알림 발송\n" f"- MOEF 인사: {moef_id_ins}건 신규, {moef_id_sent}건 알림 발송\n" ) await update.message.reply_text(summary, parse_mode=ParseMode.HTML) # ---------- 앱 빌드 ---------- # 기존: def build_app(token: str) -> Application: # 수정: 토큰을 옵션으로 받아서 없으면 환경변수에서 읽도록 def build_app(token: Optional[str] = None) -> Application: if token is None: token = os.environ.get("BOT_TOKEN", "") if not token: raise RuntimeError("BOT_TOKEN 이 설정되어 있지 않습니다.") app: Application = ApplicationBuilder().token(token).build() app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("subscribe", subscribe)) app.add_handler(CommandHandler("search", search_cmd)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, text_router)) app.add_handler(InlineQueryHandler(inline_query)) return app def _tel_href(raw: str) -> str: # tel: 링크는 숫자/플러스만 남김 digits = re.sub(r"[^0-9+]", "", raw or "") return f"tel:{digits}" if digits else "" def format_staff_results(rows: list[dict]) -> str: """ rows: [{name, position, department, task, phone}, ...] 출력 형식: 1) 이름 / position department task 원문전화번호 """ lines = [] def _src_label(src: str | None) -> str: s = (src or "").upper() return "산업부" if s == "MOTIE" else ("기재부" if s == "MOEF" else s or "-") for i, r in enumerate(rows, start=1): src = _src_label(r.get("source")) name = escape(r.get("name") or "") position = escape(r.get("position") or "") department = escape(r.get("department") or "") task = escape(r.get("task") or "") phone_display = r.get("phone") or "" phone_href = _tel_href(phone_display) phone_display_escaped = escape(phone_display) block = [ f"{i}) [{src}] {name} / {position}".strip(" /"), f"🏢 {department}" if department else "", f"📝 {task}" if task else "", (f'📞 {phone_display_escaped}' if phone_href else (f"📞 {phone_display_escaped}" if phone_display_escaped else "")), ] # 빈 줄 제거 lines.append("\n".join([b for b in block if b])) return "\n\n".join(lines) # ------------- 인라인 검색 핸들러 ------------- async def inline_query(update: Update, context: ContextTypes.DEFAULT_TYPE): query = (update.inline_query.query or "").strip() if not query: await update.inline_query.answer([], cache_time=10, is_personal=True) return loop = asyncio.get_event_loop() rows = await loop.run_in_executor(None, lambda: search_people(query, limit=10)) results = [] if not rows: res = InlineQueryResultArticle( id="no_results", title="검색 결과 없음", description=f"'{query}'에 해당하는 결과가 없습니다.", input_message_content=InputTextMessageContent(text=f"검색 결과 없음: {query}"), ) results.append(res) else: def _src_label(src: str | None) -> str: s = (src or "").upper() return "산업부" if s == "MOTIE" else ("기재부" if s == "MOEF" else s or "-") for i, r in enumerate(rows): src = _src_label(r.get("source")) title = f"[{src}] {(r.get('name') or '').strip()} / {(r.get('position') or '').strip()}".strip(" /") desc = " | ".join([ (r.get("department") or "").strip(), (r.get("task") or "").strip(), (r.get("phone") or "").strip(), ]).strip(" |") msg = format_staff_results([r]) results.append( InlineQueryResultArticle( id=f"staff_{i}", title=title or "직원", description=desc, input_message_content=InputTextMessageContent( message_text=msg, parse_mode=ParseMode.HTML, disable_web_page_preview=True, ), ) ) await update.inline_query.answer(results, cache_time=10, is_personal=True) # 파일 맨 아래 __main__ 블록도 이렇게 맞춰두면 독립 실행 시에도 OK if __name__ == "__main__": app = build_app() # 환경변수에서 읽음 logger.info("Telegram bot starting (polling)...") app.run_polling(close_loop=False)