238 lines
7.7 KiB
Python
238 lines
7.7 KiB
Python
"""Remote Docling Serve converter — delegates conversion via HTTP.
|
|
|
|
This adapter implements the DocumentConverter port by calling a remote
|
|
Docling Serve instance's REST API (v1).
|
|
|
|
API contract based on docling-serve source code:
|
|
- Options are sent as individual multipart form fields (not a JSON blob)
|
|
- Response contains document.md_content, document.html_content, document.json_content
|
|
- json_content is a serialized DoclingDocument with texts[], tables[], pictures[]
|
|
- Bounding boxes use {l, t, r, b, coord_origin} format
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import mimetypes
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
|
|
from domain.value_objects import (
|
|
ConversionOptions,
|
|
ConversionResult,
|
|
PageDetail,
|
|
PageElement,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_API_PREFIX = "/v1"
|
|
_DEFAULT_TIMEOUT = 600.0
|
|
|
|
# Docling Serve label → our element type
|
|
_LABEL_MAP = {
|
|
"table": "table",
|
|
"picture": "picture",
|
|
"figure": "picture",
|
|
"title": "title",
|
|
"section_header": "section_header",
|
|
"list_item": "list",
|
|
"formula": "formula",
|
|
"code": "code",
|
|
"caption": "text",
|
|
"footnote": "text",
|
|
"page_header": "text",
|
|
"page_footer": "text",
|
|
"paragraph": "text",
|
|
"text": "text",
|
|
"reference": "text",
|
|
}
|
|
|
|
|
|
class ServeConverter:
|
|
"""Adapter that delegates document conversion to a remote Docling Serve instance."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
api_key: str | None = None,
|
|
timeout: float = _DEFAULT_TIMEOUT,
|
|
):
|
|
self._base_url = base_url.rstrip("/")
|
|
self._api_key = api_key
|
|
self._timeout = timeout
|
|
|
|
def _headers(self) -> dict[str, str]:
|
|
headers: dict[str, str] = {}
|
|
if self._api_key:
|
|
headers["X-Api-Key"] = self._api_key
|
|
return headers
|
|
|
|
async def convert(
|
|
self,
|
|
file_path: str,
|
|
options: ConversionOptions,
|
|
) -> ConversionResult:
|
|
"""Convert a document by uploading it to Docling Serve."""
|
|
path = Path(file_path)
|
|
content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
|
|
form_data = _build_form_data(options)
|
|
url = f"{self._base_url}{_API_PREFIX}/convert/file"
|
|
|
|
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
|
with open(path, "rb") as f:
|
|
response = await client.post(
|
|
url,
|
|
files={"files": (path.name, f, content_type)},
|
|
data=form_data,
|
|
headers=self._headers(),
|
|
)
|
|
|
|
response.raise_for_status()
|
|
result_data = response.json()
|
|
|
|
return _parse_response(result_data)
|
|
|
|
async def health_check(self) -> bool:
|
|
"""Check if Docling Serve is reachable."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
resp = await client.get(
|
|
f"{self._base_url}/version",
|
|
headers=self._headers(),
|
|
)
|
|
return resp.status_code == 200
|
|
except httpx.HTTPError:
|
|
logger.warning("Docling Serve health check failed at %s", self._base_url, exc_info=True)
|
|
return False
|
|
|
|
|
|
def _build_form_data(options: ConversionOptions) -> dict[str, str | list[str]]:
|
|
"""Build form fields matching Docling Serve's multipart form contract.
|
|
|
|
Array fields (to_formats) are sent as lists — httpx encodes them as
|
|
repeated form keys (to_formats=md&to_formats=html&to_formats=json).
|
|
"""
|
|
return {
|
|
"to_formats": ["md", "html", "json"],
|
|
"do_ocr": str(options.do_ocr).lower(),
|
|
"do_table_structure": str(options.do_table_structure).lower(),
|
|
"table_mode": options.table_mode,
|
|
"do_code_enrichment": str(options.do_code_enrichment).lower(),
|
|
"do_formula_enrichment": str(options.do_formula_enrichment).lower(),
|
|
"do_picture_classification": str(options.do_picture_classification).lower(),
|
|
"do_picture_description": str(options.do_picture_description).lower(),
|
|
"include_images": str(options.generate_picture_images).lower(),
|
|
"images_scale": str(options.images_scale),
|
|
}
|
|
|
|
|
|
def _parse_response(data: dict) -> ConversionResult:
|
|
"""Parse Docling Serve v1 ConvertDocumentResponse into our domain ConversionResult."""
|
|
document = data.get("document", {})
|
|
|
|
content_md = document.get("md_content") or ""
|
|
content_html = document.get("html_content") or ""
|
|
|
|
# json_content contains the full DoclingDocument with pages, elements, bboxes
|
|
json_content = document.get("json_content")
|
|
if isinstance(json_content, str):
|
|
try:
|
|
json_content = json.loads(json_content)
|
|
except json.JSONDecodeError:
|
|
logger.warning("Failed to parse json_content as JSON, ignoring structured data")
|
|
json_content = None
|
|
|
|
pages: list[PageDetail] = []
|
|
if json_content:
|
|
pages = _extract_pages_from_docling_document(json_content)
|
|
|
|
page_count = len(pages) if pages else 1
|
|
|
|
return ConversionResult(
|
|
page_count=page_count,
|
|
content_markdown=content_md,
|
|
content_html=content_html,
|
|
pages=pages,
|
|
)
|
|
|
|
|
|
def _extract_pages_from_docling_document(doc: dict) -> list[PageDetail]:
|
|
"""Extract pages with elements from a serialized DoclingDocument.
|
|
|
|
DoclingDocument structure:
|
|
- pages: {page_no: {size: {width, height}}}
|
|
- texts: [{label, text, prov: [{page_no, bbox: {l,t,r,b,coord_origin}}]}]
|
|
- tables: [{label, prov: [...], data: {...}}]
|
|
- pictures: [{label, prov: [...]}]
|
|
"""
|
|
pages_dict: dict[int, PageDetail] = {}
|
|
|
|
# Build page dimensions
|
|
for page_key, page_data in doc.get("pages", {}).items():
|
|
page_no = int(page_key)
|
|
size = page_data.get("size", {})
|
|
pages_dict[page_no] = PageDetail(
|
|
page_number=page_no,
|
|
width=size.get("width", 612.0),
|
|
height=size.get("height", 792.0),
|
|
)
|
|
|
|
# Process all element arrays
|
|
for item in doc.get("texts", []):
|
|
_add_element(item, pages_dict)
|
|
|
|
for item in doc.get("tables", []):
|
|
_add_element(item, pages_dict)
|
|
|
|
for item in doc.get("pictures", []):
|
|
_add_element(item, pages_dict)
|
|
|
|
return sorted(pages_dict.values(), key=lambda p: p.page_number)
|
|
|
|
|
|
def _add_element(item: dict, pages: dict[int, PageDetail]) -> None:
|
|
"""Add an element from a DoclingDocument array to the correct page."""
|
|
label = item.get("label", "text")
|
|
element_type = _LABEL_MAP.get(label, "text")
|
|
content = item.get("text", "") or ""
|
|
|
|
for prov in item.get("prov", []):
|
|
page_no = prov.get("page_no", 1)
|
|
if page_no not in pages:
|
|
pages[page_no] = PageDetail(
|
|
page_number=page_no,
|
|
width=612.0,
|
|
height=792.0,
|
|
)
|
|
|
|
bbox_data = prov.get("bbox", {})
|
|
bbox = _extract_bbox(bbox_data, pages[page_no].height)
|
|
|
|
pages[page_no].elements.append(
|
|
PageElement(type=element_type, bbox=bbox, content=content, level=0)
|
|
)
|
|
|
|
|
|
def _extract_bbox(bbox_data: dict, page_height: float) -> list[float]:
|
|
"""Extract and normalize bbox to TOPLEFT [l, t, r, b] format."""
|
|
if not isinstance(bbox_data, dict):
|
|
return [0.0, 0.0, 0.0, 0.0]
|
|
|
|
left = bbox_data.get("l", 0.0)
|
|
top = bbox_data.get("t", 0.0)
|
|
right = bbox_data.get("r", 0.0)
|
|
bottom = bbox_data.get("b", 0.0)
|
|
coord_origin = bbox_data.get("coord_origin", "TOPLEFT")
|
|
|
|
if coord_origin == "BOTTOMLEFT":
|
|
# In BOTTOMLEFT: top has higher y, bottom has lower y
|
|
# In TOPLEFT: flip both — new_top = page_height - old_top
|
|
new_top = page_height - top
|
|
new_bottom = page_height - bottom
|
|
top, bottom = new_top, new_bottom
|
|
|
|
return [left, top, right, bottom]
|