""" RAG 검색 챗봇 웹 애플리케이션 - API 라우트 정의 """ import os import json import logging import tempfile import requests from flask import request, jsonify, render_template, send_from_directory, session, redirect, url_for from datetime import datetime from werkzeug.utils import secure_filename # 로거 가져오기 logger = logging.getLogger(__name__) def register_routes(app, login_required, llm_interface, retriever, stt_client, DocumentProcessor, base_retriever, app_ready, ADMIN_USERNAME, ADMIN_PASSWORD, DEVICE_SERVER_URL): """Flask 애플리케이션에 라우트 등록""" # 헬퍼 함수 def allowed_audio_file(filename): """파일이 허용된 오디오 확장자를 가지는지 확인""" ALLOWED_AUDIO_EXTENSIONS = {'mp3', 'wav', 'ogg', 'm4a'} return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_AUDIO_EXTENSIONS def allowed_doc_file(filename): """파일이 허용된 문서 확장자를 가지는지 확인""" ALLOWED_DOC_EXTENSIONS = {'txt', 'md', 'pdf', 'docx', 'csv'} return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_DOC_EXTENSIONS # 임베딩 저장 함수 def save_embeddings(base_retriever, file_path): """임베딩 데이터를 압축하여 파일에 저장""" import pickle import gzip try: # 저장 디렉토리가 없으면 생성 os.makedirs(os.path.dirname(file_path), exist_ok=True) # 타임스탬프 추가 save_data = { 'timestamp': datetime.now().isoformat(), 'retriever': base_retriever } # 압축하여 저장 (용량 줄이기) with gzip.open(file_path, 'wb') as f: pickle.dump(save_data, f) logger.info(f"임베딩 데이터를 {file_path}에 압축하여 저장했습니다.") return True except Exception as e: logger.error(f"임베딩 저장 중 오류 발생: {e}") return False @app.route('/login', methods=['GET', 'POST']) def login(): error = None next_url = request.args.get('next') logger.info(f"-------------- 로그인 페이지 접속 (Next: {next_url}) --------------") logger.info(f"Method: {request.method}") if request.method == 'POST': logger.info("로그인 시도 받음") username = request.form.get('username', '') password = request.form.get('password', '') logger.info(f"입력된 사용자명: {username}") logger.info(f"비밀번호 입력 여부: {len(password) > 0}") # 환경 변수 또는 기본값과 비교 valid_username = ADMIN_USERNAME valid_password = ADMIN_PASSWORD logger.info(f"검증용 사용자명: {valid_username}") logger.info(f"검증용 비밀번호 존재 여부: {valid_password is not None and len(valid_password) > 0}") if username == valid_username and password == valid_password: logger.info(f"로그인 성공: {username}") # 세션 설정 전 현재 세션 상태 로깅 logger.debug(f"세션 설정 전: {session}") # 세션에 로그인 정보 저장 session.permanent = True session['logged_in'] = True session['username'] = username session.modified = True logger.info(f"세션 설정 후: {session}") logger.info("세션 설정 완료, 리디렉션 시도") # 로그인 성공 후 리디렉션 redirect_to = next_url or url_for('index') logger.info(f"리디렉션 대상: {redirect_to}") response = redirect(redirect_to) return response else: logger.warning("로그인 실패: 아이디 또는 비밀번호 불일치") if username != valid_username: logger.warning("사용자명 불일치") if password != valid_password: logger.warning("비밀번호 불일치") error = '아이디 또는 비밀번호가 올바르지 않습니다.' else: logger.info("로그인 페이지 GET 요청") if 'logged_in' in session: logger.info("이미 로그인된 사용자, 메인 페이지로 리디렉션") return redirect(url_for('index')) logger.info("---------- 로그인 페이지 렌더링 ----------") return render_template('login.html', error=error, next=next_url) @app.route('/logout') def logout(): logger.info("-------------- 로그아웃 요청 --------------") logger.info(f"로그아웃 전 세션 상태: {session}") if 'logged_in' in session: username = session.get('username', 'unknown') logger.info(f"사용자 {username} 로그아웃 처리 시작") session.pop('logged_in', None) session.pop('username', None) session.modified = True logger.info(f"세션 정보 삭제 완료. 현재 세션: {session}") else: logger.warning("로그인되지 않은 상태에서 로그아웃 시도") logger.info("로그인 페이지로 리디렉션") response = redirect(url_for('login')) return response @app.route('/') @login_required def index(): """메인 페이지""" nonlocal app_ready # 앱 준비 상태 확인 - 30초 이상 지났으면 강제로 ready 상태로 변경 current_time = datetime.now() start_time = datetime.fromtimestamp(os.path.getmtime(__file__)) time_diff = (current_time - start_time).total_seconds() if not app_ready and time_diff > 30: logger.warning(f"앱이 30초 이상 초기화 중 상태입니다. 강제로 ready 상태로 변경합니다.") app_ready = True if not app_ready: logger.info("앱이 아직 준비되지 않아 로딩 페이지 표시") return render_template('loading.html'), 503 # 서비스 준비 안됨 상태 코드 logger.info("메인 페이지 요청") return render_template('index.html') @app.route('/api/status') @login_required def app_status(): """앱 초기화 상태 확인 API""" logger.info(f"앱 상태 확인 요청: {'Ready' if app_ready else 'Not Ready'}") return jsonify({"ready": app_ready}) @app.route('/api/llm', methods=['GET', 'POST']) @login_required def llm_api(): """사용 가능한 LLM 목록 및 선택 API""" if not app_ready: return jsonify({"error": "앱이 아직 초기화 중입니다. 잠시 후 다시 시도해주세요."}), 503 if request.method == 'GET': logger.info("LLM 목록 요청") try: current_details = llm_interface.get_current_llm_details() if hasattr(llm_interface, 'get_current_llm_details') else {"id": "unknown", "name": "Unknown"} supported_llms_dict = llm_interface.SUPPORTED_LLMS if hasattr(llm_interface, 'SUPPORTED_LLMS') else {} supported_list = [{ "name": name, "id": id, "current": id == current_details.get("id") } for name, id in supported_llms_dict.items()] return jsonify({ "supported_llms": supported_list, "current_llm": current_details }) except Exception as e: logger.error(f"LLM 정보 조회 오류: {e}") return jsonify({"error": "LLM 정보 조회 중 오류 발생"}), 500 elif request.method == 'POST': data = request.get_json() if not data or 'llm_id' not in data: return jsonify({"error": "LLM ID가 제공되지 않았습니다."}), 400 llm_id = data['llm_id'] logger.info(f"LLM 변경 요청: {llm_id}") try: if not hasattr(llm_interface, 'set_llm') or not hasattr(llm_interface, 'llm_clients'): raise NotImplementedError("LLM 인터페이스에 필요한 메소드/속성 없음") if llm_id not in llm_interface.llm_clients: return jsonify({"error": f"지원되지 않는 LLM ID: {llm_id}"}), 400 success = llm_interface.set_llm(llm_id) if success: new_details = llm_interface.get_current_llm_details() logger.info(f"LLM이 '{new_details.get('name', llm_id)}'로 변경되었습니다.") return jsonify({ "success": True, "message": f"LLM이 '{new_details.get('name', llm_id)}'로 변경되었습니다.", "current_llm": new_details }) else: logger.error(f"LLM 변경 실패 (ID: {llm_id})") return jsonify({"error": "LLM 변경 중 내부 오류 발생"}), 500 except Exception as e: logger.error(f"LLM 변경 처리 중 오류: {e}", exc_info=True) return jsonify({"error": f"LLM 변경 중 오류 발생: {str(e)}"}), 500 @app.route('/api/chat', methods=['POST']) @login_required def chat(): """텍스트 기반 챗봇 API""" if not app_ready or retriever is None: return jsonify({"error": "앱/검색기가 아직 초기화 중입니다. 잠시 후 다시 시도해주세요."}), 503 try: data = request.get_json() if not data or 'query' not in data: return jsonify({"error": "쿼리가 제공되지 않았습니다."}), 400 query = data['query'] logger.info(f"텍스트 쿼리 수신: {query[:100]}...") # RAG 검색 수행 if not hasattr(retriever, 'search'): raise NotImplementedError("Retriever에 search 메소드가 없습니다.") search_results = retriever.search(query, top_k=5, first_stage_k=6) # 컨텍스트 준비 if not hasattr(DocumentProcessor, 'prepare_rag_context'): raise NotImplementedError("DocumentProcessor에 prepare_rag_context 메소드가 없습니다.") context = DocumentProcessor.prepare_rag_context(search_results, field="text") if not context: logger.warning("검색 결과가 없어 컨텍스트를 생성하지 못함.") # LLM에 질의 llm_id = data.get('llm_id', None) if not hasattr(llm_interface, 'rag_generate'): raise NotImplementedError("LLMInterface에 rag_generate 메소드가 없습니다.") if not context: answer = "죄송합니다. 관련 정보를 찾을 수 없습니다." logger.info("컨텍스트 없이 기본 응답 생성") else: answer = llm_interface.rag_generate(query, context, llm_id=llm_id) logger.info(f"LLM 응답 생성 완료 (길이: {len(answer)})") # 소스 정보 추출 (CSV ID 추출 로직 포함) sources = [] if search_results: for result in search_results: if not isinstance(result, dict): logger.warning(f"예상치 못한 검색 결과 형식: {type(result)}") continue if "source" in result: source_info = { "source": result.get("source", "Unknown"), "score": result.get("rerank_score", result.get("score", 0)) } # CSV 파일 특정 처리 if "text" in result and result.get("filetype") == "csv": try: text_lines = result["text"].strip().split('\n') if text_lines: first_line = text_lines[0].strip() if ',' in first_line: first_column = first_line.split(',')[0].strip() source_info["id"] = first_column logger.debug(f"CSV 소스 ID 추출: {first_column} from {source_info['source']}") except Exception as e: logger.warning(f"CSV 소스 ID 추출 실패 ({result.get('source')}): {e}") sources.append(source_info) # 최종 응답 response_data = { "answer": answer, "sources": sources, "llm": llm_interface.get_current_llm_details() if hasattr(llm_interface, 'get_current_llm_details') else {} } return jsonify(response_data) except Exception as e: logger.error(f"채팅 처리 중 오류 발생: {e}", exc_info=True) return jsonify({"error": f"처리 중 오류가 발생했습니다: {str(e)}"}), 500