# badge_progress_manager.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ مدير تقدم الأوسمة. الإصدار V10 متوافق مع BadgeGameManager V10 ويضيف: - تسجيل حالات correct / wrong / unknown بصورة منفصلة. - عدم منح الوسام إلا عند إتقان 100% دون إجابة خاطئة أو "لا أعلم". - إبقاء الطالب في الوسام نفسه عند الحاجة إلى حصة علاجية. - تسجيل الأخطاء المتكررة وطلبات المساعدة. - فتح الوسام التالي فقط بعد إتقان الوسام الحالي. - حفظ محلي آمن مع مزامنة اختيارية إلى StudentManager. - المحافظة على أسماء الدوال القديمة التي تستخدمها app.py وSessionManager. هذا الملف لا يستخدم GPU ولا يستدعي أي نموذج ذكاء اصطناعي. """ from __future__ import annotations from copy import deepcopy from datetime import datetime, timezone import json import os from pathlib import Path import re import threading import uuid from typing import Any, Dict, Iterable, List, Optional, Tuple class BadgeProgressManager: """يدير خريطة الأوسمة ومحاولات الطالب وحالة الإتقان.""" VERSION = "badge_progress_manager_v32_evidence_ledger_arabic_system" SCHEMA_VERSION = "badge_progress_schema_v32" DEFAULT_MASTERY_REQUIRED = 100.0 STATUS_LOCKED = "locked" STATUS_AVAILABLE = "available" STATUS_IN_PROGRESS = "in_progress" STATUS_REMEDIAL_REQUIRED = "remedial_required" STATUS_EARNED = "earned" CORE_MATH_BADGES = [ "math_addition", "math_subtraction", "math_multiplication", "math_division", "math_fractions", "math_decimals", "math_percentages", "math_integers", "math_equations", "math_geometry_shapes", "math_measurement_units", "math_word_problems", ] SUBJECT_NAMES = { "math": "رياضيات", "arabic": "لغة عربية", "english": "لغة إنجليزية", "science": "علوم", "physics": "فيزياء", "chemistry": "كيمياء", "biology": "أحياء", "earth_science": "علوم الأرض", "computer": "حاسوب", } GENERIC_ROADMAPS = { "arabic": [ ("arabic_letters", "وسام الحروف والأصوات", "الحروف والأصوات"), ("arabic_words", "وسام الكلمات", "قراءة الكلمات"), ("arabic_sentences", "وسام الجمل", "فهم الجمل"), ("arabic_reading", "وسام القراءة", "القراءة والفهم"), ("arabic_grammar", "وسام القواعد", "القواعد الأساسية"), ("arabic_writing", "وسام الكتابة", "التعبير والكتابة"), ], "english": [ ("english_letters", "Letters Badge", "letters and sounds"), ("english_vocabulary", "Vocabulary Badge", "vocabulary"), ("english_sentences", "Sentences Badge", "sentences"), ("english_reading", "Reading Badge", "reading comprehension"), ("english_grammar", "Grammar Badge", "grammar"), ("english_writing", "Writing Badge", "writing"), ], "science": [ ("science_observation", "وسام الملاحظة العلمية", "الملاحظة"), ("science_living", "وسام الكائنات الحية", "الكائنات الحية"), ("science_matter", "وسام المادة", "المادة"), ("science_energy", "وسام الطاقة", "الطاقة"), ("science_earth", "وسام الأرض والفضاء", "الأرض والفضاء"), ], } def __init__( self, progress_root: str = "student_badge_progress", student_manager: Any = None, academic_manager: Any = None, mastery_required: Optional[float] = None, *args: Any, **kwargs: Any, ) -> None: self.version = self.VERSION self.schema_version = self.SCHEMA_VERSION self.progress_root = str(progress_root or "student_badge_progress") self.base_dir = self.progress_root self.mastery_required = self._safe_float( mastery_required if mastery_required is not None else os.getenv("BADGE_MASTERY_REQUIRED", self.DEFAULT_MASTERY_REQUIRED), self.DEFAULT_MASTERY_REQUIRED, ) self.mastery_required = min(100.0, max(1.0, self.mastery_required)) self.required_mastery_percentage = self.mastery_required self._student_manager = student_manager self._academic_manager = academic_manager self.storage_service = None self.arabic_skills_repo_key = os.getenv("ARABIC_DIAGNOSTIC_REPO_KEY", "system").strip() or "system" self.arabic_skills_repo_path = "diagnostics/arabic/arabic_skills_global.json" self._arabic_skills_cache: Dict[str, Any] = {} self._lock = threading.RLock() Path(self.progress_root).mkdir(parents=True, exist_ok=True) # ========================================================= # أدوات عامة # ========================================================= def now(self) -> str: return datetime.now(timezone.utc).isoformat() def set_runtime_dependencies(self, academic_service=None, student_service=None, storage_service=None): self.academic_service = academic_service self.student_service = student_service self._academic_manager = academic_service self._student_manager = student_service self.storage_service = storage_service self._arabic_skills_cache = {} return { "success": True, "dependency_owner": "SessionManager", "storage_injected": storage_service is not None, "version": self.version, } def _safe_int(self, value: Any, default: int = 0) -> int: try: return int(float(str(value).strip())) except (TypeError, ValueError): return int(default) def _safe_float(self, value: Any, default: float = 0.0) -> float: try: return float(str(value).strip()) except (TypeError, ValueError): return float(default) def _safe_bool(self, value: Any, default: bool = False) -> bool: if isinstance(value, bool): return value if value is None: return default return str(value).strip().lower() in { "1", "true", "yes", "y", "on", "enabled", "ready", "done" } def normalize_identity(self, value: Any) -> str: text = str(value or "").strip() text = re.sub(r"[^a-zA-Z0-9\u0600-\u06FF_-]+", "_", text) text = re.sub(r"_+", "_", text).strip("_-") return text[:120] or "student" def normalize_subject(self, subject: Any) -> str: text = str(subject or "").strip().lower() aliases = { "رياضيات": "math", "رياضيات أساسية": "math", "mathematics": "math", "math": "math", "عربي": "arabic", "لغة عربية": "arabic", "arabic": "arabic", "إنجليزي": "english", "انجليزي": "english", "لغة إنجليزية": "english", "english": "english", "علوم": "science", "science": "science", "فيزياء": "physics", "physics": "physics", "كيمياء": "chemistry", "chemistry": "chemistry", "أحياء": "biology", "احياء": "biology", "biology": "biology", "علوم الأرض": "earth_science", "علوم الارض": "earth_science", "earth_science": "earth_science", "حاسوب": "computer", "computer": "computer", } return aliases.get(text, re.sub(r"[^a-z0-9_]+", "_", text).strip("_") or "math") def subject_name(self, subject: Any) -> str: key = self.normalize_subject(subject) return self.SUBJECT_NAMES.get(key, str(subject or key)) def normalize_grade(self, grade: Any, default: int = 1) -> int: text = str(grade if grade is not None else "").strip().lower().translate( str.maketrans("٠١٢٣٤٥٦٧٨٩", "0123456789") ) if text in {"0", "k", "kg", "kindergarten", "روضة", "الروضة", "روضه"}: return 0 digits = "".join(char for char in text if char.isdigit()) value = self._safe_int(digits or text, default) return max(0, min(12, value)) def normalize_badge_id(self, badge_id: Any) -> str: text = str(badge_id or "").strip().lower() aliases = { "": "", "addition": "math_addition", "جمع": "math_addition", "subtraction": "math_subtraction", "طرح": "math_subtraction", "multiplication": "math_multiplication", "ضرب": "math_multiplication", "division": "math_division", "قسمة": "math_division", "fractions": "math_fractions", "fraction": "math_fractions", "كسور": "math_fractions", "decimals": "math_decimals", "decimal": "math_decimals", "percent": "math_percentages", "percentage": "math_percentages", "math_percent": "math_percentages", "integers": "math_integers", "equations": "math_equations", "geometry": "math_geometry_shapes", "math_geometry": "math_geometry_shapes", "measurement": "math_measurement_units", "math_measurement": "math_measurement_units", "word_problems": "math_word_problems", } return aliases.get(text, text) def _json_safe(self, value: Any) -> Any: if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, dict): return {str(key): self._json_safe(item) for key, item in value.items()} if isinstance(value, (list, tuple, set)): return [self._json_safe(item) for item in value] return str(value) def _read_json(self, path: str, default: Any = None) -> Any: if default is None: default = {} try: if not path or not os.path.exists(path): return deepcopy(default) with open(path, "r", encoding="utf-8") as file: value = json.load(file) return value except Exception: return deepcopy(default) def _write_json(self, path: str, data: Any) -> str: parent = os.path.dirname(path) if parent: Path(parent).mkdir(parents=True, exist_ok=True) temporary = f"{path}.tmp" with open(temporary, "w", encoding="utf-8") as file: json.dump(self._json_safe(data), file, ensure_ascii=False, indent=2) os.replace(temporary, path) return path # ========================================================= # مسارات الحفظ والهجرة # ========================================================= def progress_path(self, identity: Any, subject: Any) -> str: identity_key = self.normalize_identity(identity) subject_key = self.normalize_subject(subject) return os.path.join(self.progress_root, identity_key, f"{subject_key}.json") def legacy_paths(self, identity: Any, subject: Any) -> List[str]: identity_key = self.normalize_identity(identity) subject_key = self.normalize_subject(subject) return [ os.path.join(self.progress_root, f"{identity_key}_{subject_key}.json"), os.path.join(self.progress_root, f"{identity_key}.json"), os.path.join(self.progress_root, identity_key, "progress.json"), os.path.join(self.progress_root, identity_key, "badge_progress.json"), ] # ========================================================= # خرائط الأوسمة # ========================================================= def _badge_title_fallback(self, badge_id: str) -> str: titles = { "math_addition": "وسام الجمع", "math_subtraction": "وسام الطرح", "math_multiplication": "وسام الضرب", "math_division": "وسام القسمة", "math_fractions": "وسام الكسور", "math_decimals": "وسام الأعداد العشرية", "math_percentages": "وسام النسب المئوية", "math_integers": "وسام الأعداد الصحيحة", "math_equations": "وسام المعادلات", "math_geometry_shapes": "وسام الأشكال الهندسية", "math_measurement_units": "وسام القياس والوحدات", "math_word_problems": "وسام المسائل اللفظية", } return titles.get(badge_id, badge_id.replace("_", " ")) def _skill_title_fallback(self, badge_id: str) -> str: title = self._badge_title_fallback(badge_id) return title.replace("وسام ", "").strip() def _math_roadmap(self, *args, **kwargs): del args, kwargs return None def _registry_subject_config(self, subject_key: str) -> Dict[str, Any]: path = Path(__file__).resolve().parent / "subjects_registry.json" data = self._read_json(str(path), default={}) for item in data.get("subjects", []) if isinstance(data, dict) else []: if not isinstance(item, dict): continue if self.normalize_subject(item.get("subject_id")) == subject_key: return item if subject_key == "arabic": return { "subject_id": "arabic", "enabled": True, "skills_file": "arabic_skills_global.json", "skills_repo_path": self.arabic_skills_repo_path, } return {} def _grade_token(self, value: Any) -> int: text = str(value or "").strip().upper() if text in {"KG", "K", "0", "روضة", "الروضة"}: return 0 return self._safe_int(text, 0) def _load_subject_skills_document( self, subject_key: str, configured_file: str, ) -> Dict[str, Any]: if subject_key == "arabic": if self._arabic_skills_cache: return self._arabic_skills_cache downloader = getattr(self.storage_service, "download_json_optional", None) if callable(downloader): try: value = downloader( repo_key=self.arabic_skills_repo_key, filename=self.arabic_skills_repo_path, default={}, ) if isinstance(value, dict) and value.get("skills"): self._arabic_skills_cache = value return value except Exception: pass path = Path(__file__).resolve().parent / configured_file return self._read_json(str(path), default={}) def _subject_skill_roadmap( self, subject_key: str, grade: int, ) -> List[Dict[str, Any]]: """ يبني الأوسمة من ملف المهارات الحقيقي للمادة. العربية مثلًا تستخدم AR-S01-001 ... بدل ستة أوسمة عامة. """ config = self._registry_subject_config(subject_key) skills_file = str(config.get("skills_file") or "").strip() if not skills_file: return [] data = self._load_subject_skills_document( subject_key, skills_file ) if not isinstance(data, dict): return [] stages = {} for stage in data.get("stages", []) or []: if not isinstance(stage, dict): continue stage_id = str(stage.get("stage_id") or "").strip() grades = stage.get("typical_grade_range") or [] stage_min = 0 stage_max = 12 if isinstance(grades, list) and grades: nums = [self._grade_token(x) for x in grades] stage_min = min(nums) stage_max = max(nums) stages[stage_id] = { "sequence": self._safe_int(stage.get("sequence"), 0), "title_ar": str(stage.get("title_ar") or stage_id), "grade_min": stage_min, "grade_max": stage_max, } learning_mode = str( os.getenv("DEFAULT_LEARNING_MODE", "gap_recovery") or "gap_recovery" ).strip().lower() raw_skills = [ skill for skill in (data.get("skills") or []) if isinstance(skill, dict) ] raw_skills.sort( key=lambda skill: ( self._safe_int( (stages.get(str(skill.get("stage_id") or "")) or {}).get("sequence"), 999, ), str(skill.get("skill_id") or ""), ) ) roadmap = [] previous = "" for skill in raw_skills: skill_id = str(skill.get("skill_id") or "").strip() if not skill_id: continue stage_id = str(skill.get("stage_id") or "").strip() stage = stages.get(stage_id) or {} stage_min = self._safe_int(stage.get("grade_min"), 0) stage_max = self._safe_int(stage.get("grade_max"), 12) # في gap_recovery: الطالب يرى الأساسيات حتى المستوى المناسب له، # ولا نقفز إلى مراحل أعلى يبدأ نطاقها بعد صفه. if learning_mode == "gap_recovery": include = stage_min <= grade else: include = stage_min <= grade <= stage_max if not include: continue title_ar = str( skill.get("name_ar") or skill.get("skill_title_ar") or skill_id ).strip() roadmap.append({ "order": len(roadmap) + 1, "badge_id": skill_id, "title": f"مهارة: {title_ar}", "skill_id": skill_id, "skill_title": title_ar, "global_level": stage_id or subject_key, "stage_id": stage_id, "stage_title_ar": str(stage.get("title_ar") or ""), "cluster": str(skill.get("cluster") or ""), "description": title_ar, "objective": title_ar, "prerequisites": list(skill.get("prerequisites") or ([previous] if previous else [])), "expected_grade_min": stage_min, "expected_grade_max": stage_max, "recommended_for_grade": stage_min <= grade <= stage_max, "mastery_required": 100.0, "mastery_scope": "exact_skill", "learning_mode": learning_mode, "assessment_blueprint": deepcopy( skill.get("assessment_blueprint") or {} ), }) previous = skill_id return roadmap def _generic_roadmap(self, subject_key: str, grade: int) -> List[Dict[str, Any]]: definitions = self.GENERIC_ROADMAPS.get(subject_key) if not definitions: definitions = [ (f"{subject_key}_foundation", "وسام الأساس", "المهارات الأساسية"), (f"{subject_key}_understanding", "وسام الفهم", "الفهم والتطبيق"), (f"{subject_key}_mastery", "وسام الإتقان", "الإتقان"), ] roadmap = [] previous = "" for index, definition in enumerate(definitions): badge_id, title, skill_title = definition roadmap.append( { "order": index + 1, "badge_id": badge_id, "title": title, "skill_id": badge_id, "skill_title": skill_title, "global_level": f"{subject_key}_l{index + 1}", "description": skill_title, "objective": skill_title, "prerequisites": [previous] if previous else [], "expected_grade_min": 1, "expected_grade_max": 12, "recommended_for_grade": True, "mastery_required": self.mastery_required, } ) previous = badge_id return roadmap def base_roadmap(self, subject: Any = "math", grade: Any = 1) -> List[Dict[str, Any]]: subject_key = self.normalize_subject(subject) grade_number = self.normalize_grade(grade) if subject_key == "math": return self._math_roadmap(grade_number) exact = self._subject_skill_roadmap(subject_key, grade_number) if exact: return exact return self._generic_roadmap(subject_key, grade_number) # ========================================================= # سجل التقدم # ========================================================= def _empty_badge_record(self, badge: Dict[str, Any]) -> Dict[str, Any]: return { "badge_id": badge.get("badge_id", ""), "title": badge.get("title", ""), "skill_id": badge.get("skill_id", badge.get("badge_id", "")), "skill_title": badge.get("skill_title", ""), "badge_group_id": badge.get("badge_group_id", ""), "badge_group_ids": list(badge.get("badge_group_ids") or []), "mastery_scope": badge.get("mastery_scope", "exact_skill"), "learning_mode": badge.get("learning_mode", os.getenv("DEFAULT_LEARNING_MODE", "gap_recovery")), "global_level": badge.get("global_level", ""), "status": self.STATUS_LOCKED, "earned": False, "unlocked": False, "attempts_count": 0, "best_percentage": 0.0, "last_percentage": 0.0, "correct_total": 0, "wrong_total": 0, "unknown_total": 0, "help_requested_total": 0, "repeated_error_total": 0, "error_counts": {}, "remedial_metrics": { "questions_total": 0, "first_attempt_correct": 0, "hints_total": 0, "response_time_total_seconds": 0.0, "independent_success_count": 0, "closure_success_count": 0, "transfer_success_count": 0, "spaced_recheck_success_count": 0, "same_error_recurrence_count": 0, "last_strategy_id": "", "last_intervention_level": "", "last_updated_at": "", }, "attempts": [], "last_attempt": {}, "earned_at": "", "remedial_required": False, "remedial_completed": False, "remedial_completed_at": "", "mastery_required": self.mastery_required, "updated_at": self.now(), } def default_progress( self, identity: Any, subject: Any = "math", grade: Any = 1, ) -> Dict[str, Any]: identity_key = self.normalize_identity(identity) subject_key = self.normalize_subject(subject) grade_number = self.normalize_grade(grade) roadmap = self.base_roadmap(subject_key, grade_number) badge_records: Dict[str, Dict[str, Any]] = {} for index, badge in enumerate(roadmap): record = self._empty_badge_record(badge) if index == 0: record["status"] = self.STATUS_AVAILABLE record["unlocked"] = True badge_records[badge["badge_id"]] = record return { "success": True, "ready": True, "version": self.version, "schema_version": self.schema_version, "identity": identity_key, "subject": subject_key, "subject_name": self.subject_name(subject_key), "grade": grade_number, "current_badge_id": roadmap[0]["badge_id"] if roadmap else "", "current_index": 0, "all_completed": False, "mastery_required": self.mastery_required, "badge_order": [item["badge_id"] for item in roadmap], "badges": badge_records, "attempts_total": 0, "earned_count": 0, "last_result": {}, "last_remedial_badge_id": "", "game_assignments": [], "interactive_game_results": [], "diagnostic_history": [], "created_at": self.now(), "updated_at": self.now(), } def _extract_legacy_subject_data( self, legacy: Dict[str, Any], subject_key: str, ) -> Dict[str, Any]: if not isinstance(legacy, dict): return {} if isinstance(legacy.get("subjects"), dict): candidate = legacy["subjects"].get(subject_key) if isinstance(candidate, dict): return candidate if legacy.get("subject") == subject_key or "badges" in legacy: return legacy return {} def _migrate_progress( self, data: Any, identity: Any, subject: Any, grade: Any, ) -> Dict[str, Any]: base = self.default_progress(identity, subject, grade) if not isinstance(data, dict): return base subject_key = self.normalize_subject(subject) source = self._extract_legacy_subject_data(data, subject_key) or data source_badges = source.get("badges") if isinstance(source.get("badges"), dict) else {} earned_ids = ( source.get("earned_badges") or source.get("completed_badges") or source.get("mastered_badges") or [] ) if isinstance(earned_ids, dict): earned_ids = list(earned_ids.keys()) earned_set = {self.normalize_badge_id(item) for item in earned_ids if item} for badge_id, target in base["badges"].items(): old = source_badges.get(badge_id) if not isinstance(old, dict): old = {} for key in ( "attempts_count", "best_percentage", "last_percentage", "correct_total", "wrong_total", "unknown_total", "help_requested_total", "repeated_error_total", "error_counts", "attempts", "last_attempt", "earned_at", "remedial_required", "remedial_completed", "remedial_completed_at", ): if key in old: target[key] = deepcopy(old[key]) old_status = str(old.get("status", "")).strip().lower() old_earned = self._safe_bool(old.get("earned"), False) or self._safe_bool(old.get("mastered"), False) if "attempts_count" not in old and "attempts" in old and not isinstance(old.get("attempts"), list): target["attempts_count"] = self._safe_int(old.get("attempts"), 0) if badge_id in earned_set or old_earned or old_status in {"earned", "completed", "mastered"}: target["earned"] = True target["unlocked"] = True target["status"] = self.STATUS_EARNED target["best_percentage"] = max( self._safe_float(target.get("best_percentage"), 0.0), self.mastery_required, ) for key in ( "attempts_total", "last_result", "last_remedial_badge_id", "game_assignments", "interactive_game_results", "diagnostic_history", "created_at", ): if key in source: base[key] = deepcopy(source[key]) requested_current = self.normalize_badge_id( source.get("current_badge_id") or source.get("current_badge") or "" ) if requested_current in base["badges"] and not base["badges"][requested_current].get("earned"): base["current_badge_id"] = requested_current self._recalculate_unlocks(base) base["updated_at"] = self.now() base["version"] = self.version base["schema_version"] = self.schema_version return base def load_progress( self, identity: Any, subject: Any = "math", grade: Any = 1, create: bool = True, ) -> Dict[str, Any]: path = self.progress_path(identity, subject) with self._lock: data = self._read_json(path, default=None) if not isinstance(data, dict) or not data: data = None for legacy_path in self.legacy_paths(identity, subject): legacy = self._read_json(legacy_path, default=None) if isinstance(legacy, dict) and legacy: data = legacy break if data is None and not create: return {} progress = self._migrate_progress(data, identity, subject, grade) if create: self.save_progress(progress, sync_student=False) return progress def get_progress(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.load_progress(*args, **kwargs) def _get_academic_manager(self, *args, **kwargs): del args, kwargs return self.academic_service def _sync_to_academic_manager( self, progress: Dict[str, Any], ) -> bool: manager = self._get_academic_manager() if manager is None or not hasattr( manager, "save_badge_progress", ): return False try: result = manager.save_badge_progress( identity=progress.get("identity", "student"), subject=progress.get("subject", "math"), progress=progress, sync_cloud=True, ) return bool( isinstance(result, dict) and result.get("success") ) except Exception: return False def _get_student_manager(self, *args, **kwargs): del args, kwargs return self.student_service def _student_summary(self, progress: Dict[str, Any]) -> Dict[str, Any]: current_id = progress.get("current_badge_id", "") current = progress.get("badges", {}).get(current_id, {}) return { "version": self.version, "subject": progress.get("subject", ""), "grade": progress.get("grade", 1), "current_badge_id": current_id, "current_badge_status": current.get("status", ""), "earned_count": progress.get("earned_count", 0), "total_badges": len(progress.get("badge_order", [])), "all_completed": progress.get("all_completed", False), "last_result": progress.get("last_result", {}), "game_assignments_count": len(progress.get("game_assignments") or []), "interactive_game_results_count": len(progress.get("interactive_game_results") or []), "updated_at": progress.get("updated_at", self.now()), } def _sync_to_student_manager(self, progress: Dict[str, Any]) -> bool: manager = self._get_student_manager() if manager is None or not hasattr(manager, "update_badge_learning"): return False try: identity = progress.get("identity", "student") student = manager.get_student(identity) if hasattr(manager, "get_student") else None badge_learning = ( deepcopy(student.get("badge_learning")) if isinstance(student, dict) and isinstance(student.get("badge_learning"), dict) else {} ) subjects = badge_learning.get("subjects") if not isinstance(subjects, dict): subjects = {} subjects[progress.get("subject", "math")] = self._student_summary(progress) manager.update_badge_learning( identity=identity, subjects=subjects, current_subject=progress.get("subject", "math"), last_badge_progress=self._student_summary(progress), ) return True except Exception: return False def save_progress( self, progress: Dict[str, Any], sync_student: bool = True, ) -> Dict[str, Any]: if not isinstance(progress, dict): return { "success": False, "message": "بيانات تقدم الأوسمة غير صالحة.", "version": self.version, } progress["version"] = self.version progress["schema_version"] = self.schema_version progress["updated_at"] = self.now() self._recalculate_unlocks(progress) path = self.progress_path(progress.get("identity"), progress.get("subject")) with self._lock: self._write_json(path, progress) academic_synced = ( self._sync_to_academic_manager(progress) if sync_student else False ) compatibility_student_synced = ( self._sync_to_student_manager(progress) if sync_student else False ) return { "success": True, "message": "تم حفظ تقدم الأوسمة.", "path": path, "academic_synced": academic_synced, "student_synced": compatibility_student_synced, "academic_source_of_truth": True, "progress": progress, "version": self.version, } def record_remedial_question_metric( self, identity: Any, subject: Any, grade: Any, badge_id: Any, metric: Dict[str, Any], save: bool = True, ) -> Dict[str, Any]: progress = self.load_progress(identity, subject, grade) badge_key = self.normalize_badge_id(badge_id) record = (progress.get("badges") or {}).get(badge_key) if not isinstance(record, dict): return { "success": False, "message": "المهارة/الوسام غير موجود في سجل التقدم.", "version": self.version, } metrics = record.get("remedial_metrics") if not isinstance(metrics, dict): metrics = {} record["remedial_metrics"] = metrics metrics["questions_total"] = self._safe_int( metrics.get("questions_total"), 0 ) + 1 if self._safe_bool(metric.get("first_attempt_correct"), False): metrics["first_attempt_correct"] = self._safe_int( metrics.get("first_attempt_correct"), 0 ) + 1 metrics["hints_total"] = self._safe_int( metrics.get("hints_total"), 0 ) + self._safe_int(metric.get("hint_count"), 0) metrics["response_time_total_seconds"] = round( self._safe_float( metrics.get("response_time_total_seconds"), 0.0 ) + self._safe_float(metric.get("response_time"), 0.0), 3, ) role = str(metric.get("question_role") or "") is_correct = self._safe_bool(metric.get("is_correct"), False) independent = self._safe_bool( metric.get("independent_success"), False ) if independent: metrics["independent_success_count"] = self._safe_int( metrics.get("independent_success_count"), 0 ) + 1 if is_correct and role == "closure_check": metrics["closure_success_count"] = self._safe_int( metrics.get("closure_success_count"), 0 ) + 1 if is_correct and role == "transfer_check": metrics["transfer_success_count"] = self._safe_int( metrics.get("transfer_success_count"), 0 ) + 1 if is_correct and role == "spaced_recheck": metrics["spaced_recheck_success_count"] = self._safe_int( metrics.get("spaced_recheck_success_count"), 0 ) + 1 if self._safe_bool(metric.get("same_error_recurred"), False): metrics["same_error_recurrence_count"] = self._safe_int( metrics.get("same_error_recurrence_count"), 0 ) + 1 if metric.get("strategy_id"): metrics["last_strategy_id"] = str(metric.get("strategy_id")) if metric.get("intervention_level"): metrics["last_intervention_level"] = str( metric.get("intervention_level") ) metrics["last_updated_at"] = self.now() record["updated_at"] = self.now() progress["updated_at"] = self.now() if save: return self.save_progress(progress) return { "success": True, "progress": progress, "metrics": metrics, "version": self.version, } # ========================================================= # فتح الأوسمة والحالة الحالية # ========================================================= def _recalculate_unlocks(self, progress: Dict[str, Any]) -> None: order = list(progress.get("badge_order") or []) badges = progress.get("badges") if not isinstance(badges, dict): badges = {} progress["badges"] = badges first_unearned = "" earned_count = 0 prior_earned = True for index, badge_id in enumerate(order): record = badges.get(badge_id) if not isinstance(record, dict): continue earned = self._safe_bool(record.get("earned"), False) if earned: earned_count += 1 record["status"] = self.STATUS_EARNED record["unlocked"] = True prior_earned = prior_earned and True continue if not first_unearned: first_unearned = badge_id unlocked = index == 0 or prior_earned record["unlocked"] = unlocked if record.get("remedial_required"): record["status"] = self.STATUS_REMEDIAL_REQUIRED elif unlocked: if self._safe_int(record.get("attempts_count"), 0) > 0: record["status"] = self.STATUS_IN_PROGRESS else: record["status"] = self.STATUS_AVAILABLE else: record["status"] = self.STATUS_LOCKED prior_earned = False progress["earned_count"] = earned_count progress["all_completed"] = bool(order) and earned_count == len(order) if progress["all_completed"]: progress["current_badge_id"] = "" progress["current_index"] = len(order) else: learning_mode = str( os.getenv("DEFAULT_LEARNING_MODE", "gap_recovery") or "gap_recovery" ).strip().lower() # في مسار علاج الفاقد لا نحافظ على مؤشر قديم يتجاوز مهارات # تأسيسية أضيفت لاحقًا، مثل مهارات الروضة. نبدأ من أول مهارة # غير متقنة، بينما يبقى الصف مجرد حد أعلى للمهارات المتاحة. if learning_mode == "gap_recovery": current = first_unearned else: current = progress.get("current_badge_id") if current not in badges or badges.get(current, {}).get("earned"): current = first_unearned progress["current_badge_id"] = current or first_unearned try: progress["current_index"] = order.index(progress["current_badge_id"]) except (ValueError, TypeError): progress["current_index"] = 0 def _decorated_roadmap(self, progress: Dict[str, Any]) -> List[Dict[str, Any]]: base = self.base_roadmap(progress.get("subject"), progress.get("grade")) records = progress.get("badges") if isinstance(progress.get("badges"), dict) else {} result = [] for item in base: badge_id = item.get("badge_id", "") record = records.get(badge_id, {}) decorated = deepcopy(item) decorated.update( { "status": record.get("status", self.STATUS_LOCKED), "earned": self._safe_bool(record.get("earned"), False), "unlocked": self._safe_bool(record.get("unlocked"), False), "is_current": badge_id == progress.get("current_badge_id"), "attempts_count": self._safe_int(record.get("attempts_count"), 0), "best_percentage": self._safe_float(record.get("best_percentage"), 0.0), "last_percentage": self._safe_float(record.get("last_percentage"), 0.0), "wrong_total": self._safe_int(record.get("wrong_total"), 0), "unknown_total": self._safe_int(record.get("unknown_total"), 0), "help_requested_total": self._safe_int(record.get("help_requested_total"), 0), "remedial_required": self._safe_bool(record.get("remedial_required"), False), "remedial_completed": self._safe_bool(record.get("remedial_completed"), False), "earned_at": record.get("earned_at", ""), } ) result.append(decorated) return result def _current_stage_roadmap( self, roadmap: List[Dict[str, Any]], progress: Dict[str, Any], ) -> Tuple[List[Dict[str, Any]], str, str]: """ يعرض مرحلة المهارة الحالية فقط، لا جميع مهارات الصف/المادة. المرحلة تؤخذ من stage_id، وللرياضيات من global_level عند عدم وجود stage_id. """ if not roadmap: return [], "", "" current_id = str(progress.get("current_badge_id") or "").strip() current_item = next( (item for item in roadmap if str(item.get("badge_id") or "") == current_id), None, ) if current_item is None: current_item = next( (item for item in roadmap if not self._safe_bool(item.get("earned"), False)), roadmap[0], ) stage_id = str( current_item.get("stage_id") or current_item.get("global_level") or "" ).strip() stage_title = str( current_item.get("stage_title_ar") or stage_id or "المرحلة الحالية" ).strip() if not stage_id: return roadmap, "", stage_title stage_items = [ item for item in roadmap if str(item.get("stage_id") or item.get("global_level") or "").strip() == stage_id ] return stage_items or roadmap, stage_id, stage_title def get_roadmap( self, identity: Any = "student", subject: Any = "math", grade: Any = 1, **kwargs: Any, ) -> Dict[str, Any]: progress = self.load_progress(identity, subject, grade) full_roadmap = self._decorated_roadmap(progress) roadmap, stage_id, stage_title = self._current_stage_roadmap( full_roadmap, progress, ) total = len(roadmap) earned = sum( 1 for item in roadmap if isinstance(item, dict) and self._safe_bool(item.get("earned"), False) ) remaining = max(0, total - earned) percentage = round((earned / total) * 100, 2) if total else 0.0 return { "success": True, "ready": True, "identity": progress.get("identity"), "subject": progress.get("subject"), "subject_name": progress.get("subject_name"), "grade": progress.get("grade"), "roadmap": roadmap, "badges": roadmap, "current_badge_id": progress.get("current_badge_id", ""), "current_stage_id": stage_id, "current_stage_title": stage_title, "earned_count": earned, "remaining_count": remaining, "total_badges": total, "total_subject_badges": len(full_roadmap), "progress_percentage": percentage, "all_completed": progress.get("all_completed", False), "mastery_required": self.mastery_required, "progress": progress, "version": self.version, } def get_current_badge( self, identity: Any = "student", subject: Any = "math", grade: Any = 1, **kwargs: Any, ) -> Dict[str, Any]: roadmap_result = self.get_roadmap(identity, subject, grade) current_id = str(roadmap_result.get("current_badge_id") or "").strip() roadmap = [ item for item in (roadmap_result.get("roadmap") or []) if isinstance(item, dict) ] current = None for item in roadmap: if str(item.get("badge_id") or "").strip() == current_id: current = item break # Self-heal: بعد تغيير ملفات المهارات قد يبقى current_badge_id قديمًا. # نختار أول مهارة غير مكتسبة بدل إيقاف جلسة الطالب. if current is None and roadmap: current = next( ( item for item in roadmap if not self._safe_bool(item.get("earned"), False) ), roadmap[0], ) current_id = str(current.get("badge_id") or "").strip() try: progress = roadmap_result.get("progress") if isinstance(progress, dict) and current_id: progress["current_badge_id"] = current_id self._recalculate_unlocks(progress) self.save_progress(progress, sync_student=False) roadmap_result["current_badge_id"] = current_id roadmap_result["progress"] = progress except Exception: pass if current is None: all_completed = bool(roadmap_result.get("all_completed")) return { "success": all_completed, "ready": False, "completed": all_completed, "completed_all": all_completed, "error": "" if all_completed else "roadmap_empty", "message": ( "أتم الطالب جميع الأوسمة المتاحة." if all_completed else "لا توجد مهارات في خريطة المادة الحالية." ), "badge": None, "current_badge": None, "roadmap": roadmap_result, "version": self.version, } return { "success": True, "ready": True, "message": "المهارة الحالية جاهزة.", "badge": current, "current_badge": current, "badge_id": current_id, "blocked_by_remedial": current.get("status") == self.STATUS_REMEDIAL_REQUIRED, "roadmap": roadmap_result, "progress": roadmap_result.get("progress", {}), "version": self.version, } def can_start_badge( self, identity: Any, subject: Any, grade: Any, badge_id: Any, ) -> Dict[str, Any]: current = self.get_current_badge(identity, subject, grade) badge_id = self.normalize_badge_id(badge_id) if not current.get("success"): return current badge = current.get("badge") or {} if badge.get("badge_id") != badge_id: return { "success": False, "allowed": False, "message": "هذه المهارة ما زالت مقفلة.", "current_badge": badge, "version": self.version, } return { "success": True, "allowed": True, "message": "يمكن بدء تحدي المهارة.", "badge": badge, "version": self.version, } # ========================================================= # تسجيل نتيجة لعبة الوسام # ========================================================= def _result_state(self, value: Any) -> Dict[str, Any]: if not isinstance(value, dict): return {} state = value.get("state") or value.get("game_state") if isinstance(state, dict): return state return value def _count_answer_types(self, answers: Iterable[Any]) -> Tuple[int, int, int]: correct = wrong = unknown = 0 for answer in answers or []: if not isinstance(answer, dict): continue answer_type = str( answer.get("answer_type") or answer.get("diagnostic_status") or "" ).strip().lower() if answer_type == "correct" or answer.get("is_correct") is True: correct += 1 elif answer_type == "unknown": unknown += 1 else: wrong += 1 return correct, wrong, unknown def _extract_game_metrics(self, game_result: Dict[str, Any]) -> Dict[str, Any]: state = self._result_state(game_result) answers = state.get("answers") if isinstance(state.get("answers"), list) else [] counted_correct, counted_wrong, counted_unknown = self._count_answer_types(answers) total = self._safe_int( game_result.get("total_questions", state.get("total_questions", len(state.get("questions") or answers))), len(state.get("questions") or answers), ) total = max(1, total) correct = self._safe_int( game_result.get("correct_count", state.get("correct_count", counted_correct)), counted_correct, ) wrong = self._safe_int( game_result.get("wrong_count", state.get("wrong_count", counted_wrong)), counted_wrong, ) unknown = self._safe_int( game_result.get("unknown_count", state.get("unknown_count", counted_unknown)), counted_unknown, ) if correct + wrong + unknown < len(answers): wrong = max(wrong, len(answers) - correct - unknown) percentage = self._safe_float( game_result.get("score_percentage", state.get("score_percentage", 0.0)), 0.0, ) if percentage <= 0 and correct > 0: percentage = round((correct / total) * 100, 2) error_counts = game_result.get("error_counts") or state.get("error_counts") or {} if not isinstance(error_counts, dict): error_counts = {} repeated_error_total = sum( max(0, self._safe_int(count, 0) - 1) for count in error_counts.values() ) repeated_error = any(self._safe_int(count, 0) >= 2 for count in error_counts.values()) help_requests = state.get("help_requests") if not isinstance(help_requests, list): help_requests = [] help_requested_count = self._safe_int( state.get("help_requested_count", len(help_requests)), len(help_requests), ) # v25: # هذا المدير لا يستنتج الإتقان من النسبة أو عدد الإجابات. # القرار يأتي من StrategyManager عبر SessionManager. mastered = None needs_remedial = None return { "game_id": state.get("game_id") or game_result.get("game_id") or uuid.uuid4().hex, "badge_id": self.normalize_badge_id( state.get("badge_id") or game_result.get("badge_id") or "" ), "skill_id": state.get("skill_id") or game_result.get("skill_id") or "", "subject": state.get("subject_key") or state.get("subject") or game_result.get("subject") or "math", "grade": state.get("grade") or game_result.get("grade") or 1, "completed": self._safe_bool( game_result.get( "completed", game_result.get( "is_completed", state.get("status") == "completed", ), ), state.get("status") == "completed", ), "terminated_early": self._safe_bool( game_result.get( "terminated_early", state.get("terminated_early", False), ), False, ), "termination_reason": str( game_result.get("termination_reason") or state.get("termination_reason") or "" ), "termination_error_type": str( game_result.get("termination_error_type") or state.get("termination_error_type") or "" ), "stopped_on_first_error": self._safe_bool( game_result.get( "stopped_on_first_error", state.get("stopped_on_first_error", False), ), False, ), "attempted_questions": self._safe_int( game_result.get( "attempted_questions", state.get( "attempted_questions", len(answers), ), ), len(answers), ), "learning_mode": str( state.get("learning_mode") or game_result.get("learning_mode") or "gap_recovery" ), "total_questions": total, "correct_count": correct, "wrong_count": wrong, "unknown_count": unknown, "percentage": round(percentage, 2), "mastered": mastered, "needs_remedial": needs_remedial, "decision_source": "strategy_manager_required", "evidence_only": True, "error_counts": {str(key): self._safe_int(value, 0) for key, value in error_counts.items()}, "repeated_error": repeated_error, "repeated_error_total": repeated_error_total, "help_requested_count": help_requested_count, "help_requests": deepcopy(help_requests), "last_help_request": deepcopy(state.get("last_help_request") or {}), "answers": deepcopy(answers), "game_version": state.get("version") or game_result.get("version") or "", "created_at": self.now(), } def make_gap_report( self, identity: Any, metrics: Dict[str, Any], badge_record: Dict[str, Any], subject: Any, grade: Any, ) -> Dict[str, Any]: error_counts = metrics.get("error_counts") if isinstance(metrics.get("error_counts"), dict) else {} repeated = [ key for key, value in error_counts.items() if self._safe_int(value, 0) >= 2 ] unknown_count = self._safe_int(metrics.get("unknown_count"), 0) wrong_count = self._safe_int(metrics.get("wrong_count"), 0) causes = [] if unknown_count: causes.append("unknown_answers") if wrong_count: causes.append("wrong_answers") if repeated: causes.append("repeated_errors") return { "report_id": f"gap_{uuid.uuid4().hex}", "identity": self.normalize_identity(identity), "subject": self.normalize_subject(subject), "grade": self.normalize_grade(grade), "badge_id": metrics.get("badge_id") or badge_record.get("badge_id", ""), "badge_title": badge_record.get("title", ""), "skill_id": metrics.get("skill_id") or badge_record.get("skill_id", ""), "skill_title": badge_record.get("skill_title", ""), "global_level": badge_record.get("global_level", ""), "percentage": metrics.get("percentage", 0), "correct_count": metrics.get("correct_count", 0), "wrong_count": wrong_count, "unknown_count": unknown_count, "total_questions": metrics.get("total_questions", 0), "attempted_questions": metrics.get( "attempted_questions", 0 ), "terminated_early": metrics.get( "terminated_early", False ), "termination_reason": metrics.get( "termination_reason", "" ), "termination_error_type": metrics.get( "termination_error_type", "" ), "stopped_on_first_error": metrics.get( "stopped_on_first_error", False ), "learning_mode": metrics.get( "learning_mode", "gap_recovery" ), "wrong_items": [ { "question_id": item.get("question_id", ""), "prompt": item.get("prompt", ""), "selected": item.get("selected", ""), "correct": item.get("correct", ""), "answer_type": item.get("answer_type", ""), "error_type": item.get("error_type", ""), "skill_id": item.get("skill_id", ""), "difficulty": item.get("difficulty", 1), "explanation": item.get("explanation", ""), } for item in metrics.get("answers", []) if isinstance(item, dict) and item.get("answer_type") in {"wrong", "unknown"} ][:3], "error_counts": deepcopy(error_counts), "repeated_errors": repeated, "repeated_error": bool(repeated), "help_requested_count": metrics.get("help_requested_count", 0), "diagnostic_profile": deepcopy(metrics.get("diagnostic_profile") or {}), "causes": causes, "recommended_action": "remedial_lesson_then_retry_same_badge", "return_to_same_badge": True, "created_at": self.now(), "progress_version": self.version, } def _diagnostic_evidence_events(self, answers: Iterable[Any]) -> List[Dict[str, Any]]: events: List[Dict[str, Any]] = [] seen: set[str] = set() for answer in answers or []: if not isinstance(answer, dict): continue question = answer.get("question_data") if isinstance(answer.get("question_data"), dict) else {} signature = question.get("runtime_evidence_signature") or question.get("evidence_identity_signature") if isinstance(signature, dict): signature = signature.get("runtime_signature") or signature.get("static_condition_family_signature") or "" signature = str(signature or answer.get("question_id") or "").strip() slot = str(question.get("condition_slot_id") or (question.get("stimulus_plan") or {}).get("condition_slot_id") or "").strip() event = { "question_id": str(answer.get("question_id") or ""), "template_id": str(question.get("template_id") or ""), "skill_id": str(answer.get("skill_id") or question.get("skill_id") or ""), "is_correct": answer.get("is_correct") is True, "answer_type": str(answer.get("answer_type") or ""), "error_type": str(answer.get("error_type") or ""), "condition_slot_id": slot, "diagnostic_condition_family": str(question.get("diagnostic_condition_family") or ""), "evidence_identity_signature": signature, "independent_evidence": bool(signature and signature not in seen), "primary_target_attribute": str(question.get("primary_target_attribute") or ""), "required_attributes": deepcopy(question.get("required_attributes") or []), "diagnostic_attributes": deepcopy(question.get("diagnostic_attributes") or []), "q_vector": deepcopy(question.get("q_vector") or {}), "error_hypothesis_ids": deepcopy(question.get("error_hypothesis_ids") or []), "stimulus_identity_group": str(question.get("stimulus_identity_group") or ""), "created_at": str(answer.get("created_at") or self.now()), } if signature: seen.add(signature) events.append(event) return events def record_game_result( self, identity: Any = "student", subject: Any = "math", grade: Any = 1, badge_id: Any = "", game_result: Any = None, result: Any = None, state: Any = None, **kwargs: Any, ) -> Dict[str, Any]: raw = game_result if isinstance(game_result, dict) else result if not isinstance(raw, dict): raw = state if isinstance(state, dict) else kwargs metrics = self._extract_game_metrics(raw if isinstance(raw, dict) else {}) subject_key = self.normalize_subject(metrics.get("subject") or subject) grade_number = self.normalize_grade(metrics.get("grade") or grade) badge_key = self.normalize_badge_id( badge_id or metrics.get("badge_id") or "" ) progress = self.load_progress(identity, subject_key, grade_number) if badge_key not in progress.get("badges", {}): badge_key = progress.get("current_badge_id", badge_key) badge_record = progress.get("badges", {}).get(badge_key) if not isinstance(badge_record, dict): return { "success": False, "message": "الوسام غير موجود في خريطة التقدم.", "badge_id": badge_key, "version": self.version, } metrics["badge_id"] = badge_key metrics["skill_id"] = metrics.get("skill_id") or badge_record.get("skill_id") or badge_key diagnostic_profile = raw.get("diagnostic_profile") if isinstance(raw, dict) else {} if not isinstance(diagnostic_profile, dict): diagnostic_profile = {} metrics["diagnostic_profile"] = deepcopy(diagnostic_profile) metrics["evidence_events"] = self._diagnostic_evidence_events( metrics.get("answers") or [] ) independent_signatures = { str(x.get("evidence_identity_signature")) for x in metrics["evidence_events"] if isinstance(x, dict) and x.get("independent_evidence") and x.get("evidence_identity_signature") } condition_slots = { str(x.get("condition_slot_id")) for x in metrics["evidence_events"] if isinstance(x, dict) and x.get("condition_slot_id") } metrics["independent_evidence_count"] = len(independent_signatures) metrics["distinct_condition_slots"] = sorted(condition_slots) metrics["duplicate_evidence_count"] = max(0, len(metrics["evidence_events"]) - len(independent_signatures)) metrics["attempt_id"] = f"attempt_{uuid.uuid4().hex}" metrics["attempt_number"] = self._safe_int(badge_record.get("attempts_count"), 0) + 1 metrics["mastery_required"] = None metrics["mastery_rule_owner"] = "StrategyManager" metrics["percentage_is_decision_rule"] = False badge_record["attempts_count"] = metrics["attempt_number"] badge_record["last_percentage"] = metrics["percentage"] badge_record["best_percentage"] = max( self._safe_float(badge_record.get("best_percentage"), 0.0), metrics["percentage"], ) badge_record["correct_total"] = self._safe_int(badge_record.get("correct_total"), 0) + metrics["correct_count"] badge_record["wrong_total"] = self._safe_int(badge_record.get("wrong_total"), 0) + metrics["wrong_count"] badge_record["unknown_total"] = self._safe_int(badge_record.get("unknown_total"), 0) + metrics["unknown_count"] badge_record["help_requested_total"] = self._safe_int(badge_record.get("help_requested_total"), 0) + metrics["help_requested_count"] badge_record["repeated_error_total"] = self._safe_int(badge_record.get("repeated_error_total"), 0) + metrics["repeated_error_total"] accumulated_errors = badge_record.get("error_counts") if not isinstance(accumulated_errors, dict): accumulated_errors = {} for key, count in metrics["error_counts"].items(): accumulated_errors[key] = self._safe_int(accumulated_errors.get(key), 0) + self._safe_int(count, 0) badge_record["error_counts"] = accumulated_errors attempts = badge_record.get("attempts") if not isinstance(attempts, list): attempts = [] attempts.append(deepcopy(metrics)) badge_record["attempts"] = attempts[-50:] badge_record["last_attempt"] = deepcopy(metrics) badge_record["updated_at"] = self.now() progress["attempts_total"] = self._safe_int(progress.get("attempts_total"), 0) + 1 progress["last_result"] = deepcopy(metrics) diagnostic_history = progress.setdefault("diagnostic_history", []) diagnostic_history.append({ "badge_id": badge_key, "skill_id": metrics.get("skill_id", badge_key), "percentage": metrics.get("percentage", 0), "correct_count": metrics.get("correct_count", 0), "wrong_count": metrics.get("wrong_count", 0), "unknown_count": metrics.get("unknown_count", 0), "diagnostic_profile": deepcopy(metrics.get("diagnostic_profile") or {}), "created_at": self.now(), }) progress["diagnostic_history"] = diagnostic_history[-100:] evidence_history = progress.setdefault("evidence_history", []) evidence_history.append({ "badge_id": badge_key, "skill_id": metrics.get("skill_id", badge_key), "percentage_metric_only": metrics.get("percentage", 0), "correct_count": metrics.get("correct_count", 0), "wrong_count": metrics.get("wrong_count", 0), "unknown_count": metrics.get("unknown_count", 0), "error_counts": deepcopy(metrics.get("error_counts") or {}), "help_requested_count": metrics.get("help_requested_count", 0), "evidence_events": deepcopy(metrics.get("evidence_events") or []), "independent_evidence_count": metrics.get("independent_evidence_count", 0), "distinct_condition_slots": deepcopy(metrics.get("distinct_condition_slots") or []), "duplicate_evidence_count": metrics.get("duplicate_evidence_count", 0), "strategy_decision": {}, "created_at": self.now(), "record_type": "raw_attempt_evidence", }) progress["evidence_history"] = evidence_history[-200:] last_gap_report = {} # القرار الوحيد المسموح بتطبيقه هنا هو قرار صريح صادر # من StrategyManager (ويمرره SessionManager). strategy_decision = {} if isinstance(raw, dict): for key in ( "strategy_decision", "diagnostic_strategy_decision", "mastery_decision", ): value = raw.get(key) if isinstance(value, dict): strategy_decision = deepcopy(value) break if not strategy_decision: value = kwargs.get("strategy_decision") if isinstance(value, dict): strategy_decision = deepcopy(value) decision = str( strategy_decision.get("decision") or strategy_decision.get("next_action") or "" ).strip().lower() mastery_decisions = { "mastery_confirmed", "initial_mastery", "retained_mastery", } gap_decisions = { "gap_confirmed", "misconception_confirmed", "foundation_gap_confirmed", "start_remediation", } metrics["strategy_decision"] = deepcopy(strategy_decision) metrics["strategy_decision_name"] = decision metrics["decision_source"] = ( "strategy_manager" if strategy_decision else "pending_strategy_decision" ) if progress.get("evidence_history"): progress["evidence_history"][-1]["strategy_decision"] = deepcopy( strategy_decision ) progress["evidence_history"][-1]["decision_source"] = ( metrics["decision_source"] ) if decision in mastery_decisions: metrics["mastered"] = True metrics["needs_remedial"] = False elif decision in gap_decisions: metrics["mastered"] = False metrics["needs_remedial"] = True else: metrics["mastered"] = None metrics["needs_remedial"] = None if metrics["mastered"] is True: badge_record["earned"] = True badge_record["unlocked"] = True badge_record["status"] = self.STATUS_EARNED badge_record["earned_at"] = badge_record.get("earned_at") or self.now() badge_record["remedial_required"] = False badge_record["remedial_completed"] = False badge_record["remedial_completed_at"] = "" progress["last_remedial_badge_id"] = "" badge_record["last_gap_report"] = {} progress["last_gap_report"] = {} message = "أحسنت، حصلت على الوسام وتم فتح الوسام التالي." feedback_event = "badge_earned" sound_key = "badge_earned" elif metrics["needs_remedial"] is True: badge_record["earned"] = False badge_record["unlocked"] = True badge_record["status"] = self.STATUS_REMEDIAL_REQUIRED badge_record["remedial_required"] = True badge_record["remedial_completed"] = False badge_record["remedial_completed_at"] = "" progress["current_badge_id"] = badge_key progress["last_remedial_badge_id"] = badge_key last_gap_report = self.make_gap_report( identity=identity, metrics=metrics, badge_record=badge_record, subject=subject_key, grade=grade_number, ) badge_record["last_gap_report"] = deepcopy(last_gap_report) progress["last_gap_report"] = deepcopy(last_gap_report) message = "لم يكتمل إتقان المهارة. سيتم تجهيز حصة علاجية ثم تعود إلى الوسام نفسه." feedback_event = "remedial_required" sound_key = "stage_complete" else: # لا قرار = لا منح وسام ولا فتح علاج. # نحفظ الأدلة وننتظر StrategyManager. badge_record["unlocked"] = True if badge_record.get("earned"): badge_record["status"] = self.STATUS_EARNED else: badge_record["status"] = self.STATUS_IN_PROGRESS badge_record["remedial_required"] = False progress["current_badge_id"] = badge_key message = ( "تم تسجيل أدلة المحاولة. " "بانتظار قرار استراتيجية التشخيص." ) feedback_event = "evidence_recorded" sound_key = "" last_gap_report = {} self._recalculate_unlocks(progress) save_result = self.save_progress( progress, sync_student=True, ) academic_attempt_result = {} academic_manager = self._get_academic_manager() if academic_manager is not None and hasattr( academic_manager, "record_badge_attempt", ): try: academic_attempt_result = ( academic_manager.record_badge_attempt( identity=identity, attempt=metrics, gap_report=last_gap_report, sync_cloud=True, ) ) except Exception as exc: academic_attempt_result = { "success": False, "message": str(exc), } roadmap = self.get_roadmap(identity, subject_key, grade_number) current = self.get_current_badge(identity, subject_key, grade_number) return { "success": True, "ready": True, "message": message, "identity": progress.get("identity"), "subject": subject_key, "grade": grade_number, "badge_id": badge_key, "skill_id": metrics.get("skill_id"), "mastered": metrics["mastered"], "badge_earned": metrics["mastered"] is True, "needs_remedial": metrics["needs_remedial"], "required_before_badge_retry": metrics["needs_remedial"] is True, "return_to_same_badge_after_remedial": metrics["needs_remedial"] is True, "block_next_badge": metrics["mastered"] is not True, "strategy_decision": deepcopy( metrics.get("strategy_decision") or {} ), "decision_pending": metrics["mastered"] is None, "evidence_only_manager": True, "answer_summary": { "correct": metrics["correct_count"], "wrong": metrics["wrong_count"], "unknown": metrics["unknown_count"], "total": metrics["total_questions"], "percentage": metrics["percentage"], }, "error_counts": metrics["error_counts"], "repeated_error": metrics["repeated_error"], "help_requested_count": metrics["help_requested_count"], "attempt": metrics, "terminated_early": metrics.get( "terminated_early", False ), "termination_reason": metrics.get( "termination_reason", "" ), "stopped_on_first_error": metrics.get( "stopped_on_first_error", False ), "attempted_questions": metrics.get( "attempted_questions", 0 ), "badge": deepcopy(badge_record), "last_gap_report": deepcopy(last_gap_report), "gap_report": deepcopy(last_gap_report), "diagnostic_profile": deepcopy(metrics.get("diagnostic_profile") or {}), "text": ( f"# نتيجة الوسام\n\n" f"النتيجة: {metrics['correct_count']} من {metrics['total_questions']}\n\n" f"النسبة: {metrics['percentage']}%\n\n" f"الإجابات الخاطئة: {metrics['wrong_count']}\n\n" f"لا أعلم: {metrics['unknown_count']}\n\n" f"{message}" ), "progress": progress, "roadmap": roadmap, "current_badge": current, "feedback_event": feedback_event, "feedback_sound_key": sound_key, "save_result": save_result, "academic_attempt_result": academic_attempt_result, "academic_source_of_truth": True, "version": self.version, } def update_from_game_result(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.record_game_result(*args, **kwargs) def save_badge_result(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.record_game_result(*args, **kwargs) def register_badge_result(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.record_game_result(*args, **kwargs) def process_game_result(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.record_game_result(*args, **kwargs) def complete_badge(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.record_game_result(*args, **kwargs) def award_badge( self, identity: Any, subject: Any = "math", grade: Any = 1, badge_id: Any = "", **kwargs: Any, ) -> Dict[str, Any]: badge_id = self.normalize_badge_id(badge_id) progress = self.load_progress(identity, subject, grade) record = progress.get("badges", {}).get(badge_id) if not isinstance(record, dict): return {"success": False, "message": "الوسام غير موجود.", "version": self.version} record["earned"] = True record["unlocked"] = True record["status"] = self.STATUS_EARNED record["earned_at"] = record.get("earned_at") or self.now() record["best_percentage"] = 100.0 record["remedial_required"] = False self.save_progress(progress) return { "success": True, "message": "تم منح الوسام.", "badge": record, "roadmap": self.get_roadmap(identity, subject, grade), "version": self.version, } # ========================================================= # الحصة العلاجية وإعادة المحاولة # ========================================================= def mark_remedial_completed( self, identity: Any, subject: Any = "math", grade: Any = 1, badge_id: Any = "", task_id: Any = "", **kwargs: Any, ) -> Dict[str, Any]: progress = self.load_progress(identity, subject, grade) badge_key = self.normalize_badge_id( badge_id or progress.get("last_remedial_badge_id") or progress.get("current_badge_id") ) record = progress.get("badges", {}).get(badge_key) if not isinstance(record, dict): return {"success": False, "message": "الوسام غير موجود.", "version": self.version} record["remedial_completed"] = True record["remedial_completed_at"] = self.now() record["last_remedial_task_id"] = str(task_id or "") record["remedial_required"] = False record["status"] = self.STATUS_IN_PROGRESS progress["current_badge_id"] = badge_key progress["last_remedial_badge_id"] = badge_key self.save_progress(progress) return { "success": True, "message": "تم تسجيل مشاهدة الحصة. عد الآن إلى الوسام نفسه بسؤال جديد.", "return_to_badge": True, "badge_id": badge_key, "skill_id": record.get("skill_id", badge_key), "progress": progress, "version": self.version, } def remedial_completed(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.mark_remedial_completed(*args, **kwargs) # ========================================================= # إعادة الضبط # ========================================================= def reset_subject( self, identity: Any, subject: Any = "math", grade: Any = 1, **kwargs: Any, ) -> Dict[str, Any]: """ V28: إعادة الضبط لا تعيد تحميل التقدم بعد الحفظ ولا تنفذ مزامنتين. المسار: default_progress -> حفظ محلي واحد بدون sync -> مزامنة AcademicManager مرة واحدة -> بناء roadmap من نفس progress الموجود في الذاكرة StudentManager يبقى طبقة توافق، لكن لا نؤخر Reset بكتابة سحابية ثانية. """ progress = self.default_progress(identity, subject, grade) # حفظ محلي واحد فقط. local_save = self.save_progress( progress, sync_student=False, ) # AcademicManager هو مصدر الحقيقة الأكاديمي. academic_synced = self._sync_to_academic_manager(progress) # نبني الخريطة من نفس البيانات بدل load_progress/get_roadmap. full_roadmap = self._decorated_roadmap(progress) roadmap_items, stage_id, stage_title = self._current_stage_roadmap( full_roadmap, progress, ) total = len(roadmap_items) earned = sum( 1 for item in roadmap_items if isinstance(item, dict) and self._safe_bool(item.get("earned"), False) ) remaining = max(0, total - earned) percentage = round((earned / total) * 100, 2) if total else 0.0 roadmap = { "success": True, "ready": True, "identity": progress.get("identity"), "subject": progress.get("subject"), "subject_name": progress.get("subject_name"), "grade": progress.get("grade"), "roadmap": roadmap_items, "badges": roadmap_items, "current_badge_id": progress.get("current_badge_id", ""), "current_stage_id": stage_id, "current_stage_title": stage_title, "earned_count": earned, "remaining_count": remaining, "total_badges": total, "total_subject_badges": len(full_roadmap), "progress_percentage": percentage, "all_completed": progress.get("all_completed", False), "mastery_required": self.mastery_required, "progress": progress, "version": self.version, } return { "success": True, "ready": True, "message": "تمت إعادة تقدم المادة إلى أول وسام.", "progress": progress, "roadmap": roadmap, "save_result": local_save, "academic_synced": academic_synced, "student_manager_sync_skipped_for_speed": True, "reset_path": ( "default_progress->local_save_once" "->academic_sync_once->in_memory_roadmap" ), "version": self.version, } def reset_progress(self, *args: Any, **kwargs: Any) -> Dict[str, Any]: return self.reset_subject(*args, **kwargs) # ========================================================= # العرض النصي # ========================================================= def _status_icon(self, status: str) -> str: return { self.STATUS_EARNED: "🏅", self.STATUS_AVAILABLE: "▶️", self.STATUS_IN_PROGRESS: "🔄", self.STATUS_REMEDIAL_REQUIRED: "🧩", self.STATUS_LOCKED: "🔒", }.get(status, "•") def _status_text(self, status: str) -> str: return { self.STATUS_EARNED: "مكتسب", self.STATUS_AVAILABLE: "متاح", self.STATUS_IN_PROGRESS: "قيد المحاولة", self.STATUS_REMEDIAL_REQUIRED: "يحتاج حصة علاجية", self.STATUS_LOCKED: "مقفل", }.get(status, status) def format_roadmap(self, roadmap: Any) -> str: if isinstance(roadmap, dict): result = roadmap items = result.get("roadmap") or result.get("badges") or [] elif isinstance(roadmap, list): items = roadmap result = { "earned_count": sum(1 for item in items if isinstance(item, dict) and item.get("earned")), "total_badges": len(items), } else: return "لا توجد خريطة أوسمة متاحة." if not items: return "لا توجد أوسمة متاحة لهذه المادة." lines = [ "خريطة الأوسمة — المرحلة الحالية", ( f"المرحلة: {result.get('current_stage_title') or result.get('current_stage_id') or 'الحالية'}" if isinstance(result, dict) else "المرحلة الحالية" ), f"المكتسب: {result.get('earned_count', 0)} من {result.get('total_badges', len(items))}", f"المتبقي: {result.get('remaining_count', max(0, len(items) - result.get('earned_count', 0)))}", "", ] for item in items: if not isinstance(item, dict): continue status = str(item.get("status", self.STATUS_LOCKED)) current = " — الوسام الحالي" if item.get("is_current") else "" score = "" if self._safe_int(item.get("attempts_count"), 0) > 0: score = f" — أفضل نتيجة {self._safe_float(item.get('best_percentage'), 0.0):g}%" lines.append( f"{self._status_icon(status)} {item.get('order', '')}. " f"{item.get('title') or item.get('badge_id')} — {self._status_text(status)}{score}{current}" ) return "\n".join(lines).strip() def format_current_badge(self, current: Any) -> str: if not isinstance(current, dict): return "لا يوجد وسام حالي." if not current.get("success"): return str(current.get("message") or "أتم الطالب جميع الأوسمة المتاحة.") badge = current.get("badge") or current.get("current_badge") or {} if not isinstance(badge, dict): return "لا يوجد وسام حالي." lines = [ "الوسام الحالي", str(badge.get("title") or badge.get("badge_id") or "وسام"), f"المهارة: {badge.get('skill_title') or badge.get('skill_id') or ''}", f"الحالة: {self._status_text(str(badge.get('status', self.STATUS_AVAILABLE)))}", "قرار الإتقان أو الفجوة يصدر من StrategyManager بعد جمع أدلة كافية.", ] if self._safe_int(badge.get("attempts_count"), 0) > 0: lines.append(f"عدد المحاولات: {badge.get('attempts_count', 0)}") lines.append(f"أفضل نتيجة: {self._safe_float(badge.get('best_percentage'), 0.0):g}%") if badge.get("remedial_required"): lines.append("توجد حصة علاجية مطلوبة قبل إعادة الوسام.") elif badge.get("remedial_completed"): lines.append("تمت مشاهدة الحصة العلاجية ويمكن إعادة الوسام الآن.") return "\n".join(lines).strip() # ========================================================= # الفحص # ========================================================= # ========================================================= # تكليفات الألعاب والنتائج التفاعلية # ========================================================= def register_game_assignment( self, identity: Any = "student", subject: Any = "math", grade: Any = 1, badge_id: Any = "", game_path: str = "", title: str = "", assignment_type: str = "practice", source_repo: str = "Asem75/teacher_dataset", file_type: str = "", metadata: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> Dict[str, Any]: subject_key = self.normalize_subject(subject) grade_number = self.normalize_grade(grade) progress = self.load_progress(identity, subject_key, grade_number) path = str(game_path or kwargs.get("asset_path") or "").strip() if not path: return { "success": False, "message": "مسار ملف اللعبة مطلوب.", "version": self.version, } extension = Path(path).suffix.lower().lstrip(".") assignment = { "assignment_id": f"game_assignment_{uuid.uuid4().hex}", "identity": progress.get("identity"), "subject": subject_key, "grade": grade_number, "badge_id": self.normalize_badge_id(badge_id or progress.get("current_badge_id")), "title": str(title or Path(path).stem), "assignment_type": str(assignment_type or "practice"), "source_repo": str(source_repo or "Asem75/teacher_dataset"), "game_path": path, "file_type": str(file_type or extension or "unknown"), "interactive": extension in {"html", "htm", "json", "js"}, "status": "assigned", "metadata": deepcopy(metadata or {}), "assigned_at": self.now(), "started_at": "", "completed_at": "", } assignments = progress.setdefault("game_assignments", []) assignments.append(assignment) progress["game_assignments"] = assignments[-200:] saved = self.save_progress(progress) return { "success": True, "ready": True, "assignment": assignment, "progress": saved.get("progress", progress), "message": "تم ربط مرجع اللعبة بتقدم الطالب.", "version": self.version, } def list_game_assignments( self, identity: Any = "student", subject: Any = "math", grade: Any = 1, status: str = "", **kwargs: Any, ) -> Dict[str, Any]: progress = self.load_progress(identity, subject, grade) items = [item for item in progress.get("game_assignments", []) if isinstance(item, dict)] status_key = str(status or "").strip().lower() if status_key: items = [item for item in items if str(item.get("status", "")).lower() == status_key] return { "success": True, "ready": True, "assignments": items, "count": len(items), "version": self.version, } def update_game_assignment_status( self, identity: Any, assignment_id: str, status: str, subject: Any = "math", grade: Any = 1, **kwargs: Any, ) -> Dict[str, Any]: progress = self.load_progress(identity, subject, grade) assignment = None for item in progress.get("game_assignments", []): if isinstance(item, dict) and item.get("assignment_id") == assignment_id: assignment = item break if assignment is None: return {"success": False, "message": "تكليف اللعبة غير موجود.", "version": self.version} status_key = str(status or "assigned").strip().lower() assignment["status"] = status_key if status_key == "in_progress" and not assignment.get("started_at"): assignment["started_at"] = self.now() if status_key in {"completed", "done", "watched"}: assignment["completed_at"] = self.now() assignment["updated_at"] = self.now() self.save_progress(progress) return {"success": True, "assignment": assignment, "version": self.version} def record_interactive_game_result( self, identity: Any = "student", game_result: Optional[Dict[str, Any]] = None, subject: Any = "math", grade: Any = 1, badge_id: Any = "", assignment_id: str = "", **kwargs: Any, ) -> Dict[str, Any]: result = deepcopy(game_result or kwargs or {}) progress = self.load_progress(identity, subject, grade) entry = { "result_id": f"interactive_result_{uuid.uuid4().hex}", "assignment_id": str(assignment_id or result.get("assignment_id") or ""), "game_id": str(result.get("game_id") or result.get("id") or ""), "badge_id": self.normalize_badge_id(badge_id or result.get("badge_id") or progress.get("current_badge_id")), "subject": self.normalize_subject(result.get("subject") or subject), "score": self._safe_float(result.get("score", result.get("percentage", 0)), 0.0), "completed": self._safe_bool(result.get("completed"), False), "correct_count": self._safe_int(result.get("correct_count"), 0), "wrong_count": self._safe_int(result.get("wrong_count"), 0), "unknown_count": self._safe_int(result.get("unknown_count"), 0), "raw_result": self._json_safe(result), "created_at": self.now(), } results = progress.setdefault("interactive_game_results", []) results.append(entry) progress["interactive_game_results"] = results[-200:] if entry["assignment_id"]: self.update_game_assignment_status( identity=identity, assignment_id=entry["assignment_id"], status="completed" if entry["completed"] else "in_progress", subject=subject, grade=grade, ) progress = self.load_progress(identity, subject, grade) self.save_progress(progress) return { "success": True, "ready": True, "result": entry, "message": "تم حفظ نتيجة اللعبة التفاعلية.", "version": self.version, } def get_diagnostic_profile( self, identity: Any = "student", subject: Any = "math", grade: Any = 1, **kwargs: Any, ) -> Dict[str, Any]: progress = self.load_progress(identity, subject, grade) history = [item for item in progress.get("diagnostic_history", []) if isinstance(item, dict)] latest = history[-1] if history else {} return { "success": True, "ready": True, "latest": latest, "history": history, "count": len(history), "assessment_mode": "sequential_exact_skill_evidence_history", "profile_type": "evidence_history_not_level_test", "decision_owner": "StrategyManager", "version": self.version, } # أسماء واجهات V17 للتوافق def update_from_game_state(self, identity="student", game_state=None, **kwargs): return self.record_game_result(identity=identity, state=game_state, **kwargs) def update_progress_from_game_state(self, *args, **kwargs): return self.update_from_game_state(*args, **kwargs) def update_from_game(self, *args, **kwargs): return self.update_from_game_state(*args, **kwargs) def finalize_badge(self, *args, **kwargs): return self.update_from_game_state(*args, **kwargs) def get_badge_roadmap(self, *args, **kwargs): return self.get_roadmap(*args, **kwargs) def roadmap(self, *args, **kwargs): return self.get_roadmap(*args, **kwargs) def reset(self, *args, **kwargs): return self.reset_subject(*args, **kwargs) def reset_badge_progress(self, *args, **kwargs): return self.reset_subject(*args, **kwargs) def set_current_badge(self, identity="student", badge_id="math_addition", subject="math", grade=1, **kwargs): progress = self.load_progress(identity, subject, grade) badge_key = self.normalize_badge_id(badge_id) if badge_key not in progress.get("badges", {}): return {"success": False, "message": f"لا يوجد وسام بهذا الاسم: {badge_key}", "version": self.version} progress["current_badge_id"] = badge_key progress["all_completed"] = False record = progress["badges"][badge_key] record["unlocked"] = True if not record.get("earned"): record["status"] = self.STATUS_AVAILABLE self.save_progress(progress) return {"success": True, "current_badge_id": badge_key, "badge": record, "progress": progress, "version": self.version} def summary(self) -> Dict[str, Any]: return { "success": True, "ready": True, "version": self.version, "schema_version": self.schema_version, "progress_root": self.progress_root, "base_dir": self.base_dir, "mastery_required": None, "mastery_rule_owner": "StrategyManager", "award_rule": "applies_strategy_decision_only", "assessment_mode": "sequential_exact_skill_evidence_ledger", "manager_role": "evidence_and_progress_ledger", "roadmap_scope": "current_stage_only", "roadmap_source": "official_skill_files_only", "obsolete_skill_math_bank": False, "reset_scope": "subject_progress_to_first_official_skill", "reset_runtime": "single_local_save_single_academic_sync_no_reload", "decides_mastery": False, "decides_remediation": False, "computes_percentage_for_reporting_only": True, "stop_on_first_error": False, "early_termination_is_not_first_error": True, "decision_owner": "StrategyManager", "orchestrator": "SessionManager", "game_assignments_supported": True, "interactive_game_results_supported": True, "game_asset_discovery": "external_game_library_manager_required", "tracks": { "correct": True, "wrong": True, "unknown": True, "repeated_error": True, "help_requested": True, "remedial_completed": True, }, "next_badge_rule": ( "opens_only_after_strategy_manager_confirms_mastery" ), "failed_badge_rule": ( "remedial_only_after_strategy_manager_confirms_gap" ), "student_manager_sync": "optional_and_safe", "module_functions": [ "get_roadmap", "format_roadmap", "get_current_badge", "format_current_badge", "record_game_result", "update_from_game_result", "mark_remedial_completed", "register_game_assignment", "list_game_assignments", "update_game_assignment_status", "record_interactive_game_result", "get_diagnostic_profile", "reset_subject", "summary", ], "message": ( "BadgeProgressManager يسجل الأدلة والتقدم فقط، " "ويطبق قرار StrategyManager دون امتلاك سياسة الإتقان أو العلاج أو التشخيص." ), } def is_ready(self) -> bool: return True # أسماء توافقية ProgressManager = BadgeProgressManager MasteryBadgeProgressManager = BadgeProgressManager _MANAGER: Optional[BadgeProgressManager] = None def get_manager() -> BadgeProgressManager: global _MANAGER if _MANAGER is None: _MANAGER = BadgeProgressManager() return _MANAGER def get_roadmap(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().get_roadmap(*args, **kwargs) def get_current_badge(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().get_current_badge(*args, **kwargs) def record_game_result(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().record_game_result(*args, **kwargs) def update_from_game_result(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().update_from_game_result(*args, **kwargs) def mark_remedial_completed(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().mark_remedial_completed(*args, **kwargs) def reset_subject(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().reset_subject(*args, **kwargs) def register_game_assignment(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().register_game_assignment(*args, **kwargs) def list_game_assignments(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().list_game_assignments(*args, **kwargs) def update_game_assignment_status(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().update_game_assignment_status(*args, **kwargs) def record_interactive_game_result(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().record_interactive_game_result(*args, **kwargs) def get_diagnostic_profile(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().get_diagnostic_profile(*args, **kwargs) def summary() -> Dict[str, Any]: return get_manager().summary()