From e88f9d2986c09057bd3e000c078a1573e561b8b0 Mon Sep 17 00:00:00 2001 From: jessikitty Date: Thu, 5 Feb 2026 12:21:13 +1100 Subject: [PATCH] Phase 3.1: Add chore logs API endpoints - weekly reports, user stats, verification --- backend/app/api/v1/chore_logs.py | 166 +++++++++++++++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 backend/app/api/v1/chore_logs.py diff --git a/backend/app/api/v1/chore_logs.py b/backend/app/api/v1/chore_logs.py new file mode 100644 index 0000000..0a3b1a8 --- /dev/null +++ b/backend/app/api/v1/chore_logs.py @@ -0,0 +1,166 @@ +"""Chore Completion Log API endpoints.""" +from fastapi import APIRouter, Depends, HTTPException, status, Query +from sqlalchemy.orm import Session +from sqlalchemy import func, and_ +from typing import List, Optional +from datetime import datetime, timedelta + +from app.core.database import get_db +from app.api.v1.auth import get_current_user +from app.models.user import User +from app.models.chore import Chore +from app.models.chore_assignment import ChoreAssignment +from app.models.chore_completion_log import ChoreCompletionLog +from app.schemas import chore_completion_log as log_schemas + + +router = APIRouter() + + +def enrich_completion_log(db: Session, log: ChoreCompletionLog) -> dict: + """Add related information to completion log.""" + chore = db.query(Chore).filter(Chore.id == log.chore_id).first() + user = db.query(User).filter(User.id == log.user_id).first() + + verified_by_name = None + if log.verified_by_user_id: + verified_by = db.query(User).filter(User.id == log.verified_by_user_id).first() + if verified_by: + verified_by_name = verified_by.full_name or verified_by.username + + return { + "id": log.id, + "chore_id": log.chore_id, + "user_id": log.user_id, + "completed_at": log.completed_at, + "notes": log.notes, + "verified_by_user_id": log.verified_by_user_id, + "created_at": log.created_at, + "chore_title": chore.title if chore else None, + "user_name": user.full_name or user.username if user else None, + "user_avatar": user.avatar_url if user else None, + "verified_by_name": verified_by_name + } + + +@router.post("/{chore_id}/complete", response_model=log_schemas.ChoreCompletionLog, status_code=status.HTTP_201_CREATED) +def complete_chore(chore_id: int, notes: Optional[str] = None, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): + chore = db.query(Chore).filter(Chore.id == chore_id).first() + if not chore: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Chore not found") + + assignment = db.query(ChoreAssignment).filter(ChoreAssignment.chore_id == chore_id, ChoreAssignment.user_id == current_user.id).first() + if not assignment: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="You are not assigned to this chore") + + completion_log = ChoreCompletionLog(chore_id=chore_id, user_id=current_user.id, completed_at=datetime.utcnow(), notes=notes) + db.add(completion_log) + assignment.completed_at = datetime.utcnow() + + all_assignments = db.query(ChoreAssignment).filter(ChoreAssignment.chore_id == chore_id).all() + if all(a.completed_at is not None for a in all_assignments): + chore.completed_at = datetime.utcnow() + chore.status = "completed" + + db.commit() + db.refresh(completion_log) + return enrich_completion_log(db, completion_log) + + +@router.get("/completions", response_model=List[log_schemas.ChoreCompletionLog]) +def get_completion_logs(skip: int = 0, limit: int = 100, chore_id: Optional[int] = Query(None), user_id: Optional[int] = Query(None), start_date: Optional[datetime] = Query(None), end_date: Optional[datetime] = Query(None), db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): + query = db.query(ChoreCompletionLog) + if chore_id: query = query.filter(ChoreCompletionLog.chore_id == chore_id) + if user_id: query = query.filter(ChoreCompletionLog.user_id == user_id) + if start_date: query = query.filter(ChoreCompletionLog.completed_at >= start_date) + if end_date: query = query.filter(ChoreCompletionLog.completed_at <= end_date) + query = query.order_by(ChoreCompletionLog.completed_at.desc()) + logs = query.offset(skip).limit(limit).all() + return [enrich_completion_log(db, log) for log in logs] + + +@router.get("/reports/weekly", response_model=log_schemas.WeeklyChoreReport) +def get_weekly_report(user_id: Optional[int] = Query(None), weeks_ago: int = Query(0), db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): + today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + start_of_week = today - timedelta(days=today.weekday()) - timedelta(weeks=weeks_ago) + end_of_week = start_of_week + timedelta(days=7) + + query = db.query(ChoreCompletionLog).filter(and_(ChoreCompletionLog.completed_at >= start_of_week, ChoreCompletionLog.completed_at < end_of_week)) + if user_id: query = query.filter(ChoreCompletionLog.user_id == user_id) + logs = query.all() + + completions_by_user = {} + for log in logs: + user = db.query(User).filter(User.id == log.user_id).first() + if user: + username = user.full_name or user.username + completions_by_user[username] = completions_by_user.get(username, 0) + 1 + + completions_by_chore = {} + for log in logs: + chore = db.query(Chore).filter(Chore.id == log.chore_id).first() + if chore: completions_by_chore[chore.title] = completions_by_chore.get(chore.title, 0) + 1 + + completions_by_day = {} + for log in logs: + day_name = log.completed_at.strftime("%A") + completions_by_day[day_name] = completions_by_day.get(day_name, 0) + 1 + + user_stats = [] + for user_name, count in completions_by_user.items(): + user = db.query(User).filter((User.full_name == user_name) | (User.username == user_name)).first() + user_stats.append({"username": user_name, "count": count, "avatar_url": user.avatar_url if user else None}) + user_stats.sort(key=lambda x: x["count"], reverse=True) + top_performers = user_stats[:5] + + recent_logs = sorted(logs, key=lambda x: x.completed_at, reverse=True)[:10] + recent_completions = [enrich_completion_log(db, log) for log in recent_logs] + + return {"start_date": start_of_week, "end_date": end_of_week, "total_completions": len(logs), "completions_by_user": completions_by_user, "completions_by_chore": completions_by_chore, "completions_by_day": completions_by_day, "top_performers": top_performers, "recent_completions": recent_completions} + + +@router.get("/reports/user/{user_id}", response_model=log_schemas.UserChoreStats) +def get_user_stats(user_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): + user = db.query(User).filter(User.id == user_id).first() + if not user: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") + + total_completions = db.query(ChoreCompletionLog).filter(ChoreCompletionLog.user_id == user_id).count() + + today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0) + start_of_week = today - timedelta(days=today.weekday()) + completions_this_week = db.query(ChoreCompletionLog).filter(and_(ChoreCompletionLog.user_id == user_id, ChoreCompletionLog.completed_at >= start_of_week)).count() + + start_of_month = today.replace(day=1) + completions_this_month = db.query(ChoreCompletionLog).filter(and_(ChoreCompletionLog.user_id == user_id, ChoreCompletionLog.completed_at >= start_of_month)).count() + + favorite_chore = None + chore_counts = db.query(ChoreCompletionLog.chore_id, func.count(ChoreCompletionLog.id).label('count')).filter(ChoreCompletionLog.user_id == user_id).group_by(ChoreCompletionLog.chore_id).order_by(func.count(ChoreCompletionLog.id).desc()).first() + if chore_counts: + chore = db.query(Chore).filter(Chore.id == chore_counts[0]).first() + if chore: favorite_chore = chore.title + + recent_logs = db.query(ChoreCompletionLog).filter(ChoreCompletionLog.user_id == user_id).order_by(ChoreCompletionLog.completed_at.desc()).limit(10).all() + recent_completions = [enrich_completion_log(db, log) for log in recent_logs] + + return {"user_id": user.id, "username": user.username, "full_name": user.full_name, "avatar_url": user.avatar_url, "total_completions": total_completions, "completions_this_week": completions_this_week, "completions_this_month": completions_this_month, "favorite_chore": favorite_chore, "recent_completions": recent_completions} + + +@router.delete("/completions/{log_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_completion_log(log_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): + log = db.query(ChoreCompletionLog).filter(ChoreCompletionLog.id == log_id).first() + if not log: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Completion log not found") + if not current_user.is_admin and log.user_id != current_user.id: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized") + db.delete(log) + db.commit() + return None + + +@router.post("/completions/{log_id}/verify", response_model=log_schemas.ChoreCompletionLog) +def verify_completion(log_id: int, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)): + log = db.query(ChoreCompletionLog).filter(ChoreCompletionLog.id == log_id).first() + if not log: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Completion log not found") + if log.user_id == current_user.id: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot verify own completion") + log.verified_by_user_id = current_user.id + db.commit() + db.refresh(log) + return enrich_completion_log(db, log)