Release 22/12/2025

app/data_access/company.py

@@ -9,6 +9,9 @@
 from app.data_access.company_certification import CompanyCertificationType
 from app.data_access.employment import EmploymentOutput, OAuth2ClientOutput
 from app.data_access.mission import MissionConnection
+from app.data_access.regulation_computation import (
+    CompanyAdminRegulationComputationsByUserAndDay,
+)
 from app.data_access.regulatory_alerts_summary import (
     RegulatoryAlertsSummary,
 )
@@ -24,6 +27,9 @@
     is_employed_by_company_over_period,
     has_any_employment_with_company_or_controller,
 )
+from app.domain.regulation_computations import (
+    get_company_admin_regulation_computations,
+)
 from app.domain.regulatory_alerts_summary import get_regulatory_alerts_summary
 from app.domain.work_days import WorkDayStatsOnly
 from app.helpers.authorization import (
@@ -38,7 +44,12 @@
 )
 from app.helpers.pagination import to_connection
 from app.helpers.time import to_datetime
-from app.models import Company, User, Mission, Activity
+from app.models import (
+    Company,
+    User,
+    Mission,
+    Activity,
+)
 from app.models.activity import ActivityType
 from app.models.company_known_address import CompanyKnownAddressOutput
 from app.models.employment import (
@@ -231,6 +242,21 @@ class Meta:
             required=False, description="Identifiant du groupe sélectionné"
         ),
     )
+    admin_regulation_computations_by_user_and_by_day = graphene.List(
+        lambda: CompanyAdminRegulationComputationsByUserAndDay,
+        from_date=graphene.Date(
+            required=False,
+            description="Date de début de l'historique des alertes",
+        ),
+        to_date=graphene.Date(
+            required=False,
+            description="Date de fin de l'historique des alertes",
+        ),
+        description="Résultats de calcul de seuils règlementaires groupés par jour",
+        user_ids=graphene.List(
+            lambda: graphene.Int, description="Identifiants des utilisateurs"
+        ),
+    )
 
     def resolve_name(self, info):
         return self.name
@@ -495,3 +521,17 @@ def resolve_regulatory_alerts_recap(
         return get_regulatory_alerts_summary(
             month=month, user_ids=user_ids, unique_user_id=unique_user_id
         )
+
+    def resolve_admin_regulation_computations_by_user_and_by_day(
+        self, info, from_date=None, to_date=None, user_ids=None
+    ):
+        company_user_ids = [u.id for u in self.users]
+        if user_ids:
+            user_ids_set = set(user_ids)
+            company_user_ids = [
+                u_id for u_id in company_user_ids if u_id in user_ids_set
+            ]
+
+        return get_company_admin_regulation_computations(
+            user_ids=company_user_ids, from_date=from_date, to_date=to_date
+        )

app/data_access/regulation_computation.py

@@ -118,3 +118,24 @@ def __init__(self, day, regulation_computations):
 
     def __repr__(self):
         return f"Day {self.day} - #{len(self.regulation_computations)}"
+
+
+class CompanyAdminRegulationComputationsByUserAndDay(graphene.ObjectType):
+    day = graphene.Field(
+        graphene.Date,
+        description="Journée pour laquelle les seuils sont calculés (pour les calculs hebdomadaires, il s'agit du premier jour de la semaine en considérant qu'elle commence le lundi)",
+    )
+    user_id = graphene.Field(
+        graphene.Int, description="Identifiant de l'utilisateur"
+    )
+    nb_alerts_daily_admin = graphene.Field(
+        graphene.Int,
+        description="Nombres d'alertes gestionnaire journalières sur la journée",
+    )
+    nb_alerts_weekly_admin = graphene.Field(
+        graphene.Int,
+        description="Nombres d'alertes gestionnaire hebdomadaires sur la journée",
+    )
+
+    def __repr__(self):
+        return f"day={self.day} - user={self.user_id} #{self.nb_alerts_admin}"

app/domain/regulation_computations.py

@@ -1,6 +1,11 @@
 from itertools import groupby
 
-from app.models import RegulationComputation, RegulatoryAlert
+from app.data_access.regulation_computation import (
+    CompanyAdminRegulationComputationsByUserAndDay,
+)
+from app.helpers.submitter_type import SubmitterType
+from app.models import RegulationComputation, RegulatoryAlert, RegulationCheck
+from app.models.regulation_check import UnitType, RegulationCheckType
 
 
 def get_regulation_computations(
@@ -53,3 +58,77 @@ def get_regulatory_computations(user_id, start_date=None, end_date=None):
         .order_by(RegulationComputation.creation_time)
         .all()
     )
+
+
+def get_admin_regulatory_computations_for_users(
+    user_ids, from_date=None, to_date=None
+):
+    regulation_computations_query = RegulationComputation.query.filter(
+        RegulationComputation.user_id.in_(user_ids),
+        RegulationComputation.submitter_type == SubmitterType.ADMIN,
+    )
+    if from_date:
+        regulation_computations_query = regulation_computations_query.filter(
+            RegulationComputation.day >= from_date
+        )
+    if to_date:
+        regulation_computations_query = regulation_computations_query.filter(
+            RegulationComputation.day <= to_date
+        )
+    return regulation_computations_query.all()
+
+
+def get_company_admin_regulation_computations(
+    user_ids, from_date=None, to_date=None
+):
+    regulation_computations = get_admin_regulatory_computations_for_users(
+        user_ids=user_ids, from_date=from_date, to_date=to_date
+    )
+
+    def _get_alerts_dict(unit):
+        query = RegulatoryAlert.query.join(RegulationCheck).filter(
+            RegulatoryAlert.user_id.in_(user_ids),
+            RegulatoryAlert.submitter_type == SubmitterType.ADMIN,
+            RegulationCheck.unit == unit,
+        )
+        if from_date:
+            query = query.filter(RegulatoryAlert.day >= from_date)
+        if to_date:
+            query = query.filter(RegulatoryAlert.day <= to_date)
+        alerts = query.all()
+        alerts_dict = {}
+        for a in alerts:
+            alerts_dict.setdefault((a.user_id, a.day), []).append(a)
+        return alerts_dict
+
+    daily_alerts = _get_alerts_dict(UnitType.DAY)
+    weekly_alerts = _get_alerts_dict(UnitType.WEEK)
+
+    ret = []
+    for rc in regulation_computations:
+        day = rc.day
+        user_id = rc.user_id
+        daily_alerts_for_user = daily_alerts.get((user_id, day), [])
+        nb_alerts_daily_admin = 0
+        for a in daily_alerts_for_user:
+            if a.regulation_check.type == RegulationCheckType.ENOUGH_BREAK:
+                if a.extra["too_much_uninterrupted_work_time"]:
+                    nb_alerts_daily_admin += 1
+                if a.extra["not_enough_break"]:
+                    nb_alerts_daily_admin += 1
+            else:
+                nb_alerts_daily_admin += 1
+
+        weekly_alerts_for_user = weekly_alerts.get((user_id, day), [])
+        nb_alerts_weekly_admin = len(weekly_alerts_for_user)
+
+        ret.append(
+            CompanyAdminRegulationComputationsByUserAndDay(
+                day=day,
+                user_id=user_id,
+                nb_alerts_daily_admin=nb_alerts_daily_admin,
+                nb_alerts_weekly_admin=nb_alerts_weekly_admin,
+            )
+        )
+
+    return ret

app/seed/helpers.py

@@ -206,16 +206,12 @@ def log_and_validate_mission(
                 creation_time=work_periods[-1][1],
             )
     if admin_validating is not None:
-        db.session.add(
-            MissionValidation(
-                submitter=admin_validating,
-                mission=mission,
-                user=employee,
-                reception_time=work_periods[-1][1]
-                + datetime.timedelta(days=1),
-                is_admin=True,
-                creation_time=work_periods[-1][1] + datetime.timedelta(days=1),
-            )
+        validate_mission(
+            mission=mission,
+            is_admin_validation=True,
+            for_user=employee,
+            submitter=admin_validating,
+            creation_time=work_periods[-1][1] + datetime.timedelta(days=1),
         )
 
     return mission

app/seed/scenarios/lot_of_missions.py

@@ -26,6 +26,25 @@
 NB_HISTORY_REGULAR = 5
 NB_VEHICLES = 20
 NB_ADDRESSES = 20
+MINIMUM_HOUR = 3
+MAXIMUM_HOUR = 18
+
+
+def _get_random_work_periods(days_ago):
+    work_periods = []
+    nb_activities = random.choice([1, 2])
+    hours = random.sample(
+        range(MINIMUM_HOUR, MAXIMUM_HOUR + 1), nb_activities * 2
+    )
+    hours.sort()
+    for i in range(nb_activities):
+        work_periods.append(
+            [
+                get_time(how_many_days_ago=days_ago, hour=hours[i * 2]),
+                get_time(how_many_days_ago=days_ago, hour=hours[i * 2 + 1]),
+            ]
+        )
+    return work_periods
 
 
 def run_scenario_lot_of_missions():
@@ -67,12 +86,9 @@ def run_scenario_lot_of_missions():
                 vehicle=random.choice(company.vehicles),
                 address=random.choice(company.known_addresses),
                 add_location_entry=True,
-                work_periods=[
-                    [
-                        get_time(how_many_days_ago=nb_days_ago + 1, hour=14),
-                        get_time(how_many_days_ago=nb_days_ago + 1, hour=15),
-                    ]
-                ],
+                work_periods=_get_random_work_periods(
+                    days_ago=nb_days_ago + 1
+                ),
                 employee_comment="Commentaire du salarié",
                 employee_expenditure=random.choice(list(ExpenditureType)),
             )
@@ -85,7 +101,7 @@ def run_scenario_lot_of_missions():
     # Admin cancels missions
     for deleted_mission in deleted_missions:
         make_authenticated_request(
-            time=get_time(how_many_days_ago=1, hour=17),
+            time=get_time(how_many_days_ago=1, hour=MAXIMUM_HOUR + 2),
             submitter_id=admin.id,
             query=ApiRequests.cancel_mission,
             variables=dict(
@@ -107,16 +123,9 @@ def run_scenario_lot_of_missions():
                 vehicle=random.choice(company.vehicles),
                 address=random.choice(company.known_addresses),
                 add_location_entry=True,
-                work_periods=[
-                    [
-                        get_time(how_many_days_ago=nb_days_ago + 1, hour=6),
-                        get_time(how_many_days_ago=nb_days_ago + 1, hour=10),
-                    ],
-                    [
-                        get_time(how_many_days_ago=nb_days_ago + 1, hour=14),
-                        get_time(how_many_days_ago=nb_days_ago + 1, hour=18),
-                    ],
-                ],
+                work_periods=_get_random_work_periods(
+                    days_ago=nb_days_ago + 1
+                ),
                 employee_expenditure=random.choice(list(ExpenditureType)),
                 validate=employee_validates,
                 admin_validating=admin if admin_validates else None,

app/services/brevo/acquisition_data_finder.py

@@ -1,14 +1,14 @@
 """Acquisition funnel data finder."""
 
 from datetime import date
-from sqlalchemy import func
 from typing import List, Dict, Any
 
-from app import db
-from app.models import Employment, User
-from app.models.user import UserAccountStatus
 from ._config import BrevoFunnelConfig
-from .utils import get_companies_base_data, get_admin_info
+from .utils import (
+    get_companies_base_data,
+    get_admin_info,
+    get_creator_activation_status,
+)
 
 
 class AcquisitionDataFinder:
@@ -39,7 +39,7 @@ def find_companies(
 
         company_ids = [c["id"] for c in companies_base]
 
-        creator_activation = self._get_creator_activation_status(company_ids)
+        creator_activation = get_creator_activation_status(company_ids)
         admin_info = get_admin_info(company_ids)
 
         acquisition_companies = []
@@ -115,60 +115,6 @@ def _classify_acquisition_stage(
 
         return "entreprise inscrite sans compte activé"
 
-    def _get_creator_activation_status(
-        self, company_ids: List[int]
-    ) -> Dict[int, bool]:
-        """Check if company creators have activated their accounts.
-
-        The creator is the first admin (earliest employment ID) who registered the company.
-        A company is considered "activated" when its creator has activated their account.
-
-        Args:
-            company_ids: List of company IDs to check
-
-        Returns:
-            Dictionary mapping company_id to activation status (True/False)
-            - True: Creator account is active
-            - False: Creator account is not active or not found
-        """
-        if not company_ids:
-            return {}
-
-        first_admins = (
-            db.session.query(
-                Employment.company_id,
-                func.min(Employment.id).label("creator_id"),
-            )
-            .filter(
-                Employment.company_id.in_(company_ids),
-                Employment.has_admin_rights == True,
-            )
-            .group_by(Employment.company_id)
-            .subquery()
-        )
-
-        stats = (
-            db.session.query(
-                Employment.company_id,
-                func.count(User.id).label("active_count"),
-            )
-            .join(User, Employment.user_id == User.id)
-            .join(
-                first_admins,
-                (Employment.company_id == first_admins.c.company_id)
-                & (Employment.id == first_admins.c.creator_id),
-            )
-            .filter(
-                Employment.company_id.in_(company_ids),
-                Employment.has_admin_rights == True,
-                User.status == UserAccountStatus.ACTIVE,
-            )
-            .group_by(Employment.company_id)
-            .all()
-        )
-
-        return {stat.company_id: stat.active_count > 0 for stat in stats}
-
 
 def get_companies_acquisition_data() -> List[Dict[str, Any]]:
     finder = AcquisitionDataFinder()