468 lines
16 KiB
Python
468 lines
16 KiB
Python
"""
|
|
Presence Management Routes
|
|
User presence marking and admin management
|
|
"""
|
|
from typing import List
|
|
from datetime import datetime, timedelta
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
|
|
from database.connection import get_db
|
|
from database.models import UserPresence, User, DailyParkingAssignment
|
|
from utils.auth_middleware import get_current_user, require_manager_or_admin
|
|
from utils.helpers import generate_uuid
|
|
from services.parking import handle_presence_change, get_spot_display_name
|
|
from app import config
|
|
|
|
router = APIRouter(prefix="/api/presence", tags=["presence"])
|
|
|
|
|
|
# Request/Response Models
|
|
class PresenceMarkRequest(BaseModel):
|
|
date: str # YYYY-MM-DD
|
|
status: str # present, remote, absent
|
|
|
|
|
|
class AdminPresenceMarkRequest(BaseModel):
|
|
user_id: str
|
|
date: str
|
|
status: str
|
|
|
|
|
|
class BulkPresenceRequest(BaseModel):
|
|
start_date: str
|
|
end_date: str
|
|
status: str
|
|
days: List[int] | None = None # Optional: [0,1,2,3,4] for Mon-Fri
|
|
|
|
|
|
class AdminBulkPresenceRequest(BaseModel):
|
|
user_id: str
|
|
start_date: str
|
|
end_date: str
|
|
status: str
|
|
days: List[int] | None = None
|
|
|
|
|
|
class PresenceResponse(BaseModel):
|
|
id: str
|
|
user_id: str
|
|
date: str
|
|
status: str
|
|
created_at: str | None
|
|
updated_at: str | None
|
|
parking_spot_number: str | None = None
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
# Helper functions
|
|
def validate_status(status: str):
|
|
if status not in ["present", "remote", "absent"]:
|
|
raise HTTPException(status_code=400, detail="Status must be: present, remote, or absent")
|
|
|
|
|
|
def parse_date(date_str: str) -> datetime:
|
|
try:
|
|
return datetime.strptime(date_str, "%Y-%m-%d")
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD")
|
|
|
|
|
|
def check_manager_access(current_user: User, target_user: User, db: Session):
|
|
"""Check if current_user has access to target_user"""
|
|
if current_user.role == "admin":
|
|
return True
|
|
|
|
if current_user.role == "manager":
|
|
# Manager can access users they manage
|
|
if target_user.manager_id == current_user.id:
|
|
return True
|
|
raise HTTPException(status_code=403, detail="User is not managed by you")
|
|
|
|
raise HTTPException(status_code=403, detail="Access denied")
|
|
|
|
|
|
def _mark_presence_for_user(
|
|
user_id: str,
|
|
date: str,
|
|
status: str,
|
|
db: Session,
|
|
target_user: User
|
|
) -> UserPresence:
|
|
"""
|
|
Core presence marking logic - shared by user and admin routes.
|
|
"""
|
|
validate_status(status)
|
|
parse_date(date)
|
|
|
|
existing = db.query(UserPresence).filter(
|
|
UserPresence.user_id == user_id,
|
|
UserPresence.date == date
|
|
).first()
|
|
|
|
now = datetime.utcnow().isoformat()
|
|
old_status = existing.status if existing else None
|
|
|
|
if existing:
|
|
existing.status = status
|
|
existing.updated_at = now
|
|
db.commit()
|
|
db.refresh(existing)
|
|
presence = existing
|
|
else:
|
|
presence = UserPresence(
|
|
id=generate_uuid(),
|
|
user_id=user_id,
|
|
date=date,
|
|
status=status,
|
|
created_at=now,
|
|
updated_at=now
|
|
)
|
|
db.add(presence)
|
|
db.commit()
|
|
db.refresh(presence)
|
|
|
|
# Handle parking assignment
|
|
# Use manager_id if user has one, or user's own id if they are a manager
|
|
parking_manager_id = target_user.manager_id
|
|
if not parking_manager_id and target_user.role == "manager":
|
|
# Manager is part of their own team for parking purposes
|
|
parking_manager_id = target_user.id
|
|
|
|
if old_status != status and parking_manager_id:
|
|
try:
|
|
handle_presence_change(
|
|
user_id, date,
|
|
old_status or "absent", status,
|
|
parking_manager_id, db
|
|
)
|
|
except Exception as e:
|
|
config.logger.warning(f"Parking handler failed for user {user_id} on {date}: {e}")
|
|
|
|
return presence
|
|
|
|
|
|
def _bulk_mark_presence(
|
|
user_id: str,
|
|
start_date: str,
|
|
end_date: str,
|
|
status: str,
|
|
days: List[int] | None,
|
|
db: Session,
|
|
target_user: User
|
|
) -> List[UserPresence]:
|
|
"""
|
|
Core bulk presence marking logic - shared by user and admin routes.
|
|
"""
|
|
validate_status(status)
|
|
start = parse_date(start_date)
|
|
end = parse_date(end_date)
|
|
|
|
if end < start:
|
|
raise HTTPException(status_code=400, detail="End date must be after start date")
|
|
if (end - start).days > 90:
|
|
raise HTTPException(status_code=400, detail="Range cannot exceed 90 days")
|
|
|
|
results = []
|
|
current_date = start
|
|
now = datetime.utcnow().isoformat()
|
|
|
|
while current_date <= end:
|
|
if days is None or current_date.weekday() in days:
|
|
date_str = current_date.strftime("%Y-%m-%d")
|
|
|
|
existing = db.query(UserPresence).filter(
|
|
UserPresence.user_id == user_id,
|
|
UserPresence.date == date_str
|
|
).first()
|
|
|
|
old_status = existing.status if existing else None
|
|
|
|
if existing:
|
|
existing.status = status
|
|
existing.updated_at = now
|
|
results.append(existing)
|
|
else:
|
|
presence = UserPresence(
|
|
id=generate_uuid(),
|
|
user_id=user_id,
|
|
date=date_str,
|
|
status=status,
|
|
created_at=now,
|
|
updated_at=now
|
|
)
|
|
db.add(presence)
|
|
results.append(presence)
|
|
|
|
# Handle parking for each date
|
|
# Use manager_id if user has one, or user's own id if they are a manager
|
|
parking_manager_id = target_user.manager_id
|
|
if not parking_manager_id and target_user.role == "manager":
|
|
parking_manager_id = target_user.id
|
|
|
|
if old_status != status and parking_manager_id:
|
|
try:
|
|
handle_presence_change(
|
|
user_id, date_str,
|
|
old_status or "absent", status,
|
|
parking_manager_id, db
|
|
)
|
|
except Exception as e:
|
|
config.logger.warning(f"Parking handler failed for user {user_id} on {date_str}: {e}")
|
|
|
|
current_date += timedelta(days=1)
|
|
|
|
db.commit()
|
|
return results
|
|
|
|
|
|
def _delete_presence(
|
|
user_id: str,
|
|
date: str,
|
|
db: Session,
|
|
target_user: User
|
|
) -> dict:
|
|
"""
|
|
Core presence deletion logic - shared by user and admin routes.
|
|
"""
|
|
parse_date(date)
|
|
|
|
presence = db.query(UserPresence).filter(
|
|
UserPresence.user_id == user_id,
|
|
UserPresence.date == date
|
|
).first()
|
|
|
|
if not presence:
|
|
raise HTTPException(status_code=404, detail="Presence not found")
|
|
|
|
old_status = presence.status
|
|
db.delete(presence)
|
|
db.commit()
|
|
|
|
# Use manager_id if user has one, or user's own id if they are a manager
|
|
parking_manager_id = target_user.manager_id
|
|
if not parking_manager_id and target_user.role == "manager":
|
|
parking_manager_id = target_user.id
|
|
|
|
if parking_manager_id:
|
|
try:
|
|
handle_presence_change(
|
|
user_id, date,
|
|
old_status, "absent",
|
|
parking_manager_id, db
|
|
)
|
|
except Exception as e:
|
|
config.logger.warning(f"Parking handler failed for user {user_id} on {date}: {e}")
|
|
|
|
return {"message": "Presence deleted"}
|
|
|
|
|
|
# User Routes
|
|
@router.post("/mark", response_model=PresenceResponse)
|
|
def mark_presence(data: PresenceMarkRequest, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
|
|
"""Mark presence for a date"""
|
|
return _mark_presence_for_user(current_user.id, data.date, data.status, db, current_user)
|
|
|
|
|
|
@router.post("/mark-bulk", response_model=List[PresenceResponse])
|
|
def mark_bulk_presence(data: BulkPresenceRequest, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
|
|
"""Mark presence for a date range"""
|
|
return _bulk_mark_presence(
|
|
current_user.id, data.start_date, data.end_date,
|
|
data.status, data.days, db, current_user
|
|
)
|
|
|
|
|
|
@router.get("/my-presences", response_model=List[PresenceResponse])
|
|
def get_my_presences(start_date: str = None, end_date: str = None, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
|
|
"""Get current user's presences"""
|
|
query = db.query(UserPresence).filter(UserPresence.user_id == current_user.id)
|
|
|
|
if start_date:
|
|
parse_date(start_date)
|
|
query = query.filter(UserPresence.date >= start_date)
|
|
if end_date:
|
|
parse_date(end_date)
|
|
query = query.filter(UserPresence.date <= end_date)
|
|
|
|
return query.order_by(UserPresence.date.desc()).all()
|
|
|
|
|
|
@router.delete("/{date}")
|
|
def delete_presence(date: str, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
|
|
"""Delete presence for a date"""
|
|
return _delete_presence(current_user.id, date, db, current_user)
|
|
|
|
|
|
# Admin/Manager Routes
|
|
@router.post("/admin/mark", response_model=PresenceResponse)
|
|
def admin_mark_presence(data: AdminPresenceMarkRequest, db: Session = Depends(get_db), current_user=Depends(require_manager_or_admin)):
|
|
"""Mark presence for any user (manager/admin)"""
|
|
target_user = db.query(User).filter(User.id == data.user_id).first()
|
|
if not target_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
check_manager_access(current_user, target_user, db)
|
|
return _mark_presence_for_user(data.user_id, data.date, data.status, db, target_user)
|
|
|
|
|
|
@router.post("/admin/mark-bulk", response_model=List[PresenceResponse])
|
|
def admin_mark_bulk_presence(data: AdminBulkPresenceRequest, db: Session = Depends(get_db), current_user=Depends(require_manager_or_admin)):
|
|
"""Bulk mark presence for any user (manager/admin)"""
|
|
target_user = db.query(User).filter(User.id == data.user_id).first()
|
|
if not target_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
check_manager_access(current_user, target_user, db)
|
|
return _bulk_mark_presence(
|
|
data.user_id, data.start_date, data.end_date,
|
|
data.status, data.days, db, target_user
|
|
)
|
|
|
|
|
|
@router.delete("/admin/{user_id}/{date}")
|
|
def admin_delete_presence(user_id: str, date: str, db: Session = Depends(get_db), current_user=Depends(require_manager_or_admin)):
|
|
"""Delete presence for any user (manager/admin)"""
|
|
target_user = db.query(User).filter(User.id == user_id).first()
|
|
if not target_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
check_manager_access(current_user, target_user, db)
|
|
return _delete_presence(user_id, date, db, target_user)
|
|
|
|
|
|
@router.get("/team")
|
|
def get_team_presences(start_date: str, end_date: str, manager_id: str = None, db: Session = Depends(get_db), current_user=Depends(get_current_user)):
|
|
"""Get team presences with parking info, filtered by manager.
|
|
- Admins can see all teams
|
|
- Managers see their own team
|
|
- Employees can only see their own team (read-only view)
|
|
"""
|
|
parse_date(start_date)
|
|
parse_date(end_date)
|
|
|
|
# Get users based on permissions and manager filter
|
|
# Note: Manager is part of their own team (for parking assignment purposes)
|
|
if current_user.role == "employee":
|
|
# Employees can only see their own team (users with same manager_id + the manager)
|
|
if not current_user.manager_id:
|
|
return [] # No manager assigned, no team to show
|
|
users = db.query(User).filter(
|
|
(User.manager_id == current_user.manager_id) | (User.id == current_user.manager_id)
|
|
).all()
|
|
elif manager_id:
|
|
# Filter by specific manager (for admins/managers) - include the manager themselves
|
|
users = db.query(User).filter(
|
|
(User.manager_id == manager_id) | (User.id == manager_id)
|
|
).all()
|
|
elif current_user.role == "admin":
|
|
# Admin sees all users
|
|
users = db.query(User).all()
|
|
else:
|
|
# Manager sees their team + themselves
|
|
users = db.query(User).filter(
|
|
(User.manager_id == current_user.id) | (User.id == current_user.id)
|
|
).all()
|
|
|
|
# Batch query presences and parking for all users
|
|
user_ids = [u.id for u in users]
|
|
presences = db.query(UserPresence).filter(
|
|
UserPresence.user_id.in_(user_ids),
|
|
UserPresence.date >= start_date,
|
|
UserPresence.date <= end_date
|
|
).all()
|
|
|
|
parking_assignments = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.user_id.in_(user_ids),
|
|
DailyParkingAssignment.date >= start_date,
|
|
DailyParkingAssignment.date <= end_date
|
|
).all()
|
|
|
|
# Build lookups
|
|
parking_lookup = {}
|
|
parking_info_lookup = {}
|
|
for p in parking_assignments:
|
|
if p.user_id not in parking_lookup:
|
|
parking_lookup[p.user_id] = []
|
|
parking_info_lookup[p.user_id] = []
|
|
parking_lookup[p.user_id].append(p.date)
|
|
spot_display_name = get_spot_display_name(p.spot_id, p.manager_id, db)
|
|
parking_info_lookup[p.user_id].append({
|
|
"id": p.id,
|
|
"date": p.date,
|
|
"spot_id": p.spot_id,
|
|
"spot_display_name": spot_display_name
|
|
})
|
|
|
|
# Build manager lookup for display
|
|
manager_ids = list(set(u.manager_id for u in users if u.manager_id))
|
|
managers = db.query(User).filter(User.id.in_(manager_ids)).all() if manager_ids else []
|
|
manager_lookup = {m.id: m.name for m in managers}
|
|
|
|
# Build response
|
|
result = []
|
|
for user in users:
|
|
user_presences = [p for p in presences if p.user_id == user.id]
|
|
|
|
result.append({
|
|
"id": user.id,
|
|
"name": user.name,
|
|
"manager_id": user.manager_id,
|
|
"manager_name": manager_lookup.get(user.manager_id),
|
|
"presences": [{"date": p.date, "status": p.status} for p in user_presences],
|
|
"parking_dates": parking_lookup.get(user.id, []),
|
|
"parking_info": parking_info_lookup.get(user.id, [])
|
|
})
|
|
|
|
return result
|
|
|
|
|
|
@router.get("/admin/{user_id}")
|
|
def get_user_presences(user_id: str, start_date: str = None, end_date: str = None, db: Session = Depends(get_db), current_user=Depends(require_manager_or_admin)):
|
|
"""Get any user's presences with parking info (manager/admin)"""
|
|
target_user = db.query(User).filter(User.id == user_id).first()
|
|
if not target_user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
check_manager_access(current_user, target_user, db)
|
|
|
|
query = db.query(UserPresence).filter(UserPresence.user_id == user_id)
|
|
|
|
if start_date:
|
|
parse_date(start_date)
|
|
query = query.filter(UserPresence.date >= start_date)
|
|
if end_date:
|
|
parse_date(end_date)
|
|
query = query.filter(UserPresence.date <= end_date)
|
|
|
|
presences = query.order_by(UserPresence.date.desc()).all()
|
|
|
|
# Batch query parking assignments
|
|
date_strs = [p.date for p in presences]
|
|
parking_map = {}
|
|
if date_strs:
|
|
assignments = db.query(DailyParkingAssignment).filter(
|
|
DailyParkingAssignment.user_id == user_id,
|
|
DailyParkingAssignment.date.in_(date_strs)
|
|
).all()
|
|
for a in assignments:
|
|
parking_map[a.date] = get_spot_display_name(a.spot_id, a.manager_id, db)
|
|
|
|
# Build response
|
|
result = []
|
|
for presence in presences:
|
|
result.append({
|
|
"id": presence.id,
|
|
"user_id": presence.user_id,
|
|
"date": presence.date,
|
|
"status": presence.status,
|
|
"created_at": presence.created_at,
|
|
"updated_at": presence.updated_at,
|
|
"parking_spot_number": parking_map.get(presence.date)
|
|
})
|
|
|
|
return result
|