Create learning_loss_manager.pylearning_loss_manager.py

learning_loss_manager.pylearning_loss_manager.py

@@ -0,0 +1,581 @@
+# learning_loss_manager.py
+
+import os
+import json
+from datetime import datetime
+
+
+class LearningLossManager:
+    """
+    مدير الفاقد التعليمي.
+
+    وظيفته:
+    - يستقبل تقرير فحص المستوى.
+    - يستخرج الفجوات.
+    - يختار أول فجوة مهمة.
+    - يبني خطة علاجية.
+    - يجهز Prompt للشرح القصير بواسطة Qwen لاحقًا.
+    - يجهز Prompt للكويز بواسطة Phi لاحقًا.
+
+    ملاحظة:
+    هذا الملف لا يشغّل Qwen ولا Phi مباشرة.
+    هو فقط يجهز الخطة والطلبات حتى لا نضغط على ذاكرة المساحة الآن.
+    """
+
+    def __init__(self):
+        self.version = "learning_loss_manager_v1_0"
+        self.storage_dir = "learning_loss_sessions"
+        os.makedirs(self.storage_dir, exist_ok=True)
+
+    def now(self):
+        return datetime.utcnow().isoformat()
+
+    def safe_text(self, value, default=""):
+        text = str(value or "").strip()
+        return text if text else default
+
+    def slug(self, text):
+        text = str(text or "").strip()
+        keep = []
+
+        for ch in text:
+            if ch.isalnum() or ch in ["_", "-"]:
+                keep.append(ch)
+            elif "\u0600" <= ch <= "\u06FF":
+                keep.append(ch)
+            else:
+                keep.append("_")
+
+        result = "".join(keep).strip("_")
+        return result[:80] or "item"
+
+    def make_plan_id(self, diagnostic_report):
+        subject = diagnostic_report.get("subject_key") or diagnostic_report.get("subject") or "subject"
+        identity = diagnostic_report.get("identity") or "student"
+        stamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+        return f"loss_{self.slug(subject)}_{self.slug(identity)}_{stamp}"
+
+    def normalize_report(self, diagnostic_data):
+        """
+        يقبل:
+        - report النهائي مباشرة.
+        - state يحتوي report.
+        - dict يحتوي planning_payload.
+        """
+
+        if not isinstance(diagnostic_data, dict):
+            return {
+                "success": False,
+                "message": "بيانات الفحص غير صالحة.",
+                "report": {}
+            }
+
+        if isinstance(diagnostic_data.get("report"), dict):
+            return {
+                "success": True,
+                "message": "تم استخراج التقرير من حالة الفحص.",
+                "report": diagnostic_data.get("report")
+            }
+
+        if isinstance(diagnostic_data.get("planning_payload"), dict):
+            return {
+                "success": True,
+                "message": "البيانات هي تقرير أو تحتوي مخرجات التخطيط.",
+                "report": diagnostic_data
+            }
+
+        if diagnostic_data.get("badge_results") or diagnostic_data.get("history"):
+            return {
+                "success": True,
+                "message": "البيانات تبدو كتقرير فحص مباشر.",
+                "report": diagnostic_data
+            }
+
+        return {
+            "success": False,
+            "message": "لم أجد تقرير فحص أو planning_payload.",
+            "report": {}
+        }
+
+    def extract_planning_payload(self, report):
+        if not isinstance(report, dict):
+            return {}
+
+        payload = report.get("planning_payload")
+
+        if isinstance(payload, dict):
+            return payload
+
+        gap_badges = []
+        mastered_badges = []
+
+        for item in report.get("badge_results", []):
+            if item.get("status") == "mastered":
+                mastered_badges.append({
+                    "badge_id": item.get("badge_id"),
+                    "badge_title": item.get("badge_title"),
+                    "strengths": item.get("strengths", []),
+                })
+            else:
+                gap_badges.append({
+                    "badge_id": item.get("badge_id"),
+                    "badge_title": item.get("badge_title"),
+                    "status": item.get("status"),
+                    "decision": item.get("decision"),
+                    "first_gap": item.get("first_gap"),
+                    "gaps": item.get("gaps", []),
+                    "planner_instruction": item.get("planner_instruction"),
+                })
+
+        return {
+            "identity": report.get("identity"),
+            "subject": report.get("subject"),
+            "subject_key": report.get("subject_key"),
+            "grade": report.get("grade"),
+            "grade_number": report.get("grade_number"),
+            "mastered_badges": mastered_badges,
+            "gap_badges": gap_badges,
+            "planner_note": "تم بناء planning_payload داخل LearningLossManager.",
+        }
+
+    def choose_priority_gap(self, planning_payload):
+        gaps = planning_payload.get("gap_badges", [])
+
+        if not gaps:
+            return None
+
+        priority_order = {
+            "foundation_gap": 1,
+            "partial_gap": 2,
+            "needs_more_evidence": 3,
+            "near_mastery": 4,
+        }
+
+        def gap_score(item):
+            status = item.get("status", "needs_more_evidence")
+            return priority_order.get(status, 99)
+
+        sorted_gaps = sorted(gaps, key=gap_score)
+        return sorted_gaps[0]
+
+    def get_first_gap_detail(self, priority_gap):
+        if not isinstance(priority_gap, dict):
+            return {}
+
+        first_gap = priority_gap.get("first_gap")
+
+        if isinstance(first_gap, dict) and first_gap:
+            return first_gap
+
+        gaps = priority_gap.get("gaps", [])
+
+        if isinstance(gaps, list) and gaps:
+            if isinstance(gaps[0], dict):
+                return gaps[0]
+
+        return {
+            "micro_skill": "",
+            "micro_skill_title": "المهارة الضعيفة غير محددة بعد",
+            "reason": "تحتاج أدلة أكثر",
+        }
+
+    def build_plan_from_diagnostic_report(self, diagnostic_data):
+        normalized = self.normalize_report(diagnostic_data)
+
+        if not normalized.get("success"):
+            return {
+                "success": False,
+                "message": normalized.get("message"),
+                "plan": None,
+            }
+
+        report = normalized.get("report", {})
+        planning_payload = self.extract_planning_payload(report)
+        priority_gap = self.choose_priority_gap(planning_payload)
+
+        plan_id = self.make_plan_id(report)
+
+        if not priority_gap:
+            plan = {
+                "success": True,
+                "version": self.version,
+                "plan_id": plan_id,
+                "created_at": self.now(),
+                "updated_at": self.now(),
+                "source_test_id": report.get("test_id"),
+                "identity": planning_payload.get("identity") or report.get("identity"),
+                "subject": planning_payload.get("subject") or report.get("subject"),
+                "subject_key": planning_payload.get("subject_key") or report.get("subject_key"),
+                "grade": planning_payload.get("grade") or report.get("grade"),
+                "grade_number": planning_payload.get("grade_number") or report.get("grade_number"),
+                "status": "no_clear_gap",
+                "locked_until_done": False,
+                "message": "لا توجد فجوة واضحة من تقرير الفحص الحالي.",
+                "priority_gap": None,
+                "steps": [],
+                "qwen_prompt": "",
+                "phi_quiz_prompt": "",
+                "planning_payload": planning_payload,
+            }
+
+            self.save_plan(plan)
+            return {
+                "success": True,
+                "message": "لا توجد فجوة واضحة، وتم إنشاء سجل فارغ.",
+                "plan": plan,
+            }
+
+        first_gap = self.get_first_gap_detail(priority_gap)
+        steps = self.build_learning_loss_queue(priority_gap, first_gap)
+
+        plan = {
+            "success": True,
+            "version": self.version,
+            "plan_id": plan_id,
+            "created_at": self.now(),
+            "updated_at": self.now(),
+            "source_test_id": report.get("test_id"),
+            "identity": planning_payload.get("identity") or report.get("identity"),
+            "subject": planning_payload.get("subject") or report.get("subject"),
+            "subject_key": planning_payload.get("subject_key") or report.get("subject_key"),
+            "grade": planning_payload.get("grade") or report.get("grade"),
+            "grade_number": planning_payload.get("grade_number") or report.get("grade_number"),
+            "status": "open",
+            "locked_until_done": True,
+            "message": "تم إنشاء خطة فاقد تعليمي من تقرير فحص المستوى.",
+            "priority_gap": priority_gap,
+            "first_gap": first_gap,
+            "steps": steps,
+            "current_step_index": 0,
+            "qwen_prompt": "",
+            "phi_quiz_prompt": "",
+            "planning_payload": planning_payload,
+        }
+
+        plan["qwen_prompt"] = self.build_qwen_prompt(plan)
+        plan["phi_quiz_prompt"] = self.build_phi_quiz_prompt(plan)
+
+        self.save_plan(plan)
+
+        return {
+            "success": True,
+            "message": "تم بناء خطة الفاقد التعليمي.",
+            "plan": plan,
+        }
+
+    def build_learning_loss_queue(self, priority_gap, first_gap):
+        badge_title = priority_gap.get("badge_title", "وسام غير محدد")
+        status = priority_gap.get("status", "partial_gap")
+        micro_skill_title = first_gap.get("micro_skill_title", "المهارة الضعيفة")
+
+        return [
+            {
+                "step_id": "short_lesson",
+                "title": "شرح علاجي قصير",
+                "engine": "qwen",
+                "status": "pending",
+                "locked": False,
+                "goal": f"شرح مبسط للمهارة: {micro_skill_title}",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": True,
+            },
+            {
+                "step_id": "guided_practice",
+                "title": "تدريب موجّه",
+                "engine": "qwen",
+                "status": "pending",
+                "locked": True,
+                "goal": "حل مثالين مع الطالب خطوة بخطوة دون كشف الحل مباشرة.",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": True,
+            },
+            {
+                "step_id": "worksheet",
+                "title": "ورقة عمل قصيرة",
+                "engine": "local_or_qwen",
+                "status": "pending",
+                "locked": True,
+                "goal": "إنشاء 5 أسئلة متدرجة على نفس المهارة.",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": True,
+            },
+            {
+                "step_id": "educational_game",
+                "title": "لعبة تعليمية",
+                "engine": "games_manager",
+                "status": "pending",
+                "locked": True,
+                "goal": "اختيار لعبة مناسبة لترسيخ المهارة بعد الشرح والورقة.",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": False,
+            },
+            {
+                "step_id": "homework",
+                "title": "واجب قصير",
+                "engine": "local_or_qwen",
+                "status": "pending",
+                "locked": True,
+                "goal": "واجب من 3 أسئلة للتثبيت المنزلي.",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": True,
+            },
+            {
+                "step_id": "phi_quiz",
+                "title": "كويز تحقق",
+                "engine": "phi",
+                "status": "pending",
+                "locked": True,
+                "goal": "كويز جديد لا يكرر سؤال الفحص.",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": True,
+            },
+            {
+                "step_id": "reassessment",
+                "title": "إعادة تقييم المهارة",
+                "engine": "phi",
+                "status": "pending",
+                "locked": True,
+                "goal": "تحديد هل أُغلقت الفجوة أم تحتاج دورة علاجية ثانية.",
+                "badge_title": badge_title,
+                "gap_status": status,
+                "required_to_unlock_next": True,
+            },
+        ]
+
+    def build_qwen_prompt(self, plan):
+        priority_gap = plan.get("priority_gap") or {}
+        first_gap = plan.get("first_gap") or {}
+
+        subject = plan.get("subject") or plan.get("subject_key") or "المادة"
+        grade = plan.get("grade") or plan.get("grade_number") or "غير محدد"
+        badge_title = priority_gap.get("badge_title", "وسام غير محدد")
+        status = priority_gap.get("status", "partial_gap")
+        micro_skill_title = first_gap.get("micro_skill_title", "المهارة الضعيفة")
+        reason = first_gap.get("reason", "إجابة خاطئة أو لا أعلم")
+
+        payload = {
+            "role": "Qwen teacher",
+            "task": "نفذ شرحًا علاجيًا قصيرًا لفجوة تعليمية محددة.",
+            "student_context": {
+                "identity": plan.get("identity"),
+                "subject": subject,
+                "grade": grade,
+            },
+            "gap": {
+                "badge_title": badge_title,
+                "status": status,
+                "micro_skill_title": micro_skill_title,
+                "reason": reason,
+                "planner_instruction": priority_gap.get("planner_instruction"),
+            },
+            "teaching_rules": [
+                "ابدأ بسؤال افتتاحي قصير للتأكد من حضور الطالب.",
+                "اشرح المهارة ببساطة وبأمثلة قليلة.",
+                "لا تنتقل للخطوة التالية قبل سؤال تحقق.",
+                "إذا أجاب الطالب خطأ، أعد الشرح بطريقة أبسط.",
+                "لا تجعل الحصة طويلة.",
+                "لا تعطِ كويز نهائي، لأن الكويز سيبنيه Phi لاحقًا.",
+                "استخدم العربية الواضحة، ومع الإنجليز�� اذكر ترجمة عربية عند الحاجة.",
+            ],
+            "required_output": {
+                "opening_question": "سؤال حضور بسيط",
+                "short_explanation": "شرح قصير",
+                "example_1": "مثال محلول جزئيًا",
+                "check_question": "سؤال تحقق بعد الشرح",
+                "next_action": "انتظار إجابة الطالب",
+            },
+        }
+
+        return json.dumps(payload, ensure_ascii=False, indent=2)
+
+    def build_phi_quiz_prompt(self, plan):
+        priority_gap = plan.get("priority_gap") or {}
+        first_gap = plan.get("first_gap") or {}
+
+        payload = {
+            "role": "Phi evaluator",
+            "task": "اصنع كويز تحقق قصير بعد الحصة العلاجية.",
+            "student_context": {
+                "identity": plan.get("identity"),
+                "subject": plan.get("subject"),
+                "subject_key": plan.get("subject_key"),
+                "grade": plan.get("grade"),
+                "grade_number": plan.get("grade_number"),
+            },
+            "gap": {
+                "badge_id": priority_gap.get("badge_id"),
+                "badge_title": priority_gap.get("badge_title"),
+                "status": priority_gap.get("status"),
+                "micro_skill": first_gap.get("micro_skill"),
+                "micro_skill_title": first_gap.get("micro_skill_title"),
+                "reason": first_gap.get("reason"),
+            },
+            "rules": [
+                "أرجع JSON فقط.",
+                "اصنع 3 أسئلة جديدة لا تكرر سؤال الفحص.",
+                "كل سؤال له 4 اختيارات.",
+                "answer_index رقم من 0 إلى 3.",
+                "لا تكشف الحل للطالب قبل الإجابة.",
+                "بعد التصحيح أعط قرارًا: gap_closed أو needs_more_practice.",
+            ],
+            "required_json": {
+                "quiz_id": "quiz_unique_id",
+                "questions": [
+                    {
+                        "question_id": "q1",
+                        "question": "نص السؤال",
+                        "choices": ["أ", "ب", "ج", "د"],
+                        "answer_index": 0,
+                        "correct_answer": "أ",
+                        "micro_skill_title": "اسم المهارة",
+                    }
+                ],
+                "closing_rule": "gap_closed if score >= 2/3 else needs_more_practice",
+            },
+        }
+
+        return json.dumps(payload, ensure_ascii=False, indent=2)
+
+    def get_current_step(self, plan):
+        steps = plan.get("steps", [])
+        index = int(plan.get("current_step_index", 0))
+
+        if index < 0 or index >= len(steps):
+            return None
+
+        return steps[index]
+
+    def complete_current_step(self, plan, result=None):
+        if not isinstance(plan, dict):
+            return {
+                "success": False,
+                "message": "الخطة غير صالحة.",
+                "plan": plan,
+            }
+
+        steps = plan.get("steps", [])
+        index = int(plan.get("current_step_index", 0))
+
+        if index < 0 or index >= len(steps):
+            plan["status"] = "completed"
+            plan["locked_until_done"] = False
+            plan["updated_at"] = self.now()
+            self.save_plan(plan)
+
+            return {
+                "success": True,
+                "message": "الخطة مكتملة.",
+                "plan": plan,
+            }
+
+        steps[index]["status"] = "completed"
+        steps[index]["completed_at"] = self.now()
+        steps[index]["result"] = result or {}
+
+        next_index = index + 1
+
+        if next_index < len(steps):
+            steps[next_index]["locked"] = False
+            plan["current_step_index"] = next_index
+            plan["status"] = "open"
+            message = f"تم إنهاء الخطوة، والانتقال إلى: {steps[next_index].get('title')}"
+        else:
+            plan["current_step_index"] = next_index
+            plan["status"] = "completed"
+            plan["locked_until_done"] = False
+            message = "تم إنهاء خطة الفاقد التعليمي."
+
+        plan["steps"] = steps
+        plan["updated_at"] = self.now()
+        self.save_plan(plan)
+
+        return {
+            "success": True,
+            "message": message,
+            "plan": plan,
+        }
+
+    def save_plan(self, plan):
+        try:
+            plan_id = plan.get("plan_id") or "learning_loss_plan"
+            path = os.path.join(self.storage_dir, f"{self.slug(plan_id)}.json")
+
+            with open(path, "w", encoding="utf-8") as f:
+                json.dump(plan, f, ensure_ascii=False, indent=2)
+
+            plan["saved_path"] = path
+        except Exception:
+            pass
+
+        return plan
+
+    def load_plan(self, plan_id):
+        path = os.path.join(self.storage_dir, f"{self.slug(plan_id)}.json")
+
+        if not os.path.exists(path):
+            return None
+
+        try:
+            with open(path, "r", encoding="utf-8") as f:
+                return json.load(f)
+        except Exception:
+            return None
+
+    def format_plan(self, plan):
+        if not isinstance(plan, dict):
+            return "لا توجد خطة فاقد تعليمي."
+
+        if plan.get("status") == "no_clear_gap":
+            return f"""
+# الفاقد التعليمي
+
+✅ لا توجد فجوة واضحة حاليًا.
+
+المادة: {plan.get('subject')}
+الصف: {plan.get('grade')}
+
+الرسالة:
+{plan.get('message')}
+""".strip()
+
+        priority_gap = plan.get("priority_gap") or {}
+        first_gap = plan.get("first_gap") or {}
+        current_step = self.get_current_step(plan)
+
+        text = f"""
+# خطة الفاقد التعليمي
+
+الحالة: {plan.get('status')}
+الخطة مقفلة حتى الإكمال: {plan.get('locked_until_done')}
+
+المادة: {plan.get('subject')}
+الصف: {plan.get('grade')}
+
+## الفجوة المختارة أولًا
+
+الوسام: {priority_gap.get('badge_title')}
+الحالة: {priority_gap.get('status')}
+المهارة الصغيرة: {first_gap.get('micro_skill_title')}
+سبب الفجوة: {first_gap.get('reason')}
+تعليمات التخطيط: {priority_gap.get('planner_instruction')}
+
+## الخطوة الحالية
+
+{current_step.get('title') if current_step else 'لا توجد خطوة حالية'}
+
+## خطوات الخطة
+""".strip()
+
+        for i, step in enumerate(plan.get("steps", []), start=1):
+            lock_text = "مقفلة" if step.get("locked") else "مفتوحة"
+            text += f"""
+
+{i}. {step.get('title')}
+- المحرك: {step.get('engine')}
+- الحالة