add process cv async

main.py

@@ -1,17 +1,22 @@
 from __future__ import annotations
 
 import tempfile
+import httpx
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
-from fastapi import FastAPI, File, UploadFile
+from fastapi import BackgroundTasks, 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
 
 # ---------------------------------------------------------------------------
 # App setup
@@ -25,6 +30,7 @@ app = FastAPI(
 
 converter = CVConverter()
 matcher = JobMatcher()  # SentenceTransformer loaded once at startup
+cv_worker = CvWorker(converter)
 
 
 # ---------------------------------------------------------------------------
@@ -36,20 +42,13 @@ class APIResponse(BaseModel):
     statusCode: int
     payload: Optional[Any] = None
 
-
-# ---------------------------------------------------------------------------
-# Validation error handler
-# ---------------------------------------------------------------------------
-
-from fastapi.responses import JSONResponse
-from starlette.exceptions import HTTPException as StarletteHTTPException
-
+# 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()), 
+            message="Invalid request: " + str(exc.errors()),
             statusCode=400
         ).model_dump()
     )
@@ -77,22 +76,14 @@ async def global_exception_handler(request, exc):
 
 # ---------------------------------------------------------------------------
 # Health check
-# ---------------------------------------------------------------------------
-
 @app.get("/", response_model=APIResponse)
 def home():
     return APIResponse(message="Job Processor API is running", statusCode=200)
 
 
-# ---------------------------------------------------------------------------
-# POST /process-cv
-# Accepts a PDF / DOCX / DOC file, converts it to Markdown,
-# then runs the CV chunker to return structured sections.
-# ---------------------------------------------------------------------------
 
+# POST /process-cv  (synchronous)
 _ALLOWED_EXT = (".pdf", ".docx", ".doc")
-
-
 @app.post("/process-cv", response_model=APIResponse)
 async def process_cv(file: UploadFile = File(...)):
     """
@@ -121,12 +112,10 @@ async def process_cv(file: UploadFile = File(...)):
             tmp.write(file_bytes)
             tmp_path = Path(tmp.name)
 
-        # Convert file → Markdown
         conversion = converter.convert(tmp_path)
         if not conversion.success:
             return APIResponse(message=conversion.error or "Conversion failed", statusCode=422)
 
-        # Chunk the Markdown into structured sections
         chunks = chunk_cv(conversion.markdown)
         end_time = datetime.now(timezone.utc).isoformat()
 
@@ -149,13 +138,13 @@ async def process_cv(file: UploadFile = File(...)):
                     "projects":   chunks["chunks"]["projects"],
                     "awards":     chunks["chunks"]["awards"],
                 },
-                "file_type":        conversion.file_type,
-                "method_used":      conversion.method_used,
-                "is_scanned":       conversion.is_scanned,
-                "page_count":       conversion.page_count,
-                "warnings":         conversion.warnings,
-                "start_time":       start_time,
-                "end_time":         end_time,
+                "file_type":   conversion.file_type,
+                "method_used": conversion.method_used,
+                "is_scanned":  conversion.is_scanned,
+                "page_count":  conversion.page_count,
+                "warnings":    conversion.warnings,
+                "start_time":  start_time,
+                "end_time":    end_time,
             },
         )
 
@@ -171,11 +160,78 @@ async def process_cv(file: UploadFile = File(...)):
 
 
 # ---------------------------------------------------------------------------
-# POST /match-cv
-# Accepts a structured job JSON + a CV Markdown string,
-# returns a full qualification prediction.
+# POST /process-cv-async  (NEW — used by Java async worker)
+#
+# Accepts the CV file plus job_id and callback_url as form fields.
+# Returns 202 immediately and processes the CV in a BackgroundTask.
+# On completion (success or failure) POSTs the result to callback_url.
 # ---------------------------------------------------------------------------
 
+
+@app.post("/process-cv-async", response_model=APIResponse)
+async def process_cv_async(
+    background_tasks: BackgroundTasks,
+    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.
+    Accepts the CV file + job_id + callback_url as multipart form fields.
+    Returns 202 immediately; processes the CV in a background task.
+    """
+    if not file.filename:
+        return APIResponse(message="No file uploaded", statusCode=400)
+
+    if not file.filename.lower().endswith(_ALLOWED_EXT):
+        return APIResponse(
+            message=f"Invalid file format. Allowed formats: {', '.join(_ALLOWED_EXT)}",
+            statusCode=400,
+        )
+
+    # Read bytes before the background task starts (request closes after response)
+    file_bytes = await file.read()
+    filename   = file.filename
+
+    job_store.create_job(job_id)
+
+    background_tasks.add_task(
+        cv_worker.run_cv_processing,
+        job_id,
+        file_bytes,
+        filename,
+        callback_url,
+        callback_secret,
+    )
+
+    return APIResponse(
+        message="CV processing started",
+        statusCode=202,
+        payload={"job_id": job_id, "status": "PENDING"},
+    )
+
+
+# GET /job-status/{job_id}  (Python-side status check)
+@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.
+    Primarily useful for debugging; Java tracks state authoritatively in its DB.
+    """
+    job = job_store.get_job(job_id)
+    if job is None:
+        return APIResponse(message="Job not found", statusCode=404)
+
+    return APIResponse(
+        message=f"Job status: {job['status']}",
+        statusCode=200,
+        payload={"job_id": job_id, "status": job["status"]},
+    )
+
+
+# POST /match-cv  (ORIGINAL — unchanged)
+
 class MatchCVRequest(BaseModel):
     """Request body for POST /match-cv."""
     job: JobInput = Field(..., description="Structured job posting to match against.")

requirements.txt

@@ -5,6 +5,7 @@ fastapi
 uvicorn[standard]
 pydantic
 python-multipart
+httpx
 
 # Document parsing
 pdfplumber

services/job_store.py

@@ -0,0 +1,51 @@
+"""
+Simple in-memory job store for the async CV processing pipeline.
+
+Since the Python service runs as a single process on Hugging Face Spaces,
+an in-memory dict is sufficient. Jobs are keyed by job_id (string UUID).
+
+Each entry shape:
+{
+    "status":  "PENDING" | "PROCESSING" | "COMPLETED" | "FAILED",
+    "result":  <ProcessorResponse payload dict> | None,
+    "error":   <str> | None,
+}
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Optional
+
+_lock = threading.Lock()
+_store: dict[str, dict] = {}
+
+
+def create_job(job_id: str) -> None:
+    with _lock:
+        _store[job_id] = {"status": "PENDING", "result": None, "error": None}
+
+
+def set_processing(job_id: str) -> None:
+    with _lock:
+        if job_id in _store:
+            _store[job_id]["status"] = "PROCESSING"
+
+
+def set_completed(job_id: str, result: dict) -> None:
+    with _lock:
+        _store[job_id] = {"status": "COMPLETED", "result": result, "error": None}
+
+
+def set_failed(job_id: str, error: str) -> None:
+    with _lock:
+        if job_id in _store:
+            _store[job_id]["status"] = "FAILED"
+            _store[job_id]["error"]  = error
+        else:
+            _store[job_id] = {"status": "FAILED", "result": None, "error": error}
+
+
+def get_job(job_id: str) -> Optional[dict]:
+    with _lock:
+        return _store.get(job_id)

services/workers.py

@@ -0,0 +1,112 @@
+from __future__ import annotations
+
+import tempfile
+import httpx
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+from services import job_store
+from services.cv_chunker import chunk_cv
+from services.cv_converter import CVConverter
+
+class CvWorker:
+    """Handles background tasks for CV processing."""
+
+    def __init__(self, converter: CVConverter):
+        self.converter = converter
+
+    async def run_cv_processing(
+        self,
+        job_id: str,
+        file_bytes: bytes,
+        filename: str,
+        callback_url: str,
+        callback_secret: str = None
+    ) -> None:
+        """Background task: process CV then POST result to Java callback endpoint."""
+        tmp_path: Optional[Path] = None
+        try:
+            job_store.set_processing(job_id)
+
+            start_time = datetime.now(timezone.utc).isoformat()
+
+            suffix = Path(filename).suffix.lower() if filename else ".pdf"
+            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+                tmp.write(file_bytes)
+                tmp_path = Path(tmp.name)
+
+            conversion = self.converter.convert(tmp_path)
+
+            if not conversion.success:
+                error_msg = conversion.error or "Conversion failed"
+                job_store.set_failed(job_id, error_msg)
+                await self.post_callback(callback_url, {
+                    "message":    error_msg,
+                    "statusCode": 422,
+                    "payload":    None,
+                }, callback_secret)
+                return
+
+            chunks = chunk_cv(conversion.markdown)
+            end_time = datetime.now(timezone.utc).isoformat()
+
+            payload = {
+                "markdown":         conversion.markdown,
+                "cv_title":         chunks["cv_title"],
+                "seniority":        chunks["seniority"],
+                "years_experience": chunks["years_experience"],
+                "category":         chunks["category"],
+                "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"],
+                },
+                "file_type":   conversion.file_type,
+                "method_used": conversion.method_used,
+                "is_scanned":  conversion.is_scanned,
+                "page_count":  conversion.page_count,
+                "warnings":    conversion.warnings,
+                "start_time":  start_time,
+                "end_time":    end_time,
+            }
+
+            job_store.set_completed(job_id, payload)
+
+            await self.post_callback(callback_url, {
+                "message":    "CV processed successfully",
+                "statusCode": 200,
+                "payload":    payload,
+            }, callback_secret)
+
+        except Exception as exc:
+            error_msg = f"Processing error: {exc}"
+            job_store.set_failed(job_id, error_msg)
+            await self.post_callback(callback_url, {
+                "message":    error_msg,
+                "statusCode": 500,
+                "payload":    None,
+            }, callback_secret)
+
+        finally:
+            if tmp_path and tmp_path.exists():
+                tmp_path.unlink()
+
+    async def post_callback(self, callback_url: str, body: dict, callback_secret: str = None) -> None:
+        """POST the processing result back to Java."""
+        try:
+            headers = {}
+            if callback_secret:
+                headers["X-Callback-Secret"] = callback_secret
+                
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.post(callback_url, json=body, headers=headers)
+                response.raise_for_status()
+        except Exception as exc:
+            # Log but don't crash — Java will see job stay in PROCESSING and can time out
+            print(f"[CV-ASYNC] Failed to POST callback to {callback_url}: {exc}")