RadioPlayer/modules/play_sort.py

from . import log95, PlaylistModifierModule, Track, Path

_log_out: log95.TextIO
assert _log_out # pyright: ignore[reportUnboundVariable]

def load_play_counts(file_path: Path) -> dict[str, int]:
    """
    Loads the play counts from the file generated by the play_counter module.
    """
    counts = {}
    try:
        with open(file_path, 'r') as file:
            for line in file:
                if line.strip() == "" or line.startswith(";"):
                    continue
                try:
                    key, value = line.split(':', 1)
                    counts[key.strip()] = int(value.strip())
                except ValueError:
                    continue
        return counts
    except FileNotFoundError:
        return {}

class PopularitySorterModule(PlaylistModifierModule):
    """
    A playlist modifier that reorders tracks based on their play history.
    For every pair of tracks, it gives a 60% probability to schedule the less-played track first.
    """
    def __init__(self) -> None:
        self.logger = log95.log95("PopSort", output=_log_out)
        self.play_counts_file = Path(__file__, "..", "..", "play_counter").resolve()

    def modify(self, global_args: dict, playlist: list[Track]) -> list[Track]:
        self.logger.info("Applying popularity-based sorting to the playlist...")

        play_counts = load_play_counts(self.play_counts_file)
        if not play_counts:
            self.logger.info("Play counter file not found or is empty. No sorting will be applied.")
            return playlist

        sorted_by_play_count = sorted(play_counts.items(), key=lambda item: item[1], reverse=True)

        SORT_LEN = 100
        if len(playlist) >= SORT_LEN:
            top_paths = {path for path, count in sorted_by_play_count[:SORT_LEN]}
            least_top_paths = {path for path, count in sorted_by_play_count[-SORT_LEN:]}
            for a,b in zip(top_paths, least_top_paths):
                a_track = b_track = None
                a_i = b_i = 0
                for a_i, a_track in enumerate(playlist):
                    if not a_track.official: continue
                    if a_track.path == a: break
                if not a_track: continue
                for b_i, b_track in enumerate(playlist):
                    if not b_track.official: continue
                    if b_track.path == b: break
                if not b_track: continue
                if a_i < b_i:
                    playlist[a_i], playlist[b_i] = playlist[b_i], playlist[a_i]

        i = 0
        while i < len(playlist) - 1:
            track1 = playlist[i]
            track2 = playlist[i+1]

            if not (track1.official and track2.official):
                i += 2
                continue

            count1 = play_counts.get(track1.path.as_posix(), 0)
            count2 = play_counts.get(track2.path.as_posix(), 0)

            if count1 > count2:
                playlist[i], playlist[i+1] = track2, track1

            i += 2

        self.logger.info("Popularity sorting complete.")
        return playlist

playlistmod = (PopularitySorterModule(), 2)