Kyeongmin/_Geulbeot
1
0
Fork 0
Code Issues 37 Pull Requests Actions Packages Projects 4 Releases Wiki Activity

Upload converter pipeline: step2_extract.py

Browse Source
This commit is contained in:
이경민
2026-03-19 09:40:52 +09:00
parent dd936b230b
commit 7ebd14164a
1 changed files with 145 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
03.Code/업로드용/converters/pipeline/step2_extract.py
View File

@@ -2,12 +2,9 @@
""" """
extract_1_v2.py extract_1_v2.py
PDF에서 텍스트(md)와 이미지(png)를 추출하는 기능을 담당하는 모듈. PDF에서 텍스트(md)와 이미지(png)를 추출
- 원본 폴더 구조 유지 - 하위 폴더 구조 유지
- 이미지 추출 시 캡션(예: <그림 1>)과 연결 - 이미지 메타데이터 JSON 생성 (폴더경로, 파일명, 페이지, 위치, 캡션 등)
- 헤더/푸터 제외 로직 포함
- OCR 옵션 지원 (Tesseract 설치 필요)
- JSON 기반 메타데이터 기록 (이미지경로, 캡션 등)
""" """
import fitz # PyMuPDF import fitz # PyMuPDF
@@ -30,15 +27,14 @@ try:
TESSERACT_AVAILABLE = True TESSERACT_AVAILABLE = True
except ImportError: except ImportError:
TESSERACT_AVAILABLE = False TESSERACT_AVAILABLE = False
print("[INFO] pytesseract 미설치. 이미지 텍스트 분석 기능이 제한됩니다.") print("[INFO] pytesseract 미설치 - 텍스트 잘림 필터 비활성화")
# ===== 설정 및 상수 =====
CAPTION_PATTERN = re.compile( CAPTION_PATTERN = re.compile(
r'^\s*(?:[<\[\(\{]\s*)?(그림|figure|fig)\s*\.?\s*(?:[<\[\(\{]\s*)?0*\d+(?:\s*[-~]\s*\d+)?', r'^\s*(?:[<\[\(\{]\s*)?(그림|figure|fig)\s*\.?\s*(?:[<\[\(\{]\s*)?0*\d+(?:\s*[-–]\s*\d+)?',
re.IGNORECASE re.IGNORECASE
) )
# ===== 이미지 추출 및 캡션 매칭 핵심 로직 =====
def get_figure_rects(page): def get_figure_rects(page):
""" """
@@ -194,7 +190,7 @@ def get_figure_rects(page):
'rect': union_rect, 'rect': union_rect,
'caption_index': cap['index'], 'caption_index': cap['index'],
'caption_rect': cap['rect'], 'caption_rect': cap['rect'],
'caption_text': cap['text'].strip() # 원본 캡션 텍스트 유지 'caption_text': cap['text'].strip() # ★ 캡션 텍스트 저장
}) })
return figure_regions return figure_regions
@@ -224,27 +220,28 @@ def keep_figure(pix):
return True, nonwhite_ratio, edge_ratio, var return True, nonwhite_ratio, edge_ratio, var
# ===== 추가 이미지 필터링 알고리즘 (v2.1) ===== # ===== 추가 이미지 필터 함수들 (v2.1) =====
def pix_to_pil(pix): def pix_to_pil(pix):
"""PyMuPDF Pixmap을 PIL Image로 변환""" """PyMuPDF Pixmap을 PIL Image로 변환"""
img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples) img_data = pix.tobytes("png")
return img return Image.open(io.BytesIO(img_data))
def has_cut_text_at_boundary(pix, margin=5): def has_cut_text_at_boundary(pix, margin=5):
""" """
이미지 경계선에 텍스트가 잘려 있는지 확인 이미지 경계에서 텍스트가 잘렸는지 감지
- 이미지 주변에 근접한 텍스트 박스가 있으면 필터링 대상으로 판단 - 이미지 테두리 근처에 텍스트 박스가 있으면 잘린 것으로 판단
Args: Args:
pix: PyMuPDF Pixmap pix: PyMuPDF Pixmap
margin: 경계선으로부터의 여백 (기본 5px) margin: 경계로부터의 여유 픽셀 (기본 5px)
Returns: Returns:
bool: 텍스트가 잘린 경우 True bool: 텍스트가 잘렸으면 True
""" """
if not TESSERACT_AVAILABLE: if not TESSERACT_AVAILABLE:
return False # OCR 없으면 우선 통과 return False # OCR 없으면 필터 비활성화
try: try:
img = pix_to_pil(pix) img = pix_to_pil(pix)
@@ -255,7 +252,7 @@ def has_cut_text_at_boundary(pix, margin=5):
for i, text in enumerate(data['text']): for i, text in enumerate(data['text']):
text = str(text).strip() text = str(text).strip()
if len(text) < 2: # 너무 짧은 텍스트 무시 if len(text) < 2: # 너무 짧은 텍스트는 무시
continue continue
x = data['left'][i] x = data['left'][i]
@@ -263,14 +260,14 @@ def has_cut_text_at_boundary(pix, margin=5):
w = data['width'][i] w = data['width'][i]
h = data['height'][i] h = data['height'][i]
# 텍스트가 상하좌우 경계선에 너무 가깝다면 = 잘린 텍스트 박스일 가능성 높음 # 텍스트가 이미지 경계에 너무 가까우면 = 잘린 것
# 좌측 경계 # 왼쪽 경계
if x <= margin: if x <= margin:
return True return True
# 우측 경계 # 오른쪽 경계
if x + w >= width - margin: if x + w >= width - margin:
return True return True
# 상단 경계 (제목 형태는 제외하기 위해 높이 제한 추가) # 상단 경계 (헤더 제외를 위해 좀 더 여유)
if y <= margin and h < height * 0.3: if y <= margin and h < height * 0.3:
return True return True
# 하단 경계 # 하단 경계
@@ -280,15 +277,15 @@ def has_cut_text_at_boundary(pix, margin=5):
return False return False
except Exception as e: except Exception as e:
# OCR 실패 시 필터링 없이 통과 (보수적 접근) # OCR 실패 시 필터 통과 (이미지 유지)
return False return False
def is_decorative_background(pix, edge_threshold=0.02, color_var_threshold=500): def is_decorative_background(pix, edge_threshold=0.02, color_var_threshold=500):
""" """
배경 패턴(장식) 이미지인지 확인 배경 패턴 + 텍스트만 있는 장식용 이미지인지 감지
- 엣지 비율이 낮고 (복잡한 도형/사진이 아님) - 엣지가 적고 (복잡한 도표/사진이 아님)
- 색상 분산이 낮거나 특정 범위 내인 경우 (단조로운 그라데이션 등) - 색상 다양성이 낮으면 (단순 그라데이션 배경)
Args: Args:
pix: PyMuPDF Pixmap pix: PyMuPDF Pixmap
@@ -296,20 +293,21 @@ def is_decorative_background(pix, edge_threshold=0.02, color_var_threshold=500):
color_var_threshold: 색상 분산 임계값 color_var_threshold: 색상 분산 임계값
Returns: Returns:
bool: 배경 이미지인 경우 True bool: 장식용 배경이면 True
""" """
try: try:
nonwhite_ratio, edge_ratio, var = pixmap_metrics(pix) nonwhite_ratio, edge_ratio, var = pixmap_metrics(pix)
# 엣지 비율이 2% 미만이면서 단조로운 색상 분포라면 배경 패턴 가능성 높음 # 엣지가 거의 없고 (단순한 이미지)
# 색상 분산도 낮으면 (배경 패턴)
if edge_ratio < edge_threshold and var < color_var_threshold: if edge_ratio < edge_threshold and var < color_var_threshold:
# 추가적으로 텍스트 이미지인지 OCR로 체크 가능 # 추가 확인: 텍스트만 있는지 OCR로 체크
if TESSERACT_AVAILABLE: if TESSERACT_AVAILABLE:
try: try:
img = pix_to_pil(pix) img = pix_to_pil(pix)
text = pytesseract.image_to_string(img, lang='kor+eng').strip() text = pytesseract.image_to_string(img, lang='kor+eng').strip()
# 텍스트가 있고, 엣지 비율이 아주 낮다면 = 텍스트 배경 장식 # 텍스트가 있고, 이미지가 단순하면 = 텍스트 배경
if len(text) > 3 and edge_ratio < 0.015: if len(text) > 3 and edge_ratio < 0.015:
return True return True
except: except:
@@ -325,12 +323,13 @@ def is_decorative_background(pix, edge_threshold=0.02, color_var_threshold=500):
def is_header_footer_region(rect, page_rect, height_threshold=0.12): def is_header_footer_region(rect, page_rect, height_threshold=0.12):
""" """
헤더/푸터 영역에 포함되는지 확인 헤더/푸터 영역에 있는 이미지인지 감지
- 상단 12% 또는 하단 12%에 위치한 작은 이미지는 필터링 - 페이지 상단 12% 또는 하단 12%에 위치
- 높이가 낮은 strip 형태
Args: Args:
rect: 이미지 영역 (fitz.Rect) rect: 이미지 영역 (fitz.Rect)
page_rect: 전체 페이지 영역 (fitz.Rect) page_rect: 페이지 전체 영역 (fitz.Rect)
height_threshold: 헤더/푸터 영역 비율 (기본 12%) height_threshold: 헤더/푸터 영역 비율 (기본 12%)
Returns: Returns:
@@ -341,13 +340,13 @@ def is_header_footer_region(rect, page_rect, height_threshold=0.12):
# 상단 영역 체크 # 상단 영역 체크
if rect.y0 < page_height * height_threshold: if rect.y0 < page_height * height_threshold:
# 매우 얇은 이미지(구분선 등)나 작은 로고 등 # 높이가 페이지의 15% 미만인 strip이면 헤더
if img_height < page_height * 0.15: if img_height < page_height * 0.15:
return True return True
# 하단 영역 체크 # 하단 영역 체크
if rect.y1 > page_height * (1 - height_threshold): if rect.y1 > page_height * (1 - height_threshold):
# 푸터 영역의 작은 이미지 # 높이가 페이지의 15% 미만인 strip이면 푸터
if img_height < page_height * 0.15: if img_height < page_height * 0.15:
return True return True
@@ -356,25 +355,25 @@ def is_header_footer_region(rect, page_rect, height_threshold=0.12):
def should_filter_image(pix, rect, page_rect): def should_filter_image(pix, rect, page_rect):
""" """
여러 필터링 규칙을 종합하여 이미지 보존 여부 결정 이미지를 필터링해야 하는지 종합 판단
Args: Args:
pix: PyMuPDF Pixmap pix: PyMuPDF Pixmap
rect: 이미지 영역 rect: 이미지 영역
page_rect: 전체 페이지 영역 page_rect: 페이지 전체 영역
Returns: Returns:
tuple: (필터링 여부, 필터링 이유) tuple: (필터링 여부, 필터링 사유)
""" """
# 1. 헤더/푸터 영역 체크 # 1. 헤더/푸터 영역 체크
if is_header_footer_region(rect, page_rect): if is_header_footer_region(rect, page_rect):
return True, "header_footer" return True, "header_footer"
# 2. 잘린 텍스트 포함 여부 체크 # 2. 텍스트 잘림 체크
if has_cut_text_at_boundary(pix): if has_cut_text_at_boundary(pix):
return True, "cut_text" return True, "cut_text"
# 3. 배경 장식 여부 체크 # 3. 장식용 배경 체크
if is_decorative_background(pix): if is_decorative_background(pix):
return True, "decorative_background" return True, "decorative_background"
@@ -383,26 +382,26 @@ def should_filter_image(pix, rect, page_rect):
def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata): def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
""" """
PDF 내용 추출 메인 함수 PDF 내용 추출
Args: Args:
pdf_path: PDF 경로 pdf_path: PDF 파일 경로
output_md_path: 출력 MD 경로 output_md_path: 출력 MD 파일 경로
img_dir: 이미지 저장 폴더 img_dir: 이미지 저장 폴더
metadata: 메타데이터 정보 (폴더 경로, 파일명 등) metadata: 메타데이터 딕셔너리 (폴더 경로, 파일명 등)
Returns: Returns:
image_metadata_list: 추출된 이미지 메타데이터 리스트 image_metadata_list: 추출된 이미지들의 메타데이터 리스트
""" """
os.makedirs(img_dir, exist_ok=True) os.makedirs(img_dir, exist_ok=True)
image_metadata_list = [] # 이미지 메타데이터 정보 수집 image_metadata_list = [] # ★ 이미지 메타데이터 수집
doc = fitz.open(pdf_path) doc = fitz.open(pdf_path)
total_pages = len(doc) total_pages = len(doc)
with open(output_md_path, "w", encoding="utf-8") as md_file: with open(output_md_path, "w", encoding="utf-8") as md_file:
# 문서 메타데이터 정보 추가 # ★ 메타데이터 헤더 추가
md_file.write(f"---\n") md_file.write(f"---\n")
md_file.write(f"source_pdf: {metadata['pdf_name']}\n") md_file.write(f"source_pdf: {metadata['pdf_name']}\n")
md_file.write(f"source_folder: {metadata['relative_folder']}\n") md_file.write(f"source_folder: {metadata['relative_folder']}\n")
@@ -427,7 +426,7 @@ def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
pix = page.get_pixmap(clip=rect, dpi=150, colorspace=fitz.csRGB) pix = page.get_pixmap(clip=rect, dpi=150, colorspace=fitz.csRGB)
# 추가 필터링 로직 적용 (v2.1) # ★ 추가 필터 적용 (v2.1)
should_filter, filter_reason = should_filter_image(pix, rect, page.rect) should_filter, filter_reason = should_filter_image(pix, rect, page.rect)
if should_filter: if should_filter:
continue continue
@@ -440,7 +439,7 @@ def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
fig['img_name'] = img_name fig['img_name'] = img_name
kept_figures.append(fig) kept_figures.append(fig)
# 이미지 메타데이터 수집 # ★ 이미지 메타데이터 수집
image_metadata_list.append({ image_metadata_list.append({
"image_file": img_name, "image_file": img_name,
"image_path": str(Path(img_dir) / img_name), "image_path": str(Path(img_dir) / img_name),
@@ -541,7 +540,7 @@ def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
clip=block_rect, dpi=150, colorspace=fitz.csRGB clip=block_rect, dpi=150, colorspace=fitz.csRGB
) )
# 추가 필터링 로직 적용 (v2.1) # ★ 추가 필터 적용 (v2.1)
should_filter, filter_reason = should_filter_image(pix, block_rect, page.rect) should_filter, filter_reason = should_filter_image(pix, block_rect, page.rect)
if should_filter: if should_filter:
continue continue
@@ -564,7 +563,7 @@ def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
"md": md, "md": md,
}) })
# 이미지 메타데이터 수집 # ★ 캡션 없는 이미지 메타데이터
image_metadata_list.append({ image_metadata_list.append({
"image_file": img_name, "image_file": img_name,
"image_path": str(Path(img_dir) / img_name), "image_path": str(Path(img_dir) / img_name),
@@ -586,7 +585,7 @@ def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
uncaptioned_idx += 1 uncaptioned_idx += 1
continue continue
# 레이아웃 정렬 # 읽기 순서 정렬
text_items = [it for it in items if it["kind"] == "text"] text_items = [it for it in items if it["kind"] == "text"]
page_w = page.rect.width page_w = page.rect.width
mid = page_w / 2.0 mid = page_w / 2.0
@@ -669,16 +668,15 @@ def extract_pdf_content(pdf_path, output_md_path, img_dir, metadata):
def process_all_pdfs(input_dir, output_dir): def process_all_pdfs(input_dir, output_dir):
""" """
BASE_DIR 내의 모든 PDF를 순차적으로 처리 BASE_DIR 하위의 모든 PDF를 재귀적으로 처리
폴더 구조를 유지하여 OUTPUT_BASE에 저장 폴더 구조를 유지하면서 OUTPUT_BASE에 저장
""" """
BASE_DIR = Path(input_dir) BASE_DIR = Path(input_dir)
OUTPUT_BASE = Path(output_dir) OUTPUT_BASE = Path(output_dir)
# 출력 폴더 생성 # 출력 폴더 생성
OUTPUT_BASE.mkdir(parents=True, exist_ok=True) OUTPUT_BASE.mkdir(parents=True, exist_ok=True)
# 전체 추출 된 이미지 메타데이터 통합 # 전체 이미지 메타데이터 수집
all_image_metadata = [] all_image_metadata = []
# 처리 통계 # 처리 통계
@@ -693,16 +691,16 @@ def process_all_pdfs(input_dir, output_dir):
failed_files = [] failed_files = []
print(f"=" * 60) print(f"=" * 60)
print(f"PDF 콘텐츠 추출 시작") print(f"PDF 추출 시작")
print(f"소스 폴더: {BASE_DIR}") print(f"원본 폴더: {BASE_DIR}")
print(f"출력 폴더: {OUTPUT_BASE}") print(f"출력 폴더: {OUTPUT_BASE}")
print(f"=" * 60) print(f"=" * 60)
# 모든 PDF 파일 찾기 # 모든 PDF 파일 찾기
pdf_files = list(BASE_DIR.rglob("*.pdf")) + list(BASE_DIR.rglob("*.PDF")) pdf_files = list(BASE_DIR.rglob("*.pdf"))
stats["total_pdfs"] = len(pdf_files) stats["total_pdfs"] = len(pdf_files)
print(f"발견된 PDF: {len(pdf_files)}개\n") print(f"\n총 {len(pdf_files)}개 PDF 발견\n")
for idx, pdf_path in enumerate(pdf_files, 1): for idx, pdf_path in enumerate(pdf_files, 1):
try: try:
@@ -744,7 +742,7 @@ def process_all_pdfs(input_dir, output_dir):
stats["success"] += 1 stats["success"] += 1
stats["total_images"] += len(image_metas) stats["total_images"] += len(image_metas)
print(f" 완료 (약 {len(image_metas)}개 이미지 추출)") print(f" ✓ 완료 (이미지 {len(image_metas)}개)")
except Exception as e: except Exception as e:
stats["failed"] += 1 stats["failed"] += 1
@@ -752,14 +750,14 @@ def process_all_pdfs(input_dir, output_dir):
"file": str(pdf_path), "file": str(pdf_path),
"error": str(e) "error": str(e)
}) })
print(f" 오류 발생: {e}") print(f" ✗ 실패: {e}")
# 전체 이미지 메타데이터 저장 # 전체 이미지 메타데이터 저장
meta_output_path = OUTPUT_BASE / "image_metadata.json" meta_output_path = OUTPUT_BASE / "image_metadata.json"
with open(meta_output_path, "w", encoding="utf-8") as f: with open(meta_output_path, "w", encoding="utf-8") as f:
json.dump(all_image_metadata, f, ensure_ascii=False, indent=2) json.dump(all_image_metadata, f, ensure_ascii=False, indent=2)
# 처리 결과 요약 저장 # 처리 요약 저장
summary = { summary = {
"processed_at": datetime.now().isoformat(), "processed_at": datetime.now().isoformat(),
"source_dir": str(BASE_DIR), "source_dir": str(BASE_DIR),
@@ -774,9 +772,9 @@ def process_all_pdfs(input_dir, output_dir):
# 결과 출력 # 결과 출력
print(f"\n" + "=" * 60) print(f"\n" + "=" * 60)
print(f"추출 작업 완료!") print(f"추출 완료!")
print(f"=" * 60) print(f"=" * 60)
print(f"총 대상: {stats['total_pdfs']}개") print(f"총 PDF: {stats['total_pdfs']}개")
print(f"성공: {stats['success']}개") print(f"성공: {stats['success']}개")
print(f"실패: {stats['failed']}개") print(f"실패: {stats['failed']}개")
print(f"추출된 이미지: {stats['total_images']}개") print(f"추출된 이미지: {stats['total_images']}개")
@@ -784,4 +782,10 @@ def process_all_pdfs(input_dir, output_dir):
print(f"처리 요약: {summary_path}") print(f"처리 요약: {summary_path}")
if failed_files: if failed_files:
print(f"\n실패한 파일 목록은 summary_path에서 확인 가능합니다.") print(f"\n실패한 파일:")
for f in failed_files:
print(f" - {f['file']}: {f['error']}")
if __name__ == "__main__":
process_all_pdfs()