def _process_public_lb()

in competitions/leaderboard.py [0:0]

• 39 lines of code
• 9 McCabe index (conditional complexity)

    def _process_public_lb(self):
        start_time = time.time()
        submissions_folder = snapshot_download(
            repo_id=self.competition_id,
            allow_patterns="submission_info/*.json",
            use_auth_token=self.token,
            repo_type="dataset",
        )
        logger.info(f"Downloaded submissions in {time.time() - start_time} seconds")
        start_time = time.time()
        submissions = []
        for submission in glob.glob(os.path.join(submissions_folder, "submission_info", "*.json")):
            with open(submission, "r", encoding="utf-8") as f:
                submission_info = json.load(f)
            # only select submissions that are done
            submission_info["submissions"] = [
                sub for sub in submission_info["submissions"] if sub["status"] == SubmissionStatus.SUCCESS.value
            ]
            submission_info["submissions"] = [
                sub
                for sub in submission_info["submissions"]
                if datetime.strptime(sub["datetime"], "%Y-%m-%d %H:%M:%S") < self.end_date
            ]
            if len(submission_info["submissions"]) == 0:
                continue

            user_id = submission_info["id"]
            user_submissions = []
            for sub in submission_info["submissions"]:
                _sub = {
                    "id": user_id,
                    # "submission_id": sub["submission_id"],
                    # "submission_comment": sub["submission_comment"],
                    # "status": sub["status"],
                    # "selected": sub["selected"],
                }
                for k, v in sub["public_score"].items():
                    _sub[k] = v
                _sub["submission_datetime"] = sub["datetime"]
                user_submissions.append(_sub)

            user_submissions.sort(key=lambda x: x[self.scoring_metric], reverse=self.eval_higher_is_better)
            best_user_submission = user_submissions[0]
            submissions.append(best_user_submission)
        logger.info(f"Processed submissions in {time.time() - start_time} seconds")
        return submissions