Merge pull request 'wip/gpu-main-snapshot-2026-05-11' (#7) from wip/gpu-main-snapshot-2026-05-11 into main

Reviewed-on: #7

app/core/config.py

@@ -88,6 +88,12 @@ class Settings(BaseModel):
     # NFS 경유 별도 마운트된 Roon 라이브러리.
     roon_library_path: str = ""
 
+    # KGS Code 등 외부 작성 마크다운 자료 추가 스캔 경로 (PKM 상대 경로, 쉼표 구분).
+    # env: ADDITIONAL_WATCH_TARGETS=Knowledge/Industrial_Safety/가스기사/KGS_Code,...
+    # 모두 expected_category="library" 로 처리 (md/pdf/docx 등 문서 확장자만 수락).
+    # Inbox/Recordings/Videos 기본 스캔 외에 추가만 허용.
+    additional_watch_targets: list[str] = []
+
     # 분류 체계
     taxonomy: dict = {}
     document_types: list[str] = []
@@ -108,6 +114,10 @@ def load_settings() -> Settings:
     stt_endpoint = os.getenv("STT_ENDPOINT", "http://stt-service:3300")
     roon_library_path = os.getenv("ROON_LIBRARY_PATH", "")
 
+    # ADDITIONAL_WATCH_TARGETS — 쉼표 구분 (공백 제거)
+    awt_raw = os.getenv("ADDITIONAL_WATCH_TARGETS", "")
+    additional_watch_targets = [p.strip() for p in awt_raw.split(",") if p.strip()]
+
     # config.yaml — Docker 컨테이너 내부(/app/config.yaml) 또는 프로젝트 루트
     config_path = Path("/app/config.yaml")
     if not config_path.exists():
@@ -172,6 +182,7 @@ def load_settings() -> Settings:
         ocr_endpoint=ocr_endpoint,
         stt_endpoint=stt_endpoint,
         roon_library_path=roon_library_path,
+        additional_watch_targets=additional_watch_targets,
         taxonomy=taxonomy,
         document_types=document_types,
         upload=upload_cfg,

app/workers/classify_worker.py

@@ -21,9 +21,12 @@ PR-B B-1 tier triage (신규, 4B gemma Ollama):
 from __future__ import annotations
 
 import json
+import re
 import time
 from datetime import datetime, timezone
 
+import yaml
+
 from pydantic import BaseModel, Field, ValidationError
 from sqlalchemy import text as sql_text
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -301,8 +304,42 @@ def _distill(triage_out: TriageOutput, limit: int = 2000) -> str:
     return "\n".join(parts)[:limit]
 
 
+
+# ───────────────────── frontmatter 파싱 (옵션 C) ──────────────────────
+
+# YAML frontmatter (--- ... ---) + body 분리. body 가 없거나 frontmatter 가 형식 오류여도 안전하게 fallback.
+_FM_PATTERN = re.compile("^---\\s*\\n(.*?)\\n---\\s*\\n?(.*)$", re.DOTALL)
+
+def _parse_frontmatter(extracted_text: str) -> tuple[dict, str]:
+    """extracted_text 시작에 YAML frontmatter 가 있으면 (frontmatter_dict, body) 반환.
+    없으면 ({}, extracted_text). YAML 파싱 실패 시도 ({}, extracted_text) 로 안전 fallback.
+    """
+    if not extracted_text or not extracted_text.startswith("---"):
+        return {}, extracted_text
+    m = _FM_PATTERN.match(extracted_text)
+    if not m:
+        return {}, extracted_text
+    fm_text, body = m.group(1), m.group(2)
+    try:
+        fm = yaml.safe_load(fm_text)
+        if not isinstance(fm, dict):
+            return {}, extracted_text
+        return fm, body
+    except yaml.YAMLError:
+        return {}, extracted_text
+
+
+# frontmatter 우선 인식: code/section/source_pdf/source_pages/source_basis/verified_level/verification_pending
+# 등 원문 추적 메타데이터는 LLM 이 절대 덮어쓰지 못하게 차단.
+_FRONTMATTER_PRESERVED_KEYS = {
+    "code", "section", "source_pdf", "source_pages", "source_basis",
+    "verified_level", "verification_pending", "source_type", "kgs_code",
+}
+
+
 # ───────────────────────── main process ────────────────────────────────
 
+
 async def process(document_id: int, session: AsyncSession) -> None:
     """문서 분류 + 요약 + tier triage.
 
@@ -334,6 +371,59 @@ async def process(document_id: int, session: AsyncSession) -> None:
     if not doc.extracted_text:
         raise ValueError(f"문서 ID {document_id}: extracted_text가 비어있음")
 
+    # ─── 옵션 C: markdown frontmatter 우선 인식 ───────────────────────────
+    # KGS Code 등 외부 작성 마크다운은 frontmatter 에 정확한 메타가 있다.
+    # title / tags / ai_summary / ai_domain 은 frontmatter 에 있으면 그대로 사용,
+    # 없는 필드만 LLM 호출. code/section/source_pages/verified_level 등 원문
+    # 추적 메타는 documents.md_frontmatter JSONB 에 보존하고 LLM 이 덮어쓰지 못하게 한다.
+    fm, body = _parse_frontmatter(doc.extracted_text)
+    if fm:
+        # frontmatter 전체를 md_frontmatter JSONB 에 저장 (원문 추적용 single source)
+        doc.md_frontmatter = fm
+
+        # 우선 반영 (LLM 보다 신뢰도 높음, frontmatter 가 authoritative)
+        if fm.get("title"):
+            doc.title = str(fm["title"])
+
+        fm_tags = fm.get("tags")
+        if isinstance(fm_tags, list) and fm_tags:
+            # ai_tags 에 frontmatter 태그 우선 적재 (LLM 이 추가만 가능)
+            doc.ai_tags = [str(t) for t in fm_tags]
+
+        if fm.get("ai_domain"):
+            doc.ai_domain = str(fm["ai_domain"])
+            parts = doc.ai_domain.split("/")
+            if len(parts) > 1 and not doc.ai_sub_group:
+                doc.ai_sub_group = parts[1]
+
+        if fm.get("ai_sub_group"):
+            doc.ai_sub_group = str(fm["ai_sub_group"])
+
+        if fm.get("document_type"):
+            doc.document_type = str(fm["document_type"])
+
+        if fm.get("ai_summary"):
+            doc.ai_summary = str(fm["ai_summary"])
+
+        if fm.get("importance") in ("high", "medium", "low"):
+            doc.importance = fm["importance"]
+
+        # 핵심 메타 (title + ai_domain + ai_summary) 가 모두 frontmatter 로 채워졌으면
+        # LLM classify/summarize 스킵. tier triage 도 스킵 (frontmatter 가 더 정확).
+        # frontmatter 미커버 필드는 그대로 두어 향후 필요 시 manual UI 채움.
+        if doc.title and doc.ai_domain and doc.ai_summary:
+            if not doc.ai_confidence:
+                doc.ai_confidence = 1.0  # frontmatter 는 사람이 작성한 단정값
+            doc.ai_processed_at = datetime.now(timezone.utc)
+            doc.ai_model_version = "frontmatter@manual"
+            await session.commit()
+            logger.info(f"doc {document_id}: frontmatter 옵션 C → classify/summarize/triage 전부 skip")
+            return
+
+        # 일부만 frontmatter 에 있을 때는 LLM 으로 미설정 필드 보완. 단 _FRONTMATTER_PRESERVED_KEYS
+        # 는 이미 md_frontmatter 에 있으므로 LLM 이 ai_domain/document_type 등에 영향 못 준다.
+        logger.info(f"doc {document_id}: frontmatter 부분 인식 → LLM 으로 미설정 필드 보완")
+
     client = AIClient()
     try:
         # ─── 1. Legacy classify (primary 26B) ───
@@ -344,17 +434,19 @@ async def process(document_id: int, session: AsyncSession) -> None:
         if not parsed:
             raise ValueError(f"AI 응답에서 JSON 추출 실패: {raw_response[:200]}")
 
-        # domain 검증
+        # domain 검증 (frontmatter 가 이미 채웠으면 LLM 결과 무시)
         domain = _validate_domain(parsed.get("domain", ""))
-        doc.ai_domain = domain
+        if not doc.ai_domain:
+            doc.ai_domain = domain
 
         # sub_group은 domain 경로에서 추출 (호환성)
         parts = domain.split("/")
         doc.ai_sub_group = parts[1] if len(parts) > 1 else ""
 
-        # document_type 검증
+        # document_type 검증 (frontmatter 가 이미 채웠으면 LLM 결과 무시)
         doc_type = parsed.get("document_type", "")
-        doc.document_type = doc_type if doc_type in DOCUMENT_TYPES else "Note"
+        if not doc.document_type:
+            doc.document_type = doc_type if doc_type in DOCUMENT_TYPES else "Note"
 
         # confidence
         confidence = parsed.get("confidence", 0.5)

app/workers/file_watcher.py

@@ -31,6 +31,9 @@ AUDIO_EXTS = {".mp3", ".m4a", ".opus", ".wav", ".flac", ".ogg"}
 VIDEO_DIRECT_EXTS = {".mp4", ".webm"}                # 브라우저 direct play
 VIDEO_QUARANTINE_EXTS = {".mov", ".mkv", ".avi"}     # 변환 필요, 보관만
 
+# library (외부 작성 학습 자료) 폴더 — md/pdf/docx 등 문서 확장자만 수락
+LIBRARY_DOC_EXTS = {".md", ".pdf", ".docx", ".doc", ".txt", ".rtf", ".html", ".odt"}
+
 # 스캔 대상: (하위경로, 예상 category) — None 은 문서함(카테고리 미지정)
 SCAN_TARGETS: list[tuple[str, str | None]] = [
     ("Inbox", None),
@@ -77,6 +80,15 @@ def _route_media(path: Path, expected_category: str | None) -> tuple[str | None,
             return ("video", True, None)
         return (None, False, None)  # 기타 → skip
 
+    if expected_category == "library":
+        # 외부 작성 학습 자료 (KGS Code, 시행규칙 등). 문서 확장자만 수락.
+        # frontmatter 해석은 classify_worker (옵션 C) 가 담당. file_watcher 는 라우팅만.
+        if ext in LIBRARY_DOC_EXTS:
+            return ("library", False, "extract")
+        if ext in AUDIO_EXTS or ext in VIDEO_DIRECT_EXTS or ext in VIDEO_QUARANTINE_EXTS:
+            return (None, False, None)  # audio/video 잘못 들어오면 skip
+        return (None, False, None)  # 기타 알 수 없는 확장자 skip
+
     # Inbox: 문서 파이프 (기존). audio/video 확장자가 실수로 여기 들어오면 skip.
     if ext in AUDIO_EXTS or ext in VIDEO_DIRECT_EXTS or ext in VIDEO_QUARANTINE_EXTS:
         return (None, False, None)
@@ -92,8 +104,14 @@ async def watch_inbox():
     new_count = 0
     changed_count = 0
 
+    # 동적 스캔 대상 합성: 기본 (Inbox/Recordings/Videos) + env 로 확장된 library 경로
+    # settings.additional_watch_targets 는 PKM 상대 경로 리스트 (예: "Knowledge/Industrial_Safety/가스기사/KGS_Code")
+    targets = list(SCAN_TARGETS)
+    for extra_path in settings.additional_watch_targets:
+        targets.append((extra_path, "library"))
+
     async with async_session() as session:
-        for sub, expected_category in SCAN_TARGETS:
+        for sub, expected_category in targets:
             scan_root = pkm_root / sub
             if not scan_root.exists():
                 continue