+ {{ + analysisStore.currentChunks.length ? t('chunking.noChunksOnPage') : t('chunking.noChunks') + }} +
+diff --git a/.env.example b/.env.example index 8ff39a2..dbcdc9a 100644 --- a/.env.example +++ b/.env.example @@ -1,3 +1,27 @@ +# Docker Compose build target: "local" or "remote" (selects Dockerfile target) +# CONVERSION_MODE=local + +# Conversion engine: "local" (in-process Docling) or "remote" (Docling Serve) +# Set automatically by the Docker target — override only for local dev +# CONVERSION_ENGINE=local + +# Docling Serve settings (remote mode only) +# DOCLING_SERVE_URL=http://localhost:5001 +# DOCLING_SERVE_API_KEY= + +# Max seconds per conversion (default: 600) +# CONVERSION_TIMEOUT=600 + +# Max parallel analysis jobs (default: 3) +# MAX_CONCURRENT_ANALYSES=3 + +# Deployment mode: "self-hosted" (default) or "huggingface" +# Shows disclaimer banner when set to "huggingface" +# DEPLOYMENT_MODE=self-hosted + +# Application version (set automatically by CI/Docker build) +# APP_VERSION=dev + # CORS (comma-separated origins, only needed for custom deployments) # CORS_ORIGINS=http://localhost:3000,https://your-domain.com diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index cde2426..b58235a 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,4 +1,4 @@ -name: Release Docker Image +name: Release Docker Images on: push: @@ -11,12 +11,16 @@ env: jobs: release: - name: Build & push Docker image + name: Build & push — ${{ matrix.target }} runs-on: ubuntu-latest permissions: contents: read packages: write + strategy: + matrix: + target: [remote, local] + steps: - uses: actions/checkout@v4 @@ -28,22 +32,28 @@ jobs: username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} - - name: Extract version from tag + - name: Docker metadata id: meta uses: docker/metadata-action@v5 with: images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }} tags: | - type=semver,pattern={{version}} - type=semver,pattern={{major}}.{{minor}} - type=raw,value=latest + type=semver,pattern={{version}},suffix=-${{ matrix.target }} + type=semver,pattern={{major}}.{{minor}},suffix=-${{ matrix.target }} + type=raw,value=latest-${{ matrix.target }} + + - name: Extract version from tag + id: version + run: echo "value=${GITHUB_REF_NAME#v}" >> "$GITHUB_OUTPUT" - uses: docker/build-push-action@v6 with: context: . + target: ${{ matrix.target }} push: true platforms: linux/amd64,linux/arm64 + build-args: APP_VERSION=${{ steps.version.outputs.value }} tags: ${{ steps.meta.outputs.tags }} labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max + cache-from: type=gha,scope=${{ matrix.target }} + cache-to: type=gha,mode=max,scope=${{ matrix.target }} diff --git a/.gitignore b/.gitignore index dcc86cc..97313cf 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,8 @@ site/ .vscode/ .run/ .claude/ +CLAUDE.md +**/CLAUDE.md *.iml # OS @@ -32,6 +34,9 @@ __pycache__/ .venv/ *.egg-info/ +# Excalidraw working files +docs/excalidraw/ + # Logs *.log hs_err_pid* diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d913ad..152901f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,38 @@ All notable changes to Docling Studio will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/), and this project adheres to [Semantic Versioning](https://semver.org/). +## [0.3.0] - 2026-04-07 + +### Added + +- Chunking support: domain objects, persistence, API endpoints, and frontend Prepare mode +- Chunk-to-bbox hover highlighting in Prepare mode +- Page filtering and collapsible config in Prepare mode +- Feature flipping mechanism +- Reusable pagination composable and PaginationBar component +- Version display in sidebar, settings page, and health endpoint + +### Fixed + +- Feature flag health check blocked by CORS +- Zombie jobs and unprotected JSON parse +- Upload error not displayed in DocumentUpload component +- Serve API contract: send `to_formats` as repeated form fields +- Audit findings: security, robustness, dead code, domain-infra violation + +### Changed + +- Refactored backend to hexagonal architecture for converter extensibility +- Added ServeConverter adapter for remote Docling Serve integration +- Moved `@vitest/mocker` from dependencies to devDependencies + +## [0.2.0] - 2025-05-14 + +### Added + +- Multi-arch Docker image release pipeline (GitHub Actions) +- Docker image published to `ghcr.io/scub-france/Docling-Studio` + ## [0.1.0] - 2025-01-01 ### Added diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 5010328..baef553 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -22,7 +22,13 @@ Thank you for your interest in contributing to Docling Studio! This guide will h ```bash cd document-parser python -m venv .venv && source .venv/bin/activate + +# Remote mode (lightweight — delegates to Docling Serve) pip install -r requirements.txt + +# Local mode (full — runs Docling in-process) +pip install -r requirements-local.txt + pip install ruff pytest pytest-asyncio httpx # dev tools uvicorn main:app --reload --port 8000 ``` @@ -61,11 +67,11 @@ npx prettier --write src/ # auto-format ## Running Tests ```bash -# Backend (99 tests) +# Backend (199 tests) cd document-parser pytest tests/ -v -# Frontend (81 tests) +# Frontend (129 tests) cd frontend npm run test:run ``` @@ -80,6 +86,79 @@ All tests must pass before submitting a PR. 4. Describe **what** changed and **why** in the PR description 5. Ensure CI passes (tests + build) +## Branching Strategy + +We follow a simplified Git Flow: + +| Branch | Purpose | +|--------|---------| +| `main` | Always stable — latest release merged back | +| `release/X.Y.Z` | Release preparation (freeze, bugfixes, changelog) | +| `feature/*` | New features — PR to `main` | +| `fix/*` | Bug fixes — PR to `main` (or `release/*` for pre-release fixes) | +| `hotfix/X.Y.Z` | Urgent fix on a released version — PR to `main` | + +Rules: +- All PRs target `main` (never stack branches on other feature branches) +- `release/*` branches are created from `main` when preparing a release +- `hotfix/*` branches are created from the release tag + +## Versioning + +We use [Semantic Versioning](https://semver.org/): `MAJOR.MINOR.PATCH`. + +- **Source of truth**: the git tag (`vX.Y.Z`) +- `package.json` version should match the current release branch +- The build injects the version automatically (Vite `__APP_VERSION__` for frontend, `APP_VERSION` env var for backend) + +## Release Process + +1. **Create the release branch** from `main`: + ```bash + git checkout main && git pull + git checkout -b release/X.Y.Z + ``` + +2. **On the release branch**, only: + - Bug fixes + - Move `[Unreleased]` to `[X.Y.Z] - YYYY-MM-DD` in `CHANGELOG.md` + - Update `version` in `frontend/package.json` + +3. **Merge into `main`** via PR, then **tag on `main`**: + ```bash + git checkout main && git pull + git tag vX.Y.Z + git push origin vX.Y.Z + ``` + +4. The tag triggers the **release workflow** which builds and pushes the Docker image to `ghcr.io`. + +### Docker Image Tags + +Each release produces two image variants: + +| Tag | Description | +|-----|-------------| +| `X.Y.Z-remote` | Exact version — lightweight (Docling Serve) | +| `X.Y.Z-local` | Exact version — full (in-process Docling) | +| `X.Y-remote` | Latest patch of this minor — lightweight | +| `X.Y-local` | Latest patch of this minor — full | +| `latest-remote` | Latest stable — lightweight | +| `latest-local` | Latest stable — full | + +### Hotfix + +```bash +git checkout vX.Y.Z # from the release tag +git checkout -b hotfix/X.Y.Z+1 +# fix, commit, PR to main +git tag vX.Y.Z+1 # tag on main after merge +``` + +### Changelog + +We follow [Keep a Changelog](https://keepachangelog.com/). Every PR should add a line under `[Unreleased]` in `CHANGELOG.md`. The release branch moves `[Unreleased]` to the versioned section. + ## Pull Request Guidelines - Keep PRs focused — one feature or fix per PR diff --git a/Dockerfile b/Dockerfile index 4c33624..ff3e37b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,33 +1,36 @@ # syntax=docker/dockerfile:1 # ============================================================================= -# Docling Studio — single-image build (frontend + backend) +# Docling Studio — single-image build (frontend + backend, multi-target) # # Usage: -# docker build -t docling-studio . -# docker run -p 3000:3000 docling-studio +# docker build --target remote -t docling-studio:remote . +# docker build --target local -t docling-studio:local . # ============================================================================= # --- Stage 1: Build frontend assets --- FROM node:20-alpine AS frontend-build +ARG APP_VERSION=dev + WORKDIR /build COPY frontend/package*.json ./ RUN npm ci COPY frontend/ . -RUN npm run build +RUN VITE_APP_VERSION=${APP_VERSION} npm run build -# --- Stage 2: Runtime (Python + Nginx) --- -FROM python:3.12-slim +# --- Stage 2: Base runtime (Python + Nginx) --- +FROM python:3.12-slim AS base -# System deps: poppler (pdf2image), nginx, and OpenCV runtime libs +ARG APP_VERSION=dev +ENV APP_VERSION=${APP_VERSION} + +# System deps: poppler (pdf2image), nginx RUN apt-get update && apt-get install -y --no-install-recommends \ poppler-utils \ - libgl1 \ - libglib2.0-0 \ nginx \ && rm -rf /var/lib/apt/lists/* -# Python deps +# Python deps (common) WORKDIR /app COPY document-parser/requirements.txt . RUN pip install --no-cache-dir -r requirements.txt @@ -52,5 +55,24 @@ ENV DB_PATH=/app/data/docling_studio.db EXPOSE 3000 -# nginx needs to run as root for port binding, then drops to appuser for uvicorn CMD ["sh", "-c", "nginx && exec su appuser -c 'uvicorn main:app --host 127.0.0.1 --port 8000'"] + +# --- Remote: lightweight, delegates to Docling Serve --- +FROM base AS remote +ENV CONVERSION_ENGINE=remote + +# --- Local: full Docling in-process --- +FROM base AS local + +RUN apt-get update && apt-get install -y --no-install-recommends \ + libgl1 \ + libglib2.0-0 \ + && rm -rf /var/lib/apt/lists/* + +COPY document-parser/requirements-local.txt . +RUN pip install --no-cache-dir torch torchvision --index-url https://download.pytorch.org/whl/cpu \ + && pip install --no-cache-dir -r requirements-local.txt + +RUN chown -R appuser:appuser /app \ + && chown -R appuser:appuser /usr/local/lib/python3.12/site-packages/rapidocr/models +ENV CONVERSION_ENGINE=local diff --git a/README.md b/README.md index 2477cc3..db5b76d 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,7 @@    +[](https://github.com/scub-france/Docling-Studio) A visual document analysis studio powered by [Docling](https://github.com/DS4SD/docling). Upload a PDF, configure the extraction pipeline, and visualize the results — text, tables, images, formulas, bounding boxes — all from your browser. @@ -48,8 +49,8 @@ document-parser/ ├── main.py # FastAPI app, CORS, lifespan ├── domain/ # Pure domain — no HTTP, no DB │ ├── models.py # Document, AnalysisJob dataclasses -│ ├── parsing.py # Docling conversion & page extraction -│ └── bbox.py # Bounding box coordinate normalization +│ ├── ports.py # Abstract protocols (converter, chunker) +│ └── value_objects.py # ConversionResult, PageDetail, ChunkResult ├── api/ # HTTP layer (FastAPI routers) │ ├── schemas.py # Pydantic DTOs (camelCase serialization) │ ├── documents.py # /api/documents endpoints @@ -61,7 +62,7 @@ document-parser/ ├── services/ # Use case orchestration │ ├── document_service.py # Upload, delete, preview │ └── analysis_service.py # Async Docling processing -└── tests/ # 99 tests (pytest) +└── tests/ # 199 tests (pytest) ``` ### Frontend structure (feature-based) @@ -85,22 +86,42 @@ frontend/src/ ## Quick Start -### Docker (fastest) +Docling Studio ships two Docker image variants: + +| Variant | Image tag | Size | Description | +|---------|-----------|------|-------------| +| **remote** | `latest-remote` | ~270 MB | Lightweight — delegates to an external [Docling Serve](https://github.com/DS4SD/docling-serve) instance | +| **local** | `latest-local` | ~1.9 GB | Full — runs Docling in-process, CPU-only (downloads ML models on first run) | + +### Docker — remote mode (fastest) ```bash -docker run -p 3000:3000 ghcr.io/scub-france/docling-studio:latest +docker run -p 3000:3000 \ + -e DOCLING_SERVE_URL=http://your-docling-serve:5001 \ + ghcr.io/scub-france/docling-studio:latest-remote ``` -Open [http://localhost:3000](http://localhost:3000) +### Docker — local mode (self-contained) + +```bash +docker run -p 3000:3000 ghcr.io/scub-france/docling-studio:latest-local +``` > **Note:** The first analysis takes longer as Docling downloads its ML models (~400 MB). Subsequent runs are fast. +Open [http://localhost:3000](http://localhost:3000) + ### Docker Compose (for development) ```bash git clone https://github.com/scub-france/Docling-Studio.git cd Docling-Studio + +# Local mode (default) docker compose up --build + +# Remote mode +CONVERSION_MODE=remote DOCLING_SERVE_URL=http://your-docling-serve:5001 docker compose up --build ``` ### Local Development @@ -109,7 +130,13 @@ docker compose up --build ```bash cd document-parser python -m venv .venv && source .venv/bin/activate + +# Remote mode (lightweight) pip install -r requirements.txt + +# Local mode (with Docling) +pip install -r requirements-local.txt + uvicorn main:app --reload --port 8000 ``` @@ -123,12 +150,12 @@ npm run dev ### Running Tests ```bash -# Backend (99 tests) +# Backend (199 tests) cd document-parser pip install pytest pytest-asyncio httpx pytest tests/ -v -# Frontend (81 tests) +# Frontend (129 tests) cd frontend npm run test:run ``` @@ -156,6 +183,9 @@ All configuration is done via environment variables. See [`.env.example`](.env.e | Variable | Default | Description | |----------|---------|-------------| +| `CONVERSION_ENGINE` | `local` | `local` (in-process Docling) or `remote` (Docling Serve) | +| `DOCLING_SERVE_URL` | `http://localhost:5001` | Docling Serve endpoint (remote mode only) | +| `DOCLING_SERVE_API_KEY` | — | API key for Docling Serve (optional) | | `CORS_ORIGINS` | `http://localhost:3000,...` | CORS allowed origins (comma-separated) | | `UPLOAD_DIR` | `./uploads` | File storage directory | | `DB_PATH` | `./data/docling_studio.db` | SQLite database path | @@ -167,14 +197,11 @@ GitHub Actions pipelines (see [`.github/workflows/`](.github/workflows/)): | Workflow | Trigger | What it does | |----------|---------|--------------| -| **CI** | push to `main`, pull requests | Lint + type check + Backend tests (99) + Frontend tests (81) + build | -| **Release** | push tag `v*` | Build & push multi-arch Docker image to `ghcr.io` | +| **CI** | push to `main`, pull requests | Lint + type check + Backend tests + Frontend tests + build | +| **Release** | push tag `v*` | Build & push **two** multi-arch Docker images (`remote` + `local`) to `ghcr.io` | +| **Docs** | push to `main` (docs changes) | Build & deploy MkDocs to GitHub Pages | -To publish a new version: -```bash -git tag v0.2.0 -git push origin v0.2.0 -``` +We follow [Semantic Versioning](https://semver.org/) with a simplified Git Flow. See [CONTRIBUTING.md](CONTRIBUTING.md) for the full release process. ## Performance & System Requirements @@ -186,12 +213,11 @@ git push origin v0.2.0 ### Docker Desktop settings -The document parser needs **at least 4 GB of RAM**: - -| Resource | Minimum | Recommended | -|----------|---------|-------------| -| Memory | 6 GB | 8 GB+ | -| CPUs | 4 | 8+ | +| | Remote image | Local image | +|---|---|---| +| **Image size** | ~270 MB | ~1.9 GB | +| **Memory** | 2 GB | 6 GB (recommended 8 GB+) | +| **CPUs** | 2 | 4 (recommended 8+) | ### Platform support diff --git a/docker-compose.yml b/docker-compose.yml index 6e75d5e..f1f80aa 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -2,6 +2,7 @@ services: document-parser: build: context: ./document-parser + target: ${CONVERSION_MODE:-local} expose: - "8000" volumes: @@ -9,6 +10,8 @@ services: - db_data:/app/data environment: CORS_ORIGINS: ${CORS_ORIGINS:-http://localhost:3000,http://localhost:5173} + DOCLING_SERVE_URL: ${DOCLING_SERVE_URL:-} + DOCLING_SERVE_API_KEY: ${DOCLING_SERVE_API_KEY:-} deploy: resources: limits: diff --git a/docs/architecture.md b/docs/architecture.md index 9cfcade..764795a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -2,7 +2,7 @@ ## Overview -{ width="700" } +{ width="700" } Two services communicating via REST. The frontend is a Vue 3 SPA served by Nginx in production. The backend is a FastAPI app that wraps Docling's document conversion engine. @@ -40,37 +40,47 @@ The backend follows a strict layered architecture. Dependencies flow inward: API ``` document-parser/ -├── main.py # FastAPI app, CORS, lifespan +├── main.py # FastAPI app, CORS, lifespan, health endpoint │ ├── domain/ # Pure domain — no HTTP, no DB │ ├── models.py # Document, AnalysisJob dataclasses -│ ├── parsing.py # Docling conversion & page extraction +│ ├── ports.py # Abstract protocols (DocumentConverter, DocumentChunker) +│ ├── value_objects.py # ConversionResult, ChunkingOptions, ChunkResult │ └── bbox.py # Bounding box coordinate normalization │ ├── api/ # HTTP layer (FastAPI routers) │ ├── schemas.py # Pydantic DTOs (camelCase serialization) │ ├── documents.py # /api/documents endpoints -│ └── analyses.py # /api/analyses endpoints +│ └── analyses.py # /api/analyses endpoints (create, rechunk, delete) │ ├── persistence/ # Data layer (SQLite via aiosqlite) │ ├── database.py # Connection management, schema init │ ├── document_repo.py # Document CRUD │ └── analysis_repo.py # AnalysisJob CRUD │ +├── infra/ # Infrastructure adapters +│ ├── settings.py # Environment-based configuration +│ ├── local_converter.py # In-process Docling converter (local mode) +│ ├── serve_converter.py # HTTP client for Docling Serve (remote mode) +│ ├── local_chunker.py # In-process chunking (HierarchicalChunker, HybridChunker) +│ ├── rate_limiter.py # Sliding-window rate limiting middleware +│ └── bbox.py # Bbox coordinate normalization helpers +│ ├── services/ # Use case orchestration │ ├── document_service.py # Upload, delete, preview -│ └── analysis_service.py # Async Docling processing +│ └── analysis_service.py # Async Docling processing + chunking │ -└── tests/ # pytest +└── tests/ # pytest (199 tests) ``` ### Layer responsibilities | Layer | Role | Depends on | |-------|------|------------| -| **domain** | Dataclasses, bbox math, Docling conversion | Nothing (pure Python) | +| **domain** | Dataclasses, value objects, abstract ports | Nothing (pure Python) | | **persistence** | SQLite CRUD, aiosqlite | domain (models) | -| **services** | Orchestrate use cases, call Docling | domain + persistence | +| **infra** | Adapters: converters, chunker, rate limiter, settings | domain (ports, value objects) | +| **services** | Orchestrate use cases, call converters/chunkers | domain + persistence + infra | | **api** | HTTP endpoints, Pydantic DTOs, error handling | services | ### API contract @@ -101,7 +111,9 @@ frontend/src/ │ │ ├── AnalysisPanel.vue │ │ ├── StructureViewer.vue │ │ └── ... +│ ├── chunking/ # Chunk panel UI + rechunk action │ ├── document/ # Document store, API, upload +│ ├── feature-flags/ # Feature flag store (reads /api/health) │ ├── history/ # History store, navigation │ └── settings/ # Theme, locale, API URL │ @@ -126,3 +138,73 @@ Backend response → Pinia store state → Vue reactivity → UI update - **TypeScript strict mode** with shared interfaces in `shared/types.ts`. - **No component library** — custom CSS with CSS variables for theming. - **vue-tsc** in CI to catch type errors before merge. + +## Feature Flags + +The frontend adapts its UI based on the backend's capabilities. On startup, the feature flag store fetches `/api/health` and reads the `engine` and `deploymentMode` fields. + +| Flag | Condition | Effect | +|------|-----------|--------| +| `chunking` | `engine === 'local'` | Shows chunking options in the analysis panel | +| `disclaimer` | `deploymentMode === 'huggingface'` | Shows a disclaimer banner at the top of the app | + +This allows the same frontend build to work with both local and remote backends without conditional compilation. + +## Rate Limiting + +The backend applies a sliding-window rate limiter as middleware: + +- **60 requests** per **60 seconds** per client IP +- The `/api/health` endpoint is excluded +- When the limit is exceeded, the API returns `429 Too Many Requests` with a `Retry-After` header + +## Analysis Lifecycle + +An analysis job follows this state machine: + +``` +PENDING → RUNNING → COMPLETED + → FAILED +``` + +| Status | Description | +|--------|-------------| +| `PENDING` | Job created, waiting for a processing slot | +| `RUNNING` | Docling conversion in progress | +| `COMPLETED` | Conversion finished — results available (markdown, HTML, pages, chunks) | +| `FAILED` | Conversion error — `error_message` contains details | + +The backend limits parallel jobs via `MAX_CONCURRENT_ANALYSES` (default: 3) to avoid overloading the CPU during Docling processing. + +## Local vs Remote Mode + +The backend supports two conversion engines, selected via the `CONVERSION_ENGINE` environment variable: + +| | Local | Remote | +|---|---|---| +| **Engine** | In-process Docling (PyTorch) | HTTP client to [Docling Serve](https://github.com/DS4SD/docling-serve) | +| **Chunking** | Available (in-process) | Not available | +| **Docker image** | `latest-local` (~1.9 GB) | `latest-remote` (~270 MB) | +| **ML models** | Downloaded on first run (~400 MB) | Managed by Docling Serve | +| **CPU/RAM** | 4+ CPUs, 6+ GB RAM | 2 CPUs, 2 GB RAM | + +The converter is selected at startup in `main.py` via `_build_converter()`. The chunker (`_build_chunker()`) is only instantiated in local mode — in remote mode, the chunking feature flag is disabled and the UI hides the chunking panel. + +## Health Endpoint + +`GET /api/health` returns the backend status: + +```json +{ + "status": "ok", + "engine": "local", + "version": "0.3.0", + "deploymentMode": "self-hosted" +} +``` + +The frontend uses this response to: + +1. Verify the backend is reachable +2. Evaluate feature flags (chunking, disclaimer) +3. Display the app version diff --git a/docs/bbox-pipeline.md b/docs/bbox-pipeline.md index b0be9e1..4f2e0fe 100644 --- a/docs/bbox-pipeline.md +++ b/docs/bbox-pipeline.md @@ -38,7 +38,7 @@ The frontend converts PDF points to CSS pixels, then the canvas renders at `devi ## Transformation 1 — `to_topleft_list()` -**File:** `document-parser/domain/bbox.py` +**File:** `document-parser/infra/bbox.py` Normalizes any Docling bbox to `[left, top, right, bottom]` in TOPLEFT coordinates. diff --git a/docs/contributing.md b/docs/contributing.md index 17c6e7d..44412e3 100644 --- a/docs/contributing.md +++ b/docs/contributing.md @@ -20,7 +20,13 @@ ```bash cd document-parser python -m venv .venv && source .venv/bin/activate + + # Remote mode (lightweight — delegates to Docling Serve) pip install -r requirements.txt + + # Local mode (full — runs Docling in-process) + pip install -r requirements-local.txt + pip install ruff pytest pytest-asyncio httpx uvicorn main:app --reload --port 8000 ``` diff --git a/docs/getting-started.md b/docs/getting-started.md index cd38397..b2ee2b8 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -1,33 +1,69 @@ # Getting Started -## Docker Compose (recommended) +Docling Studio ships two Docker image variants: + +| Variant | Image tag | Size | Description | +|---------|-----------|------|-------------| +| **remote** | `latest-remote` | ~270 MB | Lightweight — delegates to an external [Docling Serve](https://github.com/DS4SD/docling-serve) instance | +| **local** | `latest-local` | ~1.9 GB | Full — runs Docling in-process, CPU-only (downloads ML models on first run) | + +{ width="600" } + +## Docker — remote mode (fastest) + +```bash +docker run -p 3000:3000 \ + -e DOCLING_SERVE_URL=http://your-docling-serve:5001 \ + ghcr.io/scub-france/docling-studio:latest-remote +``` + +## Docker — local mode (self-contained) + +```bash +docker run -p 3000:3000 ghcr.io/scub-france/docling-studio:latest-local +``` + +> **Note:** The first analysis takes longer as Docling downloads its ML models (~400 MB). Subsequent runs are fast. + +Open [http://localhost:3000](http://localhost:3000). + +## Docker Compose (recommended for development) ```bash git clone https://github.com/scub-france/Docling-Studio.git cd Docling-Studio -docker compose up --build -``` -Open [http://localhost:3000](http://localhost:3000). +# Local mode (default) +docker compose up --build + +# Remote mode +CONVERSION_MODE=remote DOCLING_SERVE_URL=http://your-docling-serve:5001 docker compose up --build +``` ## Local Development -### Backend (Python 3.12+) +=== "Backend (Python 3.12+)" -```bash -cd document-parser -python -m venv .venv && source .venv/bin/activate -pip install -r requirements.txt -uvicorn main:app --reload --port 8000 -``` + ```bash + cd document-parser + python -m venv .venv && source .venv/bin/activate -### Frontend (Node 20+) + # Remote mode (lightweight) + pip install -r requirements.txt -```bash -cd frontend -npm install -npm run dev -``` + # Local mode (with Docling) + pip install -r requirements-local.txt + + uvicorn main:app --reload --port 8000 + ``` + +=== "Frontend (Node 20+)" + + ```bash + cd frontend + npm install + npm run dev + ``` The frontend runs on `http://localhost:3000` and proxies API calls to `http://localhost:8000`. @@ -65,22 +101,51 @@ These options map directly to Docling's [`PdfPipelineOptions`](https://docling-p | `generate_page_images` | `false` | Rasterize each page as an image | | `images_scale` | `1.0` | Scale factor for generated images (0.1–10) | +## Chunking Options + +!!! note + Chunking is only available in **local** mode. The chunking UI is hidden when using remote mode (Docling Serve). + +After a document is analyzed, you can split the extracted content into semantic chunks. Chunking can be configured at analysis time or re-run later with different options via the **rechunk** action. + +| Option | Default | Description | +|--------|---------|-------------| +| `chunker_type` | `hybrid` | `hybrid` (semantic + structural), `hierarchical` (heading-based), or `page` (one chunk per page) | +| `max_tokens` | `512` | Maximum tokens per chunk | +| `merge_peers` | `true` | Merge sibling elements under the same heading | +| `repeat_table_header` | `true` | Repeat table headers when a table is split across chunks | + +Each chunk includes: + +- **text** — the chunk content +- **headings** — heading hierarchy leading to the chunk +- **source_page** — the page number the chunk originates from +- **token_count** — number of tokens in the chunk +- **bboxes** — bounding boxes of the chunk's source elements (page + coordinates) + ## Configuration All configuration is done via environment variables: | Variable | Default | Description | |----------|---------|-------------| +| `CONVERSION_ENGINE` | `local` | `local` (in-process Docling) or `remote` (Docling Serve) | +| `DOCLING_SERVE_URL` | `http://localhost:5001` | Docling Serve endpoint (remote mode only) | +| `DOCLING_SERVE_API_KEY` | — | API key for Docling Serve (optional) | | `CORS_ORIGINS` | `http://localhost:3000,...` | CORS allowed origins | | `UPLOAD_DIR` | `./uploads` | File storage directory | | `DB_PATH` | `./data/docling_studio.db` | SQLite database path | | `CONVERSION_TIMEOUT` | `600` | Max seconds per Docling conversion | +| `MAX_CONCURRENT_ANALYSES` | `3` | Maximum parallel analysis jobs | +| `DEPLOYMENT_MODE` | `self-hosted` | `self-hosted` or `huggingface` (shows disclaimer banner) | +| `APP_VERSION` | `dev` | Application version (set automatically by CI/Docker) | ## System Requirements -| Resource | Minimum | Recommended | -|----------|---------|-------------| -| Memory | 6 GB | 8 GB+ | -| CPUs | 4 | 8+ | +| | Remote image | Local image | +|---|---|---| +| **Image size** | ~270 MB | ~1.9 GB | +| **Memory** | 2 GB | 6 GB (recommended 8 GB+) | +| **CPUs** | 2 | 4 (recommended 8+) | All Docker images are multi-arch (`linux/amd64` + `linux/arm64`). No GPU required. diff --git a/docs/images/archi.png b/docs/images/archi.png deleted file mode 100644 index f81569d..0000000 Binary files a/docs/images/archi.png and /dev/null differ diff --git a/docs/images/docker.png b/docs/images/docker.png new file mode 100644 index 0000000..67ede30 Binary files /dev/null and b/docs/images/docker.png differ diff --git a/docs/images/global.png b/docs/images/global.png new file mode 100644 index 0000000..8fa4299 Binary files /dev/null and b/docs/images/global.png differ diff --git a/docs/images/logo.png b/docs/images/logo.png new file mode 100644 index 0000000..bcb4a5a Binary files /dev/null and b/docs/images/logo.png differ diff --git a/docs/index.md b/docs/index.md index 1da8358..b476021 100644 --- a/docs/index.md +++ b/docs/index.md @@ -4,18 +4,23 @@ A visual document analysis studio powered by [Docling](https://github.com/DS4SD/ Upload a PDF, configure the extraction pipeline, and visualize the results — text, tables, images, formulas, bounding boxes — all from your browser. -{ width="600" } +{ width="600" }  ## Features - **PDF viewer** with page navigation, bounding box overlay, and resizable results panel -- **Configurable Docling pipeline** — OCR, table extraction, code/formula enrichment, picture classification & description +- **Configurable Docling pipeline** — OCR, table extraction, code/formula enrichment, picture classification & description, image generation - **Bounding box visualization** — color-coded element overlay directly on the PDF +- **Chunking** — split extracted content into semantic chunks (hierarchical, hybrid, or page-based) with configurable token limits - **Markdown & HTML export** of extracted content - **Document management** — upload, list, delete - **Analysis history** — re-visit and open past analyses +- **Feature flags** — capabilities adapt to the conversion engine (local vs remote) +- **Rate limiting** — 60 requests per minute per IP to protect the backend +- **Deployment modes** — self-hosted (default) or HuggingFace Spaces (with disclaimer banner) +- **Health endpoint** — `/api/health` reports engine type, deployment mode, and database status - **Dark / Light theme** and **FR / EN** localization ## Tech Stack @@ -31,7 +36,7 @@ Upload a PDF, configure the extraction pipeline, and visualize the results — t ```bash # Docker (fastest) -docker run -p 3000:3000 ghcr.io/scub-france/docling-studio:latest +docker run -p 3000:3000 ghcr.io/scub-france/docling-studio:latest-local ``` Open [http://localhost:3000](http://localhost:3000) and upload a PDF. diff --git a/document-parser/Dockerfile b/document-parser/Dockerfile index 76cc697..3c21f9a 100644 --- a/document-parser/Dockerfile +++ b/document-parser/Dockerfile @@ -1,9 +1,17 @@ -FROM python:3.12-slim +# syntax=docker/dockerfile:1 +# ============================================================================= +# Docling Studio — backend image (multi-target: remote / local) +# +# Usage: +# docker build --target remote -t docling-studio-backend:remote . +# docker build --target local -t docling-studio-backend:local . +# ============================================================================= + +# --- Base: common deps for both targets --- +FROM python:3.12-slim AS base RUN apt-get update && apt-get install -y --no-install-recommends \ poppler-utils \ - libgl1 \ - libglib2.0-0 \ && rm -rf /var/lib/apt/lists/* WORKDIR /app @@ -24,3 +32,25 @@ ENV DB_PATH=/app/data/docling_studio.db USER appuser CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"] + +# --- Remote: lightweight, delegates to Docling Serve --- +FROM base AS remote +ENV CONVERSION_ENGINE=remote + +# --- Local: full Docling in-process --- +FROM base AS local + +USER root +RUN apt-get update && apt-get install -y --no-install-recommends \ + libgl1 \ + libglib2.0-0 \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements-local.txt . +RUN pip install --no-cache-dir torch torchvision --index-url https://download.pytorch.org/whl/cpu \ + && pip install --no-cache-dir -r requirements-local.txt + +RUN chown -R appuser:appuser /app \ + && chown -R appuser:appuser /usr/local/lib/python3.12/site-packages/rapidocr/models +USER appuser +ENV CONVERSION_ENGINE=local diff --git a/document-parser/api/analyses.py b/document-parser/api/analyses.py index 3d91805..8e6112b 100644 --- a/document-parser/api/analyses.py +++ b/document-parser/api/analyses.py @@ -3,16 +3,30 @@ from __future__ import annotations import logging +from typing import Annotated -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request -from api.schemas import AnalysisResponse, CreateAnalysisRequest -from services import analysis_service +from api.schemas import ( + AnalysisResponse, + ChunkBboxResponse, + ChunkResponse, + CreateAnalysisRequest, + RechunkRequest, +) +from services.analysis_service import AnalysisService logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/analyses", tags=["analyses"]) +def _get_service(request: Request) -> AnalysisService: + return request.app.state.analysis_service + + +ServiceDep = Annotated[AnalysisService, Depends(_get_service)] + + def _to_response(job) -> AnalysisResponse: return AnalysisResponse( id=job.id, @@ -22,6 +36,8 @@ def _to_response(job) -> AnalysisResponse: content_markdown=job.content_markdown, content_html=job.content_html, pages_json=job.pages_json, + chunks_json=job.chunks_json, + has_document_json=job.document_json is not None, error_message=job.error_message, started_at=str(job.started_at) if job.started_at else None, completed_at=str(job.completed_at) if job.completed_at else None, @@ -30,7 +46,7 @@ def _to_response(job) -> AnalysisResponse: @router.post("", response_model=AnalysisResponse) -async def create_analysis(body: CreateAnalysisRequest): +async def create_analysis(body: CreateAnalysisRequest, service: ServiceDep) -> AnalysisResponse: """Create a new analysis job for a document.""" if not body.documentId or not body.documentId.strip(): raise HTTPException(status_code=400, detail="documentId is required") @@ -39,8 +55,16 @@ async def create_analysis(body: CreateAnalysisRequest): if body.pipelineOptions: pipeline_opts = body.pipelineOptions.model_dump() + chunking_opts = None + if body.chunkingOptions: + chunking_opts = body.chunkingOptions.model_dump() + try: - job = await analysis_service.create(body.documentId, pipeline_options=pipeline_opts) + job = await service.create( + body.documentId, + pipeline_options=pipeline_opts, + chunking_options=chunking_opts, + ) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) from e @@ -48,24 +72,45 @@ async def create_analysis(body: CreateAnalysisRequest): @router.get("", response_model=list[AnalysisResponse]) -async def list_analyses(): +async def list_analyses(service: ServiceDep) -> list[AnalysisResponse]: """List all analysis jobs.""" - jobs = await analysis_service.find_all() + jobs = await 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): +async def get_analysis(job_id: str, service: ServiceDep) -> AnalysisResponse: """Get a single analysis job.""" - job = await analysis_service.find_by_id(job_id) + job = await service.find_by_id(job_id) if not job: raise HTTPException(status_code=404, detail="Analysis not found") return _to_response(job) +@router.post("/{job_id}/rechunk", response_model=list[ChunkResponse]) +async def rechunk_analysis( + job_id: str, body: RechunkRequest, service: ServiceDep +) -> list[ChunkResponse]: + """Re-chunk a completed analysis with new chunking options.""" + try: + chunks = await service.rechunk(job_id, body.chunkingOptions.model_dump()) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + return [ + ChunkResponse( + text=c.text, + headings=c.headings, + source_page=c.source_page, + token_count=c.token_count, + bboxes=[ChunkBboxResponse(page=b.page, bbox=b.bbox) for b in c.bboxes], + ) + for c in chunks + ] + + @router.delete("/{job_id}", status_code=204) -async def delete_analysis(job_id: str): +async def delete_analysis(job_id: str, service: ServiceDep) -> None: """Delete an analysis job.""" - deleted = await analysis_service.delete(job_id) + deleted = await service.delete(job_id) if not deleted: raise HTTPException(status_code=404, detail="Analysis not found") diff --git a/document-parser/api/documents.py b/document-parser/api/documents.py index 183546b..76b5c9f 100644 --- a/document-parser/api/documents.py +++ b/document-parser/api/documents.py @@ -13,6 +13,8 @@ from services import document_service logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/documents", tags=["documents"]) +_READ_CHUNK_SIZE = 64 * 1024 # 64 KB + def _to_response(doc) -> DocumentResponse: return DocumentResponse( @@ -25,8 +27,8 @@ def _to_response(doc) -> DocumentResponse: ) -@router.post("/upload", response_model=DocumentResponse) -async def upload(file: UploadFile): +@router.post("/upload", response_model=DocumentResponse, status_code=200) +async def upload(file: UploadFile) -> DocumentResponse: """Upload a PDF document.""" if not file.filename: raise HTTPException(status_code=400, detail="No filename provided") @@ -35,7 +37,15 @@ async def upload(file: UploadFile): if file.size and file.size > document_service.MAX_FILE_SIZE: raise HTTPException(status_code=413, detail="File too large (max 50 MB)") - content = await file.read() + # Read in chunks to avoid holding the full upload in a single allocation + chunks: list[bytes] = [] + total = 0 + while chunk := await file.read(_READ_CHUNK_SIZE): + total += len(chunk) + if total > document_service.MAX_FILE_SIZE: + raise HTTPException(status_code=413, detail="File too large (max 50 MB)") + chunks.append(chunk) + content = b"".join(chunks) try: doc = await document_service.upload( @@ -44,20 +54,20 @@ async def upload(file: UploadFile): file_content=content, ) except ValueError as e: - raise HTTPException(status_code=413, detail=str(e)) from e + raise HTTPException(status_code=400, detail=str(e)) from e return _to_response(doc) @router.get("", response_model=list[DocumentResponse]) -async def list_documents(): +async def list_documents() -> list[DocumentResponse]: """List all documents.""" docs = await document_service.find_all() return [_to_response(d) for d in docs] @router.get("/{doc_id}", response_model=DocumentResponse) -async def get_document(doc_id: str): +async def get_document(doc_id: str) -> DocumentResponse: """Get a single document.""" doc = await document_service.find_by_id(doc_id) if not doc: @@ -66,7 +76,7 @@ async def get_document(doc_id: str): @router.delete("/{doc_id}", status_code=204) -async def delete_document(doc_id: str): +async def delete_document(doc_id: str) -> None: """Delete a document and its file.""" deleted = await document_service.delete(doc_id) if not deleted: @@ -78,12 +88,18 @@ async def preview( doc_id: str, page: int = Query(1, ge=1), dpi: int = Query(150, ge=72, le=300), -): +) -> Response: """Generate a PNG preview of a specific PDF page.""" doc = await document_service.find_by_id(doc_id) if not doc: raise HTTPException(status_code=404, detail="Document not found") + if doc.page_count and page > doc.page_count: + raise HTTPException( + status_code=400, + detail=f"Page {page} out of range (document has {doc.page_count} pages)", + ) + try: with open(doc.storage_path, "rb") as f: file_content = f.read() @@ -91,6 +107,11 @@ async def preview( return Response(content=png_bytes, media_type="image/png") except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) from e + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail="PDF file not found on disk") from exc + except OSError as exc: + logger.exception("I/O error generating preview for %s", doc_id) + raise HTTPException(status_code=422, detail="Failed to read PDF file") from exc except Exception as exc: - logger.exception("Failed to generate preview") + logger.exception("Unexpected error generating preview for %s", doc_id) raise HTTPException(status_code=422, detail="Failed to generate preview") from exc diff --git a/document-parser/api/schemas.py b/document-parser/api/schemas.py index 4e30153..ee602c6 100644 --- a/document-parser/api/schemas.py +++ b/document-parser/api/schemas.py @@ -8,7 +8,7 @@ from __future__ import annotations from datetime import datetime -from pydantic import BaseModel, ConfigDict, field_validator +from pydantic import AliasChoices, BaseModel, ConfigDict, Field, field_validator def _to_camel(name: str) -> str: @@ -18,6 +18,7 @@ def _to_camel(name: str) -> str: class _CamelModel(BaseModel): """Base model that serializes field names to camelCase.""" + model_config = ConfigDict( alias_generator=_to_camel, populate_by_name=True, @@ -28,6 +29,7 @@ class _CamelModel(BaseModel): class DocumentResponse(_CamelModel): id: str filename: str + status: str = "uploaded" # Document status (always "uploaded" for now) content_type: str | None = None file_size: int | None = None page_count: int | None = None @@ -42,6 +44,8 @@ class AnalysisResponse(_CamelModel): content_markdown: str | None = None content_html: str | None = None pages_json: str | None = None + chunks_json: str | None = None + has_document_json: bool = False error_message: str | None = None started_at: str | datetime | None = None completed_at: str | datetime | None = None @@ -50,16 +54,40 @@ class AnalysisResponse(_CamelModel): class PipelineOptionsRequest(BaseModel): """Docling pipeline configuration options.""" - do_ocr: bool = True - do_table_structure: bool = True - table_mode: str = "accurate" # "accurate" or "fast" - 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 + + model_config = ConfigDict(populate_by_name=True) + + do_ocr: bool = Field(default=True, validation_alias=AliasChoices("do_ocr", "doOcr")) + do_table_structure: bool = Field( + default=True, validation_alias=AliasChoices("do_table_structure", "doTableStructure") + ) + table_mode: str = Field( + default="accurate", validation_alias=AliasChoices("table_mode", "tableMode") + ) + do_code_enrichment: bool = Field( + default=False, validation_alias=AliasChoices("do_code_enrichment", "doCodeEnrichment") + ) + do_formula_enrichment: bool = Field( + default=False, validation_alias=AliasChoices("do_formula_enrichment", "doFormulaEnrichment") + ) + do_picture_classification: bool = Field( + default=False, + validation_alias=AliasChoices("do_picture_classification", "doPictureClassification"), + ) + do_picture_description: bool = Field( + default=False, + validation_alias=AliasChoices("do_picture_description", "doPictureDescription"), + ) + generate_picture_images: bool = Field( + default=False, + validation_alias=AliasChoices("generate_picture_images", "generatePictureImages"), + ) + generate_page_images: bool = Field( + default=False, validation_alias=AliasChoices("generate_page_images", "generatePageImages") + ) + images_scale: float = Field( + default=1.0, validation_alias=AliasChoices("images_scale", "imagesScale") + ) @field_validator("table_mode") @classmethod @@ -76,6 +104,61 @@ class PipelineOptionsRequest(BaseModel): return v +class ChunkingOptionsRequest(BaseModel): + """Docling chunking configuration options.""" + + model_config = ConfigDict(populate_by_name=True) + + chunker_type: str = Field( + default="hybrid", validation_alias=AliasChoices("chunker_type", "chunkerType") + ) + max_tokens: int = Field(default=512, validation_alias=AliasChoices("max_tokens", "maxTokens")) + merge_peers: bool = Field( + default=True, validation_alias=AliasChoices("merge_peers", "mergePeers") + ) + repeat_table_header: bool = Field( + default=True, validation_alias=AliasChoices("repeat_table_header", "repeatTableHeader") + ) + + @field_validator("chunker_type") + @classmethod + def validate_chunker_type(cls, v: str) -> str: + if v not in ("hybrid", "hierarchical"): + raise ValueError('chunker_type must be "hybrid" or "hierarchical"') + return v + + @field_validator("max_tokens") + @classmethod + def validate_max_tokens(cls, v: int) -> int: + if v < 64 or v > 8192: + raise ValueError("max_tokens must be between 64 and 8192") + return v + + +class ChunkBboxResponse(_CamelModel): + page: int + bbox: list[float] + + +class ChunkResponse(_CamelModel): + text: str + headings: list[str] = [] + source_page: int | None = None + token_count: int = 0 + bboxes: list[ChunkBboxResponse] = [] + + class CreateAnalysisRequest(BaseModel): - documentId: str # camelCase to match existing frontend contract - pipelineOptions: PipelineOptionsRequest | None = None + documentId: str = Field(validation_alias=AliasChoices("documentId", "document_id")) + pipelineOptions: PipelineOptionsRequest | None = Field( + default=None, validation_alias=AliasChoices("pipelineOptions", "pipeline_options") + ) + chunkingOptions: ChunkingOptionsRequest | None = Field( + default=None, validation_alias=AliasChoices("chunkingOptions", "chunking_options") + ) + + +class RechunkRequest(BaseModel): + chunkingOptions: ChunkingOptionsRequest = Field( + validation_alias=AliasChoices("chunkingOptions", "chunking_options") + ) diff --git a/document-parser/conftest.py b/document-parser/conftest.py index 60a2633..ddf789c 100644 --- a/document-parser/conftest.py +++ b/document-parser/conftest.py @@ -1,3 +1 @@ - - pytest_plugins = ["pytest_asyncio"] diff --git a/document-parser/domain/models.py b/document-parser/domain/models.py index 3df9eb3..3faec9a 100644 --- a/document-parser/domain/models.py +++ b/document-parser/domain/models.py @@ -42,6 +42,8 @@ class AnalysisJob: content_markdown: str | None = None content_html: str | None = None pages_json: str | None = None + document_json: str | None = None + chunks_json: str | None = None error_message: str | None = None started_at: datetime | None = None completed_at: datetime | None = None @@ -51,19 +53,29 @@ class AnalysisJob: document_filename: str | None = None def mark_running(self) -> None: + """Transition to RUNNING and record the start timestamp.""" self.status = AnalysisStatus.RUNNING self.started_at = _utcnow() def mark_completed( - self, markdown: str, html: str, pages_json: str, + self, + markdown: str, + html: str, + pages_json: str, + document_json: str | None = None, + chunks_json: str | None = None, ) -> None: + """Transition to COMPLETED with conversion results.""" self.status = AnalysisStatus.COMPLETED self.content_markdown = markdown self.content_html = html self.pages_json = pages_json + self.document_json = document_json + self.chunks_json = chunks_json self.completed_at = _utcnow() def mark_failed(self, error: str) -> None: + """Transition to FAILED with an error message.""" self.status = AnalysisStatus.FAILED self.error_message = error self.completed_at = _utcnow() diff --git a/document-parser/domain/ports.py b/document-parser/domain/ports.py new file mode 100644 index 0000000..f4ea64d --- /dev/null +++ b/document-parser/domain/ports.py @@ -0,0 +1,44 @@ +"""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 TYPE_CHECKING, Protocol + +if TYPE_CHECKING: + from domain.value_objects import ( + ChunkingOptions, + ChunkResult, + 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: ... + + +class DocumentChunker(Protocol): + """Port for document chunking. + + Takes a serialized DoclingDocument (JSON) and returns chunks. + """ + + async def chunk( + self, + document_json: str, + options: ChunkingOptions, + ) -> list[ChunkResult]: ... diff --git a/document-parser/domain/value_objects.py b/document-parser/domain/value_objects.py new file mode 100644 index 0000000..7c91845 --- /dev/null +++ b/document-parser/domain/value_objects.py @@ -0,0 +1,80 @@ +"""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 True if all options match their defaults.""" + return self == ConversionOptions() + + +@dataclass +class ConversionResult: + page_count: int + content_markdown: str + content_html: str + pages: list[PageDetail] + skipped_items: int = 0 + document_json: str | None = None + + +@dataclass +class ChunkingOptions: + chunker_type: str = "hybrid" # "hybrid", "hierarchical", "page" + max_tokens: int = 512 + merge_peers: bool = True + repeat_table_header: bool = True + + def is_default(self) -> bool: + """Return True if all options match their defaults.""" + return self == ChunkingOptions() + + +@dataclass +class ChunkBbox: + page: int + bbox: list[float] # [left, top, right, bottom] in TOPLEFT origin + + +@dataclass +class ChunkResult: + text: str + headings: list[str] = field(default_factory=list) + source_page: int | None = None + token_count: int = 0 + bboxes: list[ChunkBbox] = field(default_factory=list) diff --git a/document-parser/infra/__init__.py b/document-parser/infra/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/document-parser/domain/bbox.py b/document-parser/infra/bbox.py similarity index 94% rename from document-parser/domain/bbox.py rename to document-parser/infra/bbox.py index 41530a7..4804df9 100644 --- a/document-parser/domain/bbox.py +++ b/document-parser/infra/bbox.py @@ -43,7 +43,11 @@ def to_topleft_list(bbox: BoundingBox, page_height: float) -> list[float]: if right <= left or bottom <= top: logger.debug( "Degenerate bbox skipped: [%.1f, %.1f, %.1f, %.1f] (page_height=%.1f)", - left, top, right, bottom, page_height, + left, + top, + right, + bottom, + page_height, ) return list(EMPTY_BBOX) diff --git a/document-parser/infra/local_chunker.py b/document-parser/infra/local_chunker.py new file mode 100644 index 0000000..670f5d0 --- /dev/null +++ b/document-parser/infra/local_chunker.py @@ -0,0 +1,96 @@ +"""Local Docling chunker — runs chunking in-process using docling-core. + +This adapter implements the DocumentChunker port. It deserializes a +DoclingDocument from JSON, applies the requested chunker, and returns +domain ChunkResult objects. +""" + +from __future__ import annotations + +import asyncio +import json +import logging + +from docling_core.transforms.chunker import HierarchicalChunker +from docling_core.transforms.chunker.hybrid_chunker import HybridChunker +from docling_core.types.doc.document import DoclingDocument + +from domain.value_objects import ChunkBbox, ChunkingOptions, ChunkResult +from infra.bbox import EMPTY_BBOX, to_topleft_list + +logger = logging.getLogger(__name__) + + +def _chunk_sync(document_json: str, options: ChunkingOptions) -> list[ChunkResult]: + if not document_json or not document_json.strip(): + raise ValueError("Empty document JSON — nothing to chunk") + + try: + doc_data = json.loads(document_json) + except json.JSONDecodeError as e: + raise ValueError(f"Malformed document JSON: {e}") from e + + doc = DoclingDocument.model_validate(doc_data) + + chunker = _build_chunker(options) + results: list[ChunkResult] = [] + + for chunk in chunker.chunk(doc): + source_page = None + token_count = 0 + bboxes: list[ChunkBbox] = [] + + if hasattr(chunk, "meta") and chunk.meta and chunk.meta.doc_items: + for doc_item in chunk.meta.doc_items: + if not hasattr(doc_item, "prov") or not doc_item.prov: + continue + for prov in doc_item.prov: + page_no = prov.page_no + if source_page is None: + source_page = page_no + if prov.bbox: + page_obj = doc.pages.get(page_no) + if page_obj: + bbox = to_topleft_list(prov.bbox, page_obj.size.height) + if bbox != EMPTY_BBOX: + bboxes.append(ChunkBbox(page=page_no, bbox=bbox)) + + if hasattr(chunker, "tokenizer") and chunker.tokenizer: + token_count = chunker.tokenizer.count_tokens(chunk.text) + + headings = list(chunk.meta.headings) if chunk.meta and chunk.meta.headings else [] + + results.append( + ChunkResult( + text=chunk.text, + headings=headings, + source_page=source_page, + token_count=token_count, + bboxes=bboxes, + ) + ) + + logger.info("Chunked document into %d chunks (chunker=%s)", len(results), options.chunker_type) + return results + + +def _build_chunker(options: ChunkingOptions) -> HierarchicalChunker | HybridChunker: + if options.chunker_type == "hierarchical": + return HierarchicalChunker() + + return HybridChunker( + max_tokens=options.max_tokens, + merge_peers=options.merge_peers, + repeat_table_header=options.repeat_table_header, + ) + + +class LocalChunker: + """Adapter that runs docling-core chunking locally.""" + + async def chunk( + self, + document_json: str, + options: ChunkingOptions, + ) -> list[ChunkResult]: + return await asyncio.to_thread(_chunk_sync, document_json, options) diff --git a/document-parser/domain/parsing.py b/document-parser/infra/local_converter.py similarity index 52% rename from document-parser/domain/parsing.py rename to document-parser/infra/local_converter.py index 7e7b4bb..ba8875d 100644 --- a/document-parser/domain/parsing.py +++ b/document-parser/infra/local_converter.py @@ -1,15 +1,17 @@ -"""Docling document extraction logic — pure domain, no HTTP concerns. +"""Local Docling converter — runs Docling as a Python library in-process. -Wraps the Docling library to convert documents and extract structured -per-page elements with bounding boxes and hierarchy levels. +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 json import logging import threading -from dataclasses import dataclass, field from docling.datamodel.base_models import InputFormat from docling.datamodel.pipeline_options import ( @@ -17,7 +19,8 @@ from docling.datamodel.pipeline_options import ( TableFormerMode, TableStructureOptions, ) -from docling.document_converter import DocumentConverter, PdfFormatOption +from docling.document_converter import DocumentConverter as DoclingConverter +from docling.document_converter import PdfFormatOption from docling_core.types.doc import ( CodeItem, DocItem, @@ -32,11 +35,17 @@ from docling_core.types.doc import ( TitleItem, ) -from domain.bbox import to_topleft_list +from domain.value_objects import ( + ConversionOptions, + ConversionResult, + PageDetail, + PageElement, +) +from infra.bbox import to_topleft_list logger = logging.getLogger(__name__) -# Thread lock — DocumentConverter is not thread-safe +# Thread lock — DoclingConverter is not thread-safe _converter_lock = threading.Lock() # US Letter page dimensions (points) — fallback when page size is unknown @@ -44,61 +53,13 @@ _DEFAULT_PAGE_WIDTH = 612.0 _DEFAULT_PAGE_HEIGHT = 792.0 # Default converter (lazy-init on first request) -_default_converter: DocumentConverter | None = None - - -# --------------------------------------------------------------------------- -# 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 +_default_converter: DoclingConverter | None = None # --------------------------------------------------------------------------- # 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"), @@ -113,7 +74,6 @@ _ELEMENT_TYPE_MAP: list[tuple[type, str]] = [ 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 @@ -124,51 +84,52 @@ def _get_element_type(item: DocItem) -> str: # Pipeline factory # --------------------------------------------------------------------------- -def build_converter(options: ConversionOptions | None = None) -> DocumentConverter: - """Build a DocumentConverter with the given pipeline options.""" - opts = options or ConversionOptions() +def _build_docling_converter(options: ConversionOptions) -> DoclingConverter: table_options = TableStructureOptions( do_cell_matching=True, - mode=TableFormerMode.ACCURATE if opts.table_mode == "accurate" else TableFormerMode.FAST, + mode=TableFormerMode.ACCURATE if options.table_mode == "accurate" else TableFormerMode.FAST, ) pipeline_options = PdfPipelineOptions( - do_ocr=opts.do_ocr, - do_table_structure=opts.do_table_structure, + do_ocr=options.do_ocr, + do_table_structure=options.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, + 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 DocumentConverter( + return DoclingConverter( format_options={ InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options), } ) -def get_default_converter() -> DocumentConverter: +def _get_default_converter() -> DoclingConverter: global _default_converter if _default_converter is None: - _default_converter = build_converter() + _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]: - """Extract per-page element details with bounding boxes from Docling result. - Returns (pages, skipped_count) for transparent error reporting. - """ +def _extract_pages_detail(doc_result) -> tuple[list[PageDetail], int]: pages: dict[int, PageDetail] = {} document = doc_result.document skipped = 0 @@ -191,9 +152,10 @@ def extract_pages_detail(doc_result) -> tuple[list[PageDetail], int]: def _process_content_item( - item: DocItem | GroupItem, level: int, pages: dict[int, PageDetail], + 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 @@ -204,14 +166,15 @@ def _process_content_item( 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, + page_no, + _DEFAULT_PAGE_WIDTH, + _DEFAULT_PAGE_HEIGHT, + ) + pages[page_no] = PageDetail( + page_number=page_no, width=_DEFAULT_PAGE_WIDTH, height=_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 @@ -242,49 +205,30 @@ def _process_content_item( # --------------------------------------------------------------------------- -# Main conversion entry point +# Synchronous conversion (called via asyncio.to_thread) # --------------------------------------------------------------------------- -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() +def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult: with _converter_lock: - conv = _select_converter(opts) + conv = _select_converter(options) result = conv.convert(file_path) doc = result.document page_count = len(doc.pages) - pages_detail, skipped = extract_pages_detail(result) + pages_detail, skipped = _extract_pages_detail(result) if not pages_detail and page_count > 0: - pages_detail = _build_fallback_pages(doc, page_count) + 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) @@ -295,4 +239,21 @@ def convert_document( content_html=doc.export_to_html(), pages=pages_detail, skipped_items=skipped, + document_json=json.dumps(doc.export_to_dict()), ) + + +# --------------------------------------------------------------------------- +# 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) diff --git a/document-parser/infra/rate_limiter.py b/document-parser/infra/rate_limiter.py new file mode 100644 index 0000000..bb82ee7 --- /dev/null +++ b/document-parser/infra/rate_limiter.py @@ -0,0 +1,89 @@ +"""Lightweight in-memory rate limiter middleware for FastAPI. + +Uses a sliding-window counter per client IP. No external dependency +required — suitable for single-process deployments with SQLite. + +For multi-process or distributed setups, replace with a Redis-backed +solution (e.g. slowapi). +""" + +from __future__ import annotations + +import logging +import time +from collections import defaultdict +from dataclasses import dataclass, field +from typing import TYPE_CHECKING + +from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint +from starlette.responses import JSONResponse, Response + +if TYPE_CHECKING: + from starlette.requests import Request + +logger = logging.getLogger(__name__) + + +@dataclass +class _ClientBucket: + """Sliding window of request timestamps for a single client.""" + + timestamps: list[float] = field(default_factory=list) + + def count_recent(self, window: float, now: float) -> int: + """Remove expired entries and return the count of recent requests.""" + cutoff = now - window + self.timestamps = [t for t in self.timestamps if t > cutoff] + return len(self.timestamps) + + def add(self, now: float) -> None: + self.timestamps.append(now) + + +class RateLimiterMiddleware(BaseHTTPMiddleware): + """Per-IP rate limiter using in-memory sliding windows. + + Args: + app: The ASGI application. + requests_per_window: Max requests allowed per window. + window_seconds: Size of the sliding window in seconds. + exclude_paths: Paths exempt from rate limiting (e.g. health checks). + """ + + def __init__( + self, + app, + *, + requests_per_window: int = 60, + window_seconds: float = 60.0, + exclude_paths: tuple[str, ...] = ("/api/health",), + ): + super().__init__(app) + self._max_requests = requests_per_window + self._window = window_seconds + self._exclude = exclude_paths + self._buckets: dict[str, _ClientBucket] = defaultdict(_ClientBucket) + + async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response: + if request.url.path in self._exclude: + return await call_next(request) + + client_ip = request.client.host if request.client else "unknown" + now = time.monotonic() + + bucket = self._buckets[client_ip] + recent = bucket.count_recent(self._window, now) + + if recent >= self._max_requests: + retry_after = int(self._window) + logger.warning( + "Rate limit exceeded for %s (%d/%d)", client_ip, recent, self._max_requests + ) + return JSONResponse( + status_code=429, + content={"detail": "Too many requests"}, + headers={"Retry-After": str(retry_after)}, + ) + + bucket.add(now) + return await call_next(request) diff --git a/document-parser/infra/serve_converter.py b/document-parser/infra/serve_converter.py new file mode 100644 index 0000000..a6a0502 --- /dev/null +++ b/document-parser/infra/serve_converter.py @@ -0,0 +1,245 @@ +"""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 docling_core.types.doc.base import BoundingBox, CoordOrigin + +from domain.value_objects import ( + ConversionOptions, + ConversionResult, + PageDetail, + PageElement, +) +from infra.bbox import to_topleft_list + +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(), + "generate_page_images": str(options.generate_page_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 + + document_json = json.dumps(json_content) if json_content else None + + return ConversionResult( + page_count=page_count, + content_markdown=content_md, + content_html=content_html, + pages=pages, + document_json=document_json, + ) + + +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. + + Delegates to the canonical to_topleft_list function via a docling-core + BoundingBox, ensuring consistent coordinate handling across all converters. + """ + if not isinstance(bbox_data, dict): + return [0.0, 0.0, 0.0, 0.0] + + origin_str = bbox_data.get("coord_origin", "TOPLEFT") + origin = CoordOrigin.BOTTOMLEFT if origin_str == "BOTTOMLEFT" else CoordOrigin.TOPLEFT + + bbox = BoundingBox( + l=bbox_data.get("l", 0.0), + t=bbox_data.get("t", 0.0), + r=bbox_data.get("r", 0.0), + b=bbox_data.get("b", 0.0), + coord_origin=origin, + ) + return to_topleft_list(bbox, page_height) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py new file mode 100644 index 0000000..4f15384 --- /dev/null +++ b/document-parser/infra/settings.py @@ -0,0 +1,43 @@ +"""Centralized application settings — loaded from environment variables.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Settings: + app_version: str = "dev" + conversion_engine: str = "local" # "local" or "remote" + deployment_mode: str = "self-hosted" # "self-hosted" or "huggingface" + docling_serve_url: str = "http://localhost:5001" + docling_serve_api_key: str | None = None + conversion_timeout: int = 600 + max_concurrent_analyses: int = 3 + 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: + """Build a Settings instance from environment variables.""" + cors_raw = os.environ.get("CORS_ORIGINS", "http://localhost:3000,http://localhost:5173") + return cls( + app_version=os.environ.get("APP_VERSION", "dev"), + conversion_engine=os.environ.get("CONVERSION_ENGINE", "local"), + deployment_mode=os.environ.get("DEPLOYMENT_MODE", "self-hosted"), + 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")), + max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")), + 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(",")], + ) + + +# Module-level singleton — import this from other modules. +settings = Settings.from_env() diff --git a/document-parser/main.py b/document-parser/main.py index 7e29776..957cada 100644 --- a/document-parser/main.py +++ b/document-parser/main.py @@ -1,14 +1,18 @@ """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 collections.abc import AsyncIterator from contextlib import asynccontextmanager from fastapi import FastAPI @@ -16,7 +20,10 @@ from fastapi.middleware.cors import CORSMiddleware from api.analyses import router as analyses_router from api.documents import router as documents_router -from persistence.database import init_db +from infra.rate_limiter import RateLimiterMiddleware +from infra.settings import settings +from persistence.database import get_connection, init_db +from services.analysis_service import AnalysisService logging.basicConfig( level=logging.INFO, @@ -25,11 +32,53 @@ logging.basicConfig( logger = logging.getLogger(__name__) +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_chunker(): + """Build the chunker adapter — only available in local mode.""" + if settings.conversion_engine == "local": + from infra.local_chunker import LocalChunker + + return LocalChunker() + return None + + +def _build_analysis_service() -> AnalysisService: + converter = _build_converter() + chunker = _build_chunker() + return AnalysisService( + converter=converter, + chunker=chunker, + conversion_timeout=settings.conversion_timeout, + max_concurrent=settings.max_concurrent_analyses, + ) + + +# --------------------------------------------------------------------------- +# FastAPI app +# --------------------------------------------------------------------------- + + @asynccontextmanager -async def lifespan(app: FastAPI): - """Startup: initialize database. Shutdown: nothing special needed.""" +async def lifespan(app: FastAPI) -> AsyncIterator[None]: await init_db() - logger.info("Docling Studio backend ready") + app.state.analysis_service = _build_analysis_service() + logger.info("Docling Studio backend ready (engine=%s)", settings.conversion_engine) yield @@ -39,25 +88,35 @@ 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"], ) +app.add_middleware(RateLimiterMiddleware, requests_per_window=100, window_seconds=60) -# Mount routers app.include_router(documents_router) app.include_router(analyses_router) -@app.get("/health") -def health(): - """Health check endpoint.""" - return {"status": "ok"} +@app.get("/api/health") +async def health() -> dict[str, str]: + """Health check endpoint — verifies database connectivity.""" + db_status = "ok" + try: + async with get_connection() as db: + await db.execute("SELECT 1") + except Exception: + db_status = "error" + logger.warning("Health check: database unreachable", exc_info=True) + + status = "ok" if db_status == "ok" else "degraded" + return { + "status": status, + "version": settings.app_version, + "engine": settings.conversion_engine, + "deploymentMode": settings.deployment_mode, + "database": db_status, + } diff --git a/document-parser/persistence/analysis_repo.py b/document-parser/persistence/analysis_repo.py index a16367d..deb08b6 100644 --- a/document-parser/persistence/analysis_repo.py +++ b/document-parser/persistence/analysis_repo.py @@ -2,11 +2,21 @@ from __future__ import annotations +from datetime import UTC, datetime + from domain.models import AnalysisJob, AnalysisStatus from persistence.database import get_connection +def _parse_dt(value: str | None) -> datetime | None: + """Parse an ISO-format datetime string back into a datetime object.""" + if not value: + return None + return datetime.fromisoformat(value) + + def _row_to_job(row) -> AnalysisJob: + keys = row.keys() return AnalysisJob( id=row["id"], document_id=row["document_id"], @@ -14,11 +24,13 @@ def _row_to_job(row) -> AnalysisJob: content_markdown=row["content_markdown"], content_html=row["content_html"], pages_json=row["pages_json"], + document_json=row["document_json"] if "document_json" in keys else None, + chunks_json=row["chunks_json"] if "chunks_json" in keys else None, error_message=row["error_message"], - started_at=row["started_at"], - completed_at=row["completed_at"], - created_at=row["created_at"], - document_filename=row["filename"] if "filename" in row.keys() else None, # noqa: SIM118 + started_at=_parse_dt(row["started_at"]), + completed_at=_parse_dt(row["completed_at"]), + created_at=_parse_dt(row["created_at"]) or datetime.now(UTC), + document_filename=row["filename"] if "filename" in keys else None, ) @@ -30,6 +42,7 @@ _SELECT_WITH_DOC = """ async def insert(job: AnalysisJob) -> None: + """Persist a new analysis job record.""" async with get_connection() as db: await db.execute( """INSERT INTO analysis_jobs (id, document_id, status, created_at) @@ -40,6 +53,7 @@ async def insert(job: AnalysisJob) -> None: async def find_all(*, limit: int = 200, offset: int = 0) -> list[AnalysisJob]: + """Return analysis jobs with document info, newest first.""" async with get_connection() as db: cursor = await db.execute( f"{_SELECT_WITH_DOC} ORDER BY aj.created_at DESC LIMIT ? OFFSET ?", @@ -50,31 +64,51 @@ async def find_all(*, limit: int = 200, offset: int = 0) -> list[AnalysisJob]: async def find_by_id(job_id: str) -> AnalysisJob | None: + """Find an analysis job by ID (with document filename), or return None.""" async with get_connection() as db: - cursor = await db.execute( - f"{_SELECT_WITH_DOC} WHERE aj.id = ?", (job_id,) - ) + cursor = await db.execute(f"{_SELECT_WITH_DOC} WHERE aj.id = ?", (job_id,)) row = await cursor.fetchone() return _row_to_job(row) if row else None async def update_status(job: AnalysisJob) -> None: + """Persist all mutable fields of an analysis job (status, results, timestamps).""" async with get_connection() as db: await db.execute( """UPDATE analysis_jobs SET status = ?, content_markdown = ?, content_html = ?, - pages_json = ?, error_message = ?, started_at = ?, completed_at = ? + pages_json = ?, document_json = ?, chunks_json = ?, + error_message = ?, started_at = ?, completed_at = ? WHERE id = ?""", - (job.status.value, job.content_markdown, job.content_html, - job.pages_json, job.error_message, - str(job.started_at) if job.started_at else None, - str(job.completed_at) if job.completed_at else None, - job.id), + ( + job.status.value, + job.content_markdown, + job.content_html, + job.pages_json, + job.document_json, + job.chunks_json, + job.error_message, + str(job.started_at) if job.started_at else None, + str(job.completed_at) if job.completed_at else None, + job.id, + ), ) await db.commit() +async def update_chunks(job_id: str, chunks_json: str) -> bool: + """Update only the chunks_json column for a completed analysis.""" + async with get_connection() as db: + cursor = await db.execute( + "UPDATE analysis_jobs SET chunks_json = ? WHERE id = ?", + (chunks_json, job_id), + ) + await db.commit() + return cursor.rowcount > 0 + + async def delete(job_id: str) -> bool: + """Delete an analysis job by ID. Returns True if a row was removed.""" async with get_connection() as db: cursor = await db.execute("DELETE FROM analysis_jobs WHERE id = ?", (job_id,)) await db.commit() @@ -84,8 +118,6 @@ async def delete(job_id: str) -> bool: async def delete_by_document(document_id: str) -> int: """Delete all analysis jobs for a given document. Returns count deleted.""" async with get_connection() as db: - cursor = await db.execute( - "DELETE FROM analysis_jobs WHERE document_id = ?", (document_id,) - ) + cursor = await db.execute("DELETE FROM analysis_jobs WHERE document_id = ?", (document_id,)) await db.commit() return cursor.rowcount diff --git a/document-parser/persistence/database.py b/document-parser/persistence/database.py index 6aaa37b..10d82d3 100644 --- a/document-parser/persistence/database.py +++ b/document-parser/persistence/database.py @@ -4,13 +4,16 @@ from __future__ import annotations import logging import os +from collections.abc import AsyncIterator from contextlib import asynccontextmanager import aiosqlite +from infra.settings import settings + logger = logging.getLogger(__name__) -DB_PATH = os.environ.get("DB_PATH", "./data/docling_studio.db") +DB_PATH = settings.db_path _SCHEMA = """ CREATE TABLE IF NOT EXISTS documents ( @@ -30,6 +33,8 @@ CREATE TABLE IF NOT EXISTS analysis_jobs ( content_markdown TEXT, content_html TEXT, pages_json TEXT, + document_json TEXT, + chunks_json TEXT, error_message TEXT, started_at TEXT, completed_at TEXT, @@ -42,11 +47,29 @@ CREATE INDEX IF NOT EXISTS idx_documents_created_at ON documents(created_at); """ +_MIGRATIONS = [ + ("document_json", "ALTER TABLE analysis_jobs ADD COLUMN document_json TEXT"), + ("chunks_json", "ALTER TABLE analysis_jobs ADD COLUMN chunks_json TEXT"), +] + + +async def _run_migrations(db: aiosqlite.Connection) -> None: + """Add columns that may be missing in older databases.""" + cursor = await db.execute("PRAGMA table_info(analysis_jobs)") + existing = {row[1] for row in await cursor.fetchall()} + for col_name, ddl in _MIGRATIONS: + if col_name not in existing: + await db.execute(ddl) + logger.info("Migration: added column %s to analysis_jobs", col_name) + await db.commit() + + async def init_db() -> None: """Create database file and tables if they don't exist.""" os.makedirs(os.path.dirname(DB_PATH) or ".", exist_ok=True) async with aiosqlite.connect(DB_PATH) as db: await db.executescript(_SCHEMA) + await _run_migrations(db) await db.commit() logger.info("Database initialized at %s", DB_PATH) @@ -60,7 +83,7 @@ async def get_db() -> aiosqlite.Connection: @asynccontextmanager -async def get_connection(): +async def get_connection() -> AsyncIterator[aiosqlite.Connection]: """Context manager that opens and auto-closes a database connection.""" db = await get_db() try: diff --git a/document-parser/persistence/document_repo.py b/document-parser/persistence/document_repo.py index 3649dc1..3c946de 100644 --- a/document-parser/persistence/document_repo.py +++ b/document-parser/persistence/document_repo.py @@ -2,11 +2,18 @@ from __future__ import annotations +from datetime import UTC, datetime + from domain.models import Document from persistence.database import get_connection def _row_to_document(row) -> Document: + created = row["created_at"] + if isinstance(created, str): + created = datetime.fromisoformat(created) + if created.tzinfo is None: + created = created.replace(tzinfo=UTC) return Document( id=row["id"], filename=row["filename"], @@ -14,22 +21,31 @@ def _row_to_document(row) -> Document: file_size=row["file_size"], page_count=row["page_count"], storage_path=row["storage_path"], - created_at=row["created_at"], + created_at=created, ) async def insert(doc: Document) -> None: + """Persist a new document record.""" async with get_connection() as db: await db.execute( """INSERT INTO documents (id, filename, content_type, file_size, page_count, storage_path, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)""", - (doc.id, doc.filename, doc.content_type, doc.file_size, - doc.page_count, doc.storage_path, str(doc.created_at)), + ( + doc.id, + doc.filename, + doc.content_type, + doc.file_size, + doc.page_count, + doc.storage_path, + str(doc.created_at), + ), ) await db.commit() async def find_all(*, limit: int = 200, offset: int = 0) -> list[Document]: + """Return documents ordered by creation date (newest first).""" async with get_connection() as db: cursor = await db.execute( "SELECT * FROM documents ORDER BY created_at DESC LIMIT ? OFFSET ?", @@ -40,6 +56,7 @@ async def find_all(*, limit: int = 200, offset: int = 0) -> list[Document]: async def find_by_id(doc_id: str) -> Document | None: + """Find a document by its ID, or return None.""" async with get_connection() as db: cursor = await db.execute("SELECT * FROM documents WHERE id = ?", (doc_id,)) row = await cursor.fetchone() @@ -47,6 +64,7 @@ async def find_by_id(doc_id: str) -> Document | None: async def update_page_count(doc_id: str, page_count: int) -> None: + """Update the page count after conversion has determined it.""" async with get_connection() as db: await db.execute( "UPDATE documents SET page_count = ? WHERE id = ?", @@ -56,6 +74,7 @@ async def update_page_count(doc_id: str, page_count: int) -> None: async def delete(doc_id: str) -> bool: + """Delete a document by ID. Returns True if a row was removed.""" async with get_connection() as db: cursor = await db.execute("DELETE FROM documents WHERE id = ?", (doc_id,)) await db.commit() diff --git a/document-parser/requirements-local.txt b/document-parser/requirements-local.txt new file mode 100644 index 0000000..fc2f3c7 --- /dev/null +++ b/document-parser/requirements-local.txt @@ -0,0 +1,2 @@ +-r requirements.txt +docling>=2.80.0,<3.0.0 diff --git a/document-parser/requirements.txt b/document-parser/requirements.txt index 46d50d0..92dcf4c 100644 --- a/document-parser/requirements.txt +++ b/document-parser/requirements.txt @@ -1,4 +1,3 @@ -docling>=2.80.0,<3.0.0 docling-core>=2.0.0,<3.0.0 fastapi>=0.115.0,<1.0.0 uvicorn[standard]>=0.32.0,<1.0.0 @@ -6,3 +5,4 @@ python-multipart>=0.0.12 pdf2image>=1.17.0,<2.0.0 pillow>=10.0.0,<11.0.0 aiosqlite>=0.20.0,<1.0.0 +httpx>=0.27.0,<1.0.0 diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 7a4d71a..93c556f 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -1,108 +1,215 @@ -"""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 import asyncio +import functools import json import logging from dataclasses import asdict +from typing import TYPE_CHECKING -from domain.models import AnalysisJob -from domain.parsing import ConversionOptions, ConversionResult, convert_document +from domain.models import AnalysisJob, AnalysisStatus +from domain.value_objects import ChunkingOptions, ChunkResult, ConversionOptions, ConversionResult + +if TYPE_CHECKING: + from domain.ports import DocumentChunker, DocumentConverter 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")) + +def _chunk_to_dict(c: ChunkResult) -> dict: + """Serialize ChunkResult to a camelCase dict matching the frontend API contract.""" + return { + "text": c.text, + "headings": c.headings, + "sourcePage": c.source_page, + "tokenCount": c.token_count, + "bboxes": [{"page": b.page, "bbox": b.bbox} for b in c.bboxes], + } -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}") - - job = AnalysisJob(document_id=document_id) - job.document_filename = doc.filename - await analysis_repo.insert(job) - - # 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) - - return job +# Maximum number of concurrent analysis jobs to prevent resource exhaustion. +_DEFAULT_MAX_CONCURRENT = 3 -def _on_task_done(task: asyncio.Task) -> None: - """Log unhandled exceptions from background analysis tasks.""" +class AnalysisService: + """Orchestrates document analysis using an injected converter.""" + + def __init__( + self, + converter: DocumentConverter, + chunker: DocumentChunker | None = None, + conversion_timeout: int = 600, + max_concurrent: int = _DEFAULT_MAX_CONCURRENT, + ): + self._converter = converter + self._chunker = chunker + self._conversion_timeout = conversion_timeout + self._semaphore = asyncio.Semaphore(max_concurrent) + + async def create( + self, + document_id: str, + *, + pipeline_options: dict | None = None, + chunking_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}") + + job = AnalysisJob(document_id=document_id) + job.document_filename = doc.filename + await analysis_repo.insert(job) + + task = asyncio.create_task( + self._run_analysis( + job.id, + doc.storage_path, + doc.filename, + pipeline_options, + chunking_options, + ) + ) + task.add_done_callback(functools.partial(_on_task_done, job_id=job.id)) + + return job + + async def find_all(self) -> list[AnalysisJob]: + """Return all analysis jobs, newest first.""" + return await analysis_repo.find_all() + + async def find_by_id(self, job_id: str) -> AnalysisJob | None: + """Find an analysis job by ID, or return None.""" + return await analysis_repo.find_by_id(job_id) + + async def delete(self, job_id: str) -> bool: + """Delete an analysis job. Returns True if it existed.""" + return await analysis_repo.delete(job_id) + + async def rechunk(self, job_id: str, chunking_options: dict) -> list[ChunkResult]: + """Re-chunk an existing completed analysis with new options.""" + job = await analysis_repo.find_by_id(job_id) + if not job: + raise ValueError(f"Analysis not found: {job_id}") + if job.status != AnalysisStatus.COMPLETED: + raise ValueError(f"Analysis is not completed: {job_id}") + if not job.document_json: + raise ValueError(f"No document data available for re-chunking: {job_id}") + if not self._chunker: + raise ValueError("Chunking is not available") + + options = ChunkingOptions(**chunking_options) + chunks = await self._chunker.chunk(job.document_json, options) + + chunks_json = json.dumps([_chunk_to_dict(c) for c in chunks]) + await analysis_repo.update_chunks(job_id, chunks_json) + + return chunks + + async def _run_analysis( + self, + job_id: str, + file_path: str, + filename: str, + pipeline_options: dict | None = None, + chunking_options: dict | None = None, + ) -> None: + """Background task: run conversion and optionally chunk. + + Acquires the concurrency semaphore to limit parallel conversions + and prevent CPU/memory exhaustion on modest hardware. + """ + async with self._semaphore: + await self._run_analysis_inner( + job_id, file_path, filename, pipeline_options, chunking_options + ) + + async def _run_analysis_inner( + self, + job_id: str, + file_path: str, + filename: str, + pipeline_options: dict | None = None, + chunking_options: dict | None = None, + ) -> None: + """Inner analysis logic — called under the concurrency semaphore.""" + 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]) + + chunks_json = None + if chunking_options and self._chunker and result.document_json: + chunk_opts = ChunkingOptions(**chunking_options) + chunks = await self._chunker.chunk(result.document_json, chunk_opts) + chunks_json = json.dumps([_chunk_to_dict(c) for c in chunks]) + logger.info("Chunking produced %d chunks for job %s", len(chunks), job_id) + + job.mark_completed( + markdown=result.content_markdown, + html=result.content_html, + pages_json=pages_json, + document_json=result.document_json, + chunks_json=chunks_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)) + + +_background_tasks: set[asyncio.Task] = set() + + +def _on_task_done(task: asyncio.Task, *, job_id: str) -> None: + """Log unhandled exceptions from background analysis tasks and mark job as FAILED.""" if task.cancelled(): - logger.warning("Analysis task was cancelled") + logger.warning("Analysis task was cancelled: %s", job_id) + _schedule_mark_failed(job_id, "Task was cancelled") return exc = task.exception() if exc: - logger.error("Unhandled exception in analysis task: %s", exc, exc_info=True) + logger.error("Unhandled exception in analysis task %s: %s", job_id, exc, exc_info=True) + _schedule_mark_failed(job_id, str(exc)) -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)) +def _schedule_mark_failed(job_id: str, error: str) -> None: + """Schedule _mark_failed as a tracked background task.""" + t = asyncio.ensure_future(_mark_failed(job_id, error)) + _background_tasks.add(t) + t.add_done_callback(_background_tasks.discard) async def _mark_failed(job_id: str, error: str) -> None: @@ -112,5 +219,7 @@ async def _mark_failed(job_id: str, error: str) -> None: if job: job.mark_failed(error) await analysis_repo.update_status(job) + except OSError: + logger.exception("Database I/O error marking job %s as failed", job_id) except Exception: - logger.exception("Could not mark job %s as failed", job_id) + logger.exception("Unexpected error marking job %s as failed", job_id) diff --git a/document-parser/services/document_service.py b/document-parser/services/document_service.py index f36825d..d54ebfb 100644 --- a/document-parser/services/document_service.py +++ b/document-parser/services/document_service.py @@ -10,11 +10,12 @@ import uuid from pdf2image import convert_from_bytes, pdfinfo_from_bytes from domain.models import Document +from infra.settings import settings from persistence import analysis_repo, document_repo logger = logging.getLogger(__name__) -UPLOAD_DIR = os.environ.get("UPLOAD_DIR", "./uploads") +UPLOAD_DIR = settings.upload_dir MAX_FILE_SIZE = 50 * 1024 * 1024 # 50 MB @@ -22,8 +23,14 @@ MAX_FILE_SIZE = 50 * 1024 * 1024 # 50 MB _PDF_MAGIC = b"%PDF" +_UPLOAD_CHUNK_SIZE = 64 * 1024 # 64 KB chunks for streaming writes + + async def upload(filename: str, content_type: str, file_content: bytes) -> Document: - """Save uploaded file to disk and persist metadata.""" + """Save uploaded file to disk and persist metadata. + + Writes the file in fixed-size chunks to keep peak memory usage low. + """ if len(file_content) > MAX_FILE_SIZE: raise ValueError("File too large (max 50 MB)") @@ -32,12 +39,14 @@ async def upload(filename: str, content_type: str, file_content: bytes) -> Docum os.makedirs(UPLOAD_DIR, exist_ok=True) - ext = os.path.splitext(filename)[1] or ".pdf" + ext = ".pdf" # Content already validated as PDF safe_name = f"{uuid.uuid4()}{ext}" file_path = os.path.join(UPLOAD_DIR, safe_name) + # Write in chunks to avoid doubling memory usage for large files with open(file_path, "wb") as f: - f.write(file_content) + for offset in range(0, len(file_content), _UPLOAD_CHUNK_SIZE): + f.write(file_content[offset : offset + _UPLOAD_CHUNK_SIZE]) # Count PDF pages page_count = _count_pages(file_content) @@ -54,10 +63,12 @@ async def upload(filename: str, content_type: str, file_content: bytes) -> Docum async def find_all() -> list[Document]: + """Return all documents, newest first.""" return await document_repo.find_all() async def find_by_id(doc_id: str) -> Document | None: + """Find a document by its ID, or return None.""" return await document_repo.find_by_id(doc_id) @@ -70,12 +81,20 @@ async def delete(doc_id: str) -> bool: # Delete associated analyses first (cascade) await analysis_repo.delete_by_document(doc_id) - # Delete file from disk + # Delete file from disk (only if inside UPLOAD_DIR) try: - if os.path.exists(doc.storage_path): - os.unlink(doc.storage_path) + real_path = os.path.realpath(doc.storage_path) + real_upload_dir = os.path.realpath(UPLOAD_DIR) + if real_path.startswith(real_upload_dir + os.sep) and os.path.exists(real_path): + os.unlink(real_path) + elif os.path.exists(doc.storage_path): + logger.warning("Refused to delete file outside upload dir: %s", doc.storage_path) + except FileNotFoundError: + logger.info("File already removed: %s", doc.storage_path) + except PermissionError: + logger.error("Permission denied deleting file: %s", doc.storage_path) except OSError: - logger.warning("Could not delete file: %s", doc.storage_path) + logger.warning("Could not delete file: %s", doc.storage_path, exc_info=True) return await document_repo.delete(doc_id) @@ -96,6 +115,9 @@ def _count_pages(file_content: bytes) -> int | None: try: info = pdfinfo_from_bytes(file_content) return info.get("Pages") - except Exception: - logger.warning("Could not count pages", exc_info=True) + except (FileNotFoundError, OSError) as exc: + logger.warning("Could not count pages: %s", exc) + return None + except Exception: + logger.warning("Unexpected error counting pages", exc_info=True) return None diff --git a/document-parser/tests/test_analysis_service.py b/document-parser/tests/test_analysis_service.py new file mode 100644 index 0000000..83f6be8 --- /dev/null +++ b/document-parser/tests/test_analysis_service.py @@ -0,0 +1,114 @@ +"""Tests for AnalysisService — callbacks, concurrency, and orchestration.""" + +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from services.analysis_service import AnalysisService, _on_task_done + + +class TestOnTaskDone: + """Bug #1: _on_task_done must call _mark_failed when the task raises.""" + + @pytest.mark.asyncio + async def test_exception_marks_job_failed(self): + """When a background task raises, the job should be marked FAILED.""" + job_id = "job-123" + + # Create a task that raises + async def failing_task(): + raise RuntimeError("boom") + + task = asyncio.create_task(failing_task()) + await asyncio.sleep(0) # let the task fail + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + # ensure_future schedules it; give the event loop a tick + await asyncio.sleep(0) + + mock_mark.assert_called_once_with(job_id, "boom") + + @pytest.mark.asyncio + async def test_cancelled_task_marks_job_failed(self): + """When a background task is cancelled, the job should be marked FAILED.""" + job_id = "job-456" + + async def slow_task(): + await asyncio.sleep(999) + + import contextlib + + task = asyncio.create_task(slow_task()) + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + await asyncio.sleep(0) + + mock_mark.assert_called_once_with(job_id, "Task was cancelled") + + @pytest.mark.asyncio + async def test_successful_task_does_not_mark_failed(self): + """When a background task succeeds, _mark_failed should not be called.""" + job_id = "job-789" + + async def ok_task(): + return "done" + + task = asyncio.create_task(ok_task()) + await task + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + await asyncio.sleep(0) + + mock_mark.assert_not_called() + + +class TestAnalysisServiceConcurrency: + """Verify that the semaphore limits concurrent analysis jobs.""" + + def test_semaphore_initialized_with_max_concurrent(self): + converter = MagicMock() + service = AnalysisService(converter=converter, max_concurrent=5) + assert service._semaphore._value == 5 + + def test_default_max_concurrent(self): + converter = MagicMock() + service = AnalysisService(converter=converter) + assert service._semaphore._value == 3 + + @pytest.mark.asyncio + async def test_semaphore_limits_parallel_jobs(self): + """Only max_concurrent jobs should run in parallel; others must wait.""" + call_order: list[str] = [] + blocker = asyncio.Event() + + converter = MagicMock() + service = AnalysisService(converter=converter, max_concurrent=1) + + async def fake_inner(self, *args, **kwargs): + call_order.append("start") + await blocker.wait() + call_order.append("end") + + with patch.object(AnalysisService, "_run_analysis_inner", fake_inner): + t1 = asyncio.create_task(service._run_analysis("j1", "/f", "f.pdf")) + t2 = asyncio.create_task(service._run_analysis("j2", "/f", "f.pdf")) + await asyncio.sleep(0.05) + + # With max_concurrent=1, only one task should have started + assert call_order.count("start") == 1 + + blocker.set() + await asyncio.gather(t1, t2) + + # Both should have completed + assert call_order.count("start") == 2 + assert call_order.count("end") == 2 diff --git a/document-parser/tests/test_api_endpoints.py b/document-parser/tests/test_api_endpoints.py index da10af9..6eae888 100644 --- a/document-parser/tests/test_api_endpoints.py +++ b/document-parser/tests/test_api_endpoints.py @@ -1,6 +1,6 @@ """Tests for FastAPI API endpoints using TestClient.""" -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi.testclient import TestClient @@ -14,11 +14,24 @@ def client(): return TestClient(app, raise_server_exceptions=False) +@pytest.fixture +def mock_analysis_service(client): + """Inject a mock AnalysisService into app.state for the duration of the test.""" + mock_svc = MagicMock() + original = getattr(app.state, "analysis_service", None) + app.state.analysis_service = mock_svc + yield mock_svc + app.state.analysis_service = original + + class TestHealthEndpoint: def test_health(self, client): - resp = client.get("/health") + resp = client.get("/api/health") assert resp.status_code == 200 - assert resp.json() == {"status": "ok"} + data = resp.json() + assert data["status"] in ("ok", "degraded") + assert "engine" in data + assert "database" in data class TestDocumentEndpoints: @@ -41,8 +54,12 @@ class TestDocumentEndpoints: @patch("services.document_service.find_by_id", new_callable=AsyncMock) def test_get_document(self, mock_find, client): mock_find.return_value = Document( - id="d1", filename="test.pdf", content_type="application/pdf", - file_size=2048, page_count=3, storage_path="/tmp/test", + id="d1", + filename="test.pdf", + content_type="application/pdf", + file_size=2048, + page_count=3, + storage_path="/tmp/test", ) resp = client.get("/api/documents/d1") @@ -62,8 +79,10 @@ class TestDocumentEndpoints: @patch("services.document_service.upload", new_callable=AsyncMock) def test_upload_document(self, mock_upload, client): mock_upload.return_value = Document( - id="new-1", filename="uploaded.pdf", - content_type="application/pdf", file_size=512, + id="new-1", + filename="uploaded.pdf", + content_type="application/pdf", + file_size=512, storage_path="/tmp/uploaded", ) @@ -84,7 +103,20 @@ class TestDocumentEndpoints: "/api/documents/upload", files={"file": ("big.pdf", b"x", "application/pdf")}, ) - assert resp.status_code == 413 + assert resp.status_code == 400 + + @patch("services.document_service.find_by_id", new_callable=AsyncMock) + def test_preview_page_out_of_range(self, mock_find, client): + mock_find.return_value = Document( + id="d1", + filename="test.pdf", + page_count=3, + storage_path="/tmp/test.pdf", + ) + + resp = client.get("/api/documents/d1/preview?page=10") + assert resp.status_code == 400 + assert "out of range" in resp.json()["detail"] @patch("services.document_service.delete", new_callable=AsyncMock) def test_delete_document(self, mock_delete, client): @@ -102,11 +134,12 @@ class TestDocumentEndpoints: class TestAnalysisEndpoints: - @patch("services.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"), - ] + def test_list_analyses(self, client, mock_analysis_service): + mock_analysis_service.find_all = AsyncMock( + return_value=[ + AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf"), + ] + ) resp = client.get("/api/analyses") assert resp.status_code == 200 @@ -117,11 +150,10 @@ class TestAnalysisEndpoints: assert data[0]["documentFilename"] == "test.pdf" assert data[0]["status"] == "PENDING" - @patch("services.analysis_service.find_by_id", new_callable=AsyncMock) - def test_get_analysis(self, mock_find, client): + def test_get_analysis(self, client, mock_analysis_service): job = AnalysisJob(id="j1", document_id="d1", document_filename="test.pdf") job.mark_running() - mock_find.return_value = job + mock_analysis_service.find_by_id = AsyncMock(return_value=job) resp = client.get("/api/analyses/j1") assert resp.status_code == 200 @@ -129,17 +161,19 @@ class TestAnalysisEndpoints: assert data["status"] == "RUNNING" assert data["startedAt"] is not None - @patch("services.analysis_service.find_by_id", new_callable=AsyncMock) - def test_get_analysis_not_found(self, mock_find, client): - mock_find.return_value = None + def test_get_analysis_not_found(self, client, mock_analysis_service): + mock_analysis_service.find_by_id = AsyncMock(return_value=None) resp = client.get("/api/analyses/missing") assert resp.status_code == 404 - @patch("services.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", + def test_create_analysis(self, client, mock_analysis_service): + mock_analysis_service.create = AsyncMock( + return_value=AnalysisJob( + id="j1", + document_id="d1", + document_filename="test.pdf", + ) ) resp = client.post("/api/analyses", json={"documentId": "d1"}) @@ -147,34 +181,44 @@ class TestAnalysisEndpoints: data = resp.json() assert data["id"] == "j1" assert data["documentId"] == "d1" - mock_create.assert_called_once_with("d1", pipeline_options=None) - - @patch("services.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", + mock_analysis_service.create.assert_called_once_with( + "d1", + pipeline_options=None, + chunking_options=None, ) - resp = client.post("/api/analyses", json={ - "documentId": "d1", - "pipelineOptions": { - "do_ocr": False, - "do_table_structure": True, - "table_mode": "fast", - "do_code_enrichment": True, - "do_formula_enrichment": False, - "do_picture_classification": False, - "do_picture_description": False, - "generate_picture_images": True, - "generate_page_images": False, - "images_scale": 2.0, - } - }) + def test_create_analysis_with_pipeline_options(self, client, mock_analysis_service): + mock_analysis_service.create = AsyncMock( + return_value=AnalysisJob( + id="j2", + document_id="d1", + document_filename="test.pdf", + ) + ) + + resp = client.post( + "/api/analyses", + json={ + "documentId": "d1", + "pipelineOptions": { + "do_ocr": False, + "do_table_structure": True, + "table_mode": "fast", + "do_code_enrichment": True, + "do_formula_enrichment": False, + "do_picture_classification": False, + "do_picture_description": False, + "generate_picture_images": True, + "generate_page_images": False, + "images_scale": 2.0, + }, + }, + ) assert resp.status_code == 200 data = resp.json() assert data["id"] == "j2" - call_kwargs = mock_create.call_args + call_kwargs = mock_analysis_service.create.call_args opts = call_kwargs.kwargs["pipeline_options"] assert opts["do_ocr"] is False assert opts["table_mode"] == "fast" @@ -182,51 +226,50 @@ class TestAnalysisEndpoints: assert opts["generate_picture_images"] is True assert opts["images_scale"] == 2.0 - @patch("services.analysis_service.create", new_callable=AsyncMock) - def test_create_analysis_with_partial_pipeline_options(self, mock_create, client): + def test_create_analysis_with_partial_pipeline_options(self, client, mock_analysis_service): """Pipeline options should use defaults for unspecified fields.""" - mock_create.return_value = AnalysisJob( - id="j3", document_id="d1", document_filename="test.pdf", + mock_analysis_service.create = AsyncMock( + return_value=AnalysisJob( + id="j3", + document_id="d1", + document_filename="test.pdf", + ) ) - resp = client.post("/api/analyses", json={ - "documentId": "d1", - "pipelineOptions": {"do_ocr": False} - }) + resp = client.post( + "/api/analyses", json={"documentId": "d1", "pipelineOptions": {"do_ocr": False}} + ) assert resp.status_code == 200 - opts = mock_create.call_args.kwargs["pipeline_options"] + opts = mock_analysis_service.create.call_args.kwargs["pipeline_options"] assert opts["do_ocr"] is False # Defaults assert opts["do_table_structure"] is True assert opts["table_mode"] == "accurate" assert opts["do_code_enrichment"] is False - @patch("services.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") + def test_create_analysis_document_not_found(self, client, mock_analysis_service): + mock_analysis_service.create = AsyncMock(side_effect=ValueError("Document not found: d99")) resp = client.post("/api/analyses", json={"documentId": "d99"}) assert resp.status_code == 404 - def test_create_analysis_empty_document_id(self, client): + def test_create_analysis_empty_document_id(self, client, mock_analysis_service): resp = client.post("/api/analyses", json={"documentId": ""}) assert resp.status_code == 400 - def test_create_analysis_whitespace_document_id(self, client): + def test_create_analysis_whitespace_document_id(self, client, mock_analysis_service): resp = client.post("/api/analyses", json={"documentId": " "}) assert resp.status_code == 400 - @patch("services.analysis_service.delete", new_callable=AsyncMock) - def test_delete_analysis(self, mock_delete, client): - mock_delete.return_value = True + def test_delete_analysis(self, client, mock_analysis_service): + mock_analysis_service.delete = AsyncMock(return_value=True) resp = client.delete("/api/analyses/j1") assert resp.status_code == 204 - @patch("services.analysis_service.delete", new_callable=AsyncMock) - def test_delete_analysis_not_found(self, mock_delete, client): - mock_delete.return_value = False + def test_delete_analysis_not_found(self, client, mock_analysis_service): + mock_analysis_service.delete = AsyncMock(return_value=False) resp = client.delete("/api/analyses/missing") assert resp.status_code == 404 diff --git a/document-parser/tests/test_bbox.py b/document-parser/tests/test_bbox.py index 9ffe67c..8d3694f 100644 --- a/document-parser/tests/test_bbox.py +++ b/document-parser/tests/test_bbox.py @@ -8,12 +8,13 @@ misaligned overlays in the UI. import pytest from docling_core.types.doc.base import BoundingBox, CoordOrigin -from domain.bbox import EMPTY_BBOX, to_topleft_list +from infra.bbox import EMPTY_BBOX, to_topleft_list # --------------------------------------------------------------------------- # Standard conversions # --------------------------------------------------------------------------- + class TestToTopleftListStandard: """Normal bbox conversions (happy path).""" @@ -29,10 +30,10 @@ class TestToTopleftListStandard: result = to_topleft_list(bbox, page_height=792.0) # After conversion: new_t = 792 - 700 = 92, new_b = 792 - 600 = 192 - assert result[0] == 50 # l unchanged - assert result[1] == pytest.approx(92.0) # t = page_height - old_t - assert result[2] == 200 # r unchanged - assert result[3] == pytest.approx(192.0) # b = page_height - old_b + assert result[0] == 50 # l unchanged + assert result[1] == pytest.approx(92.0) # t = page_height - old_t + assert result[2] == 200 # r unchanged + assert result[3] == pytest.approx(192.0) # b = page_height - old_b def test_result_has_positive_dimensions(self): """Converted bbox should always have b > t (positive height).""" @@ -60,6 +61,7 @@ class TestToTopleftListStandard: # Page format variations # --------------------------------------------------------------------------- + class TestPageFormats: """Verify correct conversion across different page sizes.""" @@ -105,6 +107,7 @@ class TestPageFormats: # Degenerate / edge-case bboxes # --------------------------------------------------------------------------- + class TestDegenerateBboxes: """Bboxes that are invalid or degenerate should return EMPTY_BBOX.""" @@ -151,6 +154,7 @@ class TestDegenerateBboxes: # Precision and boundary values # --------------------------------------------------------------------------- + class TestPrecision: """Floating-point precision and edge values.""" diff --git a/document-parser/tests/test_chunking.py b/document-parser/tests/test_chunking.py new file mode 100644 index 0000000..52e84c8 --- /dev/null +++ b/document-parser/tests/test_chunking.py @@ -0,0 +1,359 @@ +"""Tests for chunking feature — domain, schemas, service, and API endpoints.""" + +from __future__ import annotations + +import json +from dataclasses import asdict +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi.testclient import TestClient + +from api.schemas import ChunkBboxResponse, ChunkingOptionsRequest, ChunkResponse, RechunkRequest +from domain.models import AnalysisJob, AnalysisStatus +from domain.value_objects import ChunkBbox, ChunkingOptions, ChunkResult +from main import app + +# --------------------------------------------------------------------------- +# Domain: value objects +# --------------------------------------------------------------------------- + + +class TestChunkingOptions: + def test_defaults(self): + opts = ChunkingOptions() + assert opts.chunker_type == "hybrid" + assert opts.max_tokens == 512 + assert opts.merge_peers is True + assert opts.repeat_table_header is True + + def test_custom_values(self): + opts = ChunkingOptions(chunker_type="hierarchical", max_tokens=256, merge_peers=False) + assert opts.chunker_type == "hierarchical" + assert opts.max_tokens == 256 + assert opts.merge_peers is False + + def test_is_default(self): + assert ChunkingOptions().is_default() + assert not ChunkingOptions(max_tokens=256).is_default() + + +class TestChunkResult: + def test_defaults(self): + chunk = ChunkResult(text="hello") + assert chunk.text == "hello" + assert chunk.headings == [] + assert chunk.source_page is None + assert chunk.token_count == 0 + + def test_full_values(self): + chunk = ChunkResult( + text="content", + headings=["Title", "Section"], + source_page=3, + token_count=42, + ) + assert chunk.headings == ["Title", "Section"] + assert chunk.source_page == 3 + assert chunk.token_count == 42 + + def test_serializable(self): + chunk = ChunkResult(text="x", headings=["h1"], source_page=1, token_count=10) + data = asdict(chunk) + assert data == { + "text": "x", + "headings": ["h1"], + "source_page": 1, + "token_count": 10, + "bboxes": [], + } + + +class TestChunkBbox: + def test_construction(self): + bbox = ChunkBbox(page=1, bbox=[10.0, 20.0, 100.0, 80.0]) + assert bbox.page == 1 + assert bbox.bbox == [10.0, 20.0, 100.0, 80.0] + + def test_serializable(self): + bbox = ChunkBbox(page=2, bbox=[0.0, 0.0, 50.0, 50.0]) + data = asdict(bbox) + assert data == {"page": 2, "bbox": [0.0, 0.0, 50.0, 50.0]} + + def test_chunk_result_with_bboxes(self): + chunk = ChunkResult( + text="content", + bboxes=[ + ChunkBbox(page=1, bbox=[10, 20, 100, 80]), + ChunkBbox(page=2, bbox=[50, 50, 150, 250]), + ], + ) + assert len(chunk.bboxes) == 2 + assert chunk.bboxes[0].page == 1 + + +class TestChunkBboxResponse: + def test_serializes(self): + resp = ChunkBboxResponse(page=1, bbox=[10.0, 20.0, 100.0, 80.0]) + data = resp.model_dump(by_alias=True) + assert data == {"page": 1, "bbox": [10.0, 20.0, 100.0, 80.0]} + + def test_chunk_response_with_bboxes(self): + resp = ChunkResponse( + text="hello", + bboxes=[ChunkBboxResponse(page=1, bbox=[10, 20, 100, 80])], + ) + data = resp.model_dump(by_alias=True) + assert len(data["bboxes"]) == 1 + assert data["bboxes"][0]["page"] == 1 + + +# --------------------------------------------------------------------------- +# Domain: AnalysisJob with chunking fields +# --------------------------------------------------------------------------- + + +class TestAnalysisJobChunking: + def test_default_chunking_fields(self): + job = AnalysisJob() + assert job.document_json is None + assert job.chunks_json is None + + def test_mark_completed_with_chunks(self): + job = AnalysisJob() + job.mark_running() + job.mark_completed( + markdown="# Title", + html="
Text
", + "json_content": { + "pages": {"1": {"size": {"width": 612.0, "height": 792.0}}}, + "texts": [ + { + "label": "title", + "text": "Title", + "prov": [ + { + "page_no": 1, + "bbox": { + "l": 10, + "t": 20, + "r": 200, + "b": 40, + "coord_origin": "TOPLEFT", + }, + } + ], + }, + { + "label": "paragraph", + "text": "Text", + "prov": [ + { + "page_no": 1, + "bbox": { + "l": 10, + "t": 50, + "r": 200, + "b": 70, + "coord_origin": "TOPLEFT", + }, + } + ], + }, + ], + "tables": [], + "pictures": [], + }, + } + } + result = _parse_response(data) + assert len(result.pages[0].elements) == 2 + assert result.pages[0].elements[0].type == "title" + assert result.pages[0].elements[0].content == "Title" + assert result.pages[0].elements[0].bbox == [10, 20, 200, 40] + assert result.pages[0].elements[1].type == "text" + + def test_multi_page(self): + data = { + "document": { + "md_content": "", + "html_content": "", + "json_content": { + "pages": { + "1": {"size": {"width": 612.0, "height": 792.0}}, + "2": {"size": {"width": 595.0, "height": 842.0}}, + }, + "texts": [], + "tables": [], + "pictures": [], + }, + } + } + result = _parse_response(data) + assert result.page_count == 2 + assert result.pages[1].width == 595.0 + + def test_no_json_content(self): + data = { + "document": { + "md_content": "text", + "html_content": "text
", + } + } + result = _parse_response(data) + assert result.content_markdown == "text" + assert result.pages == [] + assert result.page_count == 1 + assert result.document_json is None + + def test_json_content_as_string(self): + json_doc = { + "pages": {"1": {"size": {"width": 612.0, "height": 792.0}}}, + "texts": [], + "tables": [], + "pictures": [], + } + data = { + "document": { + "md_content": "", + "html_content": "", + "json_content": json.dumps(json_doc), + } + } + result = _parse_response(data) + assert result.page_count == 1 + + def test_json_content_malformed_string_falls_back(self): + """Bug #5: malformed JSON string in json_content must not crash.""" + data = { + "document": { + "md_content": "# Hello", + "html_content": "+ {{ + analysisStore.currentChunks.length ? t('chunking.noChunksOnPage') : t('chunking.noChunks') + }} +
+
{{ t('home.subtitle') }}
{{ t('studio.subtitle') }}