312 lines
9.2 KiB
Python
312 lines
9.2 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, extract
|
|
from typing import List, Optional
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from app.db.base import get_db
|
|
from app.models.user import User
|
|
from app.models.donation import Donation, DonationGoal, TithePledge, DonationType
|
|
from app.schemas.donation import (
|
|
DonationCreate,
|
|
DonationResponse,
|
|
DonationGoalCreate,
|
|
DonationGoalResponse,
|
|
TithePledgeCreate,
|
|
TithePledgeResponse,
|
|
DonationStats,
|
|
MonthlyDonationSummary,
|
|
)
|
|
from app.api.auth import get_current_user
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/", response_model=DonationResponse)
|
|
def create_donation(
|
|
donation: DonationCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Create a new donation"""
|
|
donation_data = donation.dict()
|
|
if not donation_data.get("donation_date"):
|
|
donation_data["donation_date"] = datetime.utcnow()
|
|
|
|
# Calculate next due date for recurring donations
|
|
if donation.is_recurring and donation.recurring_frequency:
|
|
if donation.recurring_frequency == "weekly":
|
|
donation_data["next_due_date"] = donation_data["donation_date"] + timedelta(
|
|
weeks=1
|
|
)
|
|
elif donation.recurring_frequency == "monthly":
|
|
donation_data["next_due_date"] = donation_data["donation_date"] + timedelta(
|
|
days=30
|
|
)
|
|
elif donation.recurring_frequency == "yearly":
|
|
donation_data["next_due_date"] = donation_data["donation_date"] + timedelta(
|
|
days=365
|
|
)
|
|
|
|
db_donation = Donation(**donation_data, donor_id=current_user.id)
|
|
db.add(db_donation)
|
|
db.commit()
|
|
db.refresh(db_donation)
|
|
|
|
db_donation.donor_name = f"{current_user.first_name} {current_user.last_name}"
|
|
return db_donation
|
|
|
|
|
|
@router.get("/", response_model=List[DonationResponse])
|
|
def get_my_donations(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
donation_type: Optional[DonationType] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get current user's donations"""
|
|
query = db.query(Donation).filter(Donation.donor_id == current_user.id)
|
|
|
|
if donation_type:
|
|
query = query.filter(Donation.donation_type == donation_type)
|
|
|
|
donations = (
|
|
query.order_by(Donation.donation_date.desc()).offset(skip).limit(limit).all()
|
|
)
|
|
|
|
for donation in donations:
|
|
donation.donor_name = f"{current_user.first_name} {current_user.last_name}"
|
|
|
|
return donations
|
|
|
|
|
|
@router.get("/all", response_model=List[DonationResponse])
|
|
def get_all_donations(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
donation_type: Optional[DonationType] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get all donations (admin only)"""
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
|
|
query = db.query(Donation)
|
|
|
|
if donation_type:
|
|
query = query.filter(Donation.donation_type == donation_type)
|
|
|
|
donations = (
|
|
query.order_by(Donation.donation_date.desc()).offset(skip).limit(limit).all()
|
|
)
|
|
|
|
for donation in donations:
|
|
if not donation.is_anonymous:
|
|
donation.donor_name = (
|
|
f"{donation.donor.first_name} {donation.donor.last_name}"
|
|
)
|
|
else:
|
|
donation.donor_name = "Anonymous"
|
|
|
|
return donations
|
|
|
|
|
|
@router.get("/stats", response_model=DonationStats)
|
|
def get_donation_stats(
|
|
year: Optional[int] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get donation statistics"""
|
|
query = db.query(Donation)
|
|
|
|
if not current_user.is_admin:
|
|
query = query.filter(Donation.donor_id == current_user.id)
|
|
|
|
if year:
|
|
query = query.filter(extract("year", Donation.donation_date) == year)
|
|
|
|
donations = query.all()
|
|
|
|
total_donations = sum(d.amount for d in donations)
|
|
total_tithes = sum(
|
|
d.amount for d in donations if d.donation_type == DonationType.TITHE
|
|
)
|
|
total_offerings = sum(
|
|
d.amount for d in donations if d.donation_type == DonationType.OFFERING
|
|
)
|
|
donation_count = len(donations)
|
|
average_donation = (
|
|
total_donations / donation_count if donation_count > 0 else Decimal("0")
|
|
)
|
|
|
|
return DonationStats(
|
|
total_donations=total_donations,
|
|
total_tithes=total_tithes,
|
|
total_offerings=total_offerings,
|
|
donation_count=donation_count,
|
|
average_donation=average_donation,
|
|
)
|
|
|
|
|
|
@router.get("/monthly-summary", response_model=List[MonthlyDonationSummary])
|
|
def get_monthly_donation_summary(
|
|
year: int = datetime.now().year,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get monthly donation summary"""
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
|
|
results = (
|
|
db.query(
|
|
extract("month", Donation.donation_date).label("month"),
|
|
extract("year", Donation.donation_date).label("year"),
|
|
func.sum(Donation.amount).label("total_amount"),
|
|
func.count(Donation.id).label("donation_count"),
|
|
func.sum(
|
|
func.case(
|
|
(Donation.donation_type == DonationType.TITHE, Donation.amount),
|
|
else_=0,
|
|
)
|
|
).label("tithe_amount"),
|
|
func.sum(
|
|
func.case(
|
|
(Donation.donation_type == DonationType.OFFERING, Donation.amount),
|
|
else_=0,
|
|
)
|
|
).label("offering_amount"),
|
|
)
|
|
.filter(extract("year", Donation.donation_date) == year)
|
|
.group_by(
|
|
extract("month", Donation.donation_date),
|
|
extract("year", Donation.donation_date),
|
|
)
|
|
.all()
|
|
)
|
|
|
|
months = [
|
|
"",
|
|
"January",
|
|
"February",
|
|
"March",
|
|
"April",
|
|
"May",
|
|
"June",
|
|
"July",
|
|
"August",
|
|
"September",
|
|
"October",
|
|
"November",
|
|
"December",
|
|
]
|
|
|
|
summary = []
|
|
for result in results:
|
|
summary.append(
|
|
MonthlyDonationSummary(
|
|
month=months[int(result.month)],
|
|
year=int(result.year),
|
|
total_amount=result.total_amount or Decimal("0"),
|
|
donation_count=result.donation_count,
|
|
tithe_amount=result.tithe_amount or Decimal("0"),
|
|
offering_amount=result.offering_amount or Decimal("0"),
|
|
)
|
|
)
|
|
|
|
return summary
|
|
|
|
|
|
# Donation Goals
|
|
|
|
|
|
@router.post("/goals", response_model=DonationGoalResponse)
|
|
def create_donation_goal(
|
|
goal: DonationGoalCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Create a donation goal (admin only)"""
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=403, detail="Admin access required")
|
|
|
|
db_goal = DonationGoal(**goal.dict(), created_by=current_user.id)
|
|
db.add(db_goal)
|
|
db.commit()
|
|
db.refresh(db_goal)
|
|
|
|
db_goal.creator_name = f"{current_user.first_name} {current_user.last_name}"
|
|
db_goal.progress_percentage = 0.0
|
|
db_goal.days_remaining = (db_goal.end_date - datetime.utcnow()).days
|
|
|
|
return db_goal
|
|
|
|
|
|
@router.get("/goals", response_model=List[DonationGoalResponse])
|
|
def get_donation_goals(
|
|
active_only: bool = True,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get donation goals"""
|
|
query = db.query(DonationGoal)
|
|
|
|
if active_only:
|
|
query = query.filter(DonationGoal.is_active)
|
|
|
|
goals = query.order_by(DonationGoal.created_at.desc()).all()
|
|
|
|
for goal in goals:
|
|
goal.creator_name = f"{goal.creator.first_name} {goal.creator.last_name}"
|
|
goal.progress_percentage = (
|
|
(float(goal.current_amount) / float(goal.target_amount)) * 100
|
|
if goal.target_amount > 0
|
|
else 0
|
|
)
|
|
goal.days_remaining = max(0, (goal.end_date - datetime.utcnow()).days)
|
|
|
|
return goals
|
|
|
|
|
|
# Tithe Pledges
|
|
|
|
|
|
@router.post("/tithe-pledge", response_model=TithePledgeResponse)
|
|
def create_tithe_pledge(
|
|
pledge: TithePledgeCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Create or update tithe pledge"""
|
|
# Deactivate existing pledges
|
|
db.query(TithePledge).filter(
|
|
TithePledge.user_id == current_user.id, TithePledge.is_active
|
|
).update({"is_active": False})
|
|
|
|
db_pledge = TithePledge(**pledge.dict(), user_id=current_user.id)
|
|
db.add(db_pledge)
|
|
db.commit()
|
|
db.refresh(db_pledge)
|
|
|
|
return db_pledge
|
|
|
|
|
|
@router.get("/tithe-pledge", response_model=TithePledgeResponse)
|
|
def get_my_tithe_pledge(
|
|
db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Get current user's active tithe pledge"""
|
|
pledge = (
|
|
db.query(TithePledge)
|
|
.filter(TithePledge.user_id == current_user.id, TithePledge.is_active)
|
|
.first()
|
|
)
|
|
|
|
if not pledge:
|
|
raise HTTPException(status_code=404, detail="No active tithe pledge found")
|
|
|
|
return pledge
|