""" 모바일 리액트 프런트에서 호출할 REST API 서버 (FastAPI) 엔드포인트: - POST /api/advise : 종합 상담 (금융+정보+복지) - POST /api/qa : 자유 질문/챗봇 응답 - GET /api/welfare/search : 복지정책 검색 - GET /api/products/search : 금융상품 RAG 검색 - GET /health : 상태 점검 """ from fastapi import FastAPI, HTTPException, Query from fastapi.staticfiles import StaticFiles from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import List, Optional, Dict, Any from agents.multi_agent_system import get_meeting_account_advisor from agents.welfare_advisor import WelfareAdvisor from agents.universal_info_advisor import get_universal_info_advisor from models.rag_system import get_kb_rag import pandas as pd import uvicorn app = FastAPI(title="KB 금융 AI Advisor API", version="1.0.0") # CORS (리액트/모바일 웹에서 호출 허용) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 프런트엔드 정적 파일 제공 (frontend/dist가 있으면 자동 서빙) try: import os FRONT_DIST = os.path.join(os.path.dirname(__file__), "frontend", "dist") if os.path.isdir(FRONT_DIST): app.mount("/", StaticFiles(directory=FRONT_DIST, html=True), name="frontend") except Exception: pass class UserInfo(BaseModel): age_group: Optional[str] = None family_type: Optional[str] = None income_level: Optional[str] = None region: Optional[str] = None class AdviseRequest(BaseModel): purpose: str = Field(..., description="예: 스페인 어학연수, 결혼준비 등") duration_months: int = Field(..., ge=1, le=120) target_amount: int = Field(..., ge=1) detailed_purpose: Optional[str] = None current_amount: Optional[int] = 0 user_info: Optional[UserInfo] = None class AdviseResponse(BaseModel): purpose_category: Optional[str] financial_recommendation: Optional[str] detailed_info: Optional[str] welfare_recommendation: Optional[str] comprehensive_advice: Optional[str] progress_info: Optional[str] = None class QARequest(BaseModel): question: str detailed_context: Optional[str] = None class QAResponse(BaseModel): answer: str category: Optional[str] = None @app.get("/health") def health() -> Dict[str, Any]: return {"status": "ok"} @app.post("/api/advise", response_model=AdviseResponse) def api_advise(req: AdviseRequest): try: advisor = get_meeting_account_advisor() results = advisor.get_comprehensive_advice( purpose=req.purpose, duration_months=req.duration_months, target_amount=req.target_amount, detailed_purpose=req.detailed_purpose, ) # 진행 현황 (current_amount 제공 시) progress_info = None if req.current_amount and req.current_amount > 0: progress_info = advisor.get_savings_progress_tracking( req.purpose, req.duration_months, req.target_amount, req.current_amount, ) # WelfareAdvisor에 사용자 정보 전달이 필요한 경우를 위해 보조 호출 (선택) welfare_text = results.get("welfare_recommendation") if (not welfare_text or "없습니다" in welfare_text) and req.user_info: welfare = WelfareAdvisor().get_welfare_recommendations( req.purpose, req.user_info.model_dump() ) if welfare: results["welfare_recommendation"] = welfare return AdviseResponse( purpose_category=results.get("purpose_category"), financial_recommendation=results.get("financial_recommendation"), detailed_info=results.get("detailed_info"), welfare_recommendation=results.get("welfare_recommendation"), comprehensive_advice=results.get("comprehensive_advice"), progress_info=progress_info, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.post("/api/qa", response_model=QAResponse) def api_qa(req: QARequest): try: advisor = get_universal_info_advisor() info = advisor.get_comprehensive_info( purpose=req.question, detailed_purpose=req.detailed_context, ) return QAResponse( answer=info.get("comprehensive_info", "답변을 생성하지 못했습니다."), category=info.get("category"), ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/welfare/search") def api_welfare_search( keyword: Optional[str] = Query(None, description="정책명/내용/대상 검색어"), region: Optional[str] = Query(None, description="지역명 (예: 서울, 부산)"), page: int = Query(1, ge=1), size: int = Query(10, ge=1, le=100), ): try: wa = WelfareAdvisor() df = wa.welfare_data if df.empty: return {"total": 0, "items": []} filtered = df.copy() if keyword: kw = keyword.strip() mask = ( filtered["policy_name"].fillna("").str.contains(kw, case=False) | filtered["service_content_detail"].fillna("").str.contains(kw, case=False) | filtered["target_audience_description"].fillna("").str.contains(kw, case=False) ) filtered = filtered[mask] if region and region != "전체": filtered = filtered[ (filtered["region_name"].fillna("").str.contains(region, case=False)) | (filtered["region_name"] == "전국") ] total = len(filtered) start = (page - 1) * size end = start + size page_df = filtered.iloc[start:end] items = [] for _, row in page_df.iterrows(): items.append( { "policy_name": row.get("policy_name", ""), "governing_body": row.get("governing_body_name", ""), "target_audience": row.get("target_audience_description", ""), "service_content": row.get("service_content_detail", ""), "region": row.get("region_name", "전국"), "application_link": row.get("application_link", ""), "last_updated": row.get("last_updated", ""), "tags": row.get("target_audience_tags", ""), } ) return {"total": total, "items": items} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/api/products/search") def api_products_search(q: str = Query(..., description="검색 질의"), k: int = Query(5, ge=1, le=20)): try: rag = get_kb_rag() results = rag.search_products(q, k=k) or [] return {"items": results} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": uvicorn.run("api_server:app", host="0.0.0.0", port=8000, reload=True)