from __future__ import annotations import logging import os import threading import time from dataclasses import dataclass from datetime import date, datetime, time as dt_time, timedelta, timezone from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session from app.core.db.common import CommonEngine, CommonSessionLocal from app.modules.alerts.models import UserAlert from app.modules.alerts.service import create_alert from app.modules.clients.models import Client from app.modules.employees.models import Employee, EmployeeAttendance from app.modules.services.models import ClientServiceSubscription, ClientServiceTaskInstance logger = logging.getLogger("audit_firm.notifications") _COMPLETED_STATUSES = { "completed", "complete", "done", "approved", "closed", "filed", "cancelled", "inactive", } _STARTED = False _START_LOCK = threading.Lock() @dataclass(slots=True) class AutomationResult: task_due_today: int = 0 task_overdue_staff: int = 0 task_overdue_manager: int = 0 task_overdue_partner: int = 0 engagement_due: int = 0 engagement_overdue: int = 0 attendance_missing_punchout: int = 0 @property def total_alerts(self) -> int: return ( self.task_due_today + self.task_overdue_staff + self.task_overdue_manager + self.task_overdue_partner + self.engagement_due + self.engagement_overdue + self.attendance_missing_punchout ) def _table_exists(table_name: str) -> bool: try: return CommonEngine.dialect.has_table(CommonEngine.connect(), table_name) except Exception: return False def _today_bounds_utc(now_utc: datetime) -> tuple[datetime, datetime]: start = datetime.combine(now_utc.date(), dt_time.min, tzinfo=timezone.utc) end = start + timedelta(days=1) return start, end def _normalize_status(value: str | None) -> str: return (value or "").strip().lower().replace(" ", "_") def _is_open_status(value: str | None) -> bool: return _normalize_status(value) not in _COMPLETED_STATUSES def _alert_exists_today( db: Session, *, user_id: int, alert_type: str, title: str, target_url: str | None, today_start_utc: datetime, ) -> bool: q = select(UserAlert.id).where( UserAlert.user_id == user_id, UserAlert.alert_type == alert_type, UserAlert.title == title[:255], UserAlert.created_at_utc >= today_start_utc, ) if target_url: q = q.where(UserAlert.target_url == target_url) else: q = q.where(UserAlert.target_url.is_(None)) return db.execute(q.limit(1)).scalar_one_or_none() is not None def _create_alert_once( db: Session, *, user_id: int | None, title: str, message: str | None, tenant_id: int | None, branch_id: int | None, role_context: str | None, alert_type: str, priority: str, target_url: str | None, today_start_utc: datetime, ) -> bool: if not user_id: return False if _alert_exists_today( db, user_id=int(user_id), alert_type=alert_type, title=title, target_url=target_url, today_start_utc=today_start_utc, ): return False create_alert( db, user_id=int(user_id), title=title, message=message, tenant_id=tenant_id, branch_id=branch_id, role_context=role_context, alert_type=alert_type, priority=priority, target_url=target_url, commit=False, ) return True def _service_label(subscription: ClientServiceSubscription | None, client: Client | None = None) -> str: if not subscription: return "work item" service_name = getattr(getattr(subscription, "catalogue", None), "service_name", None) client_name = getattr(client or getattr(subscription, "client", None), "client_name", None) period = getattr(subscription, "financial_year", None) or getattr(subscription, "assessment_year", None) parts = [p for p in [service_name, client_name, period] if p] return " - ".join(parts) if parts else f"Engagement #{subscription.id}" def run_notification_escalation_once(*, db: Session | None = None, now: datetime | None = None) -> AutomationResult: """Create daily alerts for due work, overdue work and basic escalations. This function is intentionally idempotent for the current UTC day. It checks the existing user_alerts table before inserting, so repeated scheduler runs do not flood users with duplicate alerts. """ owns_session = db is None session = db or CommonSessionLocal() now_utc = now.astimezone(timezone.utc) if now else datetime.now(timezone.utc) today = now_utc.date() today_start_utc, _ = _today_bounds_utc(now_utc) result = AutomationResult() try: if not _table_exists("user_alerts"): logger.warning("Phase 7O skipped: user_alerts table is not available. Apply Phase 7H first.") return result # 1) Staff task due today and overdue alerts. task_rows = session.execute( select(ClientServiceTaskInstance, ClientServiceSubscription, Client) .join(ClientServiceSubscription, ClientServiceTaskInstance.subscription_id == ClientServiceSubscription.id) .join(Client, ClientServiceTaskInstance.client_id == Client.id) .where( ClientServiceTaskInstance.is_active.is_(True), ClientServiceTaskInstance.assigned_to_user_id.is_not(None), ClientServiceTaskInstance.internal_target_date.is_not(None), ClientServiceTaskInstance.internal_target_date <= today, ) .order_by(ClientServiceTaskInstance.internal_target_date.asc(), ClientServiceTaskInstance.id.asc()) ).all() for task, subscription, client in task_rows: if not _is_open_status(task.status): continue target_url = f"/work/engagements/{subscription.id}" label = _service_label(subscription, client) task_due = task.internal_target_date if task_due == today: title = f"Task due today: {task.task_name}" if _create_alert_once( session, user_id=task.assigned_to_user_id, title=title, message=f"{label} has a task due today. Please open the work details and update the task status.", tenant_id=task.tenant_id, branch_id=task.branch_id, role_context="staff", alert_type="task_due", priority="high" if task.priority in {"high", "critical"} else "normal", target_url=target_url, today_start_utc=today_start_utc, ): result.task_due_today += 1 continue overdue_days = max(1, (today - task_due).days) title = f"Task overdue: {task.task_name}" if _create_alert_once( session, user_id=task.assigned_to_user_id, title=title, message=f"{label} is overdue by {overdue_days} day(s). Please complete or update the task status.", tenant_id=task.tenant_id, branch_id=task.branch_id, role_context="staff", alert_type="task_overdue", priority="critical" if overdue_days >= 3 else "high", target_url=target_url, today_start_utc=today_start_utc, ): result.task_overdue_staff += 1 if subscription.assigned_manager_user_id and overdue_days >= 1: if _create_alert_once( session, user_id=subscription.assigned_manager_user_id, title=f"Team task overdue: {task.task_name}", message=f"{label} has an overdue task assigned to staff. Overdue by {overdue_days} day(s).", tenant_id=task.tenant_id, branch_id=task.branch_id, role_context="manager", alert_type="task_overdue", priority="high", target_url=target_url, today_start_utc=today_start_utc, ): result.task_overdue_manager += 1 partner_user_id = subscription.assigned_partner_user_id or subscription.review_partner_user_id if partner_user_id and overdue_days >= 3: if _create_alert_once( session, user_id=partner_user_id, title=f"Escalation: task overdue for {overdue_days} days", message=f"{label} has a task pending beyond escalation threshold: {task.task_name}.", tenant_id=task.tenant_id, branch_id=task.branch_id, role_context="partner", alert_type="task_overdue", priority="critical", target_url=target_url, today_start_utc=today_start_utc, ): result.task_overdue_partner += 1 # 2) Engagement/service due and overdue alerts for responsible users and client portal user. sub_rows = session.execute( select(ClientServiceSubscription, Client) .join(Client, ClientServiceSubscription.client_id == Client.id) .where( ClientServiceSubscription.is_active.is_(True), ClientServiceSubscription.current_due_date.is_not(None), ClientServiceSubscription.current_due_date <= today + timedelta(days=2), ) .order_by(ClientServiceSubscription.current_due_date.asc(), ClientServiceSubscription.id.asc()) ).all() for subscription, client in sub_rows: if not _is_open_status(subscription.status): continue due_date = subscription.current_due_date target_url = f"/work/engagements/{subscription.id}" label = _service_label(subscription, client) recipient_map = [ (subscription.assigned_staff_user_id, "staff"), (subscription.assigned_manager_user_id, "manager"), (subscription.assigned_partner_user_id or subscription.review_partner_user_id, "partner"), (getattr(client, "portal_user_id", None), "client"), ] seen: set[int] = set() if due_date < today: overdue_days = (today - due_date).days title = f"Engagement overdue: {label}" message = f"Due date was {due_date.isoformat()} and is overdue by {overdue_days} day(s)." alert_type = "task_overdue" priority = "critical" if overdue_days >= 3 else "high" else: days_left = (due_date - today).days title = f"Due date approaching: {label}" message = "Due today." if days_left == 0 else f"Due in {days_left} day(s), on {due_date.isoformat()}." alert_type = "task_due" priority = "high" if days_left == 0 else "normal" for user_id, role_context in recipient_map: if not user_id or int(user_id) in seen: continue seen.add(int(user_id)) if _create_alert_once( session, user_id=user_id, title=title, message=message, tenant_id=subscription.tenant_id, branch_id=subscription.branch_id, role_context=role_context, alert_type=alert_type, priority=priority, target_url=target_url, today_start_utc=today_start_utc, ): if due_date < today: result.engagement_overdue += 1 else: result.engagement_due += 1 # 3) Missing punch-out reminders from yesterday. yesterday = today - timedelta(days=1) attendance_rows = session.execute( select(EmployeeAttendance, Employee) .join(Employee, EmployeeAttendance.employee_id == Employee.id) .where( EmployeeAttendance.attendance_date == yesterday, EmployeeAttendance.punch_in_utc.is_not(None), EmployeeAttendance.punch_out_utc.is_(None), Employee.user_id.is_not(None), ) ).all() for attendance, employee in attendance_rows: title = "Attendance punch-out missing" target_url = "/employee/attendance" if _create_alert_once( session, user_id=employee.user_id, title=title, message=f"Punch-out is missing for {yesterday.isoformat()}. Please regularise or contact your manager.", tenant_id=attendance.tenant_id, branch_id=attendance.branch_id, role_context="staff", alert_type="attendance", priority="normal", target_url=target_url, today_start_utc=today_start_utc, ): result.attendance_missing_punchout += 1 if employee.reporting_manager_user_id: if _create_alert_once( session, user_id=employee.reporting_manager_user_id, title=f"Team attendance punch-out missing: {employee.full_name}", message=f"{employee.full_name} has no punch-out for {yesterday.isoformat()}.", tenant_id=attendance.tenant_id, branch_id=attendance.branch_id, role_context="manager", alert_type="attendance", priority="normal", target_url="/employees/attendance", today_start_utc=today_start_utc, ): result.attendance_missing_punchout += 1 session.commit() logger.info("Phase 7O notification automation completed: %s alert(s) created", result.total_alerts) return result except Exception: session.rollback() logger.exception("Phase 7O notification automation failed") return result finally: if owns_session: session.close() def _scheduler_loop(interval_seconds: int, initial_delay_seconds: int) -> None: if initial_delay_seconds > 0: time.sleep(initial_delay_seconds) while True: run_notification_escalation_once() time.sleep(interval_seconds) def start_notification_scheduler() -> None: """Start the lightweight Phase 7O background scheduler once per process. Environment variables: AF_ALERT_AUTOMATION_ENABLED=false disables the scheduler AF_ALERT_AUTOMATION_INTERVAL_SECONDS=3600 controls repeat interval AF_ALERT_AUTOMATION_INITIAL_DELAY_SECONDS=20 controls first run delay """ global _STARTED enabled = (os.getenv("AF_ALERT_AUTOMATION_ENABLED", "true") or "true").strip().lower() if enabled in {"0", "false", "no", "off"}: logger.info("Phase 7O notification automation disabled by environment setting") return with _START_LOCK: if _STARTED: return interval = int(os.getenv("AF_ALERT_AUTOMATION_INTERVAL_SECONDS", "3600") or "3600") initial_delay = int(os.getenv("AF_ALERT_AUTOMATION_INITIAL_DELAY_SECONDS", "20") or "20") interval = max(300, interval) initial_delay = max(0, initial_delay) thread = threading.Thread( target=_scheduler_loop, args=(interval, initial_delay), name="phase7o-notification-escalation", daemon=True, ) thread.start() _STARTED = True logger.info("Phase 7O notification automation started with interval=%s seconds", interval)