Update .gitignore to include Ruby and PDF files, enhance README with synchronous and asynchronous API endpoint details, and implement task management for PDF conversions in server.py with logging and error handling.

.gitignore

@@ -176,4 +176,8 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
 
-.vscode/
+.vscode/
+
+# Test scripts
+*.rb
+*.pdf

README.md

@@ -505,12 +505,58 @@ result = response.json()
 
 ### API Endpoints
 
-- `GET /` - API information page (no authentication required)
-- `GET /docs` - Interactive API documentation (no authentication required)
+**Synchronous Endpoints** (wait for result, may take a long time):
 - `POST /marker` - Convert PDF using file path (requires authentication)
 - `POST /marker/upload` - Convert PDF via file upload (requires authentication)
 
-All POST endpoints require the `Authorization: Bearer <token>` header with a valid token.
+**Asynchronous Endpoints** (return immediately with task ID):
+- `POST /marker/async` - Submit PDF conversion task using file path (returns task ID)
+- `POST /marker/upload/async` - Submit PDF conversion task via file upload (returns task ID)
+- `GET /marker/task/{task_id}` - Check task status and get results
+- `DELETE /marker/task/{task_id}` - Delete a completed or failed task
+
+**Other Endpoints**:
+- `GET /` - API information page (no authentication required)
+- `GET /docs` - Interactive API documentation (no authentication required)
+
+All POST/GET/DELETE endpoints (except `/` and `/docs`) require the `Authorization: Bearer <token>` header with a valid token.
+
+### Using Asynchronous Endpoints
+
+For long-running conversions, use the async endpoints:
+
+```python
+import requests
+import time
+
+# 1. Submit task
+url = "https://your-space.hf.space/marker/upload/async"
+headers = {"Authorization": f"Bearer {token}"}
+files = {"file": ("document.pdf", open("document.pdf", "rb"), "application/pdf")}
+data = {"output_format": "markdown"}
+
+response = requests.post(url, headers=headers, files=files, data=data)
+task = response.json()
+task_id = task["task_id"]
+print(f"Task created: {task_id}")
+
+# 2. Poll for status
+status_url = f"https://your-space.hf.space/marker/task/{task_id}"
+while True:
+    response = requests.get(status_url, headers=headers)
+    status = response.json()
+    
+    if status["status"] == "completed":
+        result = status["result"]
+        print(f"Success! Output: {result['output'][:100]}...")
+        break
+    elif status["status"] == "failed":
+        print(f"Failed: {status.get('error', 'Unknown error')}")
+        break
+    else:
+        print(f"Status: {status['status']}, elapsed: {status.get('elapsed_time', 0)}s")
+        time.sleep(2)  # Wait 2 seconds before checking again
+```
 
 # Troubleshooting
 

marker/scripts/server.py

@@ -1,10 +1,17 @@
 import traceback
+import time
+import logging
+import uuid
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from threading import Lock
+from enum import Enum
 
 import click
 import os
 
 from pydantic import BaseModel, Field
-from starlette.responses import HTMLResponse
+from starlette.responses import HTMLResponse, JSONResponse
 from fastapi import FastAPI, Form, File, UploadFile, Depends, HTTPException, status
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 
@@ -13,13 +20,29 @@ from marker.output import text_from_rendered
 
 import base64
 from contextlib import asynccontextmanager
-from typing import Optional, Annotated
+from typing import Optional, Annotated, Dict, Any
 import io
 
 from marker.converters.pdf import PdfConverter
 from marker.models import create_model_dict
 from marker.settings import settings
 
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Task status enum
+class TaskStatus(str, Enum):
+    PENDING = "pending"
+    PROCESSING = "processing"
+    COMPLETED = "completed"
+    FAILED = "failed"
+
+# Global task storage and executor
+task_storage: Dict[str, Dict[str, Any]] = {}
+task_lock = Lock()
+executor = ThreadPoolExecutor(max_workers=2)  # Limit concurrent conversions
+
 app_data = {}
 
 
@@ -174,18 +197,35 @@ class CommonParams(BaseModel):
     ] = None
 
 
-async def _convert_pdf(params: CommonParams):
+def _convert_pdf_sync(params: CommonParams, task_id: str):
+    """Synchronous PDF conversion function to run in thread pool"""
     assert params.output_format in ["markdown", "json", "html", "chunks"], (
         "Invalid output format"
     )
+    start_time = time.time()
+    logger.info(f"[Task {task_id}] Starting PDF conversion for file: {params.filepath}")
+    
+    # Update task status
+    with task_lock:
+        task_storage[task_id]["status"] = TaskStatus.PROCESSING
+        task_storage[task_id]["started_at"] = time.time()
+    
     try:
+        # Get file size for logging
+        file_size = os.path.getsize(params.filepath) if params.filepath and os.path.exists(params.filepath) else 0
+        logger.info(f"[Task {task_id}] File size: {file_size / 1024:.2f} KB")
+        
         options = params.model_dump(exclude_none=True)
         # Remove None values to avoid passing them to ConfigParser
         options = {k: v for k, v in options.items() if v is not None}
+        logger.info(f"[Task {task_id}] Conversion options: {options}")
+        
         config_parser = ConfigParser(options)
         config_dict = config_parser.generate_config_dict()
         config_dict["pdftext_workers"] = 1
         converter_cls = config_parser.get_converter_cls()
+        
+        logger.info(f"[Task {task_id}] Creating converter...")
         converter = converter_cls(
             config=config_dict,
             artifact_dict=app_data["models"],
@@ -193,36 +233,76 @@ async def _convert_pdf(params: CommonParams):
             renderer=config_parser.get_renderer(),
             llm_service=config_parser.get_llm_service(),
         )
+        
+        logger.info(f"[Task {task_id}] Starting PDF conversion (this may take a while)...")
+        conversion_start = time.time()
         rendered = converter(params.filepath)
+        conversion_time = time.time() - conversion_start
+        logger.info(f"[Task {task_id}] PDF conversion completed in {conversion_time:.2f} seconds")
+        
+        logger.info(f"[Task {task_id}] Extracting text and images...")
         text, _, images = text_from_rendered(rendered)
         metadata = rendered.metadata
+        logger.info(f"[Task {task_id}] Extracted {len(text)} characters and {len(images)} images")
+        
+        logger.info(f"[Task {task_id}] Encoding images...")
+        encoded = {}
+        for k, v in images.items():
+            byte_stream = io.BytesIO()
+            v.save(byte_stream, format=settings.OUTPUT_IMAGE_FORMAT)
+            encoded[k] = base64.b64encode(byte_stream.getvalue()).decode(
+                settings.OUTPUT_ENCODING
+            )
+
+        total_time = time.time() - start_time
+        logger.info(f"[Task {task_id}] Total processing time: {total_time:.2f} seconds")
+
+        result = {
+            "format": params.output_format,
+            "output": text,
+            "images": encoded,
+            "metadata": metadata,
+            "success": True,
+            "processing_time": round(total_time, 2),
+        }
+        
+        # Update task with result
+        with task_lock:
+            task_storage[task_id]["status"] = TaskStatus.COMPLETED
+            task_storage[task_id]["result"] = result
+            task_storage[task_id]["completed_at"] = time.time()
+        
+        return result
+        
     except Exception as e:
+        logger.error(f"[Task {task_id}] Error during conversion: {str(e)}")
         traceback.print_exc()
-        return {
+        error_result = {
             "success": False,
             "error": str(e),
         }
+        
+        # Update task with error
+        with task_lock:
+            task_storage[task_id]["status"] = TaskStatus.FAILED
+            task_storage[task_id]["result"] = error_result
+            task_storage[task_id]["completed_at"] = time.time()
+            task_storage[task_id]["error"] = str(e)
+        
+        return error_result
 
-    encoded = {}
-    for k, v in images.items():
-        byte_stream = io.BytesIO()
-        v.save(byte_stream, format=settings.OUTPUT_IMAGE_FORMAT)
-        encoded[k] = base64.b64encode(byte_stream.getvalue()).decode(
-            settings.OUTPUT_ENCODING
-        )
 
-    return {
-        "format": params.output_format,
-        "output": text,
-        "images": encoded,
-        "metadata": metadata,
-        "success": True,
-    }
+async def _convert_pdf_async(params: CommonParams, task_id: str):
+    """Async wrapper for PDF conversion"""
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(executor, _convert_pdf_sync, params, task_id)
 
 
 @app.post("/marker")
 async def convert_pdf(params: CommonParams, token_verified: bool = Depends(verify_token)):
-    return await _convert_pdf(params)
+    """Synchronous endpoint - returns result directly (may take a long time)"""
+    task_id = f"sync-{uuid.uuid4().hex[:8]}"
+    return await _convert_pdf_async(params, task_id)
 
 
 @app.post("/marker/upload")
@@ -247,6 +327,7 @@ async def convert_pdf_upload(
     ),
     token_verified: bool = Depends(verify_token),
 ):
+    """Synchronous upload endpoint - returns result directly (may take a long time)"""
     upload_path = os.path.join(UPLOAD_DIRECTORY, file.filename)
     with open(upload_path, "wb+") as upload_file:
         file_contents = await file.read()
@@ -270,11 +351,151 @@ async def convert_pdf_upload(
         converter_cls=converter_cls,
         llm_service=llm_service,
     )
-    results = await _convert_pdf(params)
+    task_id = f"upload-{uuid.uuid4().hex[:8]}"
+    results = await _convert_pdf_async(params, task_id)
     os.remove(upload_path)
     return results
 
 
+@app.post("/marker/upload/async")
+async def convert_pdf_upload_async(
+    page_range: Optional[str] = Form(default=None),
+    force_ocr: Optional[bool] = Form(default=False),
+    paginate_output: Optional[bool] = Form(default=False),
+    output_format: Optional[str] = Form(default="markdown"),
+    output_dir: Optional[str] = Form(default=None),
+    use_llm: Optional[bool] = Form(default=False),
+    block_correction_prompt: Optional[str] = Form(default=None),
+    strip_existing_ocr: Optional[bool] = Form(default=False),
+    redo_inline_math: Optional[bool] = Form(default=False),
+    disable_image_extraction: Optional[bool] = Form(default=False),
+    debug: Optional[bool] = Form(default=False),
+    processors: Optional[str] = Form(default=None),
+    config_json: Optional[str] = Form(default=None),
+    converter_cls: Optional[str] = Form(default=None),
+    llm_service: Optional[str] = Form(default=None),
+    file: UploadFile = File(
+        ..., description="The PDF file to convert.", media_type="application/pdf"
+    ),
+    token_verified: bool = Depends(verify_token),
+):
+    """Asynchronous upload endpoint - returns task ID immediately"""
+    task_id = str(uuid.uuid4())
+    upload_path = os.path.join(UPLOAD_DIRECTORY, f"{task_id}_{file.filename}")
+    
+    # Save uploaded file
+    with open(upload_path, "wb+") as upload_file:
+        file_contents = await file.read()
+        upload_file.write(file_contents)
+
+    params = CommonParams(
+        filepath=upload_path,
+        page_range=page_range,
+        force_ocr=force_ocr,
+        paginate_output=paginate_output,
+        output_format=output_format,
+        output_dir=output_dir,
+        use_llm=use_llm,
+        block_correction_prompt=block_correction_prompt,
+        strip_existing_ocr=strip_existing_ocr,
+        redo_inline_math=redo_inline_math,
+        disable_image_extraction=disable_image_extraction,
+        debug=debug,
+        processors=processors,
+        config_json=config_json,
+        converter_cls=converter_cls,
+        llm_service=llm_service,
+    )
+    
+    # Initialize task
+    with task_lock:
+        task_storage[task_id] = {
+            "status": TaskStatus.PENDING,
+            "created_at": time.time(),
+            "params": params.model_dump(),
+            "result": None,
+            "error": None,
+            "upload_path": upload_path,  # Store path for cleanup
+        }
+    
+    # Start conversion in background
+    asyncio.create_task(_convert_pdf_async_with_cleanup(params, task_id, upload_path))
+    
+    return {
+        "task_id": task_id,
+        "status": TaskStatus.PENDING,
+        "message": "Task created. Use /marker/task/{task_id} to check status and get results.",
+        "status_url": f"/marker/task/{task_id}",
+    }
+
+
+async def _convert_pdf_async_with_cleanup(params: CommonParams, task_id: str, upload_path: str):
+    """Async wrapper with file cleanup"""
+    try:
+        await _convert_pdf_async(params, task_id)
+    finally:
+        # Clean up uploaded file
+        try:
+            if os.path.exists(upload_path):
+                os.remove(upload_path)
+        except Exception as e:
+            logger.warning(f"[Task {task_id}] Failed to cleanup file {upload_path}: {e}")
+
+
+@app.get("/marker/task/{task_id}")
+async def get_task_status(task_id: str, token_verified: bool = Depends(verify_token)):
+    """Get task status and result"""
+    with task_lock:
+        task = task_storage.get(task_id)
+    
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    
+    response = {
+        "task_id": task_id,
+        "status": task["status"],
+        "created_at": task.get("created_at"),
+        "started_at": task.get("started_at"),
+        "completed_at": task.get("completed_at"),
+    }
+    
+    if task["status"] == TaskStatus.COMPLETED:
+        response["result"] = task["result"]
+    elif task["status"] == TaskStatus.FAILED:
+        response["error"] = task.get("error", "Unknown error")
+        response["result"] = task.get("result")
+    
+    # Calculate elapsed time if processing
+    if task["status"] == TaskStatus.PROCESSING and "started_at" in task:
+        response["elapsed_time"] = round(time.time() - task["started_at"], 2)
+    
+    return response
+
+
+@app.delete("/marker/task/{task_id}")
+async def delete_task(task_id: str, token_verified: bool = Depends(verify_token)):
+    """Delete a completed or failed task"""
+    with task_lock:
+        task = task_storage.get(task_id)
+        if not task:
+            raise HTTPException(status_code=404, detail="Task not found")
+        
+        if task["status"] == TaskStatus.PROCESSING:
+            raise HTTPException(status_code=400, detail="Cannot delete task that is currently processing")
+        
+        # Clean up uploaded file if exists
+        upload_path = task.get("upload_path")
+        if upload_path and os.path.exists(upload_path):
+            try:
+                os.remove(upload_path)
+            except Exception as e:
+                logger.warning(f"Failed to cleanup file {upload_path}: {e}")
+        
+        del task_storage[task_id]
+    
+    return {"message": "Task deleted successfully"}
+
+
 @click.command()
 @click.option("--port", type=int, default=8000, help="Port to run the server on")
 @click.option("--host", type=str, default="127.0.0.1", help="Host to run the server on")