392 lines
13 KiB
Python
392 lines
13 KiB
Python
"""
|
|
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"""
|
|
<html>
|
|
<body>
|
|
<h2>Parking Spot Assigned</h2>
|
|
<p>Hi {user.name},</p>
|
|
<p>You have been assigned a parking spot for {day_name}:</p>
|
|
<p style="font-size: 18px; font-weight: bold;">Spot {spot_name}</p>
|
|
<p>Best regards,<br>Parking Manager</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
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"""
|
|
<html>
|
|
<body>
|
|
<h2>Parking Spot Released</h2>
|
|
<p>Hi {user.name},</p>
|
|
<p>Your parking spot (Spot {spot_name}) for {day_name} has been released.</p>
|
|
<p>Best regards,<br>Parking Manager</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
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"""
|
|
<html>
|
|
<body>
|
|
<h2>Parking Spot Reassigned</h2>
|
|
<p>Hi {user.name},</p>
|
|
<p>Your parking spot (Spot {spot_name}) for {day_name} has been reassigned to {new_user_name}.</p>
|
|
<p>Best regards,<br>Parking Manager</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
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"""
|
|
<html>
|
|
<body>
|
|
<h2>Presence Reminder</h2>
|
|
<p>Hi {user.name},</p>
|
|
<p>This is a friendly reminder to fill your presence for the upcoming week
|
|
({start_date} - {end_date}).</p>
|
|
<p>Please log in to the Parking Manager to mark your presence.</p>
|
|
<p>Best regards,<br>Parking Manager</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
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"<li>{day_name}, {date_obj.strftime('%B %d')}: Spot {spot_name}</li>")
|
|
|
|
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"""
|
|
<html>
|
|
<body>
|
|
<h2>Weekly Parking Summary</h2>
|
|
<p>Hi {user.name},</p>
|
|
<p>Here are your parking spot assignments for the upcoming week:</p>
|
|
<ul>
|
|
{''.join(assignment_lines)}
|
|
</ul>
|
|
<p>Parking assignments are now frozen. You can still release or reassign your spots if needed.</p>
|
|
<p>Best regards,<br>Parking Manager</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
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"""
|
|
<html>
|
|
<body>
|
|
<h2>Daily Parking Reminder</h2>
|
|
<p>Hi {user.name},</p>
|
|
<p>You have a parking spot assigned for today ({day_name}):</p>
|
|
<p style="font-size: 18px; font-weight: bold;">Spot {spot_name}</p>
|
|
<p>Best regards,<br>Parking Manager</p>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
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)
|