From 30b9c0ceefb3a329526b30f0b44e23783e31356f Mon Sep 17 00:00:00 2001 From: Rockachopa Date: Thu, 30 Apr 2026 10:47:42 -0400 Subject: [PATCH] feat(crisis): add image-capable intake handler for test users (closes #201) Extend crisis.intake.handle_intake_submission to accept optional image-based signals (OCR text, object labels, operator notes) and integrate with image_screening.screen_image_signals(). Changes: - crisis/intake.py: Image support added to intake handler * Optional parameters: image_ocr, image_labels, image_notes * Calls screen_image_signals when image data provided * Extends IntakeResult with image_signals and image_distress_score * Backward-compatible: existing text-only calls unaffected * Enhanced interactive CLI with [img:...] inline syntax - Updated _log_test_intake to record image signals in test metrics This is the smallest concrete fix for #201: image-capable intake submission. The text-only handler already existed via PR #202. Tests: all 8 intake tests pass (existing behavior preserved). Smoke: manual call verifies image signals detected. --- crisis/intake.py | 134 ++++++++++++++++++++++++++++++++++++----------- 1 file changed, 103 insertions(+), 31 deletions(-) diff --git a/crisis/intake.py b/crisis/intake.py index 3d649e1..02b8130 100644 --- a/crisis/intake.py +++ b/crisis/intake.py @@ -1,22 +1,32 @@ + """ Intake submission handler for the-door. Provides a lightweight function for receiving and processing test intake submissions (QA/test user messages) with separate metrics tracking. +Supports optional image-based screening via image_screening module. Usage: from crisis.intake import handle_intake_submission result = handle_intake_submission("This is a test message", test=True) + result = handle_intake_submission( + "Check this image for crisis signals", + test=True, + image_ocr="I can't go on", + image_labels=["blade", "blood"], + image_notes="cutting marks visible" + ) """ -from dataclasses import dataclass -from typing import Optional +from dataclasses import dataclass, field +from typing import List, Optional from pathlib import Path from .gateway import check_crisis from .metrics import CrisisMetrics, SessionMetrics from .session_tracker import CrisisSessionTracker +from image_screening import screen_image_signals, ImageScreeningResult @dataclass @@ -29,6 +39,8 @@ class IntakeResult: provide_988: bool is_test: bool session_id: Optional[str] = None + image_signals: List[str] = field(default_factory=list) + image_distress_score: float = 0.0 # Separate metrics namespace for test intake tracking @@ -39,6 +51,9 @@ def handle_intake_submission( message: str, test: bool = True, session_id: Optional[str] = None, + image_ocr: str = "", + image_labels: Optional[List[str]] = None, + image_notes: str = "", ) -> IntakeResult: """ Handle an intake submission from a user (including test users). @@ -47,16 +62,36 @@ def handle_intake_submission( message: The user's message text. test: If True, marks this as a test submission (does not pollute live metrics). session_id: Optional session identifier for correlation. + image_ocr: OCR text extracted from an attached image (if any). + image_labels: Object/context labels from image analysis (if any). + image_notes: Manual operator notes about the image scene (if any). Returns: IntakeResult with crisis assessment and response data. """ - # Run crisis detection + # Run crisis detection on text crisis_result = check_crisis(message) + # Optionally process image signals + image_signals: List[str] = [] + image_distress_score: float = 0.0 + if image_ocr or image_labels or image_notes: + img_result: ImageScreeningResult = screen_image_signals( + ocr_text=image_ocr, + labels=image_labels or [], + manual_notes=image_notes, + ) + image_signals = img_result.signals_detected or [] + image_distress_score = float(img_result.distress_score or 0.0) + # For test submissions, optionally log to separate test metrics if test: - _log_test_intake(crisis_result, message) + _log_test_intake( + crisis_result, + message, + img_score=image_distress_score, + img_signals=image_signals, + ) return IntakeResult( crisis_level=crisis_result["level"], @@ -66,31 +101,27 @@ def handle_intake_submission( provide_988=crisis_result["ui"]["provide_988"], is_test=test, session_id=session_id, + image_signals=image_signals, + image_distress_score=image_distress_score, ) -def _log_test_intake(detection, message: str) -> None: - """ - Log test intake submissions to a separate metrics file. - - These do not mix with production crisis statistics. - - Args: - detection: Either a CrisisDetectionResult dataclass or dict from check_crisis. - message: The original user message. - """ +def _log_test_intake( + detection, + message: str, + img_score: float = 0.0, + img_signals: Optional[List[str]] = None, +) -> None: + """Log test intake submissions to a separate metrics file (test-only).""" import time import json - # Accept both dataclass and dict - if hasattr(detection, '__dataclass_fields__'): - level = detection.level - score = detection.score - indicators = detection.indicators - else: - level = detection.get("level") - score = detection.get("score") - indicators = detection.get("indicators", []) + if img_signals is None: + img_signals = [] + + level = detection.get("level") if isinstance(detection, dict) else detection.level + score = detection.get("score") if isinstance(detection, dict) else detection.score + indicators = detection.get("indicators", []) if isinstance(detection, dict) else detection.indicators _INTAKE_METRICS_DIR.mkdir(parents=True, exist_ok=True) today = time.strftime("%Y-%m-%d") @@ -102,6 +133,10 @@ def _log_test_intake(detection, message: str) -> None: "score": score, "message_preview": message[:80], "indicators": indicators, + "image": { + "distress_score": img_score, + "signals": img_signals, + }, } with open(filepath, "a") as f: @@ -110,21 +145,58 @@ def _log_test_intake(detection, message: str) -> None: # ── Quick test interface ──────────────────────────────────────────── -def _interactive(): +def _interactive() -> None: """Run an interactive test intake session.""" print("=== Intake Submission Test Interface ===") print("Enter test messages (Ctrl+C to exit).\n") + print("Optional: prepend 'img:' to include image data.") + print(" Example: I can't go on [img:ocr=blood on hands;labels=blade,wound;notes=visible cuts]") try: while True: - msg = input("test-user> ").strip() - if not msg: + try: + raw = input("test-user> ").strip() + except (EOFError, KeyboardInterrupt): + print("\nDone.") + break + if not raw: continue - result = handle_intake_submission(msg, test=True) - print(f"\n Level : {result.crisis_level}") - print(f" Score : {result.score}") - print(f" Timmy : {result.timmy_message}") - print(f" Overlay: {result.show_overlay}") + + # Parse optional [img:...] section + image_ocr = "" + image_labels: List[str] = [] + image_notes = "" + if "[img:" in raw: + try: + img_part = raw[raw.index("[img:") + 5 : raw.index("]")] + raw = raw.replace(f"[img:{img_part}]", "").strip() + for pair in img_part.split(";"): + if "=" in pair: + k, v = pair.split("=", 1) + k = k.strip().lower() + if k == "ocr": + image_ocr = v.strip() + elif k == "labels": + image_labels = [lbl.strip() for lbl in v.split(",") if lbl.strip()] + elif k == "notes": + image_notes = v.strip() + except Exception: + pass # malformed image section; ignore + + result = handle_intake_submission( + raw, + test=True, + image_ocr=image_ocr, + image_labels=image_labels, + image_notes=image_notes, + ) + print(f"\n Level : {result.crisis_level}") + print(f" Score : {result.score:.3f}") + print(f" Timmy : {result.timmy_message}") + print(f" Overlay : {result.show_overlay}") + if result.image_signals: + print(f" Img Sig : {', '.join(result.image_signals)}") + print(f" Img Distr: {result.image_distress_score:.3f}") print() except (EOFError, KeyboardInterrupt): print("\nDone.") -- 2.43.0