from __future__ import annotations from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, status from app.api.deps import get_session_id from app.core.logging import get_logger from app.models.analysis import AnalysisRecord from app.models.resume import ResumeRecord from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db from app.schemas.analysis import ( AnalysisDetail, AnalyzeRequest, AnalyzeResponse, BulletRewrite, ComponentBreakdown, Recommendation, ScoreWeights, SectionNote, ) from app.services.ats_scoring import ATSResult, run_ats_scoring logger = get_logger(__name__) router = APIRouter() # ── POST /api/analyze ───────────────────────────────────────────────────── @router.post( "/analyze", response_model=AnalyzeResponse, tags=["analysis"], summary="Analyse resume against a job description", ) async def analyze( body: AnalyzeRequest, session_id: Annotated[str, Depends(get_session_id)], db: Annotated[AsyncSession, Depends(get_db)], ) -> AnalyzeResponse: """ Run the full ATS-style analysis pipeline: 1. Fetch extracted resume text. 2. Parse sections. 3. Extract and compare keywords. 4. Compute semantic similarity. 5. Compute weighted ATS score. 6. Return structured result. Bullet rewrites are empty in Phase 2 — LLM populates them in Phase 4. """ # ── Fetch resume ────────────────────────────────────────────────────── record = await db.get(ResumeRecord, body.resume_id) if record is None or record.session_id != session_id: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Resume not found in this session.", ) logger.info( "analyze: resume_id=%s session=%s jd_chars=%d role=%r", body.resume_id, session_id, len(body.job_description), body.target_role, ) # ── Run scoring pipeline ────────────────────────────────────────────── try: result: ATSResult = run_ats_scoring( resume_text=record.extracted_text, jd_text=body.job_description, target_role=body.target_role, ) except Exception as exc: logger.exception("Scoring pipeline failed: %s", exc) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Analysis failed — please retry.", ) from exc # ── Build response ──────────────────────────────────────────────────── response = _build_response(result, record.file_name) # ── LLM Bullet Rewrites (Phase 4) ───────────────────────────────────── import asyncio from app.services.llm_service import rewrite_weak_bullets try: # Run sync Gemini call in threadpool to avoid blocking ASGI loop rewrites_raw = await asyncio.to_thread( rewrite_weak_bullets, bullets=result.parsed.experience_bullets, job_description=body.job_description, target_role=body.target_role, max_rewrites=3 ) # Map to Pydantic objects response.bullet_rewrites = [BulletRewrite(**rw) for rw in rewrites_raw] except Exception as exc: logger.warning("LLM rewrite failed: %s", exc) # Non-fatal, just leave rewrites empty pass # ── Persist ─────────────────────────────────────────────────────────── ar = AnalysisRecord( resume_id=body.resume_id, session_id=session_id, job_description=body.job_description, target_role=body.target_role, result=response.model_dump(), ) db.add(ar) await db.commit() await db.refresh(ar) response.analysis_id = ar.id logger.info( "analyze: analysis_id=%s overall=%d latency_ms=%d", ar.id, result.overall_score, result.latency_ms, ) return response # ── GET /api/analysis/{analysis_id} ────────────────────────────────────── @router.get( "/analysis/{analysis_id}", response_model=AnalysisDetail, tags=["analysis"], summary="Get past analysis result", ) async def get_analysis( analysis_id: str, session_id: Annotated[str, Depends(get_session_id)], db: Annotated[AsyncSession, Depends(get_db)], ) -> AnalysisDetail: """Fetch an analysis result generated in this session.""" ar = await db.get(AnalysisRecord, analysis_id) if ar is None or ar.session_id != session_id: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Analysis not found in this session.", ) return AnalysisDetail( analysis_id=ar.id, resume_id=ar.resume_id, created_at=ar.created_at, result=AnalyzeResponse(**ar.result), ) # ── Private helpers ─────────────────────────────────────────────────────── def _role_fit(score: int) -> tuple[str, str]: if score >= 80: return ( "Strong fit", "Resume aligns well with the JD across skills and experience. " "Fix the missing keywords and quantify a few bullets to push into shortlist range.", ) if score >= 65: return ( "Moderate fit", "Core skills overlap but presentation and specificity are weak. " "Prioritise the top-3 missing keywords and add measurable outcomes.", ) return ( "Needs work", "Significant gaps in skills or presentation. " "Address missing keywords first, then rewrite bullets around measurable outcomes.", ) def _recommendations(result: ATSResult, file_name: str) -> list[Recommendation]: recs: list[Recommendation] = [] comp = result.components kw = result.keywords if kw.missing: top3 = ", ".join(kw.missing[:3]) recs.append(Recommendation( priority="high", category="keywords", message=f"Add missing keywords: {top3}. These appear in the JD but not in your resume.", )) if comp.experience_alignment < 60: recs.append(Recommendation( priority="high", category="experience", message="Quantify at least 5 bullets with numbers, percentages, or impact metrics.", )) if comp.semantic_similarity < 55: recs.append(Recommendation( priority="high", category="alignment", message="Rephrase your summary and experience to mirror the language and priorities in the JD.", )) if not result.parsed.summary: recs.append(Recommendation( priority="medium", category="structure", message="Add a 2-line summary at the top tailored to this specific role.", )) if not result.parsed.certifications_raw: recs.append(Recommendation( priority="low", category="certifications", message="Even one relevant certification improves ATS signal and fills a structural gap.", )) if kw.weak: recs.append(Recommendation( priority="medium", category="keywords", message=f"Strengthen underused keywords: {', '.join(kw.weak[:3])}. Mention them in context.", )) return recs[:6] # cap at 6 recommendations def _build_response(result: ATSResult, file_name: str) -> AnalyzeResponse: verdict, summary = _role_fit(result.overall_score) comp = result.components return AnalyzeResponse( analysis_id="", # filled in after persist overall_score=result.overall_score, weights=ScoreWeights(), components=ComponentBreakdown( keyword_coverage=comp.keyword_coverage, semantic_similarity=comp.semantic_similarity, skills_overlap=comp.skills_overlap, experience_alignment=comp.experience_alignment, resume_quality=comp.resume_quality, ), matched_keywords=result.keywords.matched, missing_keywords=result.keywords.missing, weak_keywords=result.keywords.weak, section_notes=[ SectionNote( section=sn.section, status=sn.status, note=sn.note, score=sn.score, ) for sn in result.section_notes ], recommendations=_recommendations(result, file_name), bullet_rewrites=[], # Phase 4 (LLM) role_fit_verdict=verdict, role_fit_summary=summary, parsed_sections=result.parsed.to_dict(), latency_ms=result.latency_ms, )