diff --git a/Dockerfile b/Dockerfile index c8b3c01..a67b8da 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,9 +3,15 @@ FROM python:3.12-slim WORKDIR /app # Install system dependencies +# ffmpeg - required by pydub for audio transcription +# libimage-exiftool-perl - required by markitdown for image/audio metadata +# libmagic1 - content type sniffing RUN apt-get update && apt-get install -y --no-install-recommends \ gcc \ python3-dev \ + ffmpeg \ + libimage-exiftool-perl \ + libmagic1 \ && rm -rf /var/lib/apt/lists/* # Copy requirements and install dependencies diff --git a/api/index.py b/api/index.py index 3c3dc70..74326ec 100644 --- a/api/index.py +++ b/api/index.py @@ -1,9 +1,18 @@ from fastapi import FastAPI, HTTPException, Depends, Header +from fastapi.concurrency import run_in_threadpool from converter import convert +from cache import store_document, get_page, PAGE_SIZE from pydantic import BaseModel from typing import Optional +import logging import os +logging.basicConfig( + level=os.environ.get("LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) +logger = logging.getLogger(__name__) + app = FastAPI() @@ -26,13 +35,43 @@ async def verify_api_key(x_api_key: str = Header(None)): async def convert_endpoint( request: Optional[ConvertRequest] = None, ): + """ + Convert a document at the given URL to markdown. + + The markdown is paginated and stored in Redis (5 min TTL). The response + contains the first page plus pagination metadata; subsequent pages can be + fetched from ``GET /convert/{id}/pages/{page}`` using the returned ``id``. + """ + if request is None or not request.file: + raise HTTPException(status_code=422, detail="`file` (a URL) is required") + try: - result = convert(request.file) - return {"content": result.text_content} + # `convert` blocks on network IO and CPU-bound parsing; running it on the + # event loop would stall the health probes and get the pod killed. + result = await run_in_threadpool(convert, request.file) + return await run_in_threadpool(store_document, result.text_content) except Exception as e: + logger.exception("conversion failed for %s", request.file) raise HTTPException(status_code=400, detail=f"Error processing URL: {str(e)}") +@app.get("/convert/{doc_id}/pages/{page}", dependencies=[Depends(verify_api_key)]) +async def get_page_endpoint(doc_id: str, page: int): + """ + Fetch a single page of a previously converted document. + + Returns 404 if the document has expired from the cache or the page is out + of range. + """ + result = await run_in_threadpool(get_page, doc_id, page) + if result is None: + raise HTTPException( + status_code=404, + detail="Document or page not found (it may have expired).", + ) + return result + + @app.get("/") async def root(): return {"status": "ok"} diff --git a/cache.py b/cache.py new file mode 100644 index 0000000..b318ed1 --- /dev/null +++ b/cache.py @@ -0,0 +1,149 @@ +import json +import os +import uuid +from typing import Optional + +import redis + +# ---- Configuration (overridable via env) ---- +REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0") +# Number of characters per page. +PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "5000")) +# How long (seconds) paginated content lives in Redis. +CACHE_TTL = int(os.environ.get("CACHE_TTL", "300")) # 5 minutes + +_KEY_PREFIX = "markitdown:doc" + +_client: Optional[redis.Redis] = None + + +def get_client() -> redis.Redis: + """Return a lazily-initialised, shared Redis client.""" + global _client + if _client is None: + _client = redis.Redis.from_url(REDIS_URL, decode_responses=True) + return _client + + +def _meta_key(doc_id: str) -> str: + return f"{_KEY_PREFIX}:{doc_id}:meta" + + +def _page_key(doc_id: str, page: int) -> str: + return f"{_KEY_PREFIX}:{doc_id}:page:{page}" + + +def paginate(content: str, page_size: int = PAGE_SIZE) -> list[str]: + """ + Split markdown content into pages of roughly ``page_size`` characters. + + Pages are broken on paragraph boundaries (``\\n\\n``) where possible so a + paragraph is not split across pages. A single paragraph larger than + ``page_size`` is hard-split. + """ + if not content: + return [""] + + pages: list[str] = [] + current = "" + + for block in content.split("\n\n"): + # Hard-split a single block that is itself larger than a page. + while len(block) > page_size: + if current: + pages.append(current) + current = "" + pages.append(block[:page_size]) + block = block[page_size:] + + candidate = block if not current else f"{current}\n\n{block}" + if len(candidate) > page_size: + pages.append(current) + current = block + else: + current = candidate + + if current or not pages: + pages.append(current) + + return pages + + +def build_pagination(doc_id: str, page: int, total_pages: int, total_length: int, + page_size: int = PAGE_SIZE) -> dict: + has_next = page < total_pages + has_prev = page > 1 + return { + "id": doc_id, + "page": page, + "page_size": page_size, + "total_pages": total_pages, + "total_length": total_length, + "has_next": has_next, + "has_prev": has_prev, + "next_page": page + 1 if has_next else None, + "prev_page": page - 1 if has_prev else None, + } + + +def store_document(content: str, page_size: int = PAGE_SIZE, + ttl: int = CACHE_TTL) -> dict: + """ + Paginate ``content`` and store every page + metadata in Redis with a TTL. + + Returns a payload for the first page:: + + {"id", "content", "pagination": {...}} + """ + pages = paginate(content, page_size) + doc_id = uuid.uuid4().hex + total_pages = len(pages) + total_length = len(content) + + meta = { + "total_pages": total_pages, + "total_length": total_length, + "page_size": page_size, + } + + client = get_client() + pipe = client.pipeline() + pipe.set(_meta_key(doc_id), json.dumps(meta), ex=ttl) + for idx, page_content in enumerate(pages, start=1): + pipe.set(_page_key(doc_id, idx), page_content, ex=ttl) + pipe.execute() + + return { + "id": doc_id, + "content": pages[0], + "pagination": build_pagination(doc_id, 1, total_pages, total_length, page_size), + } + + +def get_page(doc_id: str, page: int) -> Optional[dict]: + """ + Fetch a single page for a previously stored document. + + Returns ``None`` if the document/page is not found (expired or invalid). + """ + client = get_client() + raw_meta = client.get(_meta_key(doc_id)) + if raw_meta is None: + return None + + meta = json.loads(raw_meta) + total_pages = meta["total_pages"] + if page < 1 or page > total_pages: + return None + + content = client.get(_page_key(doc_id, page)) + if content is None: + return None + + return { + "id": doc_id, + "content": content, + "pagination": build_pagination( + doc_id, page, total_pages, meta["total_length"], meta["page_size"] + ), + } diff --git a/converter.py b/converter.py index 5835f2c..f289e42 100644 --- a/converter.py +++ b/converter.py @@ -1,5 +1,7 @@ +import logging import os import os.path +import time from urllib.parse import urlparse from markitdown import MarkItDown, DocumentConverterResult @@ -7,11 +9,21 @@ import requests import tempfile +logger = logging.getLogger(__name__) + client = OpenAI( - base_url="https://openrouter.ai/api/v1", + base_url="https://ai-gateway.vercel.sh/v1", api_key=os.environ["OPENAI_API_KEY"], ) -model = "google/gemini-3-flash-preview" +model = os.environ.get("OPENAI_MODEL", "google/gemini-3.1-flash-lite-preview") + +# Seconds to wait for the connection and for each chunk of the response body. +DOWNLOAD_CONNECT_TIMEOUT = float(os.environ.get("DOWNLOAD_CONNECT_TIMEOUT", "10")) +DOWNLOAD_READ_TIMEOUT = float(os.environ.get("DOWNLOAD_READ_TIMEOUT", "30")) +# Hard cap on the downloaded file size, in bytes (default 100 MiB). +MAX_DOWNLOAD_BYTES = int(os.environ.get("MAX_DOWNLOAD_BYTES", str(100 * 1024 * 1024))) + +_CHUNK_SIZE = 64 * 1024 def download(url: str) -> str: @@ -24,8 +36,14 @@ def download(url: str) -> str: Returns: str: The temporary file path where the downloaded content is stored. """ - response = requests.get(url) - if response.status_code == 200: + with requests.get( + url, + stream=True, + timeout=(DOWNLOAD_CONNECT_TIMEOUT, DOWNLOAD_READ_TIMEOUT), + ) as response: + if response.status_code != 200: + raise Exception(f"Failed to download file from {url}, status code: {response.status_code}") + # Extract filename from URL or use a default name parsed_url = urlparse(url) filename = os.path.basename(parsed_url.path) @@ -36,27 +54,57 @@ def download(url: str) -> str: fd, temp_path = tempfile.mkstemp(suffix=f"_{filename}") os.close(fd) - # Write content to the temporary file - with open(temp_path, 'wb') as file: - file.write(response.content) + # Stream the content to the temporary file so memory stays bounded + written = 0 + try: + with open(temp_path, 'wb') as file: + for chunk in response.iter_content(chunk_size=_CHUNK_SIZE): + if not chunk: + continue + written += len(chunk) + if written > MAX_DOWNLOAD_BYTES: + raise Exception( + f"File from {url} exceeds the maximum size of {MAX_DOWNLOAD_BYTES} bytes" + ) + file.write(chunk) + except Exception: + if os.path.exists(temp_path): + os.remove(temp_path) + raise + return temp_path - else: - raise Exception(f"Failed to download file from {url}, status code: {response.status_code}") def convert(url: str) -> DocumentConverterResult: """ Convert a URL to a MarkItDown object. + This is fully synchronous and CPU/IO blocking; callers running inside an + event loop must offload it to a worker thread. + Args: url (str): The URL to convert. Returns: MarkItDown: The converted MarkItDown object. """ + started = time.monotonic() temp_file = download(url) + downloaded_at = time.monotonic() + logger.info("downloaded %s in %.3fs", url, downloaded_at - started) + md = MarkItDown(llm_client=client, llm_model=model) - converted = md.convert(temp_file) - if os.path.exists(temp_file): - os.remove(temp_file) + try: + converted = md.convert(temp_file) + finally: + if os.path.exists(temp_file): + os.remove(temp_file) + + logger.info( + "converted %s in %.3fs (download %.3fs, convert %.3fs)", + url, + time.monotonic() - started, + downloaded_at - started, + time.monotonic() - downloaded_at, + ) return converted diff --git a/k8s/deployment.yaml b/k8s/deployment.yaml index 0bba0d9..2490675 100644 --- a/k8s/deployment.yaml +++ b/k8s/deployment.yaml @@ -15,6 +15,9 @@ spec: labels: app: markitdown-server spec: + # Conversions can run for ~60s; let in-flight requests drain on rollout + # instead of being SIGKILLed after the default 30s. + terminationGracePeriodSeconds: 120 volumes: containers: - name: markitdown-server @@ -24,8 +27,15 @@ spec: memory: "1024Mi" cpu: "500m" requests: - memory: "128Mi" + # Baseline RSS is ~120-140Mi; a 128Mi request pinned memory + # utilization at ~100% and kept the HPA at max replicas. + memory: "384Mi" cpu: "250m" + env: + - name: REDIS_URL + value: "redis://markitdown-redis:6379/0" + - name: OPENAI_MODEL + value: "google/gemini-2.5-flash-preview-05-20" envFrom: - secretRef: name: markitdown-server-secret @@ -35,12 +45,16 @@ spec: port: 8000 initialDelaySeconds: 10 periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 readinessProbe: httpGet: path: / port: 8000 initialDelaySeconds: 10 periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 --- apiVersion: v1 diff --git a/k8s/kustomization.yaml b/k8s/kustomization.yaml index 676410c..9bd1039 100644 --- a/k8s/kustomization.yaml +++ b/k8s/kustomization.yaml @@ -5,6 +5,7 @@ namespace: markitdown-server resources: - namespace.yaml + - redis.yaml - deployment.yaml - hpa.yaml - ingress.yaml diff --git a/k8s/redis.yaml b/k8s/redis.yaml new file mode 100644 index 0000000..bb7a8c0 --- /dev/null +++ b/k8s/redis.yaml @@ -0,0 +1,62 @@ +apiVersion: apps/v1 +kind: Deployment +metadata: + name: markitdown-redis + namespace: markitdown-server + labels: + app: markitdown-redis +spec: + replicas: 1 + selector: + matchLabels: + app: markitdown-redis + template: + metadata: + labels: + app: markitdown-redis + spec: + containers: + - name: redis + image: redis:7-alpine + # Ephemeral cache only (5 min TTL); cap memory and evict on pressure. + args: + - "--maxmemory" + - "256mb" + - "--maxmemory-policy" + - "allkeys-lru" + - "--save" + - "" + ports: + - containerPort: 6379 + resources: + limits: + memory: "320Mi" + cpu: "250m" + requests: + memory: "64Mi" + cpu: "50m" + livenessProbe: + tcpSocket: + port: 6379 + initialDelaySeconds: 10 + periodSeconds: 10 + readinessProbe: + exec: + command: ["redis-cli", "ping"] + initialDelaySeconds: 5 + periodSeconds: 10 +--- +apiVersion: v1 +kind: Service +metadata: + name: markitdown-redis + namespace: markitdown-server + labels: + app: markitdown-redis +spec: + selector: + app: markitdown-redis + ports: + - port: 6379 + targetPort: 6379 + type: ClusterIP diff --git a/pyproject.toml b/pyproject.toml index 456b842..3f5c8d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,8 @@ dependencies = [ "uvicorn (>=0.34.2,<0.35.0)", "markitdown[all] (>=0.1.5,<0.2.0)", "openai (>=1.82.1,<2.0.0)", - "python-multipart (>=0.0.20,<0.0.21)" + "python-multipart (>=0.0.20,<0.0.21)", + "redis (>=5.2.1,<6.0.0)" ] diff --git a/requirements.txt b/requirements.txt index 7c413eb..a0a86f0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -51,6 +51,7 @@ python-dotenv==1.1.0 python-multipart==0.0.20 python-pptx==1.0.2 pytz==2025.2 +redis==5.2.1 requests==2.32.3 six==1.17.0 sniffio==1.3.1 diff --git a/test_main.http b/test_main.http index a2d81a9..c9ac2f6 100644 --- a/test_main.http +++ b/test_main.http @@ -5,7 +5,23 @@ Accept: application/json ### -GET http://127.0.0.1:8000/hello/User +# Convert a (possibly long) document to markdown. +# Returns the first page + pagination metadata. The full paginated content +# is cached in Redis for 5 minutes under the returned `id`. +POST http://127.0.0.1:8000/convert +Content-Type: application/json +x-api-key: NxQUCbbezCZ8KUhMtg5s0MQTQU4gDQ7 + +{ + "file": "https://example.com/some-long-document.pdf" +} + +### + +# Fetch a subsequent page of a previously converted document. +# Use the `id` and `pagination.next_page` returned by /convert. +GET http://127.0.0.1:8000/convert/REPLACE_WITH_DOC_ID/pages/2 Accept: application/json +x-api-key: NxQUCbbezCZ8KUhMtg5s0MQTQU4gDQ7 ###