import logging from datetime import date, timedelta from sqlalchemy.orm import Session, joinedload from app.core.config import get_settings from app.models.enums import NotificationType, PolicyStatus, PushPlatform from app.models.policy import Policy from app.repositories.notification_repository import NotificationRepository from app.schemas.notifications import ( NotificationItem, NotificationListResponse, NotificationPreferencesResponse, NotificationPreferencesUpdate, ) from app.core.exceptions import NotFoundError from app.services.reminder_service import today_in_timezone logger = logging.getLogger(__name__) NOTIFICATION_WINDOWS: dict[NotificationType, tuple[int, int]] = { NotificationType.EXPIRING_TODAY: (0, 0), NotificationType.EXPIRING_TOMORROW: (1, 1), NotificationType.EXPIRING_NEXT_WEEK: (2, 7), } PREFERENCE_MAP = { NotificationType.EXPIRING_TODAY: "notify_expiring_today", NotificationType.EXPIRING_TOMORROW: "notify_expiring_tomorrow", NotificationType.EXPIRING_NEXT_WEEK: "notify_expiring_next_week", } SKIP_STATUSES = {PolicyStatus.RENEWED, PolicyStatus.CANCELLED, PolicyStatus.DRAFT} class NotificationService: def __init__(self, db: Session): self.db = db self.repo = NotificationRepository(db) def generate_daily_notifications(self, run_date: date | None = None) -> dict: run_date = run_date or today_in_timezone() created = 0 skipped = 0 push_sent = 0 policies = ( self.db.query(Policy) .options(joinedload(Policy.customer), joinedload(Policy.vehicle)) .filter( Policy.deleted_at.is_(None), Policy.policy_end_date.isnot(None), Policy.status.notin_(list(SKIP_STATUSES)), ) .all() ) agencies_users: dict[int, list] = {} for policy in policies: if policy.agency_id not in agencies_users: agencies_users[policy.agency_id] = self.repo.list_active_users(policy.agency_id) days_left = (policy.policy_end_date - run_date).days notification_type = self._type_for_days_left(days_left) if notification_type is None: continue for user in agencies_users[policy.agency_id]: prefs = self.repo.get_preferences(user.id) pref_key = PREFERENCE_MAP[notification_type] if not getattr(prefs, pref_key, True): skipped += 1 continue if self.repo.exists_notification(user.id, policy.id, notification_type, run_date): skipped += 1 continue title, message = self._build_message(notification_type, policy, days_left) link_path = f"/policies/{policy.id}" notification = self.repo.create_notification( { "user_id": user.id, "agency_id": policy.agency_id, "policy_id": policy.id, "notification_type": notification_type, "title": title, "message": message, "link_path": link_path, "notification_date": run_date, } ) created += 1 if prefs.push_enabled: from app.services.push_service import PushService if PushService(self.db).send_notification(user.id, title, message, link_path, policy.id): push_sent += 1 if created: self.db.commit() return { "notification_date": run_date.isoformat(), "created": created, "skipped": skipped, "push_sent": push_sent, } def list_notifications( self, user_id: int, *, unread_only: bool = False, page: int = 1, page_size: int = 20, ) -> NotificationListResponse: offset = (page - 1) * page_size items, total = self.repo.list_for_user( user_id, unread_only=unread_only, limit=page_size, offset=offset ) unread = self.repo.unread_count(user_id) return NotificationListResponse( items=[self._to_item(n) for n in items], total=total, unread_count=unread, ) def unread_count(self, user_id: int) -> int: return self.repo.unread_count(user_id) def mark_read(self, user_id: int, notification_id: int) -> NotificationItem: notification = self.repo.get_by_id(notification_id, user_id) if not notification: raise NotFoundError("Notification not found") self.repo.mark_read(notification) self.db.commit() return self._to_item(notification) def mark_all_read(self, user_id: int) -> dict: count = self.repo.mark_all_read(user_id) self.db.commit() return {"marked_read": count} def get_preferences(self, user_id: int) -> NotificationPreferencesResponse: prefs = self.repo.get_preferences(user_id) return NotificationPreferencesResponse.model_validate(prefs) def update_preferences( self, user_id: int, payload: NotificationPreferencesUpdate ) -> NotificationPreferencesResponse: prefs = self.repo.update_preferences(user_id, payload.model_dump(exclude_unset=True)) self.db.commit() return NotificationPreferencesResponse.model_validate(prefs) def register_device( self, user_id: int, platform: str, token: str, device_name: str | None ) -> dict: record = self.repo.upsert_device_token( user_id, PushPlatform(platform), token, device_name, ) self.db.commit() return {"id": record.id, "platform": record.platform.value} def unregister_device(self, user_id: int, token: str) -> dict: removed = self.repo.delete_device_token(user_id, token) self.db.commit() return {"removed": removed} @staticmethod def _type_for_days_left(days_left: int) -> NotificationType | None: for notification_type, (low, high) in NOTIFICATION_WINDOWS.items(): if low <= days_left <= high: return notification_type return None @staticmethod def _build_message(notification_type: NotificationType, policy: Policy, days_left: int) -> tuple[str, str]: customer = policy.customer.name if policy.customer else "Customer" policy_no = policy.policy_number or f"Policy #{policy.id}" vehicle = policy.vehicle.registration_number if policy.vehicle else "" if notification_type == NotificationType.EXPIRING_TODAY: title = "Policy expiring today" message = f"{customer} — {policy_no} expires today." elif notification_type == NotificationType.EXPIRING_TOMORROW: title = "Policy expiring tomorrow" message = f"{customer} — {policy_no} expires tomorrow." else: title = "Policy expiring this week" message = f"{customer} — {policy_no} expires in {days_left} days." if vehicle: message += f" Vehicle: {vehicle}." return title, message @staticmethod def _to_item(notification) -> NotificationItem: return NotificationItem( id=notification.id, notification_type=notification.notification_type.value, title=notification.title, message=notification.message, link_path=notification.link_path, policy_id=notification.policy_id, is_read=notification.is_read, notification_date=notification.notification_date.isoformat(), created_at=notification.created_at, )