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

Compare commits

base: chan:1b96840c83d8d7929d3197f731725687b97ef575
Branches Tags
chan:main
...
compare: chan:7cf62f58750353d2ef08e59091a9679ec011ce39
Branches Tags
chan:main

4 Commits
1b96840c83 ... 7cf62f5875

Author SHA1 Message Date
chan
7cf62f5875 리드미 추가 2025-08-12 16:59:40 +09:00
chan
68bd5fb719 기존 ocr, json 비교 2025-08-12 16:59:34 +09:00
chan
40785a1b30 ui 개선 및 정답셋 편집 추가 2025-08-12 16:59:14 +09:00
chan
343ae4958a 새로운엔드포인트 추가 2025-08-12 16:58:38 +09:00
13 changed files with 1378 additions and 606 deletions
Expand all files Collapse all files

5
.gitignore vendored
View File

@@ -31,4 +31,7 @@ ENV/
/results
script_run.log
/ocr_data
/workspace/shared_sessions
/workspace/shared_sessions
/filtered_data
/josn
/data

103
README.md
View File

@@ -1,88 +1,73 @@
# 문서 정보 추출 및 비교 도구
# OCR Macro: 정답셋 워크플로우
이 프로젝트는 문서(이미지, PDF)에서 정보를 추출하고, 그 결과를 원본 문서와 나란히 비교할 수 있는 도구를 제공합니다.
이 프로젝트는 여러 OCR 모델의 결과물을 효율적으로 비교하고 수정하여, 고품질의 정답셋(Ground Truth) 데이터셋을 구축하기 위한 일련의 도구와 워크플로우를 제공합니다.
## 구성 요소
## 프로젝트 개요
1. **추출 스크립트 (`workspace/process_directory.py`)**: 지정된 디렉터리의 모든 파일에 대해 정보 추출 API를 호출하고 결과를 JSON 파일로 저장합니다.
2. **비교 웹앱 (`workspace/app.py`)**: Streamlit 기반의 웹 애플리케이션으로, 원본 문서와 추출된 JSON 결과를 업로드하여 시각적으로 비교할 수 있습니다.
프로젝트는 크게 두 가지 핵심 도구로 구성됩니다.
---
1. **`run_ocr.py`**: 대량의 문서를 OCR API로 보내고 결과를 자동으로 저장하는 CLI 스크립트.
2. **`app.py`**: OCR 결과물을 원본과 비교하며 수정하고, 정답셋을 생성하는 Streamlit 기반 웹 애플리케이션.
## 1. 추출 스크립트 사용법 (`process_directory.py`)
## 구성 요소 상세
### 사전 준비
### 1. `workspace/run_ocr.py`
스크립트를 실행하기 위해서는 Python 3.6 이상이 필요합니다.
대량의 문서에 대한 OCR을 자동화하는 커맨드 라인 인터페이스(CLI) 스크립트입니다.
#### 1.1. 환경 변수 설정
- **주요 기능**: 병렬 처리, 자동 재시도(Exponential Backoff), OCR 공급자 선택(`paddle`/`upstage`), 순차 처리 모드(API의 RPS 제한 대응).
- **사용법**: `run_ocr_readme.md` 파일 참조.
API 서버의 정보는 민감 정보이므로, `workspace` 디렉터리 안에 `.env` 파일을 생성하여 관리합니다.
### 2. `workspace/app.py`
`.env` 파일 예시 (`workspace/.env`):
OCR 결과물을 시각적으로 비교하고 수정하여 정답셋을 생성하는 Streamlit 기반 웹 애플리케이션입니다.
```env
BASE_URL="http://172.0.0.1:8888"
API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"
```
- **주요 기능**: 세션 기반 협업, 3-Way 비교 뷰, 동적 레이아웃, 작업 완료 상태 추적, 정답셋 재수정, 결과물 zip 다운로드.
- **사용법**: `workspace/app_readme.md` 파일 참조.
**중요:** 이 `.env` 파일은 `.gitignore`에 의해 버전 관리에서 자동으로 제외됩니다.
## 설치 및 실행
#### 1.2. 의존성 설치
### 설치
프로젝트 루트 디렉터리에서 다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.
프로젝트의 루트 디렉토리에서 다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.
```bash
pip install -r requirements.txt
```
### 실행 방법
### 실행
프로젝트 루트 디렉터리에서 다음 형식으로 스크립트를 실행합니다.
#### OCR 자동화 스크립트
```bash
python workspace/process_directory.py [입력_디렉터리] [옵션]
# 예시: upstage 모델을 사용하여 3초 간격으로 순차 처리
python workspace/run_ocr.py ./source_documents ./result_jsons/upstage --provider upstage --delay 3
```
#### 인자 설명
#### 정답셋 생성 웹 애플리케이션
- `input_dir` (필수): 처리할 파일들이 들어있는 입력 디렉터리의 경로입니다.
- `-o, --output_dir` (선택): 결과 JSON 파일들을 저장할 출력 디렉터리입니다. (기본값: `results`)
- `--endpoint` (선택): 호출할 API 엔드포인트를 지정합니다. (`i18n` 또는 `d6c`, 기본값: `i18n`)
- `--model` (선택): 사용할 특정 LLM 모델의 이름을 지정합니다.
#### 실행 예시
```bash
# 'source_documents/data'에 있는 파일들을 'i18n' 엔드포인트로 처리
python workspace/process_directory.py source_documents/data
# 'd6c' 엔드포인트를 사용하여 처리하고 결과를 'my_results' 폴더에 저장
python workspace/process_directory.py source_documents/data2 --endpoint d6c -o my_results
```dockerfile
docker compose up
```
---
## 디렉토리 구조 (권장)
## 2. 비교 웹앱 사용법 (`app.py`)
이 웹앱을 사용하여 원본 문서와 `process_directory.py` 실행 결과로 생성된 JSON 파일을 시각적으로 비교할 수 있습니다.
### 실행 방법
프로젝트 루트 디렉터리에서 다음 명령어를 실행합니다.
```bash
streamlit run workspace/app.py
or
docker copmose up
```
위 명령어를 실행하면 웹 브라우저에서 비교 도구가 열립니다.
### 사용 절차
1. 웹앱이 실행되면 사이드바에 파일 업로드 영역이 나타납니다.
2. **"원본 문서 파일(들)을 업로드하세요."** 버튼을 클릭하여 하나 이상의 문서 파일(PDF, PNG, JPG 등)을 업로드합니다.
3. **"결과 JSON 파일(들)을 업로드하세요."** 버튼을 클릭하여 해당 문서들의 추출 결과인 JSON 파일들을 업로드합니다.
4. 파일들이 성공적으로 매칭되면, **"비교할 파일을 선택하세요."** 드롭다운 메뉴에 파일 목록이 나타납니다.
5. 목록에서 파일을 선택하면, 왼쪽에는 원본 문서가, 오른쪽에는 JSON 데이터가 표시되어 내용을 비교할 수 있습니다.
ocr_macro/
├── source_documents/ # 원본 이미지/PDF 파일
├── result_jsons/
│ ├── paddle_ocr/ # paddle_ocr 결과 JSON 파일
│ └── upstage/ # upstage 결과 JSON 파일
├── workspace/
│ ├── app.py # 정답셋 생성 웹 앱
│ ├── run_ocr.py # OCR 자동화 스크립트
│ ├── shared_sessions/ # app.py의 작업 세션 데이터 (자동 생성)
│ │ └── {seed}/
│ │ ├── docs/
│ │ ├── jsons_paddle_ocr/
│ │ ├── jsons_upstage/
│ │ └── groundtruth/ # 최종 정답셋 (자동 생성)
│ └── ...
├── requirements.txt
└── README.md # 본 파일
```

112
copy_files.py
View File

@@ -1,112 +0,0 @@
import os
import shutil
def copy_target_files_with_structure():
# 소스 디렉토리와 대상 디렉토리 설정
root_source_dir = '/home/jackjack/test/ocr_macro/ocr_data'
dest_dir = '/home/jackjack/test/ocr_macro/filtered_data'
# 복사할 파일 목록
files_to_copy = [
"20250701101504-789-402-926.jpg", "20250707164925-895-935-673.jpg",
"20250708092450-131-769-592.jpg", "20250708124502-268-927-842.jpg",
"20250709105123-169-457-765.jpg", "20250714150847-882-936-950.jpg",
"20250721090407-508-883-473.jpg", "20250724145851-721-283-914.jpg",
"20250729105852-697-150-153.jpg", "20250730180509-798-917-821.jpg",
"20170619133715-134-321-633.jpg", "20171017141811-255-321-370.jpg",
"20180103094436-462-212-348.jpg", "20180131103459-868-481-465.jpg",
"20180411134455-600-132-301.jpg", "20180412092830-356-712-939.jpg",
"20180807102155-126-746-229.jpg", "20190507165642-222-795-363.jpg",
"20191227103340-434-827-409.jpg", "20200113103330-999-251-437.png",
"20200313140454-282-318-706.jpg", "20201203162517-973-818-382.jpg",
"20250305150305-354-816-193.jpg", "2018-0319102207-217049.pdf",
"2018-0319114254-217049.pdf", "2021-0713114710-219044.pdf",
"2021-0713114843-219044.pdf", "2024-1129132456-223033.pdf",
"2024-1202134504-223033.pdf", "2024-1202134828-223033.pdf",
"2024-1216141625-211046.pdf", "2024-1231131430-223033.pdf",
"2025-0102114806-223033.pdf", "2025-0102115602-223033.pdf",
"20250715092937-779-181-466.jpg", "20250715110944-951-537-524.jpg",
"20250715111622-358-588-698.jpg", "20250715112411-186-289-669.jpg",
"20250715135137-801-844-961.jpg", "20250715161950-712-251-637.jpg",
"20250715162045-552-568-375.jpg", "20250715165509-176-474-591.jpg",
"20250715172557-573-573-629.jpg", "20250716093130-913-217-747.jpg",
"20250716105706-162-939-389.jpg", "20250716110134-808-994-942.jpg",
"20250716134023-322-796-383.jpg", "20250716163458-700-360-433.jpg",
"20250717093052-782-277-690.jpg", "20250717103222-584-701-241.jpg",
"20250717103712-214-193-157.jpg", "20250717110901-449-871-865.jpg",
"20250717155048-253-564-315.jpg", "20250717172043-664-630-683.jpg",
"20250718080610-968-626-824.jpg", "20250718093242-193-502-326.jpg",
"20250718105942-802-175-536.jpg", "20250718154510-618-961-614.jpg",
"20250718171201-832-262-559.jpg", "20250721103440-887-127-453.jpg",
"20250721103440-949-954-201.jpg", "20250721103556-832-150-503.jpg",
"20250721111443-531-701-811.jpg", "20250721111443-912-880-634.jpg",
"20250721112249-956-647-309.jpg", "20250721130808-958-549-703.jpg",
"20250721133831-152-461-423.jpg", "20250721145455-511-434-514.jpg",
"20250721145455-875-554-320.jpg", "20250721145456-782-822-874.jpg",
"20250721155757-121-923-232.jpg", "20250721160111-763-493-901.jpg",
"20250721160359-227-567-869.jpg", "20250721160359-337-126-571.jpg",
"20250721172118-534-854-174.jpg", "20250722083248-564-741-719.jpg",
"20250722101426-428-671-780.jpg", "20250722101619-869-994-366.jpg",
"20250722113040-790-828-516.jpg", "20250722113435-988-461-994.jpg",
"20250722132834-142-640-698.jpg", "20250722151220-665-449-414.jpg",
"20250722151447-194-809-212.jpg", "20250722151659-492-562-414.jpg",
"20250722155515-295-661-246.jpg", "20250722164044-771-951-768.jpg",
"20250723090127-752-277-978.jpg", "20250723103830-197-217-803.jpg",
"20250723110935-882-617-879.jpg", "20250723113848-341-499-399.jpg",
"20250723113849-860-361-766.jpg", "20250723135403-994-597-524.jpg",
"20250723135644-957-724-435.jpg", "20250723140727-539-276-326.jpg",
"20250723151024-958-230-632.jpg", "20250723160751-628-951-424.jpg",
"20250723160846-651-369-917.jpg", "20250723162424-328-470-393.jpg",
"20250724083131-482-629-632.jpg", "20250724084439-705-558-529.jpg",
"20250724085219-940-177-263.jpg", "20250724112248-515-638-257.jpg",
"20250724140126-814-266-218.jpg", "20250724165128-348-167-761.jpg",
"20250724170756-316-660-852.jpg", "20250725084748-172-127-509.jpg",
"20250725090550-647-253-595.jpg", "20250725103854-127-797-609.jpg",
"20250725112611-877-225-953.jpg", "20250725150958-785-430-943.jpg",
"20250725160005-618-961-614.jpg", "20250725160006-645-814-611.jpg",
"20250728110536-229-869-218.jpg", "20250728110536-422-535-360.jpg",
"20250728110536-848-126-746.jpg", "20250728133331-290-838-249.jpg",
"20250728133631-893-551-661.jpg", "20250728133731-800-849-608.jpg",
"20250728133919-745-435-884.jpg", "20250728141244-723-384-786.jpg",
"20250728163719-158-329-264.jpg", "20250729091304-312-462-757.jpg",
"20250729101639-845-837-748.jpg", "20250729150847-216-665-480.jpg",
"20250729152047-863-915-863.jpg", "20250729152047-872-458-985.jpg",
"20250729152047-915-601-759.jpg", "20250730093300-400-680-981.jpg",
"20250730101956-808-881-885.jpg"
]
# 대상 디렉토리가 없으면 생성
if not os.path.exists(dest_dir):
os.makedirs(dest_dir)
print(f"'{dest_dir}' 디렉토리를 생성했습니다.")
copied_files = set()
# root_source_dir부터 시작해서 모든 하위 디렉토리를 재귀적으로 탐색
for dirpath, _, filenames in os.walk(root_source_dir):
for filename in filenames:
if filename in files_to_copy and filename not in copied_files:
source_file = os.path.join(dirpath, filename)
# 원본 디렉토리 구조를 유지하기 위한 경로 계산
relative_path = os.path.relpath(dirpath, root_source_dir)
new_dest_dir = os.path.join(dest_dir, relative_path)
# 새로운 목적지 디렉토리 생성
os.makedirs(new_dest_dir, exist_ok=True)
dest_file = os.path.join(new_dest_dir, filename)
shutil.copy2(source_file, dest_file)
print(f"'{filename}'을(를) '{new_dest_dir}'(으)로 복사했습니다.")
copied_files.add(filename)
# 복사되지 않은 파일 확인
missing_files = set(files_to_copy) - copied_files
if missing_files:
print("\n다음 파일들은 찾지 못했습니다:")
for filename in sorted(list(missing_files)):
print(f"- {filename}")
if __name__ == "__main__":
copy_target_files_with_structure()

17
dockerfile
View File

@@ -1,30 +1,17 @@
# Dockerfile
# 1. 베이스 이미지 선택
# 파이썬 3.10의 가벼운(slim) 버전을 기반으로 시작합니다.
FROM python:3.10-slim
# 2. 작업 디렉토리 설정
# 컨테이너 내에서 명령어를 실행할 기본 폴더를 설정합니다.
WORKDIR /workspace
# 3. 의존성 파일 복사 및 설치
# 먼저 의존성 목록 파일을 복사합니다. (이것만 바뀌었을 경우 빌드 속도 향상)
COPY requirements.txt .
COPY workspace/ .
# requirements.txt에 명시된 라이브러리들을 설치합니다.
# --no-cache-dir 옵션은 불필요한 캐시를 남기지 않아 이미지 크기를 줄여줍니다.
RUN pip install --no-cache-dir -r requirements.txt
COPY workspace/app.py .
# 4. 앱 소스 코드 복사
# 현재 폴더의 모든 파일을 컨테이너의 /app 폴더로 복사합니다.
COPY . .
# 5. 포트 노출
# Streamlit의 기본 포트인 8501을 외부에 노출하도록 설정합니다.
EXPOSE 8501
# 6. 컨테이너 실행 명령어 설정
# 컨테이너가 시작될 때 실행할 명령어를 정의합니다.
# --server.address=0.0.0.0 옵션은 컨테이너 외부에서의 접속을 허용하기 위해 필수입니다.
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

49
docs/app_readme.md Normal file
View File

@@ -0,0 +1,49 @@
# `app.py` - 정답셋 생성 및 비교 도구
이 문서는 `workspace/app.py` Streamlit 애플리케이션의 기능, 설정 방법, 그리고 사용법을 상세히 설명합니다.
## 1. 주요 기능
- **세션 기반 협업**: 파일 업로드 시 고유한 URL(`seed`)이 생성되어, 이 링크를 공유하면 여러 명의 작업자가 동일한 환경에서 작업할 수 있습니다.
- **3-Way 비교 뷰**: **원본 문서**, **참고용 모델 결과**, **정답셋 편집기**를 한 화면에 표시하여 직관적인 비교와 수정이 가능합니다.
- **모델 선택 및 편집**: `paddle_ocr`와 `upstage` 모델 중 하나를 정답셋의 기반으로 선택하고, `parsed` 키의 내용을 직접 수정하여 정답셋을 생성합니다.
- **작업 상태 추적**: 정답셋 생성이 완료된 파일은 목록에 `✅` 아이콘으로 표시되어 작업 중복을 방지합니다.
- **정답셋 재수정**: 이미 완료된 정답셋도 언제든지 다시 불러와 수정할 수 있습니다.
- **동적 레이아웃**: '참고용 영역 숨기기' 옵션을 통해 편집 공간을 넓게 확보할 수 있습니다.
- **결과물 다운로드**: 현재 세션에서 생성된 모든 정답셋을 하나의 `.zip` 파일로 편리하게 다운로드할 수 있습니다.
## 2. 설정 및 실행
### 실행
프로젝트의 루트 디렉토리에서 다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.
```dockerfile
docker compose up
```
실행 후 웹 브라우저에서 지정된 로컬 주소(예: `http://localhost:8501`)로 접속합니다.
## 3. 작업 흐름
1. **애플리케이션 실행**: 위의 명령어로 앱을 시작합니다.
2. **파일 업로드**:
- 사이드바의 '파일 업로드' 섹션에서 다음 세 종류의 파일을 모두 업로드합니다.
1. **원본 문서** (이미지/PDF)
2. **paddle_ocr JSON** 결과물
3. **upstage JSON** 결과물
- 파일 이름(확장자 제외)이 세 종류 모두 동일해야 목록에 정상적으로 표시됩니다.
3. **세션 생성**: '업로드 및 세션 생성' 버튼을 누르면, 고유한 `seed`가 포함된 URL이 생성되며 작업 화면으로 전환됩니다.
4. **세션 공유 (선택 사항)**: 생성된 URL을 다른 작업자와 공유하여 협업을 시작할 수 있습니다.
5. **파일 선택 및 검수**:
- 사이드바의 '파일 탐색' 드롭다운 메뉴에서 검수할 파일을 선택합니다.
- 화면 상단의 `◀ 이전` / `다음 ▶` 버튼으로 파일을 이동할 수도 있습니다.
6. **정답셋 생성**:
- 화면 우측 상단의 버튼을 통해 정답셋의 기반이 될 모델(`paddle_ocr` 또는 `upstage`)을 선택합니다.
- 오른쪽 '정답셋 편집' 영역에서 `parsed` 키의 텍스트를 정확하게 수정합니다.
- '✅ 정답셋으로 저장' 버튼을 눌러 작업을 완료합니다.
7. **작업 반복**: 모든 파일에 대해 5~6번 과정을 반복합니다. 완료된 파일은 목록에 `✅` 아이콘이 표시됩니다.
8. **정답셋 다운로드**:
- 모든 작업이 완료되면, 사이드바 하단의 '내보내기' 섹션에서 '정답셋 다운로드 (.zip)' 버튼을 눌러 결과물을 다운로드합니다.
- 다운로드된 파일은 현재 세션의 `groundtruth` 폴더에 저장된 모든 JSON 파일을 포함합니다.

62
docs/run_ocr1_readme.md Normal file
View File

@@ -0,0 +1,62 @@
# 문서 정보 추출 및 비교 도구
이 프로젝트는 문서(이미지, PDF)에서 정보를 추출하고, 그 결과를 원본 문서와 나란히 비교할 수 있는 도구를 제공합니다.
## 구성 요소
1. **추출 스크립트 (`workspace/process_directory.py`)**: 지정된 디렉터리의 모든 파일에 대해 정보 추출 API를 호출하고 결과를 JSON 파일로 저장합니다.
2. **비교 웹앱 (`workspace/app.py`)**: Streamlit 기반의 웹 애플리케이션으로, 원본 문서와 추출된 JSON 결과를 업로드하여 시각적으로 비교할 수 있습니다.
---
## 1. 추출 스크립트 사용법 (`process_directory.py`)
### 사전 준비
스크립트를 실행하기 위해서는 Python 3.6 이상이 필요합니다.
#### 1.1. 환경 변수 설정
API 서버의 정보는 민감 정보이므로, `workspace` 디렉터리 안에 `.env` 파일을 생성하여 관리합니다.
`.env` 파일 예시 (`workspace/.env`):
```env
BASE_URL="http://172.0.0.1:8888"
API_KEY="sk-xxxxxxxxxxxxxxxxxxxxxxxx"
```
**중요:** 이 `.env` 파일은 `.gitignore`에 의해 버전 관리에서 자동으로 제외됩니다.
#### 1.2. 의존성 설치
프로젝트 루트 디렉터리에서 다음 명령어를 실행하여 필요한 라이브러리를 설치합니다.
```bash
pip install -r requirements.txt
```
### 실행 방법
프로젝트 루트 디렉터리에서 다음 형식으로 스크립트를 실행합니다.
```bash
python workspace/process_directory.py [입력_디렉터리] [옵션]
```
#### 인자 설명
- `input_dir` (필수): 처리할 파일들이 들어있는 입력 디렉터리의 경로입니다.
- `-o, --output_dir` (선택): 결과 JSON 파일들을 저장할 출력 디렉터리입니다. (기본값: `results`)
- `--endpoint` (선택): 호출할 API 엔드포인트를 지정합니다. (`i18n` 또는 `d6c`, 기본값: `i18n`)
- `--model` (선택): 사용할 특정 LLM 모델의 이름을 지정합니다.
#### 실행 예시
```bash
# 'source_documents/data'에 있는 파일들을 'i18n' 엔드포인트로 처리
python workspace/process_directory.py source_documents/data
# 'd6c' 엔드포인트를 사용하여 처리하고 결과를 'my_results' 폴더에 저장
python workspace/process_directory.py source_documents/data2 --endpoint d6c -o my_results
```

70
docs/run_ocr2_readme.md Normal file
View File

@@ -0,0 +1,70 @@
# `run_ocr.py` 스크립트 사용법
이 문서는 `workspace/run_ocr.py` 스크립트를 사용하여 디렉토리 내의 여러 이미지 및 PDF 파일에 대해 OCR(광학 문자 인식)을 일괄 수행하는 방법을 설명합니다.
## 개요
`run_ocr.py`는 지정된 입력 폴더에 있는 모든 파일을 비동기 OCR API 서버로 전송하고, 처리 완료된 결과를 지정된 출력 폴더에 JSON 파일로 저장하는 커맨드 라인 도구입니다. **병렬 처리**와 **순차 처리** 두 가지 모드를 모두 지원합니다.
## 주요 기능
- **듀얼 모드 지원**:
- **병렬 처리 (기본)**: 여러 파일을 동시에 처리하여 작업 시간을 대폭 단축합니다.
- **순차 처리**: 초당 요청 수(RPS) 제한이 있는 API를 위해, 각 요청 사이에 지정된 시간만큼 대기하며 파일을 하나씩 처리합니다.
- **OCR 공급자 선택**: `paddle`과 `upstage` 두 가지 OCR 엔진을 선택할 수 있습니다.
- **자동 재시도**: 네트워크 오류나 서버의 일시적인 문제가 발생하면, "Exponential Backoff + Jitter" 방식으로 지능적으로 재시도하여 안정성을 높였습니다.
- **진행 상황 및 결과 요약**: 전체 진행 상황과 최종 성공/실패/건너뜀 건수를 명확하게 보여줍니다.
## 전제 조건
스크립트를 실행하기 위해서는 Python과 `requests` 라이브러리가 필요합니다. 프로젝트의 `requirements.txt` 파일에 필요한 라이브러리가 명시되어 있습니다.
```bash
pip install -r requirements.txt
```
## 사용법
스크립트는 프로젝트 루트 디렉토리에서 다음 형식의 명령어로 실행합니다.
```bash
python workspace/run_ocr.py [입력_디렉토리] [출력_디렉토리] [옵션...]
```
### 인자 설명
- `입력_디렉토리` (필수): OCR을 수행할 원본 이미지/PDF 파일들이 들어있는 폴더의 경로입니다.
- `출력_디렉토리` (필수): OCR 결과 JSON 파일이 저장될 폴더의 경로입니다. 폴더가 존재하지 않으면 자동으로 생성됩니다.
- `--provider` (선택): 사용할 OCR 공급자를 지정합니다. (기본값: `paddle`)
- `paddle`: Paddle OCR 사용
- `upstage`: Upstage OCR 사용
- `--workers` (선택): **병렬 처리 시** 동시에 처리할 파일의 개수(스레드 수)를 지정합니다. (기본값: 4)
- `--delay` 옵션이 활성화되면 이 값은 무시됩니다.
- `--delay` (선택): **순차 처리 시** 각 요청 사이의 대기 시간(초)을 설정합니다. (기본값: 0)
- 이 값을 0보다 크게 설정하면, 스크립트는 `--workers` 설정을 무시하고 순차 처리 모드로 강제 실행됩니다.
## 실행 예시
### 예시 1: 병렬 처리 (기본)
`source_documents` 폴더의 파일들을 **Paddle OCR**로, **4개씩 동시에** 처리하여 `result_jsons` 폴더에 저장합니다.
```bash
python workspace/run_ocr.py ./source_documents ./result_jsons
```
### 예시 2: 병렬 처리 (옵션 지정)
`source_documents` 폴더의 파일들을 **Upstage OCR**로, **10개씩 동시에** 처리하여 `result_jsons_upstage` 폴더에 저장합니다.
```bash
python workspace/run_ocr.py ./source_documents ./result_jsons_upstage --provider upstage --workers 10
```
### 예시 3: 순차 처리 (RPS 제한 대응)
`source_documents` 폴더의 파일들을 **Upstage OCR**로, **하나씩 순서대로** 분리하되 **각 요청 사이에 3초씩 대기**하여 `result_jsons_upstage` 폴더에 저장합니다.
```bash
python workspace/run_ocr.py ./source_documents ./result_jsons_upstage --provider upstage --delay 3
```

466
workspace/app.py
View File

@@ -1,205 +1,407 @@
# app.py (시드 기반 서버 사이드 세션 공유 기능)
import streamlit as st
import json
from pathlib import Path
# app.py (UI 최종 개선)
import base64
import uuid
import io
import json
import shutil
import time
import uuid
import zipfile
from pathlib import Path
import streamlit as st
# --- 상수 ---
# 스크립트 파일의 위치를 기준으로 경로 설정
SESSION_BASE_PATH = Path(__file__).parent / "shared_sessions"
EDIT_KEY = "parsed" # 수정할 키를 'parsed'로 고정
# --- 상태 관리 함수 (세션 기반) ---
def get_session_path(seed):
return SESSION_BASE_PATH / seed
def get_session_completed_log_path(seed):
return get_session_path(seed) / "completed_files.json"
def load_completed_files(seed):
log_path = get_session_completed_log_path(seed)
if not log_path.exists():
return set()
try:
with open(log_path, "r", encoding="utf-8") as f:
return set(json.load(f))
except (json.JSONDecodeError, FileNotFoundError):
return set()
def save_completed_file(seed, basename):
completed_set = load_completed_files(seed)
completed_set.add(basename)
log_path = get_session_completed_log_path(seed)
with open(log_path, "w", encoding="utf-8") as f:
json.dump(list(completed_set), f, indent=2)
# --- 헬퍼 함수 ---
def get_session_path(seed):
"""시드에 해당하는 세션 디렉토리 경로를 반환합니다."""
return SESSION_BASE_PATH / seed
def save_files_to_session(seed, doc_files, json_files):
"""업로드된 파일들을 서버의 세션 디렉토리에 저장합니다."""
def save_files_to_session(seed, doc_files, json_paddle_files, json_upstage_files):
session_path = get_session_path(seed)
doc_path = session_path / "docs"
json_path = session_path / "jsons"
json_paddle_path = session_path / "jsons_paddle_ocr"
json_upstage_path = session_path / "jsons_upstage"
# 기존 디렉토리가 있으면 삭제하고 새로 생성
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 p in [doc_path, json_paddle_path, json_upstage_path]:
p.mkdir(parents=True, exist_ok=True)
for f in doc_files:
(doc_path / f.name).write_bytes(f.getbuffer())
for f in json_paddle_files:
(json_paddle_path / f.name).write_bytes(f.getbuffer())
for f in json_upstage_files:
(json_upstage_path / f.name).write_bytes(f.getbuffer())
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"
json_paddle_path = session_path / "jsons_paddle_ocr"
json_upstage_path = session_path / "jsons_upstage"
if not all([p.is_dir() for p in [doc_path, json_paddle_path, json_upstage_path]]):
return None, None, None
return (
sorted(doc_path.iterdir()),
sorted(json_paddle_path.iterdir()),
sorted(json_upstage_path.iterdir()),
)
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 = {}
def match_files_3_way(doc_files, json_paddle_files, json_upstage_files):
matched = {}
docs_map = {f.stem: f for f in doc_files}
jsons_map = {f.stem: f for f in json_files}
jsons_paddle_map = {f.stem: f for f in json_paddle_files}
jsons_upstage_map = {f.stem: f for f in json_upstage_files}
for stem, doc_file in docs_map.items():
if stem in jsons_map:
matched_pairs[stem] = {
if stem in jsons_paddle_map and stem in jsons_upstage_map:
matched[stem] = {
"doc_file": doc_file,
"json_file": jsons_map[stem]
"paddle_ocr_file": jsons_paddle_map[stem],
"upstage_file": jsons_upstage_map[stem],
}
return matched_pairs
return matched
def display_pdf(file_path_or_obj):
"""파일 경로 또는 업로드된 파일 객체를 받아 PDF를 표시합니다."""
def display_pdf(file_path):
bytes_data = file_path.read_bytes()
base64_pdf = base64.b64encode(bytes_data).decode("utf-8")
st.markdown(
f'<iframe src="data:application/pdf;base64,{base64_pdf}" width="100%" height="800" type="application/pdf"></iframe>',
unsafe_allow_html=True,
)
def display_readonly_json(file_path, title):
st.subheader(title)
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'<iframe src="data:application/pdf;base64,{base64_pdf}" width="100%" height="800" type="application/pdf"></iframe>'
st.markdown(pdf_display, unsafe_allow_html=True)
with open(file_path, "r", encoding="utf-8") as f:
data = json.load(f)
is_list = isinstance(data, list)
result_to_display = data[0] if is_list and data else data
st.json(result_to_display)
except Exception as e:
st.error(f"PDF 파일을 표시하는 중 오류가 발생했습니다: {e}")
st.error(f"JSON 표시 중 오류: {e}")
def create_gt_editor(seed, basename, json_file_path, model_name, is_re_edit=False):
editor_title = (
"정답셋 재수정" if is_re_edit else f"정답셋 편집 (기반: {model_name})"
)
st.subheader(editor_title)
try:
with open(json_file_path, "r", encoding="utf-8") as f:
data = json.load(f)
is_list = isinstance(data, list)
result_to_display = data[0] if is_list and data else data
if not isinstance(result_to_display, dict):
st.warning("수정할 수 있는 JSON 객체(dict)가 아닙니다.")
return
if EDIT_KEY not in result_to_display:
st.error(f"편집 대상 키인 '{EDIT_KEY}'를 JSON 파일에서 찾을 수 없습니다.")
return
current_value = result_to_display.get(EDIT_KEY, "")
st.markdown(f"**`{EDIT_KEY}`** 키의 내용을 수정합니다.")
new_value_str = st.text_area(
"값 수정:",
value=str(current_value),
height=800,
key=f"gt_value_input_{model_name}",
)
if st.button("✅ 정답셋으로 저장", key=f"gt_save_button_{model_name}"):
modified_data = result_to_display.copy()
modified_data[EDIT_KEY] = new_value_str
final_data_to_save = [modified_data] if is_list else modified_data
gt_path = get_session_path(seed) / "groundtruth"
gt_path.mkdir(parents=True, exist_ok=True)
save_path = gt_path / f"{basename}.json"
with open(save_path, "w", encoding="utf-8") as f:
json.dump(final_data_to_save, f, indent=2, ensure_ascii=False)
if not is_re_edit:
save_completed_file(seed, basename)
if "re_edit_gt" in st.session_state:
del st.session_state["re_edit_gt"]
st.success(f"'{save_path.name}' 파일이 정답셋으로 저장되었습니다!")
st.info("상태가 업데이트되었습니다. 페이지를 새로고침합니다.")
time.sleep(1)
st.rerun()
except Exception as e:
st.error(f"JSON 편집기 생성 중 오류: {e}")
# --- 콜백 함수 ---
def handle_nav_button(direction, total_files):
if "re_edit_gt" in st.session_state:
del st.session_state["re_edit_gt"]
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
if "re_edit_gt" in st.session_state:
del st.session_state["re_edit_gt"]
st.session_state.current_index = (
int(st.session_state.selectbox_key.split(". ", 1)[0]) - 1
)
# --- 메인 UI 로직 ---
def main():
st.set_page_config(layout="wide", page_title="결과 비교 도구")
st.title("📑 결과 비교 및 공유 도구")
st.markdown("---")
# 세션 상태 초기화
if 'current_index' not in st.session_state:
st.set_page_config(layout="wide", page_title="정답셋 생성 도구")
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")
url_seed = st.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}'에 해당하는 공유 세션을 찾을 수 없습니다. 시드가 정확한지 확인하거나, 파일을 새로 업로드하세요.")
completed_files = load_completed_files(url_seed)
files = load_files_from_session(url_seed)
if files[0] is not None:
st.success(f"'{url_seed}' 시드에서 파일을 불러왔습니다.")
matched_files = match_files_3_way(*files)
else:
st.success(f"'{url_seed}' 시드에서 공유된 파일을 불러왔습니다.")
matched_files = match_disk_files(doc_files, json_files)
# 시드가 없거나, 시드로 로드 실패 시 파일 업로더 표시
st.error(f"'{url_seed}'에 해당하는 세션을 찾을 수 없습니다.")
else:
completed_files = set()
# --- 사이드바 ---
st.sidebar.info("화면을 넓게 보려면 오른쪽 위 화살표를 누르세요 <<")
st.sidebar.markdown("---")
st.sidebar.header("파일 업로드")
if not matched_files:
st.sidebar.header("파일 업로드")
uploaded_docs = st.sidebar.file_uploader(
"1. 원본 문서 파일(들)을 업로드하세요.",
accept_multiple_files=True,
type=['png', 'jpg', 'jpeg', 'pdf']
docs = st.sidebar.file_uploader(
"1. 원본 문서", accept_multiple_files=True, type=["png", "jpg", "pdf"]
)
uploaded_jsons = st.sidebar.file_uploader(
"2. 결과 JSON 파일(들)을 업로드하세요.",
accept_multiple_files=True,
type=['json']
jsons_paddle = st.sidebar.file_uploader(
"2. paddle_ocr JSON", accept_multiple_files=True, type=["json"]
)
jsons_upstage = st.sidebar.file_uploader(
"3. upstage JSON", accept_multiple_files=True, type=["json"]
)
if all([docs, jsons_paddle, jsons_upstage]) and st.sidebar.button(
"업로드 및 세션 생성"
):
new_seed = str(uuid.uuid4())[:8]
save_files_to_session(new_seed, docs, jsons_paddle, jsons_upstage)
st.query_params["seed"] = new_seed
st.rerun()
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:
if url_seed and matched_files is not None:
st.sidebar.header("세션 공유")
# 현재 페이지의 전체 URL을 가져오는 것은 Streamlit에서 직접 지원하지 않으므로,
# 사용자에게 주소창의 URL을 복사하라고 안내합니다.
st.sidebar.success("세션이 활성화되었습니다!")
st.sidebar.info("다른 사람과 공유하려면 현재 브라우저 주소창의 URL을 복사하여 전달하세요.")
st.sidebar.info("URL을 복사하여 다른 사람과 세션을 공유하세요.")
st.sidebar.text_input("공유 시드", url_seed, disabled=True)
# --- 결과 표시 로직 (matched_files가 있을 때만 실행) ---
if not matched_files:
st.info("사이드바에서 파일을 업로드하고 '업로드 및 세션 생성' 버튼을 누르거나, 공유받은 URL로 접속하세요.")
st.info("모든 종류의 파일을 업로드하고 세션을 생성하세요.")
if matched_files is not None and not matched_files:
st.warning(
"파일 이름(확장자 제외)이 동일한 '문서-paddle_ocr-upstage' 세트를 찾을 수 없습니다."
)
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} {'✅' if name in completed_files else ''}"
for i, name in enumerate(sorted_basenames)
]
display_options = [f"{i + 1}. {name}" for i, name in enumerate(sorted_basenames)]
st.selectbox(
"파일을 직접 선택하세요:",
options=display_options,
st.sidebar.selectbox(
"파일 선택:",
display_options,
index=st.session_state.current_index,
key='selectbox_key',
on_change=handle_selectbox_change
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"<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)
st.sidebar.header("보기 옵션")
hide_reference = st.sidebar.checkbox("참고용 영역 숨기기", key="hide_reference")
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"]
st.sidebar.header("내보내기")
gt_dir = get_session_path(url_seed) / "groundtruth"
if gt_dir.exists() and any(gt_dir.iterdir()):
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "a", zipfile.ZIP_DEFLATED, False) as zip_file:
for file_path in gt_dir.glob("*.json"):
zip_file.writestr(file_path.name, file_path.read_bytes())
res_col1, res_col2 = st.columns(2)
with res_col1:
st.sidebar.download_button(
label="정답셋 다운로드 (.zip)",
data=zip_buffer.getvalue(),
file_name=f"groundtruth_{url_seed}.zip",
mime="application/zip",
)
else:
st.sidebar.write("다운로드할 정답셋이 없습니다.")
# --- 메인 화면 ---
current_basename = (
display_options[st.session_state.current_index]
.split(" ", 1)[1]
.replace(" ✅", "")
.strip()
)
pair = matched_files[current_basename]
doc_file, paddle_file, upstage_file = (
pair["doc_file"],
pair["paddle_ocr_file"],
pair["upstage_file"],
)
top_container = st.container()
is_completed = current_basename in completed_files
is_re_editing = st.session_state.get("re_edit_gt") == current_basename
with top_container:
nav_cols = st.columns([1, 5, 1])
nav_cols[0].button(
"◀ 이전",
on_click=handle_nav_button,
args=("prev", len(sorted_basenames)),
use_container_width=True,
)
nav_cols[1].markdown(
f"<h4 style='text-align: center; white-space: nowrap; overflow: hidden; text-overflow: ellipsis;'>{current_basename} ({st.session_state.current_index + 1}/{len(sorted_basenames)})</h4>",
unsafe_allow_html=True,
)
nav_cols[2].button(
"다음 ▶",
on_click=handle_nav_button,
args=("next", len(sorted_basenames)),
use_container_width=True,
)
st.markdown("---")
if is_completed and not is_re_editing:
msg_col, btn_col = st.columns([3, 1])
msg_col.success("✅ 이 파일은 정답셋 생성이 완료되었습니다.")
if btn_col.button("🔄 정답셋 다시 수정하기"):
st.session_state.re_edit_gt = current_basename
st.rerun()
elif not is_completed:
_, radio_col = st.columns([3, 1])
source_model_name = radio_col.radio(
"정답셋 편집 기반 모델 선택:",
("paddle_ocr", "upstage"),
horizontal=True,
label_visibility="collapsed",
)
if hide_reference:
col1, col3 = st.columns([1, 1])
else:
col1, col2, col3 = st.columns([2, 1, 1])
with 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)
st.image(str(doc_file), use_container_width=True)
if is_re_editing:
gt_file_path = (
get_session_path(url_seed) / "groundtruth" / f"{current_basename}.json"
)
if gt_file_path.exists():
with col3:
create_gt_editor(
url_seed,
current_basename,
gt_file_path,
"Ground Truth",
is_re_edit=True,
)
else:
st.error("저장된 정답셋 파일을 찾을 수 없습니다. 새로 생성해주세요.")
if "re_edit_gt" in st.session_state:
del st.session_state.re_edit_gt
elif is_completed:
with col3:
gt_file_path = (
get_session_path(url_seed) / "groundtruth" / f"{current_basename}.json"
)
if gt_file_path.exists():
display_readonly_json(gt_file_path, "저장된 정답셋")
else:
st.warning("저장된 정답셋 파일을 찾을 수 없습니다.")
else:
if source_model_name == "paddle_ocr":
with col3:
create_gt_editor(url_seed, current_basename, paddle_file, "paddle_ocr")
else:
with col3:
create_gt_editor(url_seed, current_basename, upstage_file, "upstage")
if not hide_reference:
with col2:
if is_re_editing:
st.empty()
elif is_completed:
ref_model_name = st.radio(
"참고용 모델 선택:",
("paddle_ocr", "upstage"),
horizontal=True,
key=f"ref_select_{current_basename}",
)
if ref_model_name == "paddle_ocr":
display_readonly_json(paddle_file, "참고용: paddle_ocr")
else:
display_readonly_json(upstage_file, "참고용: upstage")
else:
if source_model_name == "paddle_ocr":
display_readonly_json(paddle_file, "참고용: paddle_ocr")
else:
display_readonly_json(upstage_file, "참고용: upstage")
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()

236
workspace/cmp_app.py Normal file
View File

@@ -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'<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():
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"<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,
)
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()

195
workspace/process_directory.py
View File

@@ -1,195 +0,0 @@
import requests
import time
import json
import os
import argparse
import sys
from urllib.parse import urljoin
import logging
from dotenv import load_dotenv
# --- 로거 설정 ---
# 전역 로거 객체 생성
logger = logging.getLogger(__name__)
def setup_logger():
"""
로거를 설정하여 콘솔과 고유한 타임스탬프를 가진 파일에 모두 출력하도록 합니다.
로그 파일은 'workspace' 디렉터리 내에 저장됩니다.
"""
# 기존에 추가된 핸들러가 있다면 제거하여 중복 로깅 방지
for handler in logger.handlers[:]:
logger.removeHandler(handler)
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
# 콘솔 핸들러
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
# 고유한 파일명을 위한 타임스탬프 생성
timestamp = time.strftime("%Y%m%d_%H%M%S")
log_filename = f"{timestamp}_script_run.log"
# 로그 파일을 workspace 디렉터리 안에 생성
workspace_dir = os.path.dirname(__file__)
log_filepath = os.path.join(workspace_dir, log_filename)
# 파일 핸들러
file_handler = logging.FileHandler(log_filepath, encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# 생성된 로그 파일 경로를 로깅
logger.info(f"로그 파일이 '{log_filepath}'에 생성되었습니다.")
# --- API 요청 함수 ---
def start_extraction(post_url, file_path, filename, headers, model_name=None):
"""POST /extract/inner: 문서 추출 시작"""
try:
with open(file_path, 'rb') as input_f:
files_to_upload = {'input_file': (filename, input_f)}
data_payload = {}
if model_name:
data_payload['model'] = model_name
response = requests.post(post_url, files=files_to_upload, data=data_payload, headers=headers)
response.raise_for_status()
return response.json()
except Exception:
logger.exception(f"[{filename}] POST 요청 중 오류 발생")
return None
def check_progress(base_url, progress_path, filename, headers):
"""GET /extract/progress/{request_id}: 진행 상태 확인 (로깅 적용)"""
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 = ""
while True:
try:
response = requests.get(get_url, headers=headers, timeout=30)
if response.status_code == 404:
if retries_left > 0:
logger.warning(f"[{filename}] 작업을 찾을 수 없어(404) {RETRY_DELAY_ON_404}초 후 재시도합니다... ({retries_left}회 남음)")
retries_left -= 1
time.sleep(RETRY_DELAY_ON_404)
continue
else:
logger.error(f"[{filename}] 재시도 횟수 초과 후에도 작업을 찾을 수 없습니다 (404).")
return None
response.raise_for_status()
data = response.json()
if "final_result" in data and data.get("final_result") is not None:
logger.info(f"[{filename}] 처리 완료.")
return data["final_result"]
if "progress_logs" in data and data["progress_logs"]:
status_message = data["progress_logs"][-1].get("status", "상태 확인 중...")
if status_message != last_status:
last_status = status_message
logger.info(f"[{filename}] 진행 상태: {last_status}")
time.sleep(2)
except requests.exceptions.ReadTimeout:
logger.warning(f"[{filename}] 상태 확인 타임아웃. 재시도...")
time.sleep(2)
except Exception:
logger.exception(f"[{filename}] 상태 확인 중 예측하지 못한 오류 발생")
return None
# --- 메인 실행 로직 ---
def main():
# .env 파일에서 환경 변수 로드 (workspace 디렉터리 기준)
dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
load_dotenv(dotenv_path=dotenv_path)
# 로거를 가장 먼저 설정
setup_logger()
# 환경 변수에서 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)")
parser.add_argument("--model", dest="model_name", help="사용할 LLM 모델 이름")
args = parser.parse_args()
if not os.path.isdir(args.input_dir):
logger.error(f"입력 디렉터리를 찾을 수 없습니다 - {args.input_dir}")
return
os.makedirs(args.output_dir, exist_ok=True)
headers = {'X-API-KEY': API_KEY}
post_url = f"{BASE_URL}/extract/inner/{args.endpoint}"
logger.info("="*20 + " 스크립트 시작 " + "="*20)
logger.info(f"API 서버: {BASE_URL}")
logger.info(f"요청 API: {post_url}")
logger.info(f"입력 디렉터리: {args.input_dir}")
logger.info(f"출력 디렉터리: {args.output_dir}")
# 처리할 파일 목록 준비
files_to_process = [f for f in sorted(os.listdir(args.input_dir)) if os.path.isfile(os.path.join(args.input_dir, f))]
total_files = len(files_to_process)
logger.info(f"총 {total_files}개의 파일을 처리합니다.")
for i, filename in enumerate(files_to_process):
file_path = os.path.join(args.input_dir, filename)
logger.info(f"--- ({i+1}/{total_files}) 처리 시작: {filename} ---")
initial_response = start_extraction(post_url, file_path, filename, headers, args.model_name)
if not initial_response:
logger.error(f"[{filename}] 파일 처리 실패 (추출 시작 단계)")
continue
request_id = initial_response.get("request_id")
status_check_url = initial_response.get("status_check_url")
if not request_id or not status_check_url:
logger.error(f"[{filename}] 초기 응답이 잘못되었습니다: {initial_response}")
continue
logger.info(f"[{filename}] 작업 요청 성공. Request ID: {request_id}")
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")
try:
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(final_result, f, indent=2, ensure_ascii=False)
logger.info(f"[{filename}] 결과 저장 완료: {output_path}")
except IOError:
logger.exception(f"[{filename}] 파일 저장 중 오류 발생")
else:
logger.error(f"[{filename}] 파일 처리 실패 (결과 확인 단계)")
logger.info("="*20 + " 모든 작업 완료 " + "="*20)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
logging.getLogger(__name__).warning("사용자에 의해 작업이 중단되었습니다.")

335
workspace/run_ocr1.py Normal file
View File

@@ -0,0 +1,335 @@
import argparse
import json
import logging
import logging.handlers
import os
import random
import time
from multiprocessing import Pool, Queue, current_process
from urllib.parse import urljoin
import requests
from dotenv import load_dotenv
# --- 전역 변수 및 로깅 시스템 ---
worker_log_queue = None
completed_tasks_count = 0
total_tasks = 0
def setup_main_logger(log_queue):
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
formatter = logging.Formatter(
"%(asctime)s - %(processName)s - %(levelname)s - %(message)s"
)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
timestamp = time.strftime("%Y%m%d_%H%M%S")
log_filename = f"{timestamp}_robust_run.log"
workspace_dir = os.path.dirname(os.path.abspath(__file__)) or "."
log_filepath = os.path.join(workspace_dir, log_filename)
file_handler = logging.FileHandler(log_filepath, encoding="utf-8")
file_handler.setFormatter(formatter)
queue_listener = logging.handlers.QueueListener(
log_queue, console_handler, file_handler
)
queue_listener.start()
bootstrap_logger = logging.getLogger("bootstrap")
bootstrap_logger.setLevel(logging.INFO)
bootstrap_logger.addHandler(console_handler)
bootstrap_logger.addHandler(file_handler)
bootstrap_logger.propagate = False
bootstrap_logger.info(f"메인 로거 설정 완료. 로그 파일: '{log_filepath}'")
bootstrap_logger.info("=" * 20 + " 스크립트 시작 " + "=" * 20)
return queue_listener, bootstrap_logger
def init_worker(log_queue):
global worker_log_queue
worker_log_queue = log_queue
def get_worker_logger():
global worker_log_queue
worker_logger = logging.getLogger(current_process().name)
if not worker_logger.handlers:
worker_logger.setLevel(logging.INFO)
queue_handler = logging.handlers.QueueHandler(worker_log_queue)
worker_logger.addHandler(queue_handler)
return worker_logger
# --- API 요청 함수 (변경 없음) ---
def start_extraction(post_url, file_path, filename, headers, model_name=None):
worker_logger = get_worker_logger()
worker_logger.info(f"[{filename}] 파일 업로드 및 추출 요청 시작...")
# ... (내부 로직은 이전과 동일)
start_time = time.time()
try:
with open(file_path, "rb") as input_f:
files_to_upload = {"input_file": (filename, input_f)}
data_payload = {}
if model_name:
data_payload["model"] = model_name
response = requests.post(
post_url,
files=files_to_upload,
data=data_payload,
headers=headers,
timeout=300,
)
end_time = time.time()
worker_logger.info(
f"[{filename}] 추출 요청 완료. 소요 시간: {end_time - start_time:.2f}초, 상태 코드: {response.status_code}"
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
end_time = time.time()
worker_logger.exception(
f"[{filename}] 추출 요청 중 치명적 오류 발생. 소요 시간: {end_time - start_time:.2f}초. 오류: {e}"
)
return None
def check_progress(base_url, progress_path, filename, headers):
worker_logger = get_worker_logger()
get_url = urljoin(base_url + "/", progress_path.lstrip("/"))
# ... (내부 로직은 이전과 동일)
base_delay = 5
max_delay = 60
attempt = 0
last_status = ""
while True:
delay = min(max_delay, base_delay * (2**attempt))
jitter = delay * 0.2
sleep_duration = delay + random.uniform(-jitter, jitter)
worker_logger.info(
f"[{filename}] {sleep_duration:.2f}초 후 상태 확인 (시도 {attempt + 1})..."
)
time.sleep(sleep_duration)
start_time = time.time()
try:
response = requests.get(get_url, headers=headers, timeout=60)
end_time = time.time()
if response.status_code != 200:
worker_logger.warning(
f"[{filename}] 상태 확인 응답 코드: {response.status_code}. 소요 시간: {end_time - start_time:.2f}초"
)
if response.status_code == 404:
if attempt < 5:
worker_logger.warning(
f"[{filename}] 작업을 찾을 수 없음(404). 재시도합니다."
)
attempt += 1
continue
else:
worker_logger.error(
f"[{filename}] 재시도 횟수 초과 후에도 작업을 찾을 수 없음(404). 처리 중단."
)
return None
response.raise_for_status()
data = response.json()
if "final_result" in data and data.get("final_result") is not None:
worker_logger.info(f"[{filename}] 처리 완료.")
return data["final_result"]
if "progress_logs" in data and data["progress_logs"]:
status_message = data["progress_logs"][-1].get(
"status", "상태 확인 중..."
)
if status_message != last_status:
last_status = status_message
worker_logger.info(f"[{filename}] 진행 상태: {last_status}")
attempt += 1
except requests.exceptions.ReadTimeout:
end_time = time.time()
worker_logger.warning(
f"[{filename}] 상태 확인 타임아웃. 소요 시간: {end_time - start_time:.2f}초. 재시도..."
)
attempt += 1
except requests.exceptions.RequestException as e:
end_time = time.time()
worker_logger.exception(
f"[{filename}] 상태 확인 중 치명적 오류 발생. 소요 시간: {end_time - start_time:.2f}초. 오류: {e}"
)
return None
# --- 워커 함수 및 콜백 함수 ---
# <--- 핵심 변경 1: 워커 함수가 순번(index)과 전체 개수(total)를 인자로 받음 ---
def process_file_worker(index, total, file_path, config):
"""단일 파일을 처리하는 워커 함수. 이제 순번과 전체 개수를 인자로 받습니다."""
worker_logger = get_worker_logger()
filename = os.path.basename(file_path)
# <--- 핵심 변경 2: 로그 출력 시 (순번/전체) 형식 사용 ---
worker_logger.info(f"--- ({index}/{total}) 처리 시작: {filename} ---")
initial_response = start_extraction(
config["post_url"], file_path, filename, config["headers"], config["model_name"]
)
if not initial_response:
worker_logger.error(f"[{filename}] 파일 처리 실패 (추출 시작 단계)")
return index, filename, False # <--- 핵심 변경 3: 결과 반환 시 순번도 함께 전달
request_id = initial_response.get("request_id")
status_check_url = initial_response.get("status_check_url")
if not request_id or not status_check_url:
worker_logger.error(
f"[{filename}] 초기 응답이 잘못되었습니다: {initial_response}"
)
return index, filename, False
worker_logger.info(f"[{filename}] 작업 요청 성공. Request ID: {request_id}")
final_result = check_progress(
config["base_url"], status_check_url, filename, config["headers"]
)
if final_result:
output_path = os.path.join(
config["output_dir"], f"{os.path.splitext(filename)[0]}.json"
)
try:
with open(output_path, "w", encoding="utf-8") as f:
json.dump(final_result, f, indent=2, ensure_ascii=False)
worker_logger.info(f"[{filename}] 결과 저장 완료: {output_path}")
return index, filename, True
except IOError:
worker_logger.exception(f"[{filename}] 파일 저장 중 오류 발생")
return index, filename, False
else:
worker_logger.error(f"[{filename}] 파일 처리 실패 (결과 확인 단계)")
return index, filename, False
def update_progress(result):
"""진행 상황을 업데이트하고 출력하는 콜백 함수."""
global completed_tasks_count, total_tasks
completed_tasks_count += 1
# <--- 핵심 변경 4: 콜백 함수가 순번을 함께 받아서 로그에 표시 ---
index, filename, success = result
status = "성공" if success else "실패"
logging.getLogger("bootstrap").info(
f"--- ({completed_tasks_count}/{total_tasks}) 완료: [{index}번째 파일] {filename} (상태: {status}) ---"
)
# --- 메인 실행 로직 ---
def main():
global total_tasks
# ... (인자 파싱 및 설정 부분은 이전과 동일) ...
dotenv_path = os.path.join(os.path.dirname(__file__), ".env")
load_dotenv(dotenv_path=dotenv_path)
BASE_URL = os.getenv("BASE_URL")
API_KEY = os.getenv("API_KEY")
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 엔드포인트 선택",
)
parser.add_argument("--model", dest="model_name", help="사용할 LLM 모델 이름")
parser.add_argument(
"-w", "--workers", type=int, default=4, help="동시에 실행할 워커 프로세스 수"
)
args = parser.parse_args()
log_queue = Queue()
queue_listener, logger = setup_main_logger(log_queue)
if not BASE_URL or not API_KEY:
logger.error("환경 변수(BASE_URL, API_KEY)가 .env 파일에 설정되지 않았습니다.")
queue_listener.stop()
return
if not os.path.isdir(args.input_dir):
logger.error(f"입력 디렉터리를 찾을 수 없습니다 - {args.input_dir}")
queue_listener.stop()
return
os.makedirs(args.output_dir, exist_ok=True)
headers = {"X-API-KEY": API_KEY}
post_url = f"{BASE_URL}/extract/inner/{args.endpoint}"
logger.info(f"API 서버: {BASE_URL}")
logger.info(f"요청 API: {post_url}")
logger.info(f"입력 디렉터리: {args.input_dir}")
logger.info(f"출력 디렉터리: {args.output_dir}")
logger.info(f"동시 작업 수 (워커): {args.workers}")
files_to_process = [
os.path.join(args.input_dir, f)
for f in sorted(os.listdir(args.input_dir))
if os.path.isfile(os.path.join(args.input_dir, f))
]
total_tasks = len(files_to_process)
logger.info(f"총 {total_tasks}개의 파일을 처리합니다.")
config = {
"base_url": BASE_URL,
"post_url": post_url,
"headers": headers,
"model_name": args.model_name,
"output_dir": args.output_dir,
}
results = []
pool = None
try:
pool = Pool(
processes=args.workers, initializer=init_worker, initargs=(log_queue,)
)
async_results = []
# <--- 핵심 변경 5: for 루프에서 enumerate를 사용하여 순번(i)을 함께 전달 ---
for i, file_path in enumerate(files_to_process):
# 작업 인자에 (순번, 전체 개수, 파일 경로, 설정)을 담음
task_args = (i + 1, total_tasks, file_path, config)
res = pool.apply_async(
process_file_worker, args=task_args, callback=update_progress
)
async_results.append(res)
pool.close()
pool.join()
results = [res.get() for res in async_results]
except KeyboardInterrupt:
logger.warning("사용자에 의해 작업이 중단되었습니다.")
if pool:
pool.terminate()
pool.join()
except Exception:
logging.getLogger(__name__).exception(
"메인 프로세스에서 예기치 않은 오류가 발생했습니다."
)
if pool:
pool.terminate()
pool.join()
# <--- 핵심 변경 6: 최종 결과 요약 시 순번을 제외하고 파일명만 사용 ---
successful_files = [filename for index, filename, success in results if success]
failed_files = [filename for index, filename, success in results if not success]
logger.info("=" * 20 + " 모든 작업 완료 " + "=" * 20)
logger.info(
f"총 {total_tasks}개 파일 중 {len(successful_files)}개 성공, {len(failed_files)}개 실패."
)
if failed_files:
logger.warning(f"실패한 파일 목록: {failed_files}")
queue_listener.stop()
if __name__ == "__main__":
main()

242
workspace/run_ocr2.py Normal file
View File

@@ -0,0 +1,242 @@
import argparse
import json
import random
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
import requests
# --- 설정 ---
BASE_URL = "http://172.16.10.176:8892"
POLL_INTERVAL = 2 # 상태 확인 간격 (초)
SUPPORTED_EXTENSIONS = [".png", ".jpg", ".jpeg", ".webp", ".bmp", ".pdf"]
MAX_RETRIES = 5 # 최대 재시도 횟수
BACKOFF_FACTOR = 1 # 백오프 시간 기본값 (초)
def request_with_retry(method, url, **kwargs):
"""Exponential Backoff + Jitter를 적용한 재시도 로직"""
for attempt in range(MAX_RETRIES):
try:
response = requests.request(method, url, **kwargs)
if 500 <= response.status_code < 600:
print(
f"\n⚠️ 서버 오류 ({response.status_code}). {attempt + 1}번째 재시도 중..."
)
raise requests.exceptions.HTTPError(
f"Server error: {response.status_code}"
)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt < MAX_RETRIES - 1:
backoff_time = BACKOFF_FACTOR * (2**attempt)
jitter = random.uniform(0, 1)
sleep_time = backoff_time + jitter
print(f"\n❌ 네트워크 오류: {e}. {sleep_time:.2f}초 후 재시도합니다.")
time.sleep(sleep_time)
else:
print(
f"\n❌ 최대 재시도 횟수({MAX_RETRIES})를 초과했습니다. 최종 오류: {e}"
)
raise
def run_ocr_on_file(file_path: Path, output_dir: Path, provider: str):
"""단일 파일에 대해 OCR을 수행하고 결과를 저장합니다."""
log_prefix = f"[{file_path.name}]"
print(f"{log_prefix} 📄 처리 시작 (Provider: {provider})")
output_file_path = output_dir / f"{file_path.stem}.json"
if output_file_path.exists():
print(f"{log_prefix} 이미 결과가 존재합니다. 건너뜁니다.")
return f"SKIPPED: {file_path.name}"
try:
with open(file_path, "rb") as f:
files = {"file": (file_path.name, f, "application/octet-stream")}
endpoint = f"/ocr/{provider}"
response = request_with_retry(
"POST", f"{BASE_URL}{endpoint}", files=files, timeout=30
)
initial_data = response.json()
request_id = initial_data.get("request_id")
status_check_url = initial_data.get("status_check_url")
if not request_id or not status_check_url:
print(
f"{log_prefix} ❌ 오류: 서버 응답에 'request_id' 또는 'status_check_url'이 없습니다."
)
return f"FAILED: {file_path.name} (Invalid initial response)"
print(f"{log_prefix} - 작업 접수 완료. Request ID: {request_id}")
progress_url = f"{BASE_URL}{status_check_url}"
print(f"{log_prefix} - 결과 확인 중...", end="", flush=True)
while True:
progress_response = request_with_retry("GET", progress_url, timeout=30)
progress_data = progress_response.json()
status = progress_data.get("status")
if status == "SUCCESS":
print(" 완료!")
result_data = progress_data.get("final_result")
with open(output_file_path, "w", encoding="utf-8") as f:
json.dump(result_data, f, indent=2, ensure_ascii=False)
print(f"{log_prefix} - ✅ 결과 저장 완료: {output_file_path.name}")
return f"SUCCESS: {file_path.name}"
elif status == "FAILURE":
print(" 실패!")
print(
f"{log_prefix} - ❌ 오류: OCR 작업 실패. 상세: {progress_data.get('message')}"
)
return f"FAILED: {file_path.name} (OCR Failure)"
else: # PENDING
print(".", end="", flush=True)
time.sleep(POLL_INTERVAL)
except requests.exceptions.RequestException as e:
print(f"\n{log_prefix} ❌ 최종 오류: {e}")
return f"FAILED: {file_path.name} ({e})"
except json.JSONDecodeError as e:
print(f"\n{log_prefix} ❌ 최종 오류: JSON 파싱 실패 - {e}")
return f"FAILED: {file_path.name} (JSON Parse Error)"
def main(input_dir: str, output_dir: str, provider: str, workers: int, delay: float):
"""지정된 디렉토리의 모든 지원 형식 파일에 대해 OCR을 실행합니다."""
input_path = Path(input_dir)
output_path = Path(output_dir)
if not input_path.is_dir():
print(f"오류: 입력 디렉토리를 찾을 수 없습니다: {input_dir}")
return
output_path.mkdir(parents=True, exist_ok=True)
print(f"입력 디렉토리: {input_path.resolve()}")
print(f"출력 디렉토리: {output_path.resolve()}")
print(f"OCR Provider: {provider}")
files_to_process = [
p
for p in input_path.iterdir()
if p.is_file() and p.suffix.lower() in SUPPORTED_EXTENSIONS
]
if not files_to_process:
print("처리할 파일을 찾을 수 없습니다.")
return
total_files = len(files_to_process)
results = {"SUCCESS": 0, "SKIPPED": 0, "FAILED": 0}
if delay > 0:
# --- 순차(동기) 처리 모드 ---
print(f"순차 처리 모드 활성화. 요청 간 대기 시간: {delay}초")
print("-" * 40)
print(f"총 {total_files}개의 파일을 처리합니다.")
for i, file in enumerate(files_to_process):
print(f"\n--- [{i + 1}/{total_files}] ---")
try:
result_str = run_ocr_on_file(file, output_path, provider)
status = result_str.split(":")[0]
if "SUCCESS" in status:
results["SUCCESS"] += 1
elif "SKIPPED" in status:
results["SKIPPED"] += 1
else:
results["FAILED"] += 1
except Exception as exc:
print(f"\n[Main] {file.name} 처리 중 예외 발생: {exc}")
results["FAILED"] += 1
if i < total_files - 1:
print(f"요청 간 대기... ({delay}초)")
time.sleep(delay)
else:
# --- 병렬 처리 모드 ---
print(f"병렬 처리 모드 활성화. 동시 작업 수: {workers}")
print("-" * 40)
print(f"총 {total_files}개의 파일을 처리합니다.")
with ThreadPoolExecutor(max_workers=workers) as executor:
future_to_file = {
executor.submit(run_ocr_on_file, file, output_path, provider): file
for file in files_to_process
}
for i, future in enumerate(as_completed(future_to_file)):
file = future_to_file[future]
try:
result_str = future.result()
status = result_str.split(":")[0]
if "SUCCESS" in status:
results["SUCCESS"] += 1
elif "SKIPPED" in status:
results["SKIPPED"] += 1
else:
results["FAILED"] += 1
except Exception as exc:
print(f"\n[Main] {file.name} 처리 중 예외 발생: {exc}")
results["FAILED"] += 1
print(f"--- 진행 상황: {i + 1}/{total_files} ---")
print("\n" + "=" * 40)
print("모든 작업이 완료되었습니다.")
print(f" - 성공: {results['SUCCESS']}")
print(f" - 건너뜀: {results['SKIPPED']}")
print(f" - 실패: {results['FAILED']}")
print("=" * 40)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="디렉토리 내 파일에 대해 비동기 OCR을 병렬 또는 순차적으로 수행합니다.",
formatter_class=argparse.RawTextHelpFormatter,
)
parser.add_argument(
"input_dir", type=str, help="OCR을 수행할 파일들이 있는 입력 디렉토리"
)
parser.add_argument(
"output_dir", type=str, help="OCR 결과(JSON)를 저장할 출력 디렉토리"
)
parser.add_argument(
"--provider",
type=str,
default="paddle",
choices=["paddle", "upstage"],
help="사용할 OCR 공급자 (기본값: paddle)",
)
parser.add_argument(
"--workers",
type=int,
default=4,
help="병렬 처리 시 동시에 실행할 워커(스레드) 수 (기본값: 4)\n(--delay 옵션이 0보다 크면 이 값은 무시됩니다.)",
)
parser.add_argument(
"--delay",
type=float,
default=0,
help="순차 처리 시 각 요청 사이의 대기 시간(초) (기본값: 0)\n(0보다 큰 값으로 설정하면 순차 모드로 강제 실행됩니다.)",
)
args = parser.parse_args()
main(args.input_dir, args.output_dir, args.provider, args.workers, args.delay)
# 사용 예시:
# 1. Paddle OCR, 병렬 워커 4개 사용 (기본)
# python workspace/run_ocr.py ./source_documents ./result_jsons
#
# 2. Upstage OCR, 병렬 워커 8개 사용
# python workspace/run_ocr.py ./source_documents ./result_jsons --provider upstage --workers 8
#
# 3. Upstage OCR, 순차 처리 (요청마다 3초 대기)
# python workspace/run_ocr.py ./source_documents ./result_jsons --provider upstage --delay 3

92
workspace/show_summary.py
View File

@@ -1,92 +0,0 @@
# workspace/show_summary.py
import os
import json
import argparse
import pandas as pd
def generate_summary(directory_path):
"""
지정된 디렉터리에서 모든 JSON 파일을 읽어 요약 정보를 추출하고,
pandas DataFrame으로 반환합니다.
"""
summary_data = []
if not os.path.isdir(directory_path):
print(f"오류: 디렉터리를 찾을 수 없습니다 - {directory_path}")
return None
for filename in sorted(os.listdir(directory_path)):
if filename.endswith('.json'):
file_path = os.path.join(directory_path, filename)
try:
with open(file_path, 'r', encoding='utf-8') as f:
data = json.load(f)
# JSON 파일이 리스트 형태이므로 첫 번째 항목을 사용
if isinstance(data, list) and data:
item = data[0]
else:
# 예상치 못한 형식이면 건너뛰기
continue
# 필요한 정보 추출
row_data = {
'filename': item.get('filename'),
'duration_sec': item.get('time', {}).get('duration_sec')
}
# 'processed' 딕셔너리의 모든 키-값을 row_data에 추가
processed_info = item.get('processed', {})
if isinstance(processed_info, dict):
row_data.update(processed_info)
summary_data.append(row_data)
except (json.JSONDecodeError, IndexError) as e:
print(f"파일 처리 중 오류 발생 ({filename}): {e}")
except Exception as e:
print(f"알 수 없는 오류 발생 ({filename}): {e}")
if not summary_data:
print("처리할 JSON 파일이 없습니다.")
return None
return pd.DataFrame(summary_data)
def main():
"""메인 실행 함수"""
parser = argparse.ArgumentParser(description="JSON 파일들을 읽어 요약 테이블을 생성하고 CSV로 저장하는 스크립트")
parser.add_argument("input_dir", help="JSON 파일들이 포함된 입력 디렉터리 경로")
parser.add_argument("-o", "--output", help="요약 결과를 저장할 CSV 파일 경로")
args = parser.parse_args()
# pandas 출력 옵션 설정
pd.set_option('display.max_rows', 500)
pd.set_option('display.max_columns', 50)
pd.set_option('display.width', 200)
summary_df = generate_summary(args.input_dir)
if summary_df is not None:
print("\n--- JSON 처리 결과 요약 ---")
print(summary_df)
print("\n")
# CSV 파일로 저장하는 로직 추가
if args.output:
output_path = args.output
# 파일명에 .csv 확장자가 없으면 자동으로 추가
if not output_path.lower().endswith('.csv'):
output_path += '.csv'
try:
# CSV 파일 저장 시 Excel에서 한글이 깨지지 않도록 'utf-8-sig' 인코딩 사용
summary_df.to_csv(output_path, index=False, encoding='utf-8-sig')
print(f"요약 결과가 '{output_path}' 파일로 성공적으로 저장되었습니다.")
except Exception as e:
print(f"CSV 파일 저장 중 오류가 발생했습니다: {e}")
print("\n")
if __name__ == "__main__":
main()