9047122193

Committed 11 May 2024 11:26PM UTC coverage: 79.447% (-0.3%) from 79.718%

Build # 9047122193

Build Type

push

github

Committed by

alanzhu0

Commit Message

ci(scripts): make ci.yml easier to read

Run Details

3079 of 5838 branches covered (52.74%)

15790 of 19875 relevant lines covered (79.45%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

31.25

/intranet/apps/eighth/tasks.py

import calendar
import datetime
from typing import Collection

from celery import shared_task
from celery.utils.log import get_task_logger

from django.conf import settings
from django.contrib.auth import get_user_model
from django.core.mail import EmailMessage
from django.utils import timezone

from ...utils.helpers import join_nicely
from ..groups.models import Group
from ..notifications.emails import email_send
from .models import EighthActivity, EighthRoom, EighthScheduledActivity

logger = get_task_logger(__name__)


@shared_task
def room_changed_single_email(
    sched_act: EighthScheduledActivity, old_rooms: Collection[EighthRoom], new_rooms: Collection[EighthRoom]  # pylint: disable=unsubscriptable-object
):  # pylint: disable=unsubscriptable-object
    """Notifies all the users signed up for the given EighthScheduledActivity that it is changing rooms.

    Args:
        sched_act: The activity that has changed rooms.
        old_rooms: The list of rooms that the activity used to be in.
        new_rooms: The new list of rooms that the activity is in.

    """
    date_str = sched_act.block.date.strftime("%A, %B %-d")

    emails = [signup.user.notification_email for signup in sched_act.eighthsignup_set.filter(user__receive_eighth_emails=True).distinct()]

    if not emails:
        return

    base_url = "https://ion.tjhsst.edu"

    data = {
        "sched_act": sched_act,
        "date_str": date_str,
        "base_url": base_url,
        "num_old_rooms": len(old_rooms),
        "num_new_rooms": len(new_rooms),
        "old_rooms_str": join_nicely(room.formatted_name for room in old_rooms),
        "new_rooms_str": join_nicely(room.formatted_name for room in new_rooms),
    }

    logger.debug(
        "Scheduled activity %d changed from rooms %s to %s; emailing %d of %d signed up users",
        sched_act.id,
        data["old_rooms_str"],
        data["new_rooms_str"],
        len(emails),
        sched_act.eighthsignup_set.count(),
    )

    email_send(
        "eighth/emails/room_changed_single.txt",
        "eighth/emails/room_changed_single.html",
        data,
        "Room change for {} on {}".format(sched_act.activity.name, date_str),
        emails,
        bcc=True,
    )


@shared_task
def transferred_activity_email(
    dest_act: EighthScheduledActivity, source_act: EighthScheduledActivity, duplicate_students  # pylint: disable=unsubscriptable-object
):  # pylint: disable=unsubscriptable-object
    """Notifies all the users already signed up for an EighthScheduledActivity that they have been transferred into a new activity.

    Args:
        dest_act: The activity that the students were transferred into.
        dest_act: The activity that the students were transferred from.
        duplicate_students: The students that were transferred into a new activity.

    """
    date_str = dest_act.block.date.strftime("%A, %B %-d")

    # Emails are sent regardless of notification settings so that students have the chance to ensure that they are signed up for the correct place.
    emails = [u.notification_email for u in duplicate_students]

    if not emails:
        return

    base_url = "https://ion.tjhsst.edu"

    data = {
        "dest_act": dest_act,
        "source_act": source_act,
        "date_str": date_str,
        "base_url": base_url,
    }

    logger.debug(
        "Transferring students from %s to %s resulted in duplicate signups being deleted; emailing the %d affected users",
        source_act,
        dest_act.id,
        len(emails),
    )

    email_send(
        "eighth/emails/transferred_activity.txt",
        "eighth/emails/transferred_activity.html",
        data,
        "8th Period Transfer to {} on {}".format(dest_act.activity.name, date_str),
        emails,
        bcc=True,
    )


@shared_task
def room_changed_activity_email(
    act: EighthActivity, old_rooms: Collection[EighthRoom], new_rooms: Collection[EighthRoom]  # pylint: disable=unsubscriptable-object
):  # pylint: disable=unsubscriptable-object
    """Notifies all the users signed up for the given EighthActivity on the blocks for which the room
    list is not overriden that it is changing rooms.

    Args:
        act: The activity that has changed rooms.
        old_rooms: The list of rooms that the activity used to be in.
        new_rooms: The new list of rooms that the activity is in.

    """
    all_sched_acts = act.eighthscheduledactivity_set.filter(block__date__gte=timezone.localdate())
    sched_acts = all_sched_acts.filter(rooms=None)
    users = (
        get_user_model()
        .objects.filter(
            receive_eighth_emails=True,
            eighthsignup__scheduled_activity__activity=act,
            eighthsignup__scheduled_activity__block__date__gte=timezone.localdate(),
        )
        .distinct()
    )

    base_url = "https://ion.tjhsst.edu"

    data = {
        "act": act,
        "base_url": base_url,
        "num_old_rooms": len(old_rooms),
        "num_new_rooms": len(new_rooms),
        "old_rooms_str": join_nicely(room.formatted_name for room in old_rooms),
        "new_rooms_str": join_nicely(room.formatted_name for room in new_rooms),
    }

    logger.debug(
        "Activity %d changed from rooms %s to %s; emailing %d signed up users", act.id, data["old_rooms_str"], data["new_rooms_str"], len(users)
    )

    for user in users:
        data["date_strs"] = [
            "{}, {} block".format(sa.block.date.strftime("%A, %B %-d"), sa.block.block_letter)
            for sa in sched_acts.filter(eighthsignup_set__user=user)
        ]

        email_send(
            "eighth/emails/room_changed_activity.txt",
            "eighth/emails/room_changed_activity.html",
            data,
            "Room changes for {}".format(act.name),
            [user.notification_email],
            bcc=True,
        )


@shared_task
def eighth_admin_assign_hybrid_sticky_blocks(fmtdate: str) -> None:
    """Sign all users up for z - Hybrid Sticky according to their status.

    Args:
        fmtdate: The date where users should be signed up for blocks.

    """
    # Circular dependency
    from .views.admin.blocks import perform_hybrid_block_signup  # pylint: disable=import-outside-toplevel

    perform_hybrid_block_signup(fmtdate, logger)


@shared_task
def eighth_admin_signup_group_task(*, user_id: int, group_id: int, schact_id: int, skip_users: set) -> None:
    """Sign all users in a specific group up for a specific scheduled activity
    (in the background), sending an email to the user who requested the operation when
    it is done.

    Args:
        user_id: The ID of the user who requested that this operation be performed.
        group_id: The ID of the group to sign up for the activity.
        schact_id:.The ID of the EighthScheduledActivity to add the group members to.
        skip_users: A list of users that should not be signed up for the activity,
            usually because they are stickied into another activity.
    """
    # Circular dependency
    from .views.admin.groups import eighth_admin_perform_group_signup  # pylint: disable=import-outside-toplevel

    user = get_user_model().objects.get(id=user_id)

    data = {
        "group": Group.objects.get(id=group_id),
        "scheduled_activity": EighthScheduledActivity.objects.get(id=schact_id),
        "help_email": settings.FEEDBACK_EMAIL,
        "base_url": "https://ion.tjhsst.edu",
    }

    try:
        eighth_admin_perform_group_signup(group_id=group_id, schact_id=schact_id, request=None, skip_users=skip_users)
    except Exception:
        email_send(
            "eighth/emails/group_signup_error.txt",
            "eighth/emails/group_signup_error.html",
            data,
            "Error during group signup",
            [user.notification_email],
            bcc=False,
        )

        raise  # Send to Sentry
    else:
        email_send(
            "eighth/emails/group_signup_complete.txt",
            "eighth/emails/group_signup_complete.html",
            data,
            "Group signup complete",
            [user.notification_email],
            bcc=False,
        )


@shared_task
def eighth_admin_signup_group_task_hybrid(*, user_id: int, group_id: int, schact_virtual_id: int, schact_person_id: int) -> None:
    """Sign all users in a specific group up for a specific scheduled activity
    (in the background), sending an email to the user who requested the operation when
    it is done.

    Args:
        user_id: The ID of the user who requested that this operation be performed.
        group_id: The ID of the group to sign up for the activity.
        schact_virtual_id: The ID of the EighthScheduledActivity to add the virtual group members to.
        schact_person_id: The ID of the EighthScheduledActivity to add the in-person group members to.

    """
    # Circular dependency
    from .views.admin.hybrid import eighth_admin_perform_group_signup  # pylint: disable=import-outside-toplevel

    user = get_user_model().objects.get(id=user_id)

    data = {
        "group": Group.objects.get(id=group_id),
        "hybrid": True,
        "scheduled_activity_virtual": EighthScheduledActivity.objects.get(id=schact_virtual_id),
        "scheduled_activity_person": EighthScheduledActivity.objects.get(id=schact_person_id),
        "help_email": settings.FEEDBACK_EMAIL,
        "base_url": "https://ion.tjhsst.edu",
    }

    try:
        eighth_admin_perform_group_signup(group_id=group_id, schact_virtual_id=schact_virtual_id, schact_person_id=schact_person_id, request=None)
    except Exception:
        email_send(
            "eighth/emails/group_signup_error.txt",
            "eighth/emails/group_signup_error.html",
            data,
            "Error during group signup",
            [user.notification_email],
            bcc=False,
        )

        raise  # Send to Sentry
    else:
        email_send(
            "eighth/emails/group_signup_complete.txt",
            "eighth/emails/group_signup_complete.html",
            data,
            "Group signup complete",
            [user.notification_email],
            bcc=False,
        )


@shared_task
def email_scheduled_activity_students_task(
    scheduled_activity_id: int,
    sender_id: int,
    subject: str,
    body: str,
) -> None:
    scheduled_activity = EighthScheduledActivity.objects.get(id=scheduled_activity_id)

    emails = [signup.user.notification_email for signup in scheduled_activity.eighthsignup_set.all()]

    sender = get_user_model().objects.get(id=sender_id)

    msg = EmailMessage(subject=subject, body=body, from_email=settings.EMAIL_FROM, to=[], reply_to=[sender.notification_email], bcc=emails)

    if settings.PRODUCTION or settings.FORCE_EMAIL_SEND:
        msg.send()
    else:
        logger.debug("Refusing to email in non-production environments. To force email sending, enable settings.FORCE_EMAIL_SEND.")


@shared_task
def follow_up_absence_emails():
    """Send emails to all students with uncleared absences from the past month."""
    month = datetime.datetime.now().month
    year = datetime.datetime.now().year
    if month == 1:
        month = 12
        year -= 1
    else:
        month -= 1
    first_day = datetime.date(year, month, 1)
    last_day = datetime.date(year, month, calendar.monthrange(year, month)[1])

    for student in get_user_model().objects.filter(user_type="student"):
        absences = student.absence_info().filter(scheduled_activity__date__gte=first_day, scheduled_activity__date__lte=last_day)
        num_absences = absences.count()

        if num_absences > 0:
            data = {
                "absences": absences,
                "info_link": "https://ion.tjhsst.edu/eighth/absences",
                "num_absences": num_absences,
                "base_url": "https://ion.tjhsst.edu",
            }

            email_send(
                "eighth/emails/absence_monthly.txt",
                "eighth/emails/absence_monthly.html",
                data,
                "{} Uncleared Eighth Period Absences".format(num_absences),
                [student.notification_email],
                bcc=True,
            )

1	import calendar	1✔
2	import datetime	1✔
3	from typing import Collection	1✔
4
5	from celery import shared_task	1✔
6	from celery.utils.log import get_task_logger	1✔
7
8	from django.conf import settings	1✔
9	from django.contrib.auth import get_user_model	1✔
10	from django.core.mail import EmailMessage	1✔
11	from django.utils import timezone	1✔
12
13	from ...utils.helpers import join_nicely	1✔
14	from ..groups.models import Group	1✔
15	from ..notifications.emails import email_send	1✔
16	from .models import EighthActivity, EighthRoom, EighthScheduledActivity	1✔
17
18	logger = get_task_logger(__name__)	1✔
19
20
21	@shared_task	1✔
22	def room_changed_single_email(	1!
23	sched_act: EighthScheduledActivity, old_rooms: Collection[EighthRoom], new_rooms: Collection[EighthRoom] # pylint: disable=unsubscriptable-object
24	): # pylint: disable=unsubscriptable-object
25	"""Notifies all the users signed up for the given EighthScheduledActivity that it is changing rooms.
26
27	Args:
28	sched_act: The activity that has changed rooms.
29	old_rooms: The list of rooms that the activity used to be in.
30	new_rooms: The new list of rooms that the activity is in.
31
32	"""
33	date_str = sched_act.block.date.strftime("%A, %B %-d")	×
34
35	emails = [signup.user.notification_email for signup in sched_act.eighthsignup_set.filter(user__receive_eighth_emails=True).distinct()]	×
36
37	if not emails:	×
38	return	×
39
40	base_url = "https://ion.tjhsst.edu"	×
41
42	data = {	×
43	"sched_act": sched_act,
44	"date_str": date_str,
45	"base_url": base_url,
46	"num_old_rooms": len(old_rooms),
47	"num_new_rooms": len(new_rooms),
48	"old_rooms_str": join_nicely(room.formatted_name for room in old_rooms),
49	"new_rooms_str": join_nicely(room.formatted_name for room in new_rooms),
50	}
51
52	logger.debug(	×
53	"Scheduled activity %d changed from rooms %s to %s; emailing %d of %d signed up users",
54	sched_act.id,
55	data["old_rooms_str"],
56	data["new_rooms_str"],
57	len(emails),
58	sched_act.eighthsignup_set.count(),
59	)
60
61	email_send(	×
62	"eighth/emails/room_changed_single.txt",
63	"eighth/emails/room_changed_single.html",
64	data,
65	"Room change for {} on {}".format(sched_act.activity.name, date_str),
66	emails,
67	bcc=True,
68	)
69
70
71	@shared_task	1✔
72	def transferred_activity_email(	1!
73	dest_act: EighthScheduledActivity, source_act: EighthScheduledActivity, duplicate_students # pylint: disable=unsubscriptable-object
74	): # pylint: disable=unsubscriptable-object
75	"""Notifies all the users already signed up for an EighthScheduledActivity that they have been transferred into a new activity.
76
77	Args:
78	dest_act: The activity that the students were transferred into.
79	dest_act: The activity that the students were transferred from.
80	duplicate_students: The students that were transferred into a new activity.
81
82	"""
83	date_str = dest_act.block.date.strftime("%A, %B %-d")	×
84
85	# Emails are sent regardless of notification settings so that students have the chance to ensure that they are signed up for the correct place.
86	emails = [u.notification_email for u in duplicate_students]	×
87
88	if not emails:	×
89	return	×
90
91	base_url = "https://ion.tjhsst.edu"	×
92
93	data = {	×
94	"dest_act": dest_act,
95	"source_act": source_act,
96	"date_str": date_str,
97	"base_url": base_url,
98	}
99
100	logger.debug(	×
101	"Transferring students from %s to %s resulted in duplicate signups being deleted; emailing the %d affected users",
102	source_act,
103	dest_act.id,
104	len(emails),
105	)
106
107	email_send(	×
108	"eighth/emails/transferred_activity.txt",
109	"eighth/emails/transferred_activity.html",
110	data,
111	"8th Period Transfer to {} on {}".format(dest_act.activity.name, date_str),
112	emails,
113	bcc=True,
114	)
115
116
117	@shared_task	1✔
118	def room_changed_activity_email(	1!
119	act: EighthActivity, old_rooms: Collection[EighthRoom], new_rooms: Collection[EighthRoom] # pylint: disable=unsubscriptable-object
120	): # pylint: disable=unsubscriptable-object
121	"""Notifies all the users signed up for the given EighthActivity on the blocks for which the room
122	list is not overriden that it is changing rooms.
123
124	Args:
125	act: The activity that has changed rooms.
126	old_rooms: The list of rooms that the activity used to be in.
127	new_rooms: The new list of rooms that the activity is in.
128
129	"""
130	all_sched_acts = act.eighthscheduledactivity_set.filter(block__date__gte=timezone.localdate())	×
131	sched_acts = all_sched_acts.filter(rooms=None)	×
132	users = (	×
133	get_user_model()
134	.objects.filter(
135	receive_eighth_emails=True,
136	eighthsignup__scheduled_activity__activity=act,
137	eighthsignup__scheduled_activity__block__date__gte=timezone.localdate(),
138	)
139	.distinct()
140	)
141
142	base_url = "https://ion.tjhsst.edu"	×
143
144	data = {	×
145	"act": act,
146	"base_url": base_url,
147	"num_old_rooms": len(old_rooms),
148	"num_new_rooms": len(new_rooms),
149	"old_rooms_str": join_nicely(room.formatted_name for room in old_rooms),
150	"new_rooms_str": join_nicely(room.formatted_name for room in new_rooms),
151	}
152
153	logger.debug(	×
154	"Activity %d changed from rooms %s to %s; emailing %d signed up users", act.id, data["old_rooms_str"], data["new_rooms_str"], len(users)
155	)
156
157	for user in users:	×
158	data["date_strs"] = [	×
159	"{}, {} block".format(sa.block.date.strftime("%A, %B %-d"), sa.block.block_letter)
160	for sa in sched_acts.filter(eighthsignup_set__user=user)
161	]
162
163	email_send(	×
164	"eighth/emails/room_changed_activity.txt",
165	"eighth/emails/room_changed_activity.html",
166	data,
167	"Room changes for {}".format(act.name),
168	[user.notification_email],
169	bcc=True,
170	)
171
172
173	@shared_task	1✔
174	def eighth_admin_assign_hybrid_sticky_blocks(fmtdate: str) -> None:	1!
175	"""Sign all users up for z - Hybrid Sticky according to their status.
176
177	Args:
178	fmtdate: The date where users should be signed up for blocks.
179
180	"""
181	# Circular dependency
182	from .views.admin.blocks import perform_hybrid_block_signup # pylint: disable=import-outside-toplevel	×
183
184	perform_hybrid_block_signup(fmtdate, logger)	×
185
186
187	@shared_task	1✔
188	def eighth_admin_signup_group_task(*, user_id: int, group_id: int, schact_id: int, skip_users: set) -> None:	1!
189	"""Sign all users in a specific group up for a specific scheduled activity
190	(in the background), sending an email to the user who requested the operation when
191	it is done.
192
193	Args:
194	user_id: The ID of the user who requested that this operation be performed.
195	group_id: The ID of the group to sign up for the activity.
196	schact_id:.The ID of the EighthScheduledActivity to add the group members to.
197	skip_users: A list of users that should not be signed up for the activity,
198	usually because they are stickied into another activity.
199	"""
200	# Circular dependency
201	from .views.admin.groups import eighth_admin_perform_group_signup # pylint: disable=import-outside-toplevel	×
202
203	user = get_user_model().objects.get(id=user_id)	×
204
205	data = {	×
206	"group": Group.objects.get(id=group_id),
207	"scheduled_activity": EighthScheduledActivity.objects.get(id=schact_id),
208	"help_email": settings.FEEDBACK_EMAIL,
209	"base_url": "https://ion.tjhsst.edu",
210	}
211
212	try:	×
213	eighth_admin_perform_group_signup(group_id=group_id, schact_id=schact_id, request=None, skip_users=skip_users)	×
214	except Exception:	×
215	email_send(	×
216	"eighth/emails/group_signup_error.txt",
217	"eighth/emails/group_signup_error.html",
218	data,
219	"Error during group signup",
220	[user.notification_email],
221	bcc=False,
222	)
223
224	raise # Send to Sentry	×
225	else:
226	email_send(	×
227	"eighth/emails/group_signup_complete.txt",
228	"eighth/emails/group_signup_complete.html",
229	data,
230	"Group signup complete",
231	[user.notification_email],
232	bcc=False,
233	)
234
235
236	@shared_task	1✔
237	def eighth_admin_signup_group_task_hybrid(*, user_id: int, group_id: int, schact_virtual_id: int, schact_person_id: int) -> None:	1!
238	"""Sign all users in a specific group up for a specific scheduled activity
239	(in the background), sending an email to the user who requested the operation when
240	it is done.
241
242	Args:
243	user_id: The ID of the user who requested that this operation be performed.
244	group_id: The ID of the group to sign up for the activity.
245	schact_virtual_id: The ID of the EighthScheduledActivity to add the virtual group members to.
246	schact_person_id: The ID of the EighthScheduledActivity to add the in-person group members to.
247
248	"""
249	# Circular dependency
250	from .views.admin.hybrid import eighth_admin_perform_group_signup # pylint: disable=import-outside-toplevel	×
251
252	user = get_user_model().objects.get(id=user_id)	×
253
254	data = {	×
255	"group": Group.objects.get(id=group_id),
256	"hybrid": True,
257	"scheduled_activity_virtual": EighthScheduledActivity.objects.get(id=schact_virtual_id),
258	"scheduled_activity_person": EighthScheduledActivity.objects.get(id=schact_person_id),
259	"help_email": settings.FEEDBACK_EMAIL,
260	"base_url": "https://ion.tjhsst.edu",
261	}
262
263	try:	×
264	eighth_admin_perform_group_signup(group_id=group_id, schact_virtual_id=schact_virtual_id, schact_person_id=schact_person_id, request=None)	×
265	except Exception:	×
266	email_send(	×
267	"eighth/emails/group_signup_error.txt",
268	"eighth/emails/group_signup_error.html",
269	data,
270	"Error during group signup",
271	[user.notification_email],
272	bcc=False,
273	)
274
275	raise # Send to Sentry	×
276	else:
277	email_send(	×
278	"eighth/emails/group_signup_complete.txt",
279	"eighth/emails/group_signup_complete.html",
280	data,
281	"Group signup complete",
282	[user.notification_email],
283	bcc=False,
284	)
285
286
287	@shared_task	1✔
288	def email_scheduled_activity_students_task(	1!
289	scheduled_activity_id: int,
290	sender_id: int,
291	subject: str,
292	body: str,
293	) -> None:
294	scheduled_activity = EighthScheduledActivity.objects.get(id=scheduled_activity_id)	×
295
296	emails = [signup.user.notification_email for signup in scheduled_activity.eighthsignup_set.all()]	×
297
298	sender = get_user_model().objects.get(id=sender_id)	×
299
300	msg = EmailMessage(subject=subject, body=body, from_email=settings.EMAIL_FROM, to=[], reply_to=[sender.notification_email], bcc=emails)	×
301
302	if settings.PRODUCTION or settings.FORCE_EMAIL_SEND:	×
303	msg.send()	×
304	else:
305	logger.debug("Refusing to email in non-production environments. To force email sending, enable settings.FORCE_EMAIL_SEND.")	×
306
307
308	@shared_task	1✔
309	def follow_up_absence_emails():	1!
310	"""Send emails to all students with uncleared absences from the past month."""
311	month = datetime.datetime.now().month	×
312	year = datetime.datetime.now().year	×
313	if month == 1:	×
314	month = 12	×
315	year -= 1	×
316	else:
317	month -= 1	×
318	first_day = datetime.date(year, month, 1)	×
319	last_day = datetime.date(year, month, calendar.monthrange(year, month)[1])	×
320
321	for student in get_user_model().objects.filter(user_type="student"):	×
322	absences = student.absence_info().filter(scheduled_activity__date__gte=first_day, scheduled_activity__date__lte=last_day)	×
323	num_absences = absences.count()	×
324
325	if num_absences > 0:	×
326	data = {	×
327	"absences": absences,
328	"info_link": "https://ion.tjhsst.edu/eighth/absences",
329	"num_absences": num_absences,
330	"base_url": "https://ion.tjhsst.edu",
331	}
332
333	email_send(	×
334	"eighth/emails/absence_monthly.txt",
335	"eighth/emails/absence_monthly.html",
336	data,
337	"{} Uncleared Eighth Period Absences".format(num_absences),
338	[student.notification_email],
339	bcc=True,
340	)

tjcsl / ion / 9047122193

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous