#!/usr/bin/env python3
"""
허깅페이스 Spaces 배포용 메인 파일
소방 복무관리 RAG 챗봇
"""
import os
import sys
import argparse
import gradio as gr
import pandas as pd
from pathlib import Path
from typing import List, Dict, Tuple, Optional
import hashlib
import time
import shutil
# 현재 디렉토리를 Python 경로에 추가
current_dir = os.path.dirname(os.path.abspath(__file__))
sys.path.append(current_dir)
# 모듈 임포트
from config import Config
from rag_chatbot import RAGChatbot
from document_processor import DocumentProcessor
# OpenAI/Supabase 관련 모듈은 에러 처리 후 임포트
try:
from openai_chatbot import OpenAIRAGChatbot
OPENAI_AVAILABLE = True
except ImportError as e:
print(f"OpenAI 모듈 임포트 오류: {e}")
print("FAISS 모드로 실행됩니다.")
OPENAI_AVAILABLE = False
OpenAIRAGChatbot = None
class AdminManager:
"""관리자 기능을 담당하는 클래스"""
def __init__(self):
self.admin_password = os.getenv("ADMIN_PASSWORD")
self.enabled = bool(self.admin_password)
self.authenticated = False
self.auth_time = None
self.session_timeout = 3600 # 1시간
self.uploaded_files_dir = Path("uploaded_documents")
self.uploaded_files_dir.mkdir(exist_ok=True)
def _hash_password(self, password: str) -> str:
"""비밀번호 해싱"""
return hashlib.sha256(password.encode()).hexdigest()
def authenticate(self, password: str) -> Tuple[bool, str]:
"""관리자 인증"""
if not self.enabled:
return False, "❌ 관리자 기능이 비활성화되었습니다. ADMIN_PASSWORD를 설정해주세요."
if self._hash_password(password) == self._hash_password(self.admin_password):
self.authenticated = True
self.auth_time = time.time()
return True, "✅ 인증 성공! 관리자 기능에 접근할 수 있습니다."
else:
self.authenticated = False
return False, "❌ 비밀번호가 올바르지 않습니다."
def is_authenticated(self) -> bool:
"""인증 상태 확인"""
if not self.enabled:
return False
if not self.authenticated:
return False
# 세션 타임아웃 확인
if self.auth_time and (time.time() - self.auth_time) > self.session_timeout:
self.authenticated = False
return False
return True
def logout(self) -> Tuple[bool, str]:
"""로그아웃"""
self.authenticated = False
self.auth_time = None
return False, "👋 로그아웃되었습니다."
def get_document_list(self) -> pd.DataFrame:
"""업로드된 문서 목록 가져오기"""
documents_folder = Path(Config.DOCS_FOLDER)
docs = []
if documents_folder.exists():
for file_path in documents_folder.glob("*"):
if file_path.is_file():
stat = file_path.stat()
docs.append({
"파일명": file_path.name,
"크기": f"{stat.st_size / 1024:.1f} KB",
"수정일": time.strftime("%Y-%m-%d %H:%M", time.localtime(stat.st_mtime)),
"타입": file_path.suffix.upper()
})
return pd.DataFrame(docs) if docs else pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"])
def process_uploaded_files(self, files) -> str:
"""업로드된 파일 처리"""
if not self.enabled:
return "❌ 관리자 기능이 비활성화되어 있습니다. ADMIN_PASSWORD를 설정해주세요."
if not self.is_authenticated():
return "❌ 관리자 인증이 필요합니다."
if not files:
return "⚠️ 업로드된 파일이 없습니다."
try:
# documents 폴더로 파일 복사
docs_folder = Path(Config.DOCS_FOLDER)
docs_folder.mkdir(exist_ok=True)
processed_files = []
errors = []
for file_obj in files:
try:
# Gradio File 객체에서 파일 경로 가져오기
if hasattr(file_obj, 'name'):
file_path = file_obj.name
original_name = Path(file_path).name
elif isinstance(file_obj, str):
file_path = file_obj
original_name = Path(file_path).name
else:
# file_obj이 파일 경로 문자열인 경우
file_path = str(file_obj)
original_name = Path(file_path).name
# 파일명 안전 처리
safe_name = "".join(c for c in original_name if c.isalnum() or c in "._-")
if not safe_name:
safe_name = f"document_{len(processed_files)}.txt"
dest_path = docs_folder / safe_name
# 파일 복사
shutil.copy2(file_path, dest_path)
processed_files.append(safe_name)
except Exception as e:
file_name = getattr(file_obj, 'name', str(file_obj))
errors.append(f"{Path(file_name).name}: {str(e)}")
# 결과 메시지 생성
result_msg = f"📁 **파일 처리 완료**\n\n"
if processed_files:
result_msg += f"✅ **성공 ({len(processed_files)}개)**:\n"
for file_name in processed_files:
result_msg += f"- {file_name}\n"
if errors:
result_msg += f"\n❌ **오류 ({len(errors)}개)**:\n"
for error in errors:
result_msg += f"- {error}\n"
result_msg += f"\n💡 **다음 단계**: 하단 '벡터 DB 재구축' 버튼을 눌러 임베딩을 진행해주세요."
return result_msg
except Exception as e:
return f"❌ 파일 처리 중 오류 발생: {str(e)}"
def rebuild_vector_db(self, force_rebuild: bool = True) -> str:
"""벡터 DB 재구축"""
if not self.enabled:
return "❌ 관리자 기능이 비활성화되어 있습니다. ADMIN_PASSWORD를 설정해주세요."
if not self.is_authenticated():
return "❌ 관리자 인증이 필요합니다."
try:
# 챗봇 재초기화
docs_folder = Path(Config.DOCS_FOLDER)
if not docs_folder.exists() or not list(docs_folder.glob("*")):
return "⚠️ documents 폴더에 파일이 없습니다. 먼저 파일을 업로드해주세요."
# 기존 챗봇 재초기화
success = self.chatbot.initialize(str(docs_folder), force_rebuild)
if success:
# 통계 정보 가져오기
stats = self.chatbot.get_stats()
vector_stats = stats.get("vector_store", {})
result_msg = f"🔄 **벡터 DB 재구축 완료**\n\n"
result_msg += f"- 📄 처리 문서: {vector_stats.get('total_documents', 0)}개\n"
result_msg += f"- 🧠 임베딩 모델: {vector_stats.get('embedding_model', 'N/A')}\n"
result_msg += f"- 📏 벡터 차원: {vector_stats.get('index_dimension', 'N/A')}\n"
result_msg += f"- ✅ 챗봇 업데이트 완료"
return result_msg
else:
return "❌ 벡터 DB 재구축 실패"
except Exception as e:
return f"❌ 벡터 DB 재구축 중 오류 발생: {str(e)}"
def delete_document(self, filename: str) -> str:
"""문서 삭제"""
if not self.enabled:
return "❌ 관리자 기능이 비활성화되어 있습니다. ADMIN_PASSWORD를 설정해주세요."
if not self.is_authenticated():
return "❌ 관리자 인증이 필요합니다."
try:
file_path = Path(Config.DOCS_FOLDER) / filename
if file_path.exists():
file_path.unlink()
return f"✅ '{filename}' 파일이 삭제되었습니다."
else:
return f"⚠️ '{filename}' 파일을 찾을 수 없습니다."
except Exception as e:
return f"❌ 파일 삭제 중 오류 발생: {str(e)}"
class HuggingFaceApp:
"""허깅페이스 Spaces 배포용 앱 클래스"""
def __init__(self):
# 벡터 DB 타입에 따라 챗봇 선택
if self._should_use_supabase():
try:
self.chatbot = OpenAIRAGChatbot()
except Exception as e:
print(f"Supabase 챗봇 초기화 오류: {e}")
print("FAISS 모드로 전환됩니다.")
self.chatbot = RAGChatbot()
else:
self.chatbot = RAGChatbot()
self.is_initialized = False
# 관리자 매니저 초기화
self.admin_manager = AdminManager()
# 관리자 매니저에 챗봇 참조 전달
self.admin_manager.chatbot = self.chatbot
# 예시 질문 (허깅페이스 환경 최적화)
self.example_questions = [
"연차휴가 사용 방법을 알려주세요",
"정규근무시간은 어떻게 되나요?",
"당직근무 절차가 궁금합니다",
"인사평가는 언제 진행되나요?",
"파견근무 신청 방법",
"복무규정 위반 시 처리"
]
# 앱 초기화
self._initialize_app()
def _should_use_supabase(self) -> bool:
"""Supabase 경로 사용 여부 판단"""
has_supabase_env = bool(Config.SUPABASE_URL and Config.SUPABASE_KEY)
has_openai = bool(Config.OPENAI_API_KEY)
if Config.VECTOR_DB_TYPE != "supabase":
return False
if not has_supabase_env:
print("Supabase 환경 변수가 없어 FAISS 모드로 실행됩니다.")
return False
if not has_openai:
print("OpenAI API Key가 없어 FAISS 모드로 실행됩니다.")
return False
if not (OPENAI_AVAILABLE and OpenAIRAGChatbot):
print("OpenAI 모듈을 사용할 수 없어 FAISS 모드로 실행됩니다.")
return False
return True
def _initialize_app(self):
"""앱 초기화"""
try:
print("Starting Fire Service Management RAG Chatbot...")
# 문서 폴더 확인 및 샘플 데이터 생성
self._ensure_documents()
# 챗봇 초기화
success = self.chatbot.initialize()
if success:
self.is_initialized = True
print("Chatbot initialization completed")
else:
print("Chatbot initialization failed - running in template mode")
except Exception as e:
print(f"Initialization error: {str(e)}")
# 오류가 있어도 앱은 계속 실행
def _ensure_documents(self):
"""문서 폴더 및 샘플 데이터 확인"""
docs_folder = Path("documents")
docs_folder.mkdir(exist_ok=True)
# 샘플 문서가 없으면 생성
sample_files = list(docs_folder.glob("*.txt"))
if not sample_files:
print("Creating sample documents...")
self._create_sample_documents()
def _create_sample_documents(self):
"""샘플 복무관리 문서 생성"""
sample_docs = [
{
"filename": "복무관리규정.txt",
"content": """소방공무원 복무관리 규정
제1장 총칙
제1조 (목적)
이 규정은 소방공무원의 복무에 관한 기본사항을 규정하여 직무수행의 효율성을 높이고 조직의 발전에 기여함을 목적으로 한다.
제2조 (근무시간)
1. 정규근무시간은 09:00부터 18:00까지로 한다.
2. 점심시간은 12:00부터 13:00까지로 한다.
3. 토요일, 일요일 및 법정공휴일은 휴무일로 한다.
제3조 (연차휴가)
1. 연차휴가는 1년간 정상 근무한 자에게 15일을 부여한다.
2. 연차휴가 사용 시 3일 전까지 신청서를 제출해야 한다.
3. 부서장의 승인을 받아 사용하며, 긴급한 경우에는 사후 승인도 가능하다.
제4조 (당직근무)
1. 당직근무는 정규근무시간 외에 수행하는 근무를 말한다.
2. 당직자는 비상상황에 대비한 통신장비를 항시 점검해야 한다.
3. 당직 중에는 음주를 엄금하며, 직무 수행에 지장이 없는 행위만 가능하다.
"""
},
{
"filename": "인사평가규정.txt",
"content": """소방공무원 인사평가 규정
제1장 평가의 기본원칙
제1조 (평가목적)
소방공무원의 직무수행 능력과 성과를 객관적으로 평가하여 능력위주의 인사관리를 정립하고 공정한 보상 및 승진의 기초 자료로 활용한다.
제2조 (평가주기)
1. 정기평가는 연 1회 실시하며, 평가기간은 매년 1월 1일부터 12월 31일까지로 한다.
2. 수시평가는 특별한 사유가 있을 경우 실시할 수 있다.
제3조 (평가항목)
1. 직무수행 능력 (40점)
2. 업무 성과 (30점)
3. 근무 태도 (20점)
4. 협업 능력 (10점)
제4조 (평가등급)
- 수 (90점 이상)
- 우 (80점 이상 90점 미만)
- 양 (70점 이상 80점 미만)
- 가 (60점 이상 70점 미만)
- 미 (60점 미만)
"""
},
{
"filename": "교육훈련.txt",
"content": """소방공무원 교육훈련 안내
제1조 (교육목적)
소방공무원의 전문성 향상과 직무능력 개발을 위한 체계적인 교육훈련을 실시한다.
제2조 (필수교육)
1. 신임교육: 신규 임용자 대상 2주간 집체교육
2. 직무연수: 매년 1회, 직무별 전문교육
3. 안전교육: 분기별 1회, 안전사고 예방 교육
제3조 (선택교육)
1. 외국어 교육
2. 정보통신 기술 교육
3. 리더십 교육
4. 전문 자격증 취득 지원 교육
제4조 (교육신청)
1. 교육 희망자는 소속 기관을 통해 신청한다.
2. 신청 시기는 교육 시작일 1개월 전까지이다.
3. 업무에 지장이 없는 경우 우선 선발한다.
"""
}
]
for doc in sample_docs:
file_path = Path("documents") / doc["filename"]
try:
with open(file_path, 'w', encoding='utf-8') as f:
f.write(doc["content"])
print(f"✅ {doc['filename']} 생성 완료")
except Exception as e:
print(f"❌ {doc['filename']} 생성 실패: {str(e)}")
def format_response(self, response) -> str:
"""응답을 Gradio 형식으로 변환"""
try:
# 메시지 형식으로 변환
answer = response.answer
# 신뢰도 표시
confidence = getattr(response, 'confidence', 0.0)
if confidence >= 0.8:
confidence_emoji = "🟢"
elif confidence >= 0.5:
confidence_emoji = "🟡"
else:
confidence_emoji = "🔴"
# 응답 시간
response_time = getattr(response, 'response_time', 0.0)
# 출처 정보
sources = getattr(response, 'sources', [])
source_text = ""
if sources:
source_text = "\n\n📚 **참고자료:**\n"
for i, source in enumerate(sources[:3], 1): # 최대 3개만 표시
source_name = source.get('source', '알 수 없음')
source_text += f"{i}. {source_name}\n"
# 전체 응답
full_response = f"""{answer}
---
{confidence_emoji} 신뢰도: {confidence:.1%}
⏱️ 응답시간: {response_time:.2f}초
📄 참고문서: {len(sources)}개{source_text}"""
return full_response
except Exception as e:
return f"응답 형식 변환 중 오류 발생: {str(e)}"
def chat_function(self, message: str, history: List[List[str]]) -> List[List[str]]:
"""채팅 함수"""
if not message.strip():
return history
try:
# 챗봇 응답 생성
if self.is_initialized:
response = self.chatbot.generate_answer(message, use_llm=False)
answer = self.format_response(response)
else:
answer = "죄송합니다. 챗봇이 초기화되지 않았습니다. 페이지를 새로고침해주세요."
# 히스토리에 추가
history.append([message, answer])
except Exception as e:
error_msg = f"답변 생성 중 오류 발생: {str(e)}\n\n관리자에게 문의해주세요."
history.append([message, error_msg])
return history
def create_demo(self):
"""Gradio 데모 생성"""
custom_css = """
.gradio-container {
max-width: 1200px !important;
margin: auto !important;
font-family: 'Segoe UI', 'Pretendard', system-ui, -apple-system, sans-serif;
background: #0f172a;
color: #e2e8f0;
}
.main-hero {
background: radial-gradient(circle at 20% 20%, rgba(255,94,94,0.25), rgba(15,23,42,0.4)),
radial-gradient(circle at 80% 0%, rgba(255,214,102,0.25), rgba(15,23,42,0.4)),
linear-gradient(135deg, #111827 0%, #0b1222 100%);
padding: 28px;
border-radius: 18px;
border: 1px solid #1f2937;
box-shadow: 0 20px 60px rgba(0,0,0,0.35);
margin-bottom: 18px;
}
.hero-title { font-size: 2.4rem; font-weight: 800; margin: 0 0 8px 0; color: #fff; letter-spacing: -0.5px; }
.hero-sub { font-size: 1.05rem; color: #cbd5e1; margin: 0; }
.pill {
display: inline-flex;
align-items: center;
gap: 8px;
padding: 8px 12px;
border-radius: 999px;
background: rgba(255,255,255,0.08);
color: #e2e8f0;
font-size: 0.95rem;
border: 1px solid rgba(255,255,255,0.12);
}
.stat-grid {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(180px, 1fr));
gap: 12px;
margin-top: 14px;
}
.stat-card {
background: rgba(255,255,255,0.03);
border: 1px solid rgba(255,255,255,0.08);
border-radius: 14px;
padding: 14px;
}
.stat-value { font-size: 1.4rem; font-weight: 700; color: #fff; }
.stat-label { font-size: 0.9rem; color: #cbd5e1; }
.card {
background: #0b1222;
border: 1px solid #1f2937;
border-radius: 14px;
padding: 14px;
box-shadow: 0 12px 30px rgba(0,0,0,0.3);
}
.card h3 { margin-top: 0; color: #fff; }
.badge {
display: inline-block;
padding: 6px 10px;
border-radius: 10px;
font-size: 0.85rem;
background: #f97316;
color: #0b1222;
font-weight: 700;
}
.info-note {
background: rgba(255,255,255,0.05);
border: 1px dashed rgba(255,255,255,0.15);
padding: 12px;
border-radius: 12px;
color: #cbd5e1;
}
"""
stats = self.chatbot.get_stats() if self.is_initialized else {}
vector_stats = stats.get("vector_store", {})
doc_count = vector_stats.get("total_documents", 0)
index_dim = vector_stats.get("index_dimension", "N/A")
embedding_model = vector_stats.get("embedding_model", "N/A")
status_message = (
"✅ **챗봇 준비 완료** - 복무관리 관련 질문을 입력해주세요"
if self.is_initialized else
"⚠️ **챗봇 초기화 필요** - 관리자 탭에서 문서를 업로드하고 재구축하세요"
)
admin_enabled = self.admin_manager.enabled
admin_status_init = (
"🔒 **관리자 기능 비활성화** - ADMIN_PASSWORD를 설정하세요"
if not admin_enabled else
"🔒 **인증 필요** - 관리자 비밀번호를 입력해주세요"
)
admin_notice = (
"ADMIN_PASSWORD가 설정되지 않아 관리자 기능이 비활성화되었습니다.
"
if not admin_enabled else ""
)
with gr.Blocks(
title="소방 복무관리 RAG 챗봇",
theme=gr.themes.Soft(),
css=custom_css
) as demo:
gr.HTML(f"""
🚒 소방 복무관리 · RAG
규정·절차를 바로 찾는
복무관리 AI 컨시어지
연차/당직/인사평가/교육훈련까지 신뢰할 수 있는 출처 기반 답변을 제공합니다.
🔍 실시간 검색
📚 출처 명시
🛡️ 보안 인증
실시간 상태
{'LIVE' if self.is_initialized else 'SETUP'}
{status_message}
""")
with gr.Row():
with gr.Column(scale=7):
gr.Markdown("### ⚡ 바로 묻기")
quick_note = "채팅 탭에서 추천 질문을 클릭해 입력창에 불러온 뒤 바로 전송하세요."
gr.HTML(f"{quick_note}
")
with gr.Column(scale=5):
gr.Markdown("### 🧭 사용 흐름")
gr.HTML("""
- 채팅 탭에서 질문 입력 → 답변/출처 확인
- 문서 추가 필요 시 관리자 탭에서 업로드
- “벡터 DB 재구축” 후 새 문서를 반영
챗봇이 초기화되지 않았다면 문서를 업로드하고 재구축을 실행하세요.
""")
with gr.Tabs():
with gr.Tab("💬 채팅"):
gr.HTML("""
복무관리 전문가에게 물어보세요
연차/당직/인사평가/교육훈련/징계 등 규정 기반 답변을 제공합니다.
""")
with gr.Row():
status_text = gr.HTML(status_message)
chatbot = gr.Chatbot(
height=600,
placeholder="안녕하세요! 복무관리 관련 무엇이든 물어보세요.",
avatar_images=["👤", "🤖"],
type="tuples"
)
with gr.Row():
msg = gr.Textbox(
placeholder="예: 연차휴가 사용 절차를 알려줘 / 당직근무 수당 기준이 궁금해",
container=False,
scale=7
)
submit_btn = gr.Button("전송", scale=1, variant="primary")
clear_btn = gr.Button("초기화", scale=1)
gr.Markdown("#### 💡 추천 질문")
with gr.Row():
for question in self.example_questions[:4]:
gr.Button(question, size="sm").click(fn=lambda q=question: q, outputs=msg)
with gr.Accordion("📊 시스템 정보", open=False):
if self.is_initialized:
stats = self.chatbot.get_stats()
vector_stats = stats.get("vector_store", {})
gr.Markdown(f"""
- **문서 수**: {vector_stats.get('total_documents', 0)}개
- **임베딩 모델**: {vector_stats.get('embedding_model', 'N/A')}
- **응답 모드**: 템플릿 기반
- **최대 검색 문서**: {Config.MAX_RETRIEVE_DOCS}개
""")
else:
gr.Markdown("⚠️ 챗봇이 초기화되지 않았습니다.")
with gr.Tab("🔐 관리자"):
gr.Markdown("### 🛡️ 관리자 전용 기능")
if admin_notice:
gr.HTML(admin_notice)
with gr.Row():
auth_status = gr.HTML(admin_status_init)
with gr.Group():
gr.Markdown("#### 🔑 관리자 인증")
with gr.Row():
admin_password = gr.Textbox(
placeholder="관리자 비밀번호를 입력하세요",
type="password",
label="비밀번호",
scale=3
)
auth_btn = gr.Button("🔐 인증", variant="primary", scale=1)
logout_btn = gr.Button("🚪 로그아웃", scale=1)
with gr.Group(visible=admin_enabled) as admin_panel:
gr.Markdown("#### 📁 문서 업로드")
upload_status = gr.HTML("⏳ 파일을 선택하고 업로드해주세요.")
uploaded_files = gr.File(
file_count="multiple",
file_types=[".txt", ".pdf", ".docx", ".xlsx", ".csv"],
label="복무관리 문서 파일"
)
with gr.Row():
upload_btn = gr.Button("📤 파일 업로드 및 처리", variant="primary")
refresh_btn = gr.Button("🔄 목록 새로고침")
gr.Markdown("#### 📋 현재 문서 목록")
document_list = gr.DataFrame(
headers=["파일명", "크기", "수정일", "타입"],
datatype=["str", "str", "str", "str"]
)
with gr.Row():
delete_selected = gr.Button("🗑️ 선택 삭제", variant="stop")
rebuild_btn = gr.Button("🔄 벡터 DB 재구축", variant="secondary")
gr.Markdown("#### 📊 작업 결과")
operation_result = gr.HTML("작업 결과가 여기에 표시됩니다.")
gr.Markdown("#### 📈 시스템 통계")
system_stats = gr.HTML("시스템 상태를 확인해주세요.")
with gr.Tab("📊 정보"):
gr.Markdown("### 시스템 정보 및 설정")
with gr.Row():
with gr.Column():
gr.Markdown("#### 🤖 챗봇 정보")
chatbot_info = gr.HTML("챗봇 정보 로딩 중...")
with gr.Column():
gr.Markdown("#### ⚙️ 시스템 설정")
system_settings = gr.HTML(f"""
- **벡터 DB 타입**: {Config.VECTOR_DB_TYPE}
- **청크 크기**: {Config.CHUNK_SIZE}
- **최대 검색 문서**: {Config.MAX_RETRIEVE_DOCS}
- **문서 폴더**: {Config.DOCS_FOLDER}
""")
gr.Markdown("#### 📋 사용 방법")
usage_guide = gr.HTML("""
관리자 비밀번호: 환경변수 ADMIN_PASSWORD를 설정해야 관리자 기능이 활성화됩니다.
초기화 필요 시 관리자 탭에서 문서를 업로드한 뒤 “벡터 DB 재구축”을 클릭하세요.
""")
# 이벤트 핸들러
def user_input(user_message, history):
"""사용자 입력 처리"""
return "", history + [[user_message, None]]
def bot_response(history):
"""봇 응답 처리"""
if history and history[-1][1] is None:
user_message = history[-1][0]
bot_message = self.chat_function(user_message, history[:-1])
if bot_message:
history[-1][1] = bot_message[-1][1]
else:
history[-1][1] = "죄송합니다. 응답을 생성할 수 없습니다."
return history
def authenticate_admin(password):
"""관리자 인증 처리"""
is_auth, message = self.admin_manager.authenticate(password)
if is_auth:
status_html = "🔓 **인증 성공!** 관리자 기능에 접근할 수 있습니다."
doc_list = self.admin_manager.get_document_list()
stats_html = self.get_system_stats_html()
return status_html, doc_list, stats_html, "", ""
else:
status_html = f"🔒 **인증 실패** - {message}"
return status_html, pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]), "인증이 필요합니다.", password, ""
def logout_admin():
"""관리자 로그아웃 처리"""
self.admin_manager.logout()
return "🔒 **로그아웃되었습니다** - 다시 인증해주세요", pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]), "인증이 필요합니다."
def upload_files(files):
"""파일 업로드 처리"""
result = self.admin_manager.process_uploaded_files(files)
doc_list = self.admin_manager.get_document_list()
return result, doc_list, None
def rebuild_vector_db():
"""벡터 DB 재구축"""
result = self.admin_manager.rebuild_vector_db(force_rebuild=True)
stats_html = self.get_system_stats_html()
return result, stats_html
def delete_selected_files(selected_rows):
"""선택된 파일 삭제"""
if not selected_rows or not self.admin_manager.is_authenticated():
return "❌ 관리자 인증이 필요하거나 삭제할 파일을 선택해주세요.", self.admin_manager.get_document_list()
results = []
for row in selected_rows:
filename = row[0]
result = self.admin_manager.delete_document(filename)
results.append(result)
doc_list = self.admin_manager.get_document_list()
result_msg = "🗑️ **파일 삭제 결과**\n\n" + "\n".join(results)
return result_msg, doc_list
def refresh_document_list():
"""문서 목록 새로고침"""
if self.admin_manager.is_authenticated():
return self.admin_manager.get_document_list(), self.get_system_stats_html()
else:
return pd.DataFrame(columns=["파일명", "크기", "수정일", "타입"]), "인증이 필요합니다."
def get_system_stats_html():
"""시스템 통계 HTML 생성"""
if not self.admin_manager.is_authenticated():
return "🔒 인증이 필요합니다."
try:
stats = self.chatbot.get_stats() if self.is_initialized else {}
vector_stats = stats.get("vector_store", {})
return f"""
📊 시스템 상태
- 챗봇 상태: {'✅ 정상' if self.is_initialized else '⚠️ 초기화 필요'}
- 관리자 인증: {'✅ 인증됨' if self.admin_manager.is_authenticated() else '❌ 미인증'}
- 문서 수: {vector_stats.get('total_documents', 0)}개
- 임베딩 모델: {vector_stats.get('embedding_model', 'N/A')}
- 벡터 차원: {vector_stats.get('index_dimension', 'N/A')}
- 벡터 DB 타입: {Config.VECTOR_DB_TYPE}
"""
except Exception as e:
return f"❌ 통계 정보 로딩 실패: {str(e)}"
def get_chatbot_info_html():
"""챗봇 정보 HTML 생성"""
try:
if self.is_initialized:
stats = self.chatbot.get_stats()
vector_stats = stats.get("vector_store", {})
return f"""
🤖 챗봇 정보
- 상태: ✅ 정상 작동 중
- 임베딩 모델: {vector_stats.get('embedding_model', 'N/A')}
- 문서 수: {vector_stats.get('total_documents', 0)}개
- 벡터 차원: {vector_stats.get('index_dimension', 'N/A')}
- LLM 사용: {'사용 가능' if stats.get('llm_available') else '템플릿 모드'}
"""
else:
return """
🤖 챗봇 정보
⚠️ 챗봇이 초기화되지 않았습니다.
"""
except Exception as e:
return f"❌ 정보 로딩 실패: {str(e)}"
# 메시지 전송 이벤트
msg.submit(
user_input,
[msg, chatbot],
[msg, chatbot],
queue=False
).then(
bot_response,
chatbot,
chatbot
)
submit_btn.click(
user_input,
[msg, chatbot],
[msg, chatbot],
queue=False
).then(
bot_response,
chatbot,
chatbot
)
clear_btn.click(
lambda: ([], ""),
outputs=[chatbot, msg]
)
# 관리자 이벤트 핸들러
auth_btn.click(
authenticate_admin,
inputs=[admin_password],
outputs=[auth_status, document_list, system_stats, admin_password, uploaded_files]
)
logout_btn.click(
logout_admin,
outputs=[auth_status, document_list, system_stats]
)
upload_btn.click(
upload_files,
inputs=[uploaded_files],
outputs=[operation_result, document_list, uploaded_files]
)
rebuild_btn.click(
rebuild_vector_db,
outputs=[operation_result, system_stats]
)
delete_selected.click(
delete_selected_files,
inputs=[document_list],
outputs=[operation_result, document_list]
)
refresh_btn.click(
refresh_document_list,
outputs=[document_list, system_stats]
)
# 초기 정보 로드
demo.load(
fn=lambda: (
self.admin_manager.get_document_list(),
get_system_stats_html(),
get_chatbot_info_html()
),
outputs=[document_list, system_stats, chatbot_info]
)
demo.queue() # FastAPI/Starlette 응답 길이 이슈 완화 및 안정적 이벤트 처리
return demo
def main():
"""메인 실행 함수"""
parser = argparse.ArgumentParser(description="소방 복무관리 RAG 챗봇")
parser.add_argument("--share", action="store_true", help="공유 링크 생성")
parser.add_argument("--port", type=int, default=7860, help="서버 포트")
args = parser.parse_args()
# 앱 생성
app = HuggingFaceApp()
demo = app.create_demo()
# 실행
print("🚀 허깅페이스 Spaces 앱 시작 중...")
demo.launch(
share=args.share,
server_port=args.port,
server_name="0.0.0.0",
show_error=True
)
if __name__ == "__main__":
main()