Spaces:
Running
Running
File size: 8,365 Bytes
6c24b50 19bd230 6c24b50 19bd230 62aa98f 6c24b50 65e87f1 6c24b50 5a01a63 62aa98f 6c24b50 19bd230 65e87f1 bd469c1 6c24b50 19bd230 7d1ad3f 19bd230 6c24b50 19bd230 65e87f1 6c24b50 19bd230 6c24b50 19bd230 65e87f1 19bd230 65e87f1 19bd230 6c24b50 65e87f1 6c24b50 19bd230 65e87f1 6c24b50 19bd230 bd469c1 19bd230 65e87f1 19bd230 bd469c1 19bd230 bd469c1 19bd230 6c24b50 65e87f1 6c24b50 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | from __future__ import annotations
import asyncio
import json as json_mod
import time
from typing import Annotated, List, Optional
from urllib.parse import urlparse
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, UploadFile
from app.api.deps import (
get_converter_service,
get_extraction_service,
get_text_cleaner_service,
)
from app.api.v1.convert import _build_metadata, _thread_pool
from app.config import get_settings
from app.core.logger import get_logger
from app.models.domain import ConversionError
from app.models.schemas import (
BatchFileResult,
BatchResponse,
BatchUrlRequest,
)
from app.services.converter_service import ConverterService
from app.services.extraction_service import ExtractionService
from app.services.text_cleaner_service import TextCleanerService
from app.utils.http_utils import DownloadError, download_url
router = APIRouter()
_logger = get_logger(__name__)
_settings = get_settings()
_MAX_UPLOAD_BYTES = _settings.max_upload_bytes
_MAX_BATCH_FILES = _settings.max_batch_files
def _batch_result_from_error(name: str, err: ConversionError) -> BatchFileResult:
return BatchFileResult(
filename=name,
success=False,
time_ms=round(err.duration_ms, 3),
error=err.message,
)
def _batch_result_from_ok(result) -> BatchFileResult:
return BatchFileResult(
filename=result.source,
success=True,
time_ms=round(result.duration_ms, 3),
content=result.markdown,
metadata=_build_metadata(result),
)
@router.post(
"/batch/files",
response_model=BatchResponse,
summary="Convert multiple files (up to 10)",
)
async def batch_files(
files: Annotated[List[UploadFile], File(description="Files to convert")],
return_json: bool = Form(False),
clean_content: bool = Query(False),
mappings: Optional[str] = Form(None, description="JSON string with field mappings"),
converter_service: ConverterService = Depends(get_converter_service),
extraction_service: ExtractionService = Depends(get_extraction_service),
text_cleaner_service: TextCleanerService = Depends(get_text_cleaner_service),
):
if not files:
raise HTTPException(status_code=400, detail={"success": False, "message": "No files provided."})
if len(files) > _MAX_BATCH_FILES:
raise HTTPException(status_code=400, detail={"success": False, "message": f"Maximum {_MAX_BATCH_FILES} files per batch."})
parsed_mappings = None
if mappings:
try:
parsed_mappings = json_mod.loads(mappings)
except json_mod.JSONDecodeError:
raise HTTPException(status_code=400, detail={"success": False, "message": "Invalid JSON in mappings parameter."})
batch_start = time.perf_counter()
async def process_single_file(f: UploadFile) -> BatchFileResult:
if f is None:
return BatchFileResult(filename="unknown", success=False, time_ms=0, error="File object is None.")
_logger.info("Batch processing file: %s", f.filename)
raw = await f.read()
if len(raw) > _MAX_UPLOAD_BYTES:
return BatchFileResult(
filename=f.filename or "unknown",
success=False,
time_ms=0,
error=f"File exceeds {_settings.max_upload_mb} MB limit.",
)
loop = asyncio.get_running_loop()
outcome = await loop.run_in_executor(_thread_pool, converter_service.convert_stream, raw, f.filename or "upload")
if isinstance(outcome, ConversionError):
return _batch_result_from_error(f.filename or "unknown", outcome)
content = outcome.markdown
if clean_content:
loop2 = asyncio.get_running_loop()
content = await loop2.run_in_executor(_thread_pool, text_cleaner_service.clean, content)
result = _batch_result_from_ok(outcome)
if clean_content:
result.content = content
if return_json and f.filename:
loop = asyncio.get_running_loop()
json_result = await loop.run_in_executor(
_thread_pool,
extraction_service.extract_structured,
f.filename,
outcome.markdown,
parsed_mappings,
raw,
)
result.json_content = json_result if "error" not in json_result else None
result.error = json_result.get("error") if "error" in json_result else None
return result
tasks = [process_single_file(f) for f in files]
results = await asyncio.gather(*tasks)
total_ms = round((time.perf_counter() - batch_start) * 1000, 3)
succeeded = sum(1 for r in results if r.success)
_logger.info("Batch files completed. Succeeded: %s/%s", succeeded, len(results))
return BatchResponse(
total=len(results),
succeeded=succeeded,
failed=len(results) - succeeded,
total_time_ms=total_ms,
results=results,
)
@router.post(
"/batch/urls",
response_model=BatchResponse,
summary="Convert multiple URLs (up to 20)",
)
async def batch_urls(
body: BatchUrlRequest,
clean_content: bool = Query(False),
converter_service: ConverterService = Depends(get_converter_service),
extraction_service: ExtractionService = Depends(get_extraction_service),
text_cleaner_service: TextCleanerService = Depends(get_text_cleaner_service),
):
batch_start = time.perf_counter()
async def process_single_url(url: str) -> BatchFileResult:
_logger.info("Batch processing URL: %s", url)
parsed = urlparse(url)
filename = parsed.path.split("/")[-1] or "url_content"
if body.return_json:
try:
raw_data, _ = await download_url(
url, timeout_seconds=30.0, max_size_bytes=_MAX_UPLOAD_BYTES
)
loop = asyncio.get_running_loop()
outcome = await loop.run_in_executor(_thread_pool, converter_service.convert_stream, raw_data, filename)
if isinstance(outcome, ConversionError):
return _batch_result_from_error(url, outcome)
result = _batch_result_from_ok(outcome)
if clean_content:
loop2 = asyncio.get_running_loop()
result.content = await loop2.run_in_executor(_thread_pool, text_cleaner_service.clean, outcome.markdown)
loop = asyncio.get_running_loop()
json_result = await loop.run_in_executor(
_thread_pool,
extraction_service.extract_structured,
filename,
outcome.markdown,
body.mappings,
raw_data,
)
result.json_content = json_result if "error" not in json_result else None
result.error = json_result.get("error") if "error" in json_result else None
return result
except DownloadError as exc:
if exc.is_size_error:
error = f"File exceeds {_settings.max_upload_mb} MB limit."
else:
error = f"Failed to fetch URL: {exc}"
return BatchFileResult(
filename=filename, success=False, time_ms=0, error=error,
)
loop = asyncio.get_running_loop()
outcome = await loop.run_in_executor(_thread_pool, converter_service.convert_url, url)
if isinstance(outcome, ConversionError):
return _batch_result_from_error(url, outcome)
result = _batch_result_from_ok(outcome)
if clean_content:
result.content = await loop.run_in_executor(_thread_pool, text_cleaner_service.clean, outcome.markdown)
return result
tasks = [process_single_url(url) for url in body.urls]
results = await asyncio.gather(*tasks)
total_ms = round((time.perf_counter() - batch_start) * 1000, 3)
succeeded = sum(1 for r in results if r.success)
_logger.info("Batch URLs completed. Succeeded: %s/%s", succeeded, len(results))
return BatchResponse(
total=len(results),
succeeded=succeeded,
failed=len(results) - succeeded,
total_time_ms=total_ms,
results=results,
)
|