Rafactored beets/random.py and moved into beetsplug/random.py

beets/library/models.py

@@ -616,6 +616,11 @@ def try_sync(self, write, move, inherit=True):
         for item in self.items():
             item.try_sync(write, move)
 
+    @property
+    def length(self) -> float:
+        """Return the total length of all items in this album in seconds."""
+        return sum(item.length for item in self.items())
+
 
 class Item(LibModel):
     """Represent a song or track."""

beetsplug/random.py

@@ -1,37 +1,35 @@
-# This file is part of beets.
-# Copyright 2016, Philippe Mongeau.
-#
-# Permission is hereby granted, free of charge, to any person obtaining
-# a copy of this software and associated documentation files (the
-# "Software"), to deal in the Software without restriction, including
-# without limitation the rights to use, copy, modify, merge, publish,
-# distribute, sublicense, and/or sell copies of the Software, and to
-# permit persons to whom the Software is furnished to do so, subject to
-# the following conditions:
-#
-# The above copyright notice and this permission notice shall be
-# included in all copies or substantial portions of the Software.
-
 """Get a random song or album from the library."""
 
+from __future__ import annotations
+
+import random
+from itertools import groupby, islice
+from operator import attrgetter
+from typing import TYPE_CHECKING, Any, Iterable, Sequence, TypeVar
+
 from beets.plugins import BeetsPlugin
-from beets.random import random_objs
 from beets.ui import Subcommand, print_
 
+if TYPE_CHECKING:
+    import optparse
+
+    from beets.library import LibModel, Library
+
+    T = TypeVar("T", bound=LibModel)
+
 
-def random_func(lib, opts, args):
+def random_func(lib: Library, opts: optparse.Values, args: list[str]):
     """Select some random items or albums and print the results."""
     # Fetch all the objects matching the query into a list.
-    if opts.album:
-        objs = list(lib.albums(args))
-    else:
-        objs = list(lib.items(args))
+    objs = lib.albums(args) if opts.album else lib.items(args)
 
     # Print a random subset.
-    objs = random_objs(
-        objs, opts.album, opts.number, opts.time, opts.equal_chance
-    )
-    for obj in objs:
+    for obj in random_objs(
+        objs=list(objs),
+        number=opts.number,
+        time_minutes=opts.time,
+        equal_chance=opts.equal_chance,
+    ):
         print_(format(obj))
 
 
@@ -64,3 +62,95 @@ def random_func(lib, opts, args):
 class Random(BeetsPlugin):
     def commands(self):
         return [random_cmd]
+
+
+NOT_FOUND_SENTINEL = object()
+
+
+def _equal_chance_permutation(
+    objs: Sequence[T],
+    field: str = "albumartist",
+    random_gen: random.Random | None = None,
+) -> Iterable[T]:
+    """Generate (lazily) a permutation of the objects where every group
+    with equal values for `field` have an equal chance of appearing in
+    any given position.
+    """
+    rand: random.Random = random_gen or random.Random()
+
+    # Group the objects by artist so we can sample from them.
+    key = attrgetter(field)
+
+    def get_attr(obj: T) -> Any:
+        try:
+            return key(obj)
+        except AttributeError:
+            return NOT_FOUND_SENTINEL
+
+    groups: dict[Any, list[T]] = {
+        NOT_FOUND_SENTINEL: [],
+    }
+    for k, values in groupby(objs, key=get_attr):
+        groups[k] = list(values)
+        # shuffle in category
+        rand.shuffle(groups[k])
+
+    # Remove items without the field value.
+    del groups[NOT_FOUND_SENTINEL]
+    while groups:
+        group = rand.choice(list(groups.keys()))
+        yield groups[group].pop()
+        if not groups[group]:
+            del groups[group]
+
+
+def _take_time(
+    iter: Iterable[T],
+    secs: float,
+) -> Iterable[T]:
+    """Return a list containing the first values in `iter`, which should
+    be Item or Album objects, that add up to the given amount of time in
+    seconds.
+    """
+    total_time = 0.0
+    for obj in iter:
+        length = obj.length
+        if total_time + length <= secs:
+            yield obj
+            total_time += length
+
+
+def random_objs(
+    objs: Sequence[T],
+    number: int = 1,
+    time_minutes: float | None = None,
+    equal_chance: bool = False,
+    random_gen: random.Random | None = None,
+) -> Iterable[T]:
+    """Get a random subset of items, optionally constrained by time or count.
+
+    Args:
+    - objs: The sequence of objects to choose from.
+    - number: The number of objects to select.
+    - time_minutes: If specified, the total length of selected objects
+      should not exceed this many minutes.
+    - equal_chance: If True, each artist has the same chance of being
+      selected, regardless of how many tracks they have.
+    - random_gen: An optional random generator to use for shuffling.
+    """
+    rand: random.Random = random_gen or random.Random()
+
+    # Permute the objects either in a straightforward way or an
+    # artist-balanced way.
+    perm: Iterable[T]
+    if equal_chance:
+        perm = _equal_chance_permutation(objs, random_gen=rand)
+    else:
+        perm = list(objs)
+        rand.shuffle(perm)
+
+    # Select objects by time our count.
+    if time_minutes:
+        return _take_time(perm, time_minutes * 60)
+    else:
+        return islice(perm, number)

docs/changelog.rst

@@ -107,6 +107,7 @@ Other changes:
 * Refactored library.py file by splitting it into multiple modules within the
   beets/library directory.
 * Added a test to check that all plugins can be imported without errors.
+* Moved `beets/random.py` into `beetsplug/random.py` to cleanup core module.
 
 2.3.1 (May 14, 2025)
 --------------------