def _get_monthly_locale_contributors()

in pontoon/messaging/emails.py [0:0]

• 63 lines of code
• 16 McCabe index (conditional complexity)

def _get_monthly_locale_contributors(locales, months_ago):
    month_date = timezone.now() - relativedelta(months=months_ago)

    actions = ActionLog.objects.filter(
        performed_by__profile__system_user=False,
        # Exclude system projects
        translation__entity__resource__project__system_project=False,
    )

    # Get contributors that started contributing to the locale in the given month
    first_contributions = (
        actions.values("performed_by", "translation__locale")
        .annotate(first_contribution_date=Min("created_at"))
        .filter(
            first_contribution_date__month=month_date.month,
            first_contribution_date__year=month_date.year,
        )
    )

    new_locale_contributors = {
        (entry["translation__locale"], entry["performed_by"])
        for entry in first_contributions
    }

    # Get all contributors in the given month,
    # grouped by locale and orderd by contribution count
    monthly_contributors = (
        actions.filter(
            created_at__month=month_date.month,
            created_at__year=month_date.year,
        )
        .values("translation__locale", "performed_by")
        .annotate(contribution_count=Count("id"))
        .order_by("-contribution_count")
    )

    # Group contributors by locale and user role
    results = {}
    all_locale_pks = [entry["translation__locale"] for entry in monthly_contributors]
    locales_dict = {locale.pk: locale for locale in locales}

    all_user_pks = [entry["performed_by"] for entry in monthly_contributors]
    users = User.objects.filter(pk__in=all_user_pks)
    users_dict = {user.pk: user for user in users}

    for locale_pk in all_locale_pks:
        locale_entries = [
            entry
            for entry in monthly_contributors
            if entry["translation__locale"] == locale_pk
        ]
        locale = locales_dict.get(locale_pk)

        new_contributors = []
        active_managers = []
        active_translators = []
        active_contributors = []

        for entry in locale_entries:
            user_pk = entry["performed_by"]
            user = users_dict.get(user_pk)

            if (locale_pk, user_pk) in new_locale_contributors:
                # Exclude staff users from new contributors
                if not user.is_staff:
                    new_contributors.append(user)

            if user in locale.managers_group.fetched_managers:
                active_managers.append(user)
            elif user in locale.translators_group.fetched_translators:
                active_translators.append(user)
            else:
                active_contributors.append(user)

        results[locale_pk] = {
            "new_contributors": new_contributors,
            "active_managers": active_managers,
            "active_translators": active_translators,
            "active_contributors": active_contributors,
        }

    return results