from __future__ import annotations """ Upload orchestrator for PDF conversion results. Workflow: 1. Convert PDF pages to JPEG or PNG images. 2. Upload each image to the File Upload Service concurrently. 3. Return clean ConversionResult with uploaded file URLs. Only JPEG and PNG output formats are supported. TIFF / merged-file mode has been removed. """ import asyncio import time from pathlib import Path from typing import List, Optional import structlog from app.core.config import settings from app.models.schemas import ( ConversionParams, ConversionResult, JobStatus, PageResult, UploadSummary, UploadedFileInfo, ) from app.services.conversion import conversion_service from app.services.file_service import upload_file_to_service logger = structlog.get_logger(__name__) def _format_str(fmt) -> str: return fmt.value if hasattr(fmt, "value") else str(fmt) def _resolve_local_path(download_url: str) -> Path: parts = download_url.lstrip("/").split("/") if len(parts) >= 5: return Path(settings.OUTPUT_DIR) / parts[3] / parts[4] raise ValueError(f"Cannot resolve local path from URL: {download_url}") def _build_filename(job_id: str, page: PageResult, ext: str) -> str: """Public-facing filename for an uploaded page. Stitched outputs are flattened to a single deterministic name; per-page images use a zero-padded sequence number. """ local = _resolve_local_path(page.download_url) if local.stem == "stitched": return f"{job_id}_stitched.{ext}" return f"{job_id}_page_{page.page_number:04d}.{ext}" async def convert_and_upload( pdf_bytes: bytes, params: ConversionParams, job_id: str, upload_mode: Optional[str] = None, ) -> ConversionResult: """ Full pipeline: 1. Render PDF pages to JPEG or PNG images. 2. Upload every image to the File Upload Service. 3. Return ConversionResult with file URLs per page. """ start = time.perf_counter() pages, _conv_ms = await conversion_service.convert(pdf_bytes, params, job_id) pages, upload_summary = await _upload_pages(pages, params, job_id) duration_ms = round((time.perf_counter() - start) * 1000, 2) return ConversionResult( job_id=job_id, status=JobStatus.COMPLETED, total_pages=len(pages), converted_pages=len(pages), pages=pages, duration_ms=duration_ms, upload_summary=upload_summary, ) async def _upload_pages( pages: List[PageResult], params: ConversionParams, job_id: str, ) -> tuple[List[PageResult], UploadSummary]: semaphore = asyncio.Semaphore(settings.UPLOAD_CONCURRENCY) fmt = _format_str(params.format) ext = "jpg" if fmt == "JPEG" else "png" content_type = "image/jpeg" if fmt == "JPEG" else "image/png" async def _upload_one(pr: PageResult) -> PageResult: async with semaphore: file_path = _resolve_local_path(pr.download_url) if not file_path.exists(): pr.upload_error = f"Local file not found: {file_path}" logger.error("upload_file_missing", page=pr.page_number, path=str(file_path)) return pr try: file_bytes = file_path.read_bytes() info = await upload_file_to_service( file_bytes, _build_filename(job_id, pr, ext), content_type ) pr.file_id = info.file_id pr.file_url = info.file_url logger.debug("page_uploaded", job_id=job_id, page=pr.page_number, file_id=info.file_id) except Exception as exc: pr.upload_error = str(exc) logger.error("page_upload_failed", job_id=job_id, page=pr.page_number, error=str(exc)) return pr updated_pages = list(await asyncio.gather(*[_upload_one(pr) for pr in pages])) succeeded = [p for p in updated_pages if p.file_id] failed = [p for p in updated_pages if p.upload_error] summary = UploadSummary( upload_mode="per_page", total_uploaded=len(succeeded), failed_uploads=len(failed), files=[ UploadedFileInfo( file_id=p.file_id or "", file_url=p.file_url or "", filename=_build_filename(job_id, p, ext), size_bytes=p.size_bytes, ) for p in succeeded ], ) logger.info("upload_complete", job_id=job_id, uploaded=len(succeeded), failed=len(failed)) return updated_pages, summary