"""
Notification Service
Handles email notifications for presence reminders and parking assignments.
Follows org-stack pattern: direct SMTP send with file fallback when disabled.
"""
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta, date
from typing import TYPE_CHECKING, List
from app import config
from utils.helpers import generate_uuid
from database.models import NotificationType
if TYPE_CHECKING:
from sqlalchemy.orm import Session
from database.models import User
def log_email_to_file(to: str, subject: str, body: str):
"""Log email to file when SMTP is disabled (org-stack pattern)"""
timestamp = datetime.now().isoformat()
try:
with open(config.EMAIL_LOG_FILE, 'a') as f:
f.write(f'\n{"="*80}\n')
f.write(f'Timestamp: {timestamp}\n')
f.write(f'To: {to}\n')
f.write(f'Subject: {subject}\n')
f.write(f'Body:\n{body}\n')
f.write(f'{"="*80}\n')
config.logger.info(f"[EMAIL] Logged to {config.EMAIL_LOG_FILE}: {to} - {subject}")
except Exception as e:
config.logger.error(f"[EMAIL] Failed to log email: {e}")
def send_email(to_email: str, subject: str, body_html: str, body_text: str = None) -> bool:
"""
Send an email via SMTP or log to file if SMTP disabled.
Returns True if sent/logged successfully, False otherwise.
"""
# Extract plain text from HTML if not provided
if not body_text:
import re
body_text = re.sub('<[^<]+?>', '', body_html)
if not config.SMTP_ENABLED:
log_email_to_file(to_email, subject, body_text)
return True
try:
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = config.SMTP_FROM
msg["To"] = to_email
msg.attach(MIMEText(body_text, "plain"))
msg.attach(MIMEText(body_html, "html"))
with smtplib.SMTP(config.SMTP_HOST, config.SMTP_PORT) as server:
if config.SMTP_USE_TLS:
server.starttls()
if config.SMTP_USER and config.SMTP_PASSWORD:
server.login(config.SMTP_USER, config.SMTP_PASSWORD)
server.send_message(msg)
config.logger.info(f"[EMAIL] Sent to {to_email}: {subject}")
return True
except Exception as e:
config.logger.error(f"[EMAIL] Failed to send to {to_email}: {e}")
# Fallback to file logging
log_email_to_file(to_email, subject, body_text)
return False
def get_week_dates(reference_date: date) -> list[date]:
"""Get Monday-Sunday dates for the week containing reference_date"""
monday = reference_date - timedelta(days=reference_date.weekday())
return [monday + timedelta(days=i) for i in range(7)]
def get_next_week_dates(reference_date: date) -> list[date]:
"""Get Monday-Sunday dates for the week after reference_date"""
days_until_next_monday = 7 - reference_date.weekday()
next_monday = reference_date + timedelta(days=days_until_next_monday)
return [next_monday + timedelta(days=i) for i in range(7)]
def get_week_reference(date_obj: date) -> str:
"""Get ISO week reference string (e.g., 2024-W48)"""
return date_obj.strftime("%Y-W%W")
# =============================================================================
# Notification sending functions
# =============================================================================
def notify_parking_assigned(user: "User", assignment_date: date, spot_name: str):
"""Send notification when parking spot is assigned"""
if not user.notify_parking_changes:
return
day_name = assignment_date.strftime("%d/%m/%Y")
subject = f"Assegnazione Posto Auto - {day_name}"
body_html = f"""
Posto Auto Assegnato
Ciao {user.name},
Ti è stato assegnato un posto auto per il giorno {day_name}:
Posto {spot_name}
Cordiali saluti,
Team Parking Manager
"""
send_email(user.email, subject, body_html)
def notify_parking_released(user: "User", assignment_date: date, spot_name: str):
"""Send notification when parking spot is released"""
if not user.notify_parking_changes:
return
day_name = assignment_date.strftime("%d/%m/%Y")
subject = f"Rilascio Posto Auto - {day_name}"
body_html = f"""
Posto Auto Rilasciato
Ciao {user.name},
Il tuo posto auto (Posto {spot_name}) per il giorno {day_name} è stato rilasciato.
Cordiali saluti,
Team Parking Manager
"""
send_email(user.email, subject, body_html)
def notify_parking_reassigned(user: "User", assignment_date: date, spot_name: str, new_user_name: str):
"""Send notification when parking spot is reassigned to someone else"""
if not user.notify_parking_changes:
return
day_name = assignment_date.strftime("%d/%m/%Y")
subject = f"Riassegnazione Posto Auto - {day_name}"
body_html = f"""
Posto Auto Riassegnato
Ciao {user.name},
Il tuo posto auto (Posto {spot_name}) per il giorno {day_name} è stato riassegnato a {new_user_name}.
Cordiali saluti,
Team Parking Manager
"""
send_email(user.email, subject, body_html)
def send_presence_reminder(user: "User", next_week_dates: List[date], db: "Session") -> bool:
"""Send presence compilation reminder for next week"""
from database.models import UserPresence, NotificationLog
week_ref = get_week_reference(next_week_dates[0])
# Check if already sent today for this week
today = datetime.now().date()
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == NotificationType.PRESENCE_REMINDER,
NotificationLog.reference_date == week_ref,
NotificationLog.sent_at >= datetime.combine(today, datetime.min.time())
).first()
if existing:
return False
# Check if week is compiled (at least 5 days marked)
# DB stores dates as Date objects now
presences = db.query(UserPresence).filter(
UserPresence.user_id == user.id,
UserPresence.date.in_(next_week_dates)
).all()
if len(presences) >= 5:
return False
# Send reminder
start_date = next_week_dates[0].strftime("%d/%m/%Y")
end_date = next_week_dates[-1].strftime("%d/%m/%Y")
subject = f"Promemoria Presenze - Settimana {start_date} - {end_date}"
body_html = f"""
Promemoria Compilazione Presenze
Ciao {user.name},
Ti ricordiamo di compilare le tue presenze per la prossima settimana
({start_date} - {end_date}).
Accedi al Parking Manager per segnare le tue presenze.
Cordiali saluti,
Team Parking Manager
"""
if send_email(user.email, subject, body_html):
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type=NotificationType.PRESENCE_REMINDER,
reference_date=week_ref,
sent_at=datetime.utcnow()
)
db.add(log)
db.commit()
return True
return False
def send_daily_parking_reminder(user: "User", date_obj: datetime, db: "Session") -> bool:
"""Send daily parking reminder for a specific date"""
from database.models import DailyParkingAssignment, NotificationLog
from services.parking import get_spot_display_name
config.logger.info(f"[SCHEDULER] Checking daily parking reminder for user {user.email}")
if not user.notify_daily_parking:
config.logger.debug(f"[SCHEDULER] User {user.email} has disabled daily parking notifications")
return False
date_str = date_obj.strftime("%Y-%m-%d")
assignment_date = date_obj.date()
# Check if already sent for this date
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == NotificationType.DAILY_PARKING,
NotificationLog.reference_date == date_str
).first()
if existing:
return False
# Get parking assignment for this date
assignment = db.query(DailyParkingAssignment).filter(
DailyParkingAssignment.user_id == user.id,
DailyParkingAssignment.date == assignment_date
).first()
if not assignment:
return False
spot_name = get_spot_display_name(assignment.spot_id, assignment.office_id, db)
day_name = date_obj.strftime("%d/%m/%Y")
subject = f"Promemoria Parcheggio - {day_name}"
body_html = f"""
Promemoria Parcheggio Giornaliero
Ciao {user.name},
Hai un posto auto assegnato per oggi ({day_name}):
Posto {spot_name}
Cordiali saluti,
Team Parking Manager
"""
if send_email(user.email, subject, body_html):
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type=NotificationType.DAILY_PARKING,
reference_date=date_str,
sent_at=datetime.utcnow()
)
db.add(log)
db.commit()
return True
return False
def run_scheduled_notifications(db: "Session"):
"""
Run all scheduled notifications - called by a scheduler/cron job.
Schedule:
- Thursday at 12:00: Presence reminder for next week
- Friday at 12:00: Weekly parking summary
- Daily at user's preferred time: Daily parking reminder (Only on open days)
"""
from database.models import User, OfficeWeeklyClosingDay, OfficeClosingDay
from zoneinfo import ZoneInfo
# Use configured timezone
tz = ZoneInfo(config.TIMEZONE)
now = datetime.now(tz)
current_hour = now.hour
current_minute = now.minute
current_weekday = now.weekday() # 0=Monday, 6=Sunday
today_date = now.date()
users = db.query(User).all()
for user in users:
# Thursday Reminder: DISABLED as per user request
# if current_weekday == 3 and current_hour == 12 and current_minute < 5:
# next_week = get_next_week_dates(today_date)
# send_presence_reminder(user, next_week, db)
# Daily parking reminder at user's preferred time
user_hour = user.notify_daily_parking_hour or 8
user_minute = user.notify_daily_parking_minute or 0
# Check if it's the right time for this user
if current_hour == user_hour and abs(current_minute - user_minute) < 5:
config.logger.info(f"[SCHEDULER] Triggering Daily Parking Reminder check for user {user.email} (Scheduled: {user_hour}:{user_minute})")
# Check if Office is OPEN today
is_office_open = True
if user.office:
# Check weekly closing days (e.g. Sat/Sun)
# Note: WeekDay enum matches python weekday (0=Mon)
weekly_closed = db.query(OfficeWeeklyClosingDay).filter(
OfficeWeeklyClosingDay.office_id == user.office_id,
OfficeWeeklyClosingDay.weekday == current_weekday
).first()
if weekly_closed:
is_office_open = False
# Check specific closing days (Holidays)
if is_office_open:
specific_closed = db.query(OfficeClosingDay).filter(
OfficeClosingDay.office_id == user.office_id,
OfficeClosingDay.date == today_date
).first()
if specific_closed:
is_office_open = False
else:
# Fallback if no office assigned: default to Mon-Fri open
if current_weekday >= 5:
is_office_open = False
if is_office_open:
send_daily_parking_reminder(user, now, db)