Refactor backend to hexagonal architecture for converter extensibility
Extract domain value objects and ports from parsing.py, move Docling-specific code to infra/local_converter.py, and convert analysis_service to a class with injected DocumentConverter. This prepares the codebase for plugging in alternative conversion backends (e.g. Docling Serve) via the Protocol pattern. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Pier-Jean Malandrino
parent
c9de4bbd8a
commit
3743ed4ca8
11 changed files with 579 additions and 450 deletions
|
|
@ -7,12 +7,17 @@ import logging
|
|||
from fastapi import APIRouter, HTTPException
|
||||
|
||||
from api.schemas import AnalysisResponse, CreateAnalysisRequest
|
||||
from services import analysis_service
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api/analyses", tags=["analyses"])
|
||||
|
||||
|
||||
def _get_service():
|
||||
"""Lazy import to avoid circular dependency at module level."""
|
||||
from main import analysis_service
|
||||
return analysis_service
|
||||
|
||||
|
||||
def _to_response(job) -> AnalysisResponse:
|
||||
return AnalysisResponse(
|
||||
id=job.id,
|
||||
|
|
@ -40,7 +45,7 @@ async def create_analysis(body: CreateAnalysisRequest):
|
|||
pipeline_opts = body.pipelineOptions.model_dump()
|
||||
|
||||
try:
|
||||
job = await analysis_service.create(body.documentId, pipeline_options=pipeline_opts)
|
||||
job = await _get_service().create(body.documentId, pipeline_options=pipeline_opts)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=404, detail=str(e)) from e
|
||||
|
||||
|
|
@ -50,14 +55,14 @@ async def create_analysis(body: CreateAnalysisRequest):
|
|||
@router.get("", response_model=list[AnalysisResponse])
|
||||
async def list_analyses():
|
||||
"""List all analysis jobs."""
|
||||
jobs = await analysis_service.find_all()
|
||||
jobs = await _get_service().find_all()
|
||||
return [_to_response(j) for j in jobs]
|
||||
|
||||
|
||||
@router.get("/{job_id}", response_model=AnalysisResponse)
|
||||
async def get_analysis(job_id: str):
|
||||
"""Get a single analysis job."""
|
||||
job = await analysis_service.find_by_id(job_id)
|
||||
job = await _get_service().find_by_id(job_id)
|
||||
if not job:
|
||||
raise HTTPException(status_code=404, detail="Analysis not found")
|
||||
return _to_response(job)
|
||||
|
|
@ -66,6 +71,6 @@ async def get_analysis(job_id: str):
|
|||
@router.delete("/{job_id}", status_code=204)
|
||||
async def delete_analysis(job_id: str):
|
||||
"""Delete an analysis job."""
|
||||
deleted = await analysis_service.delete(job_id)
|
||||
deleted = await _get_service().delete(job_id)
|
||||
if not deleted:
|
||||
raise HTTPException(status_code=404, detail="Analysis not found")
|
||||
|
|
|
|||
|
|
@ -1,298 +1,34 @@
|
|||
"""Docling document extraction logic — pure domain, no HTTP concerns.
|
||||
"""Backward-compatible re-exports for domain.parsing.
|
||||
|
||||
Wraps the Docling library to convert documents and extract structured
|
||||
per-page elements with bounding boxes and hierarchy levels.
|
||||
After the hexagonal architecture refactoring:
|
||||
- Value objects moved to domain.value_objects
|
||||
- Docling implementation moved to infra.local_converter
|
||||
|
||||
This module re-exports the public names so existing code and tests
|
||||
that import from domain.parsing continue to work.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import logging
|
||||
import threading
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from docling.datamodel.base_models import InputFormat
|
||||
from docling.datamodel.pipeline_options import (
|
||||
PdfPipelineOptions,
|
||||
TableFormerMode,
|
||||
TableStructureOptions,
|
||||
from domain.value_objects import ( # noqa: F401
|
||||
ConversionOptions,
|
||||
ConversionResult,
|
||||
PageDetail,
|
||||
PageElement,
|
||||
)
|
||||
from docling.document_converter import DocumentConverter, PdfFormatOption
|
||||
from docling_core.types.doc import (
|
||||
CodeItem,
|
||||
DocItem,
|
||||
FloatingItem,
|
||||
FormulaItem,
|
||||
GroupItem,
|
||||
ListItem,
|
||||
PictureItem,
|
||||
SectionHeaderItem,
|
||||
TableItem,
|
||||
TextItem,
|
||||
TitleItem,
|
||||
from infra.local_converter import (
|
||||
_build_docling_converter,
|
||||
_convert_sync,
|
||||
_extract_pages_detail as extract_pages_detail, # noqa: F401
|
||||
_get_default_converter as get_default_converter, # noqa: F401
|
||||
)
|
||||
|
||||
from domain.bbox import to_topleft_list
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Thread lock — DocumentConverter is not thread-safe
|
||||
_converter_lock = threading.Lock()
|
||||
|
||||
# US Letter page dimensions (points) — fallback when page size is unknown
|
||||
_DEFAULT_PAGE_WIDTH = 612.0
|
||||
_DEFAULT_PAGE_HEIGHT = 792.0
|
||||
|
||||
# Default converter (lazy-init on first request)
|
||||
_default_converter: DocumentConverter | None = None
|
||||
def build_converter(options: ConversionOptions | None = None):
|
||||
"""Build a Docling DocumentConverter (backward-compatible signature)."""
|
||||
return _build_docling_converter(options or ConversionOptions())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Domain value objects
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class PageElement:
|
||||
type: str
|
||||
bbox: list[float]
|
||||
content: str
|
||||
level: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class PageDetail:
|
||||
page_number: int
|
||||
width: float
|
||||
height: float
|
||||
elements: list[PageElement] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionOptions:
|
||||
do_ocr: bool = True
|
||||
do_table_structure: bool = True
|
||||
table_mode: str = "accurate"
|
||||
do_code_enrichment: bool = False
|
||||
do_formula_enrichment: bool = False
|
||||
do_picture_classification: bool = False
|
||||
do_picture_description: bool = False
|
||||
generate_picture_images: bool = False
|
||||
generate_page_images: bool = False
|
||||
images_scale: float = 1.0
|
||||
|
||||
def is_default(self) -> bool:
|
||||
return self == ConversionOptions()
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionResult:
|
||||
page_count: int
|
||||
content_markdown: str
|
||||
content_html: str
|
||||
pages: list[PageDetail]
|
||||
skipped_items: int = 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Element type detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Mapping from Docling type to element type string.
|
||||
# Order matters: most specific types before their parents.
|
||||
_ELEMENT_TYPE_MAP: list[tuple[type, str]] = [
|
||||
(TableItem, "table"),
|
||||
(PictureItem, "picture"),
|
||||
(TitleItem, "title"),
|
||||
(SectionHeaderItem, "section_header"),
|
||||
(ListItem, "list"),
|
||||
(FormulaItem, "formula"),
|
||||
(CodeItem, "code"),
|
||||
(FloatingItem, "floating"),
|
||||
(TextItem, "text"),
|
||||
]
|
||||
|
||||
|
||||
def _get_element_type(item: DocItem) -> str:
|
||||
"""Determine element type via isinstance on Docling's type hierarchy."""
|
||||
for cls, type_name in _ELEMENT_TYPE_MAP:
|
||||
if isinstance(item, cls):
|
||||
return type_name
|
||||
return "text"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pipeline factory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def build_converter(options: ConversionOptions | None = None) -> DocumentConverter:
|
||||
"""Build a DocumentConverter with the given pipeline options."""
|
||||
opts = options or ConversionOptions()
|
||||
|
||||
table_options = TableStructureOptions(
|
||||
do_cell_matching=True,
|
||||
mode=TableFormerMode.ACCURATE if opts.table_mode == "accurate" else TableFormerMode.FAST,
|
||||
)
|
||||
|
||||
pipeline_options = PdfPipelineOptions(
|
||||
do_ocr=opts.do_ocr,
|
||||
do_table_structure=opts.do_table_structure,
|
||||
table_structure_options=table_options,
|
||||
do_code_enrichment=opts.do_code_enrichment,
|
||||
do_formula_enrichment=opts.do_formula_enrichment,
|
||||
do_picture_classification=opts.do_picture_classification,
|
||||
do_picture_description=opts.do_picture_description,
|
||||
generate_page_images=opts.generate_page_images,
|
||||
generate_picture_images=opts.generate_picture_images,
|
||||
images_scale=opts.images_scale,
|
||||
)
|
||||
|
||||
return DocumentConverter(
|
||||
format_options={
|
||||
InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def get_default_converter() -> DocumentConverter:
|
||||
global _default_converter
|
||||
if _default_converter is None:
|
||||
_default_converter = build_converter()
|
||||
return _default_converter
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Page extraction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def extract_pages_detail(doc_result) -> tuple[list[PageDetail], int]:
|
||||
"""Extract per-page element details with bounding boxes from Docling result.
|
||||
|
||||
Returns (pages, skipped_count) for transparent error reporting.
|
||||
"""
|
||||
pages: dict[int, PageDetail] = {}
|
||||
document = doc_result.document
|
||||
skipped = 0
|
||||
|
||||
for page_key, page_obj in document.pages.items():
|
||||
page_no = int(page_key) if isinstance(page_key, str) else page_key
|
||||
pages[page_no] = PageDetail(
|
||||
page_number=page_no,
|
||||
width=page_obj.size.width,
|
||||
height=page_obj.size.height,
|
||||
)
|
||||
|
||||
for item, level in document.iterate_items():
|
||||
ok = _process_content_item(item, level, pages)
|
||||
if not ok:
|
||||
skipped += 1
|
||||
|
||||
sorted_pages = sorted(pages.values(), key=lambda p: p.page_number)
|
||||
return sorted_pages, skipped
|
||||
|
||||
|
||||
def _process_content_item(
|
||||
item: DocItem | GroupItem, level: int, pages: dict[int, PageDetail],
|
||||
) -> bool:
|
||||
"""Process a single content item and add it to the appropriate page."""
|
||||
if isinstance(item, GroupItem):
|
||||
return True
|
||||
|
||||
if not isinstance(item, DocItem) or not item.prov:
|
||||
return False
|
||||
|
||||
for prov in item.prov:
|
||||
try:
|
||||
page_no = prov.page_no
|
||||
if page_no not in pages:
|
||||
# Fallback: page was not found in document.pages (corrupted PDF or
|
||||
# Docling edge case). US Letter dimensions are used as a safe default.
|
||||
# This may cause slight bbox misalignment on non-Letter pages (e.g. A4).
|
||||
logger.warning(
|
||||
"Page %d not found in document metadata — using US Letter fallback (%sx%s pt)",
|
||||
page_no, _DEFAULT_PAGE_WIDTH, _DEFAULT_PAGE_HEIGHT,
|
||||
)
|
||||
pages[page_no] = PageDetail(page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_DEFAULT_PAGE_HEIGHT)
|
||||
|
||||
page_height = pages[page_no].height
|
||||
|
||||
bbox = [0.0, 0.0, 0.0, 0.0]
|
||||
if prov.bbox:
|
||||
bbox = to_topleft_list(prov.bbox, page_height)
|
||||
|
||||
element_type = _get_element_type(item)
|
||||
|
||||
content = getattr(item, "text", "") or ""
|
||||
if isinstance(item, TableItem):
|
||||
with contextlib.suppress(AttributeError, ValueError):
|
||||
content = item.export_to_markdown()
|
||||
|
||||
pages[page_no].elements.append(
|
||||
PageElement(type=element_type, bbox=bbox, content=content, level=level)
|
||||
)
|
||||
except (AttributeError, KeyError, TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Skipping item %s on page %s",
|
||||
type(item).__name__,
|
||||
getattr(prov, "page_no", "?"),
|
||||
exc_info=True,
|
||||
)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main conversion entry point
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _select_converter(options: ConversionOptions) -> DocumentConverter:
|
||||
"""Return the cached default converter or build a custom one."""
|
||||
if options.is_default():
|
||||
return get_default_converter()
|
||||
return build_converter(options)
|
||||
|
||||
|
||||
def _build_fallback_pages(doc, page_count: int) -> list[PageDetail]:
|
||||
"""Create empty PageDetail entries when extraction yields nothing."""
|
||||
return [
|
||||
PageDetail(
|
||||
page_number=i + 1,
|
||||
width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else _DEFAULT_PAGE_WIDTH,
|
||||
height=doc.pages[i + 1].size.height if (i + 1) in doc.pages else _DEFAULT_PAGE_HEIGHT,
|
||||
)
|
||||
for i in range(page_count)
|
||||
]
|
||||
|
||||
|
||||
def convert_document(
|
||||
file_path: str,
|
||||
options: ConversionOptions | None = None,
|
||||
) -> ConversionResult:
|
||||
"""Convert a document and return structured results.
|
||||
|
||||
This is the main entry point for document parsing. Runs synchronously
|
||||
(caller should use asyncio.to_thread for non-blocking execution).
|
||||
"""
|
||||
opts = options or ConversionOptions()
|
||||
|
||||
with _converter_lock:
|
||||
conv = _select_converter(opts)
|
||||
result = conv.convert(file_path)
|
||||
|
||||
doc = result.document
|
||||
page_count = len(doc.pages)
|
||||
pages_detail, skipped = extract_pages_detail(result)
|
||||
|
||||
if not pages_detail and page_count > 0:
|
||||
pages_detail = _build_fallback_pages(doc, page_count)
|
||||
|
||||
if skipped > 0:
|
||||
logger.info("Parsed: %d pages, %d items skipped", page_count, skipped)
|
||||
|
||||
return ConversionResult(
|
||||
page_count=page_count or len(pages_detail) or 1,
|
||||
content_markdown=doc.export_to_markdown(),
|
||||
content_html=doc.export_to_html(),
|
||||
pages=pages_detail,
|
||||
skipped_items=skipped,
|
||||
)
|
||||
def convert_document(file_path: str, options: ConversionOptions | None = None) -> ConversionResult:
|
||||
"""Convert a document synchronously (backward-compatible signature)."""
|
||||
return _convert_sync(file_path, options or ConversionOptions())
|
||||
|
|
|
|||
23
document-parser/domain/ports.py
Normal file
23
document-parser/domain/ports.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
"""Domain ports — abstract interfaces that infrastructure must implement.
|
||||
|
||||
These protocols define what the domain NEEDS, not how it's done.
|
||||
Infrastructure adapters (local Docling, Docling Serve, etc.) implement these.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Protocol
|
||||
|
||||
from domain.value_objects import ConversionOptions, ConversionResult
|
||||
|
||||
|
||||
class DocumentConverter(Protocol):
|
||||
"""Port for document conversion.
|
||||
|
||||
Any implementation (local Docling lib, remote Docling Serve, mock, etc.)
|
||||
must satisfy this contract.
|
||||
"""
|
||||
|
||||
async def convert(
|
||||
self, file_path: str, options: ConversionOptions,
|
||||
) -> ConversionResult: ...
|
||||
51
document-parser/domain/value_objects.py
Normal file
51
document-parser/domain/value_objects.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
"""Domain value objects — pure data structures for document conversion.
|
||||
|
||||
These types define the contract between the domain and infrastructure layers.
|
||||
They have ZERO external dependencies (no docling, no HTTP, no DB).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass
|
||||
class PageElement:
|
||||
type: str
|
||||
bbox: list[float]
|
||||
content: str
|
||||
level: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class PageDetail:
|
||||
page_number: int
|
||||
width: float
|
||||
height: float
|
||||
elements: list[PageElement] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionOptions:
|
||||
do_ocr: bool = True
|
||||
do_table_structure: bool = True
|
||||
table_mode: str = "accurate"
|
||||
do_code_enrichment: bool = False
|
||||
do_formula_enrichment: bool = False
|
||||
do_picture_classification: bool = False
|
||||
do_picture_description: bool = False
|
||||
generate_picture_images: bool = False
|
||||
generate_page_images: bool = False
|
||||
images_scale: float = 1.0
|
||||
|
||||
def is_default(self) -> bool:
|
||||
return self == ConversionOptions()
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionResult:
|
||||
page_count: int
|
||||
content_markdown: str
|
||||
content_html: str
|
||||
pages: list[PageDetail]
|
||||
skipped_items: int = 0
|
||||
0
document-parser/infra/__init__.py
Normal file
0
document-parser/infra/__init__.py
Normal file
243
document-parser/infra/local_converter.py
Normal file
243
document-parser/infra/local_converter.py
Normal file
|
|
@ -0,0 +1,243 @@
|
|||
"""Local Docling converter — runs Docling as a Python library in-process.
|
||||
|
||||
This adapter implements the DocumentConverter port using the Docling library
|
||||
directly. It wraps the blocking DocumentConverter in asyncio.to_thread for
|
||||
non-blocking execution.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import logging
|
||||
import threading
|
||||
|
||||
from docling.datamodel.base_models import InputFormat
|
||||
from docling.datamodel.pipeline_options import (
|
||||
PdfPipelineOptions,
|
||||
TableFormerMode,
|
||||
TableStructureOptions,
|
||||
)
|
||||
from docling.document_converter import DocumentConverter as DoclingConverter
|
||||
from docling.document_converter import PdfFormatOption
|
||||
from docling_core.types.doc import (
|
||||
CodeItem,
|
||||
DocItem,
|
||||
FloatingItem,
|
||||
FormulaItem,
|
||||
GroupItem,
|
||||
ListItem,
|
||||
PictureItem,
|
||||
SectionHeaderItem,
|
||||
TableItem,
|
||||
TextItem,
|
||||
TitleItem,
|
||||
)
|
||||
|
||||
from domain.bbox import to_topleft_list
|
||||
from domain.value_objects import (
|
||||
ConversionOptions,
|
||||
ConversionResult,
|
||||
PageDetail,
|
||||
PageElement,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Thread lock — DoclingConverter is not thread-safe
|
||||
_converter_lock = threading.Lock()
|
||||
|
||||
# US Letter page dimensions (points) — fallback when page size is unknown
|
||||
_DEFAULT_PAGE_WIDTH = 612.0
|
||||
_DEFAULT_PAGE_HEIGHT = 792.0
|
||||
|
||||
# Default converter (lazy-init on first request)
|
||||
_default_converter: DoclingConverter | None = None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Element type detection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_ELEMENT_TYPE_MAP: list[tuple[type, str]] = [
|
||||
(TableItem, "table"),
|
||||
(PictureItem, "picture"),
|
||||
(TitleItem, "title"),
|
||||
(SectionHeaderItem, "section_header"),
|
||||
(ListItem, "list"),
|
||||
(FormulaItem, "formula"),
|
||||
(CodeItem, "code"),
|
||||
(FloatingItem, "floating"),
|
||||
(TextItem, "text"),
|
||||
]
|
||||
|
||||
|
||||
def _get_element_type(item: DocItem) -> str:
|
||||
for cls, type_name in _ELEMENT_TYPE_MAP:
|
||||
if isinstance(item, cls):
|
||||
return type_name
|
||||
return "text"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pipeline factory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _build_docling_converter(options: ConversionOptions) -> DoclingConverter:
|
||||
table_options = TableStructureOptions(
|
||||
do_cell_matching=True,
|
||||
mode=TableFormerMode.ACCURATE if options.table_mode == "accurate" else TableFormerMode.FAST,
|
||||
)
|
||||
|
||||
pipeline_options = PdfPipelineOptions(
|
||||
do_ocr=options.do_ocr,
|
||||
do_table_structure=options.do_table_structure,
|
||||
table_structure_options=table_options,
|
||||
do_code_enrichment=options.do_code_enrichment,
|
||||
do_formula_enrichment=options.do_formula_enrichment,
|
||||
do_picture_classification=options.do_picture_classification,
|
||||
do_picture_description=options.do_picture_description,
|
||||
generate_page_images=options.generate_page_images,
|
||||
generate_picture_images=options.generate_picture_images,
|
||||
images_scale=options.images_scale,
|
||||
)
|
||||
|
||||
return DoclingConverter(
|
||||
format_options={
|
||||
InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _get_default_converter() -> DoclingConverter:
|
||||
global _default_converter
|
||||
if _default_converter is None:
|
||||
_default_converter = _build_docling_converter(ConversionOptions())
|
||||
return _default_converter
|
||||
|
||||
|
||||
def _select_converter(options: ConversionOptions) -> DoclingConverter:
|
||||
if options.is_default():
|
||||
return _get_default_converter()
|
||||
return _build_docling_converter(options)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Page extraction
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _extract_pages_detail(doc_result) -> tuple[list[PageDetail], int]:
|
||||
pages: dict[int, PageDetail] = {}
|
||||
document = doc_result.document
|
||||
skipped = 0
|
||||
|
||||
for page_key, page_obj in document.pages.items():
|
||||
page_no = int(page_key) if isinstance(page_key, str) else page_key
|
||||
pages[page_no] = PageDetail(
|
||||
page_number=page_no,
|
||||
width=page_obj.size.width,
|
||||
height=page_obj.size.height,
|
||||
)
|
||||
|
||||
for item, level in document.iterate_items():
|
||||
ok = _process_content_item(item, level, pages)
|
||||
if not ok:
|
||||
skipped += 1
|
||||
|
||||
sorted_pages = sorted(pages.values(), key=lambda p: p.page_number)
|
||||
return sorted_pages, skipped
|
||||
|
||||
|
||||
def _process_content_item(
|
||||
item: DocItem | GroupItem, level: int, pages: dict[int, PageDetail],
|
||||
) -> bool:
|
||||
if isinstance(item, GroupItem):
|
||||
return True
|
||||
|
||||
if not isinstance(item, DocItem) or not item.prov:
|
||||
return False
|
||||
|
||||
for prov in item.prov:
|
||||
try:
|
||||
page_no = prov.page_no
|
||||
if page_no not in pages:
|
||||
logger.warning(
|
||||
"Page %d not found in document metadata — using US Letter fallback (%sx%s pt)",
|
||||
page_no, _DEFAULT_PAGE_WIDTH, _DEFAULT_PAGE_HEIGHT,
|
||||
)
|
||||
pages[page_no] = PageDetail(page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_DEFAULT_PAGE_HEIGHT)
|
||||
|
||||
page_height = pages[page_no].height
|
||||
|
||||
bbox = [0.0, 0.0, 0.0, 0.0]
|
||||
if prov.bbox:
|
||||
bbox = to_topleft_list(prov.bbox, page_height)
|
||||
|
||||
element_type = _get_element_type(item)
|
||||
|
||||
content = getattr(item, "text", "") or ""
|
||||
if isinstance(item, TableItem):
|
||||
with contextlib.suppress(AttributeError, ValueError):
|
||||
content = item.export_to_markdown()
|
||||
|
||||
pages[page_no].elements.append(
|
||||
PageElement(type=element_type, bbox=bbox, content=content, level=level)
|
||||
)
|
||||
except (AttributeError, KeyError, TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Skipping item %s on page %s",
|
||||
type(item).__name__,
|
||||
getattr(prov, "page_no", "?"),
|
||||
exc_info=True,
|
||||
)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Synchronous conversion (called via asyncio.to_thread)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult:
|
||||
with _converter_lock:
|
||||
conv = _select_converter(options)
|
||||
result = conv.convert(file_path)
|
||||
|
||||
doc = result.document
|
||||
page_count = len(doc.pages)
|
||||
pages_detail, skipped = _extract_pages_detail(result)
|
||||
|
||||
if not pages_detail and page_count > 0:
|
||||
pages_detail = [
|
||||
PageDetail(
|
||||
page_number=i + 1,
|
||||
width=doc.pages[i + 1].size.width if (i + 1) in doc.pages else _DEFAULT_PAGE_WIDTH,
|
||||
height=doc.pages[i + 1].size.height if (i + 1) in doc.pages else _DEFAULT_PAGE_HEIGHT,
|
||||
)
|
||||
for i in range(page_count)
|
||||
]
|
||||
|
||||
if skipped > 0:
|
||||
logger.info("Parsed: %d pages, %d items skipped", page_count, skipped)
|
||||
|
||||
return ConversionResult(
|
||||
page_count=page_count or len(pages_detail) or 1,
|
||||
content_markdown=doc.export_to_markdown(),
|
||||
content_html=doc.export_to_html(),
|
||||
pages=pages_detail,
|
||||
skipped_items=skipped,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public adapter class
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class LocalConverter:
|
||||
"""Adapter that runs Docling locally as a Python library."""
|
||||
|
||||
async def convert(
|
||||
self, file_path: str, options: ConversionOptions,
|
||||
) -> ConversionResult:
|
||||
return await asyncio.to_thread(_convert_sync, file_path, options)
|
||||
30
document-parser/infra/settings.py
Normal file
30
document-parser/infra/settings.py
Normal file
|
|
@ -0,0 +1,30 @@
|
|||
"""Centralized application settings — loaded from environment variables."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Settings:
|
||||
conversion_engine: str = "local" # "local" or "remote"
|
||||
docling_serve_url: str = "http://localhost:5001"
|
||||
docling_serve_api_key: str | None = None
|
||||
conversion_timeout: int = 600
|
||||
upload_dir: str = "./uploads"
|
||||
db_path: str = "./data/docling_studio.db"
|
||||
cors_origins: list[str] = field(default_factory=lambda: ["http://localhost:3000", "http://localhost:5173"])
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> Settings:
|
||||
cors_raw = os.environ.get("CORS_ORIGINS", "http://localhost:3000,http://localhost:5173")
|
||||
return cls(
|
||||
conversion_engine=os.environ.get("CONVERSION_ENGINE", "local"),
|
||||
docling_serve_url=os.environ.get("DOCLING_SERVE_URL", "http://localhost:5001"),
|
||||
docling_serve_api_key=os.environ.get("DOCLING_SERVE_API_KEY"),
|
||||
conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "600")),
|
||||
upload_dir=os.environ.get("UPLOAD_DIR", "./uploads"),
|
||||
db_path=os.environ.get("DB_PATH", "./data/docling_studio.db"),
|
||||
cors_origins=[o.strip() for o in cors_raw.split(",")],
|
||||
)
|
||||
|
|
@ -1,14 +1,17 @@
|
|||
"""Docling Studio — unified FastAPI backend.
|
||||
|
||||
Single service replacing both the Spring Boot backend and the document parser.
|
||||
Provides document management (upload, CRUD), analysis orchestration (async Docling
|
||||
processing), and PDF preview — all backed by SQLite.
|
||||
Single service providing document management (upload, CRUD), analysis
|
||||
orchestration (async Docling processing), and PDF preview — all backed
|
||||
by SQLite.
|
||||
|
||||
Conversion engine is selected via CONVERSION_ENGINE env var:
|
||||
- "local" → Docling runs in-process as a Python library (default)
|
||||
- "remote" → delegates to a Docling Serve instance via HTTP
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
|
@ -16,6 +19,7 @@ from fastapi.middleware.cors import CORSMiddleware
|
|||
|
||||
from api.analyses import router as analyses_router
|
||||
from api.documents import router as documents_router
|
||||
from infra.settings import Settings
|
||||
from persistence.database import init_db
|
||||
|
||||
logging.basicConfig(
|
||||
|
|
@ -24,12 +28,49 @@ logging.basicConfig(
|
|||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Settings & dependency wiring
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
settings = Settings.from_env()
|
||||
|
||||
|
||||
def _build_converter():
|
||||
"""Build the converter adapter based on configuration."""
|
||||
if settings.conversion_engine == "remote":
|
||||
from infra.serve_converter import ServeConverter
|
||||
logger.info("Using remote Docling Serve at %s", settings.docling_serve_url)
|
||||
return ServeConverter(
|
||||
base_url=settings.docling_serve_url,
|
||||
api_key=settings.docling_serve_api_key,
|
||||
)
|
||||
else:
|
||||
from infra.local_converter import LocalConverter
|
||||
logger.info("Using local Docling converter")
|
||||
return LocalConverter()
|
||||
|
||||
|
||||
def _build_analysis_service():
|
||||
from services.analysis_service import AnalysisService
|
||||
converter = _build_converter()
|
||||
return AnalysisService(
|
||||
converter=converter,
|
||||
conversion_timeout=settings.conversion_timeout,
|
||||
)
|
||||
|
||||
|
||||
# Singleton service instance — imported by API routers
|
||||
analysis_service = _build_analysis_service()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# FastAPI app
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
"""Startup: initialize database. Shutdown: nothing special needed."""
|
||||
await init_db()
|
||||
logger.info("Docling Studio backend ready")
|
||||
logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine)
|
||||
yield
|
||||
|
||||
|
||||
|
|
@ -39,20 +80,14 @@ app = FastAPI(
|
|||
lifespan=lifespan,
|
||||
)
|
||||
|
||||
# CORS — configurable via env, defaults for local dev
|
||||
allowed_origins = os.environ.get(
|
||||
"CORS_ORIGINS", "http://localhost:3000,http://localhost:5173"
|
||||
).split(",")
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=[o.strip() for o in allowed_origins],
|
||||
allow_origins=settings.cors_origins,
|
||||
allow_credentials=True,
|
||||
allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
|
||||
allow_headers=["Content-Type", "Authorization"],
|
||||
)
|
||||
|
||||
# Mount routers
|
||||
app.include_router(documents_router)
|
||||
app.include_router(analyses_router)
|
||||
|
||||
|
|
@ -60,4 +95,4 @@ app.include_router(analyses_router)
|
|||
@app.get("/health")
|
||||
def health():
|
||||
"""Health check endpoint."""
|
||||
return {"status": "ok"}
|
||||
return {"status": "ok", "engine": settings.conversion_engine}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,8 @@
|
|||
"""Analysis service — async document parsing orchestration."""
|
||||
"""Analysis service — async document parsing orchestration.
|
||||
|
||||
Uses an injected DocumentConverter (port) so the service is decoupled
|
||||
from the conversion implementation (local Docling lib vs remote Docling Serve).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -8,32 +12,88 @@ import logging
|
|||
from dataclasses import asdict
|
||||
|
||||
from domain.models import AnalysisJob
|
||||
from domain.parsing import ConversionOptions, ConversionResult, convert_document
|
||||
from domain.ports import DocumentConverter
|
||||
from domain.value_objects import ConversionOptions, ConversionResult
|
||||
from persistence import analysis_repo, document_repo
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Maximum time (seconds) allowed for a single document conversion.
|
||||
CONVERSION_TIMEOUT = int(__import__("os").environ.get("CONVERSION_TIMEOUT", "600"))
|
||||
|
||||
class AnalysisService:
|
||||
"""Orchestrates document analysis using an injected converter."""
|
||||
|
||||
async def create(document_id: str, *, pipeline_options: dict | None = None) -> AnalysisJob:
|
||||
"""Create a new analysis job and launch background processing."""
|
||||
doc = await document_repo.find_by_id(document_id)
|
||||
if not doc:
|
||||
raise ValueError(f"Document not found: {document_id}")
|
||||
def __init__(self, converter: DocumentConverter, conversion_timeout: int = 600):
|
||||
self._converter = converter
|
||||
self._conversion_timeout = conversion_timeout
|
||||
|
||||
job = AnalysisJob(document_id=document_id)
|
||||
job.document_filename = doc.filename
|
||||
await analysis_repo.insert(job)
|
||||
async def create(self, document_id: str, *, pipeline_options: dict | None = None) -> AnalysisJob:
|
||||
"""Create a new analysis job and launch background processing."""
|
||||
doc = await document_repo.find_by_id(document_id)
|
||||
if not doc:
|
||||
raise ValueError(f"Document not found: {document_id}")
|
||||
|
||||
# Fire background task with error logging callback
|
||||
task = asyncio.create_task(
|
||||
_run_analysis(job.id, doc.storage_path, doc.filename, pipeline_options)
|
||||
)
|
||||
task.add_done_callback(_on_task_done)
|
||||
job = AnalysisJob(document_id=document_id)
|
||||
job.document_filename = doc.filename
|
||||
await analysis_repo.insert(job)
|
||||
|
||||
return job
|
||||
task = asyncio.create_task(
|
||||
self._run_analysis(job.id, doc.storage_path, doc.filename, pipeline_options)
|
||||
)
|
||||
task.add_done_callback(_on_task_done)
|
||||
|
||||
return job
|
||||
|
||||
async def find_all(self) -> list[AnalysisJob]:
|
||||
return await analysis_repo.find_all()
|
||||
|
||||
async def find_by_id(self, job_id: str) -> AnalysisJob | None:
|
||||
return await analysis_repo.find_by_id(job_id)
|
||||
|
||||
async def delete(self, job_id: str) -> bool:
|
||||
return await analysis_repo.delete(job_id)
|
||||
|
||||
async def _run_analysis(
|
||||
self, job_id: str, file_path: str, filename: str, pipeline_options: dict | None = None,
|
||||
) -> None:
|
||||
"""Background task: run conversion and update job status."""
|
||||
try:
|
||||
job = await analysis_repo.find_by_id(job_id)
|
||||
if not job:
|
||||
logger.error("Analysis job %s not found", job_id)
|
||||
return
|
||||
|
||||
job.mark_running()
|
||||
await analysis_repo.update_status(job)
|
||||
logger.info("Analysis started: %s (file: %s)", job_id, filename)
|
||||
|
||||
options = ConversionOptions(**(pipeline_options or {}))
|
||||
|
||||
result: ConversionResult = await asyncio.wait_for(
|
||||
self._converter.convert(file_path, options),
|
||||
timeout=self._conversion_timeout,
|
||||
)
|
||||
|
||||
pages_json = json.dumps([asdict(p) for p in result.pages])
|
||||
|
||||
job.mark_completed(
|
||||
markdown=result.content_markdown,
|
||||
html=result.content_html,
|
||||
pages_json=pages_json,
|
||||
)
|
||||
await analysis_repo.update_status(job)
|
||||
|
||||
if result.page_count:
|
||||
await document_repo.update_page_count(job.document_id, result.page_count)
|
||||
|
||||
logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count)
|
||||
|
||||
except TimeoutError:
|
||||
logger.error("Analysis timed out after %ds: %s", self._conversion_timeout, job_id)
|
||||
await _mark_failed(job_id, f"Conversion timed out after {self._conversion_timeout}s")
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("Analysis failed: %s", job_id)
|
||||
await _mark_failed(job_id, str(e))
|
||||
|
||||
|
||||
def _on_task_done(task: asyncio.Task) -> None:
|
||||
|
|
@ -46,65 +106,6 @@ def _on_task_done(task: asyncio.Task) -> None:
|
|||
logger.error("Unhandled exception in analysis task: %s", exc, exc_info=True)
|
||||
|
||||
|
||||
async def find_all() -> list[AnalysisJob]:
|
||||
return await analysis_repo.find_all()
|
||||
|
||||
|
||||
async def find_by_id(job_id: str) -> AnalysisJob | None:
|
||||
return await analysis_repo.find_by_id(job_id)
|
||||
|
||||
|
||||
async def delete(job_id: str) -> bool:
|
||||
return await analysis_repo.delete(job_id)
|
||||
|
||||
|
||||
async def _run_analysis(
|
||||
job_id: str, file_path: str, filename: str, pipeline_options: dict | None = None,
|
||||
) -> None:
|
||||
"""Background task: run Docling conversion and update job status."""
|
||||
try:
|
||||
job = await analysis_repo.find_by_id(job_id)
|
||||
if not job:
|
||||
logger.error("Analysis job %s not found", job_id)
|
||||
return
|
||||
|
||||
job.mark_running()
|
||||
await analysis_repo.update_status(job)
|
||||
logger.info("Analysis started: %s (file: %s)", job_id, filename)
|
||||
|
||||
# Build conversion options from pipeline dict
|
||||
options = ConversionOptions(**(pipeline_options or {}))
|
||||
|
||||
# Run blocking Docling conversion in a thread with timeout
|
||||
result: ConversionResult = await asyncio.wait_for(
|
||||
asyncio.to_thread(convert_document, file_path, options),
|
||||
timeout=CONVERSION_TIMEOUT,
|
||||
)
|
||||
|
||||
pages_json = json.dumps([asdict(p) for p in result.pages])
|
||||
|
||||
job.mark_completed(
|
||||
markdown=result.content_markdown,
|
||||
html=result.content_html,
|
||||
pages_json=pages_json,
|
||||
)
|
||||
await analysis_repo.update_status(job)
|
||||
|
||||
# Update document page count if available
|
||||
if result.page_count:
|
||||
await document_repo.update_page_count(job.document_id, result.page_count)
|
||||
|
||||
logger.info("Analysis completed: %s (%d pages)", job_id, result.page_count)
|
||||
|
||||
except TimeoutError:
|
||||
logger.error("Analysis timed out after %ds: %s", CONVERSION_TIMEOUT, job_id)
|
||||
await _mark_failed(job_id, f"Conversion timed out after {CONVERSION_TIMEOUT}s")
|
||||
|
||||
except Exception as e:
|
||||
logger.exception("Analysis failed: %s", job_id)
|
||||
await _mark_failed(job_id, str(e))
|
||||
|
||||
|
||||
async def _mark_failed(job_id: str, error: str) -> None:
|
||||
"""Safely mark a job as failed, handling DB errors gracefully."""
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -18,7 +18,9 @@ class TestHealthEndpoint:
|
|||
def test_health(self, client):
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == {"status": "ok"}
|
||||
data = resp.json()
|
||||
assert data["status"] == "ok"
|
||||
assert "engine" in data
|
||||
|
||||
|
||||
class TestDocumentEndpoints:
|
||||
|
|
@ -102,7 +104,7 @@ class TestDocumentEndpoints:
|
|||
|
||||
|
||||
class TestAnalysisEndpoints:
|
||||
@patch("services.analysis_service.find_all", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.find_all", new_callable=AsyncMock)
|
||||
def test_list_analyses(self, mock_find_all, client):
|
||||
mock_find_all.return_value = [
|
||||
AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf"),
|
||||
|
|
@ -117,7 +119,7 @@ class TestAnalysisEndpoints:
|
|||
assert data[0]["documentFilename"] == "test.pdf"
|
||||
assert data[0]["status"] == "PENDING"
|
||||
|
||||
@patch("services.analysis_service.find_by_id", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.find_by_id", new_callable=AsyncMock)
|
||||
def test_get_analysis(self, mock_find, client):
|
||||
job = AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf")
|
||||
job.mark_running()
|
||||
|
|
@ -129,14 +131,14 @@ class TestAnalysisEndpoints:
|
|||
assert data["status"] == "RUNNING"
|
||||
assert data["startedAt"] is not None
|
||||
|
||||
@patch("services.analysis_service.find_by_id", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.find_by_id", new_callable=AsyncMock)
|
||||
def test_get_analysis_not_found(self, mock_find, client):
|
||||
mock_find.return_value = None
|
||||
|
||||
resp = client.get("/api/analyses/missing")
|
||||
assert resp.status_code == 404
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_create_analysis(self, mock_create, client):
|
||||
mock_create.return_value = AnalysisJob(
|
||||
id="j1", document_id="d1", document_filename="test.pdf",
|
||||
|
|
@ -149,7 +151,7 @@ class TestAnalysisEndpoints:
|
|||
assert data["documentId"] == "d1"
|
||||
mock_create.assert_called_once_with("d1", pipeline_options=None)
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_create_analysis_with_pipeline_options(self, mock_create, client):
|
||||
mock_create.return_value = AnalysisJob(
|
||||
id="j2", document_id="d1", document_filename="test.pdf",
|
||||
|
|
@ -182,7 +184,7 @@ class TestAnalysisEndpoints:
|
|||
assert opts["generate_picture_images"] is True
|
||||
assert opts["images_scale"] == 2.0
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_create_analysis_with_partial_pipeline_options(self, mock_create, client):
|
||||
"""Pipeline options should use defaults for unspecified fields."""
|
||||
mock_create.return_value = AnalysisJob(
|
||||
|
|
@ -202,7 +204,7 @@ class TestAnalysisEndpoints:
|
|||
assert opts["table_mode"] == "accurate"
|
||||
assert opts["do_code_enrichment"] is False
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_create_analysis_document_not_found(self, mock_create, client):
|
||||
mock_create.side_effect = ValueError("Document not found: d99")
|
||||
|
||||
|
|
@ -217,14 +219,14 @@ class TestAnalysisEndpoints:
|
|||
resp = client.post("/api/analyses", json={"documentId": " "})
|
||||
assert resp.status_code == 400
|
||||
|
||||
@patch("services.analysis_service.delete", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.delete", new_callable=AsyncMock)
|
||||
def test_delete_analysis(self, mock_delete, client):
|
||||
mock_delete.return_value = True
|
||||
|
||||
resp = client.delete("/api/analyses/j1")
|
||||
assert resp.status_code == 204
|
||||
|
||||
@patch("services.analysis_service.delete", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.delete", new_callable=AsyncMock)
|
||||
def test_delete_analysis_not_found(self, mock_delete, client):
|
||||
mock_delete.return_value = False
|
||||
|
||||
|
|
|
|||
|
|
@ -131,8 +131,8 @@ class TestBuildConverter:
|
|||
class TestConvertDocumentRouting:
|
||||
"""Verify convert_document uses default converter for default opts, custom otherwise."""
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_default_converter_with_all_defaults(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -148,8 +148,8 @@ class TestConvertDocumentRouting:
|
|||
mock_get_default.assert_called_once()
|
||||
mock_build.assert_not_called()
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_ocr_disabled(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -165,8 +165,8 @@ class TestConvertDocumentRouting:
|
|||
mock_build.assert_called_once()
|
||||
mock_get_default.assert_not_called()
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_table_mode_fast(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -182,8 +182,8 @@ class TestConvertDocumentRouting:
|
|||
|
||||
mock_build.assert_called_once_with(opts)
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_code_enrichment_on(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -199,8 +199,8 @@ class TestConvertDocumentRouting:
|
|||
|
||||
mock_build.assert_called_once_with(opts)
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_formula_enrichment_on(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -215,8 +215,8 @@ class TestConvertDocumentRouting:
|
|||
|
||||
mock_build.assert_called_once()
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_picture_options_on(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -231,8 +231,8 @@ class TestConvertDocumentRouting:
|
|||
|
||||
mock_build.assert_called_once()
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_generate_images_on(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -247,8 +247,8 @@ class TestConvertDocumentRouting:
|
|||
|
||||
mock_build.assert_called_once()
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_uses_custom_converter_when_images_scale_changed(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -264,8 +264,8 @@ class TestConvertDocumentRouting:
|
|||
|
||||
mock_build.assert_called_once_with(opts)
|
||||
|
||||
@patch("domain.parsing.get_default_converter")
|
||||
@patch("domain.parsing.build_converter")
|
||||
@patch("infra.local_converter._get_default_converter")
|
||||
@patch("infra.local_converter._build_docling_converter")
|
||||
def test_forwards_all_options_to_build_converter(self, mock_build, mock_get_default):
|
||||
mock_conv = MagicMock()
|
||||
mock_result = MagicMock()
|
||||
|
|
@ -312,25 +312,21 @@ class TestServiceForwardsPipelineOptions:
|
|||
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service._run_analysis")
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_passes_pipeline_options_to_run(
|
||||
self, mock_run, mock_analysis_repo, mock_doc_repo, mock_doc,
|
||||
self, mock_analysis_repo, mock_doc_repo, mock_doc,
|
||||
):
|
||||
mock_doc_repo.find_by_id = AsyncMock(return_value=mock_doc)
|
||||
mock_analysis_repo.insert = AsyncMock()
|
||||
# Patch _run_analysis as a coroutine that we can inspect
|
||||
mock_run.return_value = None
|
||||
|
||||
from services import analysis_service
|
||||
mock_converter = AsyncMock()
|
||||
from services.analysis_service import AnalysisService
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
|
||||
opts = {"do_ocr": False, "table_mode": "fast"}
|
||||
|
||||
# We need to patch asyncio.create_task to capture the coroutine args
|
||||
with patch("services.analysis_service.asyncio.create_task") as mock_task:
|
||||
await analysis_service.create("d1", pipeline_options=opts)
|
||||
|
||||
# create_task should have been called with _run_analysis(...)
|
||||
await svc.create("d1", pipeline_options=opts)
|
||||
mock_task.assert_called_once()
|
||||
|
||||
@patch("services.analysis_service.document_repo")
|
||||
|
|
@ -342,32 +338,36 @@ class TestServiceForwardsPipelineOptions:
|
|||
mock_doc_repo.find_by_id = AsyncMock(return_value=mock_doc)
|
||||
mock_analysis_repo.insert = AsyncMock()
|
||||
|
||||
from services import analysis_service
|
||||
mock_converter = AsyncMock()
|
||||
from services.analysis_service import AnalysisService
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
|
||||
with patch("services.analysis_service.asyncio.create_task") as mock_task:
|
||||
await analysis_service.create("d1")
|
||||
await svc.create("d1")
|
||||
mock_task.assert_called_once()
|
||||
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@patch("services.analysis_service.convert_document")
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_analysis_forwards_options_to_convert(
|
||||
self, mock_convert, mock_doc_repo, mock_analysis_repo, mock_job,
|
||||
self, mock_doc_repo, mock_analysis_repo, mock_job,
|
||||
):
|
||||
from domain.parsing import ConversionResult, PageDetail
|
||||
from domain.value_objects import ConversionResult, PageDetail
|
||||
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_doc_repo.update_page_count = AsyncMock()
|
||||
mock_convert.return_value = ConversionResult(
|
||||
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.return_value = ConversionResult(
|
||||
page_count=1,
|
||||
content_markdown="# Test",
|
||||
content_html="<h1>Test</h1>",
|
||||
pages=[PageDetail(page_number=1, width=612.0, height=792.0)],
|
||||
)
|
||||
|
||||
from services.analysis_service import _run_analysis
|
||||
from services.analysis_service import AnalysisService
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
|
||||
opts = {
|
||||
"do_ocr": False,
|
||||
|
|
@ -381,10 +381,10 @@ class TestServiceForwardsPipelineOptions:
|
|||
"images_scale": 2.0,
|
||||
}
|
||||
|
||||
await _run_analysis("j1", "/tmp/test.pdf", "test.pdf", opts)
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", opts)
|
||||
|
||||
mock_convert.assert_called_once()
|
||||
call_args = mock_convert.call_args
|
||||
mock_converter.convert.assert_called_once()
|
||||
call_args = mock_converter.convert.call_args
|
||||
assert call_args[0][0] == "/tmp/test.pdf"
|
||||
conv_opts = call_args[0][1]
|
||||
assert conv_opts.do_ocr is False
|
||||
|
|
@ -395,47 +395,50 @@ class TestServiceForwardsPipelineOptions:
|
|||
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@patch("services.analysis_service.convert_document")
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_analysis_uses_defaults_when_no_options(
|
||||
self, mock_convert, mock_doc_repo, mock_analysis_repo, mock_job,
|
||||
self, mock_doc_repo, mock_analysis_repo, mock_job,
|
||||
):
|
||||
from domain.parsing import ConversionResult, PageDetail
|
||||
from domain.value_objects import ConversionResult, PageDetail
|
||||
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_doc_repo.update_page_count = AsyncMock()
|
||||
mock_convert.return_value = ConversionResult(
|
||||
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.return_value = ConversionResult(
|
||||
page_count=1,
|
||||
content_markdown="",
|
||||
content_html="",
|
||||
pages=[PageDetail(page_number=1, width=612.0, height=792.0)],
|
||||
)
|
||||
|
||||
from services.analysis_service import _run_analysis
|
||||
from services.analysis_service import AnalysisService
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
|
||||
await _run_analysis("j1", "/tmp/test.pdf", "test.pdf", None)
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", None)
|
||||
|
||||
# Called with file_path and default ConversionOptions
|
||||
mock_convert.assert_called_once()
|
||||
call_args = mock_convert.call_args
|
||||
mock_converter.convert.assert_called_once()
|
||||
call_args = mock_converter.convert.call_args
|
||||
assert call_args[0][0] == "/tmp/test.pdf"
|
||||
assert call_args[0][1] == ConversionOptions()
|
||||
|
||||
@patch("services.analysis_service.analysis_repo")
|
||||
@patch("services.analysis_service.document_repo")
|
||||
@patch("services.analysis_service.convert_document")
|
||||
@pytest.mark.asyncio
|
||||
async def test_run_analysis_marks_failed_on_error(
|
||||
self, mock_convert, mock_doc_repo, mock_analysis_repo, mock_job,
|
||||
self, mock_doc_repo, mock_analysis_repo, mock_job,
|
||||
):
|
||||
mock_analysis_repo.find_by_id = AsyncMock(return_value=mock_job)
|
||||
mock_analysis_repo.update_status = AsyncMock()
|
||||
mock_convert.side_effect = RuntimeError("Docling crashed")
|
||||
|
||||
from services.analysis_service import _run_analysis
|
||||
mock_converter = AsyncMock()
|
||||
mock_converter.convert.side_effect = RuntimeError("Docling crashed")
|
||||
|
||||
await _run_analysis("j1", "/tmp/test.pdf", "test.pdf", {"do_ocr": False})
|
||||
from services.analysis_service import AnalysisService
|
||||
svc = AnalysisService(converter=mock_converter)
|
||||
|
||||
await svc._run_analysis("j1", "/tmp/test.pdf", "test.pdf", {"do_ocr": False})
|
||||
|
||||
# Should have called update_status twice: RUNNING then FAILED
|
||||
assert mock_analysis_repo.update_status.call_count == 2
|
||||
|
|
@ -458,7 +461,7 @@ class TestAnalysisEndpointPipelineOptions:
|
|||
from main import app
|
||||
return TestClient(app, raise_server_exceptions=False)
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_no_pipeline_options_sends_none(self, mock_create, client):
|
||||
from domain.models import AnalysisJob
|
||||
mock_create.return_value = AnalysisJob(id="j1", document_id="d1")
|
||||
|
|
@ -467,7 +470,7 @@ class TestAnalysisEndpointPipelineOptions:
|
|||
|
||||
mock_create.assert_called_once_with("d1", pipeline_options=None)
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_empty_pipeline_options_object_uses_defaults(self, mock_create, client):
|
||||
from domain.models import AnalysisJob
|
||||
mock_create.return_value = AnalysisJob(id="j1", document_id="d1")
|
||||
|
|
@ -485,7 +488,7 @@ class TestAnalysisEndpointPipelineOptions:
|
|||
assert opts["do_formula_enrichment"] is False
|
||||
assert opts["images_scale"] == 1.0
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_partial_pipeline_options_merges_with_defaults(self, mock_create, client):
|
||||
from domain.models import AnalysisJob
|
||||
mock_create.return_value = AnalysisJob(id="j1", document_id="d1")
|
||||
|
|
@ -508,7 +511,7 @@ class TestAnalysisEndpointPipelineOptions:
|
|||
assert opts["generate_picture_images"] is False
|
||||
assert opts["generate_page_images"] is False
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_full_pipeline_options(self, mock_create, client):
|
||||
from domain.models import AnalysisJob
|
||||
mock_create.return_value = AnalysisJob(id="j1", document_id="d1")
|
||||
|
|
@ -535,7 +538,7 @@ class TestAnalysisEndpointPipelineOptions:
|
|||
opts = mock_create.call_args.kwargs["pipeline_options"]
|
||||
assert opts == payload["pipelineOptions"]
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_invalid_pipeline_option_type_rejected(self, mock_create, client):
|
||||
resp = client.post("/api/analyses", json={
|
||||
"documentId": "d1",
|
||||
|
|
@ -543,7 +546,7 @@ class TestAnalysisEndpointPipelineOptions:
|
|||
})
|
||||
assert resp.status_code == 422
|
||||
|
||||
@patch("services.analysis_service.create", new_callable=AsyncMock)
|
||||
@patch("main.analysis_service.create", new_callable=AsyncMock)
|
||||
def test_unknown_pipeline_option_ignored(self, mock_create, client):
|
||||
from domain.models import AnalysisJob
|
||||
mock_create.return_value = AnalysisJob(id="j1", document_id="d1")
|
||||
|
|
|
|||
Loading…
Reference in a new issue