파일분류, csv output, 세션유지 기능 추가
This commit is contained in:
203
workspace/app.py
203
workspace/app.py
@@ -1,19 +1,59 @@
|
||||
# app.py (개별 네비게이션 및 선택 드롭다운 동시 지원)
|
||||
|
||||
# app.py (시드 기반 서버 사이드 세션 공유 기능)
|
||||
import streamlit as st
|
||||
import json
|
||||
from pathlib import Path
|
||||
import base64
|
||||
import uuid
|
||||
import shutil
|
||||
|
||||
# --- 상수 ---
|
||||
# 스크립트 파일의 위치를 기준으로 경로 설정
|
||||
SESSION_BASE_PATH = Path(__file__).parent / "shared_sessions"
|
||||
|
||||
# --- 헬퍼 함수 ---
|
||||
|
||||
def match_uploaded_files(doc_files, json_files):
|
||||
"""
|
||||
업로드된 두 파일 목록을 받아, 이름(확장자 제외)을 기준으로 매칭하고
|
||||
결과를 딕셔너리로 반환합니다.
|
||||
"""
|
||||
def get_session_path(seed):
|
||||
"""시드에 해당하는 세션 디렉토리 경로를 반환합니다."""
|
||||
return SESSION_BASE_PATH / seed
|
||||
|
||||
def save_files_to_session(seed, doc_files, json_files):
|
||||
"""업로드된 파일들을 서버의 세션 디렉토리에 저장합니다."""
|
||||
session_path = get_session_path(seed)
|
||||
doc_path = session_path / "docs"
|
||||
json_path = session_path / "jsons"
|
||||
|
||||
# 기존 디렉토리가 있으면 삭제하고 새로 생성
|
||||
if session_path.exists():
|
||||
shutil.rmtree(session_path)
|
||||
doc_path.mkdir(parents=True, exist_ok=True)
|
||||
json_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for file in doc_files:
|
||||
with open(doc_path / file.name, "wb") as f:
|
||||
f.write(file.getbuffer())
|
||||
for file in json_files:
|
||||
with open(json_path / file.name, "wb") as f:
|
||||
f.write(file.getbuffer())
|
||||
|
||||
def load_files_from_session(seed):
|
||||
"""서버의 세션 디렉토리에서 파일 목록을 로드합니다."""
|
||||
session_path = get_session_path(seed)
|
||||
doc_path = session_path / "docs"
|
||||
json_path = session_path / "jsons"
|
||||
|
||||
if not session_path.is_dir():
|
||||
return None, None
|
||||
|
||||
doc_files = sorted(list(doc_path.iterdir()))
|
||||
json_files = sorted(list(json_path.iterdir()))
|
||||
return doc_files, json_files
|
||||
|
||||
def match_disk_files(doc_files, json_files):
|
||||
"""디스크에 저장된 두 파일 목록(Path 객체)을 매칭합니다."""
|
||||
matched_pairs = {}
|
||||
docs_map = {Path(f.name).stem: f for f in doc_files}
|
||||
jsons_map = {Path(f.name).stem: f for f in json_files}
|
||||
docs_map = {f.stem: f for f in doc_files}
|
||||
jsons_map = {f.stem: f for f in json_files}
|
||||
|
||||
for stem, doc_file in docs_map.items():
|
||||
if stem in jsons_map:
|
||||
@@ -23,111 +63,116 @@ def match_uploaded_files(doc_files, json_files):
|
||||
}
|
||||
return matched_pairs
|
||||
|
||||
def display_pdf(file_object):
|
||||
"""
|
||||
업로드된 파일 객체(UploadedFile)를 읽어 PDF를 표시합니다.
|
||||
"""
|
||||
def display_pdf(file_path_or_obj):
|
||||
"""파일 경로 또는 업로드된 파일 객체를 받아 PDF를 표시합니다."""
|
||||
try:
|
||||
file_object.seek(0)
|
||||
base64_pdf = base64.b64encode(file_object.read()).decode('utf-8')
|
||||
if isinstance(file_path_or_obj, Path):
|
||||
with open(file_path_or_obj, "rb") as f:
|
||||
bytes_data = f.read()
|
||||
else: # UploadedFile
|
||||
file_path_or_obj.seek(0)
|
||||
bytes_data = file_path_or_obj.read()
|
||||
|
||||
base64_pdf = base64.b64encode(bytes_data).decode('utf-8')
|
||||
pdf_display = f'<iframe src="data:application/pdf;base64,{base64_pdf}" width="100%" height="800" type="application/pdf"></iframe>'
|
||||
st.markdown(pdf_display, unsafe_allow_html=True)
|
||||
except Exception as e:
|
||||
st.error(f"PDF 파일을 표시하는 중 오류가 발생했습니다: {e}")
|
||||
|
||||
# --- 콜백 함수 ---
|
||||
|
||||
def handle_nav_button(direction, total_files):
|
||||
"""이전/다음 버튼 클릭을 처리하는 콜백"""
|
||||
if direction == "prev" and st.session_state.current_index > 0:
|
||||
st.session_state.current_index -= 1
|
||||
elif direction == "next" and st.session_state.current_index < total_files - 1:
|
||||
st.session_state.current_index += 1
|
||||
|
||||
def handle_selectbox_change():
|
||||
"""selectbox 변경을 처리하는 콜백"""
|
||||
selected_basename_with_index = st.session_state.selectbox_key
|
||||
# "1. filename" 형식에서 인덱스만 추출
|
||||
new_index = int(selected_basename_with_index.split('. ', 1)[0]) - 1
|
||||
st.session_state.current_index = new_index
|
||||
|
||||
# --- 메인 UI 로직 ---
|
||||
|
||||
def main():
|
||||
st.set_page_config(layout="wide", page_title="결과 비교 도구")
|
||||
st.title("📑 파일 업로드 기반 결과 비교 도구")
|
||||
st.title("📑 결과 비교 및 공유 도구")
|
||||
st.markdown("---")
|
||||
|
||||
# --- 1. 파일 업로드 ---
|
||||
st.sidebar.header("파일 업로드")
|
||||
|
||||
uploaded_docs = st.sidebar.file_uploader(
|
||||
"1. 원본 문서 파일(들)을 업로드하세요.",
|
||||
accept_multiple_files=True,
|
||||
type=['png', 'jpg', 'jpeg', 'pdf']
|
||||
)
|
||||
|
||||
uploaded_jsons = st.sidebar.file_uploader(
|
||||
"2. 결과 JSON 파일(들)을 업로드하세요.",
|
||||
accept_multiple_files=True,
|
||||
type=['json']
|
||||
)
|
||||
|
||||
if not uploaded_docs or not uploaded_jsons:
|
||||
st.info("사이드바에서 원본 문서와 결과 JSON 파일을 모두 업로드해주세요.")
|
||||
return
|
||||
|
||||
try:
|
||||
matched_files = match_uploaded_files(uploaded_docs, uploaded_jsons)
|
||||
except Exception as e:
|
||||
st.error(f"업로드된 파일을 매칭하는 중 오류가 발생했습니다: {e}")
|
||||
return
|
||||
|
||||
if not matched_files:
|
||||
st.warning("업로드된 파일 중 일치하는 문서-JSON 쌍을 찾을 수 없습니다. 파일 이름(확장자 제외)이 동일한지 확인하세요.")
|
||||
return
|
||||
|
||||
# --- 2. 파일 네비게이션 및 선택 ---
|
||||
st.sidebar.header("파일 탐색")
|
||||
|
||||
sorted_basenames = sorted(list(matched_files.keys()))
|
||||
total_files = len(sorted_basenames)
|
||||
|
||||
# 세션 상태 초기화
|
||||
if 'current_index' not in st.session_state:
|
||||
st.session_state.current_index = 0
|
||||
|
||||
# 세션 저장 기본 경로 생성
|
||||
SESSION_BASE_PATH.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
matched_files = None
|
||||
doc_files, json_files = None, None
|
||||
|
||||
# URL에서 시드 확인
|
||||
query_params = st.query_params
|
||||
url_seed = query_params.get("seed")
|
||||
|
||||
if url_seed:
|
||||
doc_files, json_files = load_files_from_session(url_seed)
|
||||
if doc_files is None:
|
||||
st.error(f"'{url_seed}'에 해당하는 공유 세션을 찾을 수 없습니다. 시드가 정확한지 확인하거나, 파일을 새로 업로드하세요.")
|
||||
else:
|
||||
st.success(f"'{url_seed}' 시드에서 공유된 파일을 불러왔습니다.")
|
||||
matched_files = match_disk_files(doc_files, json_files)
|
||||
|
||||
# 인덱스가 유효한 범위를 벗어나지 않도록 조정
|
||||
# 시드가 없거나, 시드로 로드 실패 시 파일 업로더 표시
|
||||
if not matched_files:
|
||||
st.sidebar.header("파일 업로드")
|
||||
uploaded_docs = st.sidebar.file_uploader(
|
||||
"1. 원본 문서 파일(들)을 업로드하세요.",
|
||||
accept_multiple_files=True,
|
||||
type=['png', 'jpg', 'jpeg', 'pdf']
|
||||
)
|
||||
uploaded_jsons = st.sidebar.file_uploader(
|
||||
"2. 결과 JSON 파일(들)을 업로드하세요.",
|
||||
accept_multiple_files=True,
|
||||
type=['json']
|
||||
)
|
||||
|
||||
if uploaded_docs and uploaded_jsons:
|
||||
if st.sidebar.button("업로드 및 세션 생성"):
|
||||
new_seed = str(uuid.uuid4())[:8]
|
||||
save_files_to_session(new_seed, uploaded_docs, uploaded_jsons)
|
||||
st.query_params["seed"] = new_seed # URL 업데이트 및 앱 재실행
|
||||
st.rerun()
|
||||
|
||||
# 공유 UI
|
||||
if url_seed and matched_files:
|
||||
st.sidebar.header("세션 공유")
|
||||
# 현재 페이지의 전체 URL을 가져오는 것은 Streamlit에서 직접 지원하지 않으므로,
|
||||
# 사용자에게 주소창의 URL을 복사하라고 안내합니다.
|
||||
st.sidebar.success("세션이 활성화되었습니다!")
|
||||
st.sidebar.info("다른 사람과 공유하려면 현재 브라우저 주소창의 URL을 복사하여 전달하세요.")
|
||||
st.sidebar.text_input("공유 시드", url_seed, disabled=True)
|
||||
|
||||
# --- 결과 표시 로직 (matched_files가 있을 때만 실행) ---
|
||||
if not matched_files:
|
||||
st.info("사이드바에서 파일을 업로드하고 '업로드 및 세션 생성' 버튼을 누르거나, 공유받은 URL로 접속하세요.")
|
||||
return
|
||||
|
||||
st.sidebar.header("파일 탐색")
|
||||
sorted_basenames = sorted(list(matched_files.keys()))
|
||||
total_files = len(sorted_basenames)
|
||||
st.session_state.current_index = max(0, min(st.session_state.current_index, total_files - 1))
|
||||
|
||||
# Selectbox
|
||||
display_options = [f"{i + 1}. {name}" for i, name in enumerate(sorted_basenames)]
|
||||
st.selectbox(
|
||||
"파일을 직접 선택하세요:",
|
||||
options=display_options,
|
||||
index=st.session_state.current_index,
|
||||
key='selectbox_key', # 콜백에서 값을 참조하기 위한 키
|
||||
key='selectbox_key',
|
||||
on_change=handle_selectbox_change
|
||||
)
|
||||
|
||||
# 네비게이션 컨트롤
|
||||
col1, col2, col3 = st.sidebar.columns([1, 2, 1])
|
||||
|
||||
col1.button(
|
||||
"◀ 이전",
|
||||
on_click=handle_nav_button,
|
||||
args=("prev", total_files),
|
||||
use_container_width=True
|
||||
)
|
||||
col1.button("◀ 이전", on_click=handle_nav_button, args=("prev", total_files), use_container_width=True)
|
||||
col2.markdown(f"<p style='text-align: center;'>{st.session_state.current_index + 1} / {total_files}</p>", unsafe_allow_html=True)
|
||||
col3.button(
|
||||
"다음 ▶",
|
||||
on_click=handle_nav_button,
|
||||
args=("next", total_files),
|
||||
use_container_width=True
|
||||
)
|
||||
col3.button("다음 ▶", on_click=handle_nav_button, args=("next", total_files), use_container_width=True)
|
||||
|
||||
# --- 3. 결과 표시 ---
|
||||
current_basename = sorted_basenames[st.session_state.current_index]
|
||||
st.header(f"🔎 비교 결과: `{current_basename}`")
|
||||
|
||||
@@ -138,24 +183,20 @@ def main():
|
||||
res_col1, res_col2 = st.columns(2)
|
||||
with res_col1:
|
||||
st.subheader(f"원본 문서: `{doc_file.name}`")
|
||||
doc_suffix = Path(doc_file.name).suffix.lower()
|
||||
|
||||
if doc_suffix == ".pdf":
|
||||
if doc_file.suffix.lower() == ".pdf":
|
||||
display_pdf(doc_file)
|
||||
else:
|
||||
st.image(doc_file, caption=f"원본 이미지: {doc_file.name}", use_container_width=True)
|
||||
st.image(str(doc_file), caption=f"원본 이미지: {doc_file.name}", use_container_width=True)
|
||||
|
||||
with res_col2:
|
||||
st.subheader(f"추출된 데이터: `{json_file.name}`")
|
||||
try:
|
||||
json_file.seek(0)
|
||||
data = json.load(json_file)
|
||||
with open(json_file, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
result_to_display = data[0] if isinstance(data, list) and data else data
|
||||
|
||||
if isinstance(result_to_display, dict) and 'fields' in result_to_display:
|
||||
del result_to_display['fields']
|
||||
|
||||
st.json(result_to_display)
|
||||
except Exception as e:
|
||||
st.error(f"JSON 파일을 읽거나 처리하는 중 오류가 발생했습니다: {e}")
|
||||
|
||||
Reference in New Issue
Block a user