tasks

main.tasks

Task definitions for project notifications using Huey.

Classes

Functions:

daily_project_status_check()

Daily task to check project statuses and notify leads.

Source code in main/tasks.py

@db_periodic_task(crontab(hour=10, minute=0))
def daily_project_status_check() -> None:
    """Daily task to check project statuses and notify leads."""
    from .models import Project

    projects = Project.objects.filter(status__in=["Active", "Maintenance"])
    for project in projects:
        project.check_and_notify_status()

email_monthly_charges_report()

Email the HoRSE the charges report for the last month.

Source code in main/tasks.py

@db_periodic_task(crontab(day=10, hour=10, minute=0))
def email_monthly_charges_report() -> None:
    """Email the HoRSE the charges report for the last month."""
    last_month_start, last_month_name, _, _ = get_current_and_last_month()
    email_monthly_charges_report_logic(
        last_month_start.month, last_month_start.year, last_month_name
    )

email_monthly_charges_report_logic(month, year, month_name)

Logic to email the HoRSE the charges report for the last month.

Source code in main/tasks.py

def email_monthly_charges_report_logic(month: int, year: int, month_name: str) -> None:
    """Logic to email the HoRSE the charges report for the last month."""
    subject = f"Charges report for {month_name}/{year}"
    head_email = get_head_email()
    message = _template_charges_report.format(month=month_name, year=year)
    csv_attachment = create_charges_report_for_attachment(month, year)

    email_attachment(
        subject,
        head_email,
        message,
        f"charges_report_{month}-{year}.csv",
        csv_attachment,
        "text/csv",
    )

notify_funding_status()

Weekly task to notify about funding status.

Source code in main/tasks.py

@db_periodic_task(crontab(day_of_week=1, hour=9, minute=0))
def notify_funding_status() -> None:
    """Weekly task to notify about funding status."""
    notify_funding_status_logic()

notify_funding_status_logic(date=None)

Logic for notifying the lead about funding status.

Source code in main/tasks.py

def notify_funding_status_logic(
    date: datetime.date | None = None,
) -> None:
    """Logic for notifying the lead about funding status."""
    funds_ran_out_not_expired, funding_expired_budget_left = get_budget_status(
        date=date
    )

    if len(funds_ran_out_not_expired) > 0:
        for funding in funds_ran_out_not_expired:
            subject = f"[Funding Update] {funding.project.name}"
            head_email = get_head_email()
            lead = funding.project.lead
            lead_name = lead.get_full_name() if lead is not None else "Project Leader"
            lead_email = lead.email if lead is not None else ""
            activity = funding.activity if funding.activity else "Funding Activity"
            message = _template_funds_ran_out_but_not_expired.format(
                lead=lead_name,
                project_name=funding.project.name,
                activity=activity,
            )
            email_user_and_cc_head(
                subject=subject,
                message=message,
                email=lead_email,
                head_email=head_email,
            )

    if len(funding_expired_budget_left) > 0:
        for funding in funding_expired_budget_left:
            subject = f"[Funding Expired] {funding.project.name}"
            head_email = get_head_email()
            lead = funding.project.lead
            lead_name = lead.get_full_name() if lead is not None else "Project Leader"
            lead_email = lead.email if lead is not None else ""
            message = _template_funding_expired_but_has_budget.format(
                lead=lead_name,
                project_name=funding.project.name,
                funding_left=funding.funding_left,
                budget=funding.budget,
            )
            email_user_and_cc_head(
                subject=subject,
                message=message,
                email=lead_email,
                head_email=head_email,
            )

notify_left_threshold(email, lead, project_name, threshold_type, threshold, value)

Huey task wrapper that calls the core notify logic.

Source code in main/tasks.py

@task()
def notify_left_threshold(
    email: str,
    lead: str,
    project_name: str,
    threshold_type: str,
    threshold: int,
    value: float,
) -> None:
    """Huey task wrapper that calls the core notify logic."""
    notify_left_threshold_logic(
        email, lead, project_name, threshold_type, threshold, value
    )

notify_left_threshold_logic(email, lead, project_name, threshold_type, threshold, value)

Logic for notifying the lead about project status.

Source code in main/tasks.py

def notify_left_threshold_logic(
    email: str,
    lead: str,
    project_name: str,
    threshold_type: str,
    threshold: int,
    value: float,
) -> None:
    """Logic for notifying the lead about project status."""
    if threshold_type not in ("effort", "weeks"):
        raise ValueError("Invalid threshold type provided.")
    unit = "days" if threshold_type == "effort" else "weeks"
    subject = f"[Project Status Update] {project_name}"
    message = _template.format(
        project_leader=lead,
        project_name=project_name,
        threshold=threshold,
        threshold_type=threshold_type.rsplit("_")[0],
        value=value,
        unit=unit,
    )

    email_user(subject, email, message)

notify_monthly_days_used_exceeding_days_left()

Monthly task to notify project leads and HoRSE if days used exceed days left.

Source code in main/tasks.py

@db_periodic_task(crontab(day=7, hour=9, minute=30))
def notify_monthly_days_used_exceeding_days_left() -> None:
    """Monthly task to notify project leads and HoRSE if days used exceed days left."""
    notify_monthly_days_used_exceeding_days_left_logic()

notify_monthly_days_used_exceeding_days_left_logic(date=None)

Logic to notify project lead and HoRSE if total days used exceed days left.

This function checks each project to see if the days used for the project exceed the days left. If they do, it sends an email notification to the project lead and HoRSE.

Source code in main/tasks.py

def notify_monthly_days_used_exceeding_days_left_logic(
    date: datetime.datetime | None = None,
) -> None:
    """Logic to notify project lead and HoRSE if total days used exceed days left.

    This function checks each project to see if the days used for the
    project exceed the days left. If they do,
    it sends an email notification to the project lead and HoRSE.
    """
    if date is None:
        date = timezone.now()

    projects = get_projects_with_days_used_exceeding_days_left()

    for project, days_left, total_effort in projects:
        lead = project.lead
        lead_name = lead.get_full_name() if lead else "Project Leader"
        lead_email = lead.email if lead else ""

        subject = f"[Monthly Days Used Exceed Days Left] {project.name}"
        message = _template_days_used_exceeded_days_left.format(
            lead=lead_name,
            project_name=project.name,
            total_effort=total_effort,
            days_left=days_left,
        )

        head_email = get_head_email()

        email_user_and_cc_head(
            subject=subject,
            message=message,
            email=lead_email,
            head_email=head_email,
        )

notify_monthly_time_logged_logic(last_month_start, last_month_name, current_month_start, current_month_name)

Logic to notify users about their monthly time logged.

Source code in main/tasks.py

def notify_monthly_time_logged_logic(
    last_month_start: datetime.date,
    last_month_name: str,
    current_month_start: datetime.date,
    current_month_name: str,
) -> None:
    """Logic to notify users about their monthly time logged."""
    from .models import TimeEntry, User

    avg_work_days_per_month = 220 / 12  # Approximately 18.33 days per month

    time_entries = TimeEntry.objects.filter(
        start_time__gte=last_month_start, end_time__lt=current_month_start
    )

    if not time_entries.exists():
        return  # No entries to process

    users = User.objects.filter(timeentry__in=time_entries).distinct()

    for user in users:
        total_hours, project_work_summary = get_logged_hours(
            time_entries.filter(user=user)
        )

        total_days = total_hours / 7  # Assuming 7 hours/workday
        percentage = round((total_days * 100) / avg_work_days_per_month, 1)

        message = _template_time_logged.format(
            user=user.get_full_name(),
            last_month_name=last_month_name,
            project_work_summary=project_work_summary,
            percentage=percentage,
            current_month_name=current_month_name,
        )

        subject = f"Your Project Time Logged Summary for {last_month_name}"

        email_user(
            subject=subject,
            message=message,
            email=user.email,
        )

notify_monthly_time_logged_summary()

Monthly task to notify users about their time logged.

Source code in main/tasks.py

@db_periodic_task(crontab(day=3, hour=10, minute=0))
def notify_monthly_time_logged_summary() -> None:
    """Monthly task to notify users about their time logged."""
    last_month_start, last_month_name, current_month_start, current_month_name = (
        get_current_and_last_month()
    )

    notify_monthly_time_logged_logic(
        last_month_start,
        last_month_name,
        current_month_start,
        current_month_name,
    )

sync_clockify_time_entries(days_back=30, end_date=timezone.now(), page_size=1000)

Task to sync time entries from Clockify API to TimeEntry model.

Parameters:

Name	Type	Description	Default
days_back	int	Number of days to look back for time entries.	30
end_date	datetime	The end date for the time entries to fetch.	now()
page_size	int	Number of entries to fetch per API call.	1000

Returns:

Type	Description
bool	True if there is any issue, and False if all went well.

Source code in main/tasks.py

def sync_clockify_time_entries(
    days_back: int = 30,
    end_date: datetime.datetime = timezone.now(),
    page_size: int = 1000,
) -> bool:
    """Task to sync time entries from Clockify API to TimeEntry model.

    Args:
        days_back (int): Number of days to look back for time entries.
        end_date (datetime.datetime): The end date for the time entries to fetch.
        page_size (int): Number of entries to fetch per API call.

    Returns:
        True if there is any issue, and False if all went well.
    """
    issues = False
    if (
        not settings.CLOCKIFY_API_KEY
        or not settings.CLOCKIFY_WORKSPACE_ID
        or not settings.CLOCKIFY_CUTOVER_DATE
    ):
        logger.warning("Clockify API information not found in environment variables")
        return False

    api = ClockifyAPI(settings.CLOCKIFY_API_KEY, settings.CLOCKIFY_WORKSPACE_ID)
    start_date = end_date - datetime.timedelta(days=days_back)

    # The end date can be, at most, the day before the cutover
    end_date = (
        end_date
        if end_date < settings.CLOCKIFY_CUTOVER_DATE
        else settings.CLOCKIFY_CUTOVER_DATE - datetime.timedelta(days=1)
    )
    if end_date < start_date:
        return False

    projects = Project.objects.filter(status__in=["Active", "Maintenance"]).exclude(
        clockify_id=""
    )
    for project in projects:
        logger.info(f"Processing project ID: {project.clockify_id}")
        payload = {
            "dateRangeStart": start_date.strftime("%Y-%m-%dT00:00:00.000Z"),
            "dateRangeEnd": end_date.strftime("%Y-%m-%dT23:59:59.000Z"),
            "detailedFilter": {"pageSize": page_size},
            "projects": {"contains": "CONTAINS", "ids": [project.clockify_id]},
        }

        try:
            response = api.get_time_entries(payload)
        except Exception as e:
            logger.error(
                f"Error fetching time entries for project {project.clockify_id}: {e}"
            )
            issues = issues or True
            continue

        entries = response.get("timeentries", [])
        if not isinstance(entries, list):
            entries = []

        seen_entry_ids: set[str] = set()
        for entry in entries:
            entry_id = entry.get("id") or entry.get("_id")
            project_id = entry.get("projectId")
            user_email = entry.get("userEmail")
            time_interval = entry.get("timeInterval", {})
            start = time_interval.get("start")
            end = time_interval.get("end")

            if not (project_id and user_email and start and end and entry_id):
                logger.warning(f"Skipping incomplete entry: {entry_id}")
                issues = issues or True
                continue

            try:
                user = User.objects.get(email=user_email)
            except User.DoesNotExist:
                logger.warning(
                    f"User {user_email} not found. Skipping entry {entry_id}."
                )
                issues = issues or True
                continue

            start_time = datetime.datetime.fromisoformat(start.replace("Z", "+00:00"))
            end_time = datetime.datetime.fromisoformat(end.replace("Z", "+00:00"))

            TimeEntry.objects.update_or_create(
                clockify_id=entry_id,
                defaults={
                    "user": user,
                    "project": project,
                    "start_time": start_time,
                    "end_time": end_time,
                },
            )
            seen_entry_ids.add(entry_id)

        stale_entries = TimeEntry.objects.filter(
            project=project,
            clockify_id__isnull=False,
            start_time__gte=start_date,
            end_time__lte=end_date,
        ).exclude(clockify_id__in=seen_entry_ids)

        if stale_entries.exists():
            deleted_count, _ = stale_entries.delete()
            logger.info(
                f"Removed {deleted_count} stale entries for project {project.clockify_id}"  # noqa E501
            )

    return issues

sync_kimai_time_entries(days_back=30, end_date=timezone.now())

Task to sync time entries from Kimai API to TimeEntry model.

Parameters:

Name	Type	Description	Default
days_back	int	Number of days to look back for time entries.	30
end_date	datetime	The end date for the time entries to fetch.	now()

Returns:

Type	Description
bool	True if there is any issue, and False if all went well.

Source code in main/tasks.py

def sync_kimai_time_entries(
    days_back: int = 30,
    end_date: datetime.datetime = timezone.now(),
) -> bool:
    """Task to sync time entries from Kimai API to TimeEntry model.

    Args:
        days_back (int): Number of days to look back for time entries.
        end_date (datetime.datetime): The end date for the time entries to fetch.

    Returns:
        True if there is any issue, and False if all went well.
    """
    issues = False
    if (
        not settings.KIMAI_API_TOKEN
        or not settings.KIMAI_BASE_URL
        or not settings.CLOCKIFY_CUTOVER_DATE
    ):
        logger.warning("Kimai API information not found in environment variables")
        return False

    api = KimaiAPI(settings.KIMAI_API_TOKEN, settings.KIMAI_BASE_URL)
    start_date = end_date - datetime.timedelta(days=days_back)

    # The start date can be, at most, the day of the cutover
    start_date = (
        start_date
        if start_date >= settings.CLOCKIFY_CUTOVER_DATE
        else settings.CLOCKIFY_CUTOVER_DATE
    )
    if end_date < start_date:
        return False

    projects = Project.objects.filter(status__in=["Active", "Maintenance"]).exclude(
        kimai_id=-1
    )

    for project in projects:
        logger.info(f"Processing project ID: {project.kimai_id} - {project.name}")
        entries = []
        try:
            entries = api.get_time_entries(start_date, end_date, project.kimai_id)
        except Exception as e:
            logger.error(
                f"Error fetching time entries for project {project.kimai_id} - "
                f"{project.name}: {e}"
            )
            issues = issues or True
            continue

        seen_entry_ids: set[str] = set()
        logger.debug(
            f"{len(entries)} entries found for project {project.kimai_id} - "
            f"{project.name}"
        )
        for entry in entries:
            entry_id = entry["entry_id"]
            user_email = entry["user_email"]
            start = entry["start"]
            end = entry["end"]

            try:
                user = User.objects.get(email=user_email)
            except User.DoesNotExist:
                logger.warning(
                    f"User {user_email} not found. Skipping entry {entry_id}."
                )
                issues = issues or True
                continue

            start_time = datetime.datetime.fromisoformat(start)
            end_time = datetime.datetime.fromisoformat(end)

            TimeEntry.objects.update_or_create(
                kimai_id=entry_id,
                defaults={
                    "user": user,
                    "project": project,
                    "start_time": start_time,
                    "end_time": end_time,
                },
            )
            seen_entry_ids.add(entry_id)

        stale_entries = TimeEntry.objects.filter(
            project=project,
            kimai_id__gt=0,
            start_time__gte=start_date,
            end_time__lte=end_date,
        ).exclude(kimai_id__in=seen_entry_ids)

        if stale_entries.exists():
            deleted_count, _ = stale_entries.delete()
            logger.info(
                f"Removed {deleted_count} stale entries for project "
                f"{project.kimai_id} - {project.name}"
            )

    return issues

sync_timesheets_task()

Scheduled task to sync timesheets.

Source code in main/tasks.py

@db_periodic_task(crontab(**settings.HUEY_TASK_SCHEDULES["SYNC_TIMESHEETS"]))
def sync_timesheets_task() -> None:
    """Scheduled task to sync timesheets."""
    sync_clockify_time_entries()
    sync_kimai_time_entries()
    logger.info("Timesheet sync completed.")