from datetime import datetime from decimal import Decimal, InvalidOperation from flask import Blueprint, abort, flash, redirect, render_template, request, url_for from flask_login import current_user from app.auth.access import roles_required from app.extensions import db from app.models import ( Activity, ActivityAttendance, ActivityImage, ActivityRegistration, ActivityType, CauseCategory, City, Location, DonationCause, DonationCauseImage, DonationSuggestedAmount, HeartRule, HeartTransaction, Partner, ) from app.models.enums import ActivityStatus, CauseStatus, RegistrationStatus from app.notifications.service import NotificationService from .services import can_transition_activity bp = Blueprint("admin", __name__, url_prefix="/admin") ADMIN_ROLES = ("SUPER_ADMIN", "PIRAMAL_ADMIN", "INDIAISUS_ADMIN") def _parse_dt(value): if not value: return None return datetime.strptime(value, "%Y-%m-%dT%H:%M") def _parse_date(value): if not value: return None return datetime.strptime(value, "%Y-%m-%d").date() def _next_code(prefix, model, field): ym = datetime.now().strftime("%Y%m") like_value = f"{prefix}-{ym}-%" count = model.query.filter(getattr(model, field).like(like_value)).count() + 1 return f"{prefix}-{ym}-{count:03d}" def _activity_form_context(activity=None): return { "activity": activity, "cities": City.query.filter_by(is_active=True).order_by(City.name).all(), "categories": CauseCategory.query.filter_by(is_active=True).order_by(CauseCategory.name).all(), "activity_types": ActivityType.query.filter_by(is_active=True).order_by(ActivityType.name).all(), "partners": Partner.query.filter_by(is_active=True).order_by(Partner.name).all(), } @bp.route("/") @roles_required(*ADMIN_ROLES) def dashboard(): return render_template( "admin/dashboard.html", activity_count=Activity.query.count(), published_activity_count=Activity.query.filter_by(status=ActivityStatus.PUBLISHED).count(), cause_count=DonationCause.query.count(), published_cause_count=DonationCause.query.filter_by(status=CauseStatus.PUBLISHED).count(), registration_count=ActivityRegistration.query.count(), ) @bp.route("/activities") @roles_required(*ADMIN_ROLES) def activities(): rows = Activity.query.order_by(Activity.created_at.desc()).all() return render_template("admin/activities/list.html", activities=rows) @bp.route("/activities/new", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def activity_new(): if request.method == "POST": try: start = _parse_dt(request.form.get("start_datetime")) end = _parse_dt(request.form.get("end_datetime")) if not start or not end or end <= start: raise ValueError("End date/time must be after the start date/time.") activity = Activity( activity_code=_next_code("ACT", Activity, "activity_code"), title=request.form.get("title", "").strip(), short_impact_statement=request.form.get("short_impact_statement", "").strip(), description=request.form.get("description", "").strip(), what_you_will_do=request.form.get("what_you_will_do", "").strip(), what_to_carry=request.form.get("what_to_carry", "").strip(), eligibility_requirements=request.form.get("eligibility_requirements", "").strip(), safety_instructions=request.form.get("safety_instructions", "").strip(), cause_category_id=int(request.form["cause_category_id"]), activity_type_id=int(request.form["activity_type_id"]) if request.form.get("activity_type_id") else None, partner_id=int(request.form["partner_id"]) if request.form.get("partner_id") else None, city_id=int(request.form["city_id"]), meeting_point=request.form.get("meeting_point", "").strip(), start_datetime=start, end_datetime=end, registration_open_at=_parse_dt(request.form.get("registration_open_at")), registration_close_at=_parse_dt(request.form.get("registration_close_at")), capacity=max(0, int(request.form.get("capacity") or 0)), manager_approval_required=bool(request.form.get("manager_approval_required")), working_hours_activity=bool(request.form.get("working_hours_activity")), cancellation_cutoff_hours=max(0, int(request.form.get("cancellation_cutoff_hours") or 24)), status=ActivityStatus.DRAFT, created_by_id=current_user.id, ) if not activity.title: raise ValueError("Activity title is required.") db.session.add(activity) db.session.flush() image_path = request.form.get("image_path", "").strip() if image_path: db.session.add(ActivityImage( activity_id=activity.id, file_path=image_path, alt_text=activity.title, is_primary=True, )) db.session.commit() flash("Activity created as Draft.", "success") return redirect(url_for("admin.activity_edit", activity_id=activity.id)) except (ValueError, KeyError) as exc: db.session.rollback() flash(str(exc), "danger") return render_template("admin/activities/form.html", **_activity_form_context()) @bp.route("/activities//edit", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def activity_edit(activity_id): activity = db.session.get(Activity, activity_id) if not activity: abort(404) if request.method == "POST": try: start = _parse_dt(request.form.get("start_datetime")) end = _parse_dt(request.form.get("end_datetime")) if not start or not end or end <= start: raise ValueError("End date/time must be after the start date/time.") activity.title = request.form.get("title", "").strip() activity.short_impact_statement = request.form.get("short_impact_statement", "").strip() activity.description = request.form.get("description", "").strip() activity.what_you_will_do = request.form.get("what_you_will_do", "").strip() activity.what_to_carry = request.form.get("what_to_carry", "").strip() activity.eligibility_requirements = request.form.get("eligibility_requirements", "").strip() activity.safety_instructions = request.form.get("safety_instructions", "").strip() activity.cause_category_id = int(request.form["cause_category_id"]) activity.activity_type_id = int(request.form["activity_type_id"]) if request.form.get("activity_type_id") else None activity.partner_id = int(request.form["partner_id"]) if request.form.get("partner_id") else None activity.city_id = int(request.form["city_id"]) activity.meeting_point = request.form.get("meeting_point", "").strip() activity.start_datetime = start activity.end_datetime = end activity.registration_open_at = _parse_dt(request.form.get("registration_open_at")) activity.registration_close_at = _parse_dt(request.form.get("registration_close_at")) activity.capacity = max(0, int(request.form.get("capacity") or 0)) activity.manager_approval_required = bool(request.form.get("manager_approval_required")) activity.working_hours_activity = bool(request.form.get("working_hours_activity")) activity.cancellation_cutoff_hours = max(0, int(request.form.get("cancellation_cutoff_hours") or 24)) if not activity.title: raise ValueError("Activity title is required.") image_path = request.form.get("image_path", "").strip() primary = next((image for image in activity.images if image.is_primary), None) if image_path: if primary: primary.file_path = image_path primary.alt_text = activity.title else: db.session.add(ActivityImage(activity_id=activity.id, file_path=image_path, alt_text=activity.title, is_primary=True)) elif primary: db.session.delete(primary) db.session.commit() flash("Activity updated.", "success") return redirect(url_for("admin.activity_edit", activity_id=activity.id)) except (ValueError, KeyError) as exc: db.session.rollback() flash(str(exc), "danger") return render_template("admin/activities/form.html", **_activity_form_context(activity)) @bp.route("/activities//transition/", methods=["POST"]) @roles_required(*ADMIN_ROLES) def activity_transition(activity_id, new_status): activity = db.session.get(Activity, activity_id) if not activity: abort(404) if not can_transition_activity(activity.status, new_status): flash(f"Cannot move activity from {activity.status} to {new_status}.", "danger") return redirect(url_for("admin.activity_edit", activity_id=activity.id)) activity.status = new_status now = datetime.utcnow() if new_status == ActivityStatus.APPROVED: activity.approved_by_id = current_user.id if new_status == ActivityStatus.PUBLISHED: activity.published_at = now if new_status == ActivityStatus.COMPLETED: activity.completed_at = now db.session.commit() flash(f"Activity moved to {new_status.replace('_', ' ').title()}.", "success") return redirect(url_for("admin.activity_edit", activity_id=activity.id)) @bp.route("/activities//attendance", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def activity_attendance(activity_id): activity = db.session.get(Activity, activity_id) if not activity: abort(404) registrations = ActivityRegistration.query.filter_by(activity_id=activity.id).order_by(ActivityRegistration.created_at).all() if request.method == "POST": reg_id = int(request.form["registration_id"]) registration = db.session.get(ActivityRegistration, reg_id) if not registration or registration.activity_id != activity.id: abort(404) attendance_status = request.form.get("status", "PRESENT") try: hours = Decimal(request.form.get("volunteer_hours") or "0") except InvalidOperation: hours = Decimal("0") hours = max(Decimal("0"), hours) attendance = registration.attendance if not attendance: attendance = ActivityAttendance(registration_id=registration.id, status=attendance_status) db.session.add(attendance) attendance.status = attendance_status attendance.volunteer_hours = hours attendance.confirmed_by_user_id = current_user.id attendance.confirmed_at = datetime.utcnow() attendance.remarks = request.form.get("remarks", "").strip() registration.volunteer_hours = hours if attendance_status == "PRESENT": registration.status = RegistrationStatus.COMPLETED _credit_activity_hearts(registration, hours) NotificationService.create_in_app( registration.employee_id, "ACTIVITY_COMPLETED", "Thank you for your Seva", f"Your attendance for {activity.title} is confirmed with {hours} volunteer hours. Your impact has been updated.", url_for("impact.dashboard"), ) elif attendance_status == "NO_SHOW": registration.status = RegistrationStatus.NO_SHOW NotificationService.create_in_app( registration.employee_id, "ACTIVITY_NO_SHOW", "Attendance updated", f"You were marked as a no-show for {activity.title}. If this is incorrect, please contact the Seva Setu team.", url_for("employee.my_seva"), ) db.session.commit() flash("Attendance updated.", "success") return redirect(url_for("admin.activity_attendance", activity_id=activity.id)) return render_template("admin/activities/attendance.html", activity=activity, registrations=registrations) def _credit_activity_hearts(registration, hours): rules = HeartRule.query.filter_by(is_active=True).all() for rule in rules: if rule.action_type == "VOLUNTEER_HOUR": hearts = int(Decimal(str(hours)) * rule.hearts / (rule.unit_value or 1)) elif rule.action_type == "ACTIVITY_COMPLETE": hearts = int(rule.hearts) else: continue existing = HeartTransaction.query.filter_by( employee_id=registration.employee_id, source_type="ACTIVITY_REGISTRATION", source_id=registration.id, rule_id=rule.id, ).first() if existing: existing.hearts = hearts else: db.session.add(HeartTransaction( employee_id=registration.employee_id, rule_id=rule.id, source_type="ACTIVITY_REGISTRATION", source_id=registration.id, hearts=hearts, description=f"{registration.activity.title} - {rule.action_type.replace('_', ' ').title()}", )) @bp.route("/causes") @roles_required(*ADMIN_ROLES) def causes(): rows = DonationCause.query.order_by(DonationCause.created_at.desc()).all() return render_template("admin/causes/list.html", causes=rows) @bp.route("/causes/new", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def cause_new(): return _cause_form(None) @bp.route("/causes//edit", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def cause_edit(cause_id): cause = db.session.get(DonationCause, cause_id) if not cause: abort(404) return _cause_form(cause) def _cause_form(cause): if request.method == "POST": try: if cause is None: cause = DonationCause( cause_code=_next_code("CAUSE", DonationCause, "cause_code"), created_by_id=current_user.id, status=CauseStatus.DRAFT, ) db.session.add(cause) cause.name = request.form.get("name", "").strip() cause.cause_category_id = int(request.form["cause_category_id"]) cause.partner_id = int(request.form["partner_id"]) if request.form.get("partner_id") else None cause.problem_statement = request.form.get("problem_statement", "").strip() cause.contribution_enables = request.form.get("contribution_enables", "").strip() target = request.form.get("fundraising_target", "").strip() cause.fundraising_target = Decimal(target) if target else None cause.start_date = _parse_date(request.form.get("start_date")) cause.end_date = _parse_date(request.form.get("end_date")) cause.receipt_configuration = request.form.get("receipt_configuration", "").strip() if not cause.name: raise ValueError("Cause name is required.") db.session.flush() image_path = request.form.get("image_path", "").strip() image = DonationCauseImage.query.filter_by(donation_cause_id=cause.id).order_by(DonationCauseImage.sort_order).first() if image_path: if image: image.file_path = image_path image.alt_text = cause.name else: db.session.add(DonationCauseImage(donation_cause_id=cause.id, file_path=image_path, alt_text=cause.name)) elif image: db.session.delete(image) DonationSuggestedAmount.query.filter_by(donation_cause_id=cause.id).delete() for index in range(1, 4): amount = request.form.get(f"amount_{index}", "").strip() if amount: db.session.add(DonationSuggestedAmount( donation_cause_id=cause.id, amount=Decimal(amount), impact_text=request.form.get(f"impact_{index}", "").strip(), sort_order=index, )) db.session.commit() flash("Cause saved.", "success") return redirect(url_for("admin.cause_edit", cause_id=cause.id)) except (ValueError, KeyError, InvalidOperation) as exc: db.session.rollback() flash(str(exc), "danger") return render_template( "admin/causes/form.html", cause=cause, categories=CauseCategory.query.filter_by(is_active=True).order_by(CauseCategory.name).all(), partners=Partner.query.filter_by(is_active=True).order_by(Partner.name).all(), ) @bp.route("/causes//transition/", methods=["POST"]) @roles_required(*ADMIN_ROLES) def cause_transition(cause_id, new_status): cause = db.session.get(DonationCause, cause_id) if not cause: abort(404) allowed = { CauseStatus.DRAFT: {CauseStatus.UNDER_REVIEW}, CauseStatus.UNDER_REVIEW: {CauseStatus.DRAFT, CauseStatus.APPROVED}, CauseStatus.APPROVED: {CauseStatus.PUBLISHED}, CauseStatus.PUBLISHED: {CauseStatus.CLOSED}, CauseStatus.CLOSED: {CauseStatus.ARCHIVED}, CauseStatus.ARCHIVED: set(), } if new_status not in allowed.get(cause.status, set()): flash(f"Cannot move cause from {cause.status} to {new_status}.", "danger") else: cause.status = new_status db.session.commit() flash(f"Cause moved to {new_status.replace('_', ' ').title()}.", "success") return redirect(url_for("admin.cause_edit", cause_id=cause.id)) # ----------------------------------------------------------------------------- # Operational masters # ----------------------------------------------------------------------------- @bp.route("/partners", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def partners(): if request.method == "POST": name = request.form.get("name", "").strip() if not name: flash("Partner name is required.", "danger") else: partner = Partner( name=name, partner_type=request.form.get("partner_type", "NGO").strip() or "NGO", registration_number=request.form.get("registration_number", "").strip() or None, contact_name=request.form.get("contact_name", "").strip() or None, contact_email=request.form.get("contact_email", "").strip() or None, contact_phone=request.form.get("contact_phone", "").strip() or None, website=request.form.get("website", "").strip() or None, is_active=True, ) db.session.add(partner) db.session.commit() flash("Partner added.", "success") return redirect(url_for("admin.partners")) rows = Partner.query.order_by(Partner.name).all() return render_template("admin/masters/partners.html", partners=rows) @bp.route("/partners//toggle", methods=["POST"]) @roles_required(*ADMIN_ROLES) def partner_toggle(partner_id): partner = db.session.get(Partner, partner_id) if not partner: abort(404) partner.is_active = not partner.is_active db.session.commit() flash("Partner status updated.", "success") return redirect(url_for("admin.partners")) @bp.route("/locations", methods=["GET", "POST"]) @roles_required(*ADMIN_ROLES) def locations(): if request.method == "POST": name = request.form.get("name", "").strip() city_id = request.form.get("city_id", type=int) if not name or not city_id: flash("Location name and city are required.", "danger") else: location = Location( city_id=city_id, name=name, address=request.form.get("address", "").strip() or None, is_active=True, ) db.session.add(location) db.session.commit() flash("Location added.", "success") return redirect(url_for("admin.locations")) rows = Location.query.order_by(Location.name).all() cities = City.query.filter_by(is_active=True).order_by(City.name).all() return render_template("admin/masters/locations.html", locations=rows, cities=cities) @bp.route("/locations//toggle", methods=["POST"]) @roles_required(*ADMIN_ROLES) def location_toggle(location_id): location = db.session.get(Location, location_id) if not location: abort(404) location.is_active = not location.is_active db.session.commit() flash("Location status updated.", "success") return redirect(url_for("admin.locations"))