import streamlit as st
from git import Repo
import json
import os
import time
from datetime import datetime, timedelta
from huggingface_hub import HfApi
from pathlib import Path
from calendar_rag import (
create_default_config,
AcademicCalendarRAG,
PipelineConfig
)
# Define repository path
os.environ["GITHUB_REPO_PATH"] = "D:\Last SWU\swu-chat-bot-project"
# Initialize global state
if 'all_history' not in st.session_state:
st.session_state.all_history = []
# Custom CSS for enhanced styling
def load_custom_css():
st.markdown("""
""", unsafe_allow_html=True)
def initialize_pipeline():
"""Initialize RAG pipeline with configurations"""
try:
openai_api_key = os.getenv('OPENAI_API_KEY') or st.secrets['OPENAI_API_KEY']
config = create_default_config(openai_api_key)
config.localization.enable_thai_normalization = True
config.retriever.top_k = 5
config.model.temperature = 0.3
pipeline = AcademicCalendarRAG(config)
with open("calendar.json", "r", encoding="utf-8") as f:
calendar_data = json.load(f)
pipeline.load_data(calendar_data)
return pipeline
except Exception as e:
st.error(f"Error initializing pipeline: {str(e)}")
return None
def load_qa_history():
"""Load QA history from local JSON file"""
try:
history_file = Path("qa_history.json")
if history_file.exists():
with open(history_file, "r", encoding="utf-8") as f:
return json.load(f)
return []
except Exception as e:
st.error(f"Error loading QA history: {str(e)}")
return []
def save_qa_history(history_entry):
"""
Save QA history to Git with improved authentication handling.
"""
try:
repo_path = Path(os.getenv('GITHUB_REPO_PATH', '')).resolve()
history_file = repo_path / "qa_history.json"
# Load existing history
history_data = []
if history_file.exists():
try:
with open(history_file, "r", encoding="utf-8") as f:
history_data = json.load(f)
except json.JSONDecodeError:
print("Creating new history file")
# Update history
if isinstance(history_entry, list):
history_data.extend(history_entry)
else:
history_data.append(history_entry)
# Save updated history
with open(history_file, "w", encoding="utf-8") as f:
json.dump(history_data, f, ensure_ascii=False, indent=2)
# Git operations
try:
repo = Repo(str(repo_path))
repo.index.add([str(history_file)])
if repo.is_dirty():
commit = repo.index.commit("Update QA history")
origin = repo.remote('origin')
try:
origin.push()
print(f"Successfully pushed commit {commit.hexsha}")
return True
except git.GitCommandError as push_error:
if "Authentication failed" in str(push_error):
print("Authentication failed - please check your Git credentials")
else:
print(f"Push failed: {push_error}")
return False
else:
print("No changes to commit")
return True
except Exception as git_error:
print(f"Git operation failed: {str(git_error)}")
return False
except Exception as e:
print(f"Save operation failed: {str(e)}")
return False
def add_to_qa_history(query: str, answer: str):
"""Add new QA pair to history and store it in session state"""
history_entry = {
"timestamp": (datetime.now() + timedelta(hours=7)).isoformat(),
"query": query,
"answer": answer
}
# Append to session state history
st.session_state.all_history.append(history_entry)
return history_entry
def clear_qa_history():
"""Clear QA history and sync with all storages"""
try:
# Clear session state
st.session_state.all_history = []
# Clear local file
with open("qa_history.json", "w", encoding="utf-8") as f:
json.dump([], f, ensure_ascii=False, indent=2)
# Push to GitHub
if st.session_state.github_sync_enabled:
save_qa_history([])
# Push to Hugging Face
try:
hf_token = os.getenv('HF_TOKEN') or st.secrets['HF_TOKEN']
api = HfApi(token=hf_token)
api.upload_file(
path_or_fileobj="qa_history.json",
path_in_repo="qa_history.json",
repo_id="JirasakJo/Questions_Graduate_Studies_Calendar_2024",
repo_type="space"
)
except Exception as hf_error:
print(f"Hugging Face sync failed: {str(hf_error)}")
except Exception as e:
st.error(f"Error clearing QA history: {str(e)}")
def add_to_history(role: str, message: str):
"""Add message to chat history and save if it's a complete QA pair"""
st.session_state.chat_history.append((role, message))
# If this is an assistant response, save the QA pair
if role == "assistant" and len(st.session_state.chat_history) >= 2:
user_query = st.session_state.chat_history[-2][1]
history_entry = add_to_qa_history(user_query, message)
# Save to storage if GitHub sync is enabled
if st.session_state.github_sync_enabled:
save_qa_history(history_entry)
def display_chat_history():
"""Display chat history with enhanced styling"""
for i, (role, message) in enumerate(st.session_state.chat_history):
if role == "user":
st.markdown(f"""
""", unsafe_allow_html=True)
def initialize_github_sync():
"""
Initialize GitHub repository connection with authentication handling.
"""
try:
# Get and normalize repository path
repo_path = Path(os.getenv('GITHUB_REPO_PATH', '')).resolve()
print(f"Initializing repo at: {repo_path}")
if not repo_path.exists():
print(f"Repository directory not found: {repo_path}")
return False
try:
# Initialize repository
repo = Repo(str(repo_path))
# Verify .git directory
if not (repo_path / '.git').exists():
print(".git directory not found")
return False
# Check remote configuration
try:
origin = repo.remote('origin')
urls = list(origin.urls)
if not urls:
print("Remote 'origin' has no URL")
return False
expected_url = "https://github.com/jirasaksaimekJijo/swu-chat-bot-project.git"
if expected_url not in urls:
print(f"Unexpected remote URL: {urls[0]}")
return False
print(f"Verified remote URL: {urls[0]}")
# Test Git operations
try:
repo.git.status()
print("Git status check passed")
# Try to create a test commit to verify write access
test_file = repo_path / '.git' / 'test_sync'
try:
test_file.touch()
repo.index.add([str(test_file)])
repo.index.commit("Test sync commit")
origin.push()
test_file.unlink() # Clean up test file
print("Push test successful")
except git.GitCommandError as push_error:
if "Authentication failed" in str(push_error):
print("Authentication failed - please check your Git credentials")
return False
print(f"Push test failed: {push_error}")
return False
return True
except Exception as git_error:
print(f"Git operations failed: {str(git_error)}")
return False
except ValueError:
print("Remote 'origin' not configured")
return False
except Exception as repo_error:
print(f"Repository validation failed: {str(repo_error)}")
return False
except Exception as e:
print(f"Initialization failed: {str(e)}")
return False
def main():
# Page config
st.set_page_config(
page_title="Academic Calendar Assistant",
page_icon="📅",
layout="wide",
initial_sidebar_state="collapsed"
)
# Load custom CSS
load_custom_css()
# Initialize session state
if 'pipeline' not in st.session_state:
st.session_state.pipeline = None
if 'chat_history' not in st.session_state:
st.session_state.chat_history = []
if 'github_sync_enabled' not in st.session_state:
st.session_state.github_sync_enabled = initialize_github_sync()
# Load QA history at startup
if 'qa_history_loaded' not in st.session_state:
st.session_state.qa_history_loaded = True
load_qa_history()
# Header
st.markdown("""
🎓 ระบบค้นหาข้อมูลปฏิทินการศึกษา
บัณฑิตวิทยาลัย มหาวิทยาลัยศรีนครินทรวิโรฒ
""", unsafe_allow_html=True)
# Initialize pipeline
if st.session_state.pipeline is None:
with st.spinner("กำลังเริ่มต้นระบบ..."):
st.session_state.pipeline = initialize_pipeline()
# Layout setup
chat_col, info_col = st.columns([7, 3])
with chat_col:
with st.container():
# Display chat history
display_chat_history()
# Query input section
st.markdown("""
""", unsafe_allow_html=True)
query = st.text_input(
"",
placeholder="เช่น: วันสุดท้ายของการสอบปากเปล่าในภาคเรียนที่ 1/2567 คือวันที่เท่าไร?",
key="query_input"
)
# Button layout
col1, col2, col3 = st.columns([1, 1, 4])
with col1:
send_query = st.button(
"📤 ส่งคำถาม",
type="primary",
use_container_width=True,
key="send_query_button"
)
with col2:
clear_history = st.button(
"🗑️ ล้างประวัติ",
type="secondary",
use_container_width=True,
key="clear_history_button"
)
# Process query
if send_query and query:
if st.session_state.pipeline is None:
st.error("❌ ไม่สามารถเชื่อมต่อกับระบบได้ กรุณาลองใหม่อีกครั้ง")
return
add_to_history("user", query)
try:
with st.spinner("🔍 กำลังค้นหาคำตอบ..."):
result = st.session_state.pipeline.process_query(query)
answer = result["answer"]
add_to_history("assistant", answer)
# Save QA pair to history
history_entry = add_to_qa_history(query, answer)
if st.session_state.github_sync_enabled:
if save_qa_history(history_entry):
print("Successfully saved QA history to GitHub")
else:
print("Failed to save QA history to GitHub")
with st.expander("📚 แสดงข้อมูลอ้างอิง", expanded=False):
for i, doc in enumerate(result["documents"], 1):
st.markdown(f"""
เอกสารที่ {i}:
{doc.content}
""", unsafe_allow_html=True)
with st.expander("🔍 รายละเอียดการวิเคราะห์คำถาม", expanded=False):
st.json(result["query_info"])
st.rerun()
except Exception as e:
st.error(f"❌ เกิดข้อผิดพลาด: {str(e)}")
elif send_query and not query:
st.warning("⚠️ กรุณาระบุคำถาม")
# Clear history
if clear_history:
st.session_state.chat_history = []
save_qa_history([]) # Clear saved history
st.rerun()
with info_col:
# System information
st.markdown("""