""" services/peoplehub_api.py ========================== Defensive HTTP client for the People Hub upstream APIs. POST {PEOPLEHUB_BASE_URL}/login/by-accesskey/{access_key} POST {PEOPLEHUB_BASE_URL}/roleplays/by-accesskey/{access_key} POST {PEOPLEHUB_BASE_URL}/codingassessment/by-accesskey/{access_key} POST {PEOPLEHUB_BASE_URL}/coding/submit POST {PEOPLEHUB_ASSESS_BASE_URL}/test/gsonTestResponse.jsp?accesskey=… POST {PEOPLEHUB_ASSESS_BASE_URL}/test/updateAssessmentScoreApi.jsp?… All endpoints are POST per upstream contract — calling them via GET returns HTTP 405. The assess base URL is *derived* from PEOPLEHUB_BASE_URL so deployments never hardcode hostnames. Notes ───── • Response shapes are treated as opaque JSON. Extractors below read fields with safe fallbacks so the rest of the platform never KeyErrors. • All calls include a configurable timeout + retry budget. • The full response is preserved verbatim so the frontend can store it in localStorage exactly as the spec requires. • No URL or credential is ever logged in cleartext at INFO level — only the path and host are logged for audit. """ from __future__ import annotations import json from dataclasses import dataclass from typing import Any from urllib.parse import quote, urlparse import httpx from config import settings from utils.logger import get_logger log = get_logger(__name__) # ── URL derivation ──────────────────────────────────────────────────────────── def _assess_base_url() -> str: """ Derive the eReKruit-assess base URL from PEOPLEHUB_BASE_URL. The two upstream surfaces live on the same host but on different application contexts: api base: https:///eReKruit/api assess base: https:///eReKruit-assess Operators only configure PEOPLEHUB_BASE_URL; we derive the assess base by replacing the /eReKruit/api context with /eReKruit-assess. Falls back to a same-host /eReKruit-assess for any non-standard base. """ base = (settings.PEOPLEHUB_BASE_URL or "").rstrip("/") if not base: return "" if "/eReKruit/api" in base: return base.replace("/eReKruit/api", "/eReKruit-assess") parsed = urlparse(base) if parsed.scheme and parsed.netloc: return f"{parsed.scheme}://{parsed.netloc}/eReKruit-assess" return base def _safe_log_url(url: str) -> str: """Strip query string and credentials from a URL before logging.""" try: p = urlparse(url) return f"{p.scheme}://{p.netloc}{p.path}" except Exception: # noqa: BLE001 return "" # ── Public result type ──────────────────────────────────────────────────────── @dataclass class ApiResult: ok: bool status_code: int data: Any # full parsed response (dict/list/scalar/str) raw: str # exact response body, for debugging error: str | None = None # human-readable error if ok=False url: str = "" def to_dict(self) -> dict: return { "ok": self.ok, "status_code": self.status_code, "data": self.data, "error": self.error, "url": self.url, } # ── Internals ───────────────────────────────────────────────────────────────── def _url(path: str, access_key: str) -> str: base = (settings.PEOPLEHUB_BASE_URL or "").rstrip("/") return f"{base}/{path.lstrip('/')}/{quote(access_key, safe='')}" _DEFAULT_HEADERS = { "Accept": "application/json", "User-Agent": "PeopleHub-Candidate-Portal/1.0", } def _post( url: str, body: dict | None = None, *, bare: bool = False, ) -> ApiResult: """Single POST attempt — wrapped by `_post_with_retry`. `bare=True` sends a body-less POST (no Content-Type, zero-length body). The People Hub assess JSPs (`updateAssessmentScoreApi.jsp` and `gsonTestResponse.jsp`) accept this form — and only this form — and reject `application/json` requests with a 500 because their internal parser tries to read form parameters. Postman's default "POST with no body" mode also produces a bare POST, which is why it works there. """ timeout = httpx.Timeout(settings.PEOPLEHUB_TIMEOUT, connect=10.0) try: with httpx.Client( timeout=timeout, verify=settings.PEOPLEHUB_VERIFY_TLS, follow_redirects=True, headers=_DEFAULT_HEADERS, ) as client: if bare: # No json/data/content kwargs → httpx sends Content-Length: 0 # and no Content-Type, matching the Postman "no body" call. resp = client.post(url) else: resp = client.post(url, json=(body or {})) except httpx.TimeoutException: return ApiResult(False, 0, None, "", "upstream timeout", url) except httpx.RequestError: # Don't leak transport-level error text into responses return ApiResult(False, 0, None, "", "upstream network error", url) raw = resp.text parsed: Any try: parsed = resp.json() except json.JSONDecodeError: parsed = raw if resp.is_success: return ApiResult(True, resp.status_code, parsed, raw, None, url) # Surface upstream status but never the upstream body verbatim — bodies # may contain stack traces or implementation hints that fail VAPT. return ApiResult( False, resp.status_code, parsed, raw, f"upstream returned HTTP {resp.status_code}", url, ) def _post_with_retry( url: str, body: dict | None = None, *, max_retries: int | None = None, bare: bool = False, ) -> ApiResult: """Retry transient failures (timeouts, 5xx) up to N times. `bare=True` sends a body-less POST every attempt (used for the assess JSP endpoints — see `_post.bare` for why). `max_retries` defaults to `settings.PEOPLEHUB_RETRIES`. Pass 0 to disable retries. Connect failures and 5xx are treated as transient. """ if max_retries is None: max_retries = settings.PEOPLEHUB_RETRIES attempts = max(1, int(max_retries) + 1) last: ApiResult | None = None for i in range(1, attempts + 1): result = _post(url, body, bare=bare) last = result if result.ok: return result transient = (result.status_code == 0) or (result.status_code >= 500) if not transient or i >= attempts: log.warning( "[PeopleHub] %s → %s (attempt %d/%d)", _safe_log_url(url), result.error, i, attempts, ) return result log.info( "[PeopleHub] %s → %s (attempt %d/%d) — retrying", _safe_log_url(url), result.error, i, attempts, ) return last # type: ignore[return-value] # ── Public API ──────────────────────────────────────────────────────────────── def login(access_key: str) -> ApiResult: """POST /login/by-accesskey/{access_key}""" url = _url("login/by-accesskey", access_key) log.info("[PeopleHub] login → %s", _safe_log_url(url)) return _post_with_retry(url) def roleplays(access_key: str) -> ApiResult: """POST /roleplays/by-accesskey/{access_key}""" url = _url("roleplays/by-accesskey", access_key) log.info("[PeopleHub] roleplays → %s", _safe_log_url(url)) return _post_with_retry(url) def coding_assessment(access_key: str) -> ApiResult: """POST /codingassessment/by-accesskey/{access_key}""" url = _url("codingassessment/by-accesskey", access_key) log.info("[PeopleHub] coding → %s", _safe_log_url(url)) return _post_with_retry(url) def get_assessment(access_key: str) -> ApiResult: """ POST {assess_base}/test/gsonTestResponse.jsp?accesskey={access_key} Sent as a *bare* POST (no body, no Content-Type) — the JSP rejects application/json with a 500 because its internal parser tries to read form parameters from the body. GET also returns 405. Postman's default "POST with no body" call matches our request exactly. """ base = _assess_base_url() if not base: return ApiResult(False, 0, None, "", "PEOPLEHUB_BASE_URL is not configured", "") url = ( f"{base}/test/gsonTestResponse.jsp" f"?accesskey={quote(access_key, safe='')}" ) log.info("[PeopleHub] get_assessment → %s", _safe_log_url(url)) return _post_with_retry(url, bare=True) def submit_assessment_score(access_key: str, testid: Any) -> ApiResult: """POST {assess_base}/test/updateAssessmentScoreApi.jsp?accesskey=…&testid=… Sent as a *bare* POST (no body, no Content-Type). Sending JSON here causes the JSP to throw 500. With a bare POST it returns: {"status":"success","accesskey":"…","testid":N,"attempt_id":…, "totalScore":…,"resultStatus":…,"message":"Test submitted successfully"} This call is the *final* submit only — it must NOT be invoked once per question. Per-question saves go through `update_assessment_answer`. """ base = _assess_base_url() if not base: return ApiResult(False, 0, None, "", "PEOPLEHUB_BASE_URL is not configured", "") url = ( f"{base}/test/updateAssessmentScoreApi.jsp" f"?accesskey={quote(str(access_key), safe='')}" f"&testid={quote(str(testid), safe='')}" ) log.info("[PeopleHub] submit_assessment_score → %s", _safe_log_url(url)) return _post_with_retry(url, bare=True) def update_assessment_answer( *, access_key: str, sectionid: Any, questionid: Any, question_flag: Any, counter: Any, answer: str, testid: Any, ) -> ApiResult: """POST {assess_base}/test/updateAssessmentApi.jsp Per-question save called on every Next click. The upstream JSP expects its parameters as application/x-www-form-urlencoded — sending JSON causes a 500. We pass them as form fields via httpx's `data=` kwarg. Expected response (loose contract — extracted defensively by callers): { "status": "success", "message": "", "answerUpdated": true, "timerUpdated": false, "isFinalSectionSubmit": false, "data": { "counter": "2865", "sectionid": "1", "questionid": "1551", "case_questionid": "0", "question_flag": "0" } } """ base = _assess_base_url() if not base: return ApiResult(False, 0, None, "", "PEOPLEHUB_BASE_URL is not configured", "") url = f"{base}/test/updateAssessmentApi.jsp" log.info("[PeopleHub] update_assessment_answer → %s", _safe_log_url(url)) form = { "accesskey": str(access_key), "sectionid": str(sectionid), "questionid": str(questionid), "question_flag": str(question_flag), "counter": str(counter), "Answer": "" if answer is None else str(answer), "testid": str(testid), } return _post_form_with_retry(url, form) def submit_coding(*, access_key: str, payload: dict) -> ApiResult: """POST {PEOPLEHUB_BASE_URL}/coding/submit Submits a single coding answer to the upstream People Hub service. Body shape (per upstream contract — matches the Postman sample): { "kind": "coding_submit", "access_key": "", "received_at": "", "request": { "access_key": "", "candidate_id": "", "question_id": , "question_text": "...", "language": "python|java|...", "code": "", "submitted_at": "" } } Response body (loose): { "status": true, "message": "Coding submission saved successfully", "data": { "id": , "kind": "coding_submit", "accessKey": "...", "receivedAt": "...", "candidateId": "...", "questionId": , "questionText": "...", ... } } """ base = (settings.PEOPLEHUB_BASE_URL or "").rstrip("/") if not base: return ApiResult(False, 0, None, "", "PEOPLEHUB_BASE_URL is not configured", "") url = f"{base}/coding/submit" log.info("[PeopleHub] submit_coding → %s", _safe_log_url(url)) # Make sure access_key inside the body matches the URL-bound key and the # envelope kind is set, so the upstream contract is always honoured even # if the caller forgets a field. body = dict(payload or {}) body["access_key"] = str(access_key) body.setdefault("kind", "coding_submit") return _post_with_retry(url, body) def save_roleplay(*, access_key: str, payload: dict) -> ApiResult: """POST {PEOPLEHUB_BASE_URL}/roleplay/save Accepts a JSON body in the form documented by the upstream contract: { "kind": "roleplay", "access_key": "", "received_at": "", "request": { "scenario": { "id": "...", "title": "..." }, "session_id": "...", "duration_seconds": 0, "turns_completed": 0, "mode": "voice|video|text", "end_reason": "completed|time_expired|abandoned" } } Response body (loose): { "status": "success", "message": "...", "saved": true } The access_key is included in the body per the upstream sample, but we also pass it as a query string for backward compatibility with older builds of the upstream service. """ base = (settings.PEOPLEHUB_BASE_URL or "").rstrip("/") if not base: return ApiResult(False, 0, None, "", "PEOPLEHUB_BASE_URL is not configured", "") url = f"{base}/roleplay/save" log.info("[PeopleHub] save_roleplay → %s", _safe_log_url(url)) # Make sure access_key inside the body matches the URL-bound key body = dict(payload or {}) body["access_key"] = str(access_key) body.setdefault("kind", "roleplay") return _post_with_retry(url, body) def save_roleplay_report(*, access_key: str, payload: dict) -> ApiResult: """POST {PEOPLEHUB_BASE_URL}/roleplay/report Called once the role play report has been generated on the candidate side. The body mirrors the upstream contract: { "kind": "roleplay", "access_key": "", "received_at": "", "request": { "scenario": { "id": "...", "title": "...", "description": "...", "context": "...", "category": "...", "cat_label": "...", "learner_role": "...", "learner_emoji": "...", "ai_character": "...", "ai_emoji": "...", "ai_personality": "...", "difficulty": "...", "turns": 8 }, "session_id": "...", "duration_seconds": 0, "turns_completed": 0, "mode": "voice|video|text", "end_reason": "completed|time_expired|abandoned", "report": { ... full generated report payload ... } } } Response body (loose): { "status": "success", "message": "...", "saved": true } """ base = (settings.PEOPLEHUB_BASE_URL or "").rstrip("/") if not base: return ApiResult(False, 0, None, "", "PEOPLEHUB_BASE_URL is not configured", "") url = f"{base}/roleplay/report" log.info("[PeopleHub] save_roleplay_report → %s", _safe_log_url(url)) # Make sure access_key inside the body matches the URL-bound key body = dict(payload or {}) body["access_key"] = str(access_key) body.setdefault("kind", "roleplay") return _post_with_retry(url, body) def _post_form(url: str, form: dict) -> ApiResult: """Single attempt POST with form-urlencoded body.""" timeout = httpx.Timeout(settings.PEOPLEHUB_TIMEOUT, connect=10.0) try: with httpx.Client( timeout=timeout, verify=settings.PEOPLEHUB_VERIFY_TLS, follow_redirects=True, headers=_DEFAULT_HEADERS, ) as client: resp = client.post(url, data=form) except httpx.TimeoutException: return ApiResult(False, 0, None, "", "upstream timeout", url) except httpx.RequestError: return ApiResult(False, 0, None, "", "upstream network error", url) raw = resp.text parsed: Any try: parsed = resp.json() except json.JSONDecodeError: parsed = raw if resp.is_success: return ApiResult(True, resp.status_code, parsed, raw, None, url) return ApiResult( False, resp.status_code, parsed, raw, f"upstream returned HTTP {resp.status_code}", url, ) def _post_form_with_retry(url: str, form: dict) -> ApiResult: attempts = max(1, int(settings.PEOPLEHUB_RETRIES) + 1) last: ApiResult | None = None for i in range(1, attempts + 1): result = _post_form(url, form) last = result if result.ok: return result transient = (result.status_code == 0) or (result.status_code >= 500) if not transient or i >= attempts: log.warning( "[PeopleHub] %s → %s (attempt %d/%d)", _safe_log_url(url), result.error, i, attempts, ) return result log.info( "[PeopleHub] %s → %s (attempt %d/%d) — retrying", _safe_log_url(url), result.error, i, attempts, ) return last # type: ignore[return-value] # ── Defensive field extractors ──────────────────────────────────────────────── # These read commonly-named fields from arbitrary response shapes so the rest # of the codebase doesn't have to guess. None of them ever raises. _STAGE_KEYS = ( "current_stage", "currentStage", "stage", "interview_stage", "interviewStage", "stage_no", "stageNo", "active_stage", "activeStage", ) _NAME_KEYS = ( "name", "candidate_name", "candidateName", "full_name", "fullName", "display_name", "displayName", "first_name", "firstName", ) _EMAIL_KEYS = ("email", "candidate_email", "candidateEmail", "email_id", "emailId") _ID_KEYS = ("id", "candidate_id", "candidateId", "participant_id", "participantId", "user_id", "userId", "uuid") _PHONE_KEYS = ("phone", "mobile", "phone_number", "phoneNumber", "contact", "contact_number", "contactNumber") _POSITION_KEYS = ("position_applied", "positionApplied", "position", "role_applied", "roleApplied", "job_role", "jobRole") _SKILLS_KEYS = ("skill_set", "skillSet", "skills", "skill_list", "skillList") def _walk_for_keys(data: Any, keys: tuple[str, ...], depth: int = 0) -> Any: """Recursively look for any of the given keys in nested dicts.""" if depth > 4 or data is None: return None if isinstance(data, dict): for k in keys: if k in data and data[k] not in (None, ""): return data[k] for v in data.values(): found = _walk_for_keys(v, keys, depth + 1) if found not in (None, ""): return found elif isinstance(data, list) and data: return _walk_for_keys(data[0], keys, depth + 1) return None def extract_current_stage(payload: Any, fallback: int | None = None) -> int | None: """Pull current_stage from anywhere in the response. Returns the integer stage if the API explicitly provided one (>=1). Returns None when the value is missing, empty, zero, or non-numeric — the frontend then routes the candidate to the "no interview scheduled" page rather than fabricating a stage. Pass an explicit `fallback` if a caller actually wants a default. """ raw = _walk_for_keys(payload, _STAGE_KEYS) if raw is None: return fallback try: s = str(raw).strip() if not s: return fallback n = int(s) return n if n >= 1 else fallback except (ValueError, TypeError): return fallback def extract_candidate_name(payload: Any, default: str = "Candidate") -> str: raw = _walk_for_keys(payload, _NAME_KEYS) return str(raw).strip() if raw else default def extract_candidate_email(payload: Any, default: str = "") -> str: raw = _walk_for_keys(payload, _EMAIL_KEYS) return str(raw).strip() if raw else default def extract_candidate_id(payload: Any, default: str = "") -> str: raw = _walk_for_keys(payload, _ID_KEYS) return str(raw).strip() if raw else default def extract_candidate_phone(payload: Any, default: str = "") -> str: raw = _walk_for_keys(payload, _PHONE_KEYS) return str(raw).strip() if raw else default def extract_position_applied(payload: Any, default: str = "") -> str: raw = _walk_for_keys(payload, _POSITION_KEYS) return str(raw).strip() if raw else default def extract_skill_set(payload: Any) -> list[str]: raw = _walk_for_keys(payload, _SKILLS_KEYS) if raw is None: return [] if isinstance(raw, list): return [str(s) for s in raw] if isinstance(raw, str): return [s.strip() for s in raw.split(",") if s.strip()] return [] def extract_candidate_summary(payload: Any) -> dict: """Single-stop dict the frontend can show on profile / toolbar. `current_stage` is None when the API has no stage assigned for this candidate; the frontend treats that as "no interview scheduled". """ return { "id": extract_candidate_id(payload), "name": extract_candidate_name(payload), "email": extract_candidate_email(payload), "phone": extract_candidate_phone(payload), "position_applied": extract_position_applied(payload), "skill_set": extract_skill_set(payload), "current_stage": extract_current_stage(payload), } def is_empty_response(payload: Any) -> bool: """True if payload is None / empty list / empty dict / empty string.""" if payload is None: return True if isinstance(payload, (list, dict, str)) and len(payload) == 0: return True # Common shape: { "data": [] } or { "items": [] } if isinstance(payload, dict): for k in ("data", "items", "roleplays", "results", "list"): if k in payload and isinstance(payload[k], (list, dict, str)) and len(payload[k]) == 0: # Only call empty if there are no other meaningful keys other = [kk for kk in payload.keys() if kk not in (k, "status", "message", "ok", "success")] if not other: return True return False