"""Plain, Celery-free reminder functions — directly callable/testable. Each function scans for objects that need a reminder, dedupes via ReminderLog, and fires a notification through NotificationHandler. """ from datetime import timedelta from django.utils import timezone from .models import ReminderLog def _already_sent_today(obj, reminder_key) -> bool: from django.contrib.contenttypes.models import ContentType ct = ContentType.objects.get_for_model(obj) _, created = ReminderLog.objects.get_or_create( content_type=ct, object_id=str(obj.pk), reminder_key=reminder_key, sent_date=timezone.localdate(), ) return not created def remind_overdue_tasks() -> int: """Daily digest: notify the assignee of every task that's overdue and not done/cancelled/archived. Re-fires once per day while still overdue. Runs per-organization so one org's overdue tasks never notify another's.""" from apps.notifications.handler import NotificationHandler from apps.companies.models import Organization from apps.tasks.models import Task sent = 0 for org in Organization.objects.all(): qs = ( Task.objects.filter(organization=org, due_date__lt=timezone.now()) .exclude(status__in=("done", "cancelled", "archived")) .select_related("assignee") ) for task in qs: if not task.assignee_id or _already_sent_today(task, "overdue"): continue try: NotificationHandler.send( user=task.assignee, title="Task overdue", message=f"\"{task.title}\" is overdue.", notification_type="task", priority="high", icon="alert-triangle", link=f"/tasks/{task.id}", broadcast=True, ) sent += 1 except Exception: pass return sent def remind_upcoming_events() -> int: """One-time reminder ~24h before a published event starts, to internal team members (EventTeam), not external registrants/invitees. Runs per-organization so one org's events never notify another's.""" from apps.notifications.handler import NotificationHandler from apps.companies.models import Organization from apps.events.models import Event, EventTeam now = timezone.now() sent = 0 for org in Organization.objects.all(): events = Event.objects.filter( organization=org, status="published", start_date__gte=now + timedelta(hours=23), start_date__lte=now + timedelta(hours=25), ) for event in events: for member in EventTeam.objects.filter(event=event).select_related("user"): if _already_sent_today(member, "event_upcoming"): continue try: NotificationHandler.send( user=member.user, title="Event starting soon", message=f"\"{event.name}\" starts tomorrow ({event.start_date:%Y-%m-%d %H:%M}).", notification_type="event", priority="medium", icon="calendar", link=f"/events/{event.id}", broadcast=True, ) sent += 1 except Exception: pass return sent def _notify_hr_role(title, message, link, organization, exclude_user_id=None): """Broadcast to users with hr+approve permission within organization — mirrors hr.views.SelfServiceApprovalMixin._notify_hr's role lookup.""" from apps.notifications.handler import NotificationHandler from apps.core.models import User as CoreUser qs = CoreUser.objects.filter( user_roles__organization=organization, user_roles__role__permissions__domain="hr", user_roles__role__permissions__action__in=("approve", "*"), ) if exclude_user_id: qs = qs.exclude(pk=exclude_user_id) users = list(qs.distinct()) if not users: return try: NotificationHandler.broadcast( users=users, title=title, message=message, notification_type="hr", icon="file-text", link=link, ) except Exception: pass def remind_expiring_contracts() -> int: """Two one-time thresholds (30d and 7d out) per contract. Notifies the employee and the HR role. Runs per-organization to avoid an unscoped full-table scan across every org's contracts.""" from apps.notifications.handler import NotificationHandler from apps.companies.models import Organization from apps.hr.models import Contract today = timezone.localdate() sent = 0 for org in Organization.objects.all(): for days, key in ((30, "contract_30d"), (7, "contract_7d")): target = today + timedelta(days=days) qs = Contract.objects.filter( status="active", end_date=target, employee__organization=org, ).select_related("employee__user") for contract in qs: if _already_sent_today(contract, key): continue employee_user = contract.employee.user message = f"{employee_user.get_full_name() or employee_user.email}'s contract expires on {contract.end_date} ({days} days)." if employee_user: try: NotificationHandler.send( user=employee_user, title="Your contract is expiring", message=f"Your contract expires on {contract.end_date} ({days} days).", notification_type="hr", priority="high" if days == 7 else "medium", icon="file-text", link=f"/hr/employees/{contract.employee_id}?tab=contracts", broadcast=True, ) except Exception: pass _notify_hr_role( "Contract expiring", message, f"/hr/employees/{contract.employee_id}?tab=contracts", org, exclude_user_id=employee_user.id if employee_user else None, ) sent += 1 return sent def remind_approaching_leave() -> int: """One-time reminder the day before an approved leave starts, to the employee and (if set) their manager. Runs per-organization to avoid an unscoped full-table scan across every org's leave requests.""" from apps.notifications.handler import NotificationHandler from apps.companies.models import Organization from apps.hr.models import LeaveRequest tomorrow = timezone.localdate() + timedelta(days=1) sent = 0 for org in Organization.objects.all(): qs = LeaveRequest.objects.filter( status="approved", start_date=tomorrow, employee__organization=org, ).select_related("employee__user", "employee__manager__user") for leave in qs: if _already_sent_today(leave, "leave_upcoming"): continue employee_user = leave.employee.user if employee_user: try: NotificationHandler.send( user=employee_user, title="Leave starting tomorrow", message=f"Your {leave.leave_type_label().lower()} starts tomorrow ({leave.start_date} – {leave.end_date}).", notification_type="hr", priority="medium", icon="calendar-off", link=f"/hr/employees/{leave.employee_id}?tab=activities", broadcast=True, ) except Exception: pass manager = leave.employee.manager if manager and manager.user_id: try: NotificationHandler.send( user=manager.user, title="Team member leave starting tomorrow", message=f"{employee_user.get_full_name() or employee_user.email}'s leave starts tomorrow ({leave.start_date} – {leave.end_date}).", notification_type="hr", priority="medium", icon="calendar-off", link=f"/hr/employees/{leave.employee_id}?tab=activities", broadcast=True, ) except Exception: pass sent += 1 return sent