""" 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()