Update app.py

app.py

@@ -604,14 +604,14 @@ def analyze_single_message(text, thresholds):
     return abuse_score, threshold_labels, top_patterns, {"label": sentiment}, stage, darvo_score, tone_tag
 
 def analyze_composite(msg1, date1, msg2, date2, msg3, date3, *answers_and_none):
+    from collections import Counter
+
     none_selected_checked = answers_and_none[-1]
     responses_checked = any(answers_and_none[:-1])
     none_selected = not responses_checked and none_selected_checked
 
-    if none_selected:
-        escalation_score = None
-        risk_level = "unknown"
-    else:
+    escalation_score = None
+    if not none_selected:
         escalation_score = sum(w for (_, w), a in zip(ESCALATION_QUESTIONS, answers_and_none[:-1]) if a)
 
     messages = [msg1, msg2, msg3]
@@ -620,22 +620,18 @@ def analyze_composite(msg1, date1, msg2, date2, msg3, date3, *answers_and_none):
     if not active:
         return "Please enter at least one message."
 
-    # Run model on messages
     results = [(analyze_single_message(m, THRESHOLDS.copy()), d) for m, d in active]
-    # --- Combined Abuse Escalation Scoring ---
 
-# Extract predicted abuse labels from all messages
-    predicted_labels = [label for r in results for label, _ in r[0][2]]
+    abuse_scores = [r[0][0] for r in results]
+    stages = [r[0][4] for r in results]
+    darvo_scores = [r[0][5] for r in results]
+    tone_tags = [r[0][6] for r in results]
+    dates_used = [r[1] or "Undated" for r in results]
 
-# Categorize by severity
+    predicted_labels = [label for r in results for label, _ in r[0][2]]
     high = {'control'}
-    moderate = {
-        'gaslighting', 'dismissiveness', 'obscure language',
-        'insults', 'contradictory statements', 'guilt tripping'
-    }
+    moderate = {'gaslighting', 'dismissiveness', 'obscure language', 'insults', 'contradictory statements', 'guilt tripping'}
     low = {'blame shifting', 'projection', 'recovery phase'}
-
-# Count severity types
     counts = {'high': 0, 'moderate': 0, 'low': 0}
     for label in predicted_labels:
         if label in high:
@@ -645,126 +641,21 @@ def analyze_composite(msg1, date1, msg2, date2, msg3, date3, *answers_and_none):
         elif label in low:
             counts['low'] += 1
 
-# Derive abuse_risk from combinations
+    # Pattern escalation logic
+    pattern_escalation_risk = "Low"
     if counts['high'] >= 2 and counts['moderate'] >= 2:
-        abuse_risk = 'Critical'
+        pattern_escalation_risk = "Critical"
     elif (counts['high'] >= 2 and counts['moderate'] >= 1) or (counts['moderate'] >= 3) or (counts['high'] >= 1 and counts['moderate'] >= 2):
-        abuse_risk = 'High'
+        pattern_escalation_risk = "High"
     elif (counts['moderate'] == 2) or (counts['high'] == 1 and counts['moderate'] == 1) or (counts['moderate'] == 1 and counts['low'] >= 2) or (counts['high'] == 1 and sum(counts.values()) == 1):
-        abuse_risk = 'Moderate'
-    else:
-        abuse_risk = 'Low'
-
-# Combine abuse_risk and checklist score into final risk_level
-    if escalation_score is not None:
-        if escalation_score >= 8 or abuse_risk == 'Critical':
-            risk_level = 'Critical'
-        elif escalation_score >= 5 or abuse_risk == 'High':
-            risk_level = 'High'
-        elif escalation_score >= 2 or abuse_risk == 'Moderate':
-            risk_level = 'Moderate'
-        else:
-            risk_level = 'Low'
-    abuse_scores = [r[0][0] for r in results]
-    top_labels = [r[0][1][0] if r[0][1] else r[0][2][0][0] for r in results]
-    top_scores = [r[0][2][0][1] for r in results]
-    sentiments = [r[0][3]['label'] for r in results]
-    stages = [r[0][4] for r in results]
-    darvo_scores = [r[0][5] for r in results]
-    tone_tags= [r[0][6] for r in results]
-    dates_used = [r[1] or "Undated" for r in results]  # Store dates for future mapping
-    # Calculate escalation bump *after* model results exist
-    escalation_bump = 0
-    for result, _ in results:
-        abuse_score, threshold_labels, top_patterns, sentiment, stage, darvo_score, tone_tag = result
-        if darvo_score > 0.65:
-            escalation_bump += 3
-        if tone_tag in ["forced accountability flip", "emotional threat"]:
-            escalation_bump += 2
-        if abuse_score > 80:
-            escalation_bump += 2
-        if stage == 2:
-            escalation_bump += 3
+        pattern_escalation_risk = "Moderate"
 
-    # Now we can safely calculate hybrid_score
-    hybrid_score = escalation_score + escalation_bump if escalation_score is not None else 0
-    risk_level = (
-        "High" if hybrid_score >= 16 else
-        "Moderate" if hybrid_score >= 8 else
+    checklist_escalation_risk = "Unknown" if escalation_score is None else (
+        "Critical" if escalation_score >= 20 else
+        "Moderate" if escalation_score >= 10 else
         "Low"
     )
 
-    # Now compute scores and allow override
-    abuse_scores = [r[0][0] for r in results]
-    stages = [r[0][4] for r in results]
-
-    # Post-check override (e.g. stage 2 or high abuse score forces Moderate risk)
-    if any(score > 70 for score in abuse_scores) or any(stage == 2 for stage in stages):
-        if risk_level == "Low":
-            risk_level = "Moderate"
-
-    for result, date in results:
-        assert len(result) == 7, "Unexpected output from analyze_single_message"
-    from collections import Counter
-
-# --- Step 1: Pattern-based escalation scoring ---
-    flat_patterns = [label for r in results for (label, _) in r[0][2]]
-
-    PATTERN_RISKS = {
-        "blame shifting": "low",
-        "contradictory statements": "moderate",
-        "control": "high",
-        "dismissiveness": "moderate",
-        "gaslighting": "moderate",
-        "guilt tripping": "moderate",
-        "insults": "moderate",
-        "obscure language": "low",
-        "projection": "low",
-    "   recovery phase": "low"
-    }
-
-    risk_counts = Counter(PATTERN_RISKS.get(p, "unknown") for p in flat_patterns)
-    num_critical = 0  # no "critical" tags defined in your label set
-    num_high     = risk_counts["high"]
-    num_moderate = risk_counts["moderate"]
-    num_low      = risk_counts["low"]
-
-# Determine pattern-based escalation risk
-    pattern_escalation_risk = "Low"
-    if num_high >= 2 and num_moderate >= 2:
-        pattern_escalation_risk = "Critical"
-    elif num_high >= 2 and num_moderate >= 1:
-        pattern_escalation_risk = "High"
-    elif num_moderate >= 3:
-        pattern_escalation_risk = "High"
-    elif num_high == 1 and num_moderate >= 1:
-        pattern_escalation_risk = "Moderate"
-    elif num_moderate == 2:
-        pattern_escalation_risk = "Moderate"
-    elif num_moderate == 1 and num_low >= 2:
-        pattern_escalation_risk = "Moderate"
-    elif num_high == 1 and (num_high + num_moderate + num_low) == 1:
-        pattern_escalation_risk = "Moderate"
-
-# --- Step 2: Checklist escalation logic ---
-    if escalation_score is None:
-        escalation_text = "🚫 **Escalation Potential: Unknown** (Checklist not completed)\n"
-        escalation_text += "⚠️ This section was not completed. Escalation potential cannot be calculated.\n"
-        hybrid_score = 0
-    else:
-        hybrid_score = escalation_score + escalation_bump
-        risk_level = (
-            "🔴 High" if hybrid_score >= 16 else
-            "🟠 Moderate" if hybrid_score >= 8 else
-            "🟢 Low"
-        )
-
-        escalation_text = f"📈 **Escalation Potential: {risk_level} ({hybrid_score}/29)**\n"
-        escalation_text += "📋 This score comes from your checklist *and* detected high-risk patterns.\n"
-        escalation_text += "🧠 It reflects both situational risk and behavioral red flags."
-        
-
-# --- Step 3: Escalation bump from DARVO, tone, abuse score, etc.
     escalation_bump = 0
     for result, _ in results:
         abuse_score, _, _, sentiment, stage, darvo_score, tone_tag = result
@@ -777,31 +668,37 @@ def analyze_composite(msg1, date1, msg2, date2, msg3, date3, *answers_and_none):
         if stage == 2:
             escalation_bump += 3
 
-# --- Step 4: Final escalation risk level
-    def rank(risk_label):
-        return ["Low", "Moderate", "High", "Critical"].index(risk_label) if risk_label else 0
+    def rank(label):
+        return {"Low": 0, "Moderate": 1, "High": 2, "Critical": 3, "Unknown": 0}.get(label, 0)
 
     combined_score = rank(pattern_escalation_risk) + rank(checklist_escalation_risk) + escalation_bump
-
     escalation_risk = (
         "Critical" if combined_score >= 6 else
         "High" if combined_score >= 4 else
         "Moderate" if combined_score >= 2 else
         "Low"
     )
-# --- Composite Abuse Score using compute_abuse_score ---
-    composite_abuse_scores = []
 
+    if escalation_score is None:
+        escalation_text = "🚫 **Escalation Potential: Unknown** (Checklist not completed)\n⚠️ This section was not completed. Escalation potential is estimated using message data only.\n"
+        hybrid_score = 0
+    else:
+        hybrid_score = escalation_score + escalation_bump
+        escalation_text = f"📈 **Escalation Potential: {escalation_risk} ({hybrid_score}/29)**\n"
+        escalation_text += "📋 This score combines your safety checklist answers *and* detected high-risk behavior.\n"
+        escalation_text += f"• Pattern Risk: {pattern_escalation_risk}\n"
+        escalation_text += f"• Checklist Risk: {checklist_escalation_risk}\n"
+        escalation_text += f"• Escalation Bump: +{escalation_bump} (from DARVO, tone, intensity, etc.)"
+
+    # Composite Abuse Score
+    composite_abuse_scores = []
     for result, _ in results:
         _, _, top_patterns, sentiment, _, _, _ = result
         matched_scores = [(label, score, PATTERN_WEIGHTS.get(label, 1.0)) for label, score in top_patterns]
         final_score = compute_abuse_score(matched_scores, sentiment["label"])
         composite_abuse_scores.append(final_score)
-
     composite_abuse = int(round(sum(composite_abuse_scores) / len(composite_abuse_scores)))
 
-
-
     most_common_stage = max(set(stages), key=stages.count)
     stage_text = RISK_STAGE_LABELS[most_common_stage]
 
@@ -813,39 +710,17 @@ def analyze_composite(msg1, date1, msg2, date2, msg3, date3, *answers_and_none):
 
     out = f"Abuse Intensity: {composite_abuse}%\n"
     out += "📊 This reflects the strength and severity of detected abuse patterns in the message(s).\n\n"
-
-# Save this line for later use at the 
-    if escalation_score is None:
-        escalation_text = "📉 Escalation Potential: Unknown (Checklist not completed)\n"
-        escalation_text += "⚠️ *This section was not completed. Escalation potential is unknown.*\n"
-        hybrid_score = 0  # ✅ fallback so it's defined for generate_risk_snippet
-    else:
-        escalation_text = f"🧨 **Escalation Potential: {risk_level} ({escalation_score}/{sum(w for _, w in ESCALATION_QUESTIONS)})**\n"
-        escalation_text += "This score comes directly from the safety checklist and functions as a standalone escalation risk score.\n"
-        escalation_text += "It indicates how many serious risk factors are present based on your answers to the safety checklist.\n"
-        # Derive top_label from the strongest top_patterns across all messages
-    top_label = None
-    if results:
-        sorted_patterns = sorted(
-            [(label, score) for r in results for label, score in r[0][2]],
-            key=lambda x: x[1],
-            reverse=True
-    )
-        if sorted_patterns:
-            top_label = f"{sorted_patterns[0][0]} – {int(round(sorted_patterns[0][1] * 100))}%"
-        if top_label is None:
-            top_label = "Unknown – 0%"
-    out += generate_risk_snippet(composite_abuse, top_label, hybrid_score if escalation_score is not None else 0, most_common_stage)
+    out += generate_risk_snippet(composite_abuse, top_labels[0], hybrid_score, most_common_stage)
     out += f"\n\n{stage_text}"
     out += darvo_blurb
     out += "\n\n🎭 **Emotional Tones Detected:**\n"
     for i, tone in enumerate(tone_tags):
-        label = tone if tone else "none"
-        out += f"• Message {i+1}: *{label}*\n"
-    print(f"DEBUG: avg_darvo = {avg_darvo}")
-    pattern_labels = [r[0][2][0][0] for r in results]  # top label for each message
+        out += f"• Message {i+1}: *{tone or 'none'}*\n"
+
+    pattern_labels = [r[0][2][0][0] for r in results]
     timeline_image = generate_abuse_score_chart(dates_used, abuse_scores, pattern_labels)
     out += "\n\n" + escalation_text
+
     return out, timeline_image
     
 message_date_pairs = [