data_layer.py

"""
Data Access Layer — Repository pattern.

هذه هي الطبقة الوحيدة في النظام التي تعرف أن البيانات مخزّنة في ملفات CSV.
باقي الكود (logic.py + الصفحات) يتعامل مع الـ Repository interface فقط، لذلك
الانتقال إلى Supabase/PostgreSQL لاحقًا = كتابة SupabaseRepository بنفس الدوال،
بدون أي تعديل في منطق الحسابات أو الواجهة.

كل ملف CSV في data/ يطابق جدولًا في PostgreSQL بنفس أسماء الأعمدة حرفيًا.

⚠️ ملاحظة النشر على Hugging Face Spaces:
الكتابة في ملفات CSV هنا تتم على قرص مؤقت (ephemeral filesystem). أي تعديل
(Sync / إضافة نقاط / اعتماد مكافأة) يظهر فورًا ويبقى طوال تشغيل الـ Space،
لكنه يُفقد عند إعادة تشغيل أو إعادة بناء الـ Space. هذا مقبول لنسخة العرض
(demo only) — في الإنتاج تحل Supabase محل هذه الطبقة وتصبح الكتابة دائمة.
"""

from __future__ import annotations

import json
import os
from abc import ABC, abstractmethod
from datetime import datetime

import pandas as pd

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(BASE_DIR, "data")
# حالة الـ Sync تُحفظ خارج data/ عمدًا: مجلد data/ يمثل schema قاعدة البيانات
# بالضبط ويجب أن يبقى قابلًا للترحيل بـ COPY FROM CSV دون أعمدة أو ملفات دخيلة.
# في الإنتاج يصبح هذا الملف جدول synced_tasks أو UNIQUE constraint على المصدر.
SYNC_STATE_FILE = os.path.join(BASE_DIR, "sync_state.json")


class Repository(ABC):
    """الواجهة التي يعتمد عليها باقي النظام. أي مصدر بيانات ينفّذها."""

    # ---- قراءة (SELECT) ----
    @abstractmethod
    def get_employees(self) -> pd.DataFrame: ...

    @abstractmethod
    def get_points(self) -> pd.DataFrame: ...

    @abstractmethod
    def get_badges(self) -> pd.DataFrame: ...

    @abstractmethod
    def get_rewards(self) -> pd.DataFrame: ...

    @abstractmethod
    def get_reward_requests(self) -> pd.DataFrame: ...

    @abstractmethod
    def get_notifications(self) -> pd.DataFrame: ...

    # ---- كتابة (INSERT / UPDATE) ----
    @abstractmethod
    def record_points(self, employee_id: str, action: str, points: int,
                      created_at: str | None = None) -> None: ...

    @abstractmethod
    def award_badge(self, employee_id: str, badge_name: str,
                    awarded_at: str | None = None) -> None: ...

    @abstractmethod
    def send_notification(self, employee_id: str, message: str,
                          channel: str = "slack") -> None: ...

    @abstractmethod
    def create_reward_request(self, employee_id: str, reward_id: str,
                              reward_name: str, points_cost: int) -> None: ...

    @abstractmethod
    def resolve_reward_request(self, request_id: int, status: str) -> None: ...

    @abstractmethod
    def get_synced_task_ids(self) -> set[str]: ...

    @abstractmethod
    def mark_task_synced(self, task_id: str) -> None: ...


def _now() -> str:
    """ISO 8601 بنفس صيغة الداتا الموجودة: 2026-07-01T09:15:00"""
    return datetime.now().isoformat(timespec="seconds")


class CSVRepository(Repository):
    """
    تنفيذ الـ Repository فوق ملفات CSV باستخدام pandas.

    ملاحظة مهمة: الطوابع الزمنية تُقرأ كنصوص (بدون parse_dates) عمدًا، حتى تعود
    إلى الملف بنفس صيغتها ISO بالضبط عند الكتابة. التحويل إلى datetime يتم في
    logic.py وقت الحساب فقط.
    """

    def __init__(self, data_dir: str = DATA_DIR):
        self.data_dir = data_dir

    def _path(self, table: str) -> str:
        return os.path.join(self.data_dir, f"{table}.csv")

    def _read(self, table: str, keep_default_na: bool = True) -> pd.DataFrame:
        return pd.read_csv(self._path(table), keep_default_na=keep_default_na)

    def _write(self, table: str, df: pd.DataFrame) -> None:
        # index=False يحافظ على أسماء وترتيب الأعمدة كما هي — شرط الترحيل لـ PostgreSQL
        df.to_csv(self._path(table), index=False)

    @staticmethod
    def _next_id(df: pd.DataFrame) -> int:
        """محاكاة SERIAL PRIMARY KEY."""
        if df.empty or "id" not in df.columns:
            return 1
        return int(pd.to_numeric(df["id"], errors="coerce").max()) + 1

    # ---------------- قراءة ----------------

    def get_employees(self) -> pd.DataFrame:
        # keep_default_na=False ضروري هنا: قيمة current_level النصية "None" مدرجة
        # ضمن قيم NA الافتراضية في pandas، فتُقرأ NaN بدل نص. بدون هذا المعامل
        # ينكسر حساب ترقية المستوى ويُكتب العمود فارغًا في الملف عند أي تحديث.
        return self._read("employees", keep_default_na=False)

    def get_points(self) -> pd.DataFrame:
        return self._read("employee_points")

    def get_badges(self) -> pd.DataFrame:
        return self._read("employee_badges")

    def get_rewards(self) -> pd.DataFrame:
        return self._read("rewards")

    def get_reward_requests(self) -> pd.DataFrame:
        return self._read("reward_requests")

    def get_notifications(self) -> pd.DataFrame:
        return self._read("notifications")

    # ---------------- كتابة ----------------

    def record_points(self, employee_id: str, action: str, points: int,
                      created_at: str | None = None) -> None:
        """
        تسجيل حركة نقاط + تحديث رصيد الموظف.

        current_points عمود denormalized (القسم 9 من الـ spec) — أي حركة يجب أن
        تحدّث السجل والرصيد معًا. في PostgreSQL هذه transaction واحدة أو trigger.
        """
        df = self.get_points()
        row = pd.DataFrame([{
            "id": self._next_id(df),
            "employee_id": employee_id,
            "action": action,
            "points": int(points),
            "created_at": created_at or _now(),
        }])
        self._write("employee_points", pd.concat([df, row], ignore_index=True))
        self._adjust_balance(employee_id, int(points))

    def _adjust_balance(self, employee_id: str, delta: int) -> None:
        """
        تعديل current_points. المستوى (current_level) يرتفع فقط ولا ينزل:
        الدروع إنجاز دائم، وصرف النقاط في المتجر لا يسحب درعًا سبق كسبه.
        """
        emp = self.get_employees()
        mask = emp["employee_id"] == employee_id
        if not mask.any():
            raise KeyError(f"موظف غير موجود: {employee_id}")

        new_balance = int(emp.loc[mask, "current_points"].iloc[0]) + delta
        emp.loc[mask, "current_points"] = new_balance

        # استيراد محلي لتفادي دائرة الاستيراد (logic تستورد data_layer)
        from logic import LEVEL_ORDER, level_from_points

        current = str(emp.loc[mask, "current_level"].iloc[0])
        earned = level_from_points(new_balance)
        if LEVEL_ORDER.index(earned) > LEVEL_ORDER.index(current):
            emp.loc[mask, "current_level"] = earned

        self._write("employees", emp)

    def award_badge(self, employee_id: str, badge_name: str,
                    awarded_at: str | None = None) -> None:
        df = self.get_badges()
        row = pd.DataFrame([{
            "id": self._next_id(df),
            "employee_id": employee_id,
            "badge_name": badge_name,
            "awarded_at": awarded_at or _now(),
        }])
        self._write("employee_badges", pd.concat([df, row], ignore_index=True))

    def send_notification(self, employee_id: str, message: str,
                          channel: str = "slack") -> None:
        """محاكاة Slack/WhatsApp API — في الإنتاج تصبح استدعاء HTTP فعلي."""
        df = self.get_notifications()
        row = pd.DataFrame([{
            "id": self._next_id(df),
            "employee_id": employee_id,
            "message": message,
            "channel": channel,
            "sent_at": _now(),
        }])
        self._write("notifications", pd.concat([df, row], ignore_index=True))

    def create_reward_request(self, employee_id: str, reward_id: str,
                              reward_name: str, points_cost: int) -> None:
        df = self.get_reward_requests()
        row = pd.DataFrame([{
            "id": self._next_id(df),
            "employee_id": employee_id,
            "reward_id": reward_id,
            "reward_name": reward_name,
            "points_cost": int(points_cost),
            "status": "pending",
            "requested_at": _now(),
            "resolved_at": "",
        }])
        self._write("reward_requests", pd.concat([df, row], ignore_index=True))

    def resolve_reward_request(self, request_id: int, status: str) -> None:
        df = self.get_reward_requests()
        mask = df["id"] == request_id
        if not mask.any():
            raise KeyError(f"طلب غير موجود: {request_id}")
        df.loc[mask, "status"] = status
        df.loc[mask, "resolved_at"] = _now()
        self._write("reward_requests", df)

    # ---------------- حالة الـ Sync ----------------

    def get_synced_task_ids(self) -> set[str]:
        if not os.path.exists(SYNC_STATE_FILE):
            return set()
        try:
            with open(SYNC_STATE_FILE, encoding="utf-8") as f:
                return set(json.load(f).get("synced_task_ids", []))
        except (json.JSONDecodeError, OSError):
            return set()

    def mark_task_synced(self, task_id: str) -> None:
        ids = self.get_synced_task_ids()
        ids.add(task_id)
        with open(SYNC_STATE_FILE, "w", encoding="utf-8") as f:
            json.dump({"synced_task_ids": sorted(ids)}, f, indent=2)

    def reset_sync_state(self) -> None:
        """للعرض فقط: يسمح بإعادة تشغيل الـ Sync من الصفر."""
        if os.path.exists(SYNC_STATE_FILE):
            os.remove(SYNC_STATE_FILE)


def get_repository() -> Repository:
    """
    نقطة التبديل الوحيدة عند الانتقال للإنتاج:
        return SupabaseRepository(url=..., key=...)
    """
    return CSVRepository()