import time
import gradio as gr
import random
import os
import threading
import pandas as pd
from datasets import load_dataset
# CSV 파일 경로와 동시 접근을 위한 Lock 선언
DATA_FILE = "Interface1.csv"
data_lock = threading.Lock()
def initialize_global_data():
"""
DATA_FILE이 존재하지 않으면, gaeunseo/Taskmaster_sample_data 데이터셋을 로드하여 DataFrame으로 변환한 후 CSV 파일로 저장합니다.
이미 파일이 있으면 파일에서 데이터를 읽어 DataFrame을 반환합니다.
"""
if not os.path.exists(DATA_FILE):
ds = load_dataset("gaeunseo/Taskmaster_sample_data", split="train")
data = ds.to_pandas()
# 필요한 컬럼이 없으면 추가합니다.
if "used" not in data.columns:
data["used"] = False
if "overlapping" not in data.columns:
data["overlapping"] = ""
if "text" not in data.columns:
data["text"] = ""
data.to_csv(DATA_FILE, index=False)
return data
else:
with data_lock:
df = pd.read_csv(DATA_FILE)
return df
def load_global_data():
"""CSV 파일에서 global_data DataFrame을 읽어옵니다."""
with data_lock:
df = pd.read_csv(DATA_FILE)
return df
def save_global_data(df):
"""DataFrame을 CSV 파일에 저장합니다."""
with data_lock:
df.to_csv(DATA_FILE, index=False)
# CSV 파일에 저장된 global_data 초기화
global_data = initialize_global_data()
def get_random_row_from_dataset():
"""
CSV 파일에 저장된 global_data에서,
1. conversation_id별로 그룹화하고,
2. 각 그룹에서 모든 행의 used 컬럼이 False이며, 그룹 내에 overlapping 컬럼이 "TT"인 행이 존재하는 그룹만 valid로 간주합니다.
valid한 그룹들 중 랜덤하게 하나의 그룹을 선택한 후,
- 해당 그룹의 모든 행의 used 값을 True로 업데이트(즉, 전체 그룹을 할당)하고 CSV 파일에 저장합니다.
- 선택된 그룹 내에서 overlapping 컬럼이 "TT"인 행(여러 개라면 첫 번째)을 반환합니다.
"""
global global_data
global_data = load_global_data() # 최신 데이터 로드
groups = global_data.groupby('conversation_id')
valid_groups = []
for cid, group in groups:
# 모든 행의 used 값이 False이고, 그룹 내에 overlapping 값이 "TT"인 행이 있는 그룹 필터링
if group['used'].apply(lambda x: bool(x) == False).all() and (group['overlapping'] == "TT").any():
valid_groups.append((cid, group))
if not valid_groups:
return None
chosen_cid, chosen_group = random.choice(valid_groups)
# 전체 그룹의 used 값을 True로 업데이트
global_data.loc[global_data['conversation_id'] == chosen_cid, 'used'] = True
save_global_data(global_data)
# 선택된 그룹 내에서 overlapping 컬럼이 "TT"인 행을 반환 (여러 개라면 첫 번째)
chosen_rows = chosen_group[chosen_group['overlapping'] == "TT"]
if chosen_rows.empty:
return None
chosen_row = chosen_rows.iloc[0]
return chosen_row.to_dict()
# --- 초기 대화 불러오기 ---
# 데이터셋의 text 컬럼은 "[turn]"을 기준으로 대화가 구분되어 있다고 가정합니다.
row = get_random_row_from_dataset()
if row is None:
human_message = "No valid conversation available."
ai_message = "No valid conversation available."
else:
raw_text = row['text']
human_message = raw_text.split("[turn]")[0].strip()
ai_message = raw_text.split("[turn]")[1].strip()
#############################################
# 채팅 인터페이스 관련 함수 (말풍선, 타이핑 효과, 편집 기능)
#############################################
def get_initial_human_html():
"""
페이지 로드 시, 빈 Human 말풍선과 오른쪽 🧑 이모티콘을 포함한 초기 HTML 반환
"""
wrapper_start = (
""""""
)
bubble_start = """
"""
bubble_end = "
"
emoji_html = "
🧑
"
wrapper_end = "
"""
)
bubble_start = """
"""
bubble_end = "
"
emoji_html = "
🧑
"
wrapper_end = "