diff --git a/workspace/cmp_app.py b/workspace/cmp_app.py new file mode 100644 index 0000000..31a1c2f --- /dev/null +++ b/workspace/cmp_app.py @@ -0,0 +1,236 @@ +# app.py (시드 기반 서버 사이드 세션 공유 기능) +import base64 +import json +import shutil +import uuid +from pathlib import Path + +import streamlit as st + +# --- 상수 --- +# 스크립트 파일의 위치를 기준으로 경로 설정 +SESSION_BASE_PATH = Path(__file__).parent / "shared_sessions" + +# --- 헬퍼 함수 --- + + +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 = {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: + matched_pairs[stem] = {"doc_file": doc_file, "json_file": jsons_map[stem]} + return matched_pairs + + +def display_pdf(file_path_or_obj): + """파일 경로 또는 업로드된 파일 객체를 받아 PDF를 표시합니다.""" + try: + 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'' + 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(): + selected_basename_with_index = st.session_state.selectbox_key + 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.markdown("---") + + # 세션 상태 초기화 + 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) + ) + + 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", + 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, + ) + col2.markdown( + f"

{st.session_state.current_index + 1} / {total_files}

", + unsafe_allow_html=True, + ) + col3.button( + "다음 ▶", + on_click=handle_nav_button, + args=("next", total_files), + use_container_width=True, + ) + + current_basename = sorted_basenames[st.session_state.current_index] + st.header(f"🔎 비교 결과: `{current_basename}`") + + selected_pair = matched_files[current_basename] + doc_file = selected_pair["doc_file"] + json_file = selected_pair["json_file"] + + res_col1, res_col2 = st.columns(2) + with res_col1: + st.subheader(f"원본 문서: `{doc_file.name}`") + if doc_file.suffix.lower() == ".pdf": + display_pdf(doc_file) + else: + st.image( + str(doc_file), + caption=f"원본 이미지: {doc_file.name}", + use_container_width=True, + ) + + with res_col2: + st.subheader(f"추출된 데이터: `{json_file.name}`") + try: + 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}") + + +if __name__ == "__main__": + main()