oval
5883565b87
- New endpoints: /api/v1/convert, /api/v1/extract, /api/v1/segment, /api/v1/ocr, /api/v1/table_rec, /api/v1/marker (deprecated), /api/v1/create-document, /api/v1/files/*, /api/v1/thumbnails - Async submit-and-poll pattern matching Datalab API spec - X-API-Key authentication via API_KEY env var - Filesystem-based request store for multi-worker support - OCR_BACKEND=deepseek mode: uses deepseek-ocr via ollama for OCR (prompt: <|grounding|>Free OCR.) - build_options() now propagates output_format to parsed_opts - Fixed docs page CSS curly brace conflict with .format() - Renamed marker_endpoint_referece.md -> marker_endpoint_reference.md - Containerfile: added poppler-utils, pandoc; 3 gunicorn workers - gunicorn.conf: post_fork hook for background worker threads
122 lines
3.5 KiB
Python
122 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
DEFAULT_RESULTS_DIR = os.environ.get("RESULTS_DIR", "/app/conversion_results")
|
|
DEFAULT_TTL = 3600 # 1 hour (matches Datalab)
|
|
|
|
_lock = threading.Lock()
|
|
|
|
|
|
def _req_path(request_id: str) -> Path:
|
|
return Path(DEFAULT_RESULTS_DIR) / request_id
|
|
|
|
|
|
def create_request(request_id: str, metadata: dict | None = None) -> dict:
|
|
d = _req_path(request_id)
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
entry = {
|
|
"request_id": request_id,
|
|
"status": "processing",
|
|
"success": None,
|
|
"created_at": time.time(),
|
|
"metadata": metadata or {},
|
|
}
|
|
with _lock:
|
|
(d / "status.json").write_text(json.dumps(entry))
|
|
return entry
|
|
|
|
|
|
def get_request(request_id: str) -> dict | None:
|
|
p = _req_path(request_id) / "status.json"
|
|
if not p.exists():
|
|
return None
|
|
with _lock:
|
|
return json.loads(p.read_text())
|
|
|
|
|
|
def update_request(request_id: str, **kwargs) -> dict | None:
|
|
d = _req_path(request_id)
|
|
p = d / "status.json"
|
|
if not p.exists():
|
|
return None
|
|
with _lock:
|
|
entry = json.loads(p.read_text())
|
|
entry.update(kwargs)
|
|
entry["updated_at"] = time.time()
|
|
p.write_text(json.dumps(entry))
|
|
return entry
|
|
|
|
|
|
def complete_request(request_id: str, result: dict):
|
|
entry = update_request(request_id, status="complete", success=True)
|
|
if entry:
|
|
(Path(DEFAULT_RESULTS_DIR) / request_id / "result.json").write_text(
|
|
json.dumps(result)
|
|
)
|
|
|
|
|
|
def fail_request(request_id: str, error: str):
|
|
update_request(request_id, status="complete", success=False, error=error)
|
|
|
|
|
|
def get_result(request_id: str) -> dict | None:
|
|
p = _req_path(request_id) / "result.json"
|
|
if not p.exists():
|
|
return None
|
|
return json.loads(p.read_text())
|
|
|
|
|
|
def _cleanup_loop(interval: int = 300, ttl: int = DEFAULT_TTL):
|
|
while True:
|
|
time.sleep(interval)
|
|
now = time.time()
|
|
try:
|
|
results_dir = Path(DEFAULT_RESULTS_DIR)
|
|
if not results_dir.exists():
|
|
continue
|
|
for child in results_dir.iterdir():
|
|
if child.is_dir():
|
|
status_p = child / "status.json"
|
|
if status_p.exists():
|
|
entry = json.loads(status_p.read_text())
|
|
if now - entry.get("created_at", 0) > ttl:
|
|
import shutil
|
|
shutil.rmtree(str(child))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def start_cleanup_thread(interval: int = 300, ttl: int = DEFAULT_TTL):
|
|
t = threading.Thread(target=_cleanup_loop, args=(interval, ttl), daemon=True)
|
|
t.start()
|
|
|
|
|
|
# ── Job queue for background processing (shared across workers via disk, but
|
|
# each worker has its own in-memory queue for picking up jobs) ──────
|
|
_job_queue: list[tuple[str, str, bytes, str, dict]] = []
|
|
_job_cond = threading.Condition()
|
|
|
|
|
|
def submit_job(endpoint: str, raw_bytes: bytes, filename: str, opts: dict) -> str:
|
|
import uuid
|
|
request_id = str(uuid.uuid4())
|
|
create_request(request_id, {"endpoint": endpoint, "filename": filename})
|
|
with _job_cond:
|
|
_job_queue.append((request_id, endpoint, raw_bytes, filename, opts))
|
|
_job_cond.notify()
|
|
return request_id
|
|
|
|
|
|
def wait_for_job(timeout: float | None = None) -> tuple | None:
|
|
with _job_cond:
|
|
if not _job_queue:
|
|
_job_cond.wait(timeout=timeout)
|
|
if not _job_queue:
|
|
return None
|
|
return _job_queue.pop(0)
|