325 lines
10 KiB
Python
325 lines
10 KiB
Python
"""
|
|
Parking Assignment Service
|
|
Manager-centric parking spot management with fairness algorithm
|
|
|
|
Key concepts:
|
|
- Managers own parking spots (defined by manager_parking_quota)
|
|
- Each manager has a spot prefix (A, B, C...) for display names
|
|
- Spots are named like A1, A2, B1, B2 based on manager prefix
|
|
- Fairness: users with lowest parking_days/presence_days ratio get priority
|
|
"""
|
|
from datetime import datetime, timezone
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_
|
|
|
|
from database.models import (
|
|
DailyParkingAssignment, User, UserPresence,
|
|
ParkingGuarantee, ParkingExclusion, ManagerClosingDay, ManagerWeeklyClosingDay
|
|
)
|
|
from utils.helpers import generate_uuid
|
|
from app import config
|
|
|
|
|
|
def get_spot_prefix(manager: User, db: Session) -> str:
|
|
"""Get the spot prefix for a manager (from manager_spot_prefix or auto-assign)"""
|
|
if manager.manager_spot_prefix:
|
|
return manager.manager_spot_prefix
|
|
|
|
# Auto-assign based on alphabetical order of managers without prefix
|
|
managers = db.query(User).filter(
|
|
User.role == "manager",
|
|
User.manager_spot_prefix == None
|
|
).order_by(User.name).all()
|
|
|
|
# Find existing prefixes
|
|
existing_prefixes = set(
|
|
m.manager_spot_prefix for m in db.query(User).filter(
|
|
User.role == "manager",
|
|
User.manager_spot_prefix != None
|
|
).all()
|
|
)
|
|
|
|
# Find first available letter
|
|
manager_index = next((i for i, m in enumerate(managers) if m.id == manager.id), 0)
|
|
letter = 'A'
|
|
count = 0
|
|
while letter in existing_prefixes or count < manager_index:
|
|
if letter not in existing_prefixes:
|
|
count += 1
|
|
letter = chr(ord(letter) + 1)
|
|
if ord(letter) > ord('Z'):
|
|
letter = 'A'
|
|
break
|
|
|
|
return letter
|
|
|
|
|
|
def get_spot_display_name(spot_id: str, manager_id: str, db: Session) -> str:
|
|
"""Get display name for a spot (e.g., 'A3' instead of 'spot-3')"""
|
|
manager = db.query(User).filter(User.id == manager_id).first()
|
|
if not manager:
|
|
return spot_id
|
|
|
|
prefix = get_spot_prefix(manager, db)
|
|
spot_number = spot_id.replace("spot-", "")
|
|
return f"{prefix}{spot_number}"
|
|
|
|
|
|
def is_closing_day(manager_id: str, date: str, db: Session) -> bool:
|
|
"""
|
|
Check if date is a closing day for this manager.
|
|
Checks both specific closing days and weekly recurring closing days.
|
|
"""
|
|
# Check specific closing day
|
|
specific = db.query(ManagerClosingDay).filter(
|
|
ManagerClosingDay.manager_id == manager_id,
|
|
ManagerClosingDay.date == date
|
|
).first()
|
|
if specific:
|
|
return True
|
|
|
|
# Check weekly closing day
|
|
date_obj = datetime.strptime(date, "%Y-%m-%d")
|
|
weekday = date_obj.weekday() # 0=Monday in Python
|
|
# Convert to our format: 0=Sunday, 1=Monday, ..., 6=Saturday
|
|
weekday_sunday_start = (weekday + 1) % 7
|
|
|
|
weekly = db.query(ManagerWeeklyClosingDay).filter(
|
|
ManagerWeeklyClosingDay.manager_id == manager_id,
|
|
ManagerWeeklyClosingDay.weekday == weekday_sunday_start
|
|
).first()
|
|
|
|
return weekly is not None
|
|
|
|
|
|
def initialize_parking_pool(manager_id: str, quota: int, date: str, db: Session) -> int:
|
|
"""Initialize empty parking spots for a manager's pool on a given date.
|
|
Returns 0 if it's a closing day (no parking available).
|
|
"""
|
|
# Don't create pool on closing days
|
|
if is_closing_day(manager_id, date, db):
|
|
return 0
|
|
|
|
existing = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.manager_id == manager_id,
|
|
DailyParkingAssignment.date == date
|
|
).count()
|
|
|
|
if existing > 0:
|
|
return existing
|
|
|
|
for i in range(1, quota + 1):
|
|
spot = DailyParkingAssignment(
|
|
id=generate_uuid(),
|
|
date=date,
|
|
spot_id=f"spot-{i}",
|
|
user_id=None,
|
|
manager_id=manager_id,
|
|
created_at=datetime.now(timezone.utc).isoformat()
|
|
)
|
|
db.add(spot)
|
|
|
|
db.commit()
|
|
config.logger.debug(f"Initialized {quota} parking spots for manager {manager_id} on {date}")
|
|
return quota
|
|
|
|
|
|
def get_user_parking_ratio(user_id: str, manager_id: str, db: Session) -> float:
|
|
"""
|
|
Calculate user's parking ratio: parking_days / presence_days
|
|
Lower ratio = higher priority for next parking spot
|
|
"""
|
|
# Count days user was present
|
|
presence_days = db.query(UserPresence).filter(
|
|
UserPresence.user_id == user_id,
|
|
UserPresence.status == "present"
|
|
).count()
|
|
|
|
if presence_days == 0:
|
|
return 0.0 # New user, highest priority
|
|
|
|
# Count days user got parking
|
|
parking_days = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.user_id == user_id,
|
|
DailyParkingAssignment.manager_id == manager_id
|
|
).count()
|
|
|
|
return parking_days / presence_days
|
|
|
|
|
|
def is_user_excluded(user_id: str, manager_id: str, date: str, db: Session) -> bool:
|
|
"""Check if user is excluded from parking for this date"""
|
|
exclusion = db.query(ParkingExclusion).filter(
|
|
ParkingExclusion.manager_id == manager_id,
|
|
ParkingExclusion.user_id == user_id
|
|
).first()
|
|
|
|
if not exclusion:
|
|
return False
|
|
|
|
# Check date range
|
|
if exclusion.start_date and date < exclusion.start_date:
|
|
return False
|
|
if exclusion.end_date and date > exclusion.end_date:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def has_guarantee(user_id: str, manager_id: str, date: str, db: Session) -> bool:
|
|
"""Check if user has a parking guarantee for this date"""
|
|
guarantee = db.query(ParkingGuarantee).filter(
|
|
ParkingGuarantee.manager_id == manager_id,
|
|
ParkingGuarantee.user_id == user_id
|
|
).first()
|
|
|
|
if not guarantee:
|
|
return False
|
|
|
|
# Check date range
|
|
if guarantee.start_date and date < guarantee.start_date:
|
|
return False
|
|
if guarantee.end_date and date > guarantee.end_date:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def get_users_wanting_parking(manager_id: str, date: str, db: Session) -> list[dict]:
|
|
"""
|
|
Get all users who want parking for this date, sorted by fairness priority.
|
|
Returns list of {user_id, has_guarantee, ratio}
|
|
|
|
Note: Manager is part of their own team and can get parking from their pool.
|
|
"""
|
|
# Get users who marked "present" for this date:
|
|
# - Users managed by this manager (User.manager_id == manager_id)
|
|
# - The manager themselves (User.id == manager_id)
|
|
present_users = db.query(UserPresence).join(User).filter(
|
|
UserPresence.date == date,
|
|
UserPresence.status == "present",
|
|
or_(User.manager_id == manager_id, User.id == manager_id)
|
|
).all()
|
|
|
|
candidates = []
|
|
for presence in present_users:
|
|
user_id = presence.user_id
|
|
|
|
# Skip excluded users
|
|
if is_user_excluded(user_id, manager_id, date, db):
|
|
continue
|
|
|
|
# Skip users who already have a spot
|
|
existing = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.date == date,
|
|
DailyParkingAssignment.user_id == user_id
|
|
).first()
|
|
if existing:
|
|
continue
|
|
|
|
candidates.append({
|
|
"user_id": user_id,
|
|
"has_guarantee": has_guarantee(user_id, manager_id, date, db),
|
|
"ratio": get_user_parking_ratio(user_id, manager_id, db)
|
|
})
|
|
|
|
# Sort: guaranteed users first, then by ratio (lowest first for fairness)
|
|
candidates.sort(key=lambda x: (not x["has_guarantee"], x["ratio"]))
|
|
|
|
return candidates
|
|
|
|
|
|
def assign_parking_fairly(manager_id: str, date: str, db: Session) -> dict:
|
|
"""
|
|
Assign parking spots fairly based on parking ratio.
|
|
Called after presence is set for a date.
|
|
Returns {assigned: [...], waitlist: [...]}
|
|
"""
|
|
manager = db.query(User).filter(User.id == manager_id).first()
|
|
if not manager or not manager.manager_parking_quota:
|
|
return {"assigned": [], "waitlist": []}
|
|
|
|
# No parking on closing days
|
|
if is_closing_day(manager_id, date, db):
|
|
return {"assigned": [], "waitlist": [], "closed": True}
|
|
|
|
# Initialize pool
|
|
initialize_parking_pool(manager_id, manager.manager_parking_quota, date, db)
|
|
|
|
# Get candidates sorted by fairness
|
|
candidates = get_users_wanting_parking(manager_id, date, db)
|
|
|
|
# Get available spots
|
|
free_spots = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.manager_id == manager_id,
|
|
DailyParkingAssignment.date == date,
|
|
DailyParkingAssignment.user_id == None
|
|
).all()
|
|
|
|
assigned = []
|
|
waitlist = []
|
|
|
|
for candidate in candidates:
|
|
if free_spots:
|
|
spot = free_spots.pop(0)
|
|
spot.user_id = candidate["user_id"]
|
|
assigned.append(candidate["user_id"])
|
|
else:
|
|
waitlist.append(candidate["user_id"])
|
|
|
|
db.commit()
|
|
|
|
return {"assigned": assigned, "waitlist": waitlist}
|
|
|
|
|
|
def release_user_spot(manager_id: str, user_id: str, date: str, db: Session) -> bool:
|
|
"""Release a user's parking spot and reassign to next in fairness queue"""
|
|
assignment = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.manager_id == manager_id,
|
|
DailyParkingAssignment.date == date,
|
|
DailyParkingAssignment.user_id == user_id
|
|
).first()
|
|
|
|
if not assignment:
|
|
return False
|
|
|
|
# Release the spot
|
|
assignment.user_id = None
|
|
db.commit()
|
|
|
|
# Try to assign to next user in fairness queue
|
|
candidates = get_users_wanting_parking(manager_id, date, db)
|
|
if candidates:
|
|
assignment.user_id = candidates[0]["user_id"]
|
|
db.commit()
|
|
|
|
return True
|
|
|
|
|
|
def handle_presence_change(user_id: str, date: str, old_status: str, new_status: str, manager_id: str, db: Session):
|
|
"""
|
|
Handle presence status change and update parking accordingly.
|
|
Uses fairness algorithm for assignment.
|
|
manager_id is the user's manager (from User.manager_id).
|
|
"""
|
|
# Don't process past dates
|
|
target_date = datetime.strptime(date, "%Y-%m-%d").date()
|
|
if target_date < datetime.now().date():
|
|
return
|
|
|
|
# Get manager
|
|
manager = db.query(User).filter(User.id == manager_id, User.role == "manager").first()
|
|
if not manager or not manager.manager_parking_quota:
|
|
return
|
|
|
|
# Initialize pool if needed
|
|
initialize_parking_pool(manager.id, manager.manager_parking_quota, date, db)
|
|
|
|
if old_status == "present" and new_status in ["remote", "absent"]:
|
|
# User no longer coming - release their spot (will auto-reassign)
|
|
release_user_spot(manager.id, user_id, date, db)
|
|
|
|
elif new_status == "present":
|
|
# User coming in - run fair assignment for this date
|
|
assign_parking_fairly(manager.id, date, db)
|