Spaces:
Sleeping
Sleeping
| """ | |
| 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) ---- | |
| def get_employees(self) -> pd.DataFrame: ... | |
| def get_points(self) -> pd.DataFrame: ... | |
| def get_badges(self) -> pd.DataFrame: ... | |
| def get_rewards(self) -> pd.DataFrame: ... | |
| def get_reward_requests(self) -> pd.DataFrame: ... | |
| def get_notifications(self) -> pd.DataFrame: ... | |
| # ---- كتابة (INSERT / UPDATE) ---- | |
| def record_points(self, employee_id: str, action: str, points: int, | |
| created_at: str | None = None) -> None: ... | |
| def award_badge(self, employee_id: str, badge_name: str, | |
| awarded_at: str | None = None) -> None: ... | |
| def send_notification(self, employee_id: str, message: str, | |
| channel: str = "slack") -> None: ... | |
| def create_reward_request(self, employee_id: str, reward_id: str, | |
| reward_name: str, points_cost: int) -> None: ... | |
| def resolve_reward_request(self, request_id: int, status: str) -> None: ... | |
| def get_synced_task_ids(self) -> set[str]: ... | |
| 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) | |
| 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() | |