"""텔레그램 메시지 명령 → 에이전트 라우팅. 새 명령을 추가하려면 AGENT_COMMAND_MAP에 등록만 하면 됨.""" from typing import Optional def parse_command(text: str) -> Optional[tuple]: """슬래시 명령 파싱. 반환: (agent_id_or_None, command, args_list) 또는 None 예시: /stock news -> ("stock", "news", []) /status -> (None, "status", []) /music compose 잔잔한 피아노 -> ("music", "compose", ["잔잔한 피아노"]) """ if not text: return None text = text.strip() if not text.startswith("/"): return None parts = text[1:].split(maxsplit=2) if not parts: return None first = parts[0].lower() # 전역 명령 if first in ("status", "agents", "help"): return (None, first, parts[1:] if len(parts) > 1 else []) # 에이전트 명령: / [args...] if len(parts) < 2: return None agent_id = first command = parts[1].lower() args = [parts[2]] if len(parts) > 2 else [] return (agent_id, command, args) # 에이전트별 텔레그램 → 내부 command 매핑 # 텔레그램에서 친숙한 이름 -> (실제 on_command의 command, 기본 params) AGENT_COMMAND_MAP = { "stock": { "news": ("fetch_news", {}), "alerts": ("list_alerts", {}), "test": ("test_telegram", {}), }, "music": { "credits": ("credits", {}), # compose는 인자 필요 — 아래 특수 케이스에서 처리 }, "realestate": { "matches": ("fetch_matches", {}), "dashboard": ("dashboard", {}), }, } def resolve_agent_command(agent_id: str, command: str, args: list) -> Optional[tuple]: """(internal_command, params) 반환. 매핑 없으면 None.""" mapping = AGENT_COMMAND_MAP.get(agent_id, {}).get(command) if mapping is None: # 특수 케이스: music compose if agent_id == "music" and command == "compose" and args: return ("compose", {"prompt": " ".join(args)}) return None internal_cmd, base_params = mapping params = dict(base_params) if args: # args가 있으면 첫 번째(합쳐진 나머지)를 message로 자동 주입 params["message"] = " ".join(args) return (internal_cmd, params) HELP_TEXT = """🤖 Agent Office 텔레그램 명령 전역 /status — 모든 에이전트 상태 /agents — 에이전트 목록 /help — 이 도움말 📈 주식 트레이더 /stock news — 뉴스 AI 요약 실행 /stock alerts — 알람 목록 /stock test — 텔레그램 테스트 🎵 음악 프로듀서 /music credits — Suno 크레딧 조회 /music compose <프롬프트> — 작곡 시작 🏢 청약 애널리스트 /realestate matches — 신규 매칭 조회 후 알림 전송 /realestate dashboard — 청약 현황 요약 """