chan/ocr_macro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

호환성 수정

Browse Source
This commit is contained in:
조찬영
2025-08-01 09:44:54 +09:00
parent 486c2ceeb8
commit e62ac36fb8
7 changed files with 227 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
workspace/.env-smaple Normal file
View File

@@ -0,0 +1,2 @@
BASE_URL=""
API_KEY=""

187
workspace/app.py
View File

@@ -1,134 +1,125 @@
# app.py (프로젝트별 디렉터리 지원 버전)
# app.py (파일 업로드 지원 버전)
import streamlit as st
import json
import os
import base64
from pathlib import Path
# --- 설정 ---
DOCS_DIR = Path("/data/documents")
JSON_DIR = Path("/data/jsons")
import base64
# --- 헬퍼 함수 ---
def scan_project_directories(docs_base_dir, json_base_dir):
def match_uploaded_files(doc_files, json_files):
"""
두 베이스 디렉터리를 스캔하여, 공통된 서브디렉터리(프로젝트)를 찾고
그 안의 파일 쌍을 매핑한 딕셔너리를 반환합니다.
업로드된 두 파일 목록을 받아, 이름(확장자 제외)을 기준으로 매칭하고
결과를 딕셔너리로 반환합니다.
"""
projects_data = {}
if not docs_base_dir.is_dir():
return projects_data
matched_pairs = {}
# 각 파일 목록을 이름(stem)을 키로 하는 딕셔너리로 변환
docs_map = {Path(f.name).stem: f for f in doc_files}
jsons_map = {Path(f.name).stem: f for f in json_files}
# 문서 디렉터리 기준으로 서브디렉터리(프로젝트)를 찾음
for project_path in docs_base_dir.iterdir():
if project_path.is_dir():
project_name = project_path.name
json_project_path = json_base_dir / project_name
# 문서 파일 기준으로 JSON 파일 찾기
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]
}
# JSON 디렉터리에도 해당 프로젝트 폴더가 있는지 확인
if json_project_path.is_dir():
# 프로젝트 내에서 파일 쌍 매칭
doc_files = {f.stem: f for f in project_path.iterdir() if f.is_file()}
json_files = {f.stem: f for f in json_project_path.iterdir() if f.is_file() and f.suffix == '.json'}
matching_pairs = {}
for base_name, doc_path in doc_files.items():
if base_name in json_files:
matching_pairs[base_name] = {
"doc_path": doc_path,
"json_path": json_files[base_name]
}
if matching_pairs:
projects_data[project_name] = matching_pairs
return projects_data
return matched_pairs
def display_pdf(file_path):
"""PDF 파일을 웹 페이지에 임베드하여 표시합니다."""
with open(file_path, "rb") as f:
base64_pdf = base64.b64encode(f.read()).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)
def display_pdf(file_object):
"""
업로드된 파일 객체(UploadedFile)를 읽어 PDF를 표시합니다.
"""
try:
# 파일 포인터를 처음으로 되돌림 (중요)
file_object.seek(0)
base64_pdf = base64.b64encode(file_object.read()).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}")
# --- 메인 UI 로직 ---
def main():
st.set_page_config(layout="wide", page_title="결과 비교 도구")
st.title("🗂️ 프로젝트별 결과 비교 도구")
st.title("📑 파일 업로드 기반 결과 비교 도구")
st.markdown("---")
if not DOCS_DIR.is_dir() or not JSON_DIR.is_dir():
st.error(f"오류: 데이터 루트 디렉터리(`{DOCS_DIR}` 또는 `{JSON_DIR}`)를 찾을 수 없습니다.")
# --- 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:
projects_data = scan_project_directories(DOCS_DIR, JSON_DIR)
matched_files = match_uploaded_files(uploaded_docs, uploaded_jsons)
except Exception as e:
st.error(f"프로젝트 목록을 읽는 중 오류가 발생했습니다: {e}")
st.error(f"업로드된 파일을 매칭하는 중 오류가 발생했습니다: {e}")
return
if not projects_data:
st.warning("비교할 프로젝트가 없습니다. 각 데이터 디렉터리 안에 동일한 이름의 하위 폴더가 있는지 확인하세요.")
if not matched_files:
st.warning("업로드된 파일 중 일치하는 문서-JSON 쌍을 찾을 수 없습니다. 파일 이름(확장자 제외)이 동일한지 확인하세요.")
return
# --- 1. 프로젝트 선택 ---
# --- 2. 파일 선택 ---
st.sidebar.header("파일 선택")
project_names = sorted(list(projects_data.keys()))
selected_project = st.sidebar.selectbox(
"1. 프로젝트를 선택하세요.",
project_names
sorted_basenames = sorted(list(matched_files.keys()))
selected_basename = st.sidebar.selectbox(
"비교할 파일을 선택하세요.",
sorted_basenames
)
if selected_project:
files_in_project = projects_data[selected_project]
if selected_basename:
st.header(f"🔎 비교 결과: `{selected_basename}`")
# --- 2. 파일 선택 ---
sorted_basenames = sorted(list(files_in_project.keys()))
display_options = [f"{i}. {name}" for i, name in enumerate(sorted_basenames, 1)]
selected_option = st.sidebar.selectbox(
f"2. '{selected_project}' 프로젝트의 파일을 선택하세요.",
display_options
)
selected_pair = matched_files[selected_basename]
doc_file = selected_pair["doc_file"]
json_file = selected_pair["json_file"]
if selected_option:
original_basename = selected_option.split('. ', 1)[1]
st.header(f"🔎 비교 결과: `{selected_project} / {original_basename}`")
# --- 결과 표시 ---
col1, col2 = st.columns(2)
with col1:
st.subheader(f"원본 문서: `{doc_file.name}`")
doc_suffix = Path(doc_file.name).suffix.lower()
selected_pair = files_in_project[original_basename]
doc_path = selected_pair["doc_path"]
json_path = selected_pair["json_path"]
if doc_suffix == ".pdf":
display_pdf(doc_file)
elif doc_suffix in ['.png', '.jpg', '.jpeg']:
st.image(doc_file, caption=f"원본 이미지: {doc_file.name}", use_container_width=True)
else:
st.warning("지원하지 않는 문서 형식입니다.")
# --- 결과 표시 (이전과 동일) ---
col1, col2 = st.columns(2)
with col1:
st.subheader("원본 문서")
try:
if doc_path.suffix.lower() == ".pdf":
display_pdf(doc_path)
elif doc_path.suffix.lower() in ['.png', '.jpg', '.jpeg']:
st.image(str(doc_path), caption=f"원본 이미지: {doc_path.name}", use_container_width=True)
else:
st.warning(f"지원하지 않는 문서 형식입니다: {doc_path.name}")
except Exception as e:
st.error(f"문서 파일을 표시하는 중 오류가 발생했습니다: {e}")
with col2:
st.subheader("추출된 데이터 (JSON)")
try:
with open(json_path, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, list) and len(data) > 0:
result_item = data[0]
else:
result_item = data
if 'fields' in result_item:
del result_item['fields']
st.json(result_item)
except Exception as e:
st.error(f"JSON 파일을 읽거나 처리하는 중 오류가 발생했습니다: {e}")
with col2:
st.subheader(f"추출된 데이터: `{json_file.name}`")
try:
# 파일 포인터를 처음으로 되돌림
json_file.seek(0)
data = json.load(json_file)
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()

31
workspace/process_directory.py
View File

@@ -6,12 +6,7 @@ import argparse
import sys
from urllib.parse import urljoin
import logging
# --- 설정 ---
BASE_URL = "http://172.16.10.176:8888"
API_KEY = 'sk-e03e060ea4ee8edf2e057fbff3e68c28'
RETRY_COUNT_ON_404 = 3
RETRY_DELAY_ON_404 = 5
from dotenv import load_dotenv
# --- 로거 설정 ---
# 전역 로거 객체 생성
@@ -54,10 +49,12 @@ def start_extraction(post_url, file_path, filename, headers, model_name=None):
logger.exception(f"[{filename}] POST 요청 중 오류 발생")
return None
def check_progress(progress_path, filename, headers):
def check_progress(base_url, progress_path, filename, headers):
"""GET /extract/progress/{request_id}: 진행 상태 확인 (로깅 적용)"""
get_url = urljoin(BASE_URL + '/', progress_path.lstrip('/'))
get_url = urljoin(base_url + '/', progress_path.lstrip('/'))
RETRY_COUNT_ON_404 = 3
RETRY_DELAY_ON_404 = 5
retries_left = RETRY_COUNT_ON_404
last_status = ""
@@ -99,10 +96,22 @@ def check_progress(progress_path, filename, headers):
# --- 메인 실행 로직 ---
def main():
# .env 파일에서 환경 변수 로드 (workspace 디렉터리 기준)
dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
load_dotenv(dotenv_path=dotenv_path)
# 로거를 가장 먼저 설정합니다.
setup_logger()
parser = argparse.ArgumentParser(description="문서 정보 추출 자동화 스크립트 (로깅 적용)")
# 환경 변수에서 API 정보 가져오기
BASE_URL = os.getenv("BASE_URL")
API_KEY = os.getenv("API_KEY")
if not BASE_URL or not API_KEY:
logger.error("환경 변수(BASE_URL, API_KEY)가 설정되지 않았습니다. workspace/.env 파일을 확인하세요.")
return
parser = argparse.ArgumentParser(description="문서 정보 추출 자동화 스크립트")
parser.add_argument("input_dir", help="입력 디렉터리 경로")
parser.add_argument("-o", "--output_dir", default="results", help="출력 디렉터리 경로")
parser.add_argument("--endpoint", choices=['i18n', 'd6c'], default='i18n', help="추출 API 엔드포인트 선택 (i18n 또는 d6c)")
@@ -145,7 +154,7 @@ def main():
logger.info(f"[{filename}] 작업 요청 성공. Request ID: {request_id}")
final_result = check_progress(status_check_url, filename, headers)
final_result = check_progress(BASE_URL, status_check_url, filename, headers)
if final_result:
output_path = os.path.join(args.output_dir, f"{os.path.splitext(filename)[0]}.json")
@@ -165,4 +174,4 @@ if __name__ == "__main__":
main()
except KeyboardInterrupt:
# KeyboardInterrupt는 main 밖에서 처리해야 할 수 있으므로 로거를 직접 호출
logging.getLogger(__name__).warning("사용자에 의해 작업이 중단되었습니다.")
logging.getLogger(__name__).warning("사용자에 의해 작업이 중단되었습니다.")