# admin_manager.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ مدير الإدارة الكامل لمنصة Divid Teacher. المسؤوليات: - إدارة تسجيل الطلاب وتغيير حالة الحساب. - تعديل ملف الطالب ومفتاح الدخول وكلمة مرور ولي الأمر. - إدارة الأجهزة الموثوقة للطالب. - عرض قوائم الطلاب والتقارير الإدارية الآمنة. - الاحتفاظ بسجل تدقيق لجميع العمليات الإدارية. - إدارة التوجيه العام والمنهج والمواد والمسارات والاستراتيجيات وPomodoro. - إدارة اللغات عند توفر LanguageManager. - توفير ملخص موحد لفحص النظام. ملاحظات أمنية: - لا يعيد كلمات المرور أو قيم Hash إلى واجهة الإدارة. - يمكن تفعيل التحقق الداخلي بمفتاح الإدارة من متغيرات البيئة. - عمليات الحذف تحتاج تأكيدًا صريحًا ولا تُنفذ بالخطأ. """ from __future__ import annotations import copy import hmac import json import os import threading from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple from curriculum_manager import CurriculumManager from strategy_manager import StrategyManager try: from language_manager import LanguageManager except Exception: LanguageManager = None try: from student_manager import StudentManager except Exception: StudentManager = None try: from device_identity_manager import DeviceIdentityManager except Exception: DeviceIdentityManager = None try: from badge_progress_manager import BadgeProgressManager except Exception: BadgeProgressManager = None try: from report_manager import ReportManager except Exception: ReportManager = None try: from ai_task_queue_manager import AITaskQueueManager except Exception: AITaskQueueManager = None try: from config import SYSTEM_FOLDER except Exception: SYSTEM_FOLDER = "system" VERSION = "admin_manager_v4_full_access_student_registration_control" SCHEMA_VERSION = "admin_manager_schema_v4" ACCOUNT_STATUSES = {"pending", "active", "suspended", "rejected"} STATUS_ALIASES = { "pending": "pending", "بانتظار": "pending", "قيد الانتظار": "pending", "انتظار": "pending", "اعادة للانتظار": "pending", "إعادة للانتظار": "pending", "active": "active", "تفعيل": "active", "مفعل": "active", "مفعّل": "active", "قبول": "active", "مقبول": "active", "suspended": "suspended", "ايقاف": "suspended", "إيقاف": "suspended", "ايقاف مؤقت": "suspended", "إيقاف مؤقت": "suspended", "موقوف": "suspended", "rejected": "rejected", "رفض": "rejected", "مرفوض": "rejected", } PRIVATE_STUDENT_FIELDS = { "parent_password", "parent_password_hash", "parent_pin", "parent_pin_hash", "password", "password_hash", "secret", "secret_hash", "token", "hf_token", "api_key", } FULL_PERMISSIONS = [ "students.read", "students.update", "students.activate", "students.suspend", "students.reject", "students.pending", "students.bulk_status", "students.reset_access_key", "students.reset_parent_password", "students.devices.read", "students.devices.remove", "students.delete", "curriculum.read", "curriculum.write", "strategies.read", "strategies.write", "languages.read", "languages.write", "reports.read", "system.read", "system.write", "audit.read", ] class AdminManager: """واجهة الإدارة المركزية.""" def __init__(self, *args: Any, **kwargs: Any) -> None: self.version = VERSION self.schema_version = SCHEMA_VERSION self._lock = threading.RLock() self.system_folder = Path(SYSTEM_FOLDER) self.system_folder.mkdir(parents=True, exist_ok=True) self.general_instruction_file = self.system_folder / "general_instruction.txt" self.admin_notes_file = self.system_folder / "admin_notes.json" self.audit_log_file = self.system_folder / "admin_audit_log.jsonl" self.require_admin_key = self._env_bool("ADMIN_MANAGER_REQUIRE_KEY", False) self.admin_key = str( os.getenv("ADMIN_PANEL_KEY", "") or os.getenv("ADMIN_MANAGER_KEY", "") or "" ).strip() self.curriculum_manager = CurriculumManager() self.strategy_manager = StrategyManager() self.language_manager = self._safe_manager(LanguageManager) self.student_manager = self._safe_manager(StudentManager) self.device_manager = self._safe_manager(DeviceIdentityManager) self.badge_progress_manager = self._safe_manager(BadgeProgressManager) self.report_manager = self._safe_manager(ReportManager) self.queue_manager = self._safe_manager(AITaskQueueManager) if not self.general_instruction_file.exists(): self.save_general_instruction( self.default_general_instruction(), changed_by="system", admin_key=self.admin_key if self.require_admin_key else "", _skip_guard=True, ) if not self.admin_notes_file.exists(): self.save_admin_notes([]) # ========================================================= # أدوات عامة وأمنية # ========================================================= def _env_bool(self, name: str, default: bool = False) -> bool: raw = str(os.getenv(name, "1" if default else "0")).strip().lower() return raw not in {"0", "false", "no", "off", ""} def _safe_manager(self, manager_class: Any) -> Any: if manager_class is None: return None try: return manager_class() except Exception: return None def now(self) -> str: return datetime.now(timezone.utc).isoformat() def _normalize_status(self, status: Any) -> str: raw = str(status or "").strip().lower() return STATUS_ALIASES.get(raw, raw) def verify_admin_key(self, admin_key: Any = "") -> bool: if not self.require_admin_key: return True supplied = str(admin_key or "").strip() if not self.admin_key or not supplied: return False return hmac.compare_digest(supplied, self.admin_key) def _permission_error(self, permission: str) -> Dict[str, Any]: return { "success": False, "authorized": False, "permission": permission, "message": "مفتاح الإدارة غير صحيح أو غير موجود.", "version": self.version, } def _guard(self, permission: str, admin_key: Any = "") -> Optional[Dict[str, Any]]: if self.verify_admin_key(admin_key): return None return self._permission_error(permission) def _require_student_manager(self) -> Optional[Dict[str, Any]]: if self.student_manager is not None: return None return { "success": False, "message": "StudentManager غير متوفر.", "version": self.version, } def _sanitize(self, value: Any) -> Any: if isinstance(value, dict): result: Dict[str, Any] = {} for key, item in value.items(): if str(key).lower() in PRIVATE_STUDENT_FIELDS: continue result[key] = self._sanitize(item) return result if isinstance(value, list): return [self._sanitize(item) for item in value] if isinstance(value, tuple): return [self._sanitize(item) for item in value] return value def _student_admin_summary(self, student: Any) -> Dict[str, Any]: if not isinstance(student, dict): return {} safe = self._sanitize(copy.deepcopy(student)) devices = safe.get("authorized_devices", []) if not isinstance(devices, list): devices = [] safe["authorized_devices_count"] = len( [ device for device in devices if isinstance(device, dict) and device.get("status", "active") == "active" ] ) return safe def _audit( self, action: str, target: str = "", actor: str = "admin", success: bool = True, details: Any = None, ) -> None: event = { "created_at": self.now(), "action": str(action or ""), "target": str(target or ""), "actor": str(actor or "admin"), "success": bool(success), "details": self._sanitize(details if details is not None else {}), "version": self.version, } try: with self._lock: with open(self.audit_log_file, "a", encoding="utf-8") as file: file.write(json.dumps(event, ensure_ascii=False) + "\n") except Exception: pass def load_audit_log( self, limit: int = 200, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("audit.read", admin_key) if denied: return denied items: List[Dict[str, Any]] = [] try: if self.audit_log_file.exists(): with open(self.audit_log_file, "r", encoding="utf-8") as file: for line in file: line = line.strip() if not line: continue try: item = json.loads(line) except Exception: continue if isinstance(item, dict): items.append(item) except Exception: items = [] limit = max(1, min(int(limit or 200), 2000)) return { "success": True, "count": min(len(items), limit), "events": items[-limit:], "version": self.version, } # ========================================================= # إدارة الطلاب والتسجيل # ========================================================= def refresh_students(self, use_cloud: bool = True) -> Dict[str, Any]: unavailable = self._require_student_manager() if unavailable: return unavailable try: students = self.student_manager.load_students(use_cloud=bool(use_cloud)) return { "success": True, "students_count": len(students) if isinstance(students, dict) else 0, "version": self.version, } except Exception as exc: return { "success": False, "message": f"فشل تحديث قائمة الطلاب: {exc}", "version": self.version, } def list_students( self, status: Any = "all", full_profiles: bool = False, search_text: Any = "", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.read", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable try: self.student_manager.load_students(use_cloud=True) normalized_status = self._normalize_status(status) if normalized_status in ACCOUNT_STATUSES: students = self.student_manager.list_students_by_status(normalized_status) else: records = self.student_manager.all_students(full_profiles=bool(full_profiles)) students = list(records.values()) if isinstance(records, dict) else [] wanted = str(search_text or "").strip().lower() safe_students: List[Dict[str, Any]] = [] for student in students: if not isinstance(student, dict): continue safe = self._student_admin_summary(student) if wanted: haystack = " ".join( str(safe.get(field, "")) for field in ( "identity", "student_id", "full_name", "name", "access_key", "email", "phone", "country", "grade", ) ).lower() if wanted not in haystack: continue safe_students.append(safe) safe_students.sort( key=lambda item: str( item.get("registration_requested_at") or item.get("created_at") or "" ), reverse=True, ) return { "success": True, "status_filter": normalized_status if normalized_status else "all", "count": len(safe_students), "students": safe_students, "version": self.version, } except Exception as exc: return { "success": False, "message": f"فشل عرض الطلاب: {exc}", "version": self.version, } def list_pending_registrations(self, admin_key: Any = "") -> Dict[str, Any]: return self.list_students(status="pending", admin_key=admin_key) def get_student_record( self, identity: Any, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.read", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable student = self.student_manager.get_student(identity) if not isinstance(student, dict): return { "success": False, "message": "لم يتم العثور على الطالب.", "identity": str(identity or ""), "version": self.version, } return { "success": True, "student": self._student_admin_summary(student), "version": self.version, } def student_status_counts(self, admin_key: Any = "") -> Dict[str, Any]: denied = self._guard("students.read", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable counts = {status: 0 for status in sorted(ACCOUNT_STATUSES)} try: records = self.student_manager.all_students(full_profiles=False) if isinstance(records, dict): for student in records.values(): if not isinstance(student, dict): continue status = self._normalize_status(student.get("account_status", "pending")) counts[status] = counts.get(status, 0) + 1 except Exception: pass return { "success": True, "counts": counts, "total": sum(counts.values()), "version": self.version, } def change_student_registration_status( self, identity: Any, new_status: Any, reason: Any = "", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: normalized_status = self._normalize_status(new_status) permission_map = { "active": "students.activate", "suspended": "students.suspend", "rejected": "students.reject", "pending": "students.pending", } permission = permission_map.get(normalized_status, "students.update") denied = self._guard(permission, admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable if normalized_status not in ACCOUNT_STATUSES: return { "success": False, "message": "حالة تسجيل الطالب غير صحيحة.", "allowed_statuses": sorted(ACCOUNT_STATUSES), "version": self.version, } actor = str(changed_by or "admin") reason_text = str(reason or "").strip() try: result = self.student_manager.set_account_status( identity=identity, status=normalized_status, changed_by=actor, reason=reason_text, ) except TypeError: if normalized_status == "active": result = self.student_manager.activate_student( identity, activated_by=actor, reason=reason_text, ) elif normalized_status == "suspended": result = self.student_manager.suspend_student( identity, suspended_by=actor, reason=reason_text, ) elif normalized_status == "rejected": result = self.student_manager.reject_student( identity, rejected_by=actor, reason=reason_text, ) else: result = self.student_manager.set_student_pending( identity, changed_by=actor, reason=reason_text, ) success = bool(isinstance(result, dict) and result.get("success", False)) self._audit( action="student.registration_status_changed", target=str(identity or ""), actor=actor, success=success, details={ "new_status": normalized_status, "reason": reason_text, "result": result, }, ) if not isinstance(result, dict): return { "success": success, "account_status": normalized_status, "message": "تم تحديث حالة تسجيل الطالب." if success else "فشل تحديث حالة الطالب.", "version": self.version, } result = self._sanitize(result) result["version"] = self.version result["admin_action"] = "change_student_registration_status" return result def activate_student_registration( self, identity: Any, reason: Any = "تم قبول التسجيل وتفعيل الحساب.", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: return self.change_student_registration_status( identity=identity, new_status="active", reason=reason, changed_by=changed_by, admin_key=admin_key, ) def suspend_student_registration( self, identity: Any, reason: Any = "تم إيقاف الحساب مؤقتًا.", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: return self.change_student_registration_status( identity=identity, new_status="suspended", reason=reason, changed_by=changed_by, admin_key=admin_key, ) def reject_student_registration( self, identity: Any, reason: Any = "تم رفض طلب التسجيل.", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: return self.change_student_registration_status( identity=identity, new_status="rejected", reason=reason, changed_by=changed_by, admin_key=admin_key, ) def return_student_to_pending( self, identity: Any, reason: Any = "أعيد الحساب إلى قائمة انتظار المراجعة.", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: return self.change_student_registration_status( identity=identity, new_status="pending", reason=reason, changed_by=changed_by, admin_key=admin_key, ) # أسماء مختصرة متوافقة مع الواجهة def activate_student(self, identity: Any, reason: Any = "", changed_by: Any = "admin", admin_key: Any = "") -> Dict[str, Any]: return self.activate_student_registration(identity, reason, changed_by, admin_key) def suspend_student(self, identity: Any, reason: Any = "", changed_by: Any = "admin", admin_key: Any = "") -> Dict[str, Any]: return self.suspend_student_registration(identity, reason, changed_by, admin_key) def reject_student(self, identity: Any, reason: Any = "", changed_by: Any = "admin", admin_key: Any = "") -> Dict[str, Any]: return self.reject_student_registration(identity, reason, changed_by, admin_key) def set_student_pending(self, identity: Any, reason: Any = "", changed_by: Any = "admin", admin_key: Any = "") -> Dict[str, Any]: return self.return_student_to_pending(identity, reason, changed_by, admin_key) def bulk_change_student_status( self, identities: Sequence[Any], new_status: Any, reason: Any = "", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.bulk_status", admin_key) if denied: return denied normalized_status = self._normalize_status(new_status) unique_identities: List[str] = [] for identity in identities or []: text = str(identity or "").strip() if text and text not in unique_identities: unique_identities.append(text) results: List[Dict[str, Any]] = [] for identity in unique_identities: result = self.change_student_registration_status( identity=identity, new_status=normalized_status, reason=reason, changed_by=changed_by, admin_key=admin_key, ) results.append({"identity": identity, "result": result}) success_count = sum( 1 for item in results if isinstance(item.get("result"), dict) and item["result"].get("success") ) return { "success": success_count == len(results) and bool(results), "requested_count": len(results), "success_count": success_count, "failed_count": len(results) - success_count, "new_status": normalized_status, "results": results, "version": self.version, } def update_student_profile( self, identity: Any, updates: Any = None, changed_by: Any = "admin", admin_key: Any = "", **kwargs: Any, ) -> Dict[str, Any]: denied = self._guard("students.update", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable data = dict(updates) if isinstance(updates, dict) else {} data.update(kwargs) forbidden = { "identity", "student_id", "parent_password_hash", "parent_pin_hash", "account_status", "authorized_devices", "created_at", } cleaned = { key: value for key, value in data.items() if key not in forbidden and key not in PRIVATE_STUDENT_FIELDS } if not cleaned: return { "success": False, "message": "لا توجد بيانات مسموح بتحديثها.", "version": self.version, } try: student = self.student_manager.update_student(identity, cleaned) success = isinstance(student, dict) except Exception as exc: student = None success = False error_message = str(exc) else: error_message = "" self._audit( action="student.profile_updated", target=str(identity or ""), actor=str(changed_by or "admin"), success=success, details={"updated_fields": sorted(cleaned.keys())}, ) return { "success": success, "message": "تم تحديث ملف الطالب." if success else f"فشل تحديث ملف الطالب: {error_message}", "student": self._student_admin_summary(student), "updated_fields": sorted(cleaned.keys()), "version": self.version, } def reset_student_access_key( self, identity: Any, new_access_key: Any = "", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.reset_access_key", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable student = self.student_manager.get_student(identity) if not isinstance(student, dict): return { "success": False, "message": "لم يتم العثور على الطالب.", "version": self.version, } candidate = str(new_access_key or "").strip() if not candidate: candidate = self.student_manager.generate_access_key( full_name=student.get("full_name") or student.get("name") or "", identity=student.get("identity") or identity, ) updated = self.student_manager.update_student(identity, {"access_key": candidate}) actual_key = str(updated.get("access_key", "")) if isinstance(updated, dict) else "" success = bool(actual_key and actual_key == self.student_manager.normalize_access_key(candidate)) self._audit( action="student.access_key_reset", target=str(identity or ""), actor=str(changed_by or "admin"), success=success, details={"access_key_changed": success}, ) return { "success": success, "message": "تم تغيير مفتاح دخول الطالب." if success else "تعذر تغيير مفتاح الدخول؛ قد يكون مستخدمًا لطالب آخر.", "identity": str(identity or ""), "access_key": actual_key, "version": self.version, } def reset_parent_password( self, identity: Any, new_parent_password: Any, changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.reset_parent_password", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable password = str(new_parent_password or "") if len(password) < 4: return { "success": False, "message": "كلمة مرور ولي الأمر يجب أن تتكون من 4 خانات على الأقل.", "version": self.version, } result = self.student_manager.set_parent_password(identity, password) success = bool(isinstance(result, dict) and result.get("success", False)) self._audit( action="student.parent_password_reset", target=str(identity or ""), actor=str(changed_by or "admin"), success=success, details={"password_hash_only": True}, ) result = result if isinstance(result, dict) else {"success": success} result = self._sanitize(result) result["version"] = self.version return result def list_student_devices( self, identity: Any, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.devices.read", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable devices = self.student_manager.get_authorized_devices(identity) safe_devices: List[Dict[str, Any]] = [] for device in devices if isinstance(devices, list) else []: if not isinstance(device, dict): continue safe_device = dict(device) fingerprint_hash = str(safe_device.get("fingerprint_hash", "")) safe_device["fingerprint_preview"] = ( fingerprint_hash[:8] + "…" if fingerprint_hash else "" ) safe_device.pop("raw_fingerprint", None) safe_devices.append(safe_device) return { "success": True, "identity": str(identity or ""), "count": len(safe_devices), "devices": safe_devices, "version": self.version, } def remove_student_device( self, identity: Any, fingerprint_hash: Any, changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.devices.remove", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable result = self.student_manager.remove_authorized_device(identity, fingerprint_hash) success = bool(isinstance(result, dict) and result.get("success", False)) self._audit( action="student.device_removed", target=str(identity or ""), actor=str(changed_by or "admin"), success=success, details={"fingerprint_preview": str(fingerprint_hash or "")[:8]}, ) result = result if isinstance(result, dict) else {"success": success} result = self._sanitize(result) result["version"] = self.version return result def clear_student_devices( self, identity: Any, changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.devices.remove", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable devices = self.student_manager.get_authorized_devices(identity) results: List[Dict[str, Any]] = [] for device in devices if isinstance(devices, list) else []: if not isinstance(device, dict): continue fingerprint_hash = device.get("fingerprint_hash", "") if not fingerprint_hash: continue result = self.student_manager.remove_authorized_device(identity, fingerprint_hash) results.append(self._sanitize(result)) self._audit( action="student.devices_cleared", target=str(identity or ""), actor=str(changed_by or "admin"), success=True, details={"devices_processed": len(results)}, ) return { "success": True, "identity": str(identity or ""), "removed_count": sum(1 for item in results if item.get("success")), "results": results, "version": self.version, } def delete_student_account( self, identity: Any, confirmation: Any, changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("students.delete", admin_key) if denied: return denied unavailable = self._require_student_manager() if unavailable: return unavailable if str(confirmation or "").strip() != "DELETE_STUDENT": return { "success": False, "message": "لحذف الطالب اكتب DELETE_STUDENT في خانة التأكيد.", "version": self.version, } existed = isinstance(self.student_manager.get_student(identity), dict) success = bool(self.student_manager.delete_student(identity)) if existed else False self._audit( action="student.deleted", target=str(identity or ""), actor=str(changed_by or "admin"), success=success, details={"explicit_confirmation": True}, ) return { "success": success, "message": "تم حذف النسخة المحلية وفهرس الطالب." if success else "لم يتم العثور على الطالب.", "warning": "قد تحتاج الملفات السحابية القديمة إلى حذف يدوي من خزنة الطلاب.", "version": self.version, } # ========================================================= # التوجيه العام والملاحظات # ========================================================= def default_general_instruction(self) -> str: return """ التوجيه العام لمنصة Divid Teacher: - لا تشرح للطالب دفعة واحدة. - قسّم الدرس إلى أجزاء قصيرة. - اسأل الطالب قبل بداية الحصة. - اسأل الطالب بعد كل جزء. - لا تنتقل إلى الجزء التالي حتى يجيب الطالب. - استخدم الخطة A للدرس الرئيسي. - استخدم الخطة B لتقليص الفجوة التعليمية عند الحاجة. - لا تخبر الطالب أنه ضعيف. - استخدم أسلوبًا مناسبًا لعمر الطالب. - إذا خرج الطالب عن الموضوع، أعده بهدوء. - إذا ظهرت فجوة في مهارة أساسية، اربطها بوسام إتقان مناسب. """.strip() def load_general_instruction(self) -> str: try: with open(self.general_instruction_file, "r", encoding="utf-8") as file: return file.read() except Exception: text = self.default_general_instruction() try: with open(self.general_instruction_file, "w", encoding="utf-8") as file: file.write(text) except Exception: pass return text def save_general_instruction( self, text: Any, changed_by: Any = "admin", admin_key: Any = "", _skip_guard: bool = False, ) -> Dict[str, Any]: if not _skip_guard: denied = self._guard("system.write", admin_key) if denied: return denied text = str(text or "").strip() with self._lock: with open(self.general_instruction_file, "w", encoding="utf-8") as file: file.write(text) self._audit( action="system.general_instruction_saved", actor=str(changed_by or "admin"), success=True, details={"characters": len(text)}, ) return { "success": True, "message": "تم حفظ التوجيه العام.", "text": text, "version": self.version, } def append_general_instruction( self, extra_text: Any, changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("system.write", admin_key) if denied: return denied current = self.load_general_instruction() extra = str(extra_text or "").strip() if not extra: return { "success": False, "message": "لا يوجد نص لإضافته.", "text": current, "version": self.version, } return self.save_general_instruction( current + "\n\n" + extra, changed_by=changed_by, admin_key=admin_key, ) def load_admin_notes(self) -> List[Dict[str, Any]]: try: with open(self.admin_notes_file, "r", encoding="utf-8") as file: data = json.load(file) return data if isinstance(data, list) else [] except Exception: self.save_admin_notes([]) return [] def save_admin_notes(self, notes: Any) -> bool: values = notes if isinstance(notes, list) else [] with self._lock: with open(self.admin_notes_file, "w", encoding="utf-8") as file: json.dump(values, file, ensure_ascii=False, indent=4) return True def add_admin_note( self, title: Any, text: Any, category: Any = "general", changed_by: Any = "admin", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("system.write", admin_key) if denied: return denied notes = self.load_admin_notes() note = { "id": f"note_{int(datetime.now(timezone.utc).timestamp())}_{len(notes) + 1}", "title": str(title or "").strip(), "text": str(text or "").strip(), "category": str(category or "general").strip(), "created_at": self.now(), "created_by": str(changed_by or "admin"), } notes.append(note) self.save_admin_notes(notes) self._audit( action="system.admin_note_added", target=note["id"], actor=str(changed_by or "admin"), success=True, ) return { "success": True, "note": note, "message": "تمت إضافة الملاحظة الإدارية.", "version": self.version, } # ========================================================= # المنهج والمواد والمسارات # ========================================================= def curriculum_summary(self) -> Any: return self.curriculum_manager.summary_for_admin() def add_track(self, grade: Any, track_name: Any, admin_key: Any = "") -> Dict[str, Any]: denied = self._guard("curriculum.write", admin_key) if denied: return denied result = self.curriculum_manager.add_track(grade=grade, track_name=track_name) self._audit("curriculum.track_added", str(track_name or ""), success=bool(result), details={"grade": grade}) return { "success": bool(result), "message": "تمت إضافة المسار." if result else "فشل إضافة المسار.", "version": self.version, } def remove_track(self, grade: Any, track_name: Any, admin_key: Any = "") -> Dict[str, Any]: denied = self._guard("curriculum.write", admin_key) if denied: return denied result = self.curriculum_manager.remove_track(grade=grade, track_name=track_name) self._audit("curriculum.track_removed", str(track_name or ""), success=bool(result), details={"grade": grade}) return { "success": bool(result), "message": "تم حذف المسار." if result else "فشل حذف المسار.", "version": self.version, } def add_subject( self, grade: Any, subject: Any, subject_type: Any = "core", admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("curriculum.write", admin_key) if denied: return denied result = self.curriculum_manager.add_subject( grade=grade, subject=subject, subject_type=subject_type, ) message = "تمت إضافة المادة كمادة ثانوية." if subject_type == "secondary" else "تمت إضافة المادة كمادة أساسية." self._audit("curriculum.subject_added", str(subject or ""), success=bool(result), details={"grade": grade, "subject_type": subject_type}) return { "success": bool(result), "message": message if result else "فشل إضافة المادة.", "version": self.version, } def remove_subject(self, grade: Any, subject: Any, admin_key: Any = "") -> Dict[str, Any]: denied = self._guard("curriculum.write", admin_key) if denied: return denied result = self.curriculum_manager.remove_subject(grade=grade, subject=subject) self._audit("curriculum.subject_removed", str(subject or ""), success=bool(result), details={"grade": grade}) return { "success": bool(result), "message": "تم حذف المادة." if result else "فشل حذف المادة.", "version": self.version, } def change_subject_type( self, grade: Any, subject: Any, new_type: Any, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("curriculum.write", admin_key) if denied: return denied result = self.curriculum_manager.change_subject_type( grade=grade, subject=subject, new_type=new_type, ) self._audit("curriculum.subject_type_changed", str(subject or ""), success=bool(result), details={"grade": grade, "new_type": new_type}) return { "success": bool(result), "message": "تم تغيير نوع المادة." if result else "فشل تغيير نوع المادة.", "version": self.version, } def get_subject_route(self, grade: Any, subject: Any) -> Any: return self.curriculum_manager.decide_subject_route(grade=grade, subject=subject) # ========================================================= # الاستراتيجيات وPomodoro # ========================================================= def strategy_summary(self) -> Any: return self.strategy_manager.summary() def add_subject_strategy( self, subject: Any, strategy_name: Any, steps_text: Any, methods_text: Any = "", pomodoro_recommended: bool = True, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("strategies.write", admin_key) if denied: return denied steps = [line.strip() for line in str(steps_text or "").splitlines() if line.strip()] methods = [line.strip() for line in str(methods_text or "").splitlines() if line.strip()] if not steps: return { "success": False, "message": "يجب إدخال خطوات الاستراتيجية.", "version": self.version, } result = self.strategy_manager.add_subject_strategy( subject=subject, strategy_name=strategy_name, steps=steps, methods=methods, pomodoro_recommended=pomodoro_recommended, ) self._audit("strategy.subject_strategy_added", str(strategy_name or ""), success=bool(result), details={"subject": subject}) return { "success": bool(result), "message": "تم حفظ استراتيجية المادة." if result else "فشل حفظ الاستراتيجية.", "version": self.version, } def update_pomodoro_settings( self, study_minutes: Any = 25, short_break_minutes: Any = 5, long_break_minutes: Any = 15, long_break_after_sessions: Any = 4, enabled: bool = True, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("strategies.write", admin_key) if denied: return denied settings = self.strategy_manager.update_pomodoro_settings( study_minutes=study_minutes, short_break_minutes=short_break_minutes, long_break_minutes=long_break_minutes, long_break_after_sessions=long_break_after_sessions, enabled=enabled, ) self._audit("strategy.pomodoro_updated", success=True, details=settings) return { "success": True, "message": "تم تحديث إعدادات Pomodoro.", "settings": settings, "version": self.version, } def choose_current_strategy( self, student: Any, subject: Any, plan_type: Any = "A", lesson_title: Any = "", gap_skill: Any = "", ) -> Any: return self.strategy_manager.choose_strategy( student=student, subject=subject, plan_type=plan_type, lesson_title=lesson_title, gap_skill=gap_skill, ) # ========================================================= # اللغات # ========================================================= def language_summary(self) -> Any: if self.language_manager is None: return "مدير اللغات غير متوفر حاليًا." return self.language_manager.summary() def add_language( self, code: Any, name: Any, direction: Any = "ltr", enabled: bool = True, admin_key: Any = "", ) -> Dict[str, Any]: denied = self._guard("languages.write", admin_key) if denied: return denied if self.language_manager is None: return { "success": False, "message": "مدير اللغات غير متوفر.", "version": self.version, } result = self.language_manager.add_supported_language( code=code, name=name, direction=direction, enabled=enabled, ) self._audit("language.added", str(code or ""), success=bool(result), details={"name": name, "direction": direction, "enabled": enabled}) return { "success": bool(result), "message": "تمت إضافة اللغة." if result else "فشل إضافة اللغة.", "version": self.version, } # ========================================================= # فحص الخدمات ولوحة الإدارة # ========================================================= def _manager_summary(self, manager: Any) -> Dict[str, Any]: if manager is None: return {"success": False, "available": False} try: if hasattr(manager, "summary"): result = manager.summary() return result if isinstance(result, dict) else {"success": True, "summary": result} return { "success": True, "available": True, "version": getattr(manager, "version", "unknown"), } except Exception as exc: return { "success": False, "available": True, "message": str(exc), } def services_summary(self) -> Dict[str, Any]: return { "student_manager": self._manager_summary(self.student_manager), "device_identity_manager": self._manager_summary(self.device_manager), "badge_progress_manager": self._manager_summary(self.badge_progress_manager), "report_manager": self._manager_summary(self.report_manager), "ai_task_queue_manager": self._manager_summary(self.queue_manager), "language_manager": self._manager_summary(self.language_manager), } def system_snapshot(self, admin_key: Any = "") -> Dict[str, Any]: denied = self._guard("system.read", admin_key) if denied: return denied try: curriculum = self.curriculum_manager.load_config() except Exception: curriculum = {} try: strategies = self.strategy_manager.load_general_strategies() except Exception: strategies = {} try: languages = self.language_manager.load_config() if self.language_manager is not None else {} except Exception: languages = {} return { "success": True, "created_at": self.now(), "general_instruction": self.load_general_instruction(), "curriculum": curriculum, "strategies": strategies, "admin_notes": self.load_admin_notes(), "languages": languages, "student_status_counts": self.student_status_counts(admin_key=admin_key), "services": self.services_summary(), "version": self.version, } def format_students_for_dashboard(self, status: Any = "pending", admin_key: Any = "") -> str: result = self.list_students(status=status, admin_key=admin_key) if not result.get("success"): return str(result.get("message", "تعذر عرض الطلاب.")) students = result.get("students", []) if not students: return f"لا يوجد طلاب بالحالة: {self._normalize_status(status)}." lines = [f"## الطلاب — {self._normalize_status(status)}", ""] for index, student in enumerate(students, 1): lines.extend( [ f"### {index}. {student.get('full_name') or student.get('name') or student.get('identity')}", f"- الهوية: {student.get('identity', '')}", f"- مفتاح الدخول: {student.get('access_key', '')}", f"- الحالة: {student.get('account_status', 'pending')}", f"- الصف: {student.get('grade', '')}", f"- البلد: {student.get('country', '')}", f"- البريد: {student.get('email', '')}", f"- الهاتف: {student.get('phone', '')}", f"- الأجهزة الموثوقة: {student.get('authorized_devices_count', 0)}", "", ] ) return "\n".join(lines).strip() def format_admin_dashboard(self, admin_key: Any = "") -> str: if not self.verify_admin_key(admin_key): return "مفتاح الإدارة غير صحيح أو غير موجود." counts_result = self.student_status_counts(admin_key=admin_key) counts = counts_result.get("counts", {}) if isinstance(counts_result, dict) else {} text = "# مكتب الإدارة\n\n" text += "## تسجيل الطلاب\n\n" text += f"- بانتظار المراجعة: {counts.get('pending', 0)}\n" text += f"- مفعّلون: {counts.get('active', 0)}\n" text += f"- موقوفون: {counts.get('suspended', 0)}\n" text += f"- مرفوضون: {counts.get('rejected', 0)}\n\n" text += "---\n\n" text += "## التوجيه العام\n\n" text += self.load_general_instruction() text += "\n\n---\n\n" text += "## المنهج والمواد\n\n" text += str(self.curriculum_summary()) text += "\n\n---\n\n" text += "## الاستراتيجيات\n\n" text += str(self.strategy_summary()) text += "\n\n---\n\n" text += "## اللغات\n\n" text += str(self.language_summary()) return text.strip() def summary(self) -> Dict[str, Any]: status_result = self.student_status_counts( admin_key=self.admin_key if self.require_admin_key else "" ) counts = status_result.get("counts", {}) if status_result.get("success") else {} return { "success": True, "ready": True, "version": self.version, "schema_version": self.schema_version, "role": "full_admin_control_plane", "admin_key_required": self.require_admin_key, "admin_key_configured": bool(self.admin_key), "permissions": list(FULL_PERMISSIONS), "student_registration_control": { "available": self.student_manager is not None, "allowed_statuses": sorted(ACCOUNT_STATUSES), "status_counts": counts, "supports_single_change": True, "supports_bulk_change": True, "supports_activate": True, "supports_suspend": True, "supports_reject": True, "supports_return_to_pending": True, }, "student_management": { "profile_update": True, "access_key_reset": True, "parent_password_reset": True, "device_management": True, "explicit_delete": True, }, "services": { "student_manager": self.student_manager is not None, "device_identity_manager": self.device_manager is not None, "badge_progress_manager": self.badge_progress_manager is not None, "report_manager": self.report_manager is not None, "ai_task_queue_manager": self.queue_manager is not None, "language_manager": self.language_manager is not None, }, "module_functions": [ "list_students", "list_pending_registrations", "get_student_record", "change_student_registration_status", "activate_student_registration", "suspend_student_registration", "reject_student_registration", "return_student_to_pending", "bulk_change_student_status", "update_student_profile", "reset_student_access_key", "reset_parent_password", "list_student_devices", "remove_student_device", "clear_student_devices", "delete_student_account", "load_audit_log", "system_snapshot", "summary", ], "message": "AdminManager جاهز بالصلاحيات الإدارية الكاملة، بما فيها تغيير حالة تسجيل الطلاب.", } def is_ready(self) -> bool: return True # ========================================================= # Singleton ودوال توافق مباشرة # ========================================================= _MANAGER: Optional[AdminManager] = None _MANAGER_LOCK = threading.Lock() def get_manager(*args: Any, **kwargs: Any) -> AdminManager: global _MANAGER if _MANAGER is None: with _MANAGER_LOCK: if _MANAGER is None: _MANAGER = AdminManager(*args, **kwargs) return _MANAGER def change_student_registration_status(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().change_student_registration_status(*args, **kwargs) def activate_student_registration(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().activate_student_registration(*args, **kwargs) def suspend_student_registration(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().suspend_student_registration(*args, **kwargs) def reject_student_registration(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().reject_student_registration(*args, **kwargs) def return_student_to_pending(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().return_student_to_pending(*args, **kwargs) def list_students(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().list_students(*args, **kwargs) def list_pending_registrations(*args: Any, **kwargs: Any) -> Dict[str, Any]: return get_manager().list_pending_registrations(*args, **kwargs) def summary() -> Dict[str, Any]: return get_manager().summary() AdminOfficeManager = AdminManager SystemAdminManager = AdminManager