""" Asset Text Parser ================== 사용자가 입력한 텍스트에서 포인트/마일리지 정보를 자동 추출합니다. 예시 입력: - "대한항공 45,000 마일" - "AMEX MR 50000점" - "체이스 UR 30,000" """ import re from typing import Dict, Any, List, Optional from datetime import datetime from .config import POINT_VALUATIONS # ============================================================================= # 프로그램 별칭 매핑 # ============================================================================= PROGRAM_ALIASES = { # 한국 항공 "대한항공": "KOREAN_AIR", "대한": "KOREAN_AIR", "kal": "KOREAN_AIR", "korean air": "KOREAN_AIR", "스카이패스": "KOREAN_AIR", "아시아나": "ASIANA", "oz": "ASIANA", "asiana": "ASIANA", # 미국 항공 "delta": "DELTA_SKYMILES", "델타": "DELTA_SKYMILES", "skymiles": "DELTA_SKYMILES", "united": "UNITED_MILEAGEPLUS", "유나이티드": "UNITED_MILEAGEPLUS", "mileageplus": "UNITED_MILEAGEPLUS", "american": "AA_AADVANTAGE", "aa": "AA_AADVANTAGE", "아메리칸": "AA_AADVANTAGE", "aadvantage": "AA_AADVANTAGE", # 신용카드 포인트 "amex": "AMEX_MR", "아멕스": "AMEX_MR", "mr": "AMEX_MR", "membership rewards": "AMEX_MR", "chase": "CHASE_UR", "체이스": "CHASE_UR", "ur": "CHASE_UR", "ultimate rewards": "CHASE_UR", "citi": "CITI_TYP", "시티": "CITI_TYP", "typ": "CITI_TYP", "thankyou": "CITI_TYP", "capital one": "CAPITAL_ONE", "캐피탈원": "CAPITAL_ONE", # 호텔 포인트 "marriott": "MARRIOTT_BONVOY", "메리어트": "MARRIOTT_BONVOY", "본보이": "MARRIOTT_BONVOY", "bonvoy": "MARRIOTT_BONVOY", "hilton": "HILTON_HONORS", "힐튼": "HILTON_HONORS", "honors": "HILTON_HONORS", "ihg": "IHG_REWARDS", "hyatt": "HYATT_WOH", "하얏트": "HYATT_WOH", "woh": "HYATT_WOH", "world of hyatt": "HYATT_WOH", } # ============================================================================= # 파싱 함수 # ============================================================================= def parse_asset_text(text: str) -> List[Dict[str, Any]]: """ 텍스트에서 포인트/마일리지 정보를 추출합니다. Args: text: 사용자 입력 텍스트 Returns: 추출된 자산 목록 [{"program": "...", "amount": ...}, ...] """ assets = [] # 줄 단위 + 쉼표 단위로 분리 lines = re.split(r'[\n,]', text) for line in lines: line = line.strip() if not line: continue result = _parse_single_line(line) if result: # 중복 제거 (같은 프로그램이면 합산) existing = next( (a for a in assets if a["program"] == result["program"]), None ) if existing: existing["amount"] += result["amount"] else: assets.append(result) return assets def _parse_single_line(line: str) -> Optional[Dict[str, Any]]: """단일 라인에서 프로그램과 수량 추출.""" line_lower = line.lower() # 숫자 추출 (쉼표 제거) numbers = re.findall(r'[\d,]+', line) if not numbers: return None # 가장 큰 숫자를 amount로 사용 amount = max(int(n.replace(',', '')) for n in numbers if n.replace(',', '').isdigit()) if amount == 0: return None # 프로그램 식별 program = None for alias, prog_id in PROGRAM_ALIASES.items(): if alias in line_lower: program = prog_id break if not program: # 숫자만 있는 경우 - 문맥으로 추론 불가 return None return {"program": program, "amount": amount} def normalize_program_name(name: str) -> Optional[str]: """프로그램 별칭을 정규화합니다.""" name_lower = name.lower().strip() return PROGRAM_ALIASES.get(name_lower) # ============================================================================= # MCP Tool 핸들러 # ============================================================================= async def handle_parse_asset_text( arguments: Dict[str, Any], user_id: str ) -> Dict[str, Any]: """ user_parse_asset_text Tool 핸들러. 사용자 입력 텍스트에서 포인트/마일리지를 자동 추출합니다. """ text = arguments.get("text", "") save_to_profile = arguments.get("save_to_profile", False) if not text: return { "success": False, "error": "파싱할 텍스트가 없습니다.", "usage": "예: '대한항공 45000 마일, AMEX MR 50000점'" } # 파싱 assets = parse_asset_text(text) if not assets: return { "success": False, "error": "인식 가능한 포인트/마일리지 정보가 없습니다.", "tip": "프로그램명과 숫자를 함께 입력해주세요. 예: '대한항공 45000'", "supported_programs": list(POINT_VALUATIONS.keys()) } # 프로그램 이름 보강 for asset in assets: prog_info = POINT_VALUATIONS.get(asset["program"], {}) asset["program_name"] = prog_info.get("name", asset["program"]) asset["currency"] = prog_info.get("currency", "USD") result = { "success": True, "parsed_assets": assets, "count": len(assets), "parsed_at": datetime.now().isoformat(), "tip": "user_get_asset_valuation을 호출하여 총 가치를 계산할 수 있습니다." } # 향후: save_to_profile이 True면 DB에 저장 if save_to_profile: result["save_status"] = "not_implemented" result["save_note"] = "자산 저장 기능은 향후 구현 예정" return result