hyungi/hyungi_document_server
1
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

feat(markdown): persist extracted images with auth routes

Browse Source 
Markdown Canonical Phase 1B.5 — marker 가 추출하던 이미지를 NAS 에 영구 저장하고
DB 메타 + 인증 라우트 + 프론트 swap 까지 wiring.

핵심 변경:
- marker-service /convert 응답에 base64 image 리스트 포함 (stateless 유지, NAS write 권한 X)
- marker_worker 가 NAS `/documents/extracted_images/{doc_id}/` 에 persist + UPSERT +
  고아 row DELETE + md_content ref 를 `docimg:img_NNN` stable scheme 으로 정규화
- /api/documents/{id}/images/{key}/raw 인증 라우트 (Cache-Control private + ETag = content_hash)
- frontend MarkdownDoc 가 placeholder card 안의 docimg ref 를 실제 <img> 로 swap

원칙:
- 이미지 binary = NAS, metadata = Postgres (학습 섹션 패턴 동일)
- image_key sequence 기반 결정적 → 재변환 idempotent
- MARKDOWN_IMAGE_PERSIST=false env 로 rollback 가능 (placeholder card 폴백 자연 유지)

기존 28건 marker success 문서는 본 PR 에서 건드리지 않음 — deploy + 신규 업로드 1건 +
sample 5건 검증 후 scripts/marker_reprocess_existing_success.py 로 targeted reprocess.

plan: ~/.claude/plans/piped-humming-crystal.md

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hyungi Ahn
2026-05-10 14:03:07 +09:00
parent 5b62c59f8a
commit 68fa86ea52
13 changed files with 891 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/documents.py
+42
View File
@@ -33,6 +33,7 @@ from core.config import settings
from core.database import get_session
from core.utils import file_hash
from models.document import Document
from models.document_image import DocumentImage
from models.queue import ProcessingQueue, enqueue_stage
from models.user import User
from services.document_telemetry import record_analyze_event, sanitize_source
@@ -670,6 +671,47 @@ async def get_document_file(
)
@router.get("/{doc_id}/images/{image_key}/raw")
async def get_document_image_raw(
doc_id: int,
image_key: str,
user: Annotated[User, Depends(get_current_user)],
session: Annotated[AsyncSession, Depends(get_session)],
):
"""marker 추출 이미지 raw bytes (Phase 1B.5).
md_content 안의 `![alt](docimg:img_NNN)` ref 를 frontend selector 가 이 라우트로 변환.
인증된 사용자만 응답 (단일 사용자 환경, ownership 컬럼 없음 — get_current_user 게이트 충분).
"""
# 문서 존재 확인 (image_key 만 있고 doc 가 사라진 케이스 차단)
doc = await session.get(Document, doc_id)
if doc is None:
raise HTTPException(status_code=404, detail="문서를 찾을 수 없습니다")
img = await session.scalar(
select(DocumentImage).where(
DocumentImage.document_id == doc_id,
DocumentImage.image_key == image_key,
)
)
if img is None:
raise HTTPException(status_code=404, detail="이미지를 찾을 수 없습니다")
file_path = Path(img.file_path)
if not file_path.is_file():
raise HTTPException(status_code=410, detail="파일이 사라졌습니다")
return FileResponse(
str(file_path),
media_type=img.mime_type,
headers={
# 인증 라우트라 CDN/공용 cache 금지. 단일 사용자라 private + 1h 충분.
"Cache-Control": "private, max-age=3600",
"ETag": f'"{img.content_hash}"',
},
)
@router.post("/", response_model=DocumentResponse, status_code=201)
async def upload_document(
request: Request,
app/models/document_image.py
+42
View File
@@ -0,0 +1,42 @@
"""document_images ORM (Phase 1B.5) — marker 추출 이미지 메타.
저장: NAS `/documents/extracted_images/{document_id}/{image_key}.{ext}`
표시: GET /api/documents/{doc_id}/images/{image_key}/raw (인증 필요)
md_content 의 ref 는 `![alt](docimg:img_001)` 형식 — image_key 가 sequence 기반 결정적이라
재변환 시 idempotent.
"""
from datetime import datetime
from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from core.database import Base
class DocumentImage(Base):
__tablename__ = "document_images"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
document_id: Mapped[int] = mapped_column(
BigInteger, ForeignKey("documents.id", ondelete="CASCADE"), nullable=False
)
image_key: Mapped[str] = mapped_column(String(32), nullable=False)
relative_path: Mapped[str] = mapped_column(Text, nullable=False)
file_path: Mapped[str] = mapped_column(Text, nullable=False)
mime_type: Mapped[str] = mapped_column(Text, nullable=False)
file_size: Mapped[int] = mapped_column(BigInteger, nullable=False)
content_hash: Mapped[str] = mapped_column(String(64), nullable=False)
width: Mapped[int | None] = mapped_column(Integer)
height: Mapped[int | None] = mapped_column(Integer)
page_index: Mapped[int | None] = mapped_column(Integer)
alt_text: Mapped[str | None] = mapped_column(Text)
source_slug: Mapped[str | None] = mapped_column(Text)
extraction_engine: Mapped[str] = mapped_column(
String(32), nullable=False, default="marker"
)
extraction_engine_version: Mapped[str | None] = mapped_column(String(32))
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=datetime.now, nullable=False
)
app/workers/marker_worker.py
+235 -8
View File
@@ -1,13 +1,19 @@
"""marker_worker — markdown stage 소비. Phase 1B Round 5.
"""marker_worker — markdown stage 소비. Phase 1B + Phase 1B.5 (ImgAuth).
플로우:
classify_worker 완료 → enqueue 'markdown' stage
classify_worker 완료 → enqueue 'markdown' stage (또는 reprocess 스크립트가 force=True 로 enqueue)
→ marker_worker.process()
→ doc_type / 확장자 / page_count 가드 → marker-service POST /convert
→ 응답 이미지 NAS persist + document_images UPSERT + md_content ref 정규화
→ md_content 저장 또는 doc-level failed (404/422) 또는 transient raise (5xx → queue retry)
plan: ~/.claude/plans/plan-idempotent-sundae.md
이미지 저장 위치: NAS `/documents/extracted_images/{document_id}/{image_key}.{ext}`
md_content ref 형식: `![alt](docimg:img_001)` — image_key 가 sequence 기반 결정적 → idempotent.
plan: ~/.claude/plans/piped-humming-crystal.md
"""
import base64
import hashlib
import logging
import os
import re
@@ -16,10 +22,13 @@ from typing import Any
import fitz # PyMuPDF
import httpx
from sqlalchemy import update
from sqlalchemy import delete, desc, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession
from models.document import Document
from models.document_image import DocumentImage
from models.queue import ProcessingQueue
logger = logging.getLogger(__name__)
@@ -27,6 +36,22 @@ MARKER_ENDPOINT = "http://marker-service:3300/convert"
MARKER_TIMEOUT = 300 # 큰 PDF 5 분 한도
MAX_PAGES = 200 # 페이지 hard limit
# Phase 1B.5: 이미지 NAS persist 토글. rollback 시 false → 응답 images 무시 + md_content
# rewrite skip → placeholder card 폴백 자연 유지. 환경변수 미설정 = 기본 활성화.
MARKDOWN_IMAGE_PERSIST = os.getenv("MARKDOWN_IMAGE_PERSIST", "true").lower() in ("1", "true", "yes")
EXTRACTED_IMAGES_ROOT = Path("/documents/extracted_images")
# md_content image ref 정규식. alt + href 캡처. 외부 URL 은 보존 (slug 매칭 안 되는 경우).
_IMAGE_REF_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
_FORMAT_TO_MIME = {
"png": "image/png",
"jpeg": "image/jpeg",
"jpg": "image/jpeg",
"webp": "image/webp",
"gif": "image/gif",
}
# Phase 1B = PDF only. DOCX 등은 후속 Phase.
SUPPORTED_EXTENSIONS = {".pdf"}
@@ -87,6 +112,11 @@ async def process(document_id: int, session: AsyncSession) -> None:
logger.warning(f"[marker] document {document_id} not found")
return
# ---- (0) force_reprocess flag — reprocess 스크립트가 queue payload 로 전달 ----
force_reprocess = await _read_force_reprocess(session, document_id)
if force_reprocess:
logger.info(f"markdown_force_reprocess id={document_id}")
# ---- (1) doc_type skip ----
if doc.document_type in SKIP_DOC_TYPES:
logger.info(
@@ -180,9 +210,33 @@ async def process(document_id: int, session: AsyncSession) -> None:
await _fail(session, document_id, str(exc)[:1000])
return
# ---- (7) success ----
md_content = data["md_content"]
# ---- (7) image persist + md_content rewrite (Phase 1B.5) ----
md_content_raw = data["md_content"]
images_resp = data.get("images") if MARKDOWN_IMAGE_PERSIST else None
saved_images: list[dict[str, Any]] = []
if images_resp:
try:
saved_images = _persist_images_to_nas(document_id, images_resp)
except OSError as exc:
# NAS 일시 끊김 등 — transient. queue retry 로 복구.
logger.warning(
f"[marker] image persist NAS write failed id={document_id}: "
f"{type(exc).__name__}: {exc}"
)
raise
# md_content 안의 ref 를 stable internal scheme `docimg:img_NNN` 으로 정규화.
slug_to_key = {img["source_slug"]: img["image_key"] for img in saved_images}
md_content = _rewrite_image_refs(md_content_raw, slug_to_key)
# quality 메트릭은 정규화 *후* md_content 기준 (실제 저장본). image_count 도 정확.
quality = _compute_quality(md_content, doc.extracted_text or "", data["raw_metrics"])
if data.get("images_truncated"):
quality.setdefault("warnings", []).append("images_truncated")
# ---- (8) DB 트랜잭션 — documents UPDATE + document_images UPSERT + 고아 row DELETE ----
orphan_paths = await _sync_document_images(session, document_id, saved_images, data)
await session.execute(
update(Document).where(Document.id == document_id).values(
@@ -191,7 +245,7 @@ async def process(document_id: int, session: AsyncSession) -> None:
md_extraction_engine=data["engine"],
md_extraction_engine_version=data["engine_version"],
md_extraction_quality=quality,
md_content_hash=data["md_content_hash"],
md_content_hash=hashlib.sha256(md_content.encode("utf-8")).hexdigest(),
md_source_hash=doc.file_hash,
md_generated_at=_now(),
md_extraction_error=None,
@@ -201,11 +255,184 @@ async def process(document_id: int, session: AsyncSession) -> None:
)
)
await session.commit()
# ---- (9) commit 후 고아 NAS 파일 unlink (best-effort, 실패해도 DB 정합 유지) ----
for orphan_path in orphan_paths:
try:
Path(orphan_path).unlink(missing_ok=True)
except Exception as exc:
logger.warning(
f"[marker] orphan image unlink failed id={document_id} path={orphan_path}: "
f"{type(exc).__name__}: {exc}"
)
logger.info(
f"[marker] success id={document_id} len={len(md_content)} elapsed_ms={data['elapsed_ms']}"
f"[marker] success id={document_id} len={len(md_content)} "
f"images={len(saved_images)} orphans_removed={len(orphan_paths)} "
f"elapsed_ms={data['elapsed_ms']}"
)
async def _read_force_reprocess(session: AsyncSession, document_id: int) -> bool:
"""현재 markdown stage queue 행의 payload.force_reprocess 조회. 없으면 False."""
row = await session.scalar(
select(ProcessingQueue)
.where(
ProcessingQueue.document_id == document_id,
ProcessingQueue.stage == "markdown",
ProcessingQueue.status == "processing",
)
.order_by(desc(ProcessingQueue.id))
.limit(1)
)
if not row or not row.payload:
return False
return bool(row.payload.get("force_reprocess"))
def _persist_images_to_nas(
document_id: int, images_resp: list[dict[str, Any]]
) -> list[dict[str, Any]]:
"""marker 응답 이미지 list 를 NAS 에 저장하고 메타 dict 리스트 반환.
image_key 는 sequence 기반 결정적 (`img_001` → `img_NNN`, marker 출력 순서 = 안정적).
같은 doc 재변환 시 같은 key 가 같은 path 에 overwrite → idempotent.
"""
img_root = EXTRACTED_IMAGES_ROOT / str(document_id)
img_root.mkdir(parents=True, exist_ok=True)
saved: list[dict[str, Any]] = []
for seq, img in enumerate(images_resp, start=1):
try:
raw_bytes = base64.b64decode(img["bytes_b64"])
except Exception as exc:
logger.warning(
f"[marker] image base64 decode failed id={document_id} "
f"seq={seq} slug={img.get('slug')}: {exc}"
)
continue
fmt = (img.get("format") or "png").lower()
ext = "jpeg" if fmt == "jpg" else fmt
image_key = f"img_{seq:03d}"
filename = f"{image_key}.{ext}"
rel_path = f"extracted_images/{document_id}/{filename}"
abs_path = img_root / filename
# NAS write — 실패 시 OSError raise (transient retry).
abs_path.write_bytes(raw_bytes)
saved.append({
"image_key": image_key,
"source_slug": img.get("slug") or "",
"relative_path": rel_path,
"file_path": str(abs_path),
"mime_type": _FORMAT_TO_MIME.get(ext, "application/octet-stream"),
"file_size": len(raw_bytes),
"content_hash": hashlib.sha256(raw_bytes).hexdigest(),
"width": img.get("width"),
"height": img.get("height"),
})
return saved
def _rewrite_image_refs(md_text: str, slug_to_key: dict[str, str]) -> str:
"""md_content 안의 `![alt](slug)` ref 를 `![alt](docimg:image_key)` 로 정규화.
- slug_to_key 에 없는 href 는 원본 유지 (외부 URL / 경로 변형 등)
- alt 그대로 보존
- 매칭 = href 가 (a) 정확히 slug 와 같음 OR (b) basename 이 slug 와 같음
(marker 가 `_page_0_Picture_3.jpeg` 또는 `subdir/_page_0_Picture_3.jpeg` 어느 쪽도 emit 가능)
"""
if not slug_to_key:
return md_text
def _replace(match: re.Match) -> str:
alt, href = match.group(1), match.group(2)
# 정확 매치 우선
if href in slug_to_key:
return f"![{alt}](docimg:{slug_to_key[href]})"
# basename 매치 fallback
basename = href.rsplit("/", 1)[-1]
if basename in slug_to_key:
return f"![{alt}](docimg:{slug_to_key[basename]})"
return match.group(0)
return _IMAGE_REF_RE.sub(_replace, md_text)
async def _sync_document_images(
session: AsyncSession,
document_id: int,
saved_images: list[dict[str, Any]],
response_data: dict[str, Any],
) -> list[str]:
"""document_images 동기화 — 신규 keys UPSERT + 고아 row DELETE.
반환: commit 후 unlink 해야 할 NAS 파일 경로 리스트.
"""
# 기존 row 조회 (file_path 보존 — 고아 파일 unlink 용)
existing = (await session.execute(
select(DocumentImage.image_key, DocumentImage.file_path)
.where(DocumentImage.document_id == document_id)
)).all()
existing_map = {key: path for key, path in existing}
new_keys = {img["image_key"] for img in saved_images}
orphan_keys = set(existing_map.keys()) - new_keys
orphan_paths = [existing_map[k] for k in orphan_keys]
# UPSERT — image_key 가 같으면 file_path/hash/dimensions 등을 갱신.
for img in saved_images:
stmt = (
pg_insert(DocumentImage)
.values(
document_id=document_id,
image_key=img["image_key"],
relative_path=img["relative_path"],
file_path=img["file_path"],
mime_type=img["mime_type"],
file_size=img["file_size"],
content_hash=img["content_hash"],
width=img.get("width"),
height=img.get("height"),
page_index=img.get("page_index"),
alt_text=img.get("alt_text"),
source_slug=img.get("source_slug"),
extraction_engine=response_data.get("engine") or "marker",
extraction_engine_version=response_data.get("engine_version"),
)
.on_conflict_do_update(
index_elements=["document_id", "image_key"],
set_={
"relative_path": img["relative_path"],
"file_path": img["file_path"],
"mime_type": img["mime_type"],
"file_size": img["file_size"],
"content_hash": img["content_hash"],
"width": img.get("width"),
"height": img.get("height"),
"page_index": img.get("page_index"),
"alt_text": img.get("alt_text"),
"source_slug": img.get("source_slug"),
"extraction_engine": response_data.get("engine") or "marker",
"extraction_engine_version": response_data.get("engine_version"),
},
)
)
await session.execute(stmt)
if orphan_keys:
await session.execute(
delete(DocumentImage).where(
DocumentImage.document_id == document_id,
DocumentImage.image_key.in_(orphan_keys),
)
)
return orphan_paths
def _compute_quality(md: str, raw_text: str, raw_metrics: dict[str, Any]) -> dict[str, Any]:
"""1B 휴리스틱 quality. 임계 판정 미적용 (Phase 1D 후행)."""
heading_lines = re.findall(r"^(#{1,6})\s", md, flags=re.MULTILINE)