"""AI screening orchestration: queueing, the call task, and webhook handling. Stage flow on the JobApplication (stages are rows in the data-driven PipelineStage master, seeded by add_ai_stages.py): -> AI Calling -> AI Screened (below threshold) -> AI Qualified (score >= AI_QUALIFY_THRESHOLD) """ import logging from django.conf import settings from django.utils import timezone from apps.candidates.models import Candidate from apps.pipeline.models import JobApplication, PipelineStage from . import queue from .models import AICall, CallEvaluation, CallTranscriptUtterance from .providers import get_provider from .providers.base import build_agent_variables logger = logging.getLogger(__name__) def _stage(code): return PipelineStage.objects.filter(code=code, is_active=True).first() def queue_screening(job, candidate_ids, user): """Create/refresh AICall rows for the selection and dispatch the calls. Returns (queued AICall list, per-candidate error list).""" queued, errors = [], [] for cid in candidate_ids: candidate = Candidate.objects.filter(pk=cid).first() if not candidate: errors.append(f"Candidate {cid} not found") continue if not (candidate.phone_number or "").strip(): errors.append(f"{candidate.first_name} {candidate.last_name}: no phone number") continue application, _ = JobApplication.objects.get_or_create( candidate=candidate, job=job, defaults={"stage": _stage("ai_calling") or None, "created_by": user}, ) call, _ = AICall.objects.get_or_create( candidate=candidate, job=job, defaults={"application": application, "created_by": user}, ) if call.status in (AICall.Status.QUEUED, AICall.Status.DIALING, AICall.Status.IN_PROGRESS) and call.provider_call_id: errors.append(f"{candidate.first_name} {candidate.last_name}: call already running") continue call.application = application call.status = AICall.Status.QUEUED call.error_message = "" call.agent_variables = build_agent_variables(candidate, job) call.created_by = user call.save() queued.append(call) for call in queued: run_ai_call.delay(call.id) return queued, errors @queue.background_task def run_ai_call(ai_call_id): """Background task: place one outbound call via the configured provider.""" logger.info("[AI_CALL] ── run_ai_call started for AICall id=%s ──", ai_call_id) call = AICall.objects.filter(pk=ai_call_id).first() if not call: logger.warning("[AI_CALL] AICall id=%s not found in DB, aborting.", ai_call_id) return call.attempts += 1 call.started_at = timezone.now() provider = None try: # Inside the try: an unconfigured provider must mark the call FAILED with # the real reason, not escape this task and leave it stuck at QUEUED. provider = get_provider() logger.info("[AI_CALL] Provider: %s | Candidate: %s | Job: %s | Phone: %s | Attempt: %s", provider.name, call.candidate, call.job, call.agent_variables.get('phone'), call.attempts) call.provider_call_id = provider.start_call(call) call.status = AICall.Status.DIALING call.save() logger.info("[AI_CALL] Call placed successfully — provider_call_id=%s, status=DIALING", call.provider_call_id) _move_stage(call, "ai_calling") from apps.audit_logs.services import log_activity log_activity(call.created_by, "AI_CALL_STARTED", f"AI Call started for candidate: {call.candidate.first_name} {call.candidate.last_name} (JD: {call.job.title})") except Exception as e: logger.exception("[AI_CALL] %s call FAILED for AICall %s: %s", provider.name if provider else "unconfigured", ai_call_id, e) call.status = AICall.Status.FAILED call.error_message = str(e)[:1000] call.save() def _move_stage(call, stage_code): stage = _stage(stage_code) if stage and call.application: call.application.stage = stage call.application.save(update_fields=["stage", "updated_at"]) def _by_call_id(data): """Find the AICall a callback refers to. Matches on the provider's call id first, then on our own `request_id` (`ats--`) or an `ai_call_id` in custom_data — Hunar's call-create payload carries a request_id rather than callback URLs, so a callback may echo only that. """ for key in ("call_id", "callId", "id", "call_uuid", "uuid", "sid"): value = str(data.get(key) or "") if value: call = AICall.objects.filter(provider_call_id=value).first() if call: return call custom = data.get("custom_data") if isinstance(data.get("custom_data"), dict) else {} request_id = str(data.get("request_id") or custom.get("request_id") or "") if request_id: call = AICall.objects.filter(provider_call_id=request_id).first() if call: return call # `ats--` — recover the row even if the stored # provider_call_id ended up being Hunar's own id instead. parts = request_id.split("-") if len(parts) >= 2 and parts[0] == "ats" and parts[1].isdigit(): call = AICall.objects.filter(pk=int(parts[1])).first() if call: return call ai_call_id = str(data.get("ai_call_id") or custom.get("ai_call_id") or "") if ai_call_id.isdigit(): return AICall.objects.filter(pk=int(ai_call_id)).first() return None # ─── Webhook event handlers (used by the real webhooks AND the mock) ───────── STATUS_MAP = { "not_started": AICall.Status.QUEUED, "scheduled": AICall.Status.QUEUED, "queued": AICall.Status.QUEUED, "initiated": AICall.Status.DIALING, "dialing": AICall.Status.DIALING, "ringing": AICall.Status.DIALING, "in_progress": AICall.Status.IN_PROGRESS, "answered": AICall.Status.IN_PROGRESS, "connected": AICall.Status.IN_PROGRESS, "completed": AICall.Status.COMPLETED, "not_connected": AICall.Status.NO_ANSWER, "no_answer": AICall.Status.NO_ANSWER, "noanswer": AICall.Status.NO_ANSWER, "no-answer": AICall.Status.NO_ANSWER, "busy": AICall.Status.BUSY, "cancelled": AICall.Status.CANCELLED, "canceled": AICall.Status.CANCELLED, "rejected": AICall.Status.BUSY, "declined": AICall.Status.BUSY, "failed": AICall.Status.FAILED, "error": AICall.Status.FAILED, } #: Placeholders Hunar puts in `result` for a field its agent never reached or #: could not determine. They carry no information, so they must never be read as #: a real value — mapping one onto a recommendation would present a verdict the #: agent never gave. ("NOT AVAILABLE" is what a too-short call returns for #: final_disposition; seen on a real 21-second call.) HUNAR_EMPTY_VALUES = { "", "none", "null", "n/a", "na", "-", "not covered", "not asked", "not available", "unavailable", "not determined", "undetermined", "unknown", "not applicable", } def _is_hunar_value(value) -> bool: """True when Hunar actually reported something for this field.""" if value is None: return False return str(value).strip().lower() not in HUNAR_EMPTY_VALUES #: Hunar's `result.final_disposition` -> AICall.Recommendation. The recruiting #: agent templates report their verdict here rather than under `recommendation`, #: which is why a completed call used to show "Not evaluated" in the summary #: popup. Anything reported but unrecognised becomes REVIEW — a human decides, #: and nothing is invented. HUNAR_DISPOSITION_MAP = { "qualified": AICall.Recommendation.QUALIFIED, "eligible": AICall.Recommendation.QUALIFIED, "selected": AICall.Recommendation.QUALIFIED, "shortlisted": AICall.Recommendation.QUALIFIED, "interested": AICall.Recommendation.QUALIFIED, "not qualified": AICall.Recommendation.NOT_QUALIFIED, "disqualified": AICall.Recommendation.NOT_QUALIFIED, "not eligible": AICall.Recommendation.NOT_QUALIFIED, "rejected": AICall.Recommendation.NOT_QUALIFIED, "not interested": AICall.Recommendation.NOT_QUALIFIED, "not suitable": AICall.Recommendation.NOT_QUALIFIED, } def _disposition_to_recommendation(value) -> str: """Map a Hunar disposition string onto AICall.Recommendation ("" if absent).""" if not _is_hunar_value(value): return "" key = str(value).strip().lower() mapped = HUNAR_DISPOSITION_MAP.get(key) if mapped: return mapped # Reported, but a wording we don't recognise — surface it for review rather # than silently dropping the agent's verdict. return AICall.Recommendation.REVIEW #: `result.eligibility_score` is rated out of 5 by the recruiting agent, while #: AICall.score (and the summary popup, AI Screening table, Reports and the #: AI_QUALIFY_THRESHOLD gate) is a percentage. Scale it rather than storing a #: raw 5, which would read as 5%. HUNAR_ELIGIBILITY_SCORE_MAX = 5 def _eligibility_score_to_percent(value): """Hunar's out-of-5 eligibility score as a 0-100 percentage (None if absent).""" if not _is_hunar_value(value): return None try: raw = float(str(value).strip()) except (TypeError, ValueError): return None if raw < 0: return None percent = round(raw * (100 / HUNAR_ELIGIBILITY_SCORE_MAX)) return max(0, min(100, int(percent))) def sync_hunar_call_details(call): """Fetch call details from Hunar's GET /external/v1/calls/{id}/ endpoint and update status, duration, summary, and evaluation in database.""" if not (call and call.provider_call_id): return call from .providers.hunar import HunarProvider provider = HunarProvider() # Passing our own id lets the provider recover a call whose Hunar UUID was # never captured (it matches on the custom_data.ai_call_id we sent). details = provider.get_call_details(call.provider_call_id, ai_call_id=call.id) if not details: return call # Self-heal: once Hunar's real UUID is known, store it so every later sync is # a single direct GET instead of a paged search. real_id = str(details.get("id") or "") if real_id and real_id != call.provider_call_id: logger.info("[HUNAR] AICall %s provider_call_id %r -> %r", call.id, call.provider_call_id, real_id) call.provider_call_id = real_id # 1. Update Status raw_status = str(details.get("status") or details.get("lifecycle_status") or "").lower() mapped_status = STATUS_MAP.get(raw_status) if mapped_status and call.status != AICall.Status.COMPLETED: call.status = mapped_status # 2. Update Timings & Duration dur_sec = details.get("duration_seconds") if dur_sec is not None: call.duration = int(float(dur_sec)) elif details.get("duration_minutes") is not None: call.duration = int(float(details["duration_minutes"]) * 60) if details.get("started_at") and not call.started_at: try: call.started_at = timezone.datetime.fromisoformat(details["started_at"].replace("Z", "+00:00")) except Exception: pass if details.get("ended_at") and not call.ended_at: try: call.ended_at = timezone.datetime.fromisoformat(details["ended_at"].replace("Z", "+00:00")) except Exception: pass # 3. Update Result / Evaluation / Summary result = details.get("result") if isinstance(details.get("result"), dict) else {} # Hunar returns the narrative under result.call_summary — the other keys are # kept as fallbacks so a different agent template still works. summary_text = str( result.get("call_summary") or details.get("call_summary") or details.get("summary") or result.get("summary") or "" ).strip() # Hunar is the source of truth: its summary REPLACES whatever is stored. # This previously only filled a blank field, so a call that had been through # the old local simulator kept the generated "…was screened for X. Overall # screening score N/100…" text and Hunar's real narrative was thrown away. if summary_text: call.summary = summary_text # Hunar returns NO transcript (its record has recording_url + result only). # So any conversation turns stored against a Hunar-synced call did not come # from the provider — they are left over from the local simulator. Drop them # rather than presenting a fabricated conversation as the real call. provider_transcript = str(details.get("transcript") or "").strip() provider_utterances = details.get("utterances") if isinstance(details.get("utterances"), list) else None if provider_utterances: _store_utterances(call, provider_utterances) call.transcript = _assemble_transcript_blob(call) elif provider_transcript: call.transcript = provider_transcript else: simulated = call.utterances.count() if simulated or call.transcript: logger.info( "[HUNAR] Call %s: provider returned no transcript — discarding %s " "locally-simulated utterance(s) so no fabricated conversation is shown.", call.id, simulated, ) call.utterances.all().delete() call.transcript = "" score = result.get("overall_score") or result.get("score") or details.get("score") if score is not None and str(score).isdigit(): call.score = int(score) else: # The recruiting agent templates report `eligibility_score` out of 5 # instead of a percentage — scaled here so the Score tile, AI Screening # table, Reports and the qualify threshold all keep working on one scale. eligibility = _eligibility_score_to_percent(result.get("eligibility_score")) if eligibility is not None: call.score = eligibility rec_str = str(result.get("recommendation") or "").upper() # `final_disposition` is where the recruiting agent templates put their # verdict ("Qualified" / "Not Qualified" / …). Checked after the canonical # `recommendation` key so existing behaviour is unchanged when that is sent. disposition_rec = _disposition_to_recommendation(result.get("final_disposition")) if rec_str in (AICall.Recommendation.QUALIFIED, AICall.Recommendation.REVIEW, AICall.Recommendation.NOT_QUALIFIED): call.recommendation = rec_str elif disposition_rec: call.recommendation = disposition_rec elif call.score is not None: threshold = int(getattr(settings, "AI_QUALIFY_THRESHOLD", 60)) call.recommendation = AICall.Recommendation.QUALIFIED if call.score >= threshold else AICall.Recommendation.NOT_QUALIFIED call.save() # Create/update CallEvaluation row if result is present if result or summary_text or call.score is not None: eval_obj, _ = CallEvaluation.objects.get_or_create(call=call) if call.score is not None: eval_obj.overall_score = call.score if result.get("technical_score") is not None: eval_obj.technical_score = int(result["technical_score"]) if result.get("communication_score") is not None: eval_obj.communication_score = int(result["communication_score"]) if result.get("experience_score") is not None: eval_obj.experience_score = int(result["experience_score"]) if result.get("confidence_score") is not None: eval_obj.confidence_score = int(result["confidence_score"]) if result.get("classification"): eval_obj.classification = str(result["classification"]) if result.get("recommendation"): eval_obj.recommendation = str(result["recommendation"]) if result.get("strengths"): eval_obj.strengths = result["strengths"] if isinstance(result["strengths"], list) else [str(result["strengths"])] if result.get("weaknesses"): eval_obj.weaknesses = result["weaknesses"] if isinstance(result["weaknesses"], list) else [str(result["weaknesses"])] if summary_text: eval_obj.summary = summary_text # Everything else Hunar actually returned, kept verbatim in the existing # `rubric` JSON field (no schema change). The popup renders these as # "Key Responses" / call metadata — all provider values, nothing derived. meta = {k: details.get(k) for k in ( "recording_url", "answered_by", "engagement_status", "call_ended_by", "language", "from_phone_number", "user_speech_duration", "duration_seconds", "status", "lifecycle_status", ) if details.get(k) not in (None, "")} eval_obj.rubric = { **(eval_obj.rubric if isinstance(eval_obj.rubric, dict) else {}), "hunar_result": result, "hunar_metadata": meta, } eval_obj.save() # Move application stage if completed if call.status == AICall.Status.COMPLETED: threshold = int(getattr(settings, "AI_QUALIFY_THRESHOLD", 60)) target_stage = "ai_qualified" if (call.score or 0) >= threshold else "ai_screened" _move_stage(call, target_stage) return call #: Statuses that end the call without a successful conversation. TERMINAL_UNSUCCESSFUL = ( AICall.Status.NO_ANSWER, AICall.Status.BUSY, AICall.Status.CANCELLED, AICall.Status.FAILED, ) def handle_status_event(data): logger.info("[WEBHOOK] ── status event received ──") logger.info("[WEBHOOK] Status payload: %s", data) call = _by_call_id(data) if not call: logger.warning("[WEBHOOK] No AICall found for call_id=%s — ignoring status event.", data.get('call_id') or data.get('id')) return False status = STATUS_MAP.get(str(data.get("status", "")).lower()) logger.info("[WEBHOOK] AICall id=%s | raw_status=%s | mapped=%s", call.id, data.get('status'), status) if status: call.status = status if status in TERMINAL_UNSUCCESSFUL: call.error_message = str(data.get("reason") or data.get("error") or "")[:1000] call.ended_at = timezone.now() logger.warning("[WEBHOOK] Call ended with %s — reason: %s", status, call.error_message) call.save() else: # An unmapped status must never silently advance the call. Record it so the # value can be added to STATUS_MAP instead of being lost. logger.warning( "[WEBHOOK] Unmapped status %r for AICall id=%s — status left at %s.", data.get("status"), call.id, call.status, ) return True def _store_utterances(call, utterances): """Idempotently upsert structured utterance rows (platform batches ~5 s).""" stored = 0 for u in utterances or []: try: seq = int(u.get("sequence")) except (TypeError, ValueError): continue CallTranscriptUtterance.objects.update_or_create( call=call, sequence=seq, defaults={ "speaker": (u.get("speaker") or "AGENT").upper()[:10], "message": str(u.get("message") or ""), "started_ms": u.get("started_ms"), "ended_ms": u.get("ended_ms"), "language": (u.get("language") or "en")[:5], "interrupted": bool(u.get("interrupted")), }, ) stored += 1 return stored def _assemble_transcript_blob(call): """Flat text form of the utterances (kept on AICall for compatibility).""" lines = [ f"{'Agent' if u.speaker == CallTranscriptUtterance.Speaker.AGENT else 'Candidate'}: {u.message}" for u in call.utterances.all() ] return "\n".join(lines) def handle_transcript_event(data): logger.info("[WEBHOOK] ── transcript event received ──") logger.info("[WEBHOOK] Transcript payload: %s", data) call = _by_call_id(data) if not call: logger.warning("[WEBHOOK] No AICall found for call_id=%s — ignoring transcript event.", data.get('call_id') or data.get('id')) return False # Structured utterances (self-hosted platform / mock) if data.get("utterances"): stored = _store_utterances(call, data["utterances"]) call.transcript = _assemble_transcript_blob(call) call.save(update_fields=["transcript", "updated_at"]) logger.info("[WEBHOOK] %d utterance(s) upserted for AICall id=%s", stored, call.id) return True # Legacy flat-text chunk chunk = data.get("transcript") or data.get("text") or "" if chunk: # providers may stream partial transcripts; append idempotently if chunk not in call.transcript: call.transcript = f"{call.transcript}\n{chunk}".strip() call.save(update_fields=["transcript", "updated_at"]) logger.info("[WEBHOOK] Transcript appended for AICall id=%s (chunk length=%d)", call.id, len(chunk)) return True def _store_evaluation(call, data): """Upsert the structured CallEvaluation from a completed payload.""" fields = {} for key in ("technical_score", "communication_score", "experience_score", "confidence_score", "overall_score"): if data.get(key) is not None: try: fields[key] = max(0, min(100, int(float(data[key])))) except (TypeError, ValueError): pass if data.get("classification") in CallEvaluation.Classification.values: fields["classification"] = data["classification"] if data.get("recommendation") in CallEvaluation.Recommendation.values: fields["recommendation"] = data["recommendation"] for key in ("strengths", "weaknesses"): if isinstance(data.get(key), list): fields[key] = [str(x) for x in data[key]][:10] if data.get("summary"): fields["summary"] = str(data["summary"]) if isinstance(data.get("rubric"), dict): fields["rubric"] = data["rubric"] evaluation, _ = CallEvaluation.objects.update_or_create(call=call, defaults=fields) return evaluation def handle_completed_event(data): logger.info("[WEBHOOK] ── completed event received ──") logger.info("[WEBHOOK] Completed payload: %s", data) call = _by_call_id(data) if not call: logger.warning("[WEBHOOK] No AICall found for call_id=%s — ignoring completed event.", data.get('call_id') or data.get('id')) return False call.ended_at = timezone.now() if data.get("duration") is not None: try: call.duration = int(float(data["duration"])) except (TypeError, ValueError): pass # COMPLETED must be EARNED, never assumed. This handler previously defaulted # to COMPLETED whenever the payload's status was missing or unrecognised, so # an unanswered/rejected call was reported as a successful one. Now: # • a recognised status is used as-is; # • an unknown/absent status only counts as COMPLETED when there is positive # evidence the call actually connected (a non-zero duration, or a # transcript), otherwise it is FAILED with the raw value recorded. raw_status = str(data.get("status") or "").strip().lower() mapped = STATUS_MAP.get(raw_status) if mapped: call.status = mapped else: connected = bool(data.get("utterances") or data.get("transcript")) or bool(call.duration) call.status = AICall.Status.COMPLETED if connected else AICall.Status.FAILED logger.warning( "[WEBHOOK] Completed event carried status=%r (unmapped) for AICall id=%s — " "resolved to %s from call evidence (duration=%s, transcript=%s).", data.get("status"), call.id, call.status, call.duration, bool(data.get("utterances") or data.get("transcript")), ) if call.status == AICall.Status.FAILED and not call.error_message: call.error_message = ( f"Provider reported completion without a success status " f"({data.get('status')!r}) and the call never connected." )[:1000] if data.get("utterances"): _store_utterances(call, data["utterances"]) call.transcript = _assemble_transcript_blob(call) elif data.get("transcript"): call.transcript = str(data["transcript"]) if data.get("summary"): call.summary = str(data["summary"]) evaluation = None if isinstance(data.get("evaluation"), dict): evaluation = _store_evaluation(call, data["evaluation"]) logger.info("[WEBHOOK] Evaluation stored for AICall id=%s — overall=%s, classification=%s", call.id, evaluation.overall_score, evaluation.classification) if data.get("score") is not None: try: call.score = max(0, min(100, int(float(data["score"])))) except (TypeError, ValueError): pass elif evaluation and evaluation.overall_score is not None: call.score = evaluation.overall_score # Evaluation-level Proceed/Hold/Reject maps onto the call-level recommendation _EVAL_REC_MAP = {"Proceed": AICall.Recommendation.QUALIFIED, "Hold": AICall.Recommendation.REVIEW, "Reject": AICall.Recommendation.NOT_QUALIFIED} rec = str(data.get("recommendation") or "").upper() if rec in AICall.Recommendation.values: call.recommendation = rec elif evaluation and evaluation.recommendation in _EVAL_REC_MAP: call.recommendation = _EVAL_REC_MAP[evaluation.recommendation] elif call.score is not None and call.status == AICall.Status.COMPLETED: # Threshold-derived recommendation is OUR inference, not the provider's, so # it is only made for a call that actually connected and completed. A # call that never connected must not carry a hiring recommendation. call.recommendation = ( AICall.Recommendation.QUALIFIED if call.score >= settings.AI_QUALIFY_THRESHOLD else AICall.Recommendation.NOT_QUALIFIED ) call.save() logger.info("[WEBHOOK] AICall id=%s completed — status=%s, score=%s, recommendation=%s, duration=%s", call.id, call.status, call.score, call.recommendation, call.duration) # Push the result into the pipeline so the rest of the ATS sees it if call.status == AICall.Status.COMPLETED: _move_stage(call, "ai_qualified" if call.recommendation == AICall.Recommendation.QUALIFIED else "ai_screened") if call.application and call.score is not None: call.application.score = call.score call.application.save(update_fields=["score", "updated_at"]) from apps.audit_logs.services import log_activity log_activity(call.created_by, "AI_CALL_COMPLETED", f"AI Call completed for candidate: {call.candidate.first_name} {call.candidate.last_name} (Score: {call.score}, Rec: {call.recommendation})") return True