Skip to content

C
ClubPython
Commit e666e669 authored by Administrator's avatar Administrator
Browse files
Options

Update 44 files via Son of Anton

parent 0e85c103
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 3308 additions and 39 deletions
backend/apps/forms_engine/admin.py
View file @ e666e669
# Admin for forms_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Forms Engine Admin
"""
from django.contrib import admin
from django.db.models import JSONField
from django_json_widget.widgets import JSONEditorWidget
from simple_history.admin import SimpleHistoryAdmin
from .models import FormSchema, FormSubmission
@admin.register(FormSchema)
class FormSchemaAdmin(SimpleHistoryAdmin):
list_display = ['code', 'name_ar', 'version', 'is_active', 'validity_days', 'fee_service_code']
list_filter = ['is_active', 'code']
search_fields = ['code', 'name_ar']
readonly_fields = ['created_at', 'updated_at']
formfield_overrides = {
JSONField: {'widget': JSONEditorWidget},
}
@admin.register(FormSubmission)
class FormSubmissionAdmin(SimpleHistoryAdmin):
list_display = [
'form_number', 'schema', 'status', 'member',
'submitted_by', 'submitted_at', 'expires_at',
]
list_filter = ['status', 'schema__code']
search_fields = ['form_number', 'member__full_name_ar', 'member__membership_number']
raw_id_fields = ['member', 'submitted_by', 'reviewed_by']
readonly_fields = ['submitted_at']
formfield_overrides = {
JSONField: {'widget': JSONEditorWidget},
}
\ No newline at end of file
backend/apps/forms_engine/management/commands/seed_forms.py 0 → 100644
View file @ e666e669
from django.core.management.base import BaseCommand
from apps.forms_engine.seed import seed_forms
class Command(BaseCommand):
help = 'Seed all form schemas'
def handle(self, *args, **options):
created = seed_forms()
self.stdout.write(self.style.SUCCESS(f'Forms seeded: {created} schemas created'))
\ No newline at end of file
backend/apps/forms_engine/models.py
View file @ e666e669
# Models for forms_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Forms Engine Models
=====================================
Dynamic form schemas stored as JSON. Form submissions tracked.
"""
from django.db import models
from django.conf import settings
from simple_history.models import HistoricalRecords
import auditlog
class FormSchema(models.Model):
code = models.CharField(max_length=50, verbose_name="الكود")
name_ar = models.CharField(max_length=200, verbose_name="الاسم بالعربي")
name_en = models.CharField(max_length=200, blank=True, default="", verbose_name="الاسم بالانجليزي")
schema = models.JSONField(verbose_name="مخطط الاستمارة")
ui_schema = models.JSONField(null=True, blank=True, verbose_name="مخطط واجهة المستخدم")
version = models.PositiveIntegerField(default=1, verbose_name="الإصدار")
fee_service_code = models.CharField(
max_length=50, blank=True, default="", verbose_name="كود رسوم الخدمة",
)
validity_days = models.PositiveIntegerField(default=15, verbose_name="صلاحية بالأيام")
is_active = models.BooleanField(default=True, verbose_name="نشط")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
updated_at = models.DateTimeField(auto_now=True, verbose_name="تاريخ التعديل")
history = HistoricalRecords()
class Meta:
verbose_name = "نموذج استمارة"
verbose_name_plural = "نماذج الاستمارات"
unique_together = [('code', 'version')]
ordering = ['code', '-version']
def __str__(self):
return f"{self.code} v{self.version} — {self.name_ar}"
class FormSubmission(models.Model):
STATUS_CHOICES = [
('DRAFT', 'مسودة'),
('SUBMITTED', 'مقدم'),
('UNDER_REVIEW', 'قيد المراجعة'),
('APPROVED', 'معتمد'),
('REJECTED', 'مرفوض'),
('EXPIRED', 'منتهي الصلاحية'),
('COMPLETED', 'مكتمل'),
]
schema = models.ForeignKey(
FormSchema, on_delete=models.PROTECT,
related_name='submissions', verbose_name="النموذج",
)
form_number = models.CharField(
max_length=20, unique=True, db_index=True, verbose_name="رقم الاستمارة",
)
data = models.JSONField(verbose_name="بيانات الاستمارة")
member = models.ForeignKey(
'members.Member', null=True, blank=True,
on_delete=models.SET_NULL, related_name='form_submissions',
verbose_name="العضو",
)
status = models.CharField(
max_length=20, choices=STATUS_CHOICES, default='DRAFT',
verbose_name="الحالة",
)
submitted_by = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, related_name='form_submissions',
verbose_name="مقدم بواسطة",
)
reviewed_by = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, related_name='form_reviews',
verbose_name="مراجع بواسطة",
)
fee_receipt_number = models.CharField(
max_length=50, blank=True, default="", verbose_name="رقم إيصال الرسوم",
)
submitted_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ التقديم")
approved_at = models.DateTimeField(null=True, blank=True, verbose_name="تاريخ الاعتماد")
expires_at = models.DateTimeField(null=True, blank=True, verbose_name="ينتهي في")
notes = models.TextField(blank=True, default="", verbose_name="ملاحظات")
history = HistoricalRecords()
class Meta:
verbose_name = "استمارة مقدمة"
verbose_name_plural = "الاستمارات المقدمة"
ordering = ['-submitted_at']
def __str__(self):
return f"{self.form_number} — {self.schema.code} — {self.get_status_display()}"
auditlog.register(FormSchema)
auditlog.register(FormSubmission)
\ No newline at end of file
backend/apps/forms_engine/seed.py
View file @ e666e669
# Management command: seed_forms — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Forms Engine Seed Data
All 8 form schemas as JSON Schema definitions.
"""
def _base_schema(title, required_fields, properties):
"""Build a JSON Schema object."""
return {
'type': 'object',
'title': title,
'required': required_fields,
'properties': properties,
}
FORM_SCHEMAS = [
{
'code': 'NEW_MEMBERSHIP',
'name_ar': 'استمارة عضوية جديدة',
'name_en': 'New Membership Application',
'fee_service_code': 'SVC_NEW_FORM',
'validity_days': 15,
'schema': _base_schema('استمارة عضوية جديدة', ['full_name_ar', 'national_id', 'phone_mobile', 'qualification'], {
'full_name_ar': {'type': 'string', 'minLength': 10, 'maxLength': 200, 'title': 'الاسم بالكامل (عربي)'},
'full_name_en': {'type': 'string', 'maxLength': 200, 'title': 'الاسم بالكامل (إنجليزي)'},
'national_id': {'type': 'string', 'pattern': '^[23]\\d{13}$', 'title': 'الرقم القومي'},
'passport_number': {'type': 'string', 'title': 'رقم جواز السفر'},
'id_type': {'type': 'string', 'enum': ['NATIONAL_ID', 'PASSPORT', 'MILITARY_ID'], 'title': 'نوع إثبات الشخصية'},
'qualification': {'type': 'string', 'enum': ['HIGH', 'MEDIUM', 'NONE'], 'title': 'المؤهل الدراسي'},
'religion': {'type': 'string', 'enum': ['MUSLIM', 'CHRISTIAN', 'OTHER'], 'title': 'الديانة'},
'marital_status': {'type': 'string', 'enum': ['SINGLE', 'MARRIED', 'DIVORCED', 'WIDOWED'], 'title': 'الحالة الاجتماعية'},
'phone_mobile': {'type': 'string', 'pattern': '^01\\d{9}$', 'title': 'رقم المحمول'},
'phone_home': {'type': 'string', 'title': 'تليفون المنزل'},
'email': {'type': 'string', 'format': 'email', 'title': 'البريد الإلكتروني'},
'emergency_name': {'type': 'string', 'title': 'شخص للطوارئ'},
'emergency_phone': {'type': 'string', 'title': 'هاتف الطوارئ'},
'residence_type': {'type': 'string', 'enum': ['RENTED', 'OWNED', 'OTHER'], 'title': 'نوع السكن'},
'residence_address': {'type': 'string', 'title': 'عنوان السكن'},
'area': {'type': 'string', 'title': 'المنطقة'},
'governorate': {'type': 'string', 'title': 'المحافظة'},
'employment_type': {'type': 'string', 'enum': ['EMPLOYED', 'SELF_EMPLOYED', 'PROFESSIONS', 'OTHER'], 'title': 'نوع التوظيف'},
'occupation': {'type': 'string', 'title': 'المهنة'},
'referral_source': {'type': 'array', 'items': {'type': 'string'}, 'title': 'كيف عرفت النادي'},
'spouses': {'type': 'array', 'items': {'type': 'object'}, 'title': 'الزوجات'},
'children': {'type': 'array', 'items': {'type': 'object'}, 'title': 'الأبناء'},
'notes': {'type': 'string', 'title': 'ملاحظات'},
}),
},
{
'code': 'TRANSFER_SEPARATION',
'name_ar': 'استمارة تحويل / فصل',
'name_en': 'Transfer / Separation Form',
'fee_service_code': 'SVC_TRANSFER_FORM',
'validity_days': 15,
'schema': _base_schema('استمارة تحويل / فصل', ['transfer_type', 'full_name_ar'], {
'transfer_type': {'type': 'string', 'enum': ['CHILD_SEPARATION', 'CHILD_MANDATORY_25', 'DIVORCE', 'DEATH', 'WAIVER', 'SPORTS_CONVERSION', 'CROSS_BRANCH'], 'title': 'نوع التحويل'},
'full_name_ar': {'type': 'string', 'title': 'الاسم بالكامل'},
'national_id': {'type': 'string', 'title': 'الرقم القومي'},
'qualification': {'type': 'string', 'enum': ['HIGH', 'MEDIUM', 'NONE'], 'title': 'المؤهل'},
'notes': {'type': 'string', 'title': 'ملاحظات'},
}),
},
{
'code': 'ADDITION_CHILD',
'name_ar': 'استمارة ضم أبناء',
'name_en': 'Children Addition Form',
'fee_service_code': 'SVC_ADDITION_FORM',
'validity_days': 15,
'schema': _base_schema('استمارة ضم أبناء', ['children'], {
'children': {'type': 'array', 'items': {'type': 'object', 'properties': {
'full_name_ar': {'type': 'string'}, 'national_id': {'type': 'string'},
'date_of_birth': {'type': 'string', 'format': 'date'}, 'gender': {'type': 'string', 'enum': ['MALE', 'FEMALE']},
'relationship': {'type': 'string', 'enum': ['SON', 'DAUGHTER', 'STEPCHILD']},
}}, 'maxItems': 3, 'title': 'الأبناء'},
'notes': {'type': 'string', 'title': 'ملاحظات'},
}),
},
{
'code': 'ADDITION_SPOUSE',
'name_ar': 'استمارة ضم زوج / زوجة',
'name_en': 'Spouse Addition Form',
'fee_service_code': 'SVC_ADDITION_FORM',
'validity_days': 15,
'schema': _base_schema('استمارة ضم زوج/زوجة', ['full_name_ar', 'marriage_date'], {
'full_name_ar': {'type': 'string', 'title': 'الاسم'},
'national_id': {'type': 'string', 'title': 'الرقم القومي'},
'marriage_date': {'type': 'string', 'format': 'date', 'title': 'تاريخ الزواج'},
'nationality': {'type': 'string', 'title': 'الجنسية'},
'qualification': {'type': 'string', 'title': 'المؤهل'},
'mobile': {'type': 'string', 'title': 'المحمول'},
}),
},
{
'code': 'ADDITION_TEMPORARY',
'name_ar': 'استمارة ضم عضو مؤقت',
'name_en': 'Temporary Member Addition Form',
'fee_service_code': 'SVC_ADDITION_FORM',
'validity_days': 15,
'schema': _base_schema('استمارة ضم عضو مؤقت', ['full_name_ar', 'category'], {
'full_name_ar': {'type': 'string', 'title': 'الاسم'},
'national_id': {'type': 'string', 'title': 'الرقم القومي'},
'category': {'type': 'string', 'enum': ['PARENT', 'SPECIAL_NEEDS_CHILD', 'UNMARRIED_DAUGHTER', 'SISTER', 'STEPCHILD', 'ORPHAN', 'DISABLED_SIBLING', 'NANNY'], 'title': 'الفئة'},
'has_championship': {'type': 'boolean', 'title': 'حاصل على بطولات جمهورية'},
}),
},
{
'code': 'SEASONAL_MEMBERSHIP',
'name_ar': 'استمارة عضوية موسمية',
'name_en': 'Seasonal Membership Form',
'fee_service_code': 'SVC_SEASONAL',
'validity_days': 180,
'schema': _base_schema('استمارة عضوية موسمية', ['full_name_ar', 'phone'], {
'full_name_ar': {'type': 'string', 'title': 'الاسم'},
'national_id': {'type': 'string', 'title': 'الرقم القومي'},
'qualification': {'type': 'string', 'title': 'المؤهل'},
'phone': {'type': 'string', 'title': 'التليفون'},
'address': {'type': 'string', 'title': 'العنوان'},
}),
},
{
'code': 'LOST_CARNET',
'name_ar': 'استمارة بدل فاقد كارنيه',
'name_en': 'Lost Carnet Replacement Form',
'fee_service_code': 'SVC_CARNET_REPLACE',
'validity_days': 7,
'schema': _base_schema('استمارة بدل فاقد كارنيه', ['declaration'], {
'declaration': {'type': 'boolean', 'title': 'إقرار بالفقد والالتزام'},
}),
},
{
'code': 'DATA_UPDATE',
'name_ar': 'استمارة تعديل بيانات',
'name_en': 'Data Update Form',
'fee_service_code': '',
'validity_days': 30,
'schema': _base_schema('استمارة تعديل بيانات', [], {
'fields_to_update': {'type': 'object', 'title': 'الحقول المراد تعديلها'},
'reason': {'type': 'string', 'title': 'سبب التعديل'},
}),
},
]
def seed_forms():
from .models import FormSchema
created = 0
for form_data in FORM_SCHEMAS:
_, was_created = FormSchema.objects.update_or_create(
code=form_data['code'],
version=1,
defaults={
'name_ar': form_data['name_ar'],
'name_en': form_data.get('name_en', ''),
'schema': form_data['schema'],
'fee_service_code': form_data.get('fee_service_code', ''),
'validity_days': form_data.get('validity_days', 15),
'is_active': True,
},
)
if was_created:
created += 1
return created
\ No newline at end of file
backend/apps/forms_engine/serializers.py
View file @ e666e669
# Serializers for forms_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Forms Engine Serializers
"""
from rest_framework import serializers
from django.utils import timezone
from datetime import timedelta
from .models import FormSchema, FormSubmission
from .validator import FormValidator
class FormSchemaSerializer(serializers.ModelSerializer):
class Meta:
model = FormSchema
fields = [
'id', 'code', 'name_ar', 'name_en', 'schema', 'ui_schema',
'version', 'fee_service_code', 'validity_days', 'is_active',
'created_at', 'updated_at',
]
read_only_fields = ['created_at', 'updated_at']
class FormSchemaListSerializer(serializers.ModelSerializer):
class Meta:
model = FormSchema
fields = ['id', 'code', 'name_ar', 'version', 'is_active', 'fee_service_code', 'validity_days']
class FormSubmissionSerializer(serializers.ModelSerializer):
schema_code = serializers.CharField(source='schema.code', read_only=True)
schema_name = serializers.CharField(source='schema.name_ar', read_only=True)
class Meta:
model = FormSubmission
fields = [
'id', 'schema', 'schema_code', 'schema_name', 'form_number',
'data', 'member', 'status', 'submitted_by', 'reviewed_by',
'fee_receipt_number', 'submitted_at', 'approved_at',
'expires_at', 'notes',
]
read_only_fields = ['submitted_at', 'form_number']
def validate_data(self, value):
schema_id = self.initial_data.get('schema')
if schema_id:
try:
form_schema = FormSchema.objects.get(id=schema_id)
result = FormValidator.validate_submission(form_schema, value)
if not result['valid']:
raise serializers.ValidationError(result['errors'])
except FormSchema.DoesNotExist:
raise serializers.ValidationError('نموذج الاستمارة غير موجود')
return value
def create(self, validated_data):
# Auto-generate form number
from django.db.models import Max
max_num = FormSubmission.objects.aggregate(
Max('form_number')
)['form_number__max']
if max_num and max_num.isdigit():
next_num = str(int(max_num) + 1).zfill(8)
else:
next_num = '00000001'
validated_data['form_number'] = next_num
# Set expiry based on schema validity_days
schema = validated_data.get('schema')
if schema:
validated_data['expires_at'] = (
timezone.now() + timedelta(days=schema.validity_days)
)
return super().create(validated_data)
\ No newline at end of file
backend/apps/forms_engine/tests.py
View file @ e666e669
# Tests for forms_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Forms Engine Tests
"""
from django.test import TestCase
from .models import FormSchema, FormSubmission
from .validator import FormValidator
from .seed import seed_forms
class FormSeedTest(TestCase):
def test_seed_creates_schemas(self):
created = seed_forms()
self.assertEqual(created, 8)
self.assertTrue(FormSchema.objects.filter(code='NEW_MEMBERSHIP').exists())
class FormValidatorTest(TestCase):
@classmethod
def setUpTestData(cls):
seed_forms()
cls.schema = FormSchema.objects.get(code='NEW_MEMBERSHIP', version=1)
def test_valid_data(self):
data = {
'full_name_ar': 'أحمد محمد أحمد علي',
'national_id': '29001011234567',
'phone_mobile': '01012345678',
'qualification': 'HIGH',
}
result = FormValidator.validate_submission(self.schema, data)
self.assertTrue(result['valid'])
def test_missing_required_field(self):
data = {'full_name_ar': 'أحمد محمد'}
result = FormValidator.validate_submission(self.schema, data)
self.assertFalse(result['valid'])
def test_invalid_national_id_pattern(self):
data = {
'full_name_ar': 'أحمد محمد أحمد علي',
'national_id': '12345',
'phone_mobile': '01012345678',
'qualification': 'HIGH',
}
result = FormValidator.validate_submission(self.schema, data)
self.assertFalse(result['valid'])
\ No newline at end of file
backend/apps/forms_engine/urls.py
View file @ e666e669
urlpatterns = [] """
\ No newline at end of file THE CLUB ERP — Forms Engine URLs
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views
router = DefaultRouter()
router.register(r'schemas', views.FormSchemaViewSet, basename='formschema')
router.register(r'submissions', views.FormSubmissionViewSet, basename='formsubmission')
urlpatterns = [
path('', include(router.urls)),
]
\ No newline at end of file
backend/apps/forms_engine/validator.py
View file @ e666e669
# Form validator — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Form Schema Validator
======================================
Validates submitted form data against JSON schema.
"""
import jsonschema
from jsonschema import ValidationError as JSONValidationError
class FormValidator:
"""Validates form submission data against its JSON schema."""
@staticmethod
def validate(schema: dict, data: dict) -> dict:
"""
Validate data against schema.
Returns {'valid': True/False, 'errors': [list of error messages]}
"""
errors = []
try:
jsonschema.validate(instance=data, schema=schema)
except JSONValidationError as e:
errors.append(str(e.message))
except jsonschema.SchemaError as e:
errors.append(f"Schema error: {e.message}")
return {
'valid': len(errors) == 0,
'errors': errors,
}
@staticmethod
def validate_submission(form_schema_obj, data: dict) -> dict:
"""Validate using a FormSchema model instance."""
schema = form_schema_obj.schema
if not isinstance(schema, dict):
return {'valid': False, 'errors': ['Invalid schema format']}
return FormValidator.validate(schema, data)
\ No newline at end of file
backend/apps/forms_engine/views.py
View file @ e666e669
# Views for forms_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Forms Engine Views
"""
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.filters import SearchFilter, OrderingFilter
from django.utils import timezone
from .models import FormSchema, FormSubmission
from .serializers import (
FormSchemaSerializer, FormSchemaListSerializer,
FormSubmissionSerializer,
)
from .validator import FormValidator
class FormSchemaViewSet(viewsets.ModelViewSet):
queryset = FormSchema.objects.all()
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter]
filterset_fields = ['code', 'is_active']
search_fields = ['code', 'name_ar']
ordering = ['code', '-version']
def get_serializer_class(self):
if self.action == 'list':
return FormSchemaListSerializer
return FormSchemaSerializer
@action(detail=True, methods=['post'], url_path='validate')
def validate_data(self, request, pk=None):
schema_obj = self.get_object()
data = request.data.get('data', {})
result = FormValidator.validate_submission(schema_obj, data)
return Response(result)
class FormSubmissionViewSet(viewsets.ModelViewSet):
queryset = FormSubmission.objects.select_related('schema', 'member', 'submitted_by').all()
serializer_class = FormSubmissionSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
filterset_fields = ['status', 'schema__code', 'member']
search_fields = ['form_number', 'member__full_name_ar', 'member__membership_number']
ordering = ['-submitted_at']
def perform_create(self, serializer):
serializer.save(submitted_by=self.request.user)
@action(detail=True, methods=['post'], url_path='approve')
def approve(self, request, pk=None):
submission = self.get_object()
if submission.status not in ('SUBMITTED', 'UNDER_REVIEW'):
return Response(
{'error': 'لا يمكن اعتماد استمارة في هذه الحالة'},
status=status.HTTP_400_BAD_REQUEST,
)
submission.status = 'APPROVED'
submission.approved_at = timezone.now()
submission.reviewed_by = request.user
submission.save()
return Response(FormSubmissionSerializer(submission).data)
@action(detail=True, methods=['post'], url_path='reject')
def reject(self, request, pk=None):
submission = self.get_object()
if submission.status not in ('SUBMITTED', 'UNDER_REVIEW'):
return Response(
{'error': 'لا يمكن رفض استمارة في هذه الحالة'},
status=status.HTTP_400_BAD_REQUEST,
)
submission.status = 'REJECTED'
submission.reviewed_by = request.user
submission.notes = request.data.get('notes', submission.notes)
submission.save()
return Response(FormSubmissionSerializer(submission).data)
\ No newline at end of file
backend/apps/pricing/admin.py
View file @ e666e669
# Admin for pricing — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Pricing Admin
"""
from django.contrib import admin
from django.db.models import JSONField
from django_json_widget.widgets import JSONEditorWidget
from simple_history.admin import SimpleHistoryAdmin
from .models import PricingConfig, PriceOverride, DiscountRule
@admin.register(PricingConfig)
class PricingConfigAdmin(SimpleHistoryAdmin):
list_display = ['branch', 'qualification', 'amount', 'currency', 'effective_from', 'effective_to']
list_filter = ['branch', 'qualification', 'currency']
search_fields = ['branch__name_ar']
readonly_fields = ['created_at', 'updated_at']
@admin.register(PriceOverride)
class PriceOverrideAdmin(SimpleHistoryAdmin):
list_display = ['member_id', 'original_amount', 'override_amount', 'approved_by', 'created_at']
search_fields = ['member_id', 'reason']
raw_id_fields = ['approved_by']
readonly_fields = ['created_at']
@admin.register(DiscountRule)
class DiscountRuleAdmin(SimpleHistoryAdmin):
list_display = ['code', 'name_ar', 'discount_type', 'is_stackable', 'is_active', 'effective_from']
list_filter = ['discount_type', 'is_active']
search_fields = ['code', 'name_ar']
list_editable = ['is_active']
formfield_overrides = {
JSONField: {'widget': JSONEditorWidget},
}
\ No newline at end of file
backend/apps/pricing/calculator.py
View file @ e666e669
# Price calculator — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Price Calculator
=================================
Lookup membership prices. Apply discounts. Calculate totals.
"""
from datetime import date
from decimal import Decimal
from typing import Optional
from django.db.models import Q
from django.utils import timezone
from utils.decimal_utils import money, percentage_of
class PriceCalculator:
@staticmethod
def get_membership_price(branch_id: int, qualification: str,
at_date: date = None) -> Decimal:
"""Look up the active membership price for branch + qualification."""
from .models import PricingConfig
if at_date is None:
at_date = timezone.now().date()
config = PricingConfig.objects.filter(
branch_id=branch_id,
qualification=qualification,
effective_from__lte=at_date,
).filter(
Q(effective_to__isnull=True) | Q(effective_to__gte=at_date)
).order_by('-effective_from').first()
if config:
return money(config.amount)
return money(0)
@staticmethod
def calculate_total_with_discounts(base_price: Decimal,
discount_codes: list = None,
member_count: int = None) -> dict:
"""Apply applicable discounts to base price."""
from .models import DiscountRule
base = money(base_price)
discounts = []
total_discount = money(0)
today = timezone.now().date()
# Group discount by member count
if member_count and member_count >= 5:
from apps.rules_engine.evaluator import RuleEvaluator
tiers = [
('GROUP_DISCOUNT_TIER_1', 5, 10),
('GROUP_DISCOUNT_TIER_2', 11, 20),
('GROUP_DISCOUNT_TIER_3', 21, 9999),
]
for tier_code, min_count, max_count in tiers:
if min_count <= member_count <= max_count:
params = RuleEvaluator.get_rule(tier_code)
pct = Decimal(str(params.get('percentage', 0)))
disc_amount = percentage_of(base, pct)
discounts.append({
'code': tier_code,
'percentage': pct,
'amount': disc_amount,
})
total_discount += disc_amount
break
# Cap total discount
max_pct_params = {}
try:
from apps.rules_engine.evaluator import RuleEvaluator
max_pct_params = RuleEvaluator.get_rule('GROUP_DISCOUNT_MAX')
except Exception:
pass
max_pct = Decimal(str(max_pct_params.get('max_percentage', 10)))
max_discount = percentage_of(base, max_pct)
if total_discount > max_discount:
total_discount = max_discount
return {
'base': base,
'discounts': discounts,
'total_discount': total_discount,
'total': money(base - total_discount),
}
@staticmethod
def get_service_price(service_code: str, branch_id: int = None) -> Decimal:
"""Look up a service price from the catalog."""
from apps.service_catalog.models import ServicePrice
today = timezone.now().date()
qs = ServicePrice.objects.filter(
code=service_code,
is_active=True,
effective_from__lte=today,
).filter(
Q(effective_to__isnull=True) | Q(effective_to__gte=today)
)
if branch_id:
branch_price = qs.filter(branch_id=branch_id).first()
if branch_price and branch_price.amount:
return money(branch_price.amount)
global_price = qs.filter(branch__isnull=True).first()
if global_price and global_price.amount:
return money(global_price.amount)
return money(0)
@staticmethod
def calculate_addition_fee(service_code: str, membership_value: Decimal,
branch_id: int = None, **kwargs) -> Decimal:
"""For percentage-based services, calculate the actual amount."""
from apps.service_catalog.models import ServicePrice
today = timezone.now().date()
qs = ServicePrice.objects.filter(
code=service_code,
is_active=True,
effective_from__lte=today,
).filter(
Q(effective_to__isnull=True) | Q(effective_to__gte=today)
)
if branch_id:
svc = qs.filter(branch_id=branch_id).first() or qs.filter(branch__isnull=True).first()
else:
svc = qs.filter(branch__isnull=True).first()
if not svc:
return money(0)
mv = money(membership_value)
if svc.price_type == 'FIXED':
return money(svc.amount or 0)
elif svc.price_type == 'PERCENTAGE':
return percentage_of(mv, svc.percentage or Decimal('0'))
elif svc.price_type == 'PERCENTAGE_PLUS_ANNUAL':
pct_fee = percentage_of(mv, svc.percentage or Decimal('0'))
return pct_fee # Annual part calculated separately by spouse/child fee calc
return money(0)
\ No newline at end of file
backend/apps/pricing/management/commands/seed_pricing.py 0 → 100644
View file @ e666e669
from django.core.management.base import BaseCommand
from apps.pricing.seed import seed_pricing
class Command(BaseCommand):
help = 'Seed membership pricing configurations'
def handle(self, *args, **options):
created = seed_pricing()
self.stdout.write(self.style.SUCCESS(f'Pricing seeded: {created} configs created'))
\ No newline at end of file
backend/apps/pricing/models.py
View file @ e666e669
# Models for pricing — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Pricing Models
================================
Membership prices per branch per qualification. Historical pricing.
Discounts. Per-member overrides.
"""
from django.db import models
from django.conf import settings
from simple_history.models import HistoricalRecords
import auditlog
class PricingConfig(models.Model):
QUALIFICATION_CHOICES = [
('HIGH', 'مؤهل عالي'),
('MEDIUM', 'مؤهل متوسط'),
('NONE', 'بدون مؤهل'),
]
branch = models.ForeignKey(
'branches.Branch', on_delete=models.CASCADE,
related_name='pricing_configs', verbose_name="الفرع",
)
qualification = models.CharField(
max_length=20, choices=QUALIFICATION_CHOICES, verbose_name="المؤهل",
)
amount = models.DecimalField(
max_digits=12, decimal_places=2, verbose_name="المبلغ",
)
currency = models.CharField(max_length=3, default='EGP', verbose_name="العملة")
effective_from = models.DateField(verbose_name="ساري من")
effective_to = models.DateField(null=True, blank=True, verbose_name="ساري حتى")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
updated_at = models.DateTimeField(auto_now=True, verbose_name="تاريخ التعديل")
history = HistoricalRecords()
class Meta:
verbose_name = "تسعير العضوية"
verbose_name_plural = "تسعيرات العضوية"
ordering = ['-effective_from', 'branch', 'qualification']
unique_together = [('branch', 'qualification', 'effective_from')]
def __str__(self):
return f"{self.branch} — {self.get_qualification_display()} — {self.amount} {self.currency}"
class PriceOverride(models.Model):
member_id = models.BigIntegerField(verbose_name="رقم العضو")
original_amount = models.DecimalField(
max_digits=12, decimal_places=2, verbose_name="المبلغ الأصلي",
)
override_amount = models.DecimalField(
max_digits=12, decimal_places=2, verbose_name="المبلغ المعدل",
)
reason = models.TextField(verbose_name="السبب")
approved_by = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, verbose_name="اعتمد بواسطة",
)
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
history = HistoricalRecords()
class Meta:
verbose_name = "استثناء تسعير"
verbose_name_plural = "استثناءات التسعير"
ordering = ['-created_at']
def __str__(self):
return f"Override member#{self.member_id}: {self.original_amount} → {self.override_amount}"
class DiscountRule(models.Model):
DISCOUNT_TYPE_CHOICES = [
('PERCENTAGE', 'نسبة مئوية'),
('FIXED', 'مبلغ ثابت'),
]
code = models.CharField(max_length=50, unique=True, verbose_name="الكود")
name_ar = models.CharField(max_length=200, verbose_name="الاسم بالعربي")
name_en = models.CharField(max_length=200, blank=True, default="", verbose_name="الاسم بالانجليزي")
discount_type = models.CharField(
max_length=20, choices=DISCOUNT_TYPE_CHOICES, verbose_name="نوع الخصم",
)
parameters = models.JSONField(verbose_name="المعاملات")
is_stackable = models.BooleanField(default=False, verbose_name="قابل للتجميع")
is_active = models.BooleanField(default=True, verbose_name="نشط")
effective_from = models.DateField(verbose_name="ساري من")
effective_to = models.DateField(null=True, blank=True, verbose_name="ساري حتى")
history = HistoricalRecords()
class Meta:
verbose_name = "قاعدة خصم"
verbose_name_plural = "قواعد الخصم"
ordering = ['code']
def __str__(self):
return f"{self.code} — {self.name_ar}"
auditlog.register(PricingConfig)
auditlog.register(PriceOverride)
auditlog.register(DiscountRule)
\ No newline at end of file
backend/apps/pricing/seed.py
View file @ e666e669
# Management command: seed_pricing — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Pricing Seed Data
"""
from datetime import date
from decimal import Decimal
def seed_pricing():
"""Seed all membership pricing configurations."""
from .models import PricingConfig
from apps.branches.models import Branch
branches = {b.code: b for b in Branch.objects.all()}
created = 0
# Previous prices: opening to 30/6/2024
previous_prices = [
('HIGH', Decimal('114000.00')),
('MEDIUM', Decimal('171000.00')),
('NONE', Decimal('228000.00')),
]
# Current prices: from 1/7/2024
current_prices = [
('HIGH', Decimal('150000.00')),
('MEDIUM', Decimal('225000.00')),
('NONE', Decimal('300000.00')),
]
for branch_code, branch in branches.items():
# Previous pricing era
for qual, amount in previous_prices:
_, was_created = PricingConfig.objects.get_or_create(
branch=branch,
qualification=qual,
effective_from=date(2023, 1, 1),
defaults={
'amount': amount,
'effective_to': date(2024, 6, 30),
},
)
if was_created:
created += 1
# Current pricing era
for qual, amount in current_prices:
_, was_created = PricingConfig.objects.get_or_create(
branch=branch,
qualification=qual,
effective_from=date(2024, 7, 1),
defaults={'amount': amount},
)
if was_created:
created += 1
return created
\ No newline at end of file
backend/apps/pricing/serializers.py
View file @ e666e669
# Serializers for pricing — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Pricing Serializers
"""
from rest_framework import serializers
from .models import PricingConfig, PriceOverride, DiscountRule
class PricingConfigSerializer(serializers.ModelSerializer):
branch_name = serializers.CharField(source='branch.name_ar', read_only=True)
class Meta:
model = PricingConfig
fields = [
'id', 'branch', 'branch_name', 'qualification', 'amount',
'currency', 'effective_from', 'effective_to',
'created_at', 'updated_at',
]
read_only_fields = ['created_at', 'updated_at']
class PriceOverrideSerializer(serializers.ModelSerializer):
class Meta:
model = PriceOverride
fields = '__all__'
read_only_fields = ['created_at']
class DiscountRuleSerializer(serializers.ModelSerializer):
class Meta:
model = DiscountRule
fields = '__all__'
class PriceLookupSerializer(serializers.Serializer):
branch_id = serializers.IntegerField()
qualification = serializers.ChoiceField(choices=['HIGH', 'MEDIUM', 'NONE'])
at_date = serializers.DateField(required=False)
\ No newline at end of file
backend/apps/pricing/tests.py
View file @ e666e669
# Tests for pricing — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Pricing Tests
"""
from datetime import date
from decimal import Decimal
from django.test import TestCase
from apps.branches.models import Branch
from .models import PricingConfig
from .calculator import PriceCalculator
from .seed import seed_pricing
class PricingCalculatorTest(TestCase):
@classmethod
def setUpTestData(cls):
cls.branch = Branch.objects.create(
code='test_branch', name_ar='فرع اختبار', is_active=True,
)
PricingConfig.objects.create(
branch=cls.branch, qualification='HIGH',
amount=Decimal('150000.00'), effective_from=date(2024, 7, 1),
)
PricingConfig.objects.create(
branch=cls.branch, qualification='MEDIUM',
amount=Decimal('225000.00'), effective_from=date(2024, 7, 1),
)
def test_lookup_high_qualification(self):
price = PriceCalculator.get_membership_price(
self.branch.id, 'HIGH', date(2024, 8, 1),
)
self.assertEqual(price, Decimal('150000.00'))
def test_lookup_no_config_returns_zero(self):
price = PriceCalculator.get_membership_price(
self.branch.id, 'NONE', date(2024, 8, 1),
)
self.assertEqual(price, Decimal('0.00'))
\ No newline at end of file
backend/apps/pricing/urls.py
View file @ e666e669
urlpatterns = [] """
\ No newline at end of file THE CLUB ERP — Pricing URLs
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views
router = DefaultRouter()
router.register(r'configs', views.PricingConfigViewSet, basename='pricingconfig')
router.register(r'overrides', views.PriceOverrideViewSet, basename='priceoverride')
router.register(r'discounts', views.DiscountRuleViewSet, basename='discountrule')
urlpatterns = [
path('', include(router.urls)),
]
\ No newline at end of file
backend/apps/pricing/views.py
View file @ e666e669
# Views for pricing — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Pricing Views
"""
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.filters import SearchFilter, OrderingFilter
from .models import PricingConfig, PriceOverride, DiscountRule
from .serializers import (
PricingConfigSerializer, PriceOverrideSerializer,
DiscountRuleSerializer, PriceLookupSerializer,
)
from .calculator import PriceCalculator
class PricingConfigViewSet(viewsets.ModelViewSet):
queryset = PricingConfig.objects.select_related('branch').all()
serializer_class = PricingConfigSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
filterset_fields = ['branch', 'qualification', 'currency']
ordering = ['-effective_from']
@action(detail=False, methods=['post'], url_path='lookup')
def lookup(self, request):
ser = PriceLookupSerializer(data=request.data)
ser.is_valid(raise_exception=True)
price = PriceCalculator.get_membership_price(
ser.validated_data['branch_id'],
ser.validated_data['qualification'],
ser.validated_data.get('at_date'),
)
return Response({'price': str(price), 'currency': 'EGP'})
class PriceOverrideViewSet(viewsets.ModelViewSet):
queryset = PriceOverride.objects.all()
serializer_class = PriceOverrideSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend]
filterset_fields = ['member_id']
ordering = ['-created_at']
class DiscountRuleViewSet(viewsets.ModelViewSet):
queryset = DiscountRule.objects.all()
serializer_class = DiscountRuleSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter]
filterset_fields = ['discount_type', 'is_active']
search_fields = ['code', 'name_ar']
ordering = ['code']
\ No newline at end of file
backend/apps/rules_engine/admin.py
View file @ e666e669
# Admin for rules_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Rules Engine Admin
"""
from django.contrib import admin
from django.db.models import JSONField
from django_json_widget.widgets import JSONEditorWidget
from simple_history.admin import SimpleHistoryAdmin
from .models import BusinessRule, RuleOverride
@admin.register(BusinessRule)
class BusinessRuleAdmin(SimpleHistoryAdmin):
list_display = [
'code', 'name_ar', 'category', 'is_active',
'branch', 'effective_from', 'effective_to',
]
list_filter = ['category', 'is_active', 'branch']
search_fields = ['code', 'name_ar', 'name_en']
list_editable = ['is_active']
date_hierarchy = 'effective_from'
readonly_fields = ['created_at', 'updated_at', 'created_by', 'updated_by']
formfield_overrides = {
JSONField: {'widget': JSONEditorWidget},
}
fieldsets = (
('التعريف', {
'fields': ('code', 'name_ar', 'name_en', 'category', 'description'),
}),
('المعاملات', {
'fields': ('parameters',),
}),
('النطاق', {
'fields': ('is_active', 'branch', 'effective_from', 'effective_to'),
}),
('المراجعة', {
'fields': ('created_at', 'updated_at', 'created_by', 'updated_by'),
'classes': ('collapse',),
}),
)
def save_model(self, request, obj, form, change):
if not change:
obj.created_by = request.user
obj.updated_by = request.user
super().save_model(request, obj, form, change)
@admin.register(RuleOverride)
class RuleOverrideAdmin(SimpleHistoryAdmin):
list_display = ['rule', 'member_id', 'effective_from', 'effective_to', 'approved_by']
list_filter = ['rule__category', 'effective_from']
search_fields = ['rule__code', 'member_id', 'reason']
raw_id_fields = ['approved_by']
readonly_fields = ['created_at']
formfield_overrides = {
JSONField: {'widget': JSONEditorWidget},
}
\ No newline at end of file
backend/apps/rules_engine/evaluator.py
View file @ e666e669
This diff is collapsed.
backend/apps/rules_engine/management/commands/seed_rules.py 0 → 100644
View file @ e666e669
"""
THE CLUB ERP — Seed Rules Management Command
"""
from django.core.management.base import BaseCommand
from apps.rules_engine.seed import seed_rules
class Command(BaseCommand):
help = 'Seed all 70+ business rules from master document'
def handle(self, *args, **options):
created, updated = seed_rules()
self.stdout.write(self.style.SUCCESS(
f'Rules seeded: {created} created, {updated} updated'
))
\ No newline at end of file
backend/apps/rules_engine/models.py
View file @ e666e669
# Models for rules_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Rules Engine Models
====================================
Every business rule, percentage, age limit, and fee structure
stored as configurable database records. ZERO hardcoded logic.
"""
from django.db import models
from django.conf import settings
from simple_history.models import HistoricalRecords
import auditlog
class BusinessRule(models.Model):
"""
Central business rule store. All fee percentages, age limits,
time windows, and financial parameters live here.
"""
CATEGORY_CHOICES = [
('age', 'قواعد السن'),
('children_fee', 'رسوم الأبناء'),
('spouse_fee', 'رسوم الزوجات'),
('separation', 'رسوم الفصل والتحويل'),
('divorce', 'قواعد الطلاق'),
('financial', 'قواعد مالية'),
('penalty', 'قواعد الجزاءات'),
('discount', 'قواعد الخصومات'),
('form_workflow', 'قواعد الاستمارات'),
('temporary', 'قواعد العضو المؤقت'),
('foreign', 'قواعد العضو الأجنبي'),
('death', 'قواعد حالات الوفاة'),
]
code = models.CharField(max_length=100, db_index=True, verbose_name="الكود")
name_ar = models.CharField(max_length=300, verbose_name="الاسم بالعربي")
name_en = models.CharField(max_length=300, blank=True, default="", verbose_name="الاسم بالانجليزي")
category = models.CharField(max_length=50, choices=CATEGORY_CHOICES, verbose_name="الفئة")
parameters = models.JSONField(verbose_name="المعاملات")
description = models.TextField(blank=True, default="", verbose_name="الوصف")
is_active = models.BooleanField(default=True, verbose_name="نشط")
branch = models.ForeignKey(
'branches.Branch', null=True, blank=True,
on_delete=models.SET_NULL, verbose_name="الفرع",
)
effective_from = models.DateField(verbose_name="ساري من")
effective_to = models.DateField(null=True, blank=True, verbose_name="ساري حتى")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
updated_at = models.DateTimeField(auto_now=True, verbose_name="تاريخ التعديل")
created_by = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, related_name='rules_created',
verbose_name="أنشأ بواسطة",
)
updated_by = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, related_name='rules_updated',
verbose_name="عدّل بواسطة",
)
history = HistoricalRecords()
class Meta:
verbose_name = "قاعدة عمل"
verbose_name_plural = "قواعد العمل"
ordering = ['category', 'code']
indexes = [
models.Index(fields=['code', 'is_active', 'effective_from']),
models.Index(fields=['category', 'is_active']),
]
def __str__(self):
return f"{self.code} — {self.name_ar}"
class RuleOverride(models.Model):
"""Per-member rule override with documented reason and approval."""
rule = models.ForeignKey(
BusinessRule, on_delete=models.CASCADE,
related_name='overrides', verbose_name="القاعدة",
)
member_id = models.BigIntegerField(verbose_name="رقم العضو")
override_parameters = models.JSONField(verbose_name="المعاملات المعدلة")
reason = models.TextField(verbose_name="السبب")
approved_by = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, verbose_name="اعتمد بواسطة",
)
effective_from = models.DateField(verbose_name="ساري من")
effective_to = models.DateField(null=True, blank=True, verbose_name="ساري حتى")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
history = HistoricalRecords()
class Meta:
verbose_name = "استثناء قاعدة"
verbose_name_plural = "استثناءات القواعد"
ordering = ['-created_at']
def __str__(self):
return f"Override: {self.rule.code} for member #{self.member_id}"
auditlog.register(BusinessRule)
auditlog.register(RuleOverride)
\ No newline at end of file
backend/apps/rules_engine/seed.py
View file @ e666e669
This diff is collapsed.
backend/apps/rules_engine/serializers.py
View file @ e666e669
# Serializers for rules_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Rules Engine Serializers
"""
from rest_framework import serializers
from .models import BusinessRule, RuleOverride
class BusinessRuleSerializer(serializers.ModelSerializer):
branch_name = serializers.CharField(source='branch.name_ar', read_only=True, default='')
class Meta:
model = BusinessRule
fields = [
'id', 'code', 'name_ar', 'name_en', 'category',
'parameters', 'description', 'is_active', 'branch',
'branch_name', 'effective_from', 'effective_to',
'created_at', 'updated_at',
]
read_only_fields = ['created_at', 'updated_at']
class BusinessRuleListSerializer(serializers.ModelSerializer):
branch_name = serializers.CharField(source='branch.name_ar', read_only=True, default='')
class Meta:
model = BusinessRule
fields = [
'id', 'code', 'name_ar', 'category', 'is_active',
'branch', 'branch_name', 'effective_from', 'effective_to',
]
class RuleOverrideSerializer(serializers.ModelSerializer):
rule_code = serializers.CharField(source='rule.code', read_only=True)
approved_by_name = serializers.CharField(
source='approved_by.full_name_ar', read_only=True, default='',
)
class Meta:
model = RuleOverride
fields = [
'id', 'rule', 'rule_code', 'member_id',
'override_parameters', 'reason', 'approved_by',
'approved_by_name', 'effective_from', 'effective_to',
'created_at',
]
read_only_fields = ['created_at']
class RuleEvaluateSerializer(serializers.Serializer):
"""For the evaluate endpoint."""
code = serializers.CharField()
branch_id = serializers.IntegerField(required=False, allow_null=True)
at_date = serializers.DateField(required=False, allow_null=True)
member_id = serializers.IntegerField(required=False, allow_null=True)
\ No newline at end of file
backend/apps/rules_engine/tests.py
View file @ e666e669
# Tests for rules_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Rules Engine Tests
"""
from datetime import date
from decimal import Decimal
from django.test import TestCase
from django.utils import timezone
from .models import BusinessRule, RuleOverride
from .evaluator import RuleEvaluator
from .seed import seed_rules
class RuleSeedTest(TestCase):
def test_seed_creates_all_rules(self):
created, _ = seed_rules()
self.assertGreaterEqual(created, 70)
self.assertTrue(BusinessRule.objects.filter(code='MIN_WORKING_AGE').exists())
def test_seed_idempotent(self):
seed_rules()
_, updated = seed_rules()
self.assertGreater(updated, 0)
class RuleEvaluatorTest(TestCase):
@classmethod
def setUpTestData(cls):
seed_rules()
def test_get_rule_min_working_age(self):
params = RuleEvaluator.get_rule('MIN_WORKING_AGE')
self.assertEqual(params['min_age'], 21)
def test_get_rule_value(self):
val = RuleEvaluator.get_rule_value('INSTALLMENT_MAX_MONTHS', 'months')
self.assertEqual(val, 30)
def test_child_fee_first_three_free(self):
result = RuleEvaluator.evaluate_child_fee(10, 1, Decimal('150000'))
self.assertEqual(result['fee'], Decimal('0.00'))
self.assertEqual(result['classification'], 'INCLUDED')
def test_child_fee_fourth_under_18(self):
result = RuleEvaluator.evaluate_child_fee(15, 4, Decimal('150000'))
self.assertEqual(result['fee'], Decimal('7500.00'))
self.assertEqual(result['classification'], 'DEPENDENT_FEE')
def test_child_fee_age_18(self):
result = RuleEvaluator.evaluate_child_fee(18, 1, Decimal('150000'))
self.assertEqual(result['fee'], Decimal('15000.00'))
def test_child_fee_age_21_temporary(self):
result = RuleEvaluator.evaluate_child_fee(21, 1, Decimal('150000'))
self.assertEqual(result['classification'], 'TEMPORARY')
self.assertEqual(result['fee'], Decimal('22500.00'))
def test_child_fee_age_25_rejected(self):
result = RuleEvaluator.evaluate_child_fee(25, 1, Decimal('150000'))
self.assertEqual(result['classification'], 'REJECTED')
def test_spouse_fee_first_free(self):
result = RuleEvaluator.evaluate_spouse_fee(
1, 'Egyptian', date(2020, 1, 1), date(2020, 1, 1), Decimal('150000'),
)
self.assertEqual(result['total'], Decimal('0.00'))
def test_spouse_fee_foreign(self):
result = RuleEvaluator.evaluate_spouse_fee(
2, 'American', date(2020, 1, 1), date(2020, 1, 1), Decimal('150000'),
)
self.assertEqual(result['total'], Decimal('22500.00'))
def test_separation_fee_year_1(self):
result = RuleEvaluator.evaluate_separation_fee(1, Decimal('150000'))
self.assertEqual(result['fee'], Decimal('45000.00'))
self.assertEqual(result['percentage'], Decimal('30'))
def test_separation_fee_year_6_plus(self):
result = RuleEvaluator.evaluate_separation_fee(10, Decimal('150000'))
self.assertEqual(result['fee'], Decimal('3750.00'))
\ No newline at end of file
backend/apps/rules_engine/urls.py
View file @ e666e669
urlpatterns = [] """
\ No newline at end of file THE CLUB ERP — Rules Engine URLs
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views
router = DefaultRouter()
router.register(r'rules', views.BusinessRuleViewSet, basename='businessrule')
router.register(r'overrides', views.RuleOverrideViewSet, basename='ruleoverride')
urlpatterns = [
path('', include(router.urls)),
]
\ No newline at end of file
backend/apps/rules_engine/views.py
View file @ e666e669
# Views for rules_engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Rules Engine Views
"""
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.filters import SearchFilter, OrderingFilter
from .models import BusinessRule, RuleOverride
from .serializers import (
BusinessRuleSerializer, BusinessRuleListSerializer,
RuleOverrideSerializer, RuleEvaluateSerializer,
)
from .evaluator import RuleEvaluator
class BusinessRuleViewSet(viewsets.ModelViewSet):
queryset = BusinessRule.objects.select_related('branch').all()
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
filterset_fields = ['category', 'is_active', 'branch', 'code']
search_fields = ['code', 'name_ar', 'name_en', 'description']
ordering_fields = ['code', 'category', 'effective_from', 'created_at']
ordering = ['category', 'code']
def get_serializer_class(self):
if self.action == 'list':
return BusinessRuleListSerializer
return BusinessRuleSerializer
def perform_create(self, serializer):
serializer.save(created_by=self.request.user, updated_by=self.request.user)
def perform_update(self, serializer):
serializer.save(updated_by=self.request.user)
@action(detail=False, methods=['post'], url_path='evaluate')
def evaluate(self, request):
"""Evaluate a rule by code, optional branch and date."""
ser = RuleEvaluateSerializer(data=request.data)
ser.is_valid(raise_exception=True)
code = ser.validated_data['code']
branch_id = ser.validated_data.get('branch_id')
at_date = ser.validated_data.get('at_date')
member_id = ser.validated_data.get('member_id')
if member_id:
params = RuleEvaluator.get_rule_with_override(
code, member_id, branch_id, at_date,
)
else:
params = RuleEvaluator.get_rule(code, branch_id, at_date)
return Response({
'code': code,
'parameters': params,
'branch_id': branch_id,
'at_date': str(at_date) if at_date else None,
})
class RuleOverrideViewSet(viewsets.ModelViewSet):
queryset = RuleOverride.objects.select_related('rule', 'approved_by').all()
serializer_class = RuleOverrideSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
filterset_fields = ['rule__code', 'member_id']
search_fields = ['rule__code', 'reason']
ordering = ['-created_at']
\ No newline at end of file
backend/apps/service_catalog/admin.py
View file @ e666e669
# Admin for service_catalog — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Service Catalog Admin
"""
from django.contrib import admin
from simple_history.admin import SimpleHistoryAdmin
from .models import ServicePrice
@admin.register(ServicePrice)
class ServicePriceAdmin(SimpleHistoryAdmin):
list_display = [
'code', 'name_ar', 'price_type', 'amount',
'percentage', 'is_active', 'branch', 'effective_from',
]
list_filter = ['price_type', 'is_active', 'branch', 'currency']
search_fields = ['code', 'name_ar', 'name_en']
list_editable = ['amount', 'percentage', 'is_active']
readonly_fields = ['created_at', 'updated_at']
\ No newline at end of file
backend/apps/service_catalog/management/commands/seed_catalog.py 0 → 100644
View file @ e666e669
from django.core.management.base import BaseCommand
from apps.service_catalog.seed import seed_catalog
class Command(BaseCommand):
help = 'Seed service price catalog'
def handle(self, *args, **options):
created = seed_catalog()
self.stdout.write(self.style.SUCCESS(f'Catalog seeded: {created} services created'))
\ No newline at end of file
backend/apps/service_catalog/models.py
View file @ e666e669
# Models for service_catalog — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Service Price Catalog
======================================
Every fee, form cost, percentage-based service stored here.
Employees NEVER type a price — they select from this catalog.
"""
from django.db import models
from simple_history.models import HistoricalRecords
import auditlog
class ServicePrice(models.Model):
PRICE_TYPE_CHOICES = [
('FIXED', 'مبلغ ثابت'),
('PERCENTAGE', 'نسبة مئوية'),
('PERCENTAGE_PLUS_ANNUAL', 'نسبة + رسوم سنوية'),
]
code = models.CharField(max_length=50, unique=True, verbose_name="الكود")
name_ar = models.CharField(max_length=200, verbose_name="الاسم بالعربي")
name_en = models.CharField(max_length=200, blank=True, default="", verbose_name="الاسم بالانجليزي")
price_type = models.CharField(
max_length=30, choices=PRICE_TYPE_CHOICES, verbose_name="نوع السعر",
)
amount = models.DecimalField(
max_digits=12, decimal_places=2, null=True, blank=True, verbose_name="المبلغ",
)
percentage = models.DecimalField(
max_digits=5, decimal_places=2, null=True, blank=True, verbose_name="النسبة",
)
annual_flat = models.DecimalField(
max_digits=10, decimal_places=2, null=True, blank=True, verbose_name="الرسوم السنوية",
)
base_reference = models.CharField(
max_length=50, blank=True, default="", verbose_name="مرجع الحساب",
)
branch = models.ForeignKey(
'branches.Branch', null=True, blank=True,
on_delete=models.SET_NULL, verbose_name="الفرع",
)
currency = models.CharField(max_length=3, default='EGP', verbose_name="العملة")
is_active = models.BooleanField(default=True, verbose_name="نشط")
effective_from = models.DateField(verbose_name="ساري من")
effective_to = models.DateField(null=True, blank=True, verbose_name="ساري حتى")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
updated_at = models.DateTimeField(auto_now=True, verbose_name="تاريخ التعديل")
history = HistoricalRecords()
class Meta:
verbose_name = "سعر الخدمة"
verbose_name_plural = "أسعار الخدمات"
ordering = ['code']
def __str__(self):
if self.price_type == 'FIXED':
return f"{self.code} — {self.name_ar} — {self.amount} {self.currency}"
elif self.price_type == 'PERCENTAGE':
return f"{self.code} — {self.name_ar} — {self.percentage}%"
return f"{self.code} — {self.name_ar}"
auditlog.register(ServicePrice)
\ No newline at end of file
backend/apps/service_catalog/seed.py
View file @ e666e669
# Management command: seed_catalog — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Service Catalog Seed Data
All 30+ service prices from Section 69.1
"""
from datetime import date
from decimal import Decimal
CATALOG_DATA = [
{'code': 'SVC_NEW_FORM', 'name_ar': 'استمارة عضوية جديدة', 'price_type': 'FIXED', 'amount': Decimal('505.00')},
{'code': 'SVC_TRANSFER_FORM', 'name_ar': 'استمارة تحويل / فصل', 'price_type': 'FIXED', 'amount': Decimal('570.00')},
{'code': 'SVC_ADDITION_FORM', 'name_ar': 'استمارة إضافة', 'price_type': 'FIXED', 'amount': Decimal('570.00')},
{'code': 'SVC_MARTYRS_STAMP', 'name_ar': 'طابع شهداء', 'price_type': 'FIXED', 'amount': Decimal('5.00')},
{'code': 'SVC_DEV_FEE', 'name_ar': 'رسوم تنمية', 'price_type': 'FIXED', 'amount': Decimal('35.00')},
{'code': 'SVC_CARNET_REPLACE', 'name_ar': 'بدل فاقد كارنيه', 'price_type': 'FIXED', 'amount': Decimal('100.00')},
{'code': 'SVC_SEASONAL', 'name_ar': 'عضوية موسمية', 'price_type': 'PERCENTAGE', 'percentage': Decimal('5.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_TEMP_MEMBER', 'name_ar': 'رسوم عضو مؤقت', 'price_type': 'PERCENTAGE', 'percentage': Decimal('10.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_CHILD_4TH', 'name_ar': 'رسوم ابن رابع', 'price_type': 'PERCENTAGE', 'percentage': Decimal('5.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_CHILD_18', 'name_ar': 'رسوم ابن 18 سنة', 'price_type': 'PERCENTAGE', 'percentage': Decimal('10.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_CHILD_19', 'name_ar': 'رسوم ابن 19 سنة', 'price_type': 'PERCENTAGE', 'percentage': Decimal('15.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_CHILD_20', 'name_ar': 'رسوم ابن 20 سنة', 'price_type': 'PERCENTAGE', 'percentage': Decimal('20.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_CHILD_21', 'name_ar': 'رسوم ابن 21 سنة', 'price_type': 'PERCENTAGE', 'percentage': Decimal('15.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPOUSE_2ND', 'name_ar': 'رسوم زوجة ثانية', 'price_type': 'PERCENTAGE_PLUS_ANNUAL', 'percentage': Decimal('10.00'), 'annual_flat': Decimal('150.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPOUSE_3RD', 'name_ar': 'رسوم زوجة ثالثة', 'price_type': 'PERCENTAGE_PLUS_ANNUAL', 'percentage': Decimal('20.00'), 'annual_flat': Decimal('200.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPOUSE_4TH', 'name_ar': 'رسوم زوجة رابعة', 'price_type': 'PERCENTAGE_PLUS_ANNUAL', 'percentage': Decimal('30.00'), 'annual_flat': Decimal('300.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPOUSE_FOREIGN', 'name_ar': 'رسوم زوج أجنبي', 'price_type': 'PERCENTAGE', 'percentage': Decimal('15.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPOUSE_ACQUIRED', 'name_ar': 'رسوم زوج مكتسب', 'price_type': 'PERCENTAGE', 'percentage': Decimal('50.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPOUSE_BASE', 'name_ar': 'رسوم زوج أساسي', 'price_type': 'PERCENTAGE', 'percentage': Decimal('15.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_WAIVER', 'name_ar': 'رسوم تنازل', 'price_type': 'PERCENTAGE', 'percentage': Decimal('30.00'), 'base_reference': 'membership_value'},
{'code': 'SVC_SPORTS_CONV', 'name_ar': 'رسوم تحويل رياضي', 'price_type': 'PERCENTAGE', 'percentage': Decimal('50.00'), 'base_reference': 'new_membership_value'},
{'code': 'SVC_ANNUAL_MEMBER', 'name_ar': 'اشتراك سنوي - عضو', 'price_type': 'FIXED', 'amount': Decimal('410.00')},
{'code': 'SVC_ANNUAL_SPOUSE', 'name_ar': 'اشتراك سنوي - زوجة', 'price_type': 'FIXED', 'amount': Decimal('410.00')},
{'code': 'SVC_ANNUAL_CHILD', 'name_ar': 'اشتراك سنوي - ابن/ابنة', 'price_type': 'FIXED', 'amount': Decimal('185.00')},
{'code': 'SVC_ANNUAL_TEMP', 'name_ar': 'اشتراك سنوي - مؤقت', 'price_type': 'FIXED', 'amount': Decimal('185.00')},
{'code': 'SVC_FOREIGN_USD', 'name_ar': 'عضوية أجنبي (دولار)', 'price_type': 'FIXED', 'amount': Decimal('10000.00'), 'currency': 'USD'},
]
def seed_catalog():
from .models import ServicePrice
created = 0
effective_from = date(2024, 7, 1)
for item in CATALOG_DATA:
defaults = {
'name_ar': item['name_ar'],
'price_type': item['price_type'],
'amount': item.get('amount'),
'percentage': item.get('percentage'),
'annual_flat': item.get('annual_flat'),
'base_reference': item.get('base_reference', ''),
'currency': item.get('currency', 'EGP'),
'is_active': True,
'effective_from': effective_from,
}
_, was_created = ServicePrice.objects.update_or_create(
code=item['code'],
branch__isnull=True,
defaults=defaults,
)
if was_created:
created += 1
return created
\ No newline at end of file
backend/apps/service_catalog/serializers.py
View file @ e666e669
# Serializers for service_catalog — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Service Catalog Serializers
"""
from rest_framework import serializers
from .models import ServicePrice
class ServicePriceSerializer(serializers.ModelSerializer):
branch_name = serializers.CharField(source='branch.name_ar', read_only=True, default='')
class Meta:
model = ServicePrice
fields = [
'id', 'code', 'name_ar', 'name_en', 'price_type',
'amount', 'percentage', 'annual_flat', 'base_reference',
'branch', 'branch_name', 'currency', 'is_active',
'effective_from', 'effective_to', 'created_at', 'updated_at',
]
read_only_fields = ['created_at', 'updated_at']
\ No newline at end of file
backend/apps/service_catalog/tests.py
View file @ e666e669
# Tests for service_catalog — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Service Catalog Tests
"""
from decimal import Decimal
from django.test import TestCase
from .seed import seed_catalog
from .models import ServicePrice
class ServiceCatalogSeedTest(TestCase):
def test_seed_creates_services(self):
created = seed_catalog()
self.assertGreaterEqual(created, 20)
def test_new_form_fee(self):
seed_catalog()
svc = ServicePrice.objects.get(code='SVC_NEW_FORM')
self.assertEqual(svc.amount, Decimal('505.00'))
self.assertEqual(svc.price_type, 'FIXED')
def test_seasonal_percentage(self):
seed_catalog()
svc = ServicePrice.objects.get(code='SVC_SEASONAL')
self.assertEqual(svc.percentage, Decimal('5.00'))
self.assertEqual(svc.price_type, 'PERCENTAGE')
\ No newline at end of file
backend/apps/service_catalog/urls.py
View file @ e666e669
urlpatterns = [] """
\ No newline at end of file THE CLUB ERP — Service Catalog URLs
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views
router = DefaultRouter()
router.register(r'', views.ServicePriceViewSet, basename='serviceprice')
urlpatterns = [
path('', include(router.urls)),
]
\ No newline at end of file
backend/apps/service_catalog/views.py
View file @ e666e669
# Views for service_catalog — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Service Catalog Views
"""
from rest_framework import viewsets, permissions
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.filters import SearchFilter, OrderingFilter
from .models import ServicePrice
from .serializers import ServicePriceSerializer
class ServicePriceViewSet(viewsets.ModelViewSet):
queryset = ServicePrice.objects.select_related('branch').all()
serializer_class = ServicePriceSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
filterset_fields = ['price_type', 'is_active', 'branch', 'code']
search_fields = ['code', 'name_ar', 'name_en']
ordering_fields = ['code', 'effective_from']
ordering = ['code']
\ No newline at end of file
backend/apps/workflows/admin.py
View file @ e666e669
# Admin for workflows — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow Admin
"""
from django.contrib import admin
from django.db.models import JSONField
from django_json_widget.widgets import JSONEditorWidget
from simple_history.admin import SimpleHistoryAdmin
from .models import WorkflowDefinition, WorkflowInstance, WorkflowTransitionLog
class TransitionLogInline(admin.TabularInline):
model = WorkflowTransitionLog
readonly_fields = ['from_state', 'to_state', 'actor', 'notes', 'timestamp']
extra = 0
can_delete = False
@admin.register(WorkflowDefinition)
class WorkflowDefinitionAdmin(SimpleHistoryAdmin):
list_display = ['code', 'name_ar', 'is_active', 'created_at']
list_filter = ['is_active']
search_fields = ['code', 'name_ar']
formfield_overrides = {
JSONField: {'widget': JSONEditorWidget},
}
@admin.register(WorkflowInstance)
class WorkflowInstanceAdmin(SimpleHistoryAdmin):
list_display = ['definition', 'entity_type', 'entity_id', 'current_state', 'is_active', 'started_at']
list_filter = ['definition', 'current_state', 'is_active', 'entity_type']
search_fields = ['entity_id']
inlines = [TransitionLogInline]
readonly_fields = ['started_at']
\ No newline at end of file
backend/apps/workflows/engine.py
View file @ e666e669
# State machine engine — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow State Machine Engine
==============================================
Manages state transitions, guards, and auto-actions.
"""
from django.utils import timezone
class WorkflowError(Exception):
pass
class StateMachineEngine:
"""
Manages workflow instances: creation, transitions, completion.
"""
@staticmethod
def start_workflow(definition_code: str, entity_type: str,
entity_id: int, actor=None, data: dict = None):
"""Start a new workflow instance."""
from .models import WorkflowDefinition, WorkflowInstance, WorkflowTransitionLog
try:
definition = WorkflowDefinition.objects.get(code=definition_code, is_active=True)
except WorkflowDefinition.DoesNotExist:
raise WorkflowError(f"Workflow definition '{definition_code}' not found or inactive")
initial_state = definition.get_initial_state()
if not initial_state:
raise WorkflowError(f"No initial state in workflow '{definition_code}'")
instance = WorkflowInstance.objects.create(
definition=definition,
entity_type=entity_type,
entity_id=entity_id,
current_state=initial_state,
data=data or {},
)
WorkflowTransitionLog.objects.create(
instance=instance,
from_state='',
to_state=initial_state,
actor=actor,
notes='بدء سير العمل',
)
return instance
@staticmethod
def transition(instance_id: int, to_state: str, actor=None, notes: str = ''):
"""Transition a workflow instance to a new state."""
from .models import WorkflowInstance, WorkflowTransitionLog
try:
instance = WorkflowInstance.objects.select_related('definition').get(
id=instance_id, is_active=True,
)
except WorkflowInstance.DoesNotExist:
raise WorkflowError(f"Active workflow instance #{instance_id} not found")
# Validate transition exists
valid_transitions = instance.definition.get_transitions_from(instance.current_state)
valid_targets = [t['to'] for t in valid_transitions]
if to_state not in valid_targets:
raise WorkflowError(
f"Invalid transition from '{instance.current_state}' to '{to_state}'. "
f"Valid targets: {valid_targets}"
)
# Find the transition definition for permission/action info
transition_def = next(
(t for t in valid_transitions if t['to'] == to_state), None,
)
from_state = instance.current_state
instance.current_state = to_state
instance.save(update_fields=['current_state'])
# Check if this is a terminal state
terminal_states = [
s['code'] for s in instance.definition.states
if s.get('is_terminal')
]
if to_state in terminal_states:
instance.completed_at = timezone.now()
instance.is_active = False
instance.save(update_fields=['completed_at', 'is_active'])
WorkflowTransitionLog.objects.create(
instance=instance,
from_state=from_state,
to_state=to_state,
actor=actor,
notes=notes,
)
return instance
@staticmethod
def get_available_transitions(instance_id: int) -> list:
"""Get available transitions from the current state."""
from .models import WorkflowInstance
try:
instance = WorkflowInstance.objects.select_related('definition').get(
id=instance_id, is_active=True,
)
except WorkflowInstance.DoesNotExist:
return []
return instance.definition.get_transitions_from(instance.current_state)
@staticmethod
def get_instance_for_entity(entity_type: str, entity_id: int, active_only: bool = True):
"""Get workflow instance for an entity."""
from .models import WorkflowInstance
qs = WorkflowInstance.objects.filter(
entity_type=entity_type, entity_id=entity_id,
)
if active_only:
qs = qs.filter(is_active=True)
return qs.order_by('-started_at').first()
\ No newline at end of file
backend/apps/workflows/management/commands/seed_workflows.py 0 → 100644
View file @ e666e669
from django.core.management.base import BaseCommand
from apps.workflows.seed import seed_workflows
class Command(BaseCommand):
help = 'Seed workflow definitions'
def handle(self, *args, **options):
created = seed_workflows()
self.stdout.write(self.style.SUCCESS(f'Workflows seeded: {created} definitions created'))
\ No newline at end of file
backend/apps/workflows/models.py
View file @ e666e669
# Models for workflows — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow Models
================================
State machines for every process in the system.
"""
from django.db import models
from django.conf import settings
from simple_history.models import HistoricalRecords
import auditlog
class WorkflowDefinition(models.Model):
code = models.CharField(max_length=50, unique=True, verbose_name="الكود")
name_ar = models.CharField(max_length=200, verbose_name="الاسم بالعربي")
name_en = models.CharField(max_length=200, blank=True, default="", verbose_name="الاسم بالانجليزي")
states = models.JSONField(verbose_name="الحالات")
transitions = models.JSONField(verbose_name="التحولات")
is_active = models.BooleanField(default=True, verbose_name="نشط")
created_at = models.DateTimeField(auto_now_add=True, verbose_name="تاريخ الإنشاء")
history = HistoricalRecords()
class Meta:
verbose_name = "تعريف سير العمل"
verbose_name_plural = "تعريفات سير العمل"
ordering = ['code']
def __str__(self):
return f"{self.code} — {self.name_ar}"
def get_initial_state(self):
for state in self.states:
if state.get('is_initial'):
return state['code']
return self.states[0]['code'] if self.states else None
def get_transitions_from(self, state_code):
return [t for t in self.transitions if t['from'] == state_code]
class WorkflowInstance(models.Model):
definition = models.ForeignKey(
WorkflowDefinition, on_delete=models.PROTECT,
related_name='instances', verbose_name="التعريف",
)
entity_type = models.CharField(max_length=50, verbose_name="نوع الكيان")
entity_id = models.BigIntegerField(verbose_name="معرف الكيان")
current_state = models.CharField(max_length=50, verbose_name="الحالة الحالية")
data = models.JSONField(default=dict, blank=True, verbose_name="بيانات إضافية")
started_at = models.DateTimeField(auto_now_add=True, verbose_name="بدأ في")
completed_at = models.DateTimeField(null=True, blank=True, verbose_name="اكتمل في")
is_active = models.BooleanField(default=True, verbose_name="نشط")
history = HistoricalRecords()
class Meta:
verbose_name = "مثيل سير العمل"
verbose_name_plural = "مثيلات سير العمل"
indexes = [
models.Index(fields=['entity_type', 'entity_id']),
]
def __str__(self):
return f"{self.definition.code}:{self.entity_type}#{self.entity_id} → {self.current_state}"
class WorkflowTransitionLog(models.Model):
instance = models.ForeignKey(
WorkflowInstance, on_delete=models.CASCADE,
related_name='transition_logs', verbose_name="المثيل",
)
from_state = models.CharField(max_length=50, verbose_name="من حالة")
to_state = models.CharField(max_length=50, verbose_name="إلى حالة")
actor = models.ForeignKey(
settings.AUTH_USER_MODEL, null=True, blank=True,
on_delete=models.SET_NULL, verbose_name="الفاعل",
)
notes = models.TextField(blank=True, default="", verbose_name="ملاحظات")
timestamp = models.DateTimeField(auto_now_add=True, verbose_name="التوقيت")
class Meta:
verbose_name = "سجل تحول"
verbose_name_plural = "سجلات التحولات"
ordering = ['timestamp']
def __str__(self):
return f"{self.from_state} → {self.to_state} @ {self.timestamp}"
auditlog.register(WorkflowDefinition)
auditlog.register(WorkflowInstance)
\ No newline at end of file
backend/apps/workflows/seed.py
View file @ e666e669
# Management command: seed_workflows — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow Seed Data
All 5 predefined workflow definitions from Section 42.2.
"""
WORKFLOW_DATA = [
{
'code': 'NEW_MEMBERSHIP',
'name_ar': 'سير عمل عضوية جديدة',
'name_en': 'New Membership Workflow',
'states': [
{'code': 'form_submitted', 'name_ar': 'تم تقديم الاستمارة', 'is_initial': True},
{'code': 'under_review', 'name_ar': 'قيد المراجعة'},
{'code': 'interview_scheduled', 'name_ar': 'تم تحديد موعد المقابلة'},
{'code': 'accepted', 'name_ar': 'مقبول'},
{'code': 'rejected', 'name_ar': 'مرفوض', 'is_terminal': True},
{'code': 'payment_pending', 'name_ar': 'في انتظار السداد'},
{'code': 'active', 'name_ar': 'نشط', 'is_terminal': True},
{'code': 'expired', 'name_ar': 'منتهي الصلاحية', 'is_terminal': True},
],
'transitions': [
{'from': 'form_submitted', 'to': 'under_review', 'permission': 'member.create'},
{'from': 'under_review', 'to': 'interview_scheduled', 'permission': 'interview.schedule'},
{'from': 'interview_scheduled', 'to': 'accepted', 'permission': 'interview.decide'},
{'from': 'interview_scheduled', 'to': 'rejected', 'permission': 'interview.decide'},
{'from': 'accepted', 'to': 'payment_pending'},
{'from': 'payment_pending', 'to': 'active', 'permission': 'payment.process_cash'},
{'from': 'payment_pending', 'to': 'expired'},
],
},
{
'code': 'ADDITION',
'name_ar': 'سير عمل إضافة',
'name_en': 'Addition Workflow',
'states': [
{'code': 'addition_requested', 'name_ar': 'طلب إضافة', 'is_initial': True},
{'code': 'under_review', 'name_ar': 'قيد المراجعة'},
{'code': 'approved', 'name_ar': 'معتمد'},
{'code': 'rejected', 'name_ar': 'مرفوض', 'is_terminal': True},
{'code': 'fee_pending', 'name_ar': 'في انتظار الرسوم'},
{'code': 'fee_paid', 'name_ar': 'تم السداد'},
{'code': 'active', 'name_ar': 'نشط', 'is_terminal': True},
],
'transitions': [
{'from': 'addition_requested', 'to': 'under_review'},
{'from': 'under_review', 'to': 'approved'},
{'from': 'under_review', 'to': 'rejected'},
{'from': 'approved', 'to': 'fee_pending'},
{'from': 'fee_pending', 'to': 'fee_paid'},
{'from': 'fee_paid', 'to': 'active'},
],
},
{
'code': 'SUBSCRIPTION_RENEWAL',
'name_ar': 'سير عمل تجديد الاشتراك',
'name_en': 'Subscription Renewal Workflow',
'states': [
{'code': 'renewal_due', 'name_ar': 'مستحق التجديد', 'is_initial': True},
{'code': 'notification_sent', 'name_ar': 'تم الإخطار'},
{'code': 'paid', 'name_ar': 'تم السداد', 'is_terminal': True},
{'code': 'overdue', 'name_ar': 'متأخر'},
{'code': 'fine_applied', 'name_ar': 'تم تطبيق الغرامة'},
{'code': 'paid_with_fine', 'name_ar': 'سداد مع غرامة', 'is_terminal': True},
{'code': 'membership_dropped', 'name_ar': 'إسقاط العضوية', 'is_terminal': True},
],
'transitions': [
{'from': 'renewal_due', 'to': 'notification_sent'},
{'from': 'notification_sent', 'to': 'paid'},
{'from': 'notification_sent', 'to': 'overdue'},
{'from': 'overdue', 'to': 'fine_applied'},
{'from': 'fine_applied', 'to': 'paid_with_fine'},
{'from': 'fine_applied', 'to': 'membership_dropped'},
],
},
{
'code': 'TRANSFER_SEPARATION',
'name_ar': 'سير عمل التحويل والفصل',
'name_en': 'Transfer/Separation Workflow',
'states': [
{'code': 'request_submitted', 'name_ar': 'تم تقديم الطلب', 'is_initial': True},
{'code': 'under_review', 'name_ar': 'قيد المراجعة'},
{'code': 'board_approved', 'name_ar': 'موافقة مجلس الأمناء'},
{'code': 'fee_calculated', 'name_ar': 'تم حساب الرسوم'},
{'code': 'fee_paid', 'name_ar': 'تم سداد الرسوم'},
{'code': 'completed', 'name_ar': 'مكتمل', 'is_terminal': True},
{'code': 'rejected', 'name_ar': 'مرفوض', 'is_terminal': True},
],
'transitions': [
{'from': 'request_submitted', 'to': 'under_review'},
{'from': 'under_review', 'to': 'board_approved', 'permission': 'transfer.approve'},
{'from': 'under_review', 'to': 'rejected'},
{'from': 'board_approved', 'to': 'fee_calculated'},
{'from': 'fee_calculated', 'to': 'fee_paid'},
{'from': 'fee_paid', 'to': 'completed'},
],
},
{
'code': 'VIOLATION_PENALTY',
'name_ar': 'سير عمل المخالفات والجزاءات',
'name_en': 'Violation/Penalty Workflow',
'states': [
{'code': 'violation_reported', 'name_ar': 'تم الإبلاغ عن مخالفة', 'is_initial': True},
{'code': 'under_review', 'name_ar': 'قيد المراجعة'},
{'code': 'penalty_decided', 'name_ar': 'تم تحديد العقوبة'},
{'code': 'penalty_active', 'name_ar': 'عقوبة سارية'},
{'code': 'appealed', 'name_ar': 'تم التظلم'},
{'code': 'appeal_reviewed', 'name_ar': 'تمت مراجعة التظلم'},
{'code': 'penalty_upheld', 'name_ar': 'تأكيد العقوبة', 'is_terminal': True},
{'code': 'penalty_modified', 'name_ar': 'تعديل العقوبة', 'is_terminal': True},
{'code': 'penalty_cancelled', 'name_ar': 'إلغاء العقوبة', 'is_terminal': True},
{'code': 'served', 'name_ar': 'تم التنفيذ', 'is_terminal': True},
],
'transitions': [
{'from': 'violation_reported', 'to': 'under_review'},
{'from': 'under_review', 'to': 'penalty_decided', 'permission': 'fine.impose'},
{'from': 'penalty_decided', 'to': 'penalty_active'},
{'from': 'penalty_decided', 'to': 'appealed'},
{'from': 'penalty_active', 'to': 'appealed'},
{'from': 'penalty_active', 'to': 'served'},
{'from': 'appealed', 'to': 'appeal_reviewed'},
{'from': 'appeal_reviewed', 'to': 'penalty_upheld'},
{'from': 'appeal_reviewed', 'to': 'penalty_modified'},
{'from': 'appeal_reviewed', 'to': 'penalty_cancelled'},
],
},
]
def seed_workflows():
from .models import WorkflowDefinition
created = 0
for wf_data in WORKFLOW_DATA:
_, was_created = WorkflowDefinition.objects.update_or_create(
code=wf_data['code'],
defaults={
'name_ar': wf_data['name_ar'],
'name_en': wf_data.get('name_en', ''),
'states': wf_data['states'],
'transitions': wf_data['transitions'],
'is_active': True,
},
)
if was_created:
created += 1
return created
\ No newline at end of file
backend/apps/workflows/serializers.py
View file @ e666e669
# Serializers for workflows — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow Serializers
"""
from rest_framework import serializers
from .models import WorkflowDefinition, WorkflowInstance, WorkflowTransitionLog
class WorkflowDefinitionSerializer(serializers.ModelSerializer):
class Meta:
model = WorkflowDefinition
fields = [
'id', 'code', 'name_ar', 'name_en', 'states',
'transitions', 'is_active', 'created_at',
]
read_only_fields = ['created_at']
class WorkflowTransitionLogSerializer(serializers.ModelSerializer):
actor_name = serializers.CharField(source='actor.full_name_ar', read_only=True, default='')
class Meta:
model = WorkflowTransitionLog
fields = ['id', 'from_state', 'to_state', 'actor', 'actor_name', 'notes', 'timestamp']
class WorkflowInstanceSerializer(serializers.ModelSerializer):
definition_code = serializers.CharField(source='definition.code', read_only=True)
definition_name = serializers.CharField(source='definition.name_ar', read_only=True)
transition_logs = WorkflowTransitionLogSerializer(many=True, read_only=True)
class Meta:
model = WorkflowInstance
fields = [
'id', 'definition', 'definition_code', 'definition_name',
'entity_type', 'entity_id', 'current_state', 'data',
'started_at', 'completed_at', 'is_active', 'transition_logs',
]
read_only_fields = ['started_at', 'completed_at']
class TransitionRequestSerializer(serializers.Serializer):
to_state = serializers.CharField()
notes = serializers.CharField(required=False, default='')
\ No newline at end of file
backend/apps/workflows/tests.py
View file @ e666e669
# Tests for workflows — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow Tests
"""
from django.test import TestCase
from django.contrib.auth import get_user_model
from .models import WorkflowDefinition, WorkflowInstance
from .engine import StateMachineEngine, WorkflowError
from .seed import seed_workflows
Employee = get_user_model()
class WorkflowSeedTest(TestCase):
def test_seed_creates_definitions(self):
created = seed_workflows()
self.assertEqual(created, 5)
self.assertTrue(WorkflowDefinition.objects.filter(code='NEW_MEMBERSHIP').exists())
class StateMachineEngineTest(TestCase):
@classmethod
def setUpTestData(cls):
seed_workflows()
cls.employee = Employee.objects.create_user(
username='workflow_test', password='testpass123',
full_name_ar='موظف اختبار',
)
def test_start_workflow(self):
instance = StateMachineEngine.start_workflow(
'NEW_MEMBERSHIP', 'member', 1, self.employee,
)
self.assertEqual(instance.current_state, 'form_submitted')
self.assertTrue(instance.is_active)
self.assertEqual(instance.transition_logs.count(), 1)
def test_valid_transition(self):
instance = StateMachineEngine.start_workflow(
'NEW_MEMBERSHIP', 'member', 2, self.employee,
)
updated = StateMachineEngine.transition(
instance.id, 'under_review', self.employee, 'مراجعة البيانات',
)
self.assertEqual(updated.current_state, 'under_review')
self.assertEqual(instance.transition_logs.count(), 2)
def test_invalid_transition_raises(self):
instance = StateMachineEngine.start_workflow(
'NEW_MEMBERSHIP', 'member', 3, self.employee,
)
with self.assertRaises(WorkflowError):
StateMachineEngine.transition(instance.id, 'active', self.employee)
def test_terminal_state_completes_workflow(self):
instance = StateMachineEngine.start_workflow(
'NEW_MEMBERSHIP', 'member', 4, self.employee,
)
StateMachineEngine.transition(instance.id, 'under_review', self.employee)
StateMachineEngine.transition(instance.id, 'interview_scheduled', self.employee)
updated = StateMachineEngine.transition(instance.id, 'rejected', self.employee)
self.assertFalse(updated.is_active)
self.assertIsNotNone(updated.completed_at)
def test_get_available_transitions(self):
instance = StateMachineEngine.start_workflow(
'NEW_MEMBERSHIP', 'member', 5, self.employee,
)
transitions = StateMachineEngine.get_available_transitions(instance.id)
self.assertEqual(len(transitions), 1)
self.assertEqual(transitions[0]['to'], 'under_review')
def test_get_instance_for_entity(self):
StateMachineEngine.start_workflow(
'NEW_MEMBERSHIP', 'member', 99, self.employee,
)
found = StateMachineEngine.get_instance_for_entity('member', 99)
self.assertIsNotNone(found)
self.assertEqual(found.entity_id, 99)
\ No newline at end of file
backend/apps/workflows/urls.py
View file @ e666e669
urlpatterns = [] """
\ No newline at end of file THE CLUB ERP — Workflow URLs
"""
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from . import views
router = DefaultRouter()
router.register(r'definitions', views.WorkflowDefinitionViewSet, basename='workflowdefinition')
router.register(r'instances', views.WorkflowInstanceViewSet, basename='workflowinstance')
urlpatterns = [
path('', include(router.urls)),
]
\ No newline at end of file
backend/apps/workflows/views.py
View file @ e666e669
# Views for workflows — implemented in Phase 3 """
\ No newline at end of file THE CLUB ERP — Workflow Views
"""
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from rest_framework.filters import SearchFilter
from .models import WorkflowDefinition, WorkflowInstance
from .serializers import (
WorkflowDefinitionSerializer, WorkflowInstanceSerializer,
TransitionRequestSerializer,
)
from .engine import StateMachineEngine, WorkflowError
class WorkflowDefinitionViewSet(viewsets.ModelViewSet):
queryset = WorkflowDefinition.objects.all()
serializer_class = WorkflowDefinitionSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter]
filterset_fields = ['is_active']
search_fields = ['code', 'name_ar']
ordering = ['code']
class WorkflowInstanceViewSet(viewsets.ModelViewSet):
queryset = WorkflowInstance.objects.select_related('definition').prefetch_related('transition_logs').all()
serializer_class = WorkflowInstanceSerializer
permission_classes = [permissions.IsAuthenticated]
filter_backends = [DjangoFilterBackend, SearchFilter]
filterset_fields = ['entity_type', 'entity_id', 'current_state', 'is_active', 'definition__code']
search_fields = ['entity_id']
ordering = ['-started_at']
@action(detail=True, methods=['post'], url_path='transition')
def do_transition(self, request, pk=None):
ser = TransitionRequestSerializer(data=request.data)
ser.is_valid(raise_exception=True)
try:
instance = StateMachineEngine.transition(
instance_id=int(pk),
to_state=ser.validated_data['to_state'],
actor=request.user,
notes=ser.validated_data.get('notes', ''),
)
return Response(WorkflowInstanceSerializer(instance).data)
except WorkflowError as e:
return Response({'error': str(e)}, status=status.HTTP_400_BAD_REQUEST)
@action(detail=True, methods=['get'], url_path='available-transitions')
def available_transitions(self, request, pk=None):
transitions = StateMachineEngine.get_available_transitions(int(pk))
return Response({'transitions': transitions})
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment