feat: 동적 문서 필드 매핑 기반 구축

.env.example

@@ -46,7 +46,19 @@ FOWOCO_INTENT_DEVICE=cpu
 FOWOCO_HF_TOKEN=hf_YOUR_HUGGINGFACE_TOKEN_HERE
 
 # ------------------------------------------------------------------------------
-# 5. CLOVA Template OCR — Stateless 내부 API
+# 5. Dynamic Document Automation — local Qwen3 field mapping (opt-in)
+# ------------------------------------------------------------------------------
+# false이면 model package/weights를 import 또는 load하지 않음
+FOWOCO_DYNAMIC_AUTOMATION_MAPPING_ENABLED=false
+
+# 미설정 시 FOWOCO_MODEL_CACHE_DIR 아래의 pinned revision 경로를 사용
+# FOWOCO_DYNAMIC_AUTOMATION_EMBEDDING_MODEL_PATH=/data/model-cache/qwen3-embedding-0.6b/97b0c614be4d77ee51c0cef4e5f07c00f9eb65b3
+# FOWOCO_DYNAMIC_AUTOMATION_RERANKER_MODEL_PATH=/data/model-cache/qwen3-reranker-0.6b/e61197ed45024b0ed8a2d74b80b4d909f1255473
+FOWOCO_DYNAMIC_AUTOMATION_MIN_RERANKER_SCORE=0.90
+FOWOCO_DYNAMIC_AUTOMATION_MIN_MARGIN=0.10
+
+# ------------------------------------------------------------------------------
+# 6. CLOVA Template OCR — Stateless 내부 API
 # ------------------------------------------------------------------------------
 # false이면 OCR API가 503을 반환하며 외부 HTTP client를 생성하지 않음
 FOWOCO_CLOVA_OCR_ENABLED=false

app/documents/dynamic_automation/__init__.py

@@ -0,0 +1,25 @@
+"""Contracts and catalog data for additive dynamic document automation."""
+
+from .catalog import CanonicalCatalog
+from .models import (
+    CanonicalFieldDefinition,
+    CanonicalMappingPlan,
+    CanonicalSource,
+    DocumentFieldContext,
+    FieldMapping,
+    MappingEvidence,
+    MappingStatus,
+    ScoredCandidate,
+)
+
+__all__ = [
+    "CanonicalCatalog",
+    "CanonicalFieldDefinition",
+    "CanonicalMappingPlan",
+    "CanonicalSource",
+    "DocumentFieldContext",
+    "FieldMapping",
+    "MappingEvidence",
+    "MappingStatus",
+    "ScoredCandidate",
+]

app/documents/dynamic_automation/catalog.py

@@ -0,0 +1,77 @@
+"""Versioned canonical field catalog with allowlisted identifiers."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator, Mapping
+from dataclasses import dataclass
+from dataclasses import field as dataclass_field
+from pathlib import Path
+from types import MappingProxyType
+
+import yaml
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+from .models import CanonicalFieldDefinition, DocumentFieldContext
+
+
+class _CatalogDocument(BaseModel):
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    version: str = Field(pattern=r"^v[1-9][0-9]*$")
+    fields: tuple[CanonicalFieldDefinition, ...]
+
+
+@dataclass(frozen=True)
+class CanonicalCatalog:
+    version: str
+    definitions: tuple[CanonicalFieldDefinition, ...]
+    _fields_by_id: Mapping[str, CanonicalFieldDefinition] = dataclass_field(
+        repr=False,
+        compare=False,
+    )
+
+    @classmethod
+    def load(cls, path: Path) -> CanonicalCatalog:
+        try:
+            raw = yaml.safe_load(path.read_text(encoding="utf-8"))
+        except (OSError, yaml.YAMLError) as error:
+            raise ValueError(f"invalid canonical catalog: {error}") from error
+
+        try:
+            document = _CatalogDocument.model_validate(raw)
+        except ValidationError as error:
+            if any("string_pattern_mismatch" == item["type"] for item in error.errors()):
+                raise ValueError("catalog contains an unapproved identifier") from error
+            raise ValueError(f"invalid canonical catalog: {error}") from error
+
+        fields_by_id: dict[str, CanonicalFieldDefinition] = {}
+        for field in document.fields:
+            if field.field_id in fields_by_id:
+                raise ValueError(f"duplicate canonical field identifier: {field.field_id}")
+            fields_by_id[field.field_id] = field
+
+        definitions = tuple(sorted(fields_by_id.values(), key=lambda item: item.field_id))
+        immutable_lookup = MappingProxyType(
+            {definition.field_id: definition for definition in definitions}
+        )
+        return cls(
+            version=document.version,
+            definitions=definitions,
+            _fields_by_id=immutable_lookup,
+        )
+
+    def __iter__(self) -> Iterator[CanonicalFieldDefinition]:
+        return iter(self.definitions)
+
+    def get(self, field_id: str) -> CanonicalFieldDefinition:
+        try:
+            return self._fields_by_id[field_id]
+        except KeyError as error:
+            raise KeyError(f"unknown canonical field: {field_id}") from error
+
+    def compatible(self, context: DocumentFieldContext) -> tuple[CanonicalFieldDefinition, ...]:
+        return tuple(
+            field
+            for field in self.definitions
+            if context.field_type in field.compatible_field_types
+        )

app/documents/dynamic_automation/config.py

@@ -0,0 +1,84 @@
+"""Package-owned settings for opt-in dynamic document mapping."""
+
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+from typing import Self
+
+from pydantic import Field, model_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from .qwen import (
+    QWEN3_EMBEDDING_CACHE_NAME,
+    QWEN3_EMBEDDING_REVISION,
+    QWEN3_RERANKER_CACHE_NAME,
+    QWEN3_RERANKER_REVISION,
+)
+
+_EMBEDDING_PATH = Path(QWEN3_EMBEDDING_CACHE_NAME) / QWEN3_EMBEDDING_REVISION
+_RERANKER_PATH = Path(QWEN3_RERANKER_CACHE_NAME) / QWEN3_RERANKER_REVISION
+
+
+class DynamicAutomationSettings(BaseSettings):
+    """Read dynamic-only settings without extending the FastAPI settings model."""
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        env_prefix="FOWOCO_",
+        extra="ignore",
+    )
+
+    model_cache_dir: Path = Field(
+        default_factory=lambda: Path(tempfile.gettempdir()) / "fowoco-model-cache"
+    )
+    dynamic_automation_mapping_enabled: bool = False
+    dynamic_automation_embedding_model_path: Path | None = None
+    dynamic_automation_reranker_model_path: Path | None = None
+    dynamic_automation_min_reranker_score: float = Field(default=0.90, ge=0, le=1)
+    dynamic_automation_min_margin: float = Field(default=0.10, ge=0, le=1)
+
+    @model_validator(mode="after")
+    def derive_model_paths(self) -> Self:
+        """Resolve and validate pinned model paths inside the managed cache."""
+        embedding_path = self.dynamic_automation_embedding_model_path or (
+            self.model_cache_dir / _EMBEDDING_PATH
+        )
+        reranker_path = self.dynamic_automation_reranker_model_path or (
+            self.model_cache_dir / _RERANKER_PATH
+        )
+        self.dynamic_automation_embedding_model_path = _managed_model_path(
+            embedding_path,
+            model_cache_dir=self.model_cache_dir,
+            pinned_suffix=_EMBEDDING_PATH,
+            setting_name="dynamic_automation_embedding_model_path",
+        )
+        self.dynamic_automation_reranker_model_path = _managed_model_path(
+            reranker_path,
+            model_cache_dir=self.model_cache_dir,
+            pinned_suffix=_RERANKER_PATH,
+            setting_name="dynamic_automation_reranker_model_path",
+        )
+        return self
+
+
+def _managed_model_path(
+    path: Path,
+    *,
+    model_cache_dir: Path,
+    pinned_suffix: Path,
+    setting_name: str,
+) -> Path:
+    resolved_cache = model_cache_dir.resolve(strict=False)
+    resolved_path = path.resolve(strict=False)
+    try:
+        relative_path = resolved_path.relative_to(resolved_cache)
+    except ValueError as err:
+        raise ValueError(f"{setting_name} must be below model_cache_dir") from err
+    if relative_path.parts[-2:] != pinned_suffix.parts:
+        raise ValueError(
+            f"{setting_name} must end in the pinned revision directory "
+            f"{pinned_suffix.as_posix()}"
+        )
+    return resolved_path