"""Interview intelligence: tracks, question generation, answer evaluation, follow-ups, session summaries.""" import json import logging import os import tempfile from typing import Dict, List, Optional import librosa import numpy as np import process_interview as pi logger = logging.getLogger("evalbot.ai") # -------------------------------------------------------------------------------------- # Catalog # -------------------------------------------------------------------------------------- TRACKS = [ # id, English, Arabic, group ("data_engineering", "Data Engineering", "هندسة البيانات", "tech"), ("data_analysis", "Data Analysis / BI", "تحليل البيانات", "tech"), ("data_science", "Data Science", "علم البيانات", "tech"), ("ai_ml", "AI / Machine Learning", "الذكاء الاصطناعي وتعلم الآلة", "tech"), ("backend", "Backend Development", "تطوير الـ Backend", "tech"), ("frontend", "Frontend Development", "تطوير الـ Frontend", "tech"), ("fullstack", "Full-Stack Development", "تطوير Full-Stack", "tech"), ("mobile", "Mobile Development (Android / iOS / Flutter)", "تطوير تطبيقات الموبايل", "tech"), ("devops_cloud", "DevOps & Cloud", "DevOps والحوسبة السحابية", "tech"), ("cybersecurity", "Cybersecurity", "الأمن السيبراني", "tech"), ("qa_testing", "Software Testing / QA", "اختبار البرمجيات", "tech"), ("embedded", "Embedded Systems & IoT", "الأنظمة المدمجة وإنترنت الأشياء", "tech"), ("networking", "Networking & IT Infrastructure", "الشبكات والبنية التحتية", "tech"), ("it_support", "IT Support", "الدعم الفني", "tech"), ("ui_ux", "UI / UX Design", "تصميم واجهات وتجربة المستخدم", "design"), ("graphic_design", "Graphic Design", "التصميم الجرافيكي", "design"), ("product_management", "Product Management", "إدارة المنتجات", "business"), ("project_management", "Project Management", "إدارة المشروعات", "business"), ("business_analysis", "Business Analysis", "تحليل الأعمال", "business"), ("marketing", "Marketing & Digital Marketing", "التسويق والتسويق الرقمي", "business"), ("sales", "Sales & Business Development", "المبيعات وتطوير الأعمال", "business"), ("hr", "Human Resources", "الموارد البشرية", "business"), ("finance", "Accounting & Finance", "المحاسبة والمالية", "business"), ("customer_service", "Customer Service", "خدمة العملاء", "business"), ("operations", "Operations & Supply Chain", "العمليات وسلاسل الإمداد", "business"), ("content", "Content Writing & Communication", "كتابة المحتوى والتواصل", "business"), ("teaching", "Teaching & Training", "التدريس والتدريب", "other"), ("engineering", "Engineering (Civil / Mechanical / Electrical)", "الهندسة (مدني / ميكانيكا / كهرباء)", "other"), ("healthcare", "Healthcare", "الرعاية الصحية", "other"), ("general", "General / Any role", "عام / أي وظيفة", "other"), ] TRACK_NAMES = {t[0]: t[1] for t in TRACKS} LEVELS = {"intern": "Internship", "junior": "Junior (0-2 years)", "mid": "Mid-level (2-5 years)", "senior": "Senior (5+ years)", "lead": "Lead / Manager"} TYPES = {"mixed": "a balanced mix of technical, behavioral and motivation questions", "technical": "mostly technical / role-specific knowledge questions, plus one opening question", "behavioral": "behavioral questions (STAR) about past experiences, teamwork, conflict, failure, leadership", "hr": "HR / fit questions: motivation, career goals, culture fit, strengths and weaknesses, salary expectations", "case": "problem-solving / case / scenario questions where the candidate must reason out loud"} def catalog() -> Dict: return {"tracks": [{"id": i, "en": en, "ar": ar, "group": g} for i, en, ar, g in TRACKS], "levels": list(LEVELS), "types": list(TYPES)} def role_name(track: str, custom_role: str = "") -> str: return (custom_role or "").strip()[:80] or TRACK_NAMES.get(track, "General / Any role") # -------------------------------------------------------------------------------------- # Question generation # -------------------------------------------------------------------------------------- QUESTION_SCHEMA = { "type": "object", "properties": {"questions": {"type": "array", "items": {"type": "object", "properties": { "text": {"type": "string"}, "kind": {"type": "string", "enum": ["opening", "technical", "behavioral", "situational", "motivation", "closing"]}, "skill": {"type": "string", "description": "the competency this question tests, 1-3 words"}, "rubric": {"type": "string", "description": "what a strong answer includes, 1-2 sentences"}}, "required": ["text", "kind", "skill", "rubric"]}}}, "required": ["questions"], } def generate_questions(track: str, level: str, itype: str, count: int = 6, custom_role: str = "", cv_text: str = "", jd_text: str = "") -> List[Dict]: role = role_name(track, custom_role) ctx = "" if jd_text.strip(): ctx += f"\nJob description (tailor questions to it):\n{jd_text.strip()[:4000]}\n" if cv_text.strip(): ctx += f"\nCandidate CV (ask about their real experience where relevant):\n{cv_text.strip()[:4000]}\n" prompt = f"""You are an expert interviewer. Write {count} spoken interview questions for this interview: - Role: {role} - Level: {LEVELS.get(level, level)} - Interview style: {TYPES.get(itype, TYPES['mixed'])} {ctx} Rules: - Questions are asked out loud, so keep each one to one or two clear sentences; no code to read, no lists. - Calibrate difficulty to the level. Technical questions must be specific to the role (real tools, concepts, trade-offs). - Each question must be answerable in about 1-2 minutes of speaking. - Start with a short opening question unless the style is purely technical; avoid near-duplicates. - Write the questions in English. """ res = pi._gemini_json(prompt, QUESTION_SCHEMA, temperature=0.8) qs = [q for q in (res or {}).get("questions", []) if q.get("text", "").strip()][:count] if not qs: # offline fallback: built-in bank bank = pi.QUESTION_BANK["General"] + pi.QUESTION_BANK["Behavioral"] qs = [{"text": t, "kind": "behavioral", "skill": "communication", "rubric": ""} for t in bank[:count]] return qs # -------------------------------------------------------------------------------------- # Answer evaluation (one recorded answer) # -------------------------------------------------------------------------------------- SOFT_SKILLS = ["communication", "teamwork", "conflict_resolution", "time_management", "adaptability", "leadership", "problem_solving", "emotional_intelligence"] ANSWER_SCHEMA = { "type": "object", "properties": { "score": {"type": "integer", "description": "0-10"}, "relevance": {"type": "string", "enum": ["on_topic", "partially", "off_topic"]}, "summary": {"type": "string"}, "strengths": {"type": "array", "items": {"type": "string"}}, "improvements": {"type": "array", "items": {"type": "string"}}, "delivery_tip": {"type": "string"}, "improved_answer": {"type": "string"}, "followup_question": {"type": "string", "description": "a natural probing follow-up, or empty string if not needed"}, "rubric": {"type": "object", "properties": { "clarity": {"type": "integer", "description": "0-10, how clearly the thoughts are articulated"}, "example_quality": {"type": "integer", "description": "0-10, relevance and depth of real examples"}, "structure": {"type": "integer", "description": "0-10, logical flow (STAR where it fits)"}, "outcome": {"type": "integer", "description": "0-10, a concrete result, resolution or learning"}}, "required": ["clarity", "example_quality", "structure", "outcome"]}, "soft_skills": {"type": "array", "items": {"type": "string", "enum": SOFT_SKILLS}, "description": "soft skills this answer gives real evidence of (may be empty)"}, }, "required": ["score", "relevance", "summary", "strengths", "improvements", "delivery_tip", "improved_answer", "followup_question", "rubric", "soft_skills"], } def transcribe_answer(audio_bytes: bytes, suffix: str = ".webm") -> Dict: """Transcribe one answer and compute its voice metrics. Returns transcript, words, duration, voice.""" with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: tmp.write(audio_bytes) src = tmp.name wav = None try: wav = pi.convert_to_wav(src) y, sr = librosa.load(wav, sr=16000) if len(y) < sr: return {"transcript": "", "duration_s": len(y) / sr, "voice": {"error": "too short"}, "silent": True} tr = pi.transcribe(wav, speakers_expected=None, diarize=True) words = tr.get("words") or [] text = (tr.get("text") or "").strip() if not words or not text: return {"transcript": "", "duration_s": len(y) / sr, "voice": {"error": "no speech"}, "silent": True} start, end = words[0]["start"], words[-1]["end"] utt = {"role": "Interviewee", "start": start, "end": end, "text": text, "prosodic_features": pi.extract_prosodic_features(y, sr, start, end)} voice = pi.analyze_interviewee_voice(y, sr, [utt]) # integrity: a second voice speaking a substantial part of the answer counts: Dict[str, int] = {} for w in words: if w.get("speaker"): counts[w["speaker"]] = counts.get(w["speaker"], 0) + 1 ranked = sorted(counts.values(), reverse=True) other = sum(ranked[1:]) voice["speakers"] = len(counts) or 1 voice["second_voice"] = bool(other >= 10 and other / max(1, len(words)) >= 0.25) return {"transcript": text, "duration_s": round((end - start) / 1000, 1), "voice": voice, "silent": False, "speech_start_ms": start, "speech_end_ms": end} finally: for p in (src, wav): if p and os.path.exists(p): os.remove(p) def evaluate_answer(question: Dict, transcript: str, voice: Dict, role: str, level: str, lang: str, allow_followup: bool) -> Dict: metrics = {k: voice.get(k) for k in ("speaking_rate", "filler_ratio", "interpretation")} if "error" not in voice else {} prompt = f"""You are a fair, encouraging but honest interview coach. A candidate for "{role}" ({LEVELS.get(level, level)}) answered this interview question out loud. Evaluate the answer. Question: {question['text']} What a strong answer includes: {question.get('rubric') or 'n/a'} Answer (speech-to-text transcript): {transcript} Voice metrics: {json.dumps(metrics)} - score 0-10 for content quality: relevance, correctness, depth, structure (STAR where it fits), specificity. - 2-4 strengths and 2-4 improvements, one sentence each, concrete and tied to what they said. - delivery_tip: one tip from the voice metrics. - improved_answer: a stronger version (4-6 sentences, first person) that keeps the candidate's own facts and uses placeholders like [number] for missing details. Keep it in English. - rubric: clarity, example_quality, structure and outcome, each 0-10 (for purely technical answers, judge example_quality by the concreteness of the examples and outcome by whether the answer reaches a clear conclusion). - soft_skills: which of {", ".join(SOFT_SKILLS)} the answer shows real evidence of (empty list if none). - followup_question: {"one short probing follow-up a real interviewer would ask about something specific the candidate said (in English), or an empty string if the answer was complete" if allow_followup else "always an empty string"}. {pi._language_rule(lang)}""" res = pi._gemini_json(prompt, ANSWER_SCHEMA) if not res: return {"score": None, "relevance": None, "summary": "", "strengths": [], "improvements": [], "delivery_tip": "", "improved_answer": "", "followup_question": "", "ai_generated": False} res["score"] = int(max(0, min(10, res.get("score", 0)))) res["ai_generated"] = True if not allow_followup: res["followup_question"] = "" return res # -------------------------------------------------------------------------------------- # Session summary # -------------------------------------------------------------------------------------- SUMMARY_SCHEMA = { "type": "object", "properties": { "headline": {"type": "string", "description": "one-sentence overall verdict"}, "executive_summary": {"type": "array", "items": {"type": "string"}}, "strengths": {"type": "array", "items": {"type": "object", "properties": { "competency": {"type": "string"}, "explanation": {"type": "string"}, "evidence": {"type": "string"}}, "required": ["competency", "explanation", "evidence"]}}, "weaknesses": {"type": "array", "items": {"type": "object", "properties": { "area": {"type": "string"}, "feedback": {"type": "string"}}, "required": ["area", "feedback"]}}, "role_fit": {"type": "array", "items": {"type": "string"}}, "recommendations": {"type": "array", "items": {"type": "object", "properties": { "category": {"type": "string", "enum": ["Communication Skills", "Content Delivery", "Professional Presentation"]}, "recommendation": {"type": "string"}, "how_to": {"type": "string"}}, "required": ["category", "recommendation", "how_to"]}}, "learning_plan": {"type": "array", "items": {"type": "object", "properties": { "topic": {"type": "string"}, "why": {"type": "string"}, "how": {"type": "string"}}, "required": ["topic", "why", "how"]}}, "next_steps": {"type": "array", "items": {"type": "string"}}, "hiring_signal": {"type": "string", "enum": ["strong_yes", "yes", "maybe", "no"]}, "soft_skills": {"type": "array", "items": {"type": "object", "properties": { "skill": {"type": "string", "enum": SOFT_SKILLS}, "score": {"type": "integer", "description": "0-10"}, "comment": {"type": "string", "description": "one sentence with evidence"}}, "required": ["skill", "score", "comment"]}, "description": "only the soft skills the interview gave evidence about"}, "nonverbal": {"type": "string", "description": "2-3 sentences on non-verbal behaviour from the camera metrics, or empty string if none"}, }, "required": ["headline", "executive_summary", "strengths", "weaknesses", "role_fit", "recommendations", "learning_plan", "next_steps", "hiring_signal", "soft_skills", "nonverbal"], } def summarize_session(role: str, level: str, lang: str, answers: List[Dict], voice: Dict, calibrated: Optional[Dict], for_recruiter: bool, camera: Optional[Dict] = None) -> Optional[Dict]: qa = "\n\n".join(f"[{i + 1}] Q: {a['question']}\nA: {a.get('transcript') or '(no answer)'}\n" f"Score: {a.get('score')}/10; rubric: {json.dumps((a.get('feedback') or {}).get('rubric') or {})}" for i, a in enumerate(answers)) cam = "" if camera: cam = ("Camera metrics (computed on the candidate's device): " + json.dumps( {k: camera.get(k) for k in ("face_visible", "eye_contact", "smile", "emotions", "dominant_emotion")}) + "\nUse them only for the 'nonverbal' field, cautiously: expressions are not reliable signs of ability.") cal = "" if calibrated: cal = "Human-calibrated delivery scores (percentile vs 138 rated interviews): " + ", ".join( f"{t['trait']} {t['percentile']}th" for t in calibrated["traits"]) vm = {k: voice.get(k) for k in ("speaking_rate", "filler_ratio", "interpretation")} if voice and "error" not in voice else {} audience = ("The report is read by the hiring team." if for_recruiter else "The report is read by the candidate, who wants to improve; next_steps are their practice steps.") prompt = f"""You are EvalBot, a senior interviewer and career coach. Write the final evaluation of this interview. Role: {role} ({LEVELS.get(level, level)}). {audience} - Base every point on the answers below (quote briefly as evidence) and the metrics. Do not invent facts. - 3 bullets executive summary, 3-6 strengths, 3-6 weaknesses, 3 role-fit bullets, exactly 3 recommendations (one per category), a 3-5 item learning_plan of topics to study for this role, and 3-5 next_steps. - hiring_signal: your recommendation for a {LEVELS.get(level, level)} {role} position. - soft_skills: score 0-10 only those of {", ".join(SOFT_SKILLS)} the answers gave evidence about, each with evidence. - nonverbal: {"2-3 sentences from the camera metrics" if camera else "an empty string (no camera)"}. {pi._language_rule(lang)} Voice metrics: {json.dumps(vm)} {cal} {cam} Interview: {qa} """ return pi._gemini_json(prompt, SUMMARY_SCHEMA) # -------------------------------------------------------------------------------------- # Session-level audio features for calibrated scores # -------------------------------------------------------------------------------------- def session_audio(answer_audio: List[bytes]) -> Optional[tuple]: """Concatenate answer recordings into one 16 kHz signal with candidate utterances.""" ys, utts, t = [], [], 0 for blob in answer_audio: with tempfile.NamedTemporaryFile(suffix=".audio", delete=False) as tmp: tmp.write(blob) src = tmp.name try: wav = pi.convert_to_wav(src) y, sr = librosa.load(wav, sr=16000) os.remove(wav) finally: os.remove(src) iv = librosa.effects.split(y, top_db=35) if len(iv): y = y[iv[0][0]: iv[-1][1]] # trim leading/trailing silence if len(y) < 16000: continue utts.append({"role": "Interviewee", "start": t, "end": t + int(len(y) / 16), "text": ""}) ys.append(y) gap = np.zeros(8000, dtype=y.dtype) ys.append(gap) t += int(len(y) / 16) + 500 if not ys: return None return np.concatenate(ys), 16000, utts