stefanosalemi/org-parking
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Primo commit

Browse Source
This commit is contained in:
Stefano Rosario Salemi
2026-01-13 11:20:12 +01:00
parent ce9e2fdf2a
commit 17453f5d13
51 changed files with 3883 additions and 2508 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
services/notifications.py
View File

@@ -6,11 +6,12 @@ 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 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
@@ -72,35 +73,34 @@ def send_email(to_email: str, subject: str, body_html: str, body_text: str = Non
return False
def get_week_dates(reference_date: datetime) -> list[datetime]:
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: datetime) -> list[datetime]:
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: datetime) -> str:
def get_week_reference(date_obj: date) -> str:
"""Get ISO week reference string (e.g., 2024-W48)"""
return date.strftime("%Y-W%W")
return date_obj.strftime("%Y-W%W")
# =============================================================================
# Notification sending functions
# =============================================================================
def notify_parking_assigned(user: "User", date: str, spot_name: str):
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
date_obj = datetime.strptime(date, "%Y-%m-%d")
day_name = date_obj.strftime("%A, %B %d")
day_name = assignment_date.strftime("%A, %B %d")
subject = f"Parking spot assigned for {day_name}"
body_html = f"""
@@ -117,13 +117,12 @@ def notify_parking_assigned(user: "User", date: str, spot_name: str):
send_email(user.email, subject, body_html)
def notify_parking_released(user: "User", date: str, spot_name: str):
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
date_obj = datetime.strptime(date, "%Y-%m-%d")
day_name = date_obj.strftime("%A, %B %d")
day_name = assignment_date.strftime("%A, %B %d")
subject = f"Parking spot released for {day_name}"
body_html = f"""
@@ -139,13 +138,12 @@ def notify_parking_released(user: "User", date: str, spot_name: str):
send_email(user.email, subject, body_html)
def notify_parking_reassigned(user: "User", date: str, spot_name: str, new_user_name: str):
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
date_obj = datetime.strptime(date, "%Y-%m-%d")
day_name = date_obj.strftime("%A, %B %d")
day_name = assignment_date.strftime("%A, %B %d")
subject = f"Parking spot reassigned for {day_name}"
body_html = f"""
@@ -161,29 +159,29 @@ def notify_parking_reassigned(user: "User", date: str, spot_name: str, new_user_
send_email(user.email, subject, body_html)
def send_presence_reminder(user: "User", next_week_dates: list[datetime], db: "Session") -> bool:
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().strftime("%Y-%m-%d")
today = datetime.now().date()
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == "presence_reminder",
NotificationLog.notification_type == NotificationType.PRESENCE_REMINDER,
NotificationLog.reference_date == week_ref,
NotificationLog.sent_at >= today
NotificationLog.sent_at >= datetime.combine(today, datetime.min.time())
).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]
# DB stores dates as Date objects now
presences = db.query(UserPresence).filter(
UserPresence.user_id == user.id,
UserPresence.date.in_(date_strs)
UserPresence.date.in_(next_week_dates)
).all()
if len(presences) >= 5:
@@ -211,9 +209,9 @@ def send_presence_reminder(user: "User", next_week_dates: list[datetime], db: "S
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type="presence_reminder",
notification_type=NotificationType.PRESENCE_REMINDER,
reference_date=week_ref,
sent_at=datetime.now().isoformat()
sent_at=datetime.utcnow()
)
db.add(log)
db.commit()
@@ -222,7 +220,7 @@ def send_presence_reminder(user: "User", next_week_dates: list[datetime], db: "S
return False
def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], db: "Session") -> bool:
def send_weekly_parking_summary(user: "User", next_week_dates: List[date], 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
@@ -235,7 +233,7 @@ def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], d
# Check if already sent for this week
existing = db.query(NotificationLog).filter(
NotificationLog.user_id == user.id,
NotificationLog.notification_type == "weekly_parking",
NotificationLog.notification_type == NotificationType.WEEKLY_PARKING,
NotificationLog.reference_date == week_ref
).first()
@@ -243,10 +241,9 @@ def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], d
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)
DailyParkingAssignment.date.in_(next_week_dates)
).all()
if not assignments:
@@ -254,11 +251,11 @@ def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], d
# Build assignment list
assignment_lines = []
# a.date is now a date object
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>")
day_name = a.date.strftime("%A")
spot_name = get_spot_display_name(a.spot_id, a.office_id, db)
assignment_lines.append(f"<li>{day_name}, {a.date.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")
@@ -283,9 +280,9 @@ def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], d
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type="weekly_parking",
notification_type=NotificationType.WEEKLY_PARKING,
reference_date=week_ref,
sent_at=datetime.now().isoformat()
sent_at=datetime.utcnow()
)
db.add(log)
db.commit()
@@ -294,7 +291,7 @@ def send_weekly_parking_summary(user: "User", next_week_dates: list[datetime], d
return False
def send_daily_parking_reminder(user: "User", date: datetime, db: "Session") -> bool:
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
@@ -302,12 +299,13 @@ def send_daily_parking_reminder(user: "User", date: datetime, db: "Session") ->
if not user.notify_daily_parking:
return False
date_str = date.strftime("%Y-%m-%d")
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 == "daily_parking",
NotificationLog.notification_type == NotificationType.DAILY_PARKING,
NotificationLog.reference_date == date_str
).first()
@@ -317,14 +315,14 @@ def send_daily_parking_reminder(user: "User", date: datetime, db: "Session") ->
# Get parking assignment for this date
assignment = db.query(DailyParkingAssignment).filter(
DailyParkingAssignment.user_id == user.id,
DailyParkingAssignment.date == date_str
DailyParkingAssignment.date == assignment_date
).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")
spot_name = get_spot_display_name(assignment.spot_id, assignment.office_id, db)
day_name = date_obj.strftime("%A, %B %d")
subject = f"Parking reminder for {day_name}"
body_html = f"""
@@ -343,9 +341,9 @@ def send_daily_parking_reminder(user: "User", date: datetime, db: "Session") ->
log = NotificationLog(
id=generate_uuid(),
user_id=user.id,
notification_type="daily_parking",
notification_type=NotificationType.DAILY_PARKING,
reference_date=date_str,
sent_at=datetime.now().isoformat()
sent_at=datetime.utcnow()
)
db.add(log)
db.commit()
@@ -369,18 +367,19 @@ def run_scheduled_notifications(db: "Session"):
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 at 12: Presence reminder
if current_weekday == 3 and current_hour == 12 and current_minute < 5:
next_week = get_next_week_dates(now)
next_week = get_next_week_dates(today_date)
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)
next_week = get_next_week_dates(today_date)
send_weekly_parking_summary(user, next_week, db)
# Daily parking reminder at user's preferred time (working days only)