(self, user_id: int)
return stats_dto
Compute project specific stats for a given user
Compute project specific stats for a given user
Compute project specific stats for a given user
def get_project_user_stats(self, user_id: int) -> ProjectUserStatsDTO:
        """Compute project specific stats for a given user"""
        stats_dto = ProjectUserStatsDTO()
        stats_dto.time_spent_mapping = 0
        stats_dto.time_spent_validating = 0
        stats_dto.total_time_spent = 0

        total_mapping_time = (
            db.session.query(
                func.sum(
                    cast(func.to_timestamp(TaskHistory.action_text, "HH24:MI:SS"), Time)
                )
            )
            .filter(
                or_(
                    TaskHistory.action == "LOCKED_FOR_MAPPING",
                    TaskHistory.action == "AUTO_UNLOCKED_FOR_MAPPING",
                )
            )
            .filter(TaskHistory.user_id == user_id)
            .filter(TaskHistory.project_id == self.id)
        )
        for time in total_mapping_time:
            total_mapping_time = time[0]
            if total_mapping_time:
                stats_dto.time_spent_mapping = total_mapping_time.total_seconds()
                stats_dto.total_time_spent += stats_dto.time_spent_mapping

        query = (
            TaskHistory.query.with_entities(
                func.date_trunc("minute", TaskHistory.action_date).label("trn"),
                func.max(TaskHistory.action_text).label("tm"),
            )
            .filter(TaskHistory.user_id == user_id)
            .filter(TaskHistory.project_id == self.id)
            .filter(TaskHistory.action == "LOCKED_FOR_VALIDATION")
            .group_by("trn")
            .subquery()
        )
        total_validation_time = db.session.query(
            func.sum(cast(func.to_timestamp(query.c.tm, "HH24:MI:SS"), Time))
        ).all()

        for time in total_validation_time:
            total_validation_time = time[0]
            if total_validation_time:
                stats_dto.time_spent_validating = total_validation_time.total_seconds()
                stats_dto.total_time_spent += stats_dto.time_spent_validating

        return stats_dto
def get_project_user_stats ( self , user_id : int ) -> ProjectUserStatsDTO : stats_dto = ProjectUserStatsDTO ( ) stats_dto . time_spent_mapping = 0 stats_dto . time_spent_validating = 0 stats_dto . total_time_spent = 0 total_mapping_time = ( db . session . query ( func . sum ( cast ( func . to_timestamp ( TaskHistory . action_text , "HH24:MI:SS" ) , Time ) ) ) . filter ( or_ ( TaskHistory . action == "LOCKED_FOR_MAPPING" , TaskHistory . action == "AUTO_UNLOCKED_FOR_MAPPING" , ) ) . filter ( TaskHistory . user_id == user_id ) . filter ( TaskHistory . project_id == self . id ) ) for time in total_mapping_time : total_mapping_time = time [ 0 ] if total_mapping_time : stats_dto . time_spent_mapping = total_mapping_time . total_seconds ( ) stats_dto . total_time_spent += stats_dto . time_spent_mapping query = ( TaskHistory . query . with_entities ( func . date_trunc ( "minute" , TaskHistory . action_date ) . label ( "trn" ) , func . max ( TaskHistory . action_text ) . label ( "tm" ) , ) . filter ( TaskHistory . user_id == user_id ) . filter ( TaskHistory . project_id == self . id ) . filter ( TaskHistory . action == "LOCKED_FOR_VALIDATION" ) . group_by ( "trn" ) . subquery ( ) ) total_validation_time = db . session . query ( func . sum ( cast ( func . to_timestamp ( query . c . tm , "HH24:MI:SS" ) , Time ) ) ) . all ( ) for time in total_validation_time : total_validation_time = time [ 0 ] if total_validation_time : stats_dto . time_spent_validating = total_validation_time . total_seconds ( ) stats_dto . total_time_spent += stats_dto . time_spent_validating return stats_dto