作业一难就放弃？把「扛得住的心韧力」做成训练闭环

# resilience_tracker.py —— 心理能量数据模型与挫折事件探测
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum

class Tracker(str, Enum):
    DIRECTING  = "心航力"     # 方向：知不知道要去哪
    FOCUS      = "心定力"     # 专注：能不能待在任务上
    RESILIENCE = "心韧力"     # 抗挫：卡住后能否复原
    EFFICACY   = "心效力"     # 胜任：相不相信能做成

@dataclass
class FrustrationEvent:
    student_id: str
    subject: str
    stuck_duration_sec: int       # 卡壳停顿秒数
    emotion_signal: str           # 情绪外显信号（叹气/揉橡皮/发呆/烦躁）
    task_name: str
    timestamp: datetime = field(default_factory=datetime.now)

def evaluate_frustration_signal(event: FrustrationEvent) -> dict:
    """根据卡壳时长与情绪表征判定是否需要启动散热保护"""
    # 规则：停顿超过300秒，或出现明确的高摩擦情绪表征
    is_critical = event.stuck_duration_sec >= 300 or event.emotion_signal in ["叹气", "摔笔", "抓头", "烦躁"]

    return {
        "student_id": event.student_id,
        "action_required": "TRIGGER_COOLING" if is_critical else "CONTINUE_OBSERVING",
        "dimension": Tracker.RESILIENCE.value,
        "urgency": "HIGH" if is_critical else "LOW",
        "reason": "前额叶认知负荷超出安全阈值" if is_critical else "正常思考区间"
    }

if __name__ == "__main__":
    test_event = FrustrationEvent(
        student_id="S_XIAOKAI_06",
        subject="数学",
        stuck_duration_sec=360,
        emotion_signal="烦躁",
        task_name="几何综合题第2问"
    )
    decision = evaluate_frustration_signal(test_event)
    print("探测决策结果:", decision)

# cooling_protocol.py —— 受控散热协议与降温规则生成器

SYSTEM_COOLING_RULES = """
【安全规则】
1. 严禁评价孩子的能力、智商或以往表现；
2. 严禁催促完成时间；
3. 第一句话必须客观确认阻滞事实并接纳负面情绪；
4. 必须给出具体物理微动作（如离椅喝水、深呼吸），时长限定在2分钟内。
"""

def generate_cooling_instruction(student_name: str, subject: str) -> dict:
    """生成标准化家庭书桌散热指令"""
    return {
        "step_1_ack": f"{student_name}，这道{subject}题步骤确实很多，想不出来很正常，先停笔。",
        "step_2_physical_reset": "把笔放下，站起来去厨房倒半杯温水喝，闭上眼睛做3次慢速深呼吸。",
        "step_3_timer_sec": 120,
        "protocol_status": "ACTIVE_COOLING"
    }

if __name__ == "__main__":
    protocol = generate_cooling_instruction("小楷", "数学")
    print("执行散热指令:", protocol)

# task_stepper.py —— 难题台阶式降解引擎
from dataclasses import dataclass
from typing import List

@dataclass
class MicroStep:
    step_no: int
    instruction: str
    difficulty_level: str
    is_completed: bool = False

def decompose_stuck_task(raw_question: str) -> List[MicroStep]:
    """将一道阻断孩子的综合大题解构为3个微小台阶"""
    # 工业实现中可接入大模型完成条件拆解，此处展示确定性降解结构
    steps = [
        MicroStep(
            step_no=1,
            instruction="不求算出最终数字，在草稿纸上抄写出已知条件中的3个核心数据点，画出受力或几何草图。",
            difficulty_level="VERY_LOW"
        ),
        MicroStep(
            step_no=2,
            instruction="找出与第1问结论相关联的公式，仅写出公式字母表达式，不代入复杂数值。",
            difficulty_level="LOW"
        ),
        MicroStep(
            step_no=3,
            instruction="尝试把已知数值代入公式第1步，若计算耗时超3分钟则先保留表达式。",
            difficulty_level="MEDIUM"
        )
    ]
    return steps

if __name__ == "__main__":
    task_steps = decompose_stuck_task("动点压轴综合题")
    for s in task_steps:
        print(f"台阶 {s.step_no} [{s.difficulty_level}]: {s.instruction}")

# resilience_store.py —— 韧性事件本地存储与周报回读
import sqlite3
from datetime import date

class ResilienceStore:
    def __init__(self, db_path=":memory:"):
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        with self.conn:
            self.conn.execute("""
            CREATE TABLE IF NOT EXISTS resilience_log (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                student_id TEXT NOT NULL,
                event_date TEXT NOT NULL,
                subject TEXT NOT NULL,
                stuck_reason TEXT NOT NULL,
                cooled_down INTEGER NOT NULL,
                recovered INTEGER NOT NULL
            );
            """)

    def record_recovery(self, student_id: str, subject: str, reason: str, cooled: bool, recovered: bool):
        with self.conn:
            self.conn.execute("""
            INSERT INTO resilience_log (student_id, event_date, subject, stuck_reason, cooled_down, recovered)
            VALUES (?, ?, ?, ?, ?, ?);
            """, (student_id, str(date.today()), subject, reason, 1 if cooled else 0, 1 if recovered else 0))

    def generate_weekly_report(self, student_id: str) -> dict:
        cur = self.conn.cursor()
        cur.execute("""
        SELECT COUNT(*), SUM(cooled_down), SUM(recovered)
        FROM resilience_log
        WHERE student_id = ?;
        """, (student_id,))
        total, cooled, recovered = cur.fetchone()

        return {
            "student_id": student_id,
            "total_frustration_events": total or 0,
            "successfully_cooled": cooled or 0,
            "successfully_recovered": recovered or 0,
            "recovery_rate_pct": round((recovered / total) * 100, 1) if total else 0.0,
            "summary_feedback": f"本周共遭遇 {total} 次卡壳阻断，经过前额叶散热，成功自主复原 {recovered} 次。心韧力正在形成稳定习惯！"
        }

if __name__ == "__main__":
    store = ResilienceStore()
    store.record_recovery("S_XIAOKAI_06", "数学", "几何辅助线卡壳", cooled=True, recovered=True)
    store.record_recovery("S_XIAOKAI_06", "科学", "实验大题推导过载", cooled=True, recovered=True)
    report = store.generate_weekly_report("S_XIAOKAI_06")
    print("周复盘回读指标:", report)