"""
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
from typing import TYPE_CHECKING
from app import config
from utils.helpers import generate_uuid
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: datetime) -> list[datetime]:
"""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: datetime) -> list[datetime]:
"""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: datetime) -> str:
"""Get ISO week reference string (e.g., 2024-W48)"""
return date.strftime("%Y-W%W")
# =============================================================================
# Notification sending functions
# =============================================================================
def notify_parking_assigned(user: "User", date: str, spot_name: str):
"""Send notification when parking spot is assigned"""
if not user.notify_parking_changes:
return
date_obj = datetime.strptime(date, "%Y-%m-%d")
day_name = date_obj.strftime("%A, %B %d")
subject = f"Parking spot assigned for {day_name}"
body_html = f"""
Parking Spot Assigned
Hi {user.name},
You have been assigned a parking spot for {day_name}:
Spot {spot_name}
Best regards,
Parking Manager
"""
send_email(user.email, subject, body_html)
def notify_parking_released(user: "User", date: str, spot_name: str):
"""Send notification when parking spot is released"""
if not user.notify_parking_changes:
return
date_obj = datetime.strptime(date, "%Y-%m-%d")
day_name = date_obj.strftime("%A, %B %d")
subject = f"Parking spot released for {day_name}"
body_html = f"""
Parking Spot Released
Hi {user.name},
Your parking spot (Spot {spot_name}) for {day_name} has been released.
Best regards,
Parking Manager
"""
send_email(user.email, subject, body_html)
def notify_parking_reassigned(user: "User", date: str, spot_name: str, new_user_name: str):
"""Send notification when parking spot is reassigned to someone else"""
if not user.notify_parking_changes:
return
date_obj = datetime.strptime(date, "%Y-%m-%d")
day_name = date_obj.strftime("%A, %B %d")
subject = f"Parking spot reassigned for {day_name}"
body_html = f"""
Parking Spot Reassigned
Hi {user.name},
Your parking spot (Spot {spot_name}) for {day_name} has been reassigned to {new_user_name}.
Best regards,
Parking Manager
"""
send_email(user.email, subject, body_html)
def send_presence_reminder(user: "User", next_week_dates: list[datetime], 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().strftime("%Y-%m-%d")
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == "presence_reminder",
NotificationLog.reference_date == week_ref,
NotificationLog.sent_at >= today
).first()
if existing:
return False
# Check if week is compiled (at least 5 days marked)
date_strs = [d.strftime("%Y-%m-%d") for d in next_week_dates]
presences = db.query(UserPresence).filter(
UserPresence.user_id == user.id,
UserPresence.date.in_(date_strs)
).all()
if len(presences) >= 5:
return False
# Send reminder
start_date = next_week_dates[0].strftime("%B %d")
end_date = next_week_dates[-1].strftime("%B %d, %Y")
subject = f"Reminder: Please fill your presence for {start_date} - {end_date}"
body_html = f"""
Presence Reminder
Hi {user.name},
This is a friendly reminder to fill your presence for the upcoming week
({start_date} - {end_date}).
Please log in to the Parking Manager to mark your presence.
Best regards,
Parking Manager
"""
if send_email(user.email, subject, body_html):
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type="presence_reminder",
reference_date=week_ref,
sent_at=datetime.now().isoformat()
)
db.add(log)
db.commit()
return True
return False
def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], db: "Session") -> bool:
"""Send weekly parking assignment summary for next week (Friday at 12)"""
from database.models import DailyParkingAssignment, NotificationLog
from services.parking import get_spot_display_name
if not user.notify_weekly_parking:
return False
week_ref = get_week_reference(next_week_dates[0])
# Check if already sent for this week
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == "weekly_parking",
NotificationLog.reference_date == week_ref
).first()
if existing:
return False
# Get parking assignments for next week
date_strs = [d.strftime("%Y-%m-%d") for d in next_week_dates]
assignments = db.query(DailyParkingAssignment).filter(
DailyParkingAssignment.user_id == user.id,
DailyParkingAssignment.date.in_(date_strs)
).all()
if not assignments:
return False
# Build assignment list
assignment_lines = []
for a in sorted(assignments, key=lambda x: x.date):
date_obj = datetime.strptime(a.date, "%Y-%m-%d")
day_name = date_obj.strftime("%A")
spot_name = get_spot_display_name(a.spot_id, a.manager_id, db)
assignment_lines.append(f"{day_name}, {date_obj.strftime('%B %d')}: Spot {spot_name}")
start_date = next_week_dates[0].strftime("%B %d")
end_date = next_week_dates[-1].strftime("%B %d, %Y")
subject = f"Your parking spots for {start_date} - {end_date}"
body_html = f"""
Weekly Parking Summary
Hi {user.name},
Here are your parking spot assignments for the upcoming week:
{''.join(assignment_lines)}
Parking assignments are now frozen. You can still release or reassign your spots if needed.
Best regards,
Parking Manager
"""
if send_email(user.email, subject, body_html):
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type="weekly_parking",
reference_date=week_ref,
sent_at=datetime.now().isoformat()
)
db.add(log)
db.commit()
return True
return False
def send_daily_parking_reminder(user: "User", date: 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
if not user.notify_daily_parking:
return False
date_str = date.strftime("%Y-%m-%d")
# Check if already sent for this date
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == "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 == date_str
).first()
if not assignment:
return False
spot_name = get_spot_display_name(assignment.spot_id, assignment.manager_id, db)
day_name = date.strftime("%A, %B %d")
subject = f"Parking reminder for {day_name}"
body_html = f"""
Daily Parking Reminder
Hi {user.name},
You have a parking spot assigned for today ({day_name}):
Spot {spot_name}
Best regards,
Parking Manager
"""
if send_email(user.email, subject, body_html):
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type="daily_parking",
reference_date=date_str,
sent_at=datetime.now().isoformat()
)
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 (Mon-Fri)
"""
from database.models import User
now = datetime.now()
current_hour = now.hour
current_minute = now.minute
current_weekday = now.weekday() # 0=Monday, 6=Sunday
users = db.query(User).all()
for user in users:
# Thursday at 12: Presence reminder
if current_weekday == 3 and current_hour == 12 and current_minute < 5:
next_week = get_next_week_dates(now)
send_presence_reminder(user, next_week, db)
# Friday at 12: Weekly parking summary
if current_weekday == 4 and current_hour == 12 and current_minute < 5:
next_week = get_next_week_dates(now)
send_weekly_parking_summary(user, next_week, db)
# Daily parking reminder at user's preferred time (working days only)
if current_weekday < 5: # Monday to Friday
user_hour = user.notify_daily_parking_hour or 8
user_minute = user.notify_daily_parking_minute or 0
if current_hour == user_hour and abs(current_minute - user_minute) < 5:
send_daily_parking_reminder(user, now, db)