Sicherhaven/eventify_backend
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d9a2af7168c9e1edab90e8173da95ea4c186dd9b
eventify_backend/admin_api/views.py

3316 lines
125 KiB
Python
Raw Blame History

from django.contrib.auth import authenticate, get_user_model
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework import status
from rest_framework_simplejwt.tokens import RefreshToken
from rest_framework_simplejwt.views import TokenRefreshView
from django.db import connection, transaction
from .serializers import UserSerializer
User = get_user_model()
class AdminLoginView(APIView):
permission_classes = [AllowAny]
def post(self, request):
identifier = request.data.get('username') or request.data.get('email')
password = request.data.get('password')
if not identifier or not password:
return Response({'error': 'username/email and password required'}, status=status.HTTP_400_BAD_REQUEST)
# Try username first, then email
user = authenticate(request, username=identifier, password=password)
if not user:
try:
u = User.objects.get(email=identifier)
user = authenticate(request, username=u.username, password=password)
except User.DoesNotExist:
pass
if not user:
_audit_log(request, 'auth.admin_login_failed', 'auth', 'failed',
{'identifier': identifier, 'reason': 'invalid_credentials'}, user=None)
return Response({'error': 'Invalid credentials'}, status=status.HTTP_401_UNAUTHORIZED)
if not user.is_active:
_audit_log(request, 'auth.admin_login_failed', 'auth', str(user.id),
{'identifier': identifier, 'reason': 'account_disabled'}, user=user)
return Response({'error': 'Account is disabled'}, status=status.HTTP_403_FORBIDDEN)
refresh = RefreshToken.for_user(user)
user_data = UserSerializer(user).data
# RBAC: prefer StaffProfile for allowed_modules and scopes
try:
sp = user.staff_profile
allowed_modules = sp.get_allowed_modules()
effective_scopes = sp.get_effective_scopes()
user_data['staff_role'] = sp.staff_role
user_data['department'] = sp.department.name if sp.department else None
user_data['squad'] = sp.squad.name if sp.squad else None
except Exception:
allowed_modules = user.get_allowed_modules()
effective_scopes = []
user_data['allowed_modules'] = allowed_modules
user_data['effective_scopes'] = effective_scopes
_audit_log(request, 'auth.admin_login', 'auth', str(user.id),
{'username': user.username, 'role': user.role}, user=user)
return Response({
'access': str(refresh.access_token),
'refresh': str(refresh),
'user': user_data,
})
class MeView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
me_data = UserSerializer(request.user).data
# RBAC: prefer StaffProfile for allowed_modules and scopes
try:
sp = request.user.staff_profile
allowed_modules = sp.get_allowed_modules()
effective_scopes = sp.get_effective_scopes()
me_data['staff_role'] = sp.staff_role
me_data['department'] = sp.department.name if sp.department else None
me_data['squad'] = sp.squad.name if sp.squad else None
except Exception:
allowed_modules = request.user.get_allowed_modules()
effective_scopes = []
me_data['allowed_modules'] = allowed_modules
me_data['effective_scopes'] = effective_scopes
return Response({'user': me_data})
class HealthView(APIView):
permission_classes = [AllowAny]
def get(self, request):
try:
connection.ensure_connection()
db_status = 'ok'
except Exception:
db_status = 'error'
return Response({'status': 'ok', 'db': db_status})
# ---------------------------------------------------------------------------
# Phase 2: Dashboard Views
# ---------------------------------------------------------------------------
class DashboardMetricsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from ledger.models import RazorpayTransaction
from partner.models import Partner
from events.models import Event
from bookings.models import Ticket, Booking
from django.db.models import Sum
from django.utils import timezone
import datetime
today = timezone.now().date()
# --- Revenue ---
total_paise = (
RazorpayTransaction.objects
.filter(status='captured')
.aggregate(total=Sum('amount'))['total'] or 0
)
total_revenue = total_paise / 100
# This-month / last-month revenue for growth
first_of_this_month = today.replace(day=1)
first_of_last_month = (first_of_this_month - datetime.timedelta(days=1)).replace(day=1)
this_month_paise = (
RazorpayTransaction.objects
.filter(status='captured', captured_at__date__gte=first_of_this_month)
.aggregate(total=Sum('amount'))['total'] or 0
)
last_month_paise = (
RazorpayTransaction.objects
.filter(
status='captured',
captured_at__date__gte=first_of_last_month,
captured_at__date__lt=first_of_this_month,
)
.aggregate(total=Sum('amount'))['total'] or 0
)
this_month_rev = this_month_paise / 100
last_month_rev = last_month_paise / 100
revenue_growth = (
round((this_month_rev - last_month_rev) / last_month_rev * 100, 2)
if last_month_rev else 0
)
# --- Partners ---
active_partners = Partner.objects.filter(status='active').count()
pending_partners = Partner.objects.filter(kyc_compliance_status='pending').count()
# --- Events ---
live_events = Event.objects.filter(event_status='live').count()
events_today = Event.objects.filter(start_date=today).count()
# --- Tickets ---
ticket_sales = Ticket.objects.count()
# This-week / last-week ticket growth
week_start = today - datetime.timedelta(days=today.weekday()) # Monday
last_week_start = week_start - datetime.timedelta(days=7)
this_week_tickets = Ticket.objects.filter(
booking__created_date__gte=week_start
).count()
last_week_tickets = Ticket.objects.filter(
booking__created_date__gte=last_week_start,
booking__created_date__lt=week_start,
).count()
ticket_growth = (
round((this_week_tickets - last_week_tickets) / last_week_tickets * 100, 2)
if last_week_tickets else 0
)
return Response({
'totalRevenue': total_revenue,
'revenueGrowth': revenue_growth,
'activePartners': active_partners,
'pendingPartners': pending_partners,
'liveEvents': live_events,
'eventsToday': events_today,
'ticketSales': ticket_sales,
'ticketGrowth': ticket_growth,
})
class DashboardRevenueView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from ledger.models import RazorpayTransaction
from banking_operations.models import PaymentTransaction
from django.db.models import Sum
from django.utils import timezone
import datetime
today = timezone.now().date()
result = []
for i in range(6, -1, -1):
day = today - datetime.timedelta(days=i)
rev_paise = (
RazorpayTransaction.objects
.filter(status='captured', captured_at__date=day)
.aggregate(total=Sum('amount'))['total'] or 0
)
revenue = rev_paise / 100
payouts = (
PaymentTransaction.objects
.filter(
payment_type='debit',
payment_transaction_status='completed',
payment_transaction_date=day,
)
.aggregate(total=Sum('payment_transaction_amount'))['total'] or 0
)
result.append({
'day': day.strftime('%a'),
'revenue': float(revenue),
'payouts': float(payouts),
})
return Response(result)
class DashboardActivityView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from partner.models import Partner
from events.models import Event
from bookings.models import Booking
from django.utils import timezone
items = []
# --- Partners (last 5 by id desc; no date field — timestamp=None, filtered out later) ---
for p in Partner.objects.order_by('-id')[:5]:
items.append({
'id': f'partner-{p.id}',
'type': 'partner',
'title': f'{p.name} registered',
'description': f'New partner — {getattr(p, "partner_type", "individual")}',
'timestamp': None,
'status': p.kyc_compliance_status,
})
# --- Events (last 5 by created_date desc) ---
for e in Event.objects.order_by('-created_date')[:5]:
display_name = e.title if getattr(e, 'title', None) else getattr(e, 'name', '')
ts = e.created_date
items.append({
'id': f'event-{e.id}',
'type': 'event',
'title': f'{display_name} created',
'description': f'Event status: {e.event_status}',
'timestamp': ts.isoformat() if ts else None,
'status': e.event_status,
})
# --- Bookings (last 5 by id desc) ---
for b in Booking.objects.order_by('-id')[:5]:
ts = b.created_date
items.append({
'id': f'booking-{b.booking_id}',
'type': 'booking',
'title': f'Booking {b.booking_id} placed',
'description': f'{b.quantity} ticket(s) at ₹{b.price}',
'timestamp': ts.isoformat() if ts else None,
'status': 'confirmed',
})
# Filter out items with no timestamp, sort desc, return top 10
dated = [item for item in items if item['timestamp'] is not None]
dated.sort(key=lambda x: x['timestamp'], reverse=True)
return Response(dated[:10])
class DashboardActionsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from partner.models import Partner
from events.models import Event
from banking_operations.models import PaymentTransaction
from django.db.models import Sum
kyc_count = Partner.objects.filter(kyc_compliance_status='pending').count()
flagged_count = Event.objects.filter(event_status='flagged').count()
pending_payouts = float(
PaymentTransaction.objects
.filter(payment_type='debit', payment_transaction_status='pending')
.aggregate(total=Sum('payment_transaction_amount'))['total'] or 0
)
actions = [
{
'id': 'kyc',
'type': 'kyc',
'count': kyc_count,
'title': 'Partner Approval Queue',
'description': f'{kyc_count} partners awaiting KYC review',
'href': '/partners?filter=pending',
'priority': 'high',
},
{
'id': 'flagged',
'type': 'flagged',
'count': flagged_count,
'title': 'Flagged Events',
'description': f'{flagged_count} events reported for review',
'href': '/events?filter=flagged',
'priority': 'high',
},
{
'id': 'payout',
'type': 'payout',
'count': pending_payouts,
'title': 'Pending Payouts',
'description': f'₹{pending_payouts:,.0f} ready for release',
'href': '/financials?tab=payouts',
'priority': 'medium',
},
]
return Response(actions)
# ---------------------------------------------------------------------------
# Phase 3: Partner helpers
# ---------------------------------------------------------------------------
_PARTNER_STATUS_MAP = {
'active': 'Active', 'pending': 'Invited', 'suspended': 'Suspended',
'archived': 'Archived', 'deleted': 'Archived', 'inactive': 'Archived',
}
_PARTNER_KYC_MAP = {'approved': 'Verified', 'rejected': 'Rejected'}
_PARTNER_TYPE_MAP = {
'venue': 'Venue', 'promoter': 'Promoter', 'sponsor': 'Sponsor',
'vendor': 'Vendor', 'affiliate': 'Affiliate', 'other': 'Other',
}
_RISK_MAP = {
'high_risk': 80, 'medium_risk': 45, 'low_risk': 15,
'rejected': 90, 'pending': 30, 'approved': 5,
}
def _partner_kyc_docs(p):
if not p.kyc_compliance_document_type:
return []
return [{
'id': f'kyc-{p.id}',
'partnerId': str(p.id),
'type': p.kyc_compliance_document_type.upper(),
'name': p.kyc_compliance_document_other_type or p.kyc_compliance_document_type,
'url': p.kyc_compliance_document_file.url if p.kyc_compliance_document_file else '',
'status': {'approved': 'APPROVED', 'rejected': 'REJECTED'}.get(p.kyc_compliance_status, 'PENDING'),
'mandatory': True,
'adminNote': p.kyc_compliance_reason or '',
'uploadedBy': p.primary_contact_person_name,
'uploadedAt': '',
}]
def _serialize_partner(p, events_count=0):
addr = ', '.join(filter(None, [p.address, p.city, p.state, p.country]))
return {
'id': str(p.id),
'name': p.name,
'type': _PARTNER_TYPE_MAP.get(p.partner_type, 'Other'),
'status': _PARTNER_STATUS_MAP.get(p.status, 'Invited'),
'primaryContact': {
'name': p.primary_contact_person_name,
'email': p.primary_contact_person_email,
'phone': p.primary_contact_person_phone,
},
'companyDetails': {
'website': p.website_url or '',
'address': addr,
},
'metrics': {
'eventsCount': events_count,
'totalRevenue': 0,
'openBalance': 0,
'activeDeals': 0,
'lastActivity': None,
},
'verificationStatus': _PARTNER_KYC_MAP.get(p.kyc_compliance_status, 'Pending'),
'kycComplianceStatus': p.kyc_compliance_status,
'riskScore': _RISK_MAP.get(p.kyc_compliance_status, 0),
'joinedAt': None,
'tags': [],
'notes': p.kyc_compliance_reason or '',
'kycDocuments': _partner_kyc_docs(p),
}
# ---------------------------------------------------------------------------
# Phase 3: Partner Views
# ---------------------------------------------------------------------------
class PartnerStatsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from partner.models import Partner
return Response({
'total': Partner.objects.count(),
'active': Partner.objects.filter(status='active').count(),
'pendingKyc': Partner.objects.filter(kyc_compliance_status='pending').count(),
'highRisk': Partner.objects.filter(kyc_compliance_status='high_risk').count(),
})
class PartnerListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from partner.models import Partner
from django.db.models import Count, Q
qs = Partner.objects.annotate(events_count=Count('event'))
if s := request.GET.get('status'): qs = qs.filter(status=s)
if k := request.GET.get('kyc_status'): qs = qs.filter(kyc_compliance_status=k)
if t := request.GET.get('partner_type'): qs = qs.filter(partner_type=t)
if q := request.GET.get('search'):
qs = qs.filter(
Q(name__icontains=q) |
Q(primary_contact_person_email__icontains=q) |
Q(primary_contact_person_name__icontains=q)
)
page = max(1, int(request.GET.get('page', 1)))
page_size = min(100, int(request.GET.get('page_size', 20)))
total = qs.count()
partners = qs.order_by('-id')[(page - 1) * page_size: page * page_size]
return Response({
'count': total,
'results': [_serialize_partner(p, p.events_count) for p in partners],
})
class PartnerDetailView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, pk):
from partner.models import Partner
from events.models import Event
from django.shortcuts import get_object_or_404
p = get_object_or_404(Partner, pk=pk)
events_count = Event.objects.filter(partner_id=pk).count()
data = _serialize_partner(p, events_count)
_EVENT_STATUS_MAP = {
'live': 'LIVE', 'published': 'LIVE', 'draft': 'DRAFT',
'cancelled': 'CANCELLED', 'flagged': 'PENDING_REVIEW', 'pending': 'PENDING_REVIEW',
}
events_qs = Event.objects.filter(partner_id=pk).order_by('-id')[:20]
data['events'] = [{
'id': str(e.id),
'partnerId': str(pk),
'title': e.title or e.name or '',
'date': e.start_date.isoformat() if e.start_date else '',
'venue': e.venue_name or '',
'category': '',
'ticketPrice': 0,
'totalTickets': 0,
'ticketsSold': 0,
'revenue': 0,
'status': _EVENT_STATUS_MAP.get(e.event_status, 'DRAFT'),
'submittedAt': e.created_date.isoformat() if e.created_date else '',
'createdAt': e.created_date.isoformat() if e.created_date else '',
'rejectionReason': '',
} for e in events_qs]
data['dealTerms'] = []
data['ledger'] = []
return Response(data)
class PartnerStatusView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
from partner.models import Partner
from django.shortcuts import get_object_or_404
p = get_object_or_404(Partner, pk=pk)
new_status = request.data.get('status')
valid = ('active', 'suspended', 'inactive', 'archived')
if new_status not in valid:
return Response({'error': f'status must be one of {valid}'}, status=400)
prev_status = p.status
p.status = new_status
p.kyc_compliance_reason = request.data.get('reason', p.kyc_compliance_reason or '')
with transaction.atomic():
p.save(update_fields=['status', 'kyc_compliance_reason'])
_audit_log(request, 'partner.status_changed', 'partner', p.id, {
'previous_status': prev_status,
'new_status': p.status,
'reason': request.data.get('reason', '') or '',
'partner_name': p.name,
})
return Response({'id': str(p.id), 'status': p.status})
class PartnerKYCReviewView(APIView):
permission_classes = [IsAuthenticated]
_DECISION_SLUG = {
'approved': 'partner.kyc.approved',
'rejected': 'partner.kyc.rejected',
'requested_info': 'partner.kyc.requested_info',
}
def post(self, request, pk):
from partner.models import Partner
from django.shortcuts import get_object_or_404
p = get_object_or_404(Partner, pk=pk)
decision = request.data.get('decision')
audit_slug = self._DECISION_SLUG.get(decision)
if audit_slug is None:
return Response(
{'error': 'decision must be approved, rejected, or requested_info'},
status=400,
)
reason = request.data.get('reason', '') or ''
docs_requested = request.data.get('docs_requested') or []
previous_status = p.kyc_compliance_status
previous_is_compliant = bool(p.is_kyc_compliant)
with transaction.atomic():
if decision in ('approved', 'rejected'):
p.kyc_compliance_status = decision
p.is_kyc_compliant = (decision == 'approved')
# `requested_info` leaves the compliance state intact — it's a
# reviewer signalling they need more documents from the partner.
p.kyc_compliance_reason = reason
p.save(update_fields=[
'kyc_compliance_status', 'is_kyc_compliant', 'kyc_compliance_reason',
])
_audit_log(
request,
action=audit_slug,
target_type='partner',
target_id=p.id,
details={
'reason': reason,
'docs_requested': docs_requested,
'previous_status': previous_status,
'new_status': p.kyc_compliance_status,
'previous_is_compliant': previous_is_compliant,
'new_is_compliant': bool(p.is_kyc_compliant),
},
)
# Preserve the existing response shape — frontend depends on it.
if decision == 'approved':
verification_status = 'Verified'
elif decision == 'rejected':
verification_status = 'Rejected'
else:
verification_status = 'Info Requested'
return Response({
'id': str(p.id),
'kyc_compliance_status': p.kyc_compliance_status,
'is_kyc_compliant': p.is_kyc_compliant,
'verificationStatus': verification_status,
})
# ---------------------------------------------------------------------------
# Phase 4: Users API
# ---------------------------------------------------------------------------
def _user_status(u):
return 'Active' if u.is_active else 'Suspended'
_USER_ROLE_MAP = {
'admin': 'Admin', 'manager': 'Admin',
'staff': 'Support Agent',
'customer': 'User', 'is_user': 'User',
'partner': 'Partner', 'partner_manager': 'Partner',
'partner_staff': 'Partner', 'partner_customer': 'Partner',
}
def _serialize_user(u):
full_name = f'{u.first_name} {u.last_name}'.strip() or u.username
role_key = u.role if u.role else ('customer' if getattr(u, 'is_customer', False) else 'staff')
try:
pic = u.profile_picture
avatar = pic.url if pic and pic.name and pic.name != 'default.png' else ''
except Exception:
avatar = ''
return {
'id': str(u.id),
'eventifyId': u.eventify_id or '',
'name': full_name,
'email': u.email,
'phone': getattr(u, 'phone_number', '') or '',
'countryCode': '+91',
'avatarUrl': avatar,
'role': _USER_ROLE_MAP.get(role_key, 'User'),
'status': _user_status(u),
'tier': 'Bronze',
'healthScore': 'warm',
'isVerified': u.is_active,
'is2FAEnabled': False,
'totalSpent': 0,
'bookingsCount': 0,
'refundRate': 0,
'averageOrderValue': 0,
'language': 'en',
'timezone': 'Asia/Kolkata',
'currency': 'INR',
'emailNotifications': True,
'pushNotifications': True,
'smsNotifications': True,
'tags': [],
'pinnedNote': '',
'createdAt': u.date_joined.isoformat() if u.date_joined else '',
'updatedAt': u.date_joined.isoformat() if u.date_joined else '',
'lastLoginAt': u.last_login.isoformat() if u.last_login else '',
'lastActivityAt': u.last_login.isoformat() if u.last_login else '',
}
class UserMetricsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from django.contrib.auth import get_user_model
from django.db.models import Q
from django.utils import timezone
import datetime
User = get_user_model()
today = timezone.now().date()
week_ago = today - datetime.timedelta(days=7)
# Customers = all non-superuser accounts (end users registered via mobile/web)
customer_qs = User.objects.filter(is_superuser=False)
return Response({
'total': customer_qs.count(),
'active': customer_qs.filter(is_active=True).count(),
'suspended': customer_qs.filter(is_active=False).count(),
'banned': 0, # Reserved for future explicit ban field
'newThisWeek': customer_qs.filter(date_joined__date__gte=week_ago).count(),
})
class UserListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from django.contrib.auth import get_user_model
from django.db.models import Q
User = get_user_model()
include_all = request.query_params.get('include_all', '0') == '1'
qs = User.objects.all() if include_all else User.objects.filter(is_superuser=False)
# Server-side search
search = request.query_params.get('search', '').strip()
if search:
qs = qs.filter(
Q(first_name__icontains=search) |
Q(last_name__icontains=search) |
Q(email__icontains=search) |
Q(username__icontains=search) |
Q(phone_number__icontains=search) |
Q(eventify_id__icontains=search)
)
# Status filter
status_filter = request.query_params.get('status', '').strip().lower()
if status_filter == 'active':
qs = qs.filter(is_active=True)
elif status_filter == 'suspended':
qs = qs.filter(is_active=False)
# Role filter
role_filter = request.query_params.get('role', '').strip()
if role_filter:
qs = qs.filter(role__iexact=role_filter)
try:
page = max(1, int(request.query_params.get('page', 1)))
page_size = min(100, int(request.query_params.get('page_size', 20)))
except (ValueError, TypeError):
page, page_size = 1, 20
total = qs.count()
users = qs.order_by('-date_joined')[(page - 1) * page_size: page * page_size]
return Response({'count': total, 'results': [_serialize_user(u) for u in users]})
class UserDetailView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, pk):
from django.contrib.auth import get_user_model
from django.shortcuts import get_object_or_404
User = get_user_model()
u = get_object_or_404(User, pk=pk)
return Response(_serialize_user(u))
class UserStatusView(APIView):
permission_classes = [IsAuthenticated]
# Map the external `action` slug sent by the admin dashboard to the
# audit-log slug we record + the `is_active` value that slug implies.
# Keeping the mapping data-driven means a new moderation verb (e.g.
# `flag`) only needs one new row, not scattered `if` branches.
_ACTION_MAP = {
'suspend': ('user.suspended', False),
'ban': ('user.banned', False),
'reinstate': ('user.reinstated', True),
'flag': ('user.flagged', None),
}
def patch(self, request, pk):
from django.contrib.auth import get_user_model
from django.shortcuts import get_object_or_404
User = get_user_model()
u = get_object_or_404(User, pk=pk)
action = request.data.get('action')
mapping = self._ACTION_MAP.get(action)
if mapping is None:
return Response(
{'error': 'action must be suspend, ban, reinstate, or flag'},
status=400,
)
audit_slug, new_active = mapping
reason = request.data.get('reason', '') or ''
previous_status = _user_status(u)
# Audit + state change must succeed or fail together — otherwise the
# log says one thing and the database says another.
with transaction.atomic():
if new_active is not None and u.is_active != new_active:
u.is_active = new_active
u.save(update_fields=['is_active'])
_audit_log(
request,
action=audit_slug,
target_type='user',
target_id=u.id,
details={
'reason': reason,
'previous_status': previous_status,
'new_status': _user_status(u),
},
)
return Response({'id': str(u.id), 'status': _user_status(u)})
# ---------------------------------------------------------------------------
# Phase 5: Events API
# ---------------------------------------------------------------------------
_EVENT_STATUS_MAP = {
'live': 'live', 'published': 'published',
'pending': 'published', 'created': 'draft',
'flagged': 'flagged', 'cancelled': 'cancelled',
'postponed': 'cancelled', 'completed': 'completed',
}
def _serialize_event(e):
partner_name = ''
if e.partner_id:
try:
partner_name = e.partner.name
except Exception:
partner_name = ''
return {
'id': str(e.id),
'title': e.title or getattr(e, 'name', '') or '',
'partnerId': str(e.partner_id) if e.partner_id else '',
'partnerName': partner_name,
'date': e.start_date.isoformat() if e.start_date else '',
'status': _EVENT_STATUS_MAP.get(e.event_status, 'draft'),
'ticketsSold': 0,
'revenue': 0,
'venueName': e.venue_name or '',
'createdAt': e.created_date.isoformat() if e.created_date else '',
'isFeatured': bool(e.is_featured),
'isTopEvent': bool(e.is_top_event),
'source': e.source or 'eventify',
'contributedBy': getattr(e, 'contributed_by', '') or '',
'eventTypeId': e.event_type_id,
'eventTypeName': e.event_type.event_type if e.event_type_id and e.event_type else '',
}
def _serialize_event_detail(e):
"""Full event serializer for detail view -- includes all fields + images."""
base = _serialize_event(e)
base.update({
'name': e.name or '',
'description': e.description or '',
'endDate': e.end_date.isoformat() if e.end_date else '',
'startTime': str(e.start_time or ''),
'endTime': str(e.end_time or ''),
'allYearEvent': bool(e.all_year_event),
'latitude': str(e.latitude) if e.latitude else '',
'longitude': str(e.longitude) if e.longitude else '',
'pincode': e.pincode or '',
'district': e.district or '',
'state': e.state or '',
'place': e.place or '',
'isBookable': bool(e.is_bookable),
'eventType': e.event_type_id,
'importantInformation': e.important_information or '',
'source': e.source or '',
'cancelledReason': e.cancelled_reason or '',
'outsideEventUrl': e.outside_event_url or '',
'images': [
{
'id': img.id,
'url': f'/media/{img.event_image}',
'isPrimary': bool(img.is_primary),
}
for img in e.eventimages_set.all()
],
})
return base
class EventStatsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from events.models import Event
return Response({
'total': Event.objects.count(),
'live': Event.objects.filter(event_status='live').count(),
'pending': Event.objects.filter(event_status__in=['pending', 'created']).count(),
'flagged': Event.objects.filter(event_status='flagged').count(),
'published': Event.objects.filter(event_status='published').count(),
})
class EventListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from events.models import Event
from django.db.models import Q
qs = Event.objects.select_related('partner', 'event_type').all()
if s := request.GET.get('status'):
reverse_map = {v: k for k, v in _EVENT_STATUS_MAP.items()}
backend_status = reverse_map.get(s, s)
qs = qs.filter(event_status=backend_status)
if etid := request.GET.get('event_type'):
qs = qs.filter(event_type_id=etid)
if pid := request.GET.get('partner_id'):
qs = qs.filter(partner_id=pid)
if q := request.GET.get('search'):
qs = qs.filter(
Q(title__icontains=q) | Q(name__icontains=q) | Q(venue_name__icontains=q)
)
if date_from := request.GET.get('date_from'):
qs = qs.filter(start_date__gte=date_from)
if date_to := request.GET.get('date_to'):
qs = qs.filter(start_date__lte=date_to)
try:
page = max(1, int(request.GET.get('page', 1)))
page_size = min(100, int(request.GET.get('page_size', 20)))
except (ValueError, TypeError):
page, page_size = 1, 20
total = qs.count()
events = qs.order_by('-id')[(page - 1) * page_size: page * page_size]
return Response({'count': total, 'results': [_serialize_event(e) for e in events]})
class EventDetailView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, pk):
from events.models import Event
from django.shortcuts import get_object_or_404
e = get_object_or_404(Event.objects.select_related('partner').prefetch_related('eventimages_set'), pk=pk)
return Response(_serialize_event_detail(e))
class EventUpdateView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
from events.models import Event
from django.shortcuts import get_object_or_404
e = get_object_or_404(Event, pk=pk)
field_map = {
'title': 'title',
'name': 'name',
'description': 'description',
'venueName': 'venue_name',
'place': 'place',
'district': 'district',
'state': 'state',
'pincode': 'pincode',
'importantInformation': 'important_information',
'source': 'source',
'contributedBy': 'contributed_by',
'cancelledReason': 'cancelled_reason',
'outsideEventUrl': 'outside_event_url',
}
bool_fields = {
'isBookable': 'is_bookable',
'isFeatured': 'is_featured',
'isTopEvent': 'is_top_event',
'allYearEvent': 'all_year_event',
}
updated_fields = []
for api_key, model_field in field_map.items():
if api_key in request.data:
setattr(e, model_field, request.data[api_key] or '')
updated_fields.append(model_field)
for api_key, model_field in bool_fields.items():
if api_key in request.data:
setattr(e, model_field, bool(request.data[api_key]))
updated_fields.append(model_field)
# Handle status
if 'status' in request.data:
reverse_map = {v: k for k, v in _EVENT_STATUS_MAP.items()}
backend_status = reverse_map.get(request.data['status'], request.data['status'])
e.event_status = backend_status
updated_fields.append('event_status')
if updated_fields:
with transaction.atomic():
e.save(update_fields=updated_fields)
_audit_log(request, 'event.updated', 'event', e.id, {
'changed_fields': updated_fields,
'partner_id': str(e.partner_id) if e.partner_id else None,
})
# Re-fetch with relations for response
e = Event.objects.select_related('partner').prefetch_related('eventimages_set').get(pk=pk)
return Response(_serialize_event_detail(e))
class EventModerationView(APIView):
permission_classes = [IsAuthenticated]
_ACTION_SLUG = {
'approve': 'event.approved',
'reject': 'event.rejected',
'flag': 'event.flagged',
'feature': 'event.featured',
'unfeature': 'event.unfeatured',
}
def patch(self, request, pk):
from events.models import Event
from django.shortcuts import get_object_or_404
e = get_object_or_404(Event, pk=pk)
action = request.data.get('action')
audit_slug = self._ACTION_SLUG.get(action)
if audit_slug is None:
return Response({'error': 'Invalid action'}, status=400)
reason = request.data.get('reason', '') or ''
previous_status = e.event_status
previous_is_featured = bool(e.is_featured)
with transaction.atomic():
if action == 'approve':
e.event_status = 'published'
e.save(update_fields=['event_status'])
elif action == 'reject':
e.event_status = 'cancelled'
e.cancelled_reason = reason
e.save(update_fields=['event_status', 'cancelled_reason'])
elif action == 'flag':
e.event_status = 'flagged'
e.save(update_fields=['event_status'])
elif action == 'feature':
e.is_featured = True
e.save(update_fields=['is_featured'])
elif action == 'unfeature':
e.is_featured = False
e.save(update_fields=['is_featured'])
_audit_log(
request,
action=audit_slug,
target_type='event',
target_id=e.id,
details={
'reason': reason,
'partner_id': str(e.partner_id) if e.partner_id else None,
'previous_status': previous_status,
'new_status': e.event_status,
'previous_is_featured': previous_is_featured,
'new_is_featured': bool(e.is_featured),
},
)
e.refresh_from_db()
return Response(_serialize_event(e))
class EventDeleteView(APIView):
permission_classes = [IsAuthenticated]
def delete(self, request, pk):
from events.models import Event
from django.shortcuts import get_object_or_404
e = get_object_or_404(Event, pk=pk)
event_title = e.title or getattr(e, 'name', '') or ''
partner_id = str(e.partner_id) if e.partner_id else None
with transaction.atomic():
e.delete()
_audit_log(request, 'event.deleted', 'event', pk, {
'title': event_title,
'partner_id': partner_id,
})
return Response({'status': 'deleted'}, status=204)
# ---------------------------------------------------------------------------
# Phase 6: Financials & Payouts
# ---------------------------------------------------------------------------
_TX_STATUS_MAP = {
'captured': 'Completed',
'created': 'Pending',
'failed': 'Failed',
'refunded': 'Failed',
}
def _serialize_transaction(t):
ts = t.captured_at or t.created_at
order_ref = getattr(t, 'razorpay_order_id', None) or getattr(t, 'transaction_id', None)
return {
'id': str(t.id),
'title': f'Payment {order_ref}' if order_ref else f'Transaction #{t.id}',
'partner': '',
'amount': t.amount / 100,
'date': ts.isoformat() if ts else '',
'type': 'in',
'method': 'Razorpay',
'fees': 0,
'net': t.amount / 100,
'status': _TX_STATUS_MAP.get(t.status, 'Pending'),
}
_SETTLEMENT_STATUS_MAP = {
'pending': 'Ready',
'failed': 'Overdue',
'cancelled': 'Overdue',
'completed': 'On Hold',
'refunded': 'On Hold',
}
def _serialize_settlement(p):
return {
'id': str(p.id),
'partnerName': '',
'eventName': '',
'amount': float(p.payment_transaction_amount),
'dueDate': p.payment_transaction_date.isoformat() if p.payment_transaction_date else '',
'status': _SETTLEMENT_STATUS_MAP.get(p.payment_transaction_status, 'On Hold'),
}
class FinancialMetricsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from ledger.models import RazorpayTransaction
from banking_operations.models import PaymentTransaction
from django.db.models import Sum
total_paise = (
RazorpayTransaction.objects.filter(status='captured')
.aggregate(t=Sum('amount'))['t'] or 0
)
total_revenue = total_paise / 100
total_payouts = float(
PaymentTransaction.objects
.filter(payment_type='debit', payment_transaction_status='completed')
.aggregate(t=Sum('payment_transaction_amount'))['t'] or 0
)
platform_earnings = round(total_revenue * 0.12, 2)
pending_count = PaymentTransaction.objects.filter(
payment_type='debit', payment_transaction_status='pending'
).count()
pending_amount = float(
PaymentTransaction.objects
.filter(payment_type='debit', payment_transaction_status='pending')
.aggregate(t=Sum('payment_transaction_amount'))['t'] or 0
)
return Response({
'totalRevenue': total_revenue,
'totalPayouts': total_payouts,
'platformEarnings': platform_earnings,
'pendingPayouts': pending_count,
'pendingPayoutAmount': pending_amount,
})
class TransactionListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from ledger.models import RazorpayTransaction
try:
page = max(1, int(request.GET.get('page', 1)))
page_size = min(100, int(request.GET.get('page_size', 20)))
except (ValueError, TypeError):
page, page_size = 1, 20
qs = RazorpayTransaction.objects.order_by('-id')
total = qs.count()
txs = qs[(page - 1) * page_size: page * page_size]
return Response({'count': total, 'results': [_serialize_transaction(t) for t in txs]})
class SettlementListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from banking_operations.models import PaymentTransaction
qs = PaymentTransaction.objects.filter(
payment_type='debit'
).order_by('-id')[:50]
return Response([_serialize_settlement(p) for p in qs])
class SettlementReleaseView(APIView):
permission_classes = [IsAuthenticated]
def post(self, request, pk):
from banking_operations.models import PaymentTransaction
from django.shortcuts import get_object_or_404
p = get_object_or_404(PaymentTransaction, pk=pk, payment_type='debit')
prev_status = p.payment_transaction_status
p.payment_transaction_status = 'completed'
with transaction.atomic():
p.save(update_fields=['payment_transaction_status'])
_audit_log(request, 'settlement.released', 'settlement', p.id, {
'previous_status': prev_status,
'new_status': 'completed',
})
return Response(_serialize_settlement(p))
# ---------------------------------------------------------------------------
# Phase 7: Reviews Moderation
# ---------------------------------------------------------------------------
def _reviewer_rank(total_reviews):
if total_reviews >= 41:
return 'Legend'
if total_reviews >= 21:
return 'Champion'
if total_reviews >= 11:
return 'Enthusiast'
if total_reviews >= 4:
return 'Contributor'
return 'Explorer'
def _serialize_review(r):
from admin_api.models import Review
try:
name = r.reviewer.get_full_name() or r.reviewer.username
email = r.reviewer.email
total = Review.objects.filter(reviewer=r.reviewer, status='live').count()
except Exception:
name, email, total = '', '', 0
try:
event_name = getattr(r.event, 'title', None) or getattr(r.event, 'name', '') or ''
event_id = str(r.event.id)
start = getattr(r.event, 'start_date', None)
event_date = start.isoformat() if start else ''
except Exception:
event_name, event_id, event_date = '', '', ''
return {
'id': str(r.id),
'reviewerName': name,
'reviewerEmail': email,
'reviewerAvatar': '',
'eventName': event_name,
'eventId': event_id,
'eventDate': event_date,
'rating': r.rating,
'reviewText': r.review_text,
'submissionDate': r.submission_date.isoformat(),
'status': r.status,
'reviewerRank': _reviewer_rank(total),
'reviewerTotalReviews': total,
'rejectReason': r.reject_reason or '',
}
class ReviewMetricsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from admin_api.models import Review
return Response({
'totalPending': Review.objects.filter(status='pending').count(),
'liveReviews': Review.objects.filter(status='live').count(),
'rejected': Review.objects.filter(status='rejected').count(),
})
class ReviewListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from admin_api.models import Review
qs = Review.objects.select_related('reviewer', 'event').order_by('-submission_date')
status = request.GET.get('status')
if status in ('pending', 'live', 'rejected'):
qs = qs.filter(status=status)
try:
page = max(1, int(request.GET.get('page', 1)))
page_size = min(100, int(request.GET.get('page_size', 20)))
except (ValueError, TypeError):
page, page_size = 1, 20
total = qs.count()
reviews = qs[(page - 1) * page_size: page * page_size]
return Response({'count': total, 'results': [_serialize_review(r) for r in reviews]})
class ReviewModerationView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
from admin_api.models import Review
from django.shortcuts import get_object_or_404
review = get_object_or_404(Review, pk=pk)
action = request.data.get('action')
previous_status = review.status
previous_text = review.review_text
audit_slug = None
details = {}
if action == 'approve':
review.status = 'live'
audit_slug = 'review.approved'
elif action == 'reject':
rr = request.data.get('reject_reason', 'spam')
valid_reasons = [c[0] for c in Review.REJECT_CHOICES]
if rr not in valid_reasons:
return Response({'error': 'Invalid reject_reason'}, status=400)
review.status = 'rejected'
review.reject_reason = rr
audit_slug = 'review.rejected'
details['reject_reason'] = rr
elif action == 'save_and_approve':
review.review_text = request.data.get('review_text', review.review_text)
review.status = 'live'
audit_slug = 'review.edited'
details['edited_text'] = True
elif action == 'save_live':
review.review_text = request.data.get('review_text', review.review_text)
audit_slug = 'review.edited'
details['edited_text'] = True
else:
return Response({'error': 'Invalid action'}, status=400)
details.update({
'previous_status': previous_status,
'new_status': review.status,
})
if previous_text != review.review_text:
details['original_text'] = previous_text
with transaction.atomic():
review.save()
_audit_log(
request,
action=audit_slug,
target_type='review',
target_id=review.id,
details=details,
)
return Response(_serialize_review(review))
class ReviewDeleteView(APIView):
permission_classes = [IsAuthenticated]
def delete(self, request, pk):
from admin_api.models import Review
from django.shortcuts import get_object_or_404
review = get_object_or_404(Review, pk=pk)
reviewer_id = str(review.reviewer_id) if review.reviewer_id else None
event_id = str(review.event_id) if review.event_id else None
rating = review.rating
with transaction.atomic():
review.delete()
_audit_log(request, 'review.deleted', 'review', pk, {
'reviewer_user_id': reviewer_id,
'event_id': event_id,
'rating': rating,
})
return Response(status=204)
# --- Payment Gateway Settings ---
def _serialize_gateway(gw, include_secret=False):
data = {
'id': gw.pk,
'payment_gateway_id': gw.payment_gateway_id,
'name': gw.payment_gateway_name,
'description': gw.payment_gateway_description,
'url': gw.payment_gateway_url,
'api_key': gw.payment_gateway_api_key,
'api_url': gw.payment_gateway_api_url,
'api_version': gw.payment_gateway_api_version,
'api_method': gw.payment_gateway_api_method,
'is_active': gw.is_active,
'gateway_priority': gw.gateway_priority,
'created_date': gw.created_date.isoformat() if gw.created_date else None,
'updated_date': gw.updated_date.isoformat() if gw.updated_date else None,
'logo': gw.payment_gateway_logo.url if gw.payment_gateway_logo else None,
}
if include_secret:
data['api_secret'] = gw.payment_gateway_api_secret
return data
class ActivePaymentGatewayView(APIView):
permission_classes = [AllowAny]
def get(self, request):
from banking_operations.models import PaymentGateway
gateway = PaymentGateway.objects.filter(is_active=True).order_by('-gateway_priority', '-id').first()
if not gateway:
return Response({'status': 'error', 'message': 'No active payment gateway configured.'}, status=404)
return Response({
'status': 'success',
'gateway': {
'name': gateway.payment_gateway_name,
'key_id': gateway.payment_gateway_api_key,
'currency': 'INR',
}
})
class PaymentGatewaySettingsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, pk=None):
from banking_operations.models import PaymentGateway
gateways = PaymentGateway.objects.all().order_by('-gateway_priority', '-id')
return Response({
'status': 'success',
'payment_gateways': [_serialize_gateway(g, include_secret=True) for g in gateways]
})
def post(self, request, pk=None):
from banking_operations.models import PaymentGateway
import uuid
d = request.data
required = ['name', 'api_key', 'api_secret']
missing = [f for f in required if not d.get(f)]
if missing:
return Response({'status': 'error', 'message': 'Missing fields: {}'.format(missing)}, status=400)
gw = PaymentGateway.objects.create(
payment_gateway_id=str(uuid.uuid4().hex[:10]).upper(),
payment_gateway_name=d['name'],
payment_gateway_description=d.get('description', ''),
payment_gateway_url=d.get('url', ''),
payment_gateway_api_key=d['api_key'],
payment_gateway_api_secret=d['api_secret'],
payment_gateway_api_url=d.get('api_url', ''),
payment_gateway_api_version=d.get('api_version', 'v1'),
payment_gateway_api_method=d.get('api_method', 'POST'),
is_active=d.get('is_active', True),
gateway_priority=int(d.get('gateway_priority', 0)),
)
_audit_log(request, 'gateway.created', 'gateway', gw.pk, {
'gateway_name': gw.payment_gateway_name,
'is_active': gw.is_active,
})
return Response({'status': 'success', 'payment_gateway': _serialize_gateway(gw, include_secret=True)}, status=201)
def patch(self, request, pk=None):
from banking_operations.models import PaymentGateway
from django.shortcuts import get_object_or_404
gw = get_object_or_404(PaymentGateway, pk=pk)
d = request.data
field_map = {
'name': 'payment_gateway_name',
'description': 'payment_gateway_description',
'url': 'payment_gateway_url',
'api_key': 'payment_gateway_api_key',
'api_secret': 'payment_gateway_api_secret',
'api_url': 'payment_gateway_api_url',
'api_version': 'payment_gateway_api_version',
'api_method': 'payment_gateway_api_method',
'is_active': 'is_active',
'gateway_priority': 'gateway_priority',
}
changed_fields = [cf for cf in field_map if cf in d]
for client_field, model_field in field_map.items():
if client_field in d:
setattr(gw, model_field, d[client_field])
gw.save()
_audit_log(request, 'gateway.updated', 'gateway', gw.pk, {
'gateway_name': gw.payment_gateway_name,
'changed_fields': changed_fields,
'is_active': gw.is_active,
})
return Response({'status': 'success', 'payment_gateway': _serialize_gateway(gw, include_secret=True)})
def delete(self, request, pk=None):
from banking_operations.models import PaymentGateway
from django.shortcuts import get_object_or_404
gw = get_object_or_404(PaymentGateway, pk=pk)
gw_name = gw.payment_gateway_name
gw_pk = gw.pk
gw.delete()
_audit_log(request, 'gateway.deleted', 'gateway', gw_pk, {'gateway_name': gw_name})
return Response({'status': 'success'}, status=200)
class EventCreateView(APIView):
permission_classes = [IsAuthenticated]
def post(self, request):
from events.models import Event, EventType
data = request.data
# Required fields
title = (data.get('title') or '').strip()
name = (data.get('name') or title).strip()
if not title:
return Response({'error': 'Title is required'}, status=400)
# Get event_type (required FK)
event_type_id = data.get('eventType')
if not event_type_id:
return Response({'error': 'Event type is required'}, status=400)
try:
event_type = EventType.objects.get(id=event_type_id)
except EventType.DoesNotExist:
return Response({'error': 'Invalid event type'}, status=400)
# Build the event
event = Event(
title=title,
name=name,
description=data.get('description', ''),
event_type=event_type,
event_status=data.get('eventStatus', 'pending'),
venue_name=data.get('venueName', ''),
place=data.get('place', ''),
district=data.get('district', ''),
state=data.get('state', ''),
pincode=data.get('pincode', ''),
latitude=data.get('latitude', 0),
longitude=data.get('longitude', 0),
is_bookable=data.get('isBookable', False),
is_featured=data.get('isFeatured', False),
is_top_event=data.get('isTopEvent', False),
all_year_event=data.get('allYearEvent', False),
source=data.get('source', 'official'),
important_information=data.get('importantInformation', ''),
cancelled_reason=data.get('cancelledReason', 'NA'),
outside_event_url=data.get('outsideEventUrl', 'NA'),
is_eventify_event=data.get('isEventifyEvent', True),
)
# Optional dates/times
if data.get('startDate'):
event.start_date = data['startDate']
if data.get('endDate'):
event.end_date = data['endDate']
if data.get('startTime'):
event.start_time = data['startTime']
if data.get('endTime'):
event.end_time = data['endTime']
# Optional partner
partner_id = data.get('partnerId')
if partner_id:
try:
from partners.models import Partner
event.partner = Partner.objects.get(id=partner_id)
event.is_partner_event = True
except Exception:
pass
event.save()
_audit_log(request, 'event.created', 'event', event.id, {
'title': event.title,
'partner_id': str(event.partner_id) if event.partner_id else None,
'source': event.source,
})
return Response(_serialize_event_detail(event), status=201)
class EventTypesView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from events.models import EventType
types = EventType.objects.all().order_by('id')
return Response([
{'id': t.id, 'name': t.event_type}
for t in types
])
class EventPrimaryImageView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
from events.models import Event, EventImages
try:
event = Event.objects.get(pk=pk)
except Event.DoesNotExist:
return Response({"error": "Event not found"}, status=404)
image_id = request.data.get("image_id")
if not image_id:
return Response({"error": "image_id is required"}, status=400)
try:
img = EventImages.objects.get(pk=image_id, event=event)
except EventImages.DoesNotExist:
return Response({"error": "Image not found for this event"}, status=404)
# Clear all primary flags for this event, then set the selected one
prev_primary = EventImages.objects.filter(event=event, is_primary=True).values_list('pk', flat=True).first()
EventImages.objects.filter(event=event).update(is_primary=False)
img.is_primary = True
img.save()
_audit_log(request, 'event.primary_image_changed', 'event', pk, {
'new_primary_image_id': str(image_id),
'previous_primary_image_id': str(prev_primary) if prev_primary else None,
})
return Response({"success": True, "primaryImageId": image_id})
# ---------------------------------------------------------------------------
# RBAC Views
# ---------------------------------------------------------------------------
from admin_api.models import Department, Squad, StaffProfile, CustomRole, AuditLog
from django.utils.text import slugify
from django.db.models import Q
import json
SCOPE_DEFINITIONS = {
'users.read': {'label': 'View Users', 'category': 'Users'},
'users.write': {'label': 'Edit Users', 'category': 'Users'},
'users.delete': {'label': 'Delete Users', 'category': 'Users'},
'users.ban': {'label': 'Ban/Suspend Users', 'category': 'Users'},
'events.read': {'label': 'View Events', 'category': 'Events'},
'events.write': {'label': 'Create/Edit Events', 'category': 'Events'},
'events.approve': {'label': 'Approve Events', 'category': 'Events'},
'events.delete': {'label': 'Delete Events', 'category': 'Events'},
'finance.read': {'label': 'View Finance Dashboard', 'category': 'Finance'},
'finance.refunds.read': {'label': 'View Refund Requests', 'category': 'Finance'},
'finance.refunds.execute': {'label': 'Process Refunds', 'category': 'Finance'},
'finance.payouts.read': {'label': 'View Payouts', 'category': 'Finance'},
'finance.payouts.execute': {'label': 'Execute Payouts', 'category': 'Finance'},
'partners.read': {'label': 'View Partners', 'category': 'Partners'},
'partners.write': {'label': 'Edit Partners', 'category': 'Partners'},
'partners.kyc': {'label': 'Verify Partner KYC', 'category': 'Partners'},
'partners.impersonate': {'label': 'Login as Partner', 'category': 'Partners'},
'partners.events.review': {'label': 'Review Partner Events', 'category': 'Partners'},
'partners.suspend': {'label': 'Suspend/Unsuspend Partners', 'category': 'Partners'},
'tickets.read': {'label': 'View Tickets', 'category': 'Support'},
'tickets.write': {'label': 'Respond to Tickets', 'category': 'Support'},
'tickets.assign': {'label': 'Assign Tickets', 'category': 'Support'},
'tickets.escalate': {'label': 'Escalate Tickets', 'category': 'Support'},
'settings.read': {'label': 'View Settings', 'category': 'Settings'},
'settings.write': {'label': 'Modify Settings', 'category': 'Settings'},
'settings.staff': {'label': 'Manage Staff', 'category': 'Settings'},
'ads.read': {'label': 'View Ad Campaigns', 'category': 'Ad Control'},
'ads.write': {'label': 'Create/Edit Campaigns', 'category': 'Ad Control'},
'ads.approve': {'label': 'Approve Campaigns', 'category': 'Ad Control'},
'ads.report': {'label': 'View Ad Reports', 'category': 'Ad Control'},
# Reviews Module
'reviews.read': {'label': 'View Reviews', 'category': 'Reviews'},
'reviews.moderate': {'label': 'Moderate Reviews', 'category': 'Reviews'},
'reviews.delete': {'label': 'Delete Reviews', 'category': 'Reviews'},
# Contributions Module
'contributions.read': {'label': 'View Contributions', 'category': 'Contributions'},
'contributions.approve': {'label': 'Approve Contributions', 'category': 'Contributions'},
'contributions.reject': {'label': 'Reject Contributions', 'category': 'Contributions'},
'contributions.award': {'label': 'Award EP Points', 'category': 'Contributions'},
# Leads Module
'leads.read': {'label': 'View Leads', 'category': 'Leads'},
'leads.write': {'label': 'Edit Lead Details', 'category': 'Leads'},
'leads.assign': {'label': 'Assign Leads', 'category': 'Leads'},
'leads.convert': {'label': 'Convert Leads', 'category': 'Leads'},
# Audit Log Module
'audit.read': {'label': 'View Audit Log', 'category': 'Audit Log'},
'audit.export': {'label': 'Export Audit Log CSV', 'category': 'Audit Log'},
}
def _audit_log(request, action, target_type, target_id, details=None, user=None):
AuditLog.objects.create(
user=user if user is not None else (request.user if request.user.is_authenticated else None),
action=action,
target_type=target_type,
target_id=str(target_id),
details=details or {},
ip_address=request.META.get('REMOTE_ADDR'),
)
def _serialize_department(dept):
return {
'id': dept.id,
'name': dept.name,
'slug': dept.slug,
'description': dept.description,
'base_scopes': dept.base_scopes,
'color': dept.color,
'squad_count': dept.squads.count(),
'member_count': dept.staff_members.count(),
'created_at': dept.created_at.isoformat() if dept.created_at else None,
'updated_at': dept.updated_at.isoformat() if dept.updated_at else None,
}
def _serialize_squad(squad):
return {
'id': squad.id,
'name': squad.name,
'department_id': squad.department_id,
'department_name': squad.department.name,
'manager': {
'id': squad.manager.id,
'username': squad.manager.username,
'first_name': squad.manager.first_name,
'last_name': squad.manager.last_name,
} if squad.manager else None,
'extra_scopes': squad.extra_scopes,
'member_count': squad.members.count(),
'created_at': squad.created_at.isoformat() if squad.created_at else None,
}
def _serialize_staff(sp):
u = sp.user
return {
'id': sp.id,
'user_id': u.id,
'username': u.username,
'email': u.email,
'name': (u.first_name + ' ' + u.last_name).strip() or u.username,
'first_name': u.first_name,
'last_name': u.last_name,
'phone_number': u.phone_number,
'profile_picture': u.profile_picture.url if u.profile_picture else None,
'staff_role': sp.staff_role,
'status': sp.status,
'department': {
'id': sp.department.id,
'name': sp.department.name,
'slug': sp.department.slug,
'color': sp.department.color,
} if sp.department else None,
'squad': {
'id': sp.squad.id,
'name': sp.squad.name,
} if sp.squad else None,
'effective_scopes': sp.get_effective_scopes(),
'allowed_modules': sp.get_allowed_modules(),
'joined_at': sp.joined_at.isoformat() if sp.joined_at else None,
}
# 1. DepartmentListCreateView
class DepartmentListCreateView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
depts = Department.objects.all()
return Response([_serialize_department(d) for d in depts])
def post(self, request):
data = request.data
name = data.get('name', '').strip()
if not name:
return Response({'error': 'name is required'}, status=400)
slug = data.get('slug') or slugify(name)
if Department.objects.filter(slug=slug).exists():
return Response({'error': 'Department with this slug already exists'}, status=400)
dept = Department.objects.create(
name=name,
slug=slug,
description=data.get('description', ''),
base_scopes=data.get('base_scopes', []),
color=data.get('color', '#3B82F6'),
)
_audit_log(request, 'department.created', 'Department', dept.id, {'name': name})
return Response(_serialize_department(dept), status=201)
# 2. DepartmentDetailView
class DepartmentDetailView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
try:
dept = Department.objects.get(pk=pk)
except Department.DoesNotExist:
return Response({'error': 'Department not found'}, status=404)
data = request.data
if 'name' in data:
dept.name = data['name']
if 'description' in data:
dept.description = data['description']
if 'base_scopes' in data:
dept.base_scopes = data['base_scopes']
if 'color' in data:
dept.color = data['color']
dept.save()
_audit_log(request, 'department.updated', 'Department', dept.id, data)
return Response(_serialize_department(dept))
def delete(self, request, pk):
try:
dept = Department.objects.get(pk=pk)
except Department.DoesNotExist:
return Response({'error': 'Department not found'}, status=404)
if dept.squads.exists():
return Response({'error': 'Cannot delete department that has squads. Remove squads first.'}, status=400)
dept_name = dept.name
dept.delete()
_audit_log(request, 'department.deleted', 'Department', pk, {'name': dept_name})
return Response({'success': True}, status=200)
# 3. SquadListCreateView
class SquadListCreateView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
qs = Squad.objects.select_related('department', 'manager').all()
dept_id = request.query_params.get('department')
if dept_id:
qs = qs.filter(department_id=dept_id)
return Response([_serialize_squad(s) for s in qs])
def post(self, request):
data = request.data
name = data.get('name', '').strip()
department_id = data.get('department_id')
if not name or not department_id:
return Response({'error': 'name and department_id are required'}, status=400)
try:
dept = Department.objects.get(pk=department_id)
except Department.DoesNotExist:
return Response({'error': 'Department not found'}, status=404)
manager_id = data.get('manager_id')
manager = None
if manager_id:
try:
manager = User.objects.get(pk=manager_id)
except User.DoesNotExist:
pass
squad = Squad.objects.create(
name=name,
department=dept,
manager=manager,
extra_scopes=data.get('extra_scopes', []),
)
_audit_log(request, 'squad.created', 'Squad', squad.id, {'name': name, 'department': dept.name})
return Response(_serialize_squad(squad), status=201)
# 4. SquadDetailView
class SquadDetailView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
try:
squad = Squad.objects.select_related('department', 'manager').get(pk=pk)
except Squad.DoesNotExist:
return Response({'error': 'Squad not found'}, status=404)
data = request.data
if 'name' in data:
squad.name = data['name']
if 'extra_scopes' in data:
squad.extra_scopes = data['extra_scopes']
if 'manager_id' in data:
if data['manager_id']:
try:
squad.manager = User.objects.get(pk=data['manager_id'])
except User.DoesNotExist:
return Response({'error': 'Manager user not found'}, status=404)
else:
squad.manager = None
squad.save()
_audit_log(request, 'squad.updated', 'Squad', squad.id, data)
return Response(_serialize_squad(squad))
def delete(self, request, pk):
try:
squad = Squad.objects.get(pk=pk)
except Squad.DoesNotExist:
return Response({'error': 'Squad not found'}, status=404)
if squad.members.exists():
return Response({'error': 'Cannot delete squad that has members. Move members first.'}, status=400)
squad_name = squad.name
squad.delete()
_audit_log(request, 'squad.deleted', 'Squad', pk, {'name': squad_name})
return Response({'success': True}, status=200)
# 5. StaffListView
class StaffListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
qs = StaffProfile.objects.select_related('user', 'department', 'squad').all()
# Filters
dept = request.query_params.get('department')
if dept:
qs = qs.filter(department_id=dept)
squad_id = request.query_params.get('squad')
if squad_id:
qs = qs.filter(squad_id=squad_id)
role = request.query_params.get('role')
if role:
qs = qs.filter(staff_role=role)
status_filter = request.query_params.get('status')
if status_filter:
qs = qs.filter(status=status_filter)
search = request.query_params.get('search')
if search:
qs = qs.filter(
Q(user__username__icontains=search) |
Q(user__email__icontains=search) |
Q(user__first_name__icontains=search) |
Q(user__last_name__icontains=search)
)
return Response([_serialize_staff(sp) for sp in qs])
# 6. StaffInviteView
class StaffInviteView(APIView):
permission_classes = [IsAuthenticated]
def post(self, request):
data = request.data
email = data.get('email', '').strip()
if not email:
return Response({'error': 'email is required'}, status=400)
if User.objects.filter(email=email).exists():
return Response({'error': 'User with this email already exists'}, status=400)
username = data.get('username') or email.split('@')[0]
first_name = data.get('first_name', '')
last_name = data.get('last_name', '')
user = User.objects.create_user(
username=username,
email=email,
first_name=first_name,
last_name=last_name,
password=data.get('password', 'TempPass123!'),
is_customer=False,
role='staff',
)
department = None
if data.get('department_id'):
try:
department = Department.objects.get(pk=data['department_id'])
except Department.DoesNotExist:
pass
squad = None
if data.get('squad_id'):
try:
squad = Squad.objects.get(pk=data['squad_id'])
except Squad.DoesNotExist:
pass
sp = StaffProfile.objects.create(
user=user,
staff_role=data.get('staff_role', 'MEMBER'),
department=department,
squad=squad,
status='invited',
)
_audit_log(request, 'staff.invited', 'StaffProfile', sp.id, {
'email': email, 'staff_role': sp.staff_role,
})
return Response(_serialize_staff(sp), status=201)
# 7. StaffUpdateView
class StaffUpdateView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
try:
sp = StaffProfile.objects.select_related('user', 'department', 'squad').get(pk=pk)
except StaffProfile.DoesNotExist:
return Response({'error': 'Staff profile not found'}, status=404)
data = request.data
if 'staff_role' in data:
sp.staff_role = data['staff_role']
if 'department_id' in data:
if data['department_id']:
try:
sp.department = Department.objects.get(pk=data['department_id'])
except Department.DoesNotExist:
return Response({'error': 'Department not found'}, status=404)
else:
sp.department = None
if 'squad_id' in data:
if data['squad_id']:
try:
sp.squad = Squad.objects.get(pk=data['squad_id'])
except Squad.DoesNotExist:
return Response({'error': 'Squad not found'}, status=404)
else:
sp.squad = None
sp.save()
_audit_log(request, 'staff.updated', 'StaffProfile', sp.id, data)
return Response(_serialize_staff(sp))
# 8. StaffDeactivateView
class StaffDeactivateView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
try:
sp = StaffProfile.objects.select_related('user', 'department', 'squad').get(pk=pk)
except StaffProfile.DoesNotExist:
return Response({'error': 'Staff profile not found'}, status=404)
sp.status = 'deactivated'
sp.save()
_audit_log(request, 'staff.deactivated', 'StaffProfile', sp.id, {
'username': sp.user.username,
})
return Response(_serialize_staff(sp))
# 9. StaffMoveView
class StaffMoveView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
try:
sp = StaffProfile.objects.select_related('user', 'department', 'squad').get(pk=pk)
except StaffProfile.DoesNotExist:
return Response({'error': 'Staff profile not found'}, status=404)
data = request.data
old_dept = sp.department.name if sp.department else None
old_squad = sp.squad.name if sp.squad else None
if 'department_id' in data:
if data['department_id']:
try:
sp.department = Department.objects.get(pk=data['department_id'])
except Department.DoesNotExist:
return Response({'error': 'Department not found'}, status=404)
else:
sp.department = None
if 'squad_id' in data:
if data['squad_id']:
try:
sp.squad = Squad.objects.get(pk=data['squad_id'])
except Squad.DoesNotExist:
return Response({'error': 'Squad not found'}, status=404)
else:
sp.squad = None
sp.save()
_audit_log(request, 'staff.moved', 'StaffProfile', sp.id, {
'from_department': old_dept,
'to_department': sp.department.name if sp.department else None,
'from_squad': old_squad,
'to_squad': sp.squad.name if sp.squad else None,
})
return Response(_serialize_staff(sp))
# 10. RoleListCreateView
class RoleListCreateView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
roles = CustomRole.objects.all()
return Response([{
'id': r.id,
'name': r.name,
'slug': r.slug,
'description': r.description,
'scopes': r.scopes,
'is_system': r.is_system,
'created_at': r.created_at.isoformat() if r.created_at else None,
} for r in roles])
def post(self, request):
data = request.data
name = data.get('name', '').strip()
if not name:
return Response({'error': 'name is required'}, status=400)
slug = data.get('slug') or slugify(name)
if CustomRole.objects.filter(slug=slug).exists():
return Response({'error': 'Role with this slug already exists'}, status=400)
role = CustomRole.objects.create(
name=name,
slug=slug,
description=data.get('description', ''),
scopes=data.get('scopes', []),
is_system=data.get('is_system', False),
)
_audit_log(request, 'role.created', 'CustomRole', role.id, {'name': name})
return Response({
'id': role.id,
'name': role.name,
'slug': role.slug,
'description': role.description,
'scopes': role.scopes,
'is_system': role.is_system,
'created_at': role.created_at.isoformat() if role.created_at else None,
}, status=201)
# 11. RoleDetailView
class RoleDetailView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
try:
role = CustomRole.objects.get(pk=pk)
except CustomRole.DoesNotExist:
return Response({'error': 'Role not found'}, status=404)
data = request.data
if 'name' in data:
role.name = data['name']
if 'description' in data:
role.description = data['description']
if 'scopes' in data:
role.scopes = data['scopes']
role.save()
_audit_log(request, 'role.updated', 'CustomRole', role.id, data)
return Response({
'id': role.id,
'name': role.name,
'slug': role.slug,
'description': role.description,
'scopes': role.scopes,
'is_system': role.is_system,
'created_at': role.created_at.isoformat() if role.created_at else None,
})
def delete(self, request, pk):
try:
role = CustomRole.objects.get(pk=pk)
except CustomRole.DoesNotExist:
return Response({'error': 'Role not found'}, status=404)
if role.is_system:
return Response({'error': 'Cannot delete system roles'}, status=400)
role_name = role.name
role.delete()
_audit_log(request, 'role.deleted', 'CustomRole', pk, {'name': role_name})
return Response({'success': True}, status=200)
# 12. ScopeListView
class ScopeListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
# Group by category
grouped = {}
for key, val in SCOPE_DEFINITIONS.items():
cat = val['category']
if cat not in grouped:
grouped[cat] = []
grouped[cat].append({
'key': key,
'label': val['label'],
})
return Response({
'scopes': SCOPE_DEFINITIONS,
'grouped': grouped,
})
# 13. OrgTreeView
class OrgTreeView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
departments = Department.objects.prefetch_related(
'squads__members__user',
'squads__manager',
'staff_members__user',
).all()
tree = []
for dept in departments:
squads_data = []
for squad in dept.squads.all():
members_data = []
for member in squad.members.filter(status='active'):
members_data.append({
'id': member.id,
'user_id': member.user.id,
'username': member.user.username,
'first_name': member.user.first_name,
'last_name': member.user.last_name,
'email': member.user.email,
'staff_role': member.staff_role,
'profile_picture': member.user.profile_picture.url if member.user.profile_picture else None,
})
squads_data.append({
'id': squad.id,
'name': squad.name,
'manager': {
'id': squad.manager.id,
'username': squad.manager.username,
'first_name': squad.manager.first_name,
'last_name': squad.manager.last_name,
} if squad.manager else None,
'extra_scopes': squad.extra_scopes,
'members': members_data,
'member_count': len(members_data),
})
# Unassigned staff (in department but no squad)
unassigned = dept.staff_members.filter(squad__isnull=True, status='active')
unassigned_data = []
for member in unassigned:
unassigned_data.append({
'id': member.id,
'user_id': member.user.id,
'username': member.user.username,
'first_name': member.user.first_name,
'last_name': member.user.last_name,
'email': member.user.email,
'staff_role': member.staff_role,
})
tree.append({
'id': dept.id,
'name': dept.name,
'slug': dept.slug,
'color': dept.color,
'description': dept.description,
'base_scopes': dept.base_scopes,
'squads': squads_data,
'unassigned_members': unassigned_data,
'total_members': dept.staff_members.filter(status='active').count(),
})
# Also include staff with no department
orphans = StaffProfile.objects.filter(
department__isnull=True, status='active'
).select_related('user')
orphan_data = []
for sp in orphans:
orphan_data.append({
'id': sp.id,
'user_id': sp.user.id,
'username': sp.user.username,
'first_name': sp.user.first_name,
'last_name': sp.user.last_name,
'email': sp.user.email,
'staff_role': sp.staff_role,
})
return Response({
'departments': tree,
'unassigned_staff': orphan_data,
})
# 14. AuditLogListView
def _serialize_audit_log(log):
return {
'id': log.id,
'user': {
'id': log.user.id,
'username': log.user.username,
'email': log.user.email,
} if log.user else None,
'action': log.action,
'target_type': log.target_type,
'target_id': log.target_id,
'details': log.details,
'ip_address': log.ip_address,
'created_at': log.created_at.isoformat(),
}
class AuditLogListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from django.db.models import Q
qs = AuditLog.objects.select_related('user').all()
# Filters
user_id = request.query_params.get('user')
if user_id:
qs = qs.filter(user_id=user_id)
action = request.query_params.get('action')
if action:
qs = qs.filter(action__icontains=action)
target_type = request.query_params.get('target_type')
if target_type:
qs = qs.filter(target_type=target_type)
date_from = request.query_params.get('date_from')
if date_from:
qs = qs.filter(created_at__date__gte=date_from)
date_to = request.query_params.get('date_to')
if date_to:
qs = qs.filter(created_at__date__lte=date_to)
# Free-text search across action, target type/id, and actor identity.
# ORed so one box can locate entries regardless of which axis the
# admin remembers (slug vs. target vs. user).
search = request.query_params.get('search')
if search:
qs = qs.filter(
Q(action__icontains=search)
| Q(target_type__icontains=search)
| Q(target_id__icontains=search)
| Q(user__username__icontains=search)
| Q(user__email__icontains=search)
)
# Pagination
try:
page = max(1, int(request.query_params.get('page', 1)))
page_size = max(1, min(200, int(request.query_params.get('page_size', 50))))
except (ValueError, TypeError):
page, page_size = 1, 50
total = qs.count()
start = (page - 1) * page_size
end = start + page_size
logs = qs[start:end]
return Response({
'results': [_serialize_audit_log(log) for log in logs],
'total': total,
'page': page,
'page_size': page_size,
'total_pages': (total + page_size - 1) // page_size if total else 0,
})
# 15. AuditLogMetricsView — aggregated counts for the admin header bar.
def _compute_action_groups(qs):
"""Bucket action slugs into UI-visible groups without a DB roundtrip per slug.
Mirrors `actionGroup()` on the frontend so the tooltip numbers match the
badge colours.
"""
groups = {
'create': 0,
'update': 0,
'delete': 0,
'moderate': 0,
'auth': 0,
'other': 0,
}
for action in qs.values_list('action', flat=True):
slug = (action or '').lower()
if '.deactivated' in slug or '.deleted' in slug or '.removed' in slug:
groups['delete'] += 1
elif '.created' in slug or '.invited' in slug or '.added' in slug:
groups['create'] += 1
elif (
'.updated' in slug or '.edited' in slug
or '.moved' in slug or '.changed' in slug
):
groups['update'] += 1
elif (
slug.startswith('user.')
or slug.startswith('event.')
or slug.startswith('review.')
or slug.startswith('partner.kyc.')
):
groups['moderate'] += 1
elif slug.startswith('auth.') or 'login' in slug or 'logout' in slug:
groups['auth'] += 1
else:
groups['other'] += 1
return groups
class AuditLogMetricsView(APIView):
"""GET /api/v1/rbac/audit-log/metrics/ — aggregated counts.
Cached for 60 s because the header bar is fetched on every page load and
the numbers are approximate by nature. If you need fresh numbers, bypass
cache with `?nocache=1` (useful from the Django shell during incident
response).
"""
permission_classes = [IsAuthenticated]
CACHE_KEY = 'admin_api:audit_log:metrics:v1'
CACHE_TTL_SECONDS = 60
def get(self, request):
from django.core.cache import cache
from django.utils import timezone
from datetime import timedelta
if request.query_params.get('nocache') == '1':
payload = self._compute()
cache.set(self.CACHE_KEY, payload, self.CACHE_TTL_SECONDS)
return Response(payload)
payload = cache.get(self.CACHE_KEY)
if payload is None:
payload = self._compute()
cache.set(self.CACHE_KEY, payload, self.CACHE_TTL_SECONDS)
return Response(payload)
def _compute(self):
from django.utils import timezone
from datetime import timedelta
now = timezone.now()
today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
week_start = today_start - timedelta(days=6)
qs = AuditLog.objects.all()
return {
'total': qs.count(),
'today': qs.filter(created_at__gte=today_start).count(),
'week': qs.filter(created_at__gte=week_start).count(),
'distinct_users': qs.exclude(user__isnull=True)
.values('user_id').distinct().count(),
'by_action_group': _compute_action_groups(qs),
}
# ---------------------------------------------------------------------------
# Partner Onboarding API
# ---------------------------------------------------------------------------
class PartnerOnboardView(APIView):
"""
POST /api/v1/partners/onboard/
Creates a Partner + its first partner_manager User atomically.
Requires admin or manager JWT auth.
"""
permission_classes = [IsAuthenticated]
def post(self, request):
from partner.models import Partner
data = request.data
# --- Validate required fields ---
required = {
"partner_name": data.get("partner_name"),
"contact_email": data.get("contact_email"),
"manager_username": data.get("manager_username"),
"manager_password": data.get("manager_password"),
}
missing = [k for k, v in required.items() if not v]
if missing:
return Response(
{"success": False, "error": "Missing required fields: " + ", ".join(missing)},
status=status.HTTP_400_BAD_REQUEST,
)
manager_username = data["manager_username"].strip()
contact_email = data["contact_email"].strip().lower()
manager_password = data["manager_password"]
# --- Check uniqueness ---
if User.objects.filter(username=manager_username).exists():
return Response(
{"success": False, "error": "Username '" + manager_username + "' already exists"},
status=status.HTTP_409_CONFLICT,
)
if User.objects.filter(email=contact_email).exists():
return Response(
{"success": False, "error": "Email '" + contact_email + "' already exists"},
status=status.HTTP_409_CONFLICT,
)
# --- Split contact_name into first/last ---
contact_name = (data.get("contact_name") or "").strip()
name_parts = contact_name.split(None, 1)
first_name = name_parts[0] if name_parts else ""
last_name = name_parts[1] if len(name_parts) > 1 else ""
# --- Validate partner_type ---
partner_type = (data.get("partner_type") or "other").strip().lower()
valid_types = ("venue", "promoter", "sponsor", "vendor", "affiliate", "other")
if partner_type not in valid_types:
partner_type = "other"
try:
with transaction.atomic():
# 1) Create Partner
partner = Partner.objects.create(
name=data["partner_name"].strip(),
partner_type=partner_type,
primary_contact_person_name=contact_name,
primary_contact_person_email=contact_email,
primary_contact_person_phone=(data.get("contact_phone") or ""),
status="pending",
is_kyc_compliant=False,
kyc_compliance_status="pending",
address=(data.get("address") or ""),
city=(data.get("city") or ""),
state=(data.get("state") or ""),
pincode=(data.get("pincode") or ""),
website_url=(data.get("website_url") or None),
)
# 2) Create partner_manager User
manager_user = User(
username=manager_username,
email=contact_email,
first_name=first_name,
last_name=last_name,
role="partner_manager",
is_customer=False,
partner=partner,
phone_number=(data.get("contact_phone") or ""),
)
manager_user.set_password(manager_password)
manager_user.save()
_audit_log(request, 'partner.onboarded', 'partner', partner.id, {
'partner_name': partner.name,
'manager_user_id': manager_user.id,
'manager_email': manager_user.email,
})
except Exception as e:
return Response(
{"success": False, "error": "Failed to create partner: " + str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return Response(
{
"success": True,
"partner": {
"id": partner.id,
"name": partner.name,
"partner_type": partner.partner_type,
"status": partner.status,
"contact_name": partner.primary_contact_person_name,
"contact_email": partner.primary_contact_person_email,
"contact_phone": partner.primary_contact_person_phone,
"address": partner.address,
"city": partner.city,
"state": partner.state,
"pincode": partner.pincode,
"website_url": partner.website_url or "",
"is_kyc_compliant": partner.is_kyc_compliant,
"kyc_compliance_status": partner.kyc_compliance_status,
},
"manager": {
"id": manager_user.id,
"username": manager_user.username,
"email": manager_user.email,
"first_name": manager_user.first_name,
"last_name": manager_user.last_name,
"role": manager_user.role,
},
"login_url": "https://partner.eventifyplus.com",
"message": "Partner onboarded successfully. Manager can login at partner.eventifyplus.com",
},
status=status.HTTP_201_CREATED,
)
class PartnerStaffCreateView(APIView):
"""
POST /api/v1/partners/<partner_id>/staff/
Allows a partner_manager (or admin) to add staff to their partner org.
"""
permission_classes = [IsAuthenticated]
def post(self, request, partner_id):
from partner.models import Partner
from django.shortcuts import get_object_or_404
# --- Authorization: must be admin or partner_manager of this partner ---
user = request.user
is_admin = user.role in ("admin", "manager") or user.is_superuser
is_partner_manager = (
user.role == "partner_manager"
and user.partner_id is not None
and user.partner_id == partner_id
)
if not is_admin and not is_partner_manager:
return Response(
{"success": False, "error": "You do not have permission to add staff to this partner"},
status=status.HTTP_403_FORBIDDEN,
)
# --- Validate partner exists ---
partner = get_object_or_404(Partner, pk=partner_id)
data = request.data
# --- Validate required fields ---
required = {
"username": data.get("username"),
"email": data.get("email"),
"password": data.get("password"),
}
missing = [k for k, v in required.items() if not v]
if missing:
return Response(
{"success": False, "error": "Missing required fields: " + ", ".join(missing)},
status=status.HTTP_400_BAD_REQUEST,
)
username = data["username"].strip()
email = data["email"].strip().lower()
password = data["password"]
# --- Check uniqueness ---
if User.objects.filter(username=username).exists():
return Response(
{"success": False, "error": "Username '" + username + "' already exists"},
status=status.HTTP_409_CONFLICT,
)
if User.objects.filter(email=email).exists():
return Response(
{"success": False, "error": "Email '" + email + "' already exists"},
status=status.HTTP_409_CONFLICT,
)
# --- Determine role (default partner_staff, allow partner_manager) ---
requested_role = (data.get("role") or "partner_staff").strip().lower()
allowed_roles = ("partner_staff", "partner_manager")
if requested_role not in allowed_roles:
return Response(
{"success": False, "error": "Role must be one of: " + ", ".join(allowed_roles)},
status=status.HTTP_400_BAD_REQUEST,
)
# Non-admin users cannot create partner_manager
if requested_role == "partner_manager" and not is_admin:
return Response(
{"success": False, "error": "Only admins can create additional partner managers"},
status=status.HTTP_403_FORBIDDEN,
)
try:
staff_user = User(
username=username,
email=email,
first_name=(data.get("first_name") or "").strip(),
last_name=(data.get("last_name") or "").strip(),
role=requested_role,
is_customer=False,
partner=partner,
phone_number=(data.get("phone_number") or ""),
)
staff_user.set_password(password)
staff_user.save()
_audit_log(request, 'partner.staff.created', 'partner', partner.id, {
'new_user_id': staff_user.id,
'username': staff_user.username,
'email': staff_user.email,
'role': staff_user.role,
})
except Exception as e:
return Response(
{"success": False, "error": "Failed to create staff user: " + str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR,
)
return Response(
{
"success": True,
"user": {
"id": staff_user.id,
"username": staff_user.username,
"email": staff_user.email,
"first_name": staff_user.first_name,
"last_name": staff_user.last_name,
"role": staff_user.role,
"phone_number": staff_user.phone_number or "",
"partner_id": partner.id,
"partner_name": partner.name,
},
"message": f"Staff user '{username}' created successfully for partner '{partner.name}'",
},
status=status.HTTP_201_CREATED,
)
class PartnerImpersonateView(APIView):
"""
POST /api/v1/partners/<pk>/impersonate/
Admin-only: generate a short-lived JWT for the partner's primary manager user.
Returns access/refresh tokens + user info so the partner portal can create a session.
"""
permission_classes = [IsAuthenticated]
def post(self, request, pk):
from partner.models import Partner as PartnerModel
from django.shortcuts import get_object_or_404
partner = get_object_or_404(PartnerModel, pk=pk)
partner_user = User.objects.filter(partner=partner, role='partner_manager').first()
if not partner_user:
return Response(
{'error': 'No partner_manager user found for this partner.'},
status=status.HTTP_404_NOT_FOUND,
)
refresh = RefreshToken.for_user(partner_user)
_audit_log(request, 'partner.impersonated', 'partner', str(pk), {
'partner_name': partner.name,
'impersonated_user': partner_user.username,
'admin': request.user.username,
})
return Response({
'access': str(refresh.access_token),
'refresh': str(refresh),
'user': {
'id': partner_user.id,
'email': partner_user.email,
'username': partner_user.username,
'role': partner_user.role,
'partnerId': str(pk),
},
})
# ─── Gamification Dashboard (stub) ───────────────────────────────────────────
class GamificationDashboardView(APIView):
permission_classes = [] # public for now; restrict when auth is wired up
def get(self, request):
user_id = request.GET.get('user_id', '')
return Response({
'status': 'success',
'profile': {
'user_id': user_id,
'current_tier': 'BRONZE',
'current_ep': 0,
'current_rp': 0,
'lifetime_ep': 0,
},
'submissions': [],
})
# ─── Gamification: Event Submission (stub) ────────────────────────────────────
class GamificationSubmitEventView(APIView):
permission_classes = []
def post(self, request):
data = request.data
return Response({
'status': 'success',
'submission': {
'id': 1,
'event_name': data.get('event_name', ''),
'status': 'PENDING',
'total_ep_awarded': 0,
'created_at': __import__('datetime').datetime.now().isoformat(),
},
'message': 'Event submitted for review. You will earn EP once approved!',
})
# ─── Reward Shop: List Items (stub) ──────────────────────────────────────────
class ShopItemsView(APIView):
permission_classes = []
def get(self, request):
return Response({
'status': 'success',
'items': [
{
'id': 1,
'name': 'BookMyShow Voucher',
'description': 'Get a Rs.100 BookMyShow gift card',
'rp_cost': 50,
'stock_quantity': 10,
},
{
'id': 2,
'name': 'Event Priority Listing',
'description': 'Feature your next event at the top for 7 days',
'rp_cost': 30,
'stock_quantity': 5,
},
{
'id': 3,
'name': 'Eventify Merch Pack',
'description': 'Exclusive stickers, badge & notebook',
'rp_cost': 100,
'stock_quantity': 3,
},
],
})
# ─── Reward Shop: Redeem (stub) ──────────────────────────────────────────────
class ShopRedeemView(APIView):
permission_classes = []
def post(self, request):
import uuid
item_id = request.data.get('item_id')
return Response({
'status': 'success',
'voucher': {
'item_id': item_id,
'voucher_code_issued': 'EVF-' + uuid.uuid4().hex[:8].upper(),
},
'message': 'Reward redeemed successfully!',
})
# ---------------------------------------------------------------------------
# Lead Manager
# ---------------------------------------------------------------------------
def _serialize_lead(lead):
assigned_name = ''
assigned_id = None
if lead.assigned_to:
assigned_name = lead.assigned_to.get_full_name() or lead.assigned_to.username
assigned_id = lead.assigned_to.pk
user_account = None
if lead.user_account:
u = lead.user_account
profile_pic = None
try:
if u.profile_picture:
profile_pic = u.profile_picture.url
except Exception:
pass
user_account = {
'id': u.pk,
'name': u.get_full_name() or u.username,
'email': u.email,
'phone': getattr(u, 'phone_number', None) or '',
'eventifyId': getattr(u, 'eventify_id', None),
'profilePicture': profile_pic,
}
return {
'id': lead.pk,
'name': lead.name,
'email': lead.email,
'phone': lead.phone,
'eventType': lead.event_type,
'message': lead.message,
'status': lead.status,
'source': lead.source,
'priority': lead.priority,
'assignedTo': assigned_id,
'assignedToName': assigned_name,
'notes': lead.notes,
'createdAt': lead.created_at.isoformat(),
'updatedAt': lead.updated_at.isoformat(),
'userAccount': user_account,
}
class LeadMetricsView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from admin_api.models import Lead
from django.utils import timezone
today = timezone.now().date()
return Response({
'total': Lead.objects.count(),
'newToday': Lead.objects.filter(created_at__date=today).count(),
'new': Lead.objects.filter(status='new').count(),
'contacted': Lead.objects.filter(status='contacted').count(),
'qualified': Lead.objects.filter(status='qualified').count(),
'converted': Lead.objects.filter(status='converted').count(),
'closed': Lead.objects.filter(status='closed').count(),
})
class LeadListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from admin_api.models import Lead
from django.db.models import Q
qs = Lead.objects.select_related('assigned_to', 'user_account').order_by('-created_at')
# Filters
status_f = request.query_params.get('status', '').strip()
if status_f and status_f in dict(Lead.STATUS_CHOICES):
qs = qs.filter(status=status_f)
priority_f = request.query_params.get('priority', '').strip()
if priority_f and priority_f in dict(Lead.PRIORITY_CHOICES):
qs = qs.filter(priority=priority_f)
source_f = request.query_params.get('source', '').strip()
if source_f and source_f in dict(Lead.SOURCE_CHOICES):
qs = qs.filter(source=source_f)
search = request.query_params.get('search', '').strip()
if search:
qs = qs.filter(
Q(name__icontains=search) |
Q(email__icontains=search) |
Q(phone__icontains=search)
)
date_from = request.query_params.get('date_from', '').strip()
if date_from:
qs = qs.filter(created_at__date__gte=date_from)
date_to = request.query_params.get('date_to', '').strip()
if date_to:
qs = qs.filter(created_at__date__lte=date_to)
# Pagination
try:
page = max(1, int(request.query_params.get('page', 1)))
page_size = min(100, int(request.query_params.get('page_size', 20)))
except (ValueError, TypeError):
page, page_size = 1, 20
total = qs.count()
leads = qs[(page - 1) * page_size: page * page_size]
return Response({'count': total, 'results': [_serialize_lead(l) for l in leads]})
class LeadDetailView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, pk):
from admin_api.models import Lead
from django.shortcuts import get_object_or_404
lead = get_object_or_404(Lead.objects.select_related('assigned_to', 'user_account'), pk=pk)
return Response(_serialize_lead(lead))
class LeadUpdateView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk):
from admin_api.models import Lead
from django.shortcuts import get_object_or_404
from eventify_logger.services import log
lead = get_object_or_404(Lead, pk=pk)
changed = []
new_status = request.data.get('status')
if new_status:
if new_status not in dict(Lead.STATUS_CHOICES):
return Response({'error': f'Invalid status: {new_status}'}, status=400)
lead.status = new_status
changed.append('status')
new_priority = request.data.get('priority')
if new_priority:
if new_priority not in dict(Lead.PRIORITY_CHOICES):
return Response({'error': f'Invalid priority: {new_priority}'}, status=400)
lead.priority = new_priority
changed.append('priority')
assigned_to_id = request.data.get('assignedTo')
if assigned_to_id is not None:
if assigned_to_id == '' or assigned_to_id is False:
lead.assigned_to = None
changed.append('assigned_to')
else:
try:
lead.assigned_to = User.objects.get(pk=int(assigned_to_id))
changed.append('assigned_to')
except (User.DoesNotExist, ValueError, TypeError):
return Response({'error': 'Invalid assignedTo user'}, status=400)
notes = request.data.get('notes')
if notes is not None:
lead.notes = notes
changed.append('notes')
lead.save()
log("info", f"Lead #{pk} updated: {', '.join(changed)}", request=request, user=request.user)
if changed:
_audit_log(request, 'lead.updated', 'lead', pk, {'changed_fields': changed})
return Response(_serialize_lead(lead))
# ---------------------------------------------------------------------------
# Notification schedules (admin-side recurring email jobs)
# ---------------------------------------------------------------------------
def _serialize_recipient(r):
return {
'id': r.pk,
'email': r.email,
'displayName': r.display_name,
'isActive': r.is_active,
'createdAt': r.created_at.isoformat(),
}
def _serialize_schedule(s, include_recipients=True):
payload = {
'id': s.pk,
'name': s.name,
'notificationType': s.notification_type,
'cronExpression': s.cron_expression,
'isActive': s.is_active,
'lastRunAt': s.last_run_at.isoformat() if s.last_run_at else None,
'lastStatus': s.last_status,
'lastError': s.last_error,
'createdAt': s.created_at.isoformat(),
'updatedAt': s.updated_at.isoformat(),
}
if include_recipients:
payload['recipients'] = [
_serialize_recipient(r) for r in s.recipients.all()
]
payload['recipientCount'] = s.recipients.filter(is_active=True).count()
return payload
def _validate_cron(expr):
from croniter import croniter
if not expr or not isinstance(expr, str):
return False, 'cronExpression is required'
if not croniter.is_valid(expr.strip()):
return False, f'Invalid cron expression: {expr!r}'
return True, None
class NotificationTypesView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from notifications.models import NotificationSchedule
return Response({
'types': [
{'value': v, 'label': l}
for v, l in NotificationSchedule.TYPE_CHOICES
],
})
class NotificationScheduleListView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
from notifications.models import NotificationSchedule
qs = NotificationSchedule.objects.prefetch_related('recipients').order_by('-created_at')
return Response({
'count': qs.count(),
'results': [_serialize_schedule(s) for s in qs],
})
def post(self, request):
from notifications.models import NotificationSchedule, NotificationRecipient
from django.db import transaction as db_tx
from eventify_logger.services import log
name = (request.data.get('name') or '').strip()
ntype = (request.data.get('notificationType') or '').strip()
cron = (request.data.get('cronExpression') or '').strip()
is_active = bool(request.data.get('isActive', True))
recipients_in = request.data.get('recipients') or []
if not name:
return Response({'error': 'name is required'}, status=400)
if ntype not in dict(NotificationSchedule.TYPE_CHOICES):
return Response({'error': f'Invalid notificationType: {ntype!r}'}, status=400)
ok, err = _validate_cron(cron)
if not ok:
return Response({'error': err}, status=400)
with db_tx.atomic():
schedule = NotificationSchedule.objects.create(
name=name,
notification_type=ntype,
cron_expression=cron,
is_active=is_active,
)
seen_emails = set()
for r in recipients_in:
email = (r.get('email') or '').strip().lower()
if not email or email in seen_emails:
continue
seen_emails.add(email)
NotificationRecipient.objects.create(
schedule=schedule,
email=email,
display_name=(r.get('displayName') or '').strip(),
is_active=bool(r.get('isActive', True)),
)
_audit_log(
request,
'notification.schedule.created',
'NotificationSchedule',
schedule.pk,
details={
'name': schedule.name,
'notification_type': schedule.notification_type,
'cron_expression': schedule.cron_expression,
'is_active': schedule.is_active,
'recipient_count': len(seen_emails),
},
)
log('info', f'Notification schedule #{schedule.pk} created: {schedule.name}',
request=request, user=request.user)
return Response(_serialize_schedule(schedule), status=201)
class NotificationScheduleDetailView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request, pk):
from notifications.models import NotificationSchedule
from django.shortcuts import get_object_or_404
s = get_object_or_404(
NotificationSchedule.objects.prefetch_related('recipients'), pk=pk,
)
return Response(_serialize_schedule(s))
def patch(self, request, pk):
from notifications.models import NotificationSchedule
from django.shortcuts import get_object_or_404
from eventify_logger.services import log
s = get_object_or_404(NotificationSchedule, pk=pk)
changed = []
if (name := request.data.get('name')) is not None:
name = str(name).strip()
if not name:
return Response({'error': 'name cannot be empty'}, status=400)
s.name = name
changed.append('name')
if (ntype := request.data.get('notificationType')) is not None:
if ntype not in dict(NotificationSchedule.TYPE_CHOICES):
return Response({'error': f'Invalid notificationType: {ntype!r}'}, status=400)
s.notification_type = ntype
changed.append('notification_type')
if (cron := request.data.get('cronExpression')) is not None:
ok, err = _validate_cron(cron)
if not ok:
return Response({'error': err}, status=400)
s.cron_expression = cron.strip()
changed.append('cron_expression')
if (is_active := request.data.get('isActive')) is not None:
s.is_active = bool(is_active)
changed.append('is_active')
s.save()
if changed:
_audit_log(
request,
'notification.schedule.updated',
'NotificationSchedule',
s.pk,
details={
'name': s.name,
'changed_fields': changed,
'cron_expression': s.cron_expression,
'notification_type': s.notification_type,
'is_active': s.is_active,
},
)
log('info', f'Notification schedule #{pk} updated: {", ".join(changed) or "no-op"}',
request=request, user=request.user)
return Response(_serialize_schedule(s))
def delete(self, request, pk):
from notifications.models import NotificationSchedule
from django.shortcuts import get_object_or_404
from eventify_logger.services import log
s = get_object_or_404(NotificationSchedule, pk=pk)
schedule_name = s.name
schedule_type = s.notification_type
schedule_cron = s.cron_expression
s.delete()
_audit_log(
request,
'notification.schedule.deleted',
'NotificationSchedule',
pk,
details={
'name': schedule_name,
'notification_type': schedule_type,
'cron_expression': schedule_cron,
},
)
log('info', f'Notification schedule #{pk} deleted', request=request, user=request.user)
return Response(status=204)
class NotificationRecipientView(APIView):
permission_classes = [IsAuthenticated]
def post(self, request, pk):
from notifications.models import NotificationSchedule, NotificationRecipient
from django.shortcuts import get_object_or_404
schedule = get_object_or_404(NotificationSchedule, pk=pk)
email = (request.data.get('email') or '').strip().lower()
if not email:
return Response({'error': 'email is required'}, status=400)
if NotificationRecipient.objects.filter(schedule=schedule, email=email).exists():
return Response({'error': f'{email} is already a recipient'}, status=409)
r = NotificationRecipient.objects.create(
schedule=schedule,
email=email,
display_name=(request.data.get('displayName') or '').strip(),
is_active=bool(request.data.get('isActive', True)),
)
_audit_log(
request,
'notification.recipient.added',
'NotificationRecipient',
r.pk,
details={
'schedule_id': schedule.pk,
'schedule_name': schedule.name,
'email': r.email,
'display_name': r.display_name,
'is_active': r.is_active,
},
)
return Response(_serialize_recipient(r), status=201)
class NotificationRecipientDetailView(APIView):
permission_classes = [IsAuthenticated]
def patch(self, request, pk, rid):
from notifications.models import NotificationRecipient
from django.shortcuts import get_object_or_404
r = get_object_or_404(NotificationRecipient, pk=rid, schedule_id=pk)
changed = []
if (email := request.data.get('email')) is not None:
email = str(email).strip().lower()
if not email:
return Response({'error': 'email cannot be empty'}, status=400)
clash = NotificationRecipient.objects.filter(
schedule_id=pk, email=email,
).exclude(pk=rid).exists()
if clash:
return Response({'error': f'{email} is already a recipient'}, status=409)
if r.email != email:
r.email = email
changed.append('email')
if (display_name := request.data.get('displayName')) is not None:
new_name = str(display_name).strip()
if r.display_name != new_name:
r.display_name = new_name
changed.append('display_name')
if (is_active := request.data.get('isActive')) is not None:
new_active = bool(is_active)
if r.is_active != new_active:
r.is_active = new_active
changed.append('is_active')
r.save()
if changed:
_audit_log(
request,
'notification.recipient.updated',
'NotificationRecipient',
r.pk,
details={
'schedule_id': pk,
'email': r.email,
'display_name': r.display_name,
'is_active': r.is_active,
'changed_fields': changed,
},
)
return Response(_serialize_recipient(r))
def delete(self, request, pk, rid):
from notifications.models import NotificationRecipient
from django.shortcuts import get_object_or_404
r = get_object_or_404(NotificationRecipient, pk=rid, schedule_id=pk)
recipient_email = r.email
recipient_name = r.display_name
r.delete()
_audit_log(
request,
'notification.recipient.removed',
'NotificationRecipient',
rid,
details={
'schedule_id': pk,
'email': recipient_email,
'display_name': recipient_name,
},
)
return Response(status=204)
class NotificationScheduleSendNowView(APIView):
"""Trigger an immediate dispatch of one schedule, bypassing cron check."""
permission_classes = [IsAuthenticated]
def post(self, request, pk):
from notifications.models import NotificationSchedule
from notifications.emails import render_and_send
from django.shortcuts import get_object_or_404
from django.utils import timezone as dj_tz
from eventify_logger.services import log
schedule = get_object_or_404(NotificationSchedule, pk=pk)
try:
recipient_count = render_and_send(schedule)
except Exception as exc: # noqa: BLE001
schedule.last_status = NotificationSchedule.STATUS_ERROR
schedule.last_error = str(exc)[:2000]
schedule.save(update_fields=['last_status', 'last_error', 'updated_at'])
log('error', f'Send-now failed for schedule #{pk}: {exc}',
request=request, user=request.user)
return Response({'error': str(exc)}, status=500)
schedule.last_run_at = dj_tz.now()
schedule.last_status = NotificationSchedule.STATUS_SUCCESS
schedule.last_error = ''
schedule.save(update_fields=[
'last_run_at', 'last_status', 'last_error', 'updated_at',
])
_audit_log(
request,
'notification.schedule.dispatched',
'NotificationSchedule',
schedule.pk,
details={
'schedule_name': schedule.name,
'notification_type': schedule.notification_type,
'recipient_count': recipient_count,
'triggered_at': schedule.last_run_at.isoformat(),
},
)
log('info', f'Send-now fired for schedule #{pk} → {recipient_count} recipient(s)',
request=request, user=request.user)
return Response({
'ok': True,
'recipientCount': recipient_count,
'schedule': _serialize_schedule(schedule),
})
class NotificationScheduleTestSendView(APIView):
"""Send a preview of this schedule's email to a single address.
Does NOT update last_run_at / last_status — purely for previewing content.
"""
permission_classes = [IsAuthenticated]
def post(self, request, pk):
from notifications.models import NotificationSchedule
from notifications.emails import BUILDERS
from django.conf import settings
from django.core.mail import EmailMessage
from django.shortcuts import get_object_or_404
from eventify_logger.services import log
schedule = get_object_or_404(NotificationSchedule, pk=pk)
email = (request.data.get('email') or '').strip().lower()
if not email:
return Response({'error': 'email is required'}, status=400)
builder = BUILDERS.get(schedule.notification_type)
if builder is None:
return Response(
{'error': f'No builder for type: {schedule.notification_type}'},
status=400,
)
try:
subject, html = builder(schedule)
subject = f'[TEST] {subject}'
msg = EmailMessage(
subject=subject,
body=html,
from_email=settings.DEFAULT_FROM_EMAIL,
to=[email],
)
msg.content_subtype = 'html'
msg.send(fail_silently=False)
except Exception as exc: # noqa: BLE001
log('error', f'Test-send failed for schedule #{pk}: {exc}',
request=request, user=request.user)
return Response({'error': str(exc)}, status=500)
log('info', f'Test email sent for schedule #{pk} → {email}',
request=request, user=request.user)
return Response({'ok': True, 'sentTo': email})
Reference in New Issue View Git Blame Copy Permalink