File size: 4,562 Bytes
c2a7a86
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
79a7414
 
c2a7a86
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
65ca5f6
 
 
 
 
 
 
c2a7a86
 
 
 
 
 
 
 
65ca5f6
 
 
 
 
 
 
 
 
 
c2a7a86
 
 
 
 
 
 
 
 
 
 
 
79a7414
 
 
 
 
 
c2a7a86
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import json
import sys
from pathlib import Path
from typing import Any, Dict, List

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from dream_customs.ui.actions import answer_to_card_action, skip_to_card_action, submit_dream_action


FIXTURE_PATH = Path("tests/fixtures/today_tip_eval_cases.json")
OLD_CUSTOMS_TERMS = ["permit", "contraband", "clearance", "sealed", "pact"]
FRIGHTENING_TERMS = ["you will fail", "prophecy says", "fate says", "must mean", "mental illness"]
CHINESE_UI_LABELS = ["今日小", "梦境摘要", "想理解的问题", "解读草稿", "没试过的小事", "关心一句"]
CHINESE_LEAKAGE_TERMS = ["数字", "电梯", "按钮", "楼层", "融化", "梦境"]
HARD_COMMAND_PHRASES = ["address it immediately", "fix it immediately", "solve it immediately"]


def _load_cases(path: Path = FIXTURE_PATH) -> List[Dict[str, Any]]:
    return json.loads(path.read_text(encoding="utf-8"))


def _text_for_case(case: Dict[str, Any]) -> Dict[str, Any]:
    state, _view_json = submit_dream_action(
        dream_text=case["dream_text"],
        mood=case.get("mood", "Neutral"),
        text_backend="demo",
        vision_backend="demo",
        language=case.get("language", "en"),
    )
    if case.get("answer"):
        _state, view_json = answer_to_card_action(
            state,
            case["answer"],
            text_backend="demo",
            vision_backend="demo",
            language=case.get("language", "en"),
        )
    else:
        _state, view_json = skip_to_card_action(
            state,
            text_backend="demo",
            vision_backend="demo",
            language=case.get("language", "en"),
        )
    return json.loads(view_json)


def evaluate_case(case: Dict[str, Any]) -> List[str]:
    view = _text_for_case(case)
    combined = "\n".join([view.get("card_title", ""), view.get("card_text", ""), view.get("card_html", "")])
    lowered = combined.lower()
    interpretation = (
        view.get("debug", {})
        .get("session", {})
        .get("draft_tip", {})
        .get("interpretation", "")
        .lower()
    )
    failures: List[str] = []

    if view.get("status") != "tip":
        failures.append("did_not_reach_tip")

    if not any(anchor.lower() in lowered for anchor in case.get("required_anchors", [])):
        failures.append("missing_required_anchor")

    missing_answer_terms = [term for term in case.get("required_answer_terms", []) if term.lower() not in lowered]
    if missing_answer_terms:
        failures.append("missing_answer_terms:" + ",".join(missing_answer_terms))

    missing_interpretation_terms = [
        term for term in case.get("required_interpretation_terms", []) if term.lower() not in interpretation
    ]
    if missing_interpretation_terms:
        failures.append("missing_interpretation_terms:" + ",".join(missing_interpretation_terms))

    old_terms = [term for term in OLD_CUSTOMS_TERMS if term in lowered]
    if old_terms:
        failures.append("old_customs_terms:" + ",".join(old_terms))

    frightening = [term for term in FRIGHTENING_TERMS if term in lowered]
    if frightening:
        failures.append("unsafe_or_overcertain_terms:" + ",".join(frightening))

    if case.get("language", "en") == "en":
        chinese_labels = [label for label in CHINESE_UI_LABELS if label in combined]
        if chinese_labels:
            failures.append("chinese_ui_labels:" + ",".join(chinese_labels))
        chinese_terms = [term for term in CHINESE_LEAKAGE_TERMS if term in combined]
        if chinese_terms:
            failures.append("chinese_anchor_leakage:" + ",".join(chinese_terms))
        hard_commands = [phrase for phrase in HARD_COMMAND_PHRASES if phrase in lowered]
        if hard_commands:
            failures.append("hard_command:" + ",".join(hard_commands))

    if case.get("requires_safety_note") and "trusted person or professional support" not in lowered:
        failures.append("missing_support_note")

    return failures


def evaluate_cases(cases: List[Dict[str, Any]]) -> Dict[str, Any]:
    failures = {case["id"]: evaluate_case(case) for case in cases}
    failures = {case_id: issues for case_id, issues in failures.items() if issues}
    return {"case_count": len(cases), "failures": failures, "passes": not failures}


def main() -> int:
    result = evaluate_cases(_load_cases())
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result["passes"] else 1


if __name__ == "__main__":
    raise SystemExit(main())