update_list.py

# update_list.py - OmenServe-style layout, generating both lists (part 1 of 2)
import os
import io
import re
import sys
import shutil
import sqlite3
import datetime
import subprocess
import tempfile
import zipfile
import time
import json
import collections

# This script is an entry point of its own (README/INSTALL.md say "python3
# update_list.py"), and stays at the repository root for that reason - but
# everything it imports below now lives in src/ (#959).
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))

import defaults as config
import library
import platform_compat
import audio_info

# BEFORE ANYTHING PRINTS A FILENAME. This runs as its own process - the daemon
# starts it with subprocess.run() and configure.py runs it directly - so
# oserve.py's guard does nothing for it, and every line it writes is a path off
# somebody's disk.
#
# On a console whose code page cannot represent a character in one of those
# paths, print() raises UnicodeEncodeError and the scan dies where it stood.
# Reported from a live Greek-Windows install: "External update_list.py failed
# (Exit Code 1): Unknown script error", with the parent's own reader thread
# then dying on the bytes that had made it out - so the operator got a failure
# with no cause, for a library containing an accented filename.
platform_compat.install_console_encoding_guard()

# Multi-disc/box-set container names the !rar album list truncates at - see
# generate_master_list()'s own comment on the box-word block for why this
# has to match a whole PATH SEGMENT, not a substring anywhere in the path.
# An optional trailing number covers "CD1", "Disc 2", "Volume III" (digits
# only - "III" survives as part of the folder name, same as before this fix).
_BOX_WORD_RE = re.compile(
    r'^(cd|disc|disk|volume|digital media|media)\s*\d*$', re.IGNORECASE)

def format_size_human(bytes_size):
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes_size < 1024.0:
            return f"{bytes_size:.1f}{unit}"
        bytes_size /= 1024.0
    return f"{bytes_size:.1f}PB"

def format_total_size(bytes_size):
    for unit in ['B', 'KiB', 'MiB', 'GiB', 'TiB']:
        if bytes_size < 1024.0:
            return f"{bytes_size:.1f}{unit}"
        bytes_size /= 1024.0
    return f"{bytes_size:.1f}PiB"

def _extension_set(setting_name):
    """One setting's extensions, normalised: lower-case and dot-leading.

    Normalised HERE rather than trusted from the setting, because these are
    reached from three directions - settings.conf, admin_config.py and the
    dashboard's Settings page - and only one of them goes anywhere near a
    validator. An operator writing "DB, .ini, tmp" means the obvious thing,
    and so does "MKV,.mp4 , avi".

    The leading dot is not cosmetic. `"Thumbs.db".endswith("db")` is already
    true, so it is not what makes a file match; it is what stops an extension
    matching the END OF A NAME. Without it, ignoring "ts" also hides every
    file called `credits` or `highlights`.

    An empty result is a real answer for every one of these sets, so there is
    no fallback anywhere: skip nothing, no video, nothing packable.
    """
    raw = getattr(config, setting_name, None)
    if isinstance(raw, str):
        raw = raw.split(",")
    cleaned = []
    for item in (raw or []):
        text = str(item).strip().lower()
        if not text:
            continue
        cleaned.append(text if text.startswith(".") else "." + text)
    return tuple(dict.fromkeys(cleaned))


def ignored_extensions():
    """Extensions left out of every list."""
    return _extension_set("LIST_IGNORED_EXTENSIONS")


def video_extensions():
    """Extensions routed to the film and series list rather than the music one."""
    return _extension_set("LIST_VIDEO_EXTENSIONS")


def rar_extensions():
    """Extensions that make a folder packable with !rar."""
    return _extension_set("RAR_EXTENSIONS")


def video_companion_extensions():
    """Extensions that follow a video into its list when they share its
    folder - subtitles, .nfo, .sfv. See LIST_VIDEO_COMPANION_EXTENSIONS."""
    return _extension_set("LIST_VIDEO_COMPANION_EXTENSIONS")


def pack_size_over(path, cap):
    """Is packing `path` going to exceed `cap` bytes? Returns (over, measured).

    RECURSIVE, because the pack is: `rar a <dir>` takes the directory and
    everything under it, so a check that only looked at the top level would
    measure something other than what gets packed.

    STOPS AS SOON AS THE CAP IS PASSED. The answer wanted here is a yes or no,
    not a total, and the folder this is most useful on is the one that is
    enormous - so walking all of it to produce a number nobody reads is the
    one cost worth avoiding. A folder a hundred times over the cap is refused
    after a few thousand entries instead of after all of them.

    `cap` of 0 or less means no limit, and returns (False, 0) without touching
    the disk at all: an operator who has not set one pays nothing for this.

    An unreadable entry is skipped rather than counted or raised on. This runs
    on the REQUEST path, where the alternative to an answer is a user who gets
    no reply - and the pack that follows would meet the same unreadable file
    and report it properly.
    """
    if not cap or cap <= 0:
        return False, 0
    measured = 0
    for root, _dirs, files in os.walk(platform_compat.long_path(path)):
        for name in files:
            try:
                measured += os.path.getsize(
                    platform_compat.long_path(os.path.join(root, name)))
            except OSError:
                continue
            if measured > cap:
                return True, measured
    return False, measured


def walk_with_sizes(top, onerror=None, workers=None):
    """Every file under `top`, with the size the directory entry already knew.

    WHY THIS EXISTS. os.walk is built on os.scandir, which gets each entry's
    size from the directory enumeration itself - and then throws it away,
    because os.walk's contract is names only. The caller then asks
    os.path.getsize() for a number the filesystem has just finished telling
    us. One redundant syscall per file, and on a network share one redundant
    ROUND TRIP per file.

    Measured on 20,000 files, local SSD, warm cache, both producing the same
    answer:

        os.walk + getsize    0.356s   (17.8 us/file)
        os.scandir + cached  0.095s   ( 4.7 us/file)   3.8x

    That is the WALK. A whole rebuild, timed the same way over 30,000 files
    and producing byte-identical lists, goes 2.51s -> 1.56s: 1.61x, because
    writing and packing are the rest of the job and this does not touch them.
    The walk's share is what grows on a network drive, where the second ask is
    a round trip rather than a cached answer.

    Local disk is the BEST case for the old shape. The library this was
    written for is 799,438 files on a mapped network drive, where a rebuild
    takes fifteen and a half minutes.

    Yields (dirpath, [(name, size), ...]), one tuple per directory, so the
    caller's loop keeps its shape.

    A file whose size cannot be read yields a size of None rather than being
    dropped here. That decision belongs to the caller, which already logs the
    path and excludes it - see the [LIST-GEN ERROR] branch - and a helper that
    silently skipped it would take that log line away.

    `entry.stat()` FOLLOWS symlinks, exactly as os.path.getsize() did, so a
    symlinked track still reports its target's size. On Windows that costs a
    syscall only when the entry really is a symlink; an ordinary file is
    answered from what the enumeration already returned.

    A symlinked DIRECTORY is classified as a directory - `entry.is_dir()`,
    following, exactly as os.walk does when deciding what goes in `dirs` - and
    then not descended into, which is os.walk's followlinks=False default. Two
    decisions, made separately, because collapsing them into
    is_dir(follow_symlinks=False) answers False for a symlinked directory and
    hands it back as a file.

    `onerror` is called with the OSError, matching os.walk's parameter of the
    same name, so an unreadable subtree is reported the way it always was
    rather than ending the scan.

    SEVERAL DIRECTORIES AT ONCE (#922). On a network mount every scandir()
    and every entry.stat() is a round trip - Linux's d_type gives the type
    but not the size - and one directory at a time, none of them overlap: a
    64,136-file NFS library spent about 80 s here, with every search paused.
    `workers` directories (LIST_SCAN_THREADS unless given) are listed and
    stat'd at once, the way QuickList - OmenServe's list maker - walks. Each
    directory is classified exactly as below whichever thread lists it; only
    the order directories come back in changes, and the caller sorts before
    writing anything (#443). onerror is called here, on the caller's thread,
    never from a worker. One worker is the walk as it always was.

    Finished directories are collected through a queue (#1124): each one's
    future puts itself on it when done, and the caller takes them off one at
    a time. Waiting with concurrent.futures.wait() on every outstanding
    directory instead rescanned - and locked - all of them on each
    completion, which made the walk quadratic in the number of directories:
    41 s against 3 s on 137k files, and 16 workers slower than one.
    """
    if workers is None:
        workers = scan_workers()
    workers = max(1, int(workers))

    def list_one(current):
        """(files, subdirs, errors) for one directory, or None for files when
        it could not be listed at all. Run by whichever thread gets it; says
        nothing itself - the errors go back to the caller's thread."""
        errors = []
        subdirs = []
        try:
            with os.scandir(current) as scanning:
                entries = list(scanning)
        except OSError as err:
            return None, subdirs, [err]

        files = []
        for entry in entries:
            try:
                # CLASSIFYING AND DESCENDING ARE TWO DECISIONS, and os.walk
                # makes them separately. A symlink to a directory IS a
                # directory - os.walk puts it in `dirs`, so it never reaches a
                # caller as a file - and with followlinks=False it simply is
                # not descended into.
                #
                # Doing both with is_dir(follow_symlinks=False) collapses them
                # and gets the first one wrong: that answers False for a
                # symlinked directory, which made this walk treat it as a FILE
                # and stat it, and would have published a directory as a
                # downloadable entry in the list.
                #
                # Caught by CI on Linux, where a symlink can be created without
                # elevation, while the same test skipped on the Windows box
                # that wrote it.
                if entry.is_dir():
                    if not entry.is_symlink():
                        subdirs.append(entry.path)
                    continue
            except OSError as err:
                # A directory entry that cannot even be classified. Report it
                # like an unreadable subtree - it is one - rather than
                # guessing it is a file and failing again on the stat.
                errors.append(err)
                continue
            try:
                files.append((entry.name, entry.stat().st_size))
            except OSError:
                files.append((entry.name, None))
        return files, subdirs, errors

    def report(errors):
        if onerror is not None:
            for err in errors:
                onerror(err)

    if workers == 1:
        pending = [top]
        while pending:
            current = pending.pop()
            files, subdirs, errors = list_one(current)
            report(errors)
            pending.extend(subdirs)
            if files is not None:
                yield current, files
        return

    import queue
    from concurrent.futures import ThreadPoolExecutor
    pool = ThreadPoolExecutor(max_workers=workers, thread_name_prefix="list-scan")
    # Each directory's future puts (directory, future) here when it is done,
    # so taking the next finished one costs the same however many are still
    # outstanding (#1124). SimpleQueue does its own locking.
    finished = queue.SimpleQueue()

    def submit(directory):
        future = pool.submit(list_one, directory)
        future.add_done_callback(lambda done, path=directory: finished.put((path, done)))

    try:
        submit(top)
        outstanding = 1
        while outstanding:
            current, future = finished.get()
            outstanding -= 1
            files, subdirs, errors = future.result()
            report(errors)
            for subdir in subdirs:
                submit(subdir)
            outstanding += len(subdirs)
            if files is not None:
                yield current, files
    finally:
        # A caller that stops early - an exception mid-scan - must not leave
        # workers listing a library nobody is reading any more. The futures
        # this cancels still run their done-callback and land in `finished`,
        # which nobody reads any more; that is harmless.
        pool.shutdown(wait=True, cancel_futures=True)


def scan_workers():
    """LIST_SCAN_THREADS, held to 1..64."""
    try:
        wanted = int(getattr(config, "LIST_SCAN_THREADS", 16) or 1)
    except (TypeError, ValueError):
        wanted = 1
    return max(1, min(64, wanted))


def _has_extension(name, extensions):
    return bool(extensions) and str(name).lower().endswith(tuple(extensions))


def folder_totals(rows):
    """{folder: (file count, bytes)} over (folder, name, bytes) rows."""
    totals = {}
    for folder, _name, size in rows:
        count, total = totals.get(folder, (0, 0))
        totals[folder] = (count + 1, total + (size or 0))
    return totals


def folder_summary_line(count, total_bytes, human):
    """The line under a folder heading that says what the folder holds (#69):

        14 files, 1.20GB

    ITS OWN LINE, AFTER THE CLOSING RULE, and never on the heading itself.
    The heading is not decoration: list.py's reader and dcc.py's request
    resolver both take the whole heading line as the folder path, and so
    does every older DCCore fetching this list - a size appended there would
    resolve to a folder that does not exist on every one of them. A line
    that is neither a rule nor a "!" request row is skipped by all of those
    readers (list.py's state machine drops it in its resting state; dcc.py
    only looks at prefix lines and "!" lines; AutoQ imports "!" rows), so
    this can say anything. Indented so it reads as belonging to the heading
    above rather than as a heading of its own.
    """
    return f"    {count:,} file{'' if count == 1 else 's'}, {human(total_bytes)}"


def is_video_file(name, video=None):
    """Is this file a video, by extension?"""
    return _has_extension(name, video_extensions() if video is None else video)


def is_video_companion_file(name, companions=None):
    """Is this a file that belongs beside a video - a subtitle, .nfo, .sfv?"""
    return _has_extension(name, video_companion_extensions() if companions is None else companions)


def belongs_in_video_list(name, folder_has_video, video=None, companions=None):
    """Does this file go in the video list rather than the music one?

    A video does, wherever it is. A companion file does only when its folder
    holds a video (#411): a scene release then travels whole - the .mkv, its
    .srt, its .nfo and its .sfv - while an album's .nfo stays with the album.
    `folder_has_video` is the caller's, decided once per folder from the
    folder's own files, so the answer for a companion cannot depend on the
    order the files were met in.
    """
    if is_video_file(name, video):
        return True
    return bool(folder_has_video) and is_video_companion_file(name, companions)


def is_packable_file(name, packable=None):
    """Does this file make its folder requestable with !rar?

    Its own set, deliberately not "anything the scan indexed" - see
    RAR_EXTENSIONS in defaults.py for what that cost.
    """
    return _has_extension(name, rar_extensions() if packable is None else packable)


def has_backslash_component(relative_path, separator=None):
    """Does any component of `relative_path` carry a literal backslash?

    The list writes folder headings with backslashes between the components
    and list.list_heading_parts() reads them back by splitting on both
    separators, so a backslash INSIDE a name is indistinguishable from the
    separator between two - and the folder a heading resolves to is then a
    different one, or none (#464).

    THE ANSWER DEPENDS ON WHAT THE SEPARATOR IS, which is the whole of the
    platform question here. On Linux the separator is "/" and a backslash is
    an ordinary filename character, so "Rock\\Metal" is ONE folder with an
    awkward name. On Windows the backslash IS the separator, so the same text
    is two folders and no name can contain one - this returns False for every
    path a Windows install can produce, correctly.

    `separator` is for the tests, the way irc.resolve_dcc_address()'s
    `lookup` is: the hazard exists on one platform, and a test that could only
    run there would be a hole on the other. Production passes nothing and gets
    os.sep, which is what generate_master_list() builds rel_dir with.
    """
    sep = os.sep if separator is None else separator
    flattened = str(relative_path).replace(sep, "/")
    return any("\\" in part for part in flattened.split("/"))


def relative_folder(root, scan_root, paths=None):
    """os.path.relpath(root, scan_root), for a `root` the walk produced.

    THE WALK BUILDS EVERY ROOT FROM scan_root ITSELF (#1138): each directory
    is its parent's path, a separator and the entry's name, so the relative
    path is simply what follows scan_root - a slice. relpath() made each one
    absolute and normalised both paths first, about 10-20 us a directory on
    Windows for an answer the string already held.

    The slice is taken only where relpath() could not answer differently:
    `root` must begin with scan_root and a separator, and what follows must
    be plain names - nothing empty, no "." or "..", and on Windows no forward
    slash, no colon (a drive or a stream to relpath(); no Windows name holds
    one) and no name ending in a dot or a space, which Windows'
    normalisation strips. Anything else asks relpath(), exactly as before.

    `paths` is for the tests, like has_backslash_component()'s `separator`:
    ntpath or posixpath, so both platforms' rules are checked on either.
    Production passes nothing and gets os.path.
    """
    paths = os.path if paths is None else paths
    if root == scan_root:
        return "."
    sep = paths.sep
    prefix = scan_root if scan_root.endswith((sep, paths.altsep or sep)) else scan_root + sep
    if root.startswith(prefix):
        rest = root[len(prefix):]
        names = rest.split(sep)
        if paths.altsep:
            plain = paths.altsep not in rest and ":" not in rest and all(
                name and name[-1] not in ". " for name in names)
        else:
            plain = all(name and name != "." and name != ".." for name in names)
        if plain:
            return rest
    return paths.relpath(root, scan_root)


def is_listed_file(name, ignored=None):
    """Does this file go into the list? Everything does, unless it is skipped.

    `ignored` is the hot-path argument: the scan resolves the setting ONCE and
    passes the result down, because this is asked of every file in the library
    and a 719k-file library would otherwise rebuild the tuple 719,000 times.
    Callers with one file to check can leave it out.

    A file with no extension is listed. So is a dotfile, and so is anything
    else the operator has not named - "every file" is the rule, and the
    setting is the only exception to it.
    """
    if ignored is None:
        ignored = ignored_extensions()
    if not ignored:
        return True
    return not str(name).lower().endswith(ignored)


def _one_line(text):
    """Flatten anything that would break the one-entry-per-line format.

    POSIX filenames may contain newlines and other control characters - only "/"
    and NUL are forbidden - so a track called "evil\nname.flac" is perfectly
    legal on the Linux box this daemon runs on. Written straight into the list it
    splits one request entry into two lines, leaving a truncated entry and an
    orphan fragment, and the file every user downloads is malformed from there
    down. Windows refuses such names at creation, which is why CI caught this on
    the ubuntu jobs only.

    Nobody can plant one remotely - the library is the operator's own mount - so
    this is robustness rather than a security boundary. Replacing with a space
    keeps the entry visible and the file structurally sound; such a track is
    already unrequestable, because the request parser splits on whitespace too.

    Also sanitises non-UTF-8 bytes. os.walk() on POSIX decodes filenames with
    the "surrogateescape" error handler by default, so a name with bytes that
    are not valid UTF-8 (a CP1252 rip, a bad extraction, a FAT copy) comes back
    as a lone surrogate codepoint - not itself a control character, but not
    valid UTF-8 either, and this text is about to be written with a strict
    UTF-8 encoder. Sanitised here rather than left to fail at the write: one
    bad name in a library of thousands now costs a mangled-but-valid name in
    the list, not the entire rebuild.

    Almost no name needs either change, so one regex search decides first
    (#1125): the per-character pass below cost about 15 us per row, and the
    largest share of a rebuild's writing time. The pattern is exactly the
    characters the slow path changes - the controls and DEL it flattens, and
    the lone surrogates that the UTF-8 'replace' round trip turns into "?",
    which is the only thing that round trip changes.
    """
    text = str(text)
    if _NEEDS_ONE_LINE_CLEANING.search(text) is None:
        return text
    text = text.encode("utf-8", "replace").decode("utf-8")
    return "".join(" " if ch < " " or ch == "\x7f" else ch for ch in text)


# What _one_line() would change: a control character, DEL, or a lone
# surrogate (#1125). A name with none of them is returned as it is.
_NEEDS_ONE_LINE_CLEANING = re.compile("[\x00-\x1f\x7f\ud800-\udfff]")


def _heading_text(folder):
    """The heading line a folder gets in the list, exactly as written.

    One function for the music list and for the rewrite of its length and
    quality (#1182), which finds each row's cache entry by the heading above
    it: two copies of this expression could drift, and a row would then miss
    its suffix in one and not the other."""
    import list as list_mod
    raw = f"{list_mod.LIST_FOLDER_PREFIX}{folder}\\" if folder else list_mod.LIST_FOLDER_PREFIX
    return _one_line(raw.replace("/", "\\"))


def _discard_temp_lists(*paths):
    """Remove half-written temporary lists so they cannot be mistaken for real ones."""
    for path in paths:
        try:
            if path and os.path.exists(path):
                os.remove(path)
        except OSError as err:
            print(f"[LIST-CLEAN ERROR] Could not remove {path}: {err}")


# How patient the swap is with a reader holding a list open (#923). Searches
# now run during the scan; the swap refuses new ones, but one that started a
# moment before may still be reading the list - a second or two on a very
# large one - and on Windows the rename waits for it. Ten attempts back off to
# about ten seconds in all; POSIX never retries at all.
PUBLISH_REPLACE_ATTEMPTS = 10


def _publish_artifacts(swaps):
    """Move every (temporary, destination) pair into place, or none of them.

    THE PUBLISH USED TO BE FIVE INDEPENDENT REPLACEMENTS, so a failure partway
    through left the bot advertising one scan and handing out another. The
    ordinary way to reach it, on Windows: os.replace onto a file another
    handle has open raises PermissionError, and dcc.py holds the published
    artifact open for the whole duration of a DCC send. PAUSE_ON_UPDATE only
    refuses NEW requests, so a transfer already in flight keeps that handle -
    and somebody downloading the list when the scheduled rebuild lands is not
    an edge case on a bot with several slots.

    What that produced: the master index replaced, so @find, the advert count
    and commands.count_from_master_list() all reported the new scan - while
    the archive users actually received was the previous one, the size side
    files still carried the previous numbers, the base-name marker was not
    updated, and the prune never ran. Nothing recovered it; the artifact
    stayed stale until some later rebuild happened to run with no transfer in
    progress. And the failure branch then printed "The previous list was left
    untouched and is still in use", which by then was false.

    Each destination is moved ASIDE before its replacement lands, so a failure
    can put back exactly what was there. That also makes the locked case fail
    at the safest possible moment: renaming a file another process holds open
    fails on Windows too, so the lock is discovered while moving the old file
    out of the way - before anything observable has changed.

    Raises whatever the underlying replace raised, after rolling back. The
    caller's message about the previous list still being in use is then true
    again, which is the point.
    """
    done = []
    try:
        for temporary, destination in swaps:
            backup = None
            if os.path.exists(platform_compat.long_path(destination)):
                backup = destination + ".previous"
                platform_compat.replace_with_retry(destination, backup,
                                                   attempts=PUBLISH_REPLACE_ATTEMPTS)
            platform_compat.replace_with_retry(temporary, destination,
                                               attempts=PUBLISH_REPLACE_ATTEMPTS)
            done.append((destination, backup))
    except Exception:
        # Reverse order because that is the convention for undoing a
        # sequence, not because it is required here: each swap touches only
        # its own destination and that destination's .previous, so no two of
        # them can collide. A mutation run flipped the order and nothing
        # failed, which is the honest reading - the comment that used to sit
        # here claimed a necessity there is not one of.
        for destination, backup in reversed(done):
            try:
                if backup:
                    platform_compat.replace_with_retry(backup, destination)
                else:
                    # There was nothing here before; leaving the new file
                    # would publish half a rebuild.
                    os.remove(platform_compat.long_path(destination))
            except OSError as undo_err:
                print(f"[LIST-GEN ERROR] Could not roll {destination} back: "
                      f"{undo_err}")
        raise

    for _destination, backup in done:
        if not backup:
            continue
        try:
            os.remove(platform_compat.long_path(backup))
        except OSError:
            # A leftover .previous is clutter, not a failure - the publish
            # itself succeeded and that is what the caller is waiting on.
            pass
    return True


def _prune_superseded_lists(keep, directory=None):
    """Delete older generated lists once the new ones are safely in place.

    Runs AFTER the swap, never before: the previous index has to stay usable for the whole
    scan, which can take minutes on a large NFS mount.
    """
    # The list's OWN directory. Every list prunes only what it wrote: with
    # more than one, a prune reaching across them would delete another list's
    # current index the moment their base names matched, which they always do.
    directory = directory or config.LOCAL_LIST_DIR
    removed = 0
    try:
        entries = os.listdir(directory)
    except OSError as err:
        print(f"[LIST-CLEAN ERROR] Could not read {directory}: {err}")
        return

    for item in entries:
        if item in keep:
            continue
        # The HYPHEN is the point. Every generated list is
        # f"{LIST_BASE_NAME}-{today}.txt" or f"{LIST_BASE_NAME}-RAR-{today}.txt",
        # so the separator is always there - and matching on the bare prefix
        # also matched the side files, which live in this same directory.
        #
        # With LIST_BASE_NAME derived from a nickname like "dccore" or "dcc",
        # the size side file starts with that prefix too, so every rebuild
        # wrote the side files and then deleted them again. The library's size
        # then disappeared from every public surface permanently: the advert
        # published "Files (0B)" and the CTCP SLOTS payload published 0 raw
        # bytes, on every interval, for ever - and the log line for it read
        # "[LIST-CLEAN] Removed 2 superseded list(s)", which sounds like
        # housekeeping working. Found by audit.
        if not item.startswith(config.LIST_BASE_NAME + "-"):
            continue

        # Belt and braces: never remove a file this run just wrote, whatever
        # the name matching decides.
        if item in (os.path.basename(str(getattr(config, "LIST_SIZE_FILE", ""))),
                    os.path.basename(str(getattr(config, "LIST_RAWBYTES_FILE", "")))):
            continue
        # ".rar" is here because LIST_FORMAT can publish one. Only names that
        # also start with LIST_BASE_NAME are considered, and this is the lists
        # directory, so no album archive a user is waiting on is in reach.
        if not item.endswith((".txt", ".zip", ".rar")):
            continue
        try:
            os.remove(os.path.join(directory, item))
            removed += 1
        except OSError as err:
            print(f"[LIST-CLEAN ERROR] Could not remove {item}: {err}")

    if removed:
        print(f"[LIST-CLEAN] Removed {removed} superseded list(s).")


# The literal shipped default defaults.py derives LIST_BASE_NAME away FROM -
# see its own comment on why the derivation compares against this same
# literal rather than a shared constant (a snapshot taken before overrides
# apply, the way SHIPPED_DEFAULTS does for settings_file.REQUIRED, would
# need LIST_BASE_NAME added to REQUIRED just to get one, which it deliberately
# is not).
_SHIPPED_LIST_BASE_NAME = "DCCore"


# The name the artifacts in LOCAL_LIST_DIR were last published under.
#
# #213: the migration below could only ever carry files across from the shipped
# default, so it worked exactly once. Rename the bot a second time - or from any
# value that was never "DCCore" - and the artifacts on disk keep the old name
# while the bot looks for the new one. A restart does not recover it: the list
# is right there and invisible, and the advert says 0 files.
#
# The previous name has to be remembered somewhere. A marker file in the lists
# directory rather than settings.conf/admin_config.py, because it travels with
# the thing it describes: a value in the config can be hand-edited, replaced
# wholesale on an upgrade, or restored from a backup taken before the rename,
# and each of those silently orphans the lists again - which is the exact
# failure being closed. A file in the directory cannot drift from the directory
# it names, because it is in it.
#
# Absent means an install from before this existed, which is precisely when
# falling back to _SHIPPED_LIST_BASE_NAME is the right guess.
#
# The leading dot and the lack of a .txt/.zip/.rar suffix keep it out of every
# artifact scan in list.py and out of the glob in find_latest_list().
_LIST_BASE_MARKER = ".dccore-list-base"


def list_base_marker_path(directory=None):
    directory = directory or getattr(config, "LOCAL_LIST_DIR", "./lists")
    return os.path.join(directory, _LIST_BASE_MARKER)


def read_list_base_marker(directory=None):
    """The LIST_BASE_NAME the artifacts on disk were published under, or None.

    None on any read problem, deliberately: an unreadable marker must fall back
    to the shipped-default guess, never raise into startup.
    """
    try:
        with io.open(list_base_marker_path(directory), encoding="utf-8") as handle:
            return handle.read().strip() or None
    except (OSError, UnicodeDecodeError):
        return None


def write_list_base_marker(name=None, directory=None, log=print):
    """Record the name the artifacts are now published under. Returns True on
    success.

    Written through db._atomic_write, like the advert side files: a marker
    truncated by a crash mid-write would read as absent, sending the next
    migration back to the shipped-default guess and stranding the artifacts
    this call exists to keep findable.
    """
    import db

    name = config.LIST_BASE_NAME if name is None else name
    try:
        db._atomic_write(list_base_marker_path(directory), str(name) + "\n")
        return True
    except Exception as err:
        log(f"[MIGRATE] Could not record the list base name: {err}. "
            f"A future rename may not find these files.")
        return False

def migrate_list_base_name(log=print):
    """Carry existing list files across when LIST_BASE_NAME changed out from
    under them - #184's review: defaults.py's LIST_BASE_NAME
    derivation (an untouched value takes NICKNAME's own value once NICKNAME
    is set) means every existing install's list files, generated before that
    derivation existed, are sitting on disk as "DCCore-<date>.*" while
    LIST_BASE_NAME now resolves to the operator's nickname instead.

    Without this, find_latest_list() globs for the NEW base name, finds
    nothing, and the daemon boots, joins its channels and advertises with no
    list at all - not because there is no list, but because the one on disk
    is filed under a name nothing is looking for any more. It stays that way
    until the next successful !update, which on a weekly rebuild schedule is
    up to a week of a bot that looks healthy and answers every request with
    "not found". The side-file migration's own docstring described
    this exact failure shape for the side-file rename; this is the same
    problem, for a prefix rather than a single filename.

    Deliberately narrow, matching that function's safety properties:

      * only when LIST_BASE_NAME no longer equals what defaults.py ships -
        an install that never had any "DCCore-*" files (fresh, or one that
        chose its own LIST_BASE_NAME from the very first run) has nothing to
        move, and this is a no-op for it.
      * only a file whose new name does not already exist is moved - a
        rebuild that has already happened under the new name wins over
        anything left behind from the old one.
      * os.replace, so an interrupted run leaves one intact file rather than
        two halves; a failure is logged and swallowed per file, because a
        daemon that will not start over a rename is a worse outcome than the
        rename not happening for one file.

    Returns the list of (old, new) basenames actually moved, for the tests
    and for the startup log.
    """
    # EVERY list's directory, not only the primary's. A list's files live in
    # its own directory - list.list_dir(name), with the primary keeping
    # LOCAL_LIST_DIR itself - and the marker that records what they are called
    # is already per-directory. This function simply never looked anywhere but
    # the primary, so renaming the bot orphaned every other list's artifacts:
    # they kept the old base name, nothing on the next startup knew to look
    # for it, and each of those channels advertised a library it no longer had
    # a list for.
    #
    # A failure in one directory must not stop the others, for the same reason
    # a failure on one file does not stop the rest of that directory: a daemon
    # that will not start over a rename is worse than the rename not
    # happening.
    import list as list_mod
    import library

    # No de-duplication of directories. It was written, and then deleted for
    # being unreachable: list_dir() answers LOCAL_LIST_DIR for any list marked
    # primary, so two primaries would share a directory - but load_lists()
    # normalises a hand-edited file down to exactly one primary before this
    # ever sees it. A guard that cannot be reached is a guard no test can
    # falsify, and this file has deleted two of those already.
    moved = []
    for served in library.lists():
        try:
            directory = list_mod.list_dir(served.name)
        except Exception as err:  # a malformed lists.json is not fatal here
            log(f"[MIGRATE] Could not resolve the directory for list "
                f"{served.name!r}: {err}")
            continue
        moved.extend(_migrate_one_list_directory(directory, log=log))
    return moved


def _migrate_one_list_directory(directory, log=print):
    """migrate_list_base_name() for ONE list's directory. Returns the
    (old, new) basenames actually moved."""

    # What the files on disk are actually called, not what they were called
    # when the bot shipped. Absent means an install from before the marker
    # existed, where the shipped default is the right guess (#213).
    previous = read_list_base_marker(directory) or _SHIPPED_LIST_BASE_NAME

    if previous == config.LIST_BASE_NAME:
        # Nothing to move. Still record it: an install that has never been
        # renamed has no marker, and writing one now means its FIRST rename is
        # migrated from the right name rather than from the shipped guess.
        write_list_base_marker(config.LIST_BASE_NAME, directory, log=log)
        return []

    try:
        entries = os.listdir(directory)
    except OSError as err:
        log(f"[MIGRATE] Could not read {directory}: {err}")
        return []

    # The "-" is required, not just the bare prefix: every real artifact is
    # named "DCCore-<date>.ext" or "DCCore-RAR-<date>.ext", always with a
    # hyphen immediately after the base name. A bare startswith("DCCore")
    # would also match a file already renamed to the NEW LIST_BASE_NAME when
    # that name itself happens to start with "DCCore" - e.g. "DCCoreTest" -
    # corrupting an already-correct file instead of leaving it alone.
    old_prefix = previous + "-"

    moved = []
    for item in entries:
        if not item.startswith(old_prefix):
            continue
        if not item.endswith((".txt", ".zip", ".rar")):
            continue

        new_name = config.LIST_BASE_NAME + item[len(previous):]
        old_path = os.path.join(directory, item)
        new_path = os.path.join(directory, new_name)
        if os.path.exists(new_path):
            continue
        try:
            platform_compat.replace_with_retry(old_path, new_path)
            moved.append((item, new_name))
        except OSError as err:
            log(f"[MIGRATE] Could not rename {item} to {new_name}: {err}. "
                f"It will not be found until the next successful !update.")

    for old_name, new_name in moved:
        log(f"[MIGRATE] Renamed {old_name} to {new_name}.")

    # After the move, not before: if the renames failed the marker must still
    # say what the files on disk are really called, or the next startup would
    # look for them under a name nothing has.
    if moved or not read_list_base_marker(directory):
        write_list_base_marker(config.LIST_BASE_NAME, directory, log=log)
    return moved


def _artifact_paths(fmt, date_str, directory=None, staging=".new"):
    """Where the download artifact for `fmt` is published, and staged. The
    audio reading's rewrite stages under its own suffix (#1182).
    """
    import list as list_mod
    final = os.path.join(directory or config.LOCAL_LIST_DIR,
                         list_mod.list_artifact_name(fmt, date_str))
    return final, final + staging


def _write_text_artifact(tmp_path, members):
    """Both lists as one text file.

    A plain .txt can only be one file where the .zip is two, so the album
    section is appended to the file list rather than dropped. The format is a
    choice about packaging; it is not a request to hand out less. This is a
    copy and not the master index itself precisely because the two must not be
    the same file - see list.FULL_LIST_MARKER.

    THE OPERATOR'S BANNER APPEARS ONCE, NOT ONCE PER SECTION

    Each source list carries its own banner, which is right when they are
    downloaded separately - as .zip and .rar hand them out, and as the !rar
    list is served on its own. Concatenated, that put the operator's ASCII art
    in the middle of the file as well as at the top, which reads as a bug
    rather than as a design.

    The identity line deliberately still repeats. It is a section header - the
    album half of this file should say what is serving it too, and one line is
    not noise. The banner can be any height, which is the difference.

    Removed by matching the exact text read_operator_header() returned rather
    than by recognising a banner in the output, because only the first is
    knowable: the banner is free-form and could otherwise be anything,
    including something that looks like a folder heading.
    """
    banner = read_operator_header()
    # Exactly as generate_master_list() wrote it: a leading blank line, the
    # banner, then the newline ending its last line.
    banner_block = "\n" + banner + "\n" if banner else ""

    with io.open(tmp_path, "w", encoding="utf-8", newline="\n") as out:
        for index, (source, _name) in enumerate(members):
            if index:
                out.write("\n\n")
            _copy_removing_first(source, out,
                                 banner_block if index else "")


# A megabyte at a time. Large enough that the read count is irrelevant next to
# the work of writing, small enough to be noise beside what the scan is
# already holding.
_COPY_CHUNK_CHARS = 1 << 20


def _copy_removing_first(source, out, strip=""):
    """Copy `source` into `out`, removing the FIRST occurrence of `strip`.

    In chunks (#463). This was `handle.read()` followed by
    `text.replace(strip, "", 1)`, and the file it reads is a list - several
    hundred megabytes on a library big enough for the operator to have chosen
    the txt format in the first place, held whole in memory on top of
    everything the rebuild is already holding.

    THE OVERLAP IS WHAT KEEPS IT FAITHFUL. A chunk boundary can fall inside
    the text being removed, so the last len(strip) - 1 characters of each
    chunk are held back and carried into the next rather than written out.
    Without that, a banner straddling a boundary would be written through -
    and the result would be a list with the operator's banner repeated in the
    middle of it, which is the exact thing this removal exists to prevent.
    """
    pending = ""
    with io.open(source, encoding="utf-8") as handle:
        while True:
            chunk = handle.read(_COPY_CHUNK_CHARS)
            if not chunk:
                break
            if not strip:
                out.write(chunk)
                continue
            pending += chunk
            at = pending.find(strip)
            if at >= 0:
                out.write(pending[:at] + pending[at + len(strip):])
                pending = ""
                strip = ""
                continue
            carry = len(strip) - 1
            if carry:
                out.write(pending[:-carry])
                pending = pending[-carry:]
            else:
                out.write(pending)
                pending = ""
    if pending:
        out.write(pending)


def _write_zip_artifact(tmp_path, members):
    """Store the temp files under their FINAL names, so the archive users
    download is identical to what it always was."""
    # One more heartbeat before the longest silent step in the run.
    # Deflating a list this size is minutes on a machine that has just
    # walked 80 TB, and the daemon's stall watch has nothing else to go
    # on until it finishes - see commands.run_watching_for_a_stall().
    write_progress("packing", force=True)
    with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zipf:
        for source, name in members:
            zipf.write(source, arcname=name)


_STAGING_PREFIX = ".listpack-"


# How often a running scan may write its progress file. A 719k-file library
# would otherwise spend a meaningful part of the run serialising JSON nobody
# read - the dashboard polls every couple of seconds, so anything finer is
# work done for no reader.
PROGRESS_WRITE_SECONDS = 0.5

_progress_last_write = [0.0]

# WHEN THIS RUN BEGAN, stamped once and repeated in every write.
#
# The dashboard cannot work it out for itself: update_list.py is a SUBPROCESS,
# so the only thing the two share is this file, and the daemon may have been
# restarted - or the page opened - long after the rebuild started. Deriving it
# from when the file first appeared would be wrong for the same reason, since
# the file survives the run that wrote it.
#
# Module scope, not per call: a rebuild is one process, so "when did this
# process start" is the honest answer to "how long has this been running", and
# it cannot drift as the run proceeds.
_started_at = time.time()


def progress_path():
    """Where a rebuild reports what it is doing. Resolved per call, like every
    other configurable path here - !rehash reloads config."""
    return getattr(config, "LIST_PROGRESS_FILE", os.path.join("data", "list_progress.json"))


def write_progress(phase, folder="", folder_index=0, folder_count=0,
                   files=0, force=False, rate=None):
    """Report the current step, for the dashboard to poll.

    NEVER RAISES AND NEVER BLOCKS THE SCAN. This is a progress bar: a full
    disk, a read-only data/ or a permissions problem must cost the operator
    the bar, not the list rebuild they actually asked for.

    Throttled to PROGRESS_WRITE_SECONDS, except when `force` marks a step an
    operator would notice missing - the start, each folder, and the end.

    `token` is the one the bot gave this run (#1182, RUN_TOKEN_ENV): the bot
    hands a rebuild over to the background reading only when THIS child
    reports "reading", never on a file a stopped reading left behind - by the
    token and not the pid, which a launcher such as a Windows venv's
    python.exe does not share with the script. `pid` is for display. `rate`
    is the reading's files a second.
    """
    now = time.time()
    if not force and (now - _progress_last_write[0]) < PROGRESS_WRITE_SECONDS:
        return
    _progress_last_write[0] = now
    payload = {"phase": phase, "folder": folder, "folder_index": folder_index,
               "folder_count": folder_count, "files": files, "at": now,
               "started_at": _started_at, "pid": os.getpid(),
               "token": os.environ.get("DCCORE_RUN_TOKEN", "")}
    if rate is not None:
        payload["rate"] = rate
    try:
        path = progress_path()
        directory = os.path.dirname(os.path.abspath(path))
        os.makedirs(platform_compat.long_path(directory), exist_ok=True)
        # Written whole and renamed into place: the dashboard polls this
        # while it is being written, and half a JSON object is a parse error
        # every couple of seconds rather than a progress bar.
        temp = path + ".new"
        with io.open(platform_compat.long_path(temp), "w", encoding="utf-8") as handle:
            json.dump(payload, handle)
        platform_compat.replace_with_retry(temp, path)
    except Exception:
        pass


# The phases in which the published list is untouched (#923): the scan and
# the writing work on temporary names, and so do the background audio reading
# and the rewrite of its rows (#1182) - "reading" and "rewriting". Anything
# else - "packing" and "publishing", which cover building the archive and the
# swap, or no phase at all - is a moment searches and file requests wait out.
PHASES_BEFORE_THE_SWAP = ("scanning", "writing", "reading", "rewriting")

# The phase of the background audio reading (#1182), after the list is
# published. The daemon hands a rebuild over to it on seeing this.
READING_PHASE = "reading"


def read_progress():
    """The progress file as written, or None when there is nothing readable."""
    try:
        with io.open(platform_compat.long_path(progress_path()), encoding="utf-8") as handle:
            loaded = json.load(handle)
        return loaded if isinstance(loaded, dict) else None
    except (OSError, ValueError):
        return None


def read_phase():
    """The phase the running rebuild last reported, or None when it reported
    nothing readable. The daemon asks this; the rebuild is another process."""
    loaded = read_progress()
    phase = loaded.get("phase") if loaded else None
    return str(phase) if phase else None


def stop_request_path():
    """The file that asks a background audio reading to stop (#1182). A file
    rather than a signal: it means the same on Windows, and it reaches a
    reading whichever process started it."""
    return progress_path() + ".stop"


def request_stop():
    """Ask the reading to stop. True when the request was written."""
    try:
        with io.open(platform_compat.long_path(stop_request_path()), "w", encoding="utf-8") as handle:
            handle.write("stop\n")
        return True
    except OSError:
        return False


def stop_requested():
    return os.path.exists(platform_compat.long_path(stop_request_path()))


def clear_stop_request():
    try:
        os.remove(platform_compat.long_path(stop_request_path()))
    except OSError:
        pass


def clear_progress():
    """Drop the progress file once a run is over.

    A leftover from a killed process would otherwise read as a rebuild that
    is still going, and the dashboard would show a bar that never moves.
    """
    try:
        os.remove(platform_compat.long_path(progress_path()))
    except OSError:
        pass


def _discard_stale_temps(directory=None):
    """Remove ".new" staging files a previous run was killed in the middle of.

    Only names the builder itself stages: "<LIST_BASE_NAME>-...new", in the
    lists directory. Nothing reads these - find_latest_list() globs "*.txt"
    and is_list_artifact() ends in the real extension, so a leftover has never
    been served or counted - which is exactly why nothing removed them either.

    Called at the START of a build, where this run has staged nothing yet, so
    the only file it can reach belongs to a run that is no longer alive. (Two
    builds running at once already write the SAME temp paths as each other, so
    this adds no hazard that concurrency does not already have.)
    """
    directory = directory or config.LOCAL_LIST_DIR
    base = str(getattr(config, "LIST_BASE_NAME", "") or "")
    if not base:
        return
    try:
        entries = os.listdir(directory)
    except OSError:
        return
    removed = 0
    for name in entries:
        if not (name.startswith(base + "-") and name.endswith(".new")):
            continue
        try:
            os.remove(os.path.join(directory, name))
            removed += 1
        except OSError as err:
            print(f"[LIST-CLEAN ERROR] Could not remove {name}: {err}")
    if removed:
        print(f"[LIST-CLEAN] Removed {removed} leftover staging file(s) "
              f"from an interrupted run.")


def _discard_stale_staging(directory=None):
    """Remove staging directories a previous run was killed in the middle of."""
    directory = directory or config.LOCAL_LIST_DIR
    try:
        entries = os.listdir(directory)
    except OSError:
        return
    for name in entries:
        if not name.startswith(_STAGING_PREFIX):
            continue
        path = os.path.join(directory, name)
        if os.path.isdir(path):
            shutil.rmtree(path, ignore_errors=True)


def _write_rar_artifact(tmp_path, members, directory=None):
    """Pack with the rar binary. False if it could not be done, with the reason.

    rar has no equivalent of zipfile's arcname, so the members are copied to
    their published names in a scratch directory first - packing the temporary
    files directly would put "DCCore-2026-08-30.txt.new" inside the archive and
    every reader would see the name the build happened to use that run.

    The scratch directory sits inside LOCAL_LIST_DIR so the finished archive
    lands on the same filesystem: os.replace is only atomic within one.
    """
    rar_bin = platform_compat.rar_command(getattr(config, "RAR_BINARY", None))
    if not rar_bin:
        print("[LIST-GEN] LIST_FORMAT is 'rar' but no rar binary was found - "
              "set RAR_BINARY or put rar on PATH. Packing a .zip instead.")
        return False

    # The finally below removes this run's directory. A daemon killed between
    # mkdtemp and that finally leaves one behind, and nothing else in lists/
    # ever looks at directories, so it would sit there until somebody noticed.
    _discard_stale_staging(directory)
    directory = directory or config.LOCAL_LIST_DIR
    staging = tempfile.mkdtemp(prefix=_STAGING_PREFIX, dir=directory)
    try:
        staged = []
        for source, name in members:
            target = os.path.join(staging, name)
            shutil.copyfile(source, target)
            staged.append(os.path.abspath(target))

        # rar names the archive itself when the given name has no extension,
        # so it is built inside the staging directory under a name it cannot
        # reinterpret and moved into place afterwards.
        built = os.path.join(staging, "list.rar")
        # -ep stores the names with no path at all, which is the arcname
        # behaviour above. A list of arguments and never a shell, and a
        # timeout, both for the same reasons dcc.py packs albums that way: a
        # hung rar here would wedge every list rebuild after it.
        cmd = [rar_bin, "a", "-ep", "-w" + os.path.abspath(directory),
               os.path.abspath(built)] + staged
        # utf-8/replace for the same reason as dcc.py's rar call.
        result = subprocess.run(cmd, capture_output=True, text=True,
                                encoding="utf-8", errors="replace",
                                timeout=getattr(config, "RAR_TIMEOUT", 1800),
                                **platform_compat.no_console_window())
        if result.returncode == 0 and os.path.exists(built):
            platform_compat.replace_with_retry(built, tmp_path)
            return True
        detail = (result.stderr or result.stdout or "").strip().splitlines()
        print(f"[LIST-GEN] rar exited {result.returncode} packing the list - "
              f"packing a .zip instead. {detail[-1] if detail else ''}")
        return False
    except (OSError, ValueError, subprocess.SubprocessError) as err:
        print(f"[LIST-GEN] Could not pack the list with rar ({err}) - "
              f"packing a .zip instead.")
        return False
    finally:
        shutil.rmtree(staging, ignore_errors=True)


def build_list_artifact(fmt, members, date_str, directory=None, staging=".new"):
    """Build what a user downloads. Returns (fmt_used, tmp_path, final_path).

    `fmt_used` is not always `fmt`: "rar" falls back to "zip" when the binary
    is missing or the pack fails. That is deliberate. The list is the whole
    point of the bot, and refusing to publish one over a packaging preference
    would take it off the air; the fallback is loud, and the operator still
    has something to serve while they sort the binary out.
    """
    final, tmp = _artifact_paths(fmt, date_str, directory, staging)
    if fmt == "rar":
        if _write_rar_artifact(tmp, members, directory):
            return fmt, tmp, final
        fmt = "zip"
        final, tmp = _artifact_paths(fmt, date_str, directory, staging)

    if fmt == "txt":
        _write_text_artifact(tmp, members)
    else:
        _write_zip_artifact(tmp, members)
    return fmt, tmp, final


def list_nick():
    """The nick every request line in the list names: the CONFIGURED one.

    A list outlives the connection it was built on - people save it and paste
    from it days later - so it must name the nick the bot comes back to, not
    whichever one it happens to hold (#376). On its alternate nick the bot is
    config.NICKNAME = the alternate; irc.py keeps what was configured in
    ORIGINAL_NICK, answers "!<ORIGINAL_NICK> <file>" all the same
    (get_bot_aliases()), and takes the name back as soon as it is free.

    Every rebuild today runs this module as a subprocess, which reads the
    configured nick fresh and never sees the daemon's live one - so this held
    by accident of how the rebuild is launched. Built in-process, the list was
    stamped with the alternate. ORIGINAL_NICK first makes it hold either way;
    a subprocess, where irc.py never ran, has no ORIGINAL_NICK and takes
    NICKNAME, which there is the configured value.
    """
    return str(getattr(config, "ORIGINAL_NICK", None) or getattr(config, "NICKNAME", "") or "")


def list_identity_line(nickname=None):
    """"Served by <nick> - <version> - <url>", with absent parts left out.

    Shared by the .txt and the !rar list so the two cannot drift, and built by
    joining on what is actually present: an operator who blanks SCRIPT_VERSION
    or PROJECT_URL should get a shorter line, never a dangling " - " with
    nothing after it.

    The fallbacks here are empty strings rather than a literal like "DCCore".
    A non-empty fallback would be a second opinion about a value config.py
    already declares, and which one won would depend on how the value happened
    to be read - see tests/test_config_fallbacks.py, which enforces exactly
    that. An empty fallback declines to have an opinion instead.
    """
    if nickname is None:
        nickname = list_nick()

    parts = [str(getattr(config, "SCRIPT_VERSION", "") or "").strip(),
             str(getattr(config, "PROJECT_URL", "") or "").strip()]
    detail = " - ".join(part for part in parts if part)

    return f"Served by {nickname} - {detail}" if detail else f"Served by {nickname}"


def read_operator_header(path=None, max_bytes=None, log=print):
    """The operator's own banner for the top of the list, or "" if there is none.

    Free-form: several lines, ASCII art, a channel name - whatever should greet
    whoever opens the file. Everything else written into the list goes through
    _one_line(), which flattens control characters; doing that here would
    destroy the art this exists to carry, so the banner deliberately bypasses
    it.

    TWO LINE SHAPES ARE NEUTRALISED, because the banner is written into the
    body of the master index and the index has a grammar of its own. Found by
    audit, both with the banner working exactly as documented:

      A line that is entirely "=" is a folder RULE to every reader. A banner
      with an odd number of them leaves list.find_matching_entries()'s
      rule/heading/rule machine mid-block, so the next banner line is taken as
      a folder heading and the REAL heading after it is swallowed - every file
      in the first folder is then attributed to a folder that does not exist.
      That value is not cosmetic: the dashboard's duplicate-finder and the
      File Lists view both resolve it. Rewritten to the same width in "-", so
      the box still looks like a box.

      A line beginning with "!" is a request ROW to every reader - the count in
      the advert, the CTCP SLOTS payload, and @find. A banner line like
      "!!! NEW RELEASES !!!" was counted as a file and returned as a genuine
      search hit, so a user could request it and receive nothing. Dropped
      rather than rewritten: there is no edit that keeps such a line looking
      like itself while stopping it being read as a row.

    Both are reported with the offending line, so the operator can change the
    banner rather than wonder what happened to it. Neutralised HERE rather
    than at the point of writing because _split_master_list() removes the
    banner from each split part by matching the exact text this function
    returned - so both sides have to be the same text.

    "" on anything unreadable, matching read_list_base_marker()'s reasoning: a
    banner is decoration, and no decoration should ever stop a list being
    built. A missing file is the normal state, not an error.

    Line endings are normalised to "\\n" because the list is opened without
    newline="", so the writer translates "\\n" to os.linesep itself. A CRLF
    banner written through that untouched would come out "\\r\\r\\n" on Windows.
    """
    if path is None:
        path = getattr(config, "LIST_HEADER_FILE", "./data/list_header.txt")
    if max_bytes is None:
        max_bytes = getattr(config, "LIST_HEADER_MAX_BYTES", 8192)

    try:
        with io.open(platform_compat.long_path(path), "rb") as handle:
            raw = handle.read(int(max_bytes) + 1)
    except (OSError, ValueError, TypeError):
        return ""

    if not raw:
        return ""

    truncated = len(raw) > int(max_bytes)
    if truncated:
        raw = raw[:int(max_bytes)]
        print(f"[LIST-GEN] {path} is larger than {max_bytes} bytes; "
              f"the banner was truncated.")

    # errors="replace": this is operator-supplied and may be in any encoding.
    # A banner in the wrong code page should come out looking odd, not abort
    # the list build.
    text = raw.decode("utf-8", errors="replace")

    # Normalise line endings, drop trailing blank lines so this function owns
    # the spacing around the banner rather than inheriting whatever the file
    # happened to end with.
    text = "\n".join(text.splitlines()).rstrip("\n")
    return _neutralise_banner(text, log=log)


def _neutralise_banner(text, log=print):
    """The banner with any line the index's own grammar would misread made
    safe. See read_operator_header()'s docstring for what and why.

    Returns the text unchanged when there is nothing to do, which is the
    overwhelmingly common case - most banners are prose and ASCII art that
    happens not to be a bare rule.
    """
    if not text:
        return text

    safe = []
    for line in text.split("\n"):
        stripped = line.strip()

        # A row, to every reader in the project.
        if stripped.startswith("!"):
            log(f"[LIST-GEN] A banner line starting with '!' was left out of "
                f"the list: every reader counts it as a file, and @find would "
                f"return it as a match nobody can download. The line was: "
                f"{stripped[:60]!r}")
            continue

        # A folder rule, to every reader in the project.
        if stripped and set(stripped) == {"="}:
            log(f"[LIST-GEN] A banner line of only '=' was rewritten with '-': "
                f"it reads as a folder rule and would shift every folder "
                f"heading after it by one. The line was {len(stripped)} "
                f"characters wide.")
            safe.append(line.replace("=", "-"))
            continue

        safe.append(line)

    return "\n".join(safe).rstrip("\n")


def generate_master_list(list_name=None, reading_jobs=None):
    """Scan one list's folders, clear its old files first, and build its lists.

    `reading_jobs`, a list, collects what is left for the background audio
    reading once this list is published (#1182): an AudioReadingJob when
    LIST_SHOW_AUDIO_INFO is on and new or changed audio files were listed.

    `list_name` picks which list to build; without one this is the primary,
    which on a single-list install is the only one there is - so every existing
    caller, and the whole of this function's previous behaviour, is unchanged.

    EVERY PATH BELOW COMES FROM `directory`, not from config.LOCAL_LIST_DIR.
    That is the single change that makes more than one list possible: for the
    primary the two are the same string, and for any other list it is a
    subdirectory. See list.list_dir() for why the list is in the path and not
    in the filename.
    """
    import os
    import sys
    import time
    import datetime
    import zipfile
    import re
    import defaults as config
    import db
    import library
    import list as list_mod

    directory = list_mod.list_dir(list_name)
    if not os.path.exists(directory):
        os.makedirs(directory)

    today = datetime.datetime.now().strftime("%Y-%m-%d")
    txt_filename = f"{config.LIST_BASE_NAME}-{today}.txt"
    rar_filename = f"{config.LIST_BASE_NAME}-RAR-{today}.txt"
    # Same shape as the album list's name, and it has to be: list.py's
    # find_latest_list() globs "<base>-*.txt" and takes the LAST one, so this
    # file sorts after the master list and would silently become "the" list
    # that @find searches and the advert counts. VIDEO_LIST_MARKER is how that
    # function knows to leave it alone - the same guard "-RAR-" already needs.
    video_filename = f"{config.LIST_BASE_NAME}-{list_mod.VIDEO_LIST_MARKER}-{today}.txt"

    txt_path = os.path.join(directory, txt_filename)
    rar_path = os.path.join(directory, rar_filename)
    video_path = os.path.join(directory, video_filename)
    
    # config.RAR_ENABLED (#140) refuses every !rar request. Building an album
    # list anyway, and shipping it inside the zip every user downloads, hands
    # people a file whose every line is an instruction to use a command this
    # bot will refuse - they paste one, get "Folder packing is disabled", and
    # reasonably conclude the bot is broken rather than that the feature is
    # off. Raised on #140 and left as a follow-up there.
    serve_albums = bool(getattr(config, "RAR_ENABLED", True))
    if not serve_albums:
        print("[LIST-GEN] RAR_ENABLED is off - skipping the album list entirely.")

    SIZE_FILE_PATH = os.path.join(directory, config.LIST_SIZE_FILE)
    RAWBYTES_FILE_PATH = os.path.join(directory, config.LIST_RAWBYTES_FILE)

    scan_start = time.time()

    # The old lists are NOT deleted here any more. This ran before the scan, so an NFS mount
    # that went away, a disk that filled, or any exception below left the daemon with no index
    # at all: find_latest_list() returns None, @find answers "No MasterList found" and the
    # advert publicly announces 0 files, until someone notices and runs a successful !update.
    #
    # The new lists are written to temporary names alongside the real ones and swapped in only
    # once the whole generation has succeeded (see finalise below). A failed run now leaves the
    # previous index exactly as it was.
    tmp_txt_path = txt_path + ".new"
    tmp_rar_path = rar_path + ".new"
    tmp_video_path = video_path + ".new"
    # Every format's temporary, not just the configured one: a run that asked
    # for .rar and fell back to .zip has staged two, and the guards below have
    # no way of knowing which without repeating the decision.
    tmp_index_paths = (tmp_txt_path, tmp_rar_path, tmp_video_path)
    tmp_artifact_paths = tuple(_artifact_paths(f, today, directory)[1]
                               for f in list_mod.LIST_FORMATS)
    tmp_all_paths = tmp_index_paths + tmp_artifact_paths
    # Before anything of THIS run exists. A run that fails discards its own
    # temporaries; a run that is KILLED - the machine goes down mid-scan,
    # which on a large library is a window of minutes - cannot, and those
    # files carry the date they were staged on, so the next day's run stages
    # different names and never touches them again. They accumulate one set
    # per killed run, for ever, in the directory the operator looks at to see
    # whether their lists are being built.
    _discard_stale_temps(directory)

    scan_folders = library.folders(list_name)
    print("[LIST-GEN] Scanning the library in "
          + ", ".join(f"{f.path} ({f.name})" for f in scan_folders)
          + f", {scan_workers()} folder(s) at a time...")

    all_files_data = []
    # The music list, the film-and-series list, and which folders !rar may
    # pack. All three filled by the one walk below - the !rar album list has
    # been built from the same pass since long before this.
    video_files_data = []
    packable_folders = set()
    total_bytes = 0

    # A single-folder leftover used to sit here: `scan_root` built from
    # config.FILE_DIRECTORY and then immediately overwritten inside the
    # per-folder loop below. Dead since #164, and reading a setting that is
    # only the FALLBACK for an install with no folder list - the same
    # confusion an operator raised about it still sitting on the Settings page.
    # The real scan root, and why it is long_path()-wrapped, are in the
    # loop.

    # Every subtree os.walk() could not read (a stale NFS handle, EIO, a
    # revoked ACL) used to be skipped in total silence under the default
    # onerror=None - total_files_count came out non-zero, so the zero-files
    # guard further down never caught it, and a truncated index was
    # published over the previous good one. Collected here and checked
    # once the walk finishes, so one bad subtree costs the whole run
    # (keeping the previous index) rather than a silent partial one.
    walk_errors = []
    denied_dirs = []
    # Folders left out because their own name cannot survive the list format
    # (#464) - see has_backslash_component() at the point it is used.
    unlistable_dirs = []

    def _on_walk_error(err):
        # TWO DIFFERENT FAILURES, AND ONLY ONE OF THEM IS WORTH ABORTING FOR
        # (#441).
        #
        # A directory that is PERMANENTLY unreadable - a Windows volume root's
        # System Volume Information, a POSIX lost+found, anything whose ACL
        # excludes the account the daemon runs as - fails identically on every
        # future scan. Treating it as "keep the previous index" meant the list
        # could never be rebuilt again on that install: every !update ran the
        # whole scan and then threw the result away, for ever, and the log line
        # said only that part of the library could not be read.
        #
        # A directory that has VANISHED mid-scan is the opposite: it means the
        # library changed underneath us, the scan we just did is of a state
        # that no longer exists, and publishing it would be publishing a
        # snapshot known to be wrong. That one still aborts.
        #
        # The test is whether the path is still there. Permission denied on a
        # directory that exists is a fact about the ACL, not about the scan.
        denied = isinstance(err, PermissionError)
        if denied:
            try:
                denied = os.path.isdir(err.filename)
            except Exception:
                denied = False

        if denied:
            denied_dirs.append(err.filename)
            print(f"[LIST-GEN] Skipping {err.filename!r}: permission denied, and "
                  f"it is still there - excluded from the list rather than "
                  f"failing the whole rebuild.")
            return

        walk_errors.append(err)
        print(f"[LIST-GEN ERROR] Could not read {err.filename!r} during the scan: {err}")

    # Once, here - see is_listed_file()'s note. Resolving these inside the
    # walk asked the question per file rather than per scan, and a 719k-file
    # library would have rebuilt each tuple 719,000 times.
    ignored = ignored_extensions()
    video_exts = video_extensions()
    companion_exts = video_companion_extensions()
    packable_exts = rar_extensions()
    split_video = bool(getattr(config, "SEPARATE_VIDEO_LIST", True))

    print("[LIST-GEN] Indexing every file"
          + (f", except {len(ignored)} type(s): {', '.join(ignored)}"
             if ignored else " - no extensions are being skipped")
          + ".")
    if split_video:
        print(f"[LIST-GEN] Film and series go in their own list "
              f"({len(video_exts)} extension(s)).")
    else:
        print("[LIST-GEN] One combined list - SEPARATE_VIDEO_LIST is off.")

    write_progress("scanning", folder_count=len(scan_folders), force=True)

    # Duration and quality on the file rows (#567), when asked for. The cache
    # is what keeps it affordable: only files new or changed since the last
    # rebuild are opened. Unopenable, it says so and the list is size-only.
    audio = None
    if getattr(config, "LIST_SHOW_AUDIO_INFO", False):
        # THE LIST'S OWN NAME, however the rebuild was asked for (#979). A
        # lone list is built with no name and each of several by its name, so
        # the primary's scope was "" until a second list was added and its
        # name after: its whole cache missed at once - re-read within
        # LIST_AUDIO_INFO_MINUTES, the rest size-only for rebuilds - and the
        # "" rows were never pruned. Only the primary was ever built as "",
        # so it takes those rows over.
        primary = library.primary_list().name
        scope = list_name or primary
        audio = audio_info.Cache.open(scope=scope, formerly="" if scope == primary else None)

    for folder_number, scan_folder in enumerate(scan_folders, start=1):
        # Reported per folder because the folder COUNT is the one total known
        # before the walk starts - a file total would need a full pass to
        # find, which is the work being measured. So the bar advances a
        # folder at a time and the file counter shows life in between.
        write_progress("scanning", folder=scan_folder.name,
                       folder_index=folder_number,
                       folder_count=len(scan_folders),
                       files=len(all_files_data) + len(video_files_data),
                       force=True)

        # A folder that is not there RIGHT NOW is skipped, not fatal: an
        # unplugged drive or an unmounted share should cost its own contents,
        # not take the whole list - and the bot - off the air. Deliberately
        # different from walk_errors below, which is a subtree going unreadable
        # DURING a walk of a folder that was present: that is a systemic
        # failure of a library we are meant to be reading, and it keeps the
        # previous index rather than publishing a truncated one.
        #
        # Every folder missing is still caught, further down: the zero-files
        # guard refuses to publish an empty list.
        if not os.path.isdir(platform_compat.long_path(scan_folder.path)):
            print(f"[LIST-GEN] Skipping {scan_folder.name} - "
                  f"{scan_folder.path} is not available right now. The list "
                  f"is being built without it.")
            continue

        # long_path()-wrapped, and the SAME wrapped value is relpath()'s base
        # below - see the note above scan_root's original single-folder form.
        scan_root = platform_compat.long_path(scan_folder.path)

        # walk_with_sizes(), not os.walk: the size comes back with the name
        # rather than being asked for again per file. See its docstring for
        # the measurement. Traversal ORDER differs from os.walk's and cannot
        # matter - all_files_data is sorted by (folder, filename) below before
        # anything is written.
        for root, files in walk_with_sizes(scan_root, onerror=_on_walk_error):
            # Throttled inside write_progress(), so this costs a clock read
            # per directory rather than a file write.
            write_progress("scanning", folder=scan_folder.name,
                           folder_index=folder_number,
                           folder_count=len(scan_folders),
                           files=len(all_files_data) + len(video_files_data))

            # ONCE PER DIRECTORY, NOT ONCE PER FILE (#463).
            #
            # rel_dir is a property of the directory - it was computed inside
            # the per-file loop below, so a folder of 20 tracks paid 20
            # relpath() calls and 20 join() calls for the same answer, and
            # stored 20 separate equal strings in all_files_data. Hoisted, the
            # rows share one object: 149.9 B/file against 206.9 B/file
            # measured, about 0.29 GiB at 5.4M files, and the calls go from
            # one per file to one per directory.
            #
            # The folder's LABEL leads every path (#164). Two folders can hold
            # the same relative path - the same album in flac and in mp3 is
            # the ordinary case, not a corner one - and without the label
            # their headings would be identical text that resolution could not
            # tell apart.
            #
            # Written for a single folder too. Labelling only at two or more
            # would mean an operator who adds a second folder after weeks of
            # serving changes every path anyone already saved; doing it once,
            # at the upgrade, is one break instead of two.
            #
            # relative_folder() rather than os.path.relpath(): the same
            # answer, sliced off the string when it safely can be (#1138).
            rel_dir = relative_folder(root, scan_root)
            if rel_dir == ".":
                rel_dir = ""
            rel_dir = (os.path.join(scan_folder.name, rel_dir)
                       if rel_dir else scan_folder.name)

            # A LITERAL BACKSLASH IN A FOLDER NAME IS NOT A SEPARATOR (#464).
            #
            # On the Linux box this daemon runs on, a backslash is an ordinary
            # filename character, and unzipping a Windows-made archive
            # produces one routinely. The list writes its headings with
            # backslashes as separators, and list.list_heading_parts() reads
            # them back by splitting on both separators - so a directory named
            # "Rock\Metal" is written as ...\music\Rock\Metal\ and read back as
            # ["music", "Rock", "Metal"]. If music/Rock/Metal exists, every
            # file under it - and its !rar row - resolves into that unrelated
            # folder and the requester silently gets the wrong album. If it
            # does not, nothing under the real folder can ever be served.
            #
            # Excluded rather than escaped. The list format is read by other
            # bots and by AutoQ, so an escape convention would have to be
            # understood by readers that already exist and never will be.
            # library.problems() already refuses a backslash in a folder LABEL
            # for exactly this reason ("the label travels further than the
            # machine that made it"); this is the same rule applied to the
            # components below it. A folder that cannot be named in the list
            # cannot be served from it, so listing it would advertise
            # something unreachable.
            if has_backslash_component(rel_dir):
                # No dedup needed: the walk visits each directory once, so
                # rel_dir is new every time it gets here.
                unlistable_dirs.append(rel_dir)
                continue

            # Decided once per folder, before the per-file loop: a companion
            # file's list depends on whether a video sits beside it (#411),
            # and that must not depend on which file the walk handed over
            # first. is_listed_file() is not consulted here on purpose - a
            # video the operator has ignored still says what kind of folder
            # this is.
            #
            # EACH NAME IS LOWER-CASED ONCE (#1138). is_listed_file(),
            # is_packable_file(), belongs_in_video_list() and
            # audio_info.is_audio() each lower-cased the name again - up to
            # five times a file - and rebuilt a tuple of the extensions each
            # time. The checks below are those helpers' own, made on `low`
            # against the tuples resolved once per scan; an empty tuple
            # matches nothing, as it did there. The helpers stay, for every
            # other caller. A folder's packable flag is likewise added once,
            # after its files, instead of once per packable file.
            folder_has_video = split_video and any(
                str(name).lower().endswith(video_exts) for name, _bytes in files)
            packable_here = False

            # Keep every track under its exact, complete path on disk
            for file, file_bytes in files:
                low = str(file).lower()
                if not low.endswith(ignored):
                    if file_bytes is None:
                        full_file_path = os.path.join(root, file)
                        # #228: a bare `except: pass` left file_bytes at 0 and
                        # the entry was published anyway - permission denied, a
                        # dangling symlink, or a file removed mid-scan all read
                        # as a legitimate 0-byte track. The list is the thing
                        # the bot HANDS OUT: publishing it meant offering a
                        # download that can only ever fail once someone
                        # actually requests it, with the library's reported
                        # total size quietly short and no log line saying why.
                        # Excluded and logged instead - one bad file costs
                        # itself, not the whole scan (walk_errors above is for
                        # a whole SUBTREE going unreadable, a more systemic
                        # failure than one file).
                        print(f"[LIST-GEN ERROR] Skipping {full_file_path!r}, "
                              f"its size could not be read.")
                        continue
                    total_bytes += file_bytes

                    # WHICH folder may be packed, decided per FILE and
                    # remembered per folder. A folder earns its !rar row from
                    # holding something worth packing, not from holding
                    # anything at all - see RAR_EXTENSIONS.
                    if low.endswith(packable_exts):
                        packable_here = True

                    # WHICH list the row goes in. With the split off, video
                    # lands in the same list as everything else, which is the
                    # behaviour this had before the setting existed.
                    if split_video and (low.endswith(video_exts) or (
                            folder_has_video and low.endswith(companion_exts))):
                        video_files_data.append((rel_dir, file, file_bytes))
                    else:
                        all_files_data.append((rel_dir, file, file_bytes))
                        if audio is not None and low.endswith(audio_info.AUDIO_EXTENSIONS):
                            # No request here: an unchanged file is answered
                            # from the cache by its size, the rest are read
                            # after the walk, many at once (#914). The path
                            # goes as (root, name), joined only for a file
                            # that is read - on an unchanged library that is
                            # almost none of them (#1138).
                            audio.note(audio_info.row_key(rel_dir, file),
                                       (root, file), file_bytes)
            if packable_here:
                packable_folders.add(rel_dir)

    if walk_errors:
        print(f"[LIST-GEN ERROR] {len(walk_errors)} part(s) of the library could not be "
              "read - keeping the previous index rather than publishing a truncated one.")
        if audio is not None:
            audio.close()
        return False

    # The audio files new or changed since the last rebuild (#567, #914) are
    # NOT read here any more (#1182). The list is written with what the cache
    # already knows - a file not read yet shows its size alone - and they are
    # read after every list is published, in the background, by
    # run_audio_reading(). Nothing waits for them, and nothing is paused.

    if denied_dirs:
        # Said once, with the count, because it is a standing condition rather
        # than an event: these folders will be missing from every list until
        # the operator changes their permissions, and an operator comparing
        # the file count against their own expectation deserves to know why
        # it is short.
        print(f"[LIST-GEN] {len(denied_dirs)} folder(s) were excluded because "
              f"the daemon cannot read them. They are not in this list and "
              f"will not be in the next one until their permissions change: "
              f"{', '.join(repr(p) for p in denied_dirs[:5])}"
              f"{' ...' if len(denied_dirs) > 5 else ''}")

    if unlistable_dirs:
        # Same shape and same reasoning as denied_dirs above: a standing
        # condition rather than an event, and an operator comparing the file
        # count against their own expectation deserves to know why it is
        # short.
        print(f"[LIST-GEN] {len(unlistable_dirs)} folder(s) were excluded "
              f"because a literal backslash in the name cannot be told from a "
              f"path separator once the list is written - anything under them "
              f"would resolve to a different folder or to nothing. Rename "
              f"them to list them: "
              f"{', '.join(repr(p) for p in unlistable_dirs[:5])}"
              f"{' ...' if len(unlistable_dirs) > 5 else ''}")

    # Sort by the real folder and file names
    all_files_data.sort(key=lambda x: (str(x[0]).lower(), str(x[1]).lower()))
    # THE SAME ORDER, FOR THE SAME REASON (#443). all_files_data has been
    # sorted here since the beginning; video_files_data is built by the same
    # walk and was only ever appended to, so the film and series list came out
    # in whatever order the filesystem handed the directories over - which is
    # reverse-ish on NTFS and arbitrary elsewhere. A reader scanning for a
    # title has no way to guess where it is.
    video_files_data.sort(key=lambda x: (str(x[0]).lower(), str(x[1]).lower()))

    total_files_count = len(all_files_data)
    scan_end = time.time()
    elapsed_seconds = scan_end - scan_start
    if elapsed_seconds <= 0:
        elapsed_seconds = 0.1

    files_per_second = int((total_files_count + len(video_files_data))
                           / elapsed_seconds)
    def format_total_size(b):
        for unit in ['B','KB','MB','GB','TB']:
            if b < 1024.0: return f"{b:.2f}{unit}"
            b /= 1024.0
        return f"{b:.2f}PB"
    def format_size_human(b):
        return format_total_size(b)

    # The WHOLE library, which is what the side files and the advert want.
    formatted_size = format_total_size(total_bytes)
    # This list's own share of it. The header below describes this file, and
    # a music list announcing a size that includes films it does not contain
    # is a number no reader can reconcile with what they are looking at.
    music_bytes = sum(size for _folder, _name, size in all_files_data)
    formatted_music_size = format_total_size(music_bytes)
    
    time_struct = time.gmtime(elapsed_seconds)
    duration_str = time.strftime("%H:%M:%S", time_struct)

    day = datetime.datetime.now().day
    suffix = "th" if 11 <= day <= 13 else {1: "st", 2: "nd", 3: "rd"}.get(day % 10, "th")
    date_header_str = datetime.datetime.now().strftime(f"{day}{suffix} %b %Y")
    
    # A LIST THAT CANNOT PRODUCE AN ALBUM DOES NOT GET AN ALBUM SECTION.
    #
    # serve_albums is one global read of RAR_ENABLED, applied to every list a
    # build produces. That is right for the switch itself - an operator turning
    # packing off means everywhere - and wrong as the only question asked,
    # because a list can be incapable of packing anything regardless of the
    # switch.
    #
    # A video-only list is the case that showed it. Since #288, a folder earns
    # its !rar row by holding a file in RAR_EXTENSIONS, and those are the audio
    # formats; LIST_VIDEO_EXTENSIONS and RAR_EXTENSIONS share not one entry. So
    # its album list is not "usually empty" - it can never hold a row, for any
    # library, under any configuration.
    #
    # It was still built, and still shipped: the masthead makes the file
    # non-empty, so the `getsize > 0` test below let it into the archive every
    # user downloads. Four dead lines and a heading with nothing behind it,
    # once per download, forever.
    #
    # DECIDED AFTER THE SCAN, not from the extension sets. What actually
    # matters is whether this list produced a packable folder, and the scan has
    # just finished answering that exactly - where reasoning from the
    # configured extensions would be predicting it, and would be wrong for a
    # list whose folders simply hold no albums today and might tomorrow.
    #
    # Turning serve_albums off HERE rather than adding a second flag is what
    # makes this small: every consequence already exists below - the masthead
    # is skipped, no rows are written, the empty temp file fails the size test
    # for the archive, and the `if not serve_albums:` branch removes it instead
    # of publishing it.
    if serve_albums and not packable_folders:
        serve_albums = False
        print("[LIST-GEN] No folder in this list can be packed - skipping the "
              "album list rather than shipping an empty one.")

    # Set the moment the swap completes, and read by the handler at the
    # bottom - which has to tell a failure BEFORE the swap from one
    # after it. Declared out here so the early failures do not meet an
    # unbound name in the very handler meant to report them.
    published = False
    try:
        with open(tmp_txt_path, "w", encoding="utf-8") as f, \
             open(tmp_rar_path, "w", encoding="utf-8") as f_rar:
                 
            f.write(f"List of {total_files_count:,} Files ({formatted_music_size}) generated on {date_header_str} in {duration_str} ( {files_per_second:,} Files Per Second )\n")
            f.write(f"To request a file, copy/paste to the channel... !{list_nick()} FILENAME eg. !{list_nick()} Songname.flac\n")

            # The operator's banner and the bot's identity go BELOW the two
            # lines above and above the first folder - not at the very top.
            # commands.count_from_master_list() reads one readline() and pulls
            # "List of N Files" out of it, so anything inserted above that line
            # silently zeroes the count feeding !update and the channel advert:
            # no exception, no empty file, just a list that advertises nothing.
            #
            # The identity line exists because this file travels. It gets sent
            # to strangers over DCC and reopened weeks later in a text editor
            # with no memory of which bot produced it.
            # The identity line sits directly under the two functional lines,
            # ABOVE the operator's banner. The banner is free-form and can be
            # any height, so putting it first would push the attribution off
            # the reader's first screen exactly on the installs that decorate
            # the most. Fixed position, always visible.
            #
            # Still not line 1: commands.count_from_master_list() does one
            # readline() and regexes "List of N Files" out of it, so anything
            # above that line silently zeroes the count feeding !update and the
            # channel advert - no exception, no empty file, just a list that
            # advertises nothing.
            operator_header = read_operator_header()
            f.write(list_identity_line() + "\n")
            if operator_header:
                f.write("\n" + operator_header + "\n")
            f.write("\n")

            if serve_albums:
                f_rar.write(f"List of Entire Album Folders (!rar) for !{list_nick()} generated on {date_header_str}\n")
                f_rar.write(f"To request an entire album, copy/paste the line... eg. "
                            f"!{list_nick()} !rar {list_mod.LIST_FOLDER_PREFIX}Album\\\n")
                # Same order as the .txt above, and for the same reason. The
                # !rar list is a separate download that travels on its own, so
                # it carries its own copy rather than inheriting one.
                f_rar.write(list_identity_line() + "\n")
                if operator_header:
                    f_rar.write("\n" + operator_header + "\n")
                f_rar.write("="*90 + "\n\n")

            current_folder = None
            written_rar_folders = set()  # Keeps the !rar list free of duplicate rows
            # Per-folder count and size, for the line under each heading
            # (#69). One pass over rows already in memory; the heading is
            # written before its rows, so the total has to be known first.
            music_totals = folder_totals(all_files_data)
            # ONCE PER BUILD, AND ONE WRITE PER FOLDER (#1125). The nick is
            # the same for every row of one build - a rebuild is its own
            # process and nothing changes the config under it - so asking
            # list_nick() per row only cost time. Each folder's heading and
            # rows are gathered in `chunk` and written together: one write()
            # per row was a measurable share of a million-row list.
            row_head = f"!{list_nick()} "
            rar_head = f"!{list_nick()} !rar "
            chunk = []

            for folder, filename, bytes_size in all_files_data:
                if folder != current_folder:
                    current_folder = folder
                    if chunk:
                        f.write("".join(chunk))
                        chunk = []
                    
                    # The text list gets the complete subfolder (e.g. \Digital Media 1\)
                    # The rule is drawn to the width of the folder line it wraps, not
                    # to a fixed 53 characters. Against the real library every one of
                    # the 4,107 folder headers is wider than 53 - they run 54 to 136,
                    # averaging 80 - so the fixed rule never once matched the line it
                    # was framing.
                    #
                    # Measured on _one_line()'s output rather than the raw string:
                    # that is what actually gets written, and it flattens control
                    # characters, which changes the length. _heading_text() is
                    # the one spelling of it, shared with the rewrite of the
                    # length and quality (#1182).
                    folder_line = _heading_text(folder)
                    folder_rule = "=" * len(folder_line)
                    chunk.append(f"\n{folder_rule}\n{folder_line}\n{folder_rule}\n")
                    chunk.append(folder_summary_line(*music_totals[folder], format_size_human) + "\n")
                    
                    # Strip multi-disc suffixes, for the !rar album list ONLY.
                    #
                    # Matched as a whole PATH SEGMENT (split on the same
                    # separators the folder can carry), not a substring
                    # anywhere in the path - a substring match let "\disc"
                    # fire on "\Discography" and "\media" fire on "\Media
                    # Markt Hits", collapsing a real album folder to the
                    # ARTIST root, which dcc.py refuses outright ("Artist
                    # root folders cannot be requested"). The album then had
                    # no requestable row at all: written_rar_folders
                    # deduplicates on the truncated (wrong) string.
                    #
                    # Also requires the truncation to leave at least two
                    # segments below FILE_DIRECTORY - i.e. never collapse to
                    # the artist root - and finds the EARLIEST matching
                    # segment by walking the path in order, rather than the
                    # old "first box word in LIST order" behaviour, which
                    # made the truncation point depend on the order this
                    # list happened to be written in.
                    # A folder earns a !rar row from holding something
                    # worth PACKING, not from holding anything at all. Before
                    # this the only conditions were "has a folder" and
                    # RAR_ENABLED, which read as "album folders" while the
                    # scan took .mp3 and .flac and nothing else. Once the scan
                    # took every file, every folder in the library became
                    # packable - a season of a series, or a folder holding one
                    # text note - with no size cap anywhere behind a line
                    # anybody in the channel can paste. See RAR_EXTENSIONS.
                    if folder and serve_albums and folder in packable_folders:
                        folder_segments = re.split(r'[\\/]', folder)
                        truncate_at = None
                        for seg_index, segment in enumerate(folder_segments):
                            if _BOX_WORD_RE.match(segment.strip()):
                                truncate_at = seg_index
                                break
                        # Three, not two. The threshold means "leave at least
                        # two segments below the FOLDER" - artist and album -
                        # and rel_dir now begins with the folder's label
                        # (#164), so every index shifted by one. Left at 2, a
                        # library shaped <label>/<artist>/<disc 1> would
                        # truncate to <label>/<artist>: the artist root, which
                        # dcc.py refuses outright, leaving that album with no
                        # requestable row at all - the exact failure the
                        # threshold was added to prevent.
                        if truncate_at is not None and truncate_at >= 3:
                            rar_folder_clean = "/".join(folder_segments[:truncate_at])
                        else:
                            # Truncating here would collapse to the artist
                            # root (or nothing) - offering the untruncated
                            # real path, box word and all, is still a
                            # request dcc.py will actually serve; the
                            # refused artist root is not.
                            rar_folder_clean = folder

                        raw_rar_str = f"{list_mod.LIST_FOLDER_PREFIX}{rar_folder_clean}\\"
                        display_rar_folder = raw_rar_str.replace("/", "\\")
                        
                        # Write the row exactly once per album to the .rar text file.
                        #
                        # NOTHING MAY BE APPENDED TO THIS LINE.
                        #
                        # It is not a display row, it is a command. AutoQ.mrc -
                        # the mIRC script people use to auto-download files and
                        # folders - copies this line out of the list and sends it
                        # verbatim, so anything after the folder stops it matching.
                        #
                        # The exact mechanism, read out of AutoQ.mrc's own
                        # `on *:filercvd:*:` handler rather than inferred:
                        #
                        #   $left($trunc($nopath($filename)),-4)
                        #       == $nopath($left($trunc($3-),-1))
                        #
                        # `$3-` is "token 3 to END OF LINE" - greedy, not just
                        # the folder token. Clean row: $3- is the folder,
                        # $left(...,-1) drops the trailing backslash, $nopath()
                        # takes "Album", and it matches the received filename
                        # minus ".rar", so the row is dequeued. Append anything
                        # and $3- swallows it too; $trunc() strips the spaces,
                        # so the character $left(...,-1) removes is the last
                        # letter of what was appended rather than the
                        # backslash, and $nopath() then extracts a substring
                        # that can never match. filercvd never fires, and the
                        # album stays queued in AutoQ forever despite having
                        # actually arrived. Verified against the script and
                        # reproduced (#256).
                        #
                        # Updating AutoQ is not a way out: people install old
                        # copies from mirrors, so the deployed versions cannot
                        # be assumed current.
                        #
                        # A folder size here is the obvious and recurring idea, and
                        # it was wanted. dcc.py already cites AutoQ compatibility
                        # for archive and filename shape, but nothing said it
                        # constrained this line, so the objection had to be
                        # rediscovered.
                        #
                        # The size belongs in the MAIN list instead, and it is
                        # there now (#69): a summary line under each folder
                        # heading - see folder_summary_line() for why it is its
                        # own line and NOT on the heading, which every reader
                        # takes as the folder path. Putting it there leaves
                        # this file at exactly one line per album, which a
                        # second ::INFO:: line per row would not.
                        #
                        # THE REASON, corrected against AutoQ.mrc itself rather
                        # than the second-hand version this comment used to give.
                        # A !rar row takes AutoQ's verbatim branch -
                        # `if (($2 == !rar) && ($3 != $null)) { aline -n $1- }` -
                        # the whole line, untouched. So anything appended here is
                        # sent as part of the request. That is absolute and has
                        # nothing to do with extensions.
                        #
                        # The file rows below end "::INFO:: <size>", and AutoQ does
                        # not merely cope with that: its file branch truncates at
                        # the end of the extension, so it never sees the tail at
                        # all. A further trailing field on a FILE row is therefore
                        # safe for AutoQ - the constraint there is its accept list
                        # (*.mp3 and *.rar), not the tail. See defaults.py's note
                        # above LIST_IGNORED_EXTENSIONS for the quoted source.
                        if display_rar_folder not in written_rar_folders:
                            f_rar.write(f"{rar_head}{_one_line(display_rar_folder)}\n")
                            written_rar_folders.add(display_rar_folder)
                single_file_size = format_size_human(bytes_size)
                # "4m31s 320/44.1/JS" after the size (#567), or nothing. After
                # the size, where AutoQ never looks (see the !rar note above)
                # and where other servers' lists already put it.
                if audio is not None:
                    tail = audio.suffix(audio_info.row_key(folder, filename))
                    if tail:
                        single_file_size = f"{single_file_size} {tail}"
                chunk.append(f"{row_head}{_one_line(filename)}  ::INFO:: {single_file_size}\n")
            if chunk:
                f.write("".join(chunk))

        # The film and series list. Written after the music one and from the
        # same walk, exactly as the album list is - a separate file with its
        # own header, travelling in the same archive.
        #
        # Only written when there is something to put in it: a music-only
        # library should not gain an empty file it has no use for, and every
        # reader below treats "no video list" as the ordinary case rather than
        # as a fault.
        # The one decision the three sites below all ask about: did THIS run
        # publish a film list? Asking the filesystem instead let a stale file
        # from an earlier rebuild answer for a run that produced nothing.
        wrote_video_list = bool(video_files_data)
        if wrote_video_list:
            video_bytes = sum(size for _f, _n, size in video_files_data)
            with open(tmp_video_path, "w", encoding="utf-8") as f_video:
                f_video.write(
                    f"List of {len(video_files_data):,} Films & Series "
                    f"({format_total_size(video_bytes)}) generated on "
                    f"{date_header_str}\n")
                f_video.write(
                    f"To request one, copy/paste to the channel... "
                    f"!{list_nick()} FILENAME eg. !{list_nick()} "
                    f"Some.Film.2021.mkv\n")
                # Same order as the .txt above, and for the same reason: this
                # file travels on its own once it is out of the archive, so it
                # carries its own attribution rather than inheriting one.
                f_video.write(list_identity_line() + "\n")
                if operator_header:
                    f_video.write("\n" + operator_header + "\n")
                f_video.write("\n")

                video_folder = None
                video_totals = folder_totals(video_files_data)
                # As in the music list above (#1125): the nick once, and one
                # write per folder.
                row_head = f"!{list_nick()} "
                chunk = []
                for folder, filename, bytes_size in video_files_data:
                    if folder != video_folder:
                        video_folder = folder
                        if chunk:
                            f_video.write("".join(chunk))
                            chunk = []
                        raw = (f"{list_mod.LIST_FOLDER_PREFIX}{folder}\\"
                               if folder else list_mod.LIST_FOLDER_PREFIX)
                        line = _one_line(raw.replace("/", "\\"))
                        rule = "=" * len(line)
                        chunk.append(f"\n{rule}\n{line}\n{rule}\n")
                        chunk.append(folder_summary_line(*video_totals[folder], format_size_human) + "\n")
                    chunk.append(
                        f"{row_head}{_one_line(filename)}"
                        f"  ::INFO:: {format_size_human(bytes_size)}\n")
                if chunk:
                    f_video.write("".join(chunk))
            print(f"[LIST-GEN] Film & series list created: {tmp_video_path}")

        print(f"[LIST-GEN] Text list created: {tmp_txt_path}")
        # The walk is over by here, so the folder bar has nothing left to say.
        # The phase does: writing and packing a 719k-row list is not instant,
        # and a bar that sat at 100% through it would look stalled.
        write_progress("writing", files=len(all_files_data) + len(video_files_data),
                       force=True)
        if serve_albums:
            print(f"[LIST-GEN] RAR album list created: {tmp_rar_path}")
        
        # The size and rawbytes side files used to be published here. They are now
        # written with the lists in the finalise section below - see the comment
        # there for why writing them ahead of the guards was half a rollback.
            
        # A scan that found NOTHING is almost always an unavailable mount, not an empty
        # library - and publishing it would replace a good index with an empty one, which is
        # exactly the failure this rewrite exists to prevent. Size cannot be the test: the
        # file always carries two header lines, so it is never zero bytes.
        #
        # Accept an empty result only when there is no working index to lose, so a genuine
        # first run on an empty library still succeeds.
        if total_files_count == 0 and not video_files_data:
            try:
                existing = [f for f in os.listdir(directory)
                            if f.startswith(config.LIST_BASE_NAME)
                            and f.endswith(".txt") and "-RAR-" not in f]
            except OSError:
                existing = []
            if existing:
                print("[LIST-GEN ERROR] Scan found 0 files but an index already exists "
                      "(mount unavailable?). Keeping the previous index.")
                _discard_temp_lists(*tmp_all_paths)
                return False
            print("[LIST-GEN] Scan found 0 files and there is no previous index; "
                  "publishing an empty list.")

        if not os.path.exists(tmp_txt_path):
            print("[LIST-GEN ERROR] No list was written. Keeping the previous index.")
            _discard_temp_lists(*tmp_all_paths)
            return False

        # ---- what a user actually downloads -----------------------------
        # config.LIST_FORMAT decides how the two text lists are handed over:
        # as one plain .txt, packed into a .zip, or packed into a .rar.
        # OmenServe has offered the same three for years, and which of them a
        # given person's client opens without complaint still differs.
        #
        # This is deliberately NOT tied to RAR_ENABLED. That switch governs
        # whether the bot will pack an ALBUM FOLDER for a stranger on demand -
        # minutes of CPU and a large temporary file, once per request. Packing
        # two text files once per rebuild, on the operator's own schedule, is
        # a different job and does not belong behind the same switch.
        members = [(tmp_txt_path, os.path.basename(txt_path))]
        # The film and series list rides in the same archive, on the same
        # terms as the album list: present when it has content, absent when
        # the library has no video, and never a reason to fail the build.
        if wrote_video_list:
            members.append((tmp_video_path, os.path.basename(video_path)))
        if os.path.exists(tmp_rar_path) and os.path.getsize(tmp_rar_path) > 0:
            members.append((tmp_rar_path, os.path.basename(rar_path)))

        wanted_format = list_mod.list_format()
        print(f"[LIST-GEN] Packing the list for download as .{wanted_format}...")
        artifact_format, tmp_artifact_path, artifact_path = build_list_artifact(
            wanted_format, members, today, directory)

        # Everything generated cleanly. Swap the new files in, THEN remove the superseded
        # ones. os.replace overwrites atomically on both POSIX and Windows, where os.rename
        # would raise because the destination already exists.
        #
        if not serve_albums:
            # Not published, and not left behind either: an empty album list
            # in lists/ reads as "this bot offers no albums" to anything
            # counting the file, which is a different claim from "it does
            # not offer them at all".
            _discard_temp_lists(tmp_rar_path)

        # THE DOWNLOAD ARTIFACT FIRST, because it is the one a DCC send holds
        # open - so the likeliest failure is discovered before anything else
        # has moved. All of them go together or none of them do; see
        # _publish_artifacts().
        swaps = [(tmp_artifact_path, artifact_path), (tmp_txt_path, txt_path)]
        # Only when it was written. A library with no video has no video list,
        # and the publish must not try to move a file that was never staged.
        if wrote_video_list:
            swaps.append((tmp_video_path, video_path))
        if serve_albums:
            swaps.append((tmp_rar_path, rar_path))
        # Said BEFORE the swap (#923): from here until this list is done -
        # the swap and the prune of what it replaced - the daemon refuses new
        # searches and file requests, since on Windows a file somebody is
        # reading cannot be renamed or removed.
        write_progress("publishing", force=True)
        _publish_artifacts(swaps)

        # THE POINT OF NO RETURN (#442). Everything from here to the except is
        # tail work - side files, cleanup, bookkeeping - and any of it can
        # raise. The swap has already happened, so the NEW list is the one in
        # use from this line onward, and the handler must not go on claiming
        # the old one was kept.
        published = True

        # The two side files are published HERE, AFTER every swap above has
        # already succeeded, and atomically. They used to be written before
        # the swaps - so a failure partway through (a Windows os.replace can
        # raise PermissionError if something else has the destination open)
        # rolled the list itself back to the previous index while the size
        # and byte count it wears had already been overwritten with the new
        # scan's numbers: an old index publishing a new scan's size. And a
        # plain open(..., "w") truncates first, so an interruption left a
        # readable but EMPTY file, which is unparseable and used to cost the
        # caller the file count and the list date as well as the size.
        db._atomic_write(SIZE_FILE_PATH, formatted_size)
        db._atomic_write(RAWBYTES_FILE_PATH, str(total_bytes))
        # #213: the artifacts just published are named after the CURRENT
        # LIST_BASE_NAME, so this is the moment that fact becomes true. Written
        # here rather than only in the migration, so a rename between two
        # rebuilds is still migrated from the right name.
        write_list_base_marker(config.LIST_BASE_NAME, directory=directory,
                               log=print)

        # DUPLICATE FILENAMES, SAID OUT LOUD AT BUILD TIME.
        #
        # The dashboard has answered this since #164 - Tools > Verify list,
        # build_verify_list_payload() - but only when somebody goes and looks,
        # and the operators most likely to have collisions are the ones running
        # several folders without watching a web page. The scan is holding the
        # data already, so saying it here costs nothing and no second walk.
        #
        # Why it matters: a request names a FILE, not a path, because a bare
        # filename is all the list gives a requester to copy.
        # dcc.handle_download_request() resolves that name against the list and
        # serves the FIRST folder it finds it under - so every later copy is
        # listed, looks requestable, and can never be sent. The requester does
        # not get an error either; they get the other file.
        #
        # Both lists, in the order all_list_paths() hands them to the resolver:
        # a name in the music list and the film list resolves to the music one.
        # Counting them separately would miss exactly the collisions the split
        # introduced.
        #
        # The COUNT only. The detail belongs to the view that already presents
        # it properly, resolved to paths this machine has; printing folder
        # headings here would be a second presentation to keep in step.
        # THE COUNT, NOT THE ANSWER (#463). This used to build a second full
        # copy of the library - one dict per file - and hand it to
        # find_duplicate_filenames(), whose result was passed to len() and
        # dropped. At 5.4M files that copy measured 2.2 GiB and took the peak
        # for the whole rebuild to 3.5 GiB, on top of the sort's own peak.
        # count_duplicate_filenames() holds one entry per distinct NAME and
        # answers the same question by the same definition.
        def _every_listed_row():
            # A GENERATOR, not a list. count_duplicate_filenames() reads its
            # input once and keeps one entry per distinct NAME, so handing it
            # a materialised copy of the library would put back a smaller
            # version of the very thing this change removes.
            for folder, name, _size in all_files_data:
                yield folder, name
            for folder, name, _size in video_files_data:
                yield folder, name

        duplicate_count = list_mod.count_duplicate_filenames(_every_listed_row())
        if duplicate_count:
            print(f"[LIST-GEN] WARNING: {duplicate_count} filename(s) appear "
                  f"under more than one folder. A request names a file, not a "
                  f"path, so only the first copy of each can ever be sent. "
                  f"Dashboard: Tools > Verify list.")

        print(f"[LIST-GEN] New lists activated: {os.path.basename(txt_path)} "
              f"(download: {os.path.basename(artifact_path)})")

        keep = {os.path.basename(txt_path), os.path.basename(artifact_path)}
        # The film list too, when this run published one. It is named
        # "<base>-VIDEO-<date>.txt", so the prefix match below picks it up as
        # a generated list - correctly - and without this it would be swapped
        # in and deleted again in the same run, every run. The only trace was
        # a "[LIST-CLEAN] Removed 1 superseded list(s)" line that reads like
        # housekeeping working, which is exactly how the side files were lost
        # once already (see _prune_superseded_lists).
        # THIS RUN's decision, not "a file is sitting at that path". video_path
        # carries today's date, so on a second rebuild the same day that file
        # is the EARLIER run's output - and a run that found no video would
        # have kept it, leaving a film list describing films that are gone:
        # still searchable, still counted by the advert, and absent from the
        # archive users actually download. It self-corrected across a date
        # boundary, which is why a single-build test never saw it.
        if wrote_video_list:
            keep.add(os.path.basename(video_path))
        if serve_albums:
            keep.add(os.path.basename(rar_path))
        # Yesterday's artifact in another format goes with the rest. It is not
        # only clutter: find_latest_list_file() falls back to whatever HAS been
        # built when the configured format has not been yet, so a stale .zip
        # left beside a fresh .rar would go on being handed out to somebody the
        # day the operator switched formats and the build failed.
        _prune_superseded_lists(keep=keep, directory=directory)
        if audio is not None:
            # Only a PUBLISHED rebuild forgets the files it did not see; a
            # failed one may have seen half the library.
            #
            # AN OPTIONAL CACHE DOES NOT FAIL A LIVE LIST (#980). The swap has
            # happened: the new list is what users are getting. A locked or
            # unwritable cache here went to the except below, and the rebuild
            # was reported failed - with several lists, as "still serving
            # what they last built", which was untrue. close() still keeps
            # what this rebuild read; the next one reads the rest again.
            try:
                audio.publish()
            except (sqlite3.Error, OSError) as cache_err:
                print(f"[LIST-GEN] The list is published, but the audio info cache could not be "
                      f"updated ({cache_err}): the next rebuild reads again what it could not keep.")
            # What is left to read goes to the background reading (#1182),
            # which starts once every list is published.
            if audio.pending and reading_jobs is not None:
                reading_jobs.append(AudioReadingJob(list_name, audio.scope, list(audio.pending)))
            print(f"[LIST-GEN] Audio info: {audio.reused_count:,} file(s) unchanged since the "
                  f"last rebuild"
                  f"{f', {len(audio.pending):,} new or changed to read in the background' if audio.pending else ''}.")
        return True
            
    except Exception as e:
        print(f"[LIST-GEN ERROR] Failed to generate the lists: {e}")
        if published:
            # THE SWAP ALREADY HAPPENED. Saying the old list was kept would be
            # exactly backwards: the new list is live and serving, and only
            # the tail work after the swap failed. An operator told "the
            # previous list is still in use" reasonably concludes nothing
            # changed and that the rebuild can simply be retried - while the
            # bot is already serving the new one.
            print("[LIST-GEN] The new list WAS published and is in use - the "
                  "failure above happened afterwards, in the tail work. Do "
                  "not treat this as 'nothing changed'.")
        else:
            print("[LIST-GEN] The previous list was left untouched and is still in use.")
        _discard_temp_lists(*tmp_all_paths)
        return False
    finally:
        if audio is not None:
            audio.close()

# ---------------------------------------------------------------------------
# THE BACKGROUND AUDIO READING (#1182).
#
# A rebuild with LIST_SHOW_AUDIO_INFO on used to read every new or changed
# audio file before writing the list, capped by LIST_AUDIO_INFO_MINUTES, with
# one line at the start, one at the end, and nobody able to tell how far it had
# got. Now:
#
#   1. The rebuild publishes first, with what the cache already knows.
#   2. Started by the bot, the same process then reads what is left, with no
#      time limit, saving to the cache in batches as it goes - so a reading
#      that is stopped keeps what it read. Run by hand (a terminal, cron,
#      configure.py), it publishes and says how to read them instead: an
#      unattended hours-long wait in a terminal is not what anybody asked for.
#   3. When it is done, and only if a row's length and quality changed, those
#      rows are rewritten into the CURRENT published list: no new scan (the
#      slow part on a network drive), the same atomic swap, the archive built
#      the way a rebuild builds it, and the list's date kept - other bots
#      decide to fetch a list again by that date (#286), so they do not fetch
#      it twice.
#
# `update_list.py --read-audio-info` runs 2 and 3 alone, for the lists as they
# are published, and writes in any length the cache holds and the list does
# not show yet - a rewrite that could not be swapped in last time. The
# dashboard's Tools page, the console's `audioinfo` and dccore.mrc's Library
# menu start it; by hand it reads in the foreground with progress lines.
#
# ONE AT A TIME, ACROSS PROCESSES. Every run holds the list lock beside the
# progress file (an OS lock: it dies with its process). A rebuild that finds a
# reading holding it asks that reading to stop - whoever started it, the bot,
# a terminal or a bot that has since died - and waits for it; a rebuild that
# finds another rebuild, or a reading-only run that finds anything, does not
# start. The reading's rewrite stages under its own names (".audio.new"), so
# it and a rebuild could never write the same temporary file anyway.
# ---------------------------------------------------------------------------

# One published list's audio files left to read: the list (None for the only
# one), its cache scope, and (key, path, size) for each file - the size None
# when the file was named by the published list rather than by a walk.
AudioReadingJob = collections.namedtuple("AudioReadingJob", "list_name scope pending")

# How many files are read between two saves to the cache. A stopped or killed
# reading loses at most this many reads, and a save is one transaction.
AUDIO_SAVE_EVERY = 200

# Seconds between two progress lines printed by the reading itself.
AUDIO_PRINT_SECONDS = 30.0

# How long a rebuild waits for a reading it asked to stop.
AUDIO_STOP_WAIT = 60.0

# How long the rewrite keeps trying to swap the list in while something still
# holds the old one open (#1182 audit). On Windows a list archive that a DCC
# send started before the swap holds it; new sends are refused from the
# rewrite on, so the running ones can only drain. Past this, the result says
# which list it could not write, and the bot tries again once no list send is
# running.
READING_SWAP_WAIT = 600.0

# The staging suffix of the reading's rewrite: never a rebuild's ".new".
AUDIO_STAGING = ".audio.new"

# What the rewrite answers.
REWRITE_UPDATED = "updated"
REWRITE_UNCHANGED = "unchanged"
REWRITE_FAILED = "failed"
REWRITE_STOPPED = "stopped"

# What the rewrite looks for in a file row: the size and anything after it.
_INFO_SEPARATOR = "  ::INFO:: "

# Set by the bot in the environment of the update_list.py it starts: a token
# its progress file then carries, so the bot knows the report is its own child's
# whatever launcher stands between them (a Windows venv's python.exe starts the
# real interpreter as a second process, with another pid), and the bot's own
# pid, so a reading whose bot has gone stops by itself.
RUN_TOKEN_ENV = "DCCORE_RUN_TOKEN"
DAEMON_PID_ENV = "DCCORE_DAEMON_PID"

# Seconds between two checks that the bot that started the reading is alive.
DAEMON_CHECK_SECONDS = 10.0

_list_lock = [None]


def run_token():
    """The token the bot gave this run, or "" for a run started by hand."""
    return os.environ.get(RUN_TOKEN_ENV, "")


def started_by_the_bot():
    return bool(run_token())


def list_lock_path():
    return progress_path() + ".lock"


def list_lock_holder():
    """(pid, "rebuild" or "reading") of the run holding the list lock, or None
    when no run does. Asked by the bot before it starts a reading."""
    path = list_lock_path()
    if not os.path.exists(platform_compat.long_path(path)):
        return None
    try:
        held = platform_compat.file_lock_held(platform_compat.long_path(path))
    except OSError:
        return None
    if not held:
        return None
    return platform_compat.file_lock_note(platform_compat.long_path(path))


def take_list_lock(kind, log=print, wait=None, clock=time.monotonic, sleep=time.sleep):
    """Hold the list lock for this run, as `kind`. Returns None when it is
    held, else the holder's (pid, kind) it could not take it from.

    A rebuild asks a reading that holds it to stop, and waits up to
    AUDIO_STOP_WAIT for it to let go: what the reading read is saved."""
    path = platform_compat.long_path(list_lock_path())
    try:
        _list_lock[0] = platform_compat.take_file_lock(path, kind)
        # Nobody else holds the list, so a stop request lying about was meant
        # for a run that has ended: it must not stop this one at its start.
        clear_stop_request()
        return None
    except platform_compat.AlreadyRunning:
        holder = platform_compat.file_lock_note(path)
    if kind != "rebuild" or holder[1] != "reading":
        return holder
    log(f"[LIST-GEN] An audio reading (pid {holder[0]}) holds the list; asking it to stop. "
        f"What it read is kept.")
    request_stop()
    deadline = clock() + (AUDIO_STOP_WAIT if wait is None else wait)
    try:
        while clock() < deadline:
            sleep(0.5)
            try:
                _list_lock[0] = platform_compat.take_file_lock(path, kind)
                return None
            except platform_compat.AlreadyRunning:
                continue
        return platform_compat.file_lock_note(path)
    finally:
        clear_stop_request()


def mark_list_lock(kind):
    """Say, in the held lock, what this run is doing now."""
    if _list_lock[0] is not None:
        platform_compat.set_file_lock_note(_list_lock[0], kind)


def release_list_lock():
    platform_compat.release_file_lock(_list_lock[0])
    _list_lock[0] = None


def reading_should_stop(clock=time.monotonic, _state={"checked": 0.0, "gone": False}):
    """The reading's stop question: a stop was asked, or the bot that started
    it is no longer running (#1182 audit) - a bot that crashed or whose window
    was closed ran no shutdown to ask. The bot's liveness is checked every
    DAEMON_CHECK_SECONDS, not on every read."""
    if stop_requested():
        return True
    daemon = os.environ.get(DAEMON_PID_ENV, "")
    if not daemon:
        return False
    if not _state["gone"] and clock() - _state["checked"] >= DAEMON_CHECK_SECONDS:
        _state["checked"] = clock()
        _state["gone"] = not platform_compat.pid_alive(daemon)
    return _state["gone"]


def _scope_of(list_name):
    """The audio cache scope of a list: its own name, the primary's for the
    only list (#979) - the same as generate_master_list() opens.
    """
    return list_name or library.primary_list().name


def _published_list_names():
    """The list name to give rewrite_audio_info() for each configured list:
    None for the only one, each name when there are several."""
    every = library.lists()
    return [None] if len(every) == 1 else [entry.name for entry in every]


def _is_rule(line):
    return bool(line) and not line.strip("=")


def _list_lines(path):
    """(body, ending, heading) for each line of a published music list, the
    heading being the folder heading the line sits under (None above the
    first). Opened with newline="" so the line endings come back exactly as
    written - the rewrite must not turn a Windows list's CRLF into LF."""
    heading = None
    back2 = back1 = None
    with io.open(platform_compat.long_path(path), encoding="utf-8", newline="") as handle:
        for raw in handle:
            body = raw.rstrip("\r\n")
            ending = raw[len(body):]
            if not body.startswith("!") and _is_rule(body) and _is_rule(back2) \
                    and back1 is not None and not _is_rule(back1):
                heading = back1
            yield body, ending, heading
            back2, back1 = back1, body


def _row_name(body):
    """(name, tail) of a file row - the tail being its size and anything
    after it - or None for any other line. The nick never holds a space, and
    the tail never holds the separator, so the first space and the LAST
    separator bound the name whatever it contains."""
    if not body.startswith("!"):
        return None
    space = body.find(" ")
    at = body.rfind(_INFO_SEPARATOR)
    if space < 0 or at <= space:
        return None
    return body[space + 1:at], body[at + len(_INFO_SEPARATOR):]


# The phase while a reading-only run reads the published list to find what to
# read - before any file is opened.
FINDING_PHASE = "finding"


def unread_audio_jobs(log=print):
    """The reading-only run's work: in each published list, the audio rows
    the cache has no entry for.

    No walk: the rows are the published list's, each heading is turned back
    into its folder - the label first (#164), then the path below it - and a
    file's size is asked for only when it is read. A row whose folder label
    is no longer configured is left alone; the next rebuild drops it."""
    import list as list_mod

    write_progress(FINDING_PHASE, force=True)
    every = library.lists()
    primary = library.primary_list().name
    jobs = []
    for entry in every:
        list_name = None if len(every) == 1 else entry.name
        path = list_mod.find_latest_list(list_name)
        if not path:
            continue
        scope = _scope_of(list_name)
        cache = audio_info.Cache.open(scope=scope, formerly="" if scope == primary else None, log=log)
        if cache is None:
            continue
        known = cache.known
        cache.close()
        pending = []
        resolved = {}
        for body, _ending, heading in _list_lines(path):
            row = _row_name(body)
            if row is None or heading is None or not audio_info.is_audio(row[0]):
                continue
            if heading not in resolved:
                parts = list_mod.list_heading_parts(heading)
                folder = library.folder_for_label(parts[0], list_name) if parts else None
                resolved[heading] = None if folder is None else (
                    os.path.join(folder.name, *parts[1:]), os.path.join(folder.path, *parts[1:]))
            where = resolved[heading]
            if where is None:
                continue
            key = audio_info.row_key(where[0], row[0])
            if key not in known:
                pending.append((key, platform_compat.long_path(os.path.join(where[1], row[0])), None))
        if pending:
            jobs.append(AudioReadingJob(list_name, scope, pending))
    return jobs


def _rewrite_rows(source, target, suffixes):
    """Write `source` to `target` with each audio row's length and quality
    taken from `suffixes` ({heading: {name: suffix}}). Every other byte is
    copied as it is. Returns how many rows changed."""
    changed = 0
    chunk = []
    with io.open(platform_compat.long_path(target), "w", encoding="utf-8", newline="") as out:
        for body, ending, heading in _list_lines(source):
            row = _row_name(body)
            if row is not None and audio_info.is_audio(row[0]):
                name, tail = row
                size = tail.split(" ", 1)[0]
                suffix = suffixes.get(heading, {}).get(name, "")
                wanted = f"{size} {suffix}" if suffix else size
                if wanted != tail:
                    body = body[:len(body) - len(tail)] + wanted
                    changed += 1
            chunk.append(body + ending)
            if len(chunk) >= 4096:
                out.write("".join(chunk))
                chunk = []
        out.write("".join(chunk))
    return changed


def _swap_waiting_for_readers(swaps, unchanged, stop, wait, clock, sleep, log):
    """_publish_artifacts(), tried again while the old list is held open - a
    list download already running when the reading reached its swap, on
    Windows - for up to `wait` seconds. Each failed try has rolled back
    completely, so trying again is safe; `unchanged()` and `stop()` are asked
    before every try. Raises the last error when time runs out."""
    deadline = clock() + wait
    said = False
    while True:
        if stop() or not unchanged():
            return False
        write_progress("publishing", force=True)
        try:
            _publish_artifacts(swaps)
            return True
        except OSError as err:
            if clock() >= deadline:
                raise
            if not said:
                said = True
                log(f"[AUDIO-INFO] The list is open elsewhere ({err}) - a download still running. "
                    f"New ones wait; trying again for up to {int(wait)} s.")
            sleep(5.0)


def rewrite_audio_info(list_name=None, stop=None, log=print, swap_wait=None,
                       clock=time.monotonic, sleep=time.sleep):
    """Step 3: the length and quality the cache now holds, written into the
    published list. Returns REWRITE_UPDATED when the list was swapped,
    REWRITE_UNCHANGED when no row would change, REWRITE_STOPPED when a stop or
    a rebuild that published meanwhile got there first, and REWRITE_FAILED
    when it could not be written - the lengths stay in the cache, and the
    next reading-only run writes them in.

    The result is what a full rebuild with the same cache would publish: each
    audio row's suffix is the cache's for that row, found by the heading and
    name exactly as the rebuild writes them (_heading_text(), _one_line()).
    The list's modification time, which is the date the advert gives it, is
    put back afterwards."""
    import list as list_mod

    stop = stop or (lambda: False)
    directory = list_mod.list_dir(list_name)
    txt_path = list_mod.find_latest_list(list_name)
    if not txt_path:
        return REWRITE_UNCHANGED
    base_name = os.path.basename(txt_path)
    date_str = base_name[len(config.LIST_BASE_NAME) + 1:-len(".txt")]
    try:
        before = os.stat(platform_compat.long_path(txt_path))
    except OSError:
        return REWRITE_UNCHANGED

    cache = audio_info.Cache.open(scope=_scope_of(list_name), log=log)
    if cache is None:
        return REWRITE_FAILED
    known = cache.known
    cache.close()
    suffixes = {}
    for key, (_size, suffix) in known.items():
        if suffix:
            folder, _sep, filename = key.partition("\x00")
            suffixes.setdefault(_heading_text(folder), {})[_one_line(filename)] = suffix

    # The archive in the format it was published in: a rebuild whose .rar
    # fell back to .zip published a .zip, and that is the file to replace.
    published = [fmt for fmt in list_mod.LIST_FORMATS
                 if os.path.exists(_artifact_paths(fmt, date_str, directory)[0])]
    if not published:
        log(f"[AUDIO-INFO] {base_name} has no archive beside it; the list is left as it is.")
        return REWRITE_FAILED
    fmt = list_mod.list_format() if list_mod.list_format() in published else published[0]

    def unchanged():
        try:
            now = os.stat(platform_compat.long_path(txt_path))
        except OSError:
            return False
        return (now.st_mtime_ns, now.st_size) == (before.st_mtime_ns, before.st_size)

    write_progress("rewriting", force=True)
    tmp_txt = txt_path + AUDIO_STAGING
    tmp_artifact = None
    try:
        changed = _rewrite_rows(txt_path, tmp_txt, suffixes)
        if not changed:
            _discard_temp_lists(tmp_txt)
            return REWRITE_UNCHANGED
        # The members a rebuild packs, in its order: this list, then the film
        # list and the album list of the same build when they were published.
        members = [(tmp_txt, base_name)]
        for marker in (list_mod.VIDEO_LIST_MARKER, "RAR"):
            other = os.path.join(directory, f"{config.LIST_BASE_NAME}-{marker}-{date_str}.txt")
            if os.path.exists(other) and os.path.getsize(other) > 0:
                members.append((other, os.path.basename(other)))
        used, tmp_artifact, artifact_path = build_list_artifact(
            fmt, members, date_str, directory, staging=AUDIO_STAGING)
        if used != fmt:
            log(f"[AUDIO-INFO] The .{fmt} archive could not be built again; the list is left as it is.")
            _discard_temp_lists(tmp_txt, tmp_artifact)
            return REWRITE_FAILED
        artifact_before = os.stat(platform_compat.long_path(artifact_path))
        swapped = _swap_waiting_for_readers(
            [(tmp_artifact, artifact_path), (tmp_txt, txt_path)], unchanged, stop,
            READING_SWAP_WAIT if swap_wait is None else swap_wait, clock, sleep, log)
        if not swapped:
            _discard_temp_lists(tmp_txt, tmp_artifact)
            return REWRITE_STOPPED
    except Exception as err:
        log(f"[AUDIO-INFO] Could not write the length and quality into {base_name} ({err}); "
            f"they are kept, and the bot writes them in once nothing holds the list.")
        _discard_temp_lists(tmp_txt, tmp_artifact)
        return REWRITE_FAILED

    for path, stamp in ((txt_path, before), (artifact_path, artifact_before)):
        try:
            os.utime(platform_compat.long_path(path), ns=(stamp.st_atime_ns, stamp.st_mtime_ns))
        except OSError:
            pass
    log(f"[AUDIO-INFO] {changed:,} row(s) of {base_name} now show their length and quality; "
        f"the list keeps its date.")
    return REWRITE_UPDATED


def read_audio_info(jobs, stop=None, log=print, rewrite_all=False):
    """Step 2, then step 3 for each list that gained a length: read every
    job's files, with no time limit. Returns what happened:

        {"outcome": "nothing" | "done" | "stopped", "read", "unreadable",
         "total", "updated": [list names], "unwritten": [list names]}

    `rewrite_all` (the reading-only run) also writes into every published
    list the lengths the cache already holds - one a rewrite could not swap
    in last time - even when nothing is left to read.

    Progress goes to the progress file as phase "reading": folder_index is
    the files read, folder_count the files to read, rate the files a second -
    written when a read completes, so a mount that stops answering is a stall
    the daemon can see (#968), and forced after each rewrite, so the swap's
    "publishing" never outlives the swap into the next list's reading.
    """
    global _started_at

    stop = reading_should_stop if stop is None else stop
    total = sum(len(job.pending) for job in jobs)
    result = {"outcome": "nothing", "read": 0, "unreadable": 0, "total": total,
              "updated": [], "unwritten": []}
    workers = max(1, min(128, int(getattr(config, "LIST_AUDIO_INFO_THREADS", 64) or 1)))
    # The reading's own start, for the elapsed time the dashboard shows.
    _started_at = time.time()
    began = time.perf_counter()
    printed = [time.monotonic()]

    attempted = set()

    def rewrite(list_name, offset):
        answer = rewrite_audio_info(list_name, stop=stop, log=log)
        name = _scope_of(list_name)
        attempted.add(name)
        if answer == REWRITE_UPDATED:
            result["updated"].append(name)
        elif answer == REWRITE_FAILED:
            result["unwritten"].append(name)
        # Whatever happened, the swap is over: "publishing" must not stay on
        # disk into the next list's reading, where it would pause searches.
        write_progress(READING_PHASE, folder_index=offset, folder_count=total, force=True)
        return answer

    if total:
        where = "in the background" if started_by_the_bot() else "(Ctrl-C stops it; what was read is kept)"
        log(f"[AUDIO-INFO] Audio info: reading {total:,} file(s) {where}, {workers} at a time.")
        write_progress(READING_PHASE, folder_index=0, folder_count=total, rate=0, force=True)

    read_before = 0
    for job in jobs:
        cache = audio_info.Cache.open(scope=job.scope, log=log)
        if cache is None:
            continue
        cache.pending = list(job.pending)

        def progress(done, _of, offset=read_before):
            count = offset + done
            elapsed = time.perf_counter() - began
            rate = int(count / elapsed) if elapsed > 0 else 0
            write_progress(READING_PHASE, folder_index=count, folder_count=total, rate=rate)
            if time.monotonic() - printed[0] >= AUDIO_PRINT_SECONDS:
                printed[0] = time.monotonic()
                log(f"[AUDIO-INFO] Audio info: {count:,} of {total:,} read"
                    f"{f', {rate:,}/s' if rate else ''}.")

        try:
            cache.read_pending(workers=workers, progress=progress, stop=stop,
                               save_every=AUDIO_SAVE_EVERY)
        finally:
            cache.close()
        result["read"] += cache.read_count
        result["unreadable"] += cache.unreadable_count
        read_before += cache.read_count
        # What an operator compares to choose LIST_AUDIO_INFO_THREADS: the
        # files a second over the whole reading, and how many at a time.
        seconds = time.perf_counter() - began
        result["rate"] = int(read_before / seconds) if seconds > 0 else 0
        result["workers"] = workers
        if cache.left_count or stop():
            result["outcome"] = "stopped"
            return result
        # A row can only have changed if something was read with a length;
        # rewrite_audio_info() then counts the rows that really did.
        if any(suffix for _size, suffix in cache.fresh.values()):
            if rewrite(job.list_name, read_before) == REWRITE_STOPPED:
                result["outcome"] = "stopped"
                return result

    if rewrite_all:
        for list_name in _published_list_names():
            if _scope_of(list_name) in attempted:
                continue
            if rewrite(list_name, read_before) == REWRITE_STOPPED:
                result["outcome"] = "stopped"
                return result
    if total or result["updated"] or result["unwritten"]:
        result["outcome"] = "done"
    return result


def describe_audio_result(result):
    """The reading's result as one line, for the console, the debug channel
    and the @DCCore window. The daemon says it from the JSON the child
    prints, so the wording lives here once."""
    outcome = result.get("outcome")
    read = int(result.get("read") or 0)
    unreadable = int(result.get("unreadable") or 0)
    total = int(result.get("total") or 0)
    if outcome == "nothing":
        return "Audio info: nothing new to read."
    if outcome == "off":
        return ("Audio info: length and quality are off (LIST_SHOW_AUDIO_INFO), "
                "so there is nothing to read.")
    if outcome == "busy":
        return ("Audio info: a list rebuild or another reading is already running "
                f"(pid {result.get('pid') or '?'}); not starting a second one.")
    if outcome == "stopped":
        return (f"Audio info: stopped: {read:,} of {total:,} read and kept; "
                f"the rest are read next time.")
    if outcome == "failed":
        return f"Audio info: failed: {result.get('error') or 'unknown error'}."
    updated = list(result.get("updated") or [])
    unwritten = list(result.get("unwritten") or [])
    if unwritten:
        changed = ("the list is open elsewhere, so the lengths are written in later"
                   if len(unwritten) == 1 else
                   "these lists are open elsewhere, so the lengths are written in later: "
                   + ", ".join(unwritten))
    elif not updated:
        changed = "the list did not change"
    elif len(updated) == 1:
        changed = "list updated"
    else:
        changed = "lists updated: " + ", ".join(updated)
    if not total and not read:
        return f"Audio info: nothing new to read; lengths read earlier: {changed}."
    rate = result.get("rate")
    speed = (f" Read at {rate:,} files a second, {result.get('workers')} at a time."
             if rate else "")
    return f"Audio info: done: {read - unreadable:,} read, {unreadable:,} unreadable, {changed}.{speed}"


# The child's last line: the result, for the daemon to read back.
AUDIO_RESULT_TAG = "[AUDIO-INFO-RESULT] "


def run_audio_reading(jobs=None, log=print):
    """Steps 2 and 3 from start to end, never raising: for a rebuild's jobs,
    or - with none given - for the lists as they are published, writing in
    too what the cache holds and the list does not show yet. Prints the
    result line and the JSON the daemon reads back, and clears the progress
    file when done."""
    try:
        if not getattr(config, "LIST_SHOW_AUDIO_INFO", False):
            result = {"outcome": "off", "read": 0, "unreadable": 0, "total": 0, "updated": []}
        else:
            rewrite_all = jobs is None
            if jobs is None:
                jobs = unread_audio_jobs(log)
            result = read_audio_info(jobs, log=log, rewrite_all=rewrite_all)
    except Exception as err:
        result = {"outcome": "failed", "error": str(err), "read": 0, "unreadable": 0,
                  "total": 0, "updated": []}
    finally:
        clear_progress()
    if result.get("outcome") == "stopped":
        clear_stop_request()
    log(f"[AUDIO-INFO] {describe_audio_result(result)}")
    log(AUDIO_RESULT_TAG + json.dumps(result))
    return result


def read_audio_info_main(log=print):
    """`update_list.py --read-audio-info`: the reading alone, under the list
    lock - refused, not queued, when a rebuild or another reading holds it."""
    holder = take_list_lock("reading", log=log)
    if holder is not None:
        result = {"outcome": "busy", "pid": holder[0], "read": 0, "unreadable": 0,
                  "total": 0, "updated": []}
        log(f"[AUDIO-INFO] {describe_audio_result(result)}")
        log(AUDIO_RESULT_TAG + json.dumps(result))
        return result
    try:
        return run_audio_reading(log=log)
    finally:
        release_list_lock()


def generate_all_lists(log=print, reading_jobs=None):
    """Build every configured list. True only if every one of them succeeded.

    `reading_jobs` collects each published list's audio files left to read
    (#1182); see generate_master_list().

    On a single-list install this is one call to generate_master_list() with no
    name - byte for byte what running this script has always done. The loop is
    what a second list turns on.

    EACH LIST IS BUILT INDEPENDENTLY, and one failing does not stop the rest. A
    list whose folder is on an unavailable mount must not take down the list
    whose folder is on a local disk: generate_master_list() already refuses to
    publish an empty scan over a working index, so a failed list goes on serving
    what it last built while its neighbours move on. Returning False is what
    tells the operator - and !update - that something needs looking at.

    The failures are NAMED, not counted. "1 of 3 lists failed" sends somebody to
    read a log they already have open; naming it tells them which folder to go
    and look at.
    """
    import library

    # The progress file is dropped HERE, not only in __main__, so every entry
    # point clears it and the behaviour is reachable from a test. A leftover
    # from a finished run reads as a rebuild still in progress, and the
    # dashboard would show a bar that never moves.
    try:
        return _generate_all_lists(log, reading_jobs)
    finally:
        clear_progress()


def _generate_all_lists(log, reading_jobs=None):
    import library

    every = library.lists()
    # The jobs are passed only when asked for, so a caller that builds lists
    # without the background reading calls generate_master_list() as before.
    extra = {} if reading_jobs is None else {"reading_jobs": reading_jobs}
    if len(every) == 1:
        return generate_master_list(**extra)

    failed = []
    for entry in every:
        log(f"[LIST-GEN] Building {entry.name!r} ({len(entry.folders)} folder(s))...")
        try:
            if not generate_master_list(entry.name, **extra):
                failed.append(entry.name)
        except Exception as err:
            # One list's unexpected failure is not every list's. The scan
            # handles the failures it can predict; this is the backstop for the
            # ones it cannot, and catching it here is the difference between
            # "two of your three lists rebuilt" and "the update crashed".
            log(f"[LIST-GEN ERROR] {entry.name!r} failed: {err}")
            failed.append(entry.name)

    if failed:
        log("[LIST-GEN ERROR] These lists were not rebuilt and are still serving "
            "what they last built: " + ", ".join(repr(n) for n in failed))
        return False
    log(f"[LIST-GEN] All {len(every)} lists rebuilt.")
    return True


def rebuild_main(log=print):
    """`update_list.py` with no argument: rebuild every list, then - started
    by the bot - read the audio lengths left. Returns the exit code."""
    log("--- Starting the scheduled weekly file-list update ---")
    # FILE_DIRECTORY is not in settings_file.REQUIRED (see its own comment) -
    # a blank value is a supported "not chosen yet" state the daemon itself
    # boots fine with, so os.path.exists(None) here (a TypeError, not a
    # clean failure) must be guarded against explicitly rather than assuming
    # a real string ever reaches this point.
    # Across EVERY list, not just the primary. The two guards below ask
    # "is there anything to build at all", and with several lists the
    # answer is yes if any one of them has somewhere to look - a single
    # list pointed at an unplugged drive is generate_all_lists()'s
    # problem to report, not a reason to refuse the whole run.
    configured = [folder for entry in library.lists() for folder in entry.folders]
    if not configured:
        log("[CRITICAL] No music directory configured yet - set FILE_DIRECTORY "
            "from the web dashboard's Settings page, settings.conf, or "
            "admin_config.py before running this.")
        return 1
    # Every one missing, not any one: a single unavailable folder is skipped
    # during the scan with a warning, and only a library with nothing readable
    # in it at all is worth refusing to run for.
    if not any(os.path.isdir(platform_compat.long_path(f.path)) for f in configured):
        log("[CRITICAL] None of the configured music folders exist: "
            + ", ".join(f.path for f in configured))
        return 1

    # ONE AT A TIME (#1182 audit): a reading holding the list - the bot's, a
    # terminal's, or one whose bot died - is asked to stop and waited for; a
    # second rebuild is not started at all. Two of them wrote the same
    # temporary files and could publish a list made of both.
    holder = take_list_lock("rebuild")
    if holder is not None:
        log(f"[LIST-GEN ERROR] Another list rebuild is already running (pid {holder[0] or '?'}); "
            f"not starting a second one.")
        log("--- ERROR: could not generate the list. ---")
        return 1

    # The file is removed whichever way the run ends, including the failure
    # branch: a leftover from a crashed or killed process reads as a rebuild
    # still in progress, and the dashboard would show a bar that never moves.
    # generate_all_lists() clears the progress file in its own finally, so a
    # crash inside it still leaves nothing behind for the dashboard to read.
    try:
        return _rebuild_holding_the_lock(log)
    finally:
        release_list_lock()


def _rebuild_holding_the_lock(log):
    reading_jobs = []
    success = generate_all_lists(log=log, reading_jobs=reading_jobs)
    if success:
        log("--- The list was updated successfully. ---")
        # Every list is published; what is left to read is read now, in this
        # same process, while the bot serves the new list (#1182). The bot
        # sees the "reading" phase and stops waiting for us. A rebuild that
        # failed reads nothing: the next one, or a reading-only run, does.
        # Run by hand (a terminal, cron, configure.py) it does not: hours of
        # silent waiting there is nobody's intention, so it says how instead.
        if reading_jobs and started_by_the_bot():
            mark_list_lock("reading")
            run_audio_reading(reading_jobs, log=log)
        elif reading_jobs:
            waiting = sum(len(job.pending) for job in reading_jobs)
            log(f"[AUDIO-INFO] {waiting:,} audio file(s) have no length and quality in the list "
                f"yet. The bot reads them: Tools > Read audio info on the dashboard, `audioinfo` "
                f"in its console, Library > Read audio info in dccore.mrc - or run "
                f"`update_list.py --read-audio-info`, which reads them here.")
        return 0
    else:
        log("--- ERROR: could not generate the list. ---")
        return 1


if __name__ == "__main__":
    # The reading-only run (#1182): the audio files the published lists have
    # no length for yet, read and written in - no scan, no rebuild.
    if "--read-audio-info" in sys.argv[1:]:
        outcome = read_audio_info_main()["outcome"]
        sys.exit(1 if outcome == "failed" else 0)

    sys.exit(rebuild_main())