ADR-002: Tech Stack Evaluation für DocFlow¶

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any


class BlockType(StrEnum):
    HEADING = "heading"
    PARAGRAPH = "paragraph"
    TABLE = "table"
    LIST = "list"
    IMAGE_REF = "image_ref"


@dataclass(frozen=True)
class ContentBlock:
    block_type: BlockType
    content: str
    level: int | None = None
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class ExtractionResult:
    document_id: str
    raw_text: str
    structured_blocks: list[ContentBlock]
    metadata: dict[str, Any]
    extractor_used: str
    confidence: float  # 0.0 – 1.0


class ExtractorPort(ABC):
    """Outbound Port für Dokumenten-Extraktion.

    Jeder Adapter implementiert genau dieses Interface.
    Die Domain kennt keine konkreten Extraktoren.
    """

    @abstractmethod
    async def extract(self, file_data: bytes, mime_type: str) -> ExtractionResult:
        """Extrahiert Rohtext und Struktur aus einem Dokument."""
        ...

    @abstractmethod
    def supports(self, mime_type: str) -> bool:
        """Prüft ob dieser Extraktor den MIME-Type verarbeiten kann."""
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        """Eindeutiger Adapter-Name für Registry und Logging."""
        ...

import httpx

from docflow.domain.ports.extractor import ExtractorPort, ExtractionResult


class TikaExtractorAdapter(ExtractorPort):
    """Adapter: Apache Tika via HTTP API."""

    _SUPPORTED_TYPES = frozenset({
        "application/pdf",
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        "application/vnd.openxmlformats-officedocument.presentationml.presentation",
        "text/html",
        "text/plain",
    })

    def __init__(self, base_url: str = "http://tika:9998", timeout: float = 120.0) -> None:
        self._client = httpx.AsyncClient(base_url=base_url, timeout=timeout)

    async def extract(self, file_data: bytes, mime_type: str) -> ExtractionResult:
        response = await self._client.put(
            "/tika",
            content=file_data,
            headers={"Content-Type": mime_type, "Accept": "text/plain"},
        )
        response.raise_for_status()
        raw_text = response.text
        # ... parse structured_blocks from Tika metadata endpoint ...
        return ExtractionResult(
            document_id="",  # Set by caller
            raw_text=raw_text,
            structured_blocks=[],
            metadata={"source": "tika"},
            extractor_used=self.name,
            confidence=0.85,
        )

    def supports(self, mime_type: str) -> bool:
        return mime_type in self._SUPPORTED_TYPES

    @property
    def name(self) -> str:
        return "tika"

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field


@dataclass(frozen=True)
class OCRRegion:
    """Ein erkannter Textbereich mit Bounding Box."""
    text: str
    confidence: float
    bbox: tuple[int, int, int, int]  # x, y, width, height


@dataclass(frozen=True)
class OCRResult:
    full_text: str
    regions: list[OCRRegion]
    language_detected: str
    average_confidence: float
    engine_used: str
    metadata: dict[str, object] = field(default_factory=dict)


class OCRPort(ABC):
    """Outbound Port für Optical Character Recognition.

    Empfängt Bilddaten (PNG/JPEG/TIFF), liefert strukturierten Text.
    """

    @abstractmethod
    async def recognize(
        self,
        image_data: bytes,
        language: str = "deu",
        *,
        dpi: int = 300,
    ) -> OCRResult:
        """Führt OCR auf den übergebenen Bilddaten durch."""
        ...

    @abstractmethod
    def supported_languages(self) -> frozenset[str]:
        """Menge der unterstützten Sprachcodes (ISO 639-3)."""
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        """Adapter-Name für Registry und Konfiguration."""
        ...

import asyncio
from functools import partial
from pathlib import Path

from PIL import Image
import pytesseract

from docflow.domain.ports.ocr import OCRPort, OCRResult, OCRRegion


class TesseractOCRAdapter(OCRPort):
    """Adapter: Tesseract 5 via pytesseract."""

    def __init__(
        self,
        executable: str = "/usr/bin/tesseract",
        languages: list[str] | None = None,
    ) -> None:
        pytesseract.pytesseract.tesseract_cmd = executable
        self._languages = frozenset(languages or ["deu", "eng"])

    async def recognize(
        self,
        image_data: bytes,
        language: str = "deu",
        *,
        dpi: int = 300,
    ) -> OCRResult:
        # Blocking Tesseract-Call in Thread-Pool auslagern
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(
            None,
            partial(self._run_tesseract, image_data, language, dpi),
        )
        return result

    def _run_tesseract(self, image_data: bytes, language: str, dpi: int) -> OCRResult:
        import io
        image = Image.open(io.BytesIO(image_data))
        # TSV-Output für Bounding Boxes + Confidence
        data = pytesseract.image_to_data(
            image, lang=language, output_type=pytesseract.Output.DICT,
            config=f"--dpi {dpi}",
        )
        full_text = pytesseract.image_to_string(image, lang=language)
        regions = self._parse_regions(data)
        avg_conf = sum(r.confidence for r in regions) / max(len(regions), 1)
        return OCRResult(
            full_text=full_text.strip(),
            regions=regions,
            language_detected=language,
            average_confidence=avg_conf,
            engine_used=self.name,
        )

    def _parse_regions(self, data: dict) -> list[OCRRegion]:
        regions: list[OCRRegion] = []
        for i, text in enumerate(data["text"]):
            conf = float(data["conf"][i])
            if conf > 0 and text.strip():
                regions.append(OCRRegion(
                    text=text.strip(),
                    confidence=conf / 100.0,
                    bbox=(data["left"][i], data["top"][i],
                          data["width"][i], data["height"][i]),
                ))
        return regions

    def supported_languages(self) -> frozenset[str]:
        return self._languages

    @property
    def name(self) -> str:
        return "tesseract"

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any


@dataclass(frozen=True)
class ProcessingContext:
    """Kontext für den Processor: Sprache, gewünschte Struktur, etc."""
    language: str = "deu"
    source_mime_type: str = "application/pdf"
    extraction_confidence: float = 1.0
    options: dict[str, Any] = field(default_factory=dict)


class PostProcessorPort(ABC):
    """Outbound Port für Text-Nachbearbeitung.

    Wird in einer Pipeline sequenziell aufgerufen.
    Jeder Processor transformiert den Text und gibt ihn weiter.
    """

    @abstractmethod
    async def process(self, text: str, context: ProcessingContext) -> str:
        """Verarbeitet den Text. Gibt den transformierten Text zurück."""
        ...

    @property
    @abstractmethod
    def name(self) -> str:
        """Eindeutiger Name für Pipeline-Konfiguration."""
        ...


class LLMPort(ABC):
    """Spezialisierter Port für LLM-basierte Verarbeitung.

    Erweitert PostProcessorPort um LLM-spezifische Capabilities.
    Wird vom LLMPostProcessor-Adapter intern genutzt.
    """

    @abstractmethod
    async def complete(
        self,
        prompt: str,
        *,
        system_message: str | None = None,
        max_tokens: int = 4096,
        temperature: float = 0.1,
    ) -> str:
        """Sendet Prompt an LLM und gibt die Antwort zurück."""
        ...

    @abstractmethod
    async def is_available(self) -> bool:
        """Health-Check: Ist das LLM erreichbar?"""
        ...

    @property
    @abstractmethod
    def model_name(self) -> str:
        """Name des verwendeten Modells."""
        ...

from openai import AsyncOpenAI

from docflow.domain.ports.post_processor import LLMPort


class OpenAILLMAdapter(LLMPort):
    """Adapter: OpenAI API (GPT-4o, GPT-4o-mini)."""

    def __init__(
        self,
        api_key: str,
        model: str = "gpt-4o",
        base_url: str | None = None,
    ) -> None:
        self._client = AsyncOpenAI(api_key=api_key, base_url=base_url)
        self._model = model

    async def complete(
        self,
        prompt: str,
        *,
        system_message: str | None = None,
        max_tokens: int = 4096,
        temperature: float = 0.1,
    ) -> str:
        messages: list[dict[str, str]] = []
        if system_message:
            messages.append({"role": "system", "content": system_message})
        messages.append({"role": "user", "content": prompt})

        response = await self._client.chat.completions.create(
            model=self._model,
            messages=messages,
            max_tokens=max_tokens,
            temperature=temperature,
        )
        return response.choices[0].message.content or ""

    async def is_available(self) -> bool:
        try:
            await self._client.models.retrieve(self._model)
            return True
        except Exception:
            return False

    @property
    def model_name(self) -> str:
        return self._model

import httpx

from docflow.domain.ports.post_processor import LLMPort


class OllamaLLMAdapter(LLMPort):
    """Adapter: Ollama (lokale LLMs – llama3, mistral, etc.)."""

    def __init__(
        self,
        base_url: str = "http://ollama:11434",
        model: str = "llama3.1:8b",
    ) -> None:
        self._client = httpx.AsyncClient(base_url=base_url, timeout=120.0)
        self._model = model

    async def complete(
        self,
        prompt: str,
        *,
        system_message: str | None = None,
        max_tokens: int = 4096,
        temperature: float = 0.1,
    ) -> str:
        payload: dict = {
            "model": self._model,
            "prompt": prompt,
            "stream": False,
            "options": {"temperature": temperature, "num_predict": max_tokens},
        }
        if system_message:
            payload["system"] = system_message

        response = await self._client.post("/api/generate", json=payload)
        response.raise_for_status()
        return response.json()["response"]

    async def is_available(self) -> bool:
        try:
            resp = await self._client.get("/api/tags")
            return resp.status_code == 200
        except httpx.HTTPError:
            return False

    @property
    def model_name(self) -> str:
        return self._model

from docflow.domain.ports.post_processor import PostProcessorPort, ProcessingContext, LLMPort


class LLMPostProcessor(PostProcessorPort):
    """Nutzt ein LLM zur intelligenten Text-Strukturierung.

    Kapselt den LLMPort — der LLMPostProcessor ist selbst
    ein PostProcessorPort und fügt sich in die Pipeline ein.
    """

    SYSTEM_PROMPT = (
        "Du bist ein Dokumenten-Strukturierungs-Experte. "
        "Transformiere den folgenden extrahierten Rohtext in sauberes, "
        "strukturiertes Markdown. Behalte alle Informationen bei."
    )

    def __init__(self, llm: LLMPort) -> None:
        self._llm = llm

    async def process(self, text: str, context: ProcessingContext) -> str:
        if not await self._llm.is_available():
            # Graceful Degradation: Text unverändert durchreichen
            return text

        prompt = f"Sprache: {context.language}\n\nRohtext:\n{text}"
        return await self._llm.complete(
            prompt,
            system_message=self.SYSTEM_PROMPT,
            temperature=0.1,
        )

    @property
    def name(self) -> str:
        return "llm"

from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings


class TikaConfig(BaseSettings):
    """Konfiguration für den Tika-Adapter. Liest aus ENV + YAML."""
    url: str = Field(default="http://tika:9998", alias="TIKA_URL")
    timeout_seconds: float = Field(default=120.0, gt=0)

    model_config = {"env_prefix": "DOCFLOW_TIKA_"}


class ProcessDocumentRequest(BaseModel):
    """API Request Schema."""
    output_format: str = Field(default="markdown", pattern=r"^(markdown|json|plaintext)$")
    ocr_enabled: bool = True
    language: str = Field(default="deu", min_length=2, max_length=3)
    processors: list[str] = Field(default=["cleanup", "structure"])

# Phase 1: Simple async background processing
import asyncio
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field


@dataclass
class JobQueue:
    """Einfache in-process Job Queue für Phase 1."""
    _tasks: dict[str, asyncio.Task] = field(default_factory=dict)

    async def enqueue(
        self,
        job_id: str,
        coro: Awaitable,
    ) -> None:
        task = asyncio.create_task(coro)
        self._tasks[job_id] = task

    def get_status(self, job_id: str) -> str:
        if job_id not in self._tasks:
            return "not_found"
        task = self._tasks[job_id]
        if task.done():
            return "failed" if task.exception() else "completed"
        return "processing"

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class DocumentRecord(Base):
    """Persistence Model — NICHT das Domain-Entity!"""
    __tablename__ = "documents"

    id: Mapped[str] = mapped_column(primary_key=True)
    filename: Mapped[str]
    mime_type: Mapped[str]
    size_bytes: Mapped[int]
    status: Mapped[str]
    storage_reference: Mapped[str | None]
    created_at: Mapped[datetime]

from docflow.domain.ports.storage import StoragePort


class S3StorageAdapter(StoragePort):
    """Adapter: S3-kompatibler Object Storage (AWS S3, MinIO, etc.)."""

    def __init__(
        self,
        bucket: str,
        endpoint_url: str | None = None,  # MinIO: "http://minio:9000"
        region: str = "eu-central-1",
    ) -> None:
        import aiobotocore.session
        self._session = aiobotocore.session.get_session()
        self._bucket = bucket
        self._endpoint_url = endpoint_url
        self._region = region

    async def store(self, document_id: str, data: bytes, metadata: dict) -> str:
        key = f"documents/{document_id}"
        async with self._session.create_client(
            "s3",
            endpoint_url=self._endpoint_url,
            region_name=self._region,
        ) as client:
            await client.put_object(
                Bucket=self._bucket,
                Key=key,
                Body=data,
                Metadata={k: str(v) for k, v in metadata.items()},
            )
        return f"s3://{self._bucket}/{key}"

    async def retrieve(self, reference: str) -> bytes:
        _, _, bucket, key = reference.replace("s3://", "").split("/", 1)
        # ... retrieve implementation ...

    async def delete(self, reference: str) -> None:
        # ... delete implementation ...