def _process_private_lb()

in competitions/leaderboard.py [0:0]

• 58 lines of code
• 17 McCabe index (conditional complexity)

    def _process_private_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)
                submission_info["submissions"] = [
                    sub for sub in submission_info["submissions"] if sub["status"] == SubmissionStatus.SUCCESS.value
                ]
                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[f"public_{k}"] = v
                    for k, v in sub["private_score"].items():
                        _sub[f"private_{k}"] = v
                    _sub["submission_datetime"] = sub["datetime"]
                    user_submissions.append(_sub)

                # count the number of submissions which are selected
                selected_submissions = 0
                for sub in user_submissions:
                    if sub["selected"]:
                        selected_submissions += 1

                if selected_submissions == 0:
                    # select submissions with best public score
                    user_submissions.sort(
                        key=lambda x: x[f"public_{self.scoring_metric}"], reverse=self.eval_higher_is_better
                    )
                    # select only the best submission
                    best_user_submission = user_submissions[0]

                elif selected_submissions <= self.max_selected_submissions:
                    # select only the selected submissions
                    user_submissions = [sub for sub in user_submissions if sub["selected"]]
                    # sort by private score
                    user_submissions.sort(
                        key=lambda x: x[f"private_{self.scoring_metric}"], reverse=self.eval_higher_is_better
                    )
                    # select only the best submission
                    best_user_submission = user_submissions[0]
                else:
                    logger.warning(
                        f"User {user_id} has more than {self.max_selected_submissions} selected submissions. Skipping user..."
                    )
                    continue

                # remove all keys that start with "public_"
                best_user_submission = {k: v for k, v in best_user_submission.items() if not k.startswith("public_")}

                # remove private_ from the keys
                best_user_submission = {k.replace("private_", ""): v for k, v in best_user_submission.items()}

                # remove selected key
                best_user_submission.pop("selected")
                submissions.append(best_user_submission)
        logger.info(f"Processed submissions in {time.time() - start_time} seconds")
        return submissions