memory.py

# memory.py
# -*- coding: utf-8 -*-
"""
ذاكرة المستخدم/العامة + قاعدة المعرفة في ملف واحد، بدون أي استيراد عكسي (circular import).
يوفّر:
- load_memory(user_id) / save_memory(user_id, data)
- load_global_memory() / save_global_memory(data)
- load_knowledge_base() / save_knowledge_base(data)
- learn_from_unknown(prompt) / update_knowledge_base(prompt, answer)
"""

from __future__ import annotations
import json
import os
import tempfile
import threading
from typing import Any, Dict

# ========= مسارات التخزين =========
BASE_DIR = os.path.abspath(os.path.dirname(__file__))

# مجلد بيانات الذاكرة
DATA_DIR = os.path.join(BASE_DIR, "data")
os.makedirs(DATA_DIR, exist_ok=True)

# ملفات الذاكرة
GLOBAL_MEMORY_FILE = os.path.join(DATA_DIR, "global_memory.json")
USER_MEMORY_PREFIX = "memory_"  # memory_<user_id>.json

# ملف قاعدة المعرفة (متوافق مع نسختك السابقة)
knowledge_base_path = os.path.join(BASE_DIR, "knowledge_base.json")

# قفل للتزامن
_lock = threading.RLock()

# ========= أدوات JSON آمنة =========
def _safe_write_json(path: str, data: Dict[str, Any]) -> None:
    """
    كتابة ذرّية لضمان عدم تلف الملف حتى مع انقطاع مفاجئ.
    """
    with _lock:
        os.makedirs(os.path.dirname(path), exist_ok=True)
        fd, tmp_path = tempfile.mkstemp(prefix=".tmp_", dir=os.path.dirname(path))
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp_path, path)
        finally:
            if os.path.exists(tmp_path):
                try:
                    os.remove(tmp_path)
                except OSError:
                    pass

def _read_json(path: str) -> Dict[str, Any]:
    """
    قراءة JSON بأمان. يعيد dict فارغ عند عدم وجود الملف أو في حال تلفه.
    """
    with _lock:
        if not os.path.exists(path):
            return {}
        try:
            with open(path, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception:
            return {}

def _user_file(user_id: str) -> str:
    """
    تكوين اسم ملف آمن للمستخدم.
    """
    safe_id = "".join(ch for ch in str(user_id) if ch.isalnum() or ch in ("-", "_")) or "default"
    return os.path.join(DATA_DIR, f"{USER_MEMORY_PREFIX}{safe_id}.json")

# ========= واجهة الذاكرة (لـ learner.py) =========
def load_memory(user_id: str) -> Dict[str, Any]:
    """
    قراءة ذاكرة مستخدم (dict).
    """
    return _read_json(_user_file(user_id))

def save_memory(user_id: str, data: Dict[str, Any]) -> None:
    """
    حفظ ذاكرة مستخدم (dict).
    """
    if not isinstance(data, dict):
        raise TypeError("save_memory expects a dict")
    _safe_write_json(_user_file(user_id), data)

def load_global_memory() -> Dict[str, Any]:
    """
    قراءة الذاكرة العامة (dict).
    """
    return _read_json(GLOBAL_MEMORY_FILE)

def save_global_memory(data: Dict[str, Any]) -> None:
    """
    حفظ الذاكرة العامة (dict).
    """
    if not isinstance(data, dict):
        raise TypeError("save_global_memory expects a dict")
    _safe_write_json(GLOBAL_MEMORY_FILE, data)

# ========= واجهة قاعدة المعرفة (متوافقة مع نسختك السابقة) =========
def load_knowledge_base() -> Dict[str, Any]:
    return _read_json(knowledge_base_path)

def save_knowledge_base(data: Dict[str, Any]) -> None:
    if not isinstance(data, dict):
        raise TypeError("save_knowledge_base expects a dict")
    _safe_write_json(knowledge_base_path, data)

def learn_from_unknown(prompt: str):
    """
    نفس السلوك السابق: يطبع رسالة ويطلب إجابة ليتم تعلمها.
    """
    print("نورا: لا أملك إجابة لهذا السؤال حاليًا. سأبحث عن إجابة وأتعلم.")
    answer = input("يرجى تزويدي بالإجابة المناسبة لأتعلمها: ")
    if answer and answer.strip():
        return answer
    return None

def update_knowledge_base(prompt: str, answer: str) -> None:
    kb = load_knowledge_base()
    kb[prompt] = answer
    save_knowledge_base(kb)

# ========= تصدير الأسماء =========
__all__ = [
    "load_memory", "save_memory", "load_global_memory", "save_global_memory",
    "load_knowledge_base", "save_knowledge_base",
    "learn_from_unknown", "update_knowledge_base",
]

# ========= اختبار سريع =========
if __name__ == "__main__":
    uid = "osama"

    # اختبار ذاكرة المستخدم/العامة
    u = load_memory(uid)
    u["counter"] = u.get("counter", 0) + 1
    save_memory(uid, u)

    g = load_global_memory()
    g["last_user"] = uid
    save_global_memory(g)

    print("User memory:", load_memory(uid))
    print("Global memory:", load_global_memory())

    # اختبار قاعدة المعرفة
    update_knowledge_base("ما هي نورا؟", "مساعد ذكي")
    print("KB size:", len(load_knowledge_base()))