from __future__ import annotations import tempfile import uuid import httpx from datetime import datetime, timezone from pathlib import Path from typing import Any, List, Optional from contextlib import asynccontextmanager from fastapi import FastAPI, File, Form, UploadFile from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from starlette.exceptions import HTTPException as StarletteHTTPException from services import job_store from services.cv_chunker import chunk_cv from services.cv_converter import CVConverter from services.job_matcher import JobInput, JobMatcher from services.workers import CvWorker, QueueManager, CvTask from services.taxonomy_client import load_taxonomy # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- MAX_FILE_BYTES = 20 * 1024 * 1024 # 20 MB # Magic byte signatures for MIME sniffing _MIME_SIGS = { b"%PDF": ".pdf", b"PK\x03\x04": ".docx", # DOCX (ZIP-based) b"\xd0\xcf\x11\xe0": ".doc", # Legacy OLE compound (DOC, XLS) } _ALLOWED_EXT = (".pdf", ".docx", ".doc") def _check_mime(file_bytes: bytes, declared_ext: str) -> bool: """Return True when magic bytes agree with the declared extension.""" sig = file_bytes[:4] for magic, expected_ext in _MIME_SIGS.items(): if sig.startswith(magic): # DOCX declared but is actually a ZIP: both .docx and (rarely) .doc if expected_ext == ".docx": return declared_ext in (".docx", ".doc") return declared_ext == expected_ext return True # unknown signature — let the converter decide # --------------------------------------------------------------------------- # App setup # --------------------------------------------------------------------------- converter = CVConverter() matcher = JobMatcher() # SentenceTransformer loaded once at startup cv_worker = CvWorker(converter, matcher) queue_manager = QueueManager(cv_worker, concurrency=1) @asynccontextmanager async def lifespan(app: FastAPI): load_taxonomy(max_retries=15, initial_delay=3.0) await queue_manager.start() yield await queue_manager.stop() app = FastAPI( title="Job Processor API", description="CV parsing, chunking, and job-match prediction service.", version="3.0.0", lifespan=lifespan, ) # --------------------------------------------------------------------------- # Shared response model # --------------------------------------------------------------------------- class APIResponse(BaseModel): message: str statusCode: int payload: Optional[Any] = None # --------------------------------------------------------------------------- # Exception handlers # --------------------------------------------------------------------------- @app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): return JSONResponse( status_code=400, content=APIResponse( message="Invalid request: " + str(exc.errors()), statusCode=400, ).model_dump(), ) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request, exc): return JSONResponse( status_code=exc.status_code, content=APIResponse( message=str(exc.detail), statusCode=exc.status_code, ).model_dump(), ) @app.exception_handler(Exception) async def global_exception_handler(request, exc): return JSONResponse( status_code=500, content=APIResponse( message="Internal server error: " + str(exc), statusCode=500, ).model_dump(), ) # --------------------------------------------------------------------------- # Health check # --------------------------------------------------------------------------- @app.get("/", response_model=APIResponse) def home(): return APIResponse(message="Job Processor API is running", statusCode=200) # --------------------------------------------------------------------------- # POST /process-cv (synchronous, blocking — intentional during testing) # --------------------------------------------------------------------------- @app.post("/process-cv", response_model=APIResponse) async def process_cv(file: UploadFile = File(...)): """ Upload a CV file (PDF, DOCX, DOC) and receive: - Raw Markdown text - Structured chunks (summary, experience, education, skills, projects, awards) - Detected CV metadata (title, seniority, category, years of experience) - CV quality completeness_score (0–100) - section_embeddings — pre-computed ML vectors for fast /match-cv reuse - File metadata (type, method, page count, warnings) - Per-phase timing (conversion_ms, chunking_ms, total_ms) """ request_id = str(uuid.uuid4()) if not file.filename: return APIResponse(message="No file uploaded", statusCode=400) ext = Path(file.filename).suffix.lower() if ext not in _ALLOWED_EXT: return APIResponse( message=f"Invalid file format. Allowed: {', '.join(_ALLOWED_EXT)}", statusCode=400, ) tmp_path: Optional[Path] = None t_start = datetime.now(timezone.utc) try: file_bytes = await file.read() # P1 — file size guard if len(file_bytes) > MAX_FILE_BYTES: return APIResponse( message=f"File too large. Maximum size is {MAX_FILE_BYTES // (1024*1024)} MB.", statusCode=413, ) # P4 — MIME sniff if not _check_mime(file_bytes, ext): return APIResponse( message="File content does not match the declared extension.", statusCode=400, ) # Write temp file with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp: tmp.write(file_bytes) tmp_path = Path(tmp.name) # P7 — Conversion phase timer t_conv_start = datetime.now(timezone.utc) conversion = converter.convert(tmp_path) t_conv_end = datetime.now(timezone.utc) conversion_ms = int((t_conv_end - t_conv_start).total_seconds() * 1000) if not conversion.success: return APIResponse( message=conversion.error or "Conversion failed", statusCode=422, ) # P7 — Chunking phase timer (includes ML embedding) t_chunk_start = datetime.now(timezone.utc) chunks = chunk_cv(conversion.markdown, embedder=matcher._embed) t_chunk_end = datetime.now(timezone.utc) chunking_ms = int((t_chunk_end - t_chunk_start).total_seconds() * 1000) t_end = datetime.now(timezone.utc) total_ms = int((t_end - t_start).total_seconds() * 1000) return APIResponse( message="CV processed successfully", statusCode=200, payload={ "request_id": request_id, "markdown": conversion.markdown, "cv_title": chunks["cv_title"], "seniority": chunks["seniority"], "years_experience": chunks["years_experience"], "category": chunks["category"], "completeness_score": chunks["completeness_score"], "chunks": { "summary": chunks["chunks"]["summary"], "contact": chunks["chunks"]["contact"], "links": chunks["chunks"]["links"], "skills": chunks["chunks"]["skills"], "experience": chunks["chunks"]["experience"], "education": chunks["chunks"]["education"], "projects": chunks["chunks"]["projects"], "awards": chunks["chunks"]["awards"], }, # Pre-computed ML section vectors — pass these to /match-cv or # /batch-match-cv to avoid re-encoding the CV on every match request. "section_embeddings": chunks.get("section_embeddings"), "file_type": conversion.file_type, "method_used": conversion.method_used, "is_scanned": conversion.is_scanned, "page_count": conversion.page_count, "warnings": conversion.warnings, "timing": { "conversion_ms": conversion_ms, "chunking_ms": chunking_ms, "total_ms": total_ms, }, "processed_at": t_end.isoformat(), }, ) except Exception as exc: return APIResponse( message=f"An error occurred during processing: {exc}", statusCode=500, ) finally: # P3 — guaranteed temp file cleanup if tmp_path is not None: try: tmp_path.unlink(missing_ok=True) except Exception: pass # --------------------------------------------------------------------------- # POST /process-cv-async (queued, used by Java async worker) # --------------------------------------------------------------------------- @app.post("/process-cv-async", response_model=APIResponse) async def process_cv_async( file: UploadFile = File(...), job_id: str = Form(...), callback_url: str = Form(...), callback_secret: str = Form(None), ): """ Async entry point called by the Java async worker. Returns 202 immediately; processes the CV in a queued background task. """ if not file.filename: return APIResponse(message="No file uploaded", statusCode=400) ext = Path(file.filename).suffix.lower() if ext not in _ALLOWED_EXT: return APIResponse( message=f"Invalid file format. Allowed: {', '.join(_ALLOWED_EXT)}", statusCode=400, ) file_bytes = await file.read() # Size guard (async path) if len(file_bytes) > MAX_FILE_BYTES: return APIResponse( message=f"File too large. Maximum size is {MAX_FILE_BYTES // (1024*1024)} MB.", statusCode=413, ) # MIME sniff (async path) if not _check_mime(file_bytes, ext): return APIResponse( message="File content does not match the declared extension.", statusCode=400, ) filename = file.filename job_store.create_job(job_id) task = CvTask( job_id=job_id, file_bytes=file_bytes, filename=filename, callback_url=callback_url, callback_secret=callback_secret, ) try: await queue_manager.enqueue(task) except ValueError as exc: return APIResponse( message=str(exc), statusCode=409, payload={ "job_id": job_id, "status": job_store.get_job(job_id)["status"], }, ) return APIResponse( message="CV processing started", statusCode=202, payload={"job_id": job_id, "status": "QUEUED"}, ) # --------------------------------------------------------------------------- # GET /job-status/{job_id} # --------------------------------------------------------------------------- @app.get("/job-status/{job_id}", response_model=APIResponse) def get_job_status(job_id: str): """Returns the current status of a CV processing job from the in-memory store.""" job = job_store.get_job(job_id) if job is None: return APIResponse(message="Job not found", statusCode=404) status = job["status"] if status == "QUEUED": pos = queue_manager.get_queue_position(job_id) status = f"QUEUE({pos})" if pos is not None else "QUEUE(1)" return APIResponse( message=f"Job status: {status}", statusCode=200, payload={"job_id": job_id, "status": status}, ) # --------------------------------------------------------------------------- # POST /batch-match-cv (one CV vs. array of jobs — optimised) # --------------------------------------------------------------------------- class BatchMatchCVRequest(BaseModel): """Request body for POST /batch-match-cv.""" jobs: List[JobInput] = Field( ..., min_length=1, max_length=100, description="Array of job postings (1–100).", ) cv_markdown: str cv_title: str cv_years: int cv_seniority: str cv_skills_canonical: List[str] cv_skills_technical: List[str] cv_skills_soft: List[str] cv_category: str @app.post("/batch-match-cv", response_model=APIResponse) def batch_match_cv(payload: BatchMatchCVRequest): """ Match one CV against an array of job postings in a single request. """ try: results = matcher.predict_batch( jobs=[j.model_dump() for j in payload.jobs], cv_markdown=payload.cv_markdown, cv_title=payload.cv_title, cv_years=payload.cv_years, cv_seniority=payload.cv_seniority, cv_skills_canonical=payload.cv_skills_canonical, cv_skills_technical=payload.cv_skills_technical, cv_skills_soft=payload.cv_skills_soft, cv_category=payload.cv_category, ) return APIResponse( message=f"Batch match completed: {len(results)} result(s)", statusCode=200, payload={ "matches": results, "total": len(results), }, ) except Exception as exc: return APIResponse( message=f"Batch match error: {exc}", statusCode=500, )