""" Healthie webhook event handlers. Each handler receives a WebhookPayload, stores the raw ingestion record, then fetches the full resource via GraphQL and normalizes into Core DB models. """ import logging from typing import Any import requests # Module-level singleton for connection pooling _healthie_client = None def _get_healthie_client(): """Lazy singleton — reuses requests.Session for connection pooling.""" global _healthie_client if _healthie_client is None: from access_core.services.healthie import HealthieClient _healthie_client = HealthieClient() return _healthie_client # NOTE: These imports are placed after _get_healthie_client() to avoid circular # dependency issues at module load time (E402/E305 suppressed intentionally). from access_core.intake.handlers import register_handler from access_core.intake.models import RawIngestionRecord from access_core.services.healthie_webhook import WebhookPayload logger = logging.getLogger(__name__) # ------------------------------------------------------------------ # Helpers # ------------------------------------------------------------------ def _store_raw_ingestion( payload: WebhookPayload, processing_status: str = "received" ) -> RawIngestionRecord: """Store the raw webhook payload as a RawIngestionRecord (idempotent on retry).""" record, created = RawIngestionRecord.objects.get_or_create( source_system="healthie", source_record_id=payload.resource_id, defaults={ "raw_payload": payload.raw_body, "webhook_signature_valid": True, "processing_status": processing_status, }, ) if not created and record.processing_status in ("received", "processing", "failed"): # Retry: reset to processing record.processing_status = "processing" record.raw_payload = payload.raw_body record.save() return record def _check_graphql_errors(result: dict, context: str = "") -> bool: """Check for GraphQL errors in the response. Returns True if there are errors.""" errors = result.get("errors") if errors: logger.error(f"GraphQL error {context}: {errors}") return True return False def _fetch_healthie_resource(resource_id: str, resource_type: str) -> dict: """ Fetch the full resource from Healthie via GraphQL. Healthie webhooks are thin — only resource_id + event_type. We must make a follow-up GraphQL query to get the full data. Error handling: - 4xx HTTP errors (bad/expired API key, missing resource) are permanent — return {} so the caller marks the raw record "failed" and we ack with 200 (no point retrying until an operator fixes the credential). - Transient failures (connection error, timeout, Healthie 5xx) propagate to the view, which returns 500 to trigger Healthie's webhook retry logic. """ client = _get_healthie_client() queries = { "FormAnswerGroup": _query_form_answer_group, "User": _query_user, "Document": _query_document, "Entry": _query_entry, "CarePlan": _query_care_plan, } query_fn = queries.get(resource_type) if not query_fn: logger.warning(f"No GraphQL query defined for resource type: {resource_type}") return {} try: result = query_fn(client, resource_id) except requests.exceptions.HTTPError as exc: status = getattr(exc.response, "status_code", None) if status is not None and 400 <= status < 500: # HEALTHIE_FETCH_PERMANENT_FAILURE is a stable token matched by the # CloudWatch metric filter in infra/modules/ecs — keep it verbatim. logger.error( f"HEALTHIE_FETCH_PERMANENT_FAILURE {resource_type}({resource_id}): " f"HTTP {status}. Marking ingestion as failed; reprocess with " f"retry_failed_ingestions once the Healthie API key is fixed." ) return {} raise if _check_graphql_errors(result, f"{resource_type}({resource_id})"): return {} return result def reprocess_raw_record(record: RawIngestionRecord) -> str: """ Re-run the registered handler for a stored raw webhook record. Used by the retry_failed_ingestions command and the Django admin action to recover records that failed (e.g., Healthie API key expiry). The handler's own _store_raw_ingestion call resets the record to "processing" first, so this is safe against received/processing/failed rows. Returns one of: "reprocessed", "skipped_no_handler", "skipped_bad_payload". Raises on handler failure (caller decides how to record it). """ body = record.raw_payload or {} event_type = str(body.get("event_type", "")) if not event_type: logger.warning(f"Reprocess skipped for {record}: raw payload has no event_type") return "skipped_bad_payload" from access_core.intake.handlers import get_handler handler_fn = get_handler(event_type) if not handler_fn: logger.warning(f"Reprocess skipped for {record}: no handler for {event_type}") return "skipped_no_handler" payload = WebhookPayload( resource_id=str(body.get("resource_id", record.source_record_id)), resource_id_type=str(body.get("resource_id_type", "")), event_type=event_type, changed_fields=body.get("changed_fields", []), raw_body=record.raw_payload, ) handler_fn(payload) return "reprocessed" # ------------------------------------------------------------------ # GraphQL query functions # ------------------------------------------------------------------ def _query_form_answer_group(client: Any, form_answer_group_id: str) -> dict: """Fetch a FormAnswerGroup with all its answers.""" query = """ query FormAnswerGroup($id: ID) { formAnswerGroup(id: $id) { id finished created_at user { id first_name last_name email phone_number dob gender } custom_module_form { id name } form_answers { id label answer displayed_answer custom_module { id label mod_type required } } } } """ return client.query(query, variables={"id": form_answer_group_id}) def _query_user(client: Any, user_id: str) -> dict: """Fetch a Healthie user (patient).""" query = """ query User($id: ID) { user(id: $id) { id first_name last_name email phone_number dob gender } } """ return client.query(query, variables={"id": user_id}) def _query_document(client: Any, document_id: str) -> dict: """Fetch a Healthie document.""" query = """ query Document($id: ID) { document(id: $id) { id display_name file_content_type expiring_url owner { id email } } } """ return client.query(query, variables={"id": document_id}) def _query_entry(client: Any, entry_id: str) -> dict: """Fetch a Healthie journal entry.""" query = """ query Entry($id: ID) { entry(id: $id) { id type category metric_stat description created_at client_id added_by_user { id is_patient } } } """ return client.query(query, variables={"id": entry_id}) def _query_care_plan(client: Any, care_plan_id: str) -> dict: """Fetch a Healthie care plan.""" query = """ query CarePlan($id: ID) { carePlan(id: $id) { id name description is_active is_template patient_id recommendations { id recommendation_body recommendation_type } } } """ return client.query(query, variables={"id": care_plan_id}) # ------------------------------------------------------------------ # Event handlers # ------------------------------------------------------------------ @register_handler("form_answer_group.created") def handle_form_answer_group_created(payload: WebhookPayload) -> str: """ Handle form submission webhook. This is the primary intake event — a patient submitted a form (intake, daily check-in, etc.). Process: 1. Store raw webhook payload 2. Fetch full FormAnswerGroup via GraphQL 3. Check if this is the intake form (form ID match) 4. If intake: create/update IntakeRecord + sub-models 5. If daily check-in: create DailyCheckinSubmission + responses """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "FormAnswerGroup") if not full_data.get("data", {}).get("formAnswerGroup"): raw_record.processing_status = "failed" raw_record.save() return "no_data" form_group = full_data["data"]["formAnswerGroup"] form_template_id = form_group.get("custom_module_form", {}).get("id", "") form_answers = form_group.get("form_answers", []) patient_data = form_group.get("user", {}) # Import the form ID from settings from django.conf import settings intake_form_id = getattr(settings, "HEALTHIE_INTAKE_FORM_ID", "2360739") if str(form_template_id) == str(intake_form_id): # This is the intake form — normalize into intake models _normalize_intake_form(raw_record, form_group, form_answers, patient_data) else: # This is a daily check-in or other form _normalize_other_form(raw_record, form_group, form_answers) raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("patient.created") @register_handler("patient.updated") def handle_patient_event(payload: WebhookPayload) -> str: """ Handle patient creation/update webhook. On create: Create IntakeRecord if not exists, link via external_system_identifiers. On update: Check changed_fields for status changes. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "User") if not full_data.get("data", {}).get("user"): raw_record.processing_status = "failed" raw_record.save() return "no_data" user = full_data["data"]["user"] _link_patient_to_intake(user, payload.event_type, payload.changed_fields) raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("document.created") @register_handler("document.updated") def handle_document_event(payload: WebhookPayload) -> str: """ Handle document upload/update webhook. Fetch full document via GraphQL, store reference in clinical_documents. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "Document") if not full_data.get("data", {}).get("document"): raw_record.processing_status = "failed" raw_record.save() return "no_data" document = full_data["data"]["document"] _store_document_reference(document) raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("metric_entry.created") def handle_metric_entry_created(payload: WebhookPayload) -> str: """ Handle metric entry (BP, weight, A1C, etc.) webhook. Fetch entry via GraphQL, create clinical_observations + oap_measure_observations. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "Entry") if not full_data.get("data", {}).get("entry"): raw_record.processing_status = "failed" raw_record.save() return "no_data" entry = full_data["data"]["entry"] _store_metric_observation(entry, raw_record) raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("entry.created") def handle_entry_created(payload: WebhookPayload) -> str: """ Handle generic journal entry webhook (catches all entry types). Routes to appropriate handler based on entry type field. MetricEntry is excluded — handled by handle_metric_entry_created instead. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "Entry") if not full_data.get("data", {}).get("entry"): raw_record.processing_status = "failed" raw_record.save() return "no_data" entry = full_data["data"]["entry"] entry_type = entry.get("type", "") if entry_type == "MetricEntry": # Already handled by handle_metric_entry_created — skip to avoid duplicate processing logger.debug( "MetricEntry via entry.created: skipping (handled by metric_entry.created)" ) elif entry_type == "SymptomEntry": _store_symptom_entry(entry, raw_record) else: # Generic entry — store as clinical observation logger.info(f"Storing generic entry type: {entry_type}") raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("care_plan.created") @register_handler("care_plan.updated") @register_handler("care_plan.activated") @register_handler("care_plan.deactivated") @register_handler("care_plan.deleted") def handle_care_plan_event(payload: WebhookPayload) -> str: """Handle care plan lifecycle events. Fetches care plan data via GraphQL and creates MonitoringPlan records when beneficiary and care_period can be resolved. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "CarePlan") if not full_data.get("data", {}).get("carePlan"): raw_record.processing_status = "failed" raw_record.save() return "no_data" cp = full_data["data"]["carePlan"] cp_id = cp.get("id", "") cp_name = cp.get("name", "") cp_active = cp.get("is_active", False) cp_template = cp.get("is_template", False) cp_patient = cp.get("patient_id", "") recommendations = cp.get("recommendations", []) cp_description = cp.get("description", "") logger.info( f"Care plan processed: id={cp_id} name={cp_name} active={cp_active} " f"template={cp_template} patient={cp_patient} " f"event={payload.event_type} recommendations={len(recommendations)}" ) # Store care plan reference in raw_payload raw_record.raw_payload = { **raw_record.raw_payload, "care_plan_reference": { "id": cp_id, "name": cp_name, "is_active": cp_active, "is_template": cp_template, "patient_id": cp_patient, "recommendations": recommendations, "event_type": payload.event_type, }, } # Attempt to create MonitoringPlan if beneficiary + care_period resolved if not cp_template: _create_monitoring_plan_from_care_plan( cp_id, cp_name, cp_description, cp_active, cp_patient, recommendations, payload.event_type, raw_record, ) raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("patient.merged") def handle_patient_merged(payload: WebhookPayload) -> str: """Handle patient record merge event. Attempts to find intake records for both source and destination users via external_system_identifiers lookup, then records the merge event. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") source_id = payload.raw_body.get("source_user_id", "") dest_id = payload.raw_body.get("destination_user_id", "") logger.info(f"Patient merged: source={source_id} destination={dest_id}") # Attempt to find intake records by looking up external IDs try: from access_core.identity.models import ExternalSystemIdentifier source_beneficiary = None dest_beneficiary = None try: ext_source = ExternalSystemIdentifier.objects.get( system_name="healthie", external_id=source_id ) source_beneficiary = ext_source.beneficiary except ExternalSystemIdentifier.DoesNotExist: pass try: ext_dest = ExternalSystemIdentifier.objects.get( system_name="healthie", external_id=dest_id ) dest_beneficiary = ext_dest.beneficiary except ExternalSystemIdentifier.DoesNotExist: pass if source_beneficiary or dest_beneficiary: logger.info( f"Merge linked: source_beneficiary={source_beneficiary and source_beneficiary.id} " f"dest_beneficiary={dest_beneficiary and dest_beneficiary.id}" ) except Exception as e: # noqa: BLE001 - merge lookup failure is non-fatal logger.error(f"Error during patient merge lookup: {e}") raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("metric_entry.updated") def handle_metric_entry_updated(payload: WebhookPayload) -> str: """Handle metric entry update webhook.""" raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "Entry") if not full_data.get("data", {}).get("entry"): raw_record.processing_status = "failed" raw_record.save() return "no_data" entry = full_data["data"]["entry"] _store_metric_observation(entry, raw_record) raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("entry.updated") def handle_entry_updated(payload: WebhookPayload) -> str: """Handle generic journal entry update webhook. MetricEntry is excluded — handled by handle_metric_entry_updated instead. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "Entry") if not full_data.get("data", {}).get("entry"): raw_record.processing_status = "failed" raw_record.save() return "no_data" entry = full_data["data"]["entry"] entry_type = entry.get("type", "") if entry_type == "MetricEntry": logger.debug( "MetricEntry via entry.updated: skipping (handled by metric_entry.updated)" ) elif entry_type == "SymptomEntry": _store_symptom_entry(entry, raw_record) else: logger.info(f"Generic entry updated: type={entry_type}") raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("applied_tag.created") @register_handler("applied_tag.deleted") def handle_applied_tag_event(payload: WebhookPayload) -> str: """Handle patient tag applied/removed webhook.""" raw_record = _store_raw_ingestion(payload, processing_status="processing") tag_id = payload.raw_body.get("tag_id", "") user_id = payload.raw_body.get("user_id", "") logger.info(f"Tag event: {payload.event_type} tag={tag_id} user={user_id}") raw_record.processing_status = "normalized" raw_record.save() return "normalized" @register_handler("form_answer_group.signed") def handle_form_signed(payload: WebhookPayload) -> str: """Handle form signed by provider webhook. Fetches the full FormAnswerGroup via GraphQL and updates the submission status to reflect the provider signature. """ raw_record = _store_raw_ingestion(payload, processing_status="processing") full_data = _fetch_healthie_resource(payload.resource_id, "FormAnswerGroup") form_group = full_data.get("data", {}).get("formAnswerGroup") if form_group: logger.info( f"Form signed: id={form_group.get('id')} " f"user={form_group.get('user', {}).get('id', '')}" ) # If a matching intake record exists, track the signing event if raw_record.intake_record: from access_core.intake.models import FormSubmissionEvent FormSubmissionEvent.objects.create( intake_record=raw_record.intake_record, raw_ingestion=raw_record, submitted_at=form_group.get("created_at"), source_system="healthie", status="normalized", ) else: logger.info( f"Form signed: raw record stored (no GraphQL data for id={payload.resource_id})" ) raw_record.processing_status = "normalized" raw_record.save() return "normalized" # ------------------------------------------------------------------ # Normalization helpers # ------------------------------------------------------------------ def _normalize_intake_form( raw_record: RawIngestionRecord, form_group: dict, form_answers: list, patient_data: dict, ): """Normalize intake form answers into IntakeRecord + sub-models.""" from access_core.intake.models import ( IntakeAssessment, IntakeIdentifier, IntakeLifestyle, IntakeRecord, ) # Parse form answers into key-value map first (needed for DOB/sex fallback) answers = {} for fa in form_answers: module = fa.get("custom_module", {}) module_id = module.get("id", "") answers[module_id] = { "label": fa.get("label", ""), "answer": fa.get("answer") or fa.get("displayed_answer", ""), "mod_type": module.get("mod_type", ""), } # Extract DOB: try user profile first, then fall back to form answers dob = patient_data.get("dob") if not dob: dob_str = _get_answer(answers, "Date of Birth") if dob_str: from datetime import date as date_type try: dob = date_type.fromisoformat(dob_str) except (ValueError, TypeError): logger.warning(f"Could not parse DOB from form answer: {dob_str}") # Extract sex: try user profile first, then fall back to form answers profile_gender = patient_data.get("gender", "") sex = _map_gender(profile_gender) if sex == "unknown": gender_answer = ( _get_answer(answers, "Gender") or _get_answer(answers, "Sex") or "" ) sex = _map_gender(gender_answer) # Look up or create IntakeRecord by external_form_id external_form_id = str(form_group.get("id", "")) healthie_user_id = patient_data.get("id", "") intake, created = IntakeRecord.objects.get_or_create( external_form_id=external_form_id, defaults={ "first_name": patient_data.get("first_name", ""), "last_name": patient_data.get("last_name", ""), "dob": dob, "sex": sex, "submitted_at": form_group.get("created_at"), "status": "intake_in_progress", }, ) # Merge: if a prospective intake exists for this Healthie user (from patient.created), # transfer its identifiers and sub-models then delete the duplicate. if healthie_user_id and created: prospective_key = f"healthie-user-{healthie_user_id}" try: old_intake = IntakeRecord.objects.get(external_form_id=prospective_key) # Transfer identifiers to the new intake for ident in old_intake.identifiers.all(): ident.intake_record = intake ident.save(update_fields=["intake_record"]) # Transfer sub-models for rel_name in ( "assessments", "lifestyle", "female_health", "male_health", "msk", "safety_screening", "participation_readiness", ): try: qs = getattr(old_intake, rel_name).all() for obj in qs: obj.intake_record = intake obj.save(update_fields=["intake_record"]) except Exception as e: # noqa: BLE001 - merge must not fail on one relation logger.warning( f"Failed to reassign {rel_name} records during intake merge: {e}" ) # Transfer linked raw ingestion records RawIngestionRecord.objects.filter(intake_record=old_intake).update( intake_record=intake ) old_intake.delete() logger.info( f"Merged duplicate prospective intake {old_intake.id} " f"into form-based intake {intake.id}" ) except IntakeRecord.DoesNotExist: pass # Update DOB if it was missing on first creation (race: patient.created may have created it earlier) if not intake.dob and dob: intake.dob = dob intake.save(update_fields=["dob"]) # Link raw ingestion raw_record.intake_record = intake raw_record.save() # Track the form submission event from access_core.intake.models import FormSubmissionEvent FormSubmissionEvent.objects.create( intake_record=intake, raw_ingestion=raw_record, submitted_at=form_group.get("created_at"), source_system="healthie", status="processing", ) # Store identifiers if patient_data.get("email"): IntakeIdentifier.objects.update_or_create( intake_record=intake, identifier_type="email", defaults={"identifier_value": patient_data["email"], "is_primary": True}, ) if patient_data.get("phone_number"): IntakeIdentifier.objects.update_or_create( intake_record=intake, identifier_type="phone", defaults={"identifier_value": patient_data["phone_number"]}, ) # Get or create IntakeAssessment (idempotent on retry) assessment, _ = IntakeAssessment.objects.get_or_create( intake_record=intake, defaults={"raw_ingestion": raw_record}, ) if not assessment.raw_ingestion: assessment.raw_ingestion = raw_record assessment.save() # Get or create IntakeLifestyle (idempotent on retry) lifestyle, _ = IntakeLifestyle.objects.get_or_create( intake_record=intake, defaults={"raw_ingestion": raw_record}, ) if not lifestyle.raw_ingestion: lifestyle.raw_ingestion = raw_record lifestyle.save() # Map answers to model fields based on module label matching _map_answers_to_assessment(assessment, answers) _map_answers_to_lifestyle(lifestyle, answers) _map_answers_to_female_health(intake, raw_record, answers, profile_gender or sex) _map_answers_to_male_health(intake, raw_record, answers, profile_gender or sex) _map_answers_to_msk(intake, raw_record, answers) _map_answers_to_safety_screening(intake, raw_record, answers) _map_answers_to_participation_readiness(intake, answers) logger.info( f"Intake normalized: {intake.id} form={external_form_id} answers={len(answers)}" ) def _normalize_other_form( raw_record: RawIngestionRecord, form_group: dict, form_answers: list ): """Normalize non-intake form submissions (daily check-ins, etc.). Process: 1. Match Healthie user to a Beneficiary via ExternalSystemIdentifier 2. Create DailyCheckinSubmission + DailyCheckinResponses 3. Detect safety signals from answers 4. Create SafetySignalEvent if red/yellow flags detected """ patient_data = form_group.get("user", {}) healthie_user_id = patient_data.get("id", "") email = patient_data.get("email", "") from access_core.intake.models import FormSubmissionEvent # Try to match Healthie user to a Beneficiary beneficiary = None care_period = None if healthie_user_id: try: from access_core.identity.models import ( ExternalSystemIdentifier, ) ext_id = ExternalSystemIdentifier.objects.get( system_name="healthie", external_id=healthie_user_id, ) beneficiary = ext_id.beneficiary cp_qs = beneficiary.care_periods.filter(operational_status="active") if cp_qs.exists(): care_period = cp_qs.latest("start_date") except ExternalSystemIdentifier.DoesNotExist: pass except Exception as e: # noqa: BLE001 - continue with unresolved beneficiary logger.error( f"Error matching beneficiary for healthie_user={healthie_user_id}: {e}" ) # If no beneficiary found (patient not yet promoted), store raw for later if not beneficiary: logger.info( f"Daily check-in for non-beneficiary user={healthie_user_id} email={email}. " f"Storing raw submission for later processing." ) from access_core.monitoring.models import RawDailyCheckinSubmission RawDailyCheckinSubmission.objects.create( beneficiary=beneficiary, raw_payload={ "form_group_id": form_group.get("id"), "template_id": form_group.get("custom_module_form", {}).get("id"), "user_id": healthie_user_id, "email": email, "form_answers": form_answers, "created_at": form_group.get("created_at"), }, source_system="healthie", source_record_id=form_group.get("id", ""), form_version=form_group.get("custom_module_form", {}).get("id"), ) return # Create DailyCheckinSubmission from access_core.monitoring.models import ( DailyCheckinItem, DailyCheckinResponse, DailyCheckinSubmission, DailyCheckinTemplate, ) template, _ = DailyCheckinTemplate.objects.get_or_create( template_name=form_group.get("custom_module_form", {}).get( "name", "Daily Check-in" ), clinical_track="all", defaults={"version": form_group.get("custom_module_form", {}).get("id", "1")}, ) # Parse submission date from created_at created_at = form_group.get("created_at", "") submission_date = None if created_at: try: from datetime import datetime dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) submission_date = dt.date() except (ValueError, AttributeError): pass if submission_date is None: from django.utils import timezone submission_date = timezone.now().date() submission = DailyCheckinSubmission.objects.create( beneficiary=beneficiary, care_period=care_period, template=template, submission_date=submission_date, source_system="healthie", status="submitted", ) # Parse answers into key-value map answers = {} for fa in form_answers: module = fa.get("custom_module", {}) module_id = module.get("id", "") answers[module_id] = { "label": fa.get("label", ""), "answer": fa.get("answer") or fa.get("displayed_answer", ""), "mod_type": module.get("mod_type", ""), } # Create responses for each answer for module_id, ans_data in answers.items(): item, _ = DailyCheckinItem.objects.get_or_create( template=template, item_key=module_id, defaults={ "item_label": ans_data["label"], "item_type": ans_data["mod_type"], "is_required": False, "display_order": 0, }, ) DailyCheckinResponse.objects.create( submission=submission, item=item, value=ans_data["answer"], ) # Check for safety signals in answers _check_safety_signals(submission, answers, care_period) # Track as form submission event on intake record if one exists if raw_record.intake_record: FormSubmissionEvent.objects.create( intake_record=raw_record.intake_record, raw_ingestion=raw_record, submitted_at=created_at, source_system="healthie", status="normalized", ) logger.info( f"Daily check-in normalized: submission={submission.id} " f"beneficiary={beneficiary.id} answers={len(answers)}" ) def _check_safety_signals(submission, answers: dict, care_period): """Check daily check-in answers for safety signals. Scans for answers containing keywords that indicate safety concerns. Creates SafetySignalEvent and CareTeamReviewTask if flagged. """ safety_keywords = { "red": [ "chest pain", "shortness of breath", "bleeding", "stroke", "heart attack", "suicidal", "seizure", "unconscious", "hospital", ], "yellow": [ "dizziness", "fever", "swelling", "rash", "nausea", "vomiting", "confusion", "fall", "worsening", "new pain", ], } for ans_data in answers.values(): answer_text = str(ans_data.get("answer", "")).lower() for severity, keywords in safety_keywords.items(): for keyword in keywords: if keyword in answer_text: _create_safety_signal( submission.beneficiary, care_period, severity, keyword, ans_data["label"], answer_text, ) break else: continue break def _create_safety_signal( beneficiary, care_period, severity: str, trigger_keyword: str, field_label: str, answer_text: str, ): """Create a SafetySignalEvent for a detected safety concern.""" from access_core.governance.models import CareTeamReviewTask, SafetySignalEvent signal, created = SafetySignalEvent.objects.get_or_create( beneficiary=beneficiary, care_period=care_period, signal_severity=severity, signal_type="symptom_escalation" if severity == "yellow" else "adverse_event", defaults={ "signal_details": { "trigger_keyword": trigger_keyword, "field_label": field_label, "answer_text": answer_text[:500], "source": "daily_checkin", }, "auto_generated": True, "review_status": "pending", }, ) if created: priority_map = {"red": "urgent", "yellow": "high"} CareTeamReviewTask.objects.create( beneficiary=beneficiary, care_period=care_period, trigger_type="safety_signal", trigger_record_id=signal.id, priority=priority_map.get(severity, "medium"), status="pending", ) logger.warning( f"Safety signal created: severity={severity} keyword={trigger_keyword} " f"beneficiary={beneficiary.id}" ) def _link_patient_to_intake(user: dict, event_type: str, changed_fields: list): """Link a Healthie patient to an IntakeRecord via external identifier. On patient.created: Creates an IntakeRecord if none exists. On patient.updated: Links to existing IntakeRecord by email/phone. """ email = user.get("email", "") phone = user.get("phone_number", "") healthie_id = user.get("id", "") first_name = user.get("first_name", "") last_name = user.get("last_name", "") dob = user.get("dob") gender = user.get("gender", "") from access_core.intake.models import IntakeIdentifier, IntakeRecord # Attempt to find matching IntakeRecord by email or phone lookup_kwargs = {} if email: lookup_kwargs = {"identifiers__identifier_value__iexact": email} elif phone: lookup_kwargs = {"identifiers__identifier_value": phone} if lookup_kwargs: try: intakes = list( IntakeRecord.objects.prefetch_related("identifiers").filter( **lookup_kwargs ) ) if not intakes: raise IntakeRecord.DoesNotExist # Update the most recently modified intake (most likely the active one) intake = max(intakes, key=lambda r: r.updated_at or r.created_at) logger.info( f"Patient linked: healthie_id={healthie_id} " f"-> intake_record={intake.id} via {email or phone}" ) # Update patient info on update events — apply to all matching intakes if event_type == "patient.updated": updates = {} if first_name: updates["first_name"] = first_name if last_name: updates["last_name"] = last_name if dob: updates["dob"] = dob if updates: IntakeRecord.objects.filter(pk__in=[r.pk for r in intakes]).update( **updates ) return except IntakeRecord.DoesNotExist: pass # No matching intake record found — create one for patient.created events if event_type == "patient.created" and healthie_id: external_form_id = f"healthie-user-{healthie_id}" intake, created = IntakeRecord.objects.get_or_create( external_form_id=external_form_id, defaults={ "first_name": first_name, "last_name": last_name, "dob": dob, "sex": _map_gender(gender), "status": "prospective", }, ) if created: if email: IntakeIdentifier.objects.create( intake_record=intake, identifier_type="email", identifier_value=email, is_primary=True, ) if phone: IntakeIdentifier.objects.create( intake_record=intake, identifier_type="phone", identifier_value=phone, ) logger.info( f"Patient created intake: healthie_id={healthie_id} " f"-> intake_record={intake.id} (form_id={external_form_id})" ) else: logger.info(f"Patient intake already exists: {intake.id}") else: logger.info( f"Patient event: {event_type} id={healthie_id} " f"name={first_name} {last_name} changed={changed_fields}" ) def _store_document_reference(document: dict): """Store a document reference in raw_ingestion for later normalization.""" doc_id = document.get("id") display_name = document.get("display_name") content_type = document.get("file_content_type") owner = document.get("owner", {}) logger.info( f"Document stored: id={doc_id} name={display_name} " f"type={content_type} owner={owner.get('id', '')}" ) def _resolve_beneficiary_and_care_period(healthie_client_id: str): """Resolve Healthie client_id to Beneficiary + active CarePeriod. Returns (beneficiary, care_period) or (None, None). """ if not healthie_client_id: return None, None try: from access_core.identity.models import ExternalSystemIdentifier ext_id = ExternalSystemIdentifier.objects.get( system_name="healthie", external_id=str(healthie_client_id), ) beneficiary = ext_id.beneficiary cp_qs = beneficiary.care_periods.filter(operational_status="active") care_period = cp_qs.latest("start_date") if cp_qs.exists() else None return beneficiary, care_period except ExternalSystemIdentifier.DoesNotExist: logger.info( f"No ExternalSystemIdentifier for healthie_client={healthie_client_id}" ) return None, None except Exception as e: # noqa: BLE001 - continue with unresolved beneficiary logger.error( f"Error resolving beneficiary for healthie_client={healthie_client_id}: {e}" ) return None, None def _get_system_user(): """Get or create a system user for automated events.""" from access_core.identity.models import User system_user, _ = User.objects.get_or_create( email="system@metaflow.local", defaults={ "full_name": "System Automation", "is_active": True, }, ) return system_user def _parse_healthie_date(date_str: str): """Parse ISO date string from Healthie. Returns date or None.""" if not date_str: return None try: from datetime import datetime dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")) return dt.date() except (ValueError, AttributeError): return None def _store_metric_observation(entry: dict, raw_record: RawIngestionRecord): """Store a metric entry as a ClinicalObservation + OAPMeasureObservation. Metric stat from Healthie can be: - Simple: {"value": 120, "unit": "mmHg"} - Compound: {"systolic": 120, "diastolic": 80} """ entry_id = entry.get("id") category = entry.get("category", "Unknown") metric_stat = entry.get("metric_stat", {}) created_at_str = entry.get("created_at") client_id = entry.get("client_id") added_by = entry.get("added_by_user", {}) beneficiary, care_period = _resolve_beneficiary_and_care_period(client_id) if not beneficiary: logger.warning( f"MetricEntry skipped: no beneficiary for healthie_client={client_id}. " f"Raw data stored in ingestion record {raw_record.id}." ) raw_record.raw_payload = { **raw_record.raw_payload, "metric_pending_normalization": { "entry_id": entry_id, "category": category, "metric_stat": metric_stat, "client_id": client_id, "created_at": created_at_str, }, } raw_record.save() return collection_date = _parse_healthie_date(created_at_str) if collection_date is None: from django.utils import timezone collection_date = timezone.now().date() metric_value, value_text, unit = _extract_metric_values(metric_stat, category) observation_type = "vital_sign" if category and any( k in category.lower() for k in ("a1c", "glucose", "lipid", "cholesterol", "albumin", "creatinine") ): observation_type = "lab" observation = _create_clinical_observation( beneficiary=beneficiary, care_period=care_period, observation_type=observation_type, measure_name=category, value_text=value_text, value_numeric=metric_value, unit=unit, collection_date=collection_date, raw_record=raw_record, is_patient_entry=added_by.get("is_patient", False), ) _create_oap_measure_observation( observation=observation, beneficiary=beneficiary, care_period=care_period, measure_name=category, collection_date=collection_date, raw_record=raw_record, ) logger.info( f"Metric normalized: entry={entry_id} category={category} " f"beneficiary={beneficiary.id} observation={observation.id}" ) def _extract_metric_values(metric_stat: dict, category: str): """Extract numeric value, text value, and unit from Healthie metric_stat. Returns (value_numeric, value_text, unit). """ if not isinstance(metric_stat, dict): try: return float(metric_stat), str(metric_stat), None except (ValueError, TypeError): return None, str(metric_stat), None if "systolic" in metric_stat and "diastolic" in metric_stat: sys_val = metric_stat["systolic"] dia_val = metric_stat["diastolic"] unit = metric_stat.get("unit", "mmHg") combined = f"{sys_val}/{dia_val}" return None, f"{combined} {unit}", unit value = metric_stat.get("value") unit = metric_stat.get("unit") try: numeric = float(value) return numeric, f"{value} {unit}" if unit else str(value), unit except (ValueError, TypeError): return None, str(value) if value is not None else "", unit def _create_clinical_observation( beneficiary, care_period, observation_type: str, measure_name: str, value_text: str | None, value_numeric: float | None, unit: str | None, collection_date, raw_record: RawIngestionRecord, is_patient_entry: bool, ): """Create a ClinicalObservation record.""" from access_core.clinical.models import ClinicalObservation return ClinicalObservation.objects.create( beneficiary=beneficiary, care_period=care_period, observation_type=observation_type, measure_name=measure_name, value_text=value_text, value_numeric=value_numeric, unit=unit, collection_date=collection_date, data_source="patient_upload" if is_patient_entry else "device", source_record=raw_record, is_synthetic=True, status="imported", ) def _create_oap_measure_observation( observation, beneficiary, care_period, measure_name: str, collection_date, raw_record: RawIngestionRecord, ): """Create OAPMeasureObservation if care_period exists and measure definition can be resolved.""" if not care_period: logger.info(f"OAP observation skipped: no care_period for {beneficiary.id}") return from access_core.clinical.models import OAPMeasureDefinition, OAPMeasureObservation try: measure_def, _ = OAPMeasureDefinition.objects.get_or_create( measure_name=measure_name, defaults={ "clinical_track": "all", "is_active": True, }, ) except Exception as e: # noqa: BLE001 - fall back to default measure handling logger.warning( f"Could not resolve OAP measure definition for '{measure_name}': {e}" ) return OAPMeasureObservation.objects.create( measure_definition=measure_def, beneficiary=beneficiary, care_period=care_period, canonical_observation=observation, reporting_type="baseline", collection_date=collection_date, validity_window_status="within_window", data_source=observation.data_source, collection_method="patient_reported" if observation.data_source == "patient_upload" else "device", physician_review_status="not_required", status="draft", ) def _store_symptom_entry(entry: dict, raw_record: RawIngestionRecord): """Store a symptom entry as a SafetySignalEvent. Healthie symptom entries contain a free-text description. We scan for severity keywords and create an appropriate signal. """ entry_id = entry.get("id") description = entry.get("description", "") client_id = entry.get("client_id") created_at_str = entry.get("created_at") beneficiary, care_period = _resolve_beneficiary_and_care_period(client_id) if not beneficiary: logger.warning( f"SymptomEntry skipped: no beneficiary for healthie_client={client_id}. " f"Raw data stored in ingestion record {raw_record.id}." ) raw_record.raw_payload = { **raw_record.raw_payload, "symptom_pending_normalization": { "entry_id": entry_id, "description": description, "client_id": client_id, "created_at": created_at_str, }, } raw_record.save() return if not care_period: logger.warning( f"SymptomEntry skipped: no active care_period for beneficiary={beneficiary.id}." ) return severity = _assess_symptom_severity(description) signal_type = "adverse_event" if severity == "red" else "symptom_escalation" from access_core.governance.models import SafetySignalEvent SafetySignalEvent.objects.create( beneficiary=beneficiary, care_period=care_period, signal_severity=severity, signal_type=signal_type, signal_details={ "entry_id": entry_id, "description": description[:1000], "source": "healthie_symptom_entry", "created_at": created_at_str, }, source_record_id=raw_record.id, auto_generated=True, review_status="pending", ) logger.info( f"Symptom normalized: entry={entry_id} severity={severity} " f"beneficiary={beneficiary.id}" ) def _assess_symptom_severity(description: str) -> str: """Assess severity from symptom description text. Returns 'red', 'yellow', or 'green'. """ desc_lower = description.lower() red_keywords = [ "chest pain", "shortness of breath", "bleeding", "stroke", "heart attack", "suicidal", "seizure", "unconscious", "hospital", "emergency", "can't breathe", ] yellow_keywords = [ "dizziness", "fever", "swelling", "rash", "nausea", "vomiting", "confusion", "fall", "worsening", "new pain", "back pain", "numbness", "tingling", "headache", ] for kw in red_keywords: if kw in desc_lower: return "red" for kw in yellow_keywords: if kw in desc_lower: return "yellow" return "green" def _get_answer(answers: dict, label: str) -> str: """Find an answer value by matching its module label (case-insensitive partial match).""" for ans_data in answers.values(): if label.lower() in ans_data.get("label", "").lower(): val = ans_data.get("answer", "") return str(val) if val is not None else "" return "" def _safe_float(val: str): """Parse string to float, stripping common units. Returns None on failure.""" if not val: return None cleaned = ( val.replace(",", "") .replace("lbs", "") .replace('"', "") .replace("'", "") .replace("in", "") .strip() ) try: return float(cleaned) except (ValueError, TypeError): return None def _map_answers_to_assessment(assessment, answers: dict): """Map form answers to IntakeAssessment fields.""" assessment.primary_concerns = _get_answer(answers, "What health concerns") assessment.other_concerns = _get_answer(answers, "Other Concerns") assessment.symptom_duration = _get_answer(answers, "How long have these concerns") assessment.primary_symptoms = _get_answer( answers, "What physical symptoms or sensations" ) assessment.medical_conditions = _get_answer( answers, "Have you ever been diagnosed by a physician with any of the following conditions", ) assessment.infection_immune_conditions = _get_answer( answers, "Infection, Liver, Immune, and Other Conditions" ) assessment.autoimmune_conditions = _get_answer( answers, "Autoimmune, Immune-Inflammatory" ) active_raw = _get_answer(answers, "Are any of these infection") assessment.conditions_active = active_raw.strip().lower().startswith("yes") assessment.medications = _get_answer( answers, "Are you currently using any of the following medications" ) assessment.medication_details = _get_answer( answers, "please list the name, dose, frequency" ) assessment.allergies = _get_answer(answers, "Allergies") assessment.save() def _map_answers_to_lifestyle(lifestyle, answers: dict): """Map form answers to IntakeLifestyle fields.""" lifestyle.weight_lbs = _safe_float(_get_answer(answers, "Weight in pounds")) lifestyle.height_inches = _safe_float(_get_answer(answers, "Height in inches")) lifestyle.waist_circumference_inches = _safe_float( _get_answer(answers, "Waist circumference") ) if lifestyle.weight_lbs and lifestyle.height_inches and lifestyle.height_inches > 0: lifestyle.bmi = round( 703 * lifestyle.weight_lbs / (lifestyle.height_inches**2), 1 ) lifestyle.weight_change_6mo = _get_answer( answers, "Weight change over the past 6 months" ) lifestyle.weight_control_patterns = _get_answer( answers, "weight-control or eating-control" ) lifestyle.activity_level = _get_answer(answers, "usual physical activity level") lifestyle.activity_types = _get_answer(answers, "What types of physical activity") lifestyle.post_activity_response = _get_answer( answers, "respond after physical activity" ) lifestyle.digestive_patterns = _get_answer( answers, "digestive patterns do you commonly experience" ) lifestyle.diet_patterns = _get_answer(answers, "usual diet and lifestyle patterns") lifestyle.digestive_conditions = _get_answer( answers, "diagnosed by a physician with any of the following digestive" ) lifestyle.bowel_patterns = _get_answer(answers, "typical bowel movements") lifestyle.urination_patterns = _get_answer(answers, "typical urination pattern") lifestyle.hemorrhoids = _get_answer(answers, "hemorrhoids") lifestyle.sleep_quality = _get_answer(answers, "sleep quality") lifestyle.stress_emotional_state = _get_answer(answers, "stress or emotional state") lifestyle.sweating_patterns = _get_answer(answers, "sweating pattern") lifestyle.skin_patterns = _get_answer(answers, "skin patterns apply") lifestyle.skin_longterm_meds = _get_answer(answers, "long-term for skin") lifestyle.save() def _map_answers_to_female_health(intake, raw_record, answers: dict, gender: str): """Map form answers to IntakeFemaleHealth if applicable.""" if gender != "Female": return from access_core.intake.models import IntakeFemaleHealth fh, _ = IntakeFemaleHealth.objects.get_or_create( intake_record=intake, defaults={"raw_ingestion": raw_record}, ) if not fh.raw_ingestion: fh.raw_ingestion = raw_record fh.menstrual_status = _get_answer(answers, "menstrual status") fh.menstrual_flow = _get_answer(answers, "menstrual flow") fh.menstrual_blood_color = _get_answer(answers, "menstrual blood usually closest") fh.menstrual_clots_cramps = _get_answer(answers, "menstrual clots") fh.discharge_discomfort = _get_answer(answers, "vaginal discharge") fh.diagnosed_conditions = _get_answer(answers, "diagnosed with or had a history of") fh.free_text = _get_answer(answers, "gynecologic health") fh.save() def _map_answers_to_male_health(intake, raw_record, answers: dict, gender: str): """Map form answers to IntakeMaleHealth if applicable.""" if gender != "Male": return from access_core.intake.models import IntakeMaleHealth mh, _ = IntakeMaleHealth.objects.get_or_create( intake_record=intake, defaults={"raw_ingestion": raw_record}, ) if not mh.raw_ingestion: mh.raw_ingestion = raw_record mh.prostate_conditions = _get_answer(answers, "prostate-related condition") mh.sexual_function = _get_answer(answers, "sexual function") mh.ejaculation_concerns = _get_answer(answers, "ejaculation or semen") mh.perineal_testicular_symptoms = _get_answer(answers, "perineal, testicular") mh.pelvic_surgery = _get_answer(answers, "pelvic surgery or procedure") mh.fertility_status = _get_answer(answers, "trying to conceive") mh.free_text = _get_answer(answers, "urinary, prostate, sexual") mh.save() def _map_answers_to_msk(intake, raw_record, answers: dict): """Map form answers to IntakeMSK.""" from access_core.intake.models import IntakeMSK msk, _ = IntakeMSK.objects.get_or_create( intake_record=intake, defaults={"raw_ingestion": raw_record}, ) if not msk.raw_ingestion: msk.raw_ingestion = raw_record msk.pain_locations = _get_answer(answers, "experience soreness, tightness") msk.msk_diagnoses = _get_answer( answers, "diagnosed by a physician with any of the following" ) msk.injury_history = _get_answer(answers, "sports injury, trauma") msk.functional_impact = _get_answer( answers, "Does the pain affect your daily function" ) msk.pain_quality = _get_answer(answers, "best match the quality of your pain") msk.pain_duration = _get_answer(answers, "long has this pain been present") msk.pain_severity = _safe_float(_get_answer(answers, "severity of your main pain")) msk.save() def _map_answers_to_safety_screening(intake, raw_record, answers: dict): """Map form answers to SafetyScreening.""" from access_core.intake.models import SafetyScreening ss, _ = SafetyScreening.objects.get_or_create( intake_record=intake, defaults={"raw_ingestion": raw_record}, ) if not ss.raw_ingestion: ss.raw_ingestion = raw_record safety_stop = _get_answer(answers, "Do any of the above current conditions apply") ss.safety_stop_applies = safety_stop.strip().lower().startswith("yes") ss.free_text = _get_answer( answers, "Is there anything else you would like the MetaFlow review team to know", ) if ss.safety_stop_applies: ss.screening_status = "flagged" else: ss.screening_status = "needs_review" ss.save() def _map_answers_to_participation_readiness(intake, answers: dict): """Map form answers to ParticipationReadiness.""" from access_core.intake.models import ParticipationReadiness acknowledged = _get_answer(answers, "I have read and understand") is_acknowledged = acknowledged.lower() in ("yes", "true", "1", "i agree") ParticipationReadiness.objects.update_or_create( intake_record=intake, defaults={ "acknowledged": is_acknowledged, "status": "acknowledged" if is_acknowledged else "pending", }, ) def _map_gender(healthie_gender: str) -> str: """Map Healthie gender field to our sex enum.""" gender_map = { "male": "male", "female": "female", "intersex": "other", } cleaned = (healthie_gender or "").strip().lower().replace(" ", "") if cleaned in ("nonbinary", "non-binary"): return "other" if cleaned in gender_map: return gender_map[cleaned] return "unknown" def _create_monitoring_plan_from_care_plan( cp_id: str, cp_name: str, cp_description: str, cp_active: bool, cp_patient: str, recommendations: list, event_type: str, raw_record: RawIngestionRecord, ): """Create/update MonitoringPlan from a Healthie care plan. If beneficiary + care_period can be resolved, creates MonitoringPlan + items. Otherwise, stores pending data in raw_payload for later normalization. """ from access_core.monitoring.models import ( MonitoringPlan, MonitoringPlanItem, MonitoringPlanTemplate, ) beneficiary, care_period = _resolve_beneficiary_and_care_period(cp_patient) if not beneficiary or not care_period: logger.info( f"MonitoringPlan skipped: beneficiary={beneficiary is not None}, " f"care_period={care_period is not None} for healthie_patient={cp_patient}." ) return system_user = _get_system_user() plan_status = "active" if cp_active else "completed" if event_type == "care_plan.deleted": plan_status = "archived" template, _ = MonitoringPlanTemplate.objects.get_or_create( template_name=cp_name, clinical_track="all", defaults={ "trigger_description": cp_description[:500] if cp_description else None, "is_active": cp_active, }, ) plan, created = MonitoringPlan.objects.get_or_create( beneficiary=beneficiary, care_period=care_period, clinical_trigger=cp_name, defaults={ "template": template, "created_by": system_user, "baseline_abnormality": cp_description[:1000] if cp_description else None, "status": plan_status, }, ) if not created: plan.status = plan_status plan.save() for rec in recommendations: rec_id = rec.get("id", "") rec_body = rec.get("recommendation_body", "") rec_type = rec.get("recommendation_type", "") MonitoringPlanItem.objects.update_or_create( monitoring_plan=plan, item_description=rec_body[:500] if rec_body else f"Recommendation {rec_id}", defaults={ "monitoring_category": rec_type or "general", "measure_name": rec_type, "evidence_layer": "care_plan_recommendation", "status": "scheduled" if cp_active else "completed", }, ) logger.info( f"MonitoringPlan {'created' if created else 'updated'}: plan={plan.id} " f"name={cp_name} beneficiary={beneficiary.id} " f"items={plan.items.count()}" )