scripts/audit_fusion_behavior.py

import json
import os
import sys
from pathlib import Path


BACKEND_DIR = Path(__file__).resolve().parents[1]
PROJECT_DIR = BACKEND_DIR.parent
LOCAL_EMOTION = PROJECT_DIR / "saved_models" / "emotion_v2"
LOCAL_SARCASM = PROJECT_DIR / "saved_models" / "sarcasm_v4"

if LOCAL_EMOTION.is_dir():
    os.environ.setdefault("USE_LOCAL_MODELS", "true")
    os.environ.setdefault("MOODLENS_EMOTION_MODEL_ID", str(LOCAL_EMOTION))

if LOCAL_SARCASM.is_dir():
    os.environ.setdefault("USE_LOCAL_MODELS", "true")
    os.environ.setdefault("MOODLENS_SARCASM_MODEL_ID", str(LOCAL_SARCASM))

sys.path.insert(0, str(BACKEND_DIR))
sys.path.insert(0, str(Path(__file__).resolve().parent))

from app.services.fusion_engine import analyze_text  # noqa: E402
from fusion_balance_cases import BALANCE_CASES  # noqa: E402


def compact_audit(result):
    return {
        "input": result["input_text"],
        "overall_mood": result["overall"]["overall_mood"],
        "overall_mood_score": result["overall"]["mood_score"],
        "dominant_emotion": result["overall"]["dominant_emotion"],
        "statements": [
            {
                "text": statement["text"],
                "raw_emotion_top3": statement["top3_emotions"],
                "raw_sarcasm_probabilities": {
                    "Not Sarcastic": statement["raw_sarcasm"]["model_not_sarcasm_score"],
                    "Sarcastic": statement["raw_sarcasm"]["model_sarcasm_score"],
                },
                "sarcastic_index": statement.get("sarcastic_index", 1),
                "raw_sarcasm_score": statement["raw_sarcasm"]["model_sarcasm_score"],
                "threshold_used": statement.get("sarcasm_threshold_used"),
                "raw_sarcasm_label": statement["raw_sarcasm"]["label"],
                "rule_applied": statement.get("rule_applied"),
                "calibration_reason": statement.get("calibration_reason"),
                "final_sarcasm_label": statement["sarcasm_label"],
                "final_primary_emotion": statement["primary_emotion"],
                "final_sentiment": statement["sentiment"],
                "final_mood_score": statement["mood_score"],
                "interpretation": statement["interpretation"],
                "reason": statement["sarcasm_reason"],
            }
            for statement in result["statements"]
        ],
    }


def main():
    texts = sys.argv[1:]
    if not texts:
        texts = [
            text
            for case in BALANCE_CASES
            for text in case["texts"]
        ]

    for text in texts:
        print(json.dumps(compact_audit(analyze_text(text)), indent=2))


if __name__ == "__main__":
    main()