Source code for intranet.apps.eighth.views.admin.blocks

import datetime
import logging
import pickle
import re

from cacheops import invalidate_model
from django import http
from django.conf import settings
from django.contrib import messages
from django.http import HttpResponse
from django.shortcuts import redirect, render
from django.urls import reverse

from ....auth.decorators import eighth_admin_required
from ....users.models import Group
from ...forms.admin.blocks import BlockForm, QuickBlockForm
from ...models import EighthActivity, EighthBlock, EighthScheduledActivity, EighthSignup
from ...tasks import eighth_admin_assign_hybrid_sticky_blocks
from ..attendance import generate_roster_pdf

logger = logging.getLogger(__name__)


[docs]@eighth_admin_required def add_block_view(request): if request.method == "POST" and "custom_block" in request.POST: form = QuickBlockForm(request.POST) if form.is_valid(): form.save() messages.success(request, "Successfully added block.") return redirect("eighth_admin_dashboard") else: messages.error(request, "Error adding block.") request.session["add_block_form"] = pickle.dumps(form) date = None show_letters = None signup_time = request.POST.get("signup_time") if "signup_time" in request.POST else datetime.time(12, 40) if "date" in request.GET: date = request.GET.get("date") if "date" in request.POST: date = request.POST.get("date") title_suffix = "" if date: date_format = re.compile(r"([0-9]{2})\/([0-9]{2})\/([0-9]{4})") fmtdate = date_format.sub(r"\3-\1-\2", date) title_suffix = f" - {fmtdate}" show_letters = True if "modify_blocks" in request.POST: letters = request.POST.getlist("blocks") current_letters = [] blocks_day = EighthBlock.objects.filter(date=fmtdate) for day in blocks_day: current_letters.append(day.block_letter) for ltr in letters: if not ltr: continue if ltr not in current_letters: EighthBlock.objects.create(date=fmtdate, block_letter=ltr, signup_time=signup_time) messages.success( request, f"Successfully added {ltr} Block on {fmtdate} with signups shown to students as closing at {signup_time}" ) for ltr in current_letters: if not ltr: continue if ltr not in letters: EighthBlock.objects.get(date=fmtdate, block_letter=ltr).delete() messages.success(request, f"Successfully removed {ltr} Block on {fmtdate}") else: blk = EighthBlock.objects.get(date=fmtdate, block_letter=ltr) blk.signup_time = signup_time blk.save() messages.success(request, f"Successfully changed the signup time of {ltr} Block on {fmtdate}") invalidate_model(EighthBlock) if request.POST.get("assign_withdrawn", "off") == "on": try: grp = Group.objects.get(name="Withdrawn from TJ") act = EighthActivity.objects.get(name="Z - Withdrawn from TJ") for block in EighthBlock.objects.filter(date=fmtdate): sch_act = EighthScheduledActivity.objects.update_or_create( block=block, activity=act, defaults={"attendance_taken": True, "administrative": True, "sticky": True, "capacity": 100, "restricted": True}, )[0] for u in grp.user_set.all(): sch_act.add_user(u, request=None, force=True) messages.success(request, "Successfully signed withdrawn students up for the withdrawn activity.") except (Group.DoesNotExist, EighthActivity.DoesNotExist): messages.error( request, 'Unable to assign withdrawn students; either the "Withdrawn from TJ" group or "Z - Withdrawn from TJ" activity does not exist.', ) ####### if settings.ENABLE_HYBRID_EIGHTH: if request.POST.get("assign_hybrid", "off") == "on": blocks = EighthBlock.objects.filter(date=fmtdate) block_names = {b.block_letter[4:] for b in blocks if "P1" in b.block_letter or "P2" in b.block_letter or "Virt" in b.block_letter} if "Virt" in block_names and ("P1" in block_names or "P2" in block_names): eighth_admin_assign_hybrid_sticky_blocks.delay(fmtdate=fmtdate) messages.success(request, "Assigned groups successfully.") elif "Virt" in block_names and len(block_names) == 1: # everyone is virtual messages.warning(request, "Only virtual blocks have been created, so not assigning any groups.") else: # something is wrong messages.error(request, "Created blocks successfully, but couldn't assign groups. Please do this manually.") ####### letters = [] visible_blocks = ["A", "B", "C", "D", "E", "F", "G", "H"] ####### if settings.ENABLE_HYBRID_EIGHTH: visible_blocks = ["A - Virt", "A - P1", "A - P2", "B - Virt", "B - P1", "B - P2"] ####### if show_letters: onday = EighthBlock.objects.filter(date=fmtdate) if onday: signup_time = onday[0].signup_time for ltr in visible_blocks: exists = onday.filter(block_letter=ltr) letters.append({"name": ltr, "exists": exists}) for blk in onday: if blk.block_letter not in visible_blocks: visible_blocks.append(blk.block_letter) letters.append({"name": blk.block_letter, "exists": True}) context = { "admin_page_title": f"Add or Remove Blocks{title_suffix}", "date": date, "letters": letters, "show_letters": show_letters, "signup_time": str(signup_time)[:5], "add_block_form": QuickBlockForm, } ####### if settings.ENABLE_HYBRID_EIGHTH: context.update({"hybrid": True}) ####### return render(request, "eighth/admin/add_block.html", context)
[docs]def perform_hybrid_block_signup(fmtdate, celery_logger): blocks = EighthBlock.objects.filter(date=fmtdate) block_names = {b.block_letter[4:] for b in blocks if "P1" in b.block_letter or "P2" in b.block_letter or "Virt" in b.block_letter} sticky_act = EighthActivity.objects.get_or_create(name="z - Hybrid Sticky", sticky=True, administrative=True)[0] failed_users = set() if "P1" in block_names and "Virt" in block_names and "P2" not in block_names: # sticky `virtual` and `in-person (l-z)` for `* - P1` p1_blocks = {b for b in blocks if "P1" in b.block_letter} for b in p1_blocks: sch_act = EighthScheduledActivity.objects.create(block=b, activity=sticky_act, attendance_taken=True) for g in [Group.objects.get(name="virtual"), Group.objects.get(name="in-person (l-z)")]: for user in g.user_set.all(): try: sch_act.add_user(user, request=None, force=True, no_after_deadline=True) except Exception: failed_users.add(user) # sticky `in-person (a-k)` and `in-person` for `* - Virt` virtual_blocks = {b for b in blocks if "Virt" in b.block_letter} for b in virtual_blocks: sch_act = EighthScheduledActivity.objects.create(block=b, activity=sticky_act, attendance_taken=True) for g in [Group.objects.get(name="in-person (a-k)"), Group.objects.get(name="in-person")]: for user in g.user_set.all(): try: sch_act.add_user(user, request=None, force=True, no_after_deadline=True) except Exception: failed_users.add(user) if failed_users: # this shouldn't happen because only students should be in the groups, but we don't want to 500 celery_logger.debug( "Some users could not be stickied. Please handle them manually and let the Ion devs know: {}".format( ", ".join([f"{u.first_name} {u.last_name}" for u in failed_users]) ) ) elif "P2" in block_names and "Virt" in block_names and "P1" not in block_names: # sticky `virtual` and `in-person (a-k)` for `* - P2` p2_blocks = {b for b in blocks if "P2" in b.block_letter} for b in p2_blocks: sch_act = EighthScheduledActivity.objects.create(block=b, activity=sticky_act, attendance_taken=True) for g in [Group.objects.get(name="virtual"), Group.objects.get(name="in-person (a-k)")]: for user in g.user_set.all(): try: sch_act.add_user(user, request=None, force=True, no_after_deadline=True) except Exception: failed_users.add(user) # sticky `in-person (l-z)` and `in-person` for `* - Virt` virtual_blocks = {b for b in blocks if "Virt" in b.block_letter} for b in virtual_blocks: sch_act = EighthScheduledActivity.objects.create(block=b, activity=sticky_act, attendance_taken=True) for g in [Group.objects.get(name="in-person (l-z)"), Group.objects.get(name="in-person")]: for user in g.user_set.all(): try: sch_act.add_user(user, request=None, force=True, no_after_deadline=True) except Exception: failed_users.add(user) if failed_users: # this shouldn't happen because only students should be in the groups, but we don't want to 500 celery_logger.debug( "Some users could not be stickied. Please handle them manually and let the Ion devs know: {}".format( ", ".join([f"{u.first_name} {u.last_name}" for u in failed_users]) ) )
[docs]@eighth_admin_required def edit_block_view(request, block_id): try: block = EighthBlock.objects.get(id=block_id) except EighthBlock.DoesNotExist as e: raise http.Http404 from e if request.method == "POST": form = BlockForm(request.POST, instance=block) if form.is_valid(): form.save() invalidate_model(EighthBlock) messages.success(request, "Successfully edited block.") ####### if settings.ENABLE_HYBRID_EIGHTH: messages.warning( request, """Changing the block name here did not change the groups that were assigned to it. If you need to do this, please delete this block and create the correct one.""", ) ####### return redirect("eighth_admin_dashboard") else: messages.error(request, "Error adding block.") else: form = BlockForm(instance=block) context = { "form": form, "delete_url": reverse("eighth_admin_delete_block", args=[block_id]), "admin_page_title": "Edit Block", "block_id": block_id, } return render(request, "eighth/admin/edit_form.html", context)
[docs]@eighth_admin_required def copy_block_view(request, block_id): try: block = EighthBlock.objects.get(id=block_id) except EighthBlock.DoesNotExist as e: raise http.Http404 from e if request.method == "POST": copy_signups = request.POST.get("signups", False) new_block_id = request.POST.get("block", None) if new_block_id and not new_block_id == block_id: new_block = None try: new_block = EighthBlock.objects.get(id=new_block_id) except EighthBlock.DoesNotExist: messages.error(request, "That block does not exist!") if new_block: # Delete previous EighthScheduledActivities and EighthSignups EighthScheduledActivity.objects.filter(block=block).delete() EighthSignup.objects.filter(scheduled_activity__block=block).delete() for schact in EighthScheduledActivity.objects.filter(block=new_block, cancelled=False).prefetch_related("rooms", "sponsors"): new_schact = EighthScheduledActivity.objects.create( block=block, activity=schact.activity, both_blocks=schact.both_blocks, special=schact.special ) new_schact.sponsors.set(schact.sponsors.all()) new_schact.rooms.set(schact.rooms.all()) new_schact.save() if copy_signups: EighthSignup.objects.bulk_create( [EighthSignup(user=s.user, scheduled_activity=new_schact) for s in EighthSignup.objects.filter(scheduled_activity=schact)] ) context = { "new_activities": EighthScheduledActivity.objects.filter(block=block).count(), "new_signups": EighthSignup.objects.filter(scheduled_activity__block=block).count(), "success": True, "admin_page_title": f"Finished Copy Block - {block.formatted_date} ({block.block_letter})", "block_id": block_id, } return render(request, "eighth/admin/copy_form.html", context) else: messages.error(request, "Please enter a valid block to copy activities from.") context = { "existing_activities": EighthScheduledActivity.objects.filter(block=block).count(), "existing_signups": EighthSignup.objects.filter(scheduled_activity__block=block).count(), "blocks": EighthBlock.objects.all().order_by("date"), "admin_page_title": f"Copy Block - {block.formatted_date} ({block.block_letter})", "to_block": f"{block.id}: {block.formatted_date} ({block.block_letter})", "block_id": block_id, "locked": block.locked, "success": False, } return render(request, "eighth/admin/copy_form.html", context)
[docs]@eighth_admin_required def delete_block_view(request, block_id): try: block = EighthBlock.objects.get(id=block_id) except EighthBlock.DoesNotExist as e: raise http.Http404 from e if request.method == "POST": block.delete() invalidate_model(EighthBlock) messages.success(request, "Successfully deleted block.") return redirect("eighth_admin_dashboard") else: context = { "admin_page_title": "Delete Block", "item_name": str(block), "help_text": "Deleting this block will remove all records " "of it related to eighth period.", } return render(request, "eighth/admin/delete_form.html", context)