updated series_index to float and added UniqueConstraint to audiobook and audiobook_author

frontend/.gitignore

@@ -1,24 +0,0 @@
-# Logs
-logs
-*.log
-npm-debug.log*
-yarn-debug.log*
-yarn-error.log*
-pnpm-debug.log*
-lerna-debug.log*
-
-node_modules
-dist
-dist-ssr
-*.local
-
-# Editor directories and files
-.vscode/*
-!.vscode/extensions.json
-.idea
-.DS_Store
-*.suo
-*.ntvs*
-*.njsproj
-*.sln
-*.sw?

python/alembic/richie/versions/2026_06_03-adding_audiobook_libreary_metadata_d7864d1ffc17.py

@@ -0,0 +1,93 @@
+"""adding audiobook libreary metadata.
+
+Revision ID: d7864d1ffc17
+Revises: c8a794340928
+Create Date: 2026-06-03 20:24:09.200837
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+
+from python.orm import RichieBase
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "d7864d1ffc17"
+down_revision: str | None = "c8a794340928"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+schema = RichieBase.schema_name
+
+
+def upgrade() -> None:
+    """Upgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "audiobook_author",
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_audiobook_author")),
+        sa.UniqueConstraint("name", name=op.f("uq_audiobook_author_name")),
+        schema=schema,
+    )
+    op.create_table(
+        "audiobook_series",
+        sa.Column("name", sa.String(), nullable=False),
+        sa.Column("author_id", sa.Integer(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["author_id"],
+            [f"{schema}.audiobook_author.id"],
+            name=op.f("fk_audiobook_series_author_id_audiobook_author"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_audiobook_series")),
+        sa.UniqueConstraint("author_id", "name", name=op.f("uq_audiobook_series_author_id")),
+        schema=schema,
+    )
+    op.create_table(
+        "audiobook",
+        sa.Column("title", sa.String(), nullable=False),
+        sa.Column("author_id", sa.Integer(), nullable=False),
+        sa.Column("series_id", sa.Integer(), nullable=True),
+        sa.Column("series_index", sa.Integer(), nullable=False),
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("created", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["author_id"],
+            [f"{schema}.audiobook_author.id"],
+            name=op.f("fk_audiobook_author_id_audiobook_author"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["series_id"],
+            [f"{schema}.audiobook_series.id"],
+            name=op.f("fk_audiobook_series_id_audiobook_series"),
+            ondelete="SET NULL",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_audiobook")),
+        schema=schema,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("audiobook", schema=schema)
+    op.drop_table("audiobook_series", schema=schema)
+    op.drop_table("audiobook_author", schema=schema)
+    # ### end Alembic commands ###

python/alembic/richie/versions/2026_06_10-updated_series_index_to_float_and_added__b3c60cc5beb5.py

@@ -0,0 +1,63 @@
+"""updated series_index to float and added UniqueConstraint to audiobook and audiobook_author.
+
+Revision ID: b3c60cc5beb5
+Revises: d7864d1ffc17
+Create Date: 2026-06-10 20:02:43.073725
+
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import sqlalchemy as sa
+from alembic import op
+
+from python.orm import RichieBase
+
+if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "b3c60cc5beb5"
+down_revision: str | None = "d7864d1ffc17"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+schema = RichieBase.schema_name
+
+
+def upgrade() -> None:
+    """Upgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column(
+        "audiobook",
+        "series_index",
+        existing_type=sa.INTEGER(),
+        type_=sa.Float(),
+        existing_nullable=False,
+        schema=schema,
+    )
+    op.create_unique_constraint(
+        op.f("uq_audiobook_author_id"),
+        "audiobook",
+        ["author_id", "series_id", "title"],
+        schema=schema,
+        postgresql_nulls_not_distinct=True,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(op.f("uq_audiobook_author_id"), "audiobook", schema=schema, type_="unique")
+    op.alter_column(
+        "audiobook",
+        "series_index",
+        existing_type=sa.Float(),
+        type_=sa.INTEGER(),
+        existing_nullable=False,
+        schema=schema,
+    )
+    # ### end Alembic commands ###

python/installer/old_installer.py

@@ -16,9 +16,13 @@ from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
+logger = logging.getLogger(__name__)
+ESCAPE_KEY = 27
+
 
 def configure_logger(level: str = "INFO") -> None:
     """Configure the logger.
+
     Args:
         level (str, optional): The logging level. Defaults to "INFO".
     """
@@ -32,15 +36,17 @@ def configure_logger(level: str = "INFO") -> None:
 
 def bash_wrapper(command: str) -> str:
     """Execute a bash command and capture the output.
+
     Args:
         command (str): The bash command to be executed.
+
     Returns:
         Tuple[str, int]: A tuple containing the output of the command (stdout) as a string,
         the error output (stderr) as a string (optional), and the return code as an integer.
     """
-    logging.debug(f"running {command=}")
+    logger.debug(f"running {command=}")
     # This is a acceptable risk
-    process = Popen(command.split(), stdout=PIPE, stderr=PIPE)  # noqa: S603
+    process = Popen(command.split(), stdout=PIPE, stderr=PIPE)
     output, _ = process.communicate()
     if process.returncode != 0:
         error = f"Failed to run command {command=} return code {process.returncode=}"
@@ -51,6 +57,7 @@ def bash_wrapper(command: str) -> str:
 
 def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None:
     """Partition a disk.
+
     Args:
         disk (str): The disk to partition.
         swap_size (int): The size of the swap partition in GB.
@@ -58,7 +65,7 @@ def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None:
         reserve (int, optional): The size of the reserve partition in GB. Defaults to 0.
             minimum value is 0.
     """
-    logging.info(f"partitioning {disk=}")
+    logger.info(f"partitioning {disk=}")
     swap_size = max(swap_size, 1)
     reserve = max(reserve, 0)
 
@@ -66,16 +73,16 @@ def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None:
 
     if reserve > 0:
         msg = f"Creating swap partition on {disk=} with size {swap_size=}GiB and reserve {reserve=}GiB"
-        logging.info(msg)
+        logger.info(msg)
 
         swap_start = swap_size + reserve
         swap_partition = f"mkpart swap -{swap_start}GiB -{reserve}GiB "
     else:
-        logging.info(f"Creating swap partition on {disk=} with size {swap_size=}GiB")
+        logger.info(f"Creating swap partition on {disk=} with size {swap_size=}GiB")
         swap_start = swap_size
         swap_partition = f"mkpart swap -{swap_start}GiB 100% "
 
-    logging.debug(f"{swap_partition=}")
+    logger.debug(f"{swap_partition=}")
 
     create_partitions = (
         f"parted --script --align=optimal {disk} -- "
@@ -87,13 +94,14 @@ def partition_disk(disk: str, swap_size: int, reserve: int = 0) -> None:
     )
     bash_wrapper(create_partitions)
 
-    logging.info(f"{disk=} successfully partitioned")
+    logger.info(f"{disk=} successfully partitioned")
 
 
 def create_zfs_pool(pool_disks: Sequence[str], mnt_dir: str) -> None:
     """Create a ZFS pool.
+
     Args:
-        disks (Sequence[str]): A tuple of disks to use for the pool.
+        pool_disks (Sequence[str]): A tuple of disks to use for the pool.
         mnt_dir (str): The mount directory.
     """
     if len(pool_disks) <= 0:
@@ -125,13 +133,12 @@ def create_zfs_pool(pool_disks: Sequence[str], mnt_dir: str) -> None:
     bash_wrapper(zpool_create)
     zpools = bash_wrapper("zpool list -o name")
     if "root_pool" not in zpools.splitlines():
-        logging.critical("Failed to create root_pool")
+        logger.critical("Failed to create root_pool")
         sys.exit(1)
 
 
 def create_zfs_datasets() -> None:
     """Create ZFS datasets."""
-
     bash_wrapper("zfs create -o canmount=noauto -o reservation=10G root_pool/root")
     bash_wrapper("zfs create root_pool/home")
     bash_wrapper("zfs create root_pool/var -o reservation=1G")
@@ -146,7 +153,7 @@ def create_zfs_datasets() -> None:
     }
     missing_datasets = expected_datasets.difference(datasets.splitlines())
     if missing_datasets:
-        logging.critical(f"Failed to create pools {missing_datasets}")
+        logger.critical(f"Failed to create pools {missing_datasets}")
         sys.exit(1)
 
 
@@ -159,6 +166,8 @@ def get_cpu_manufacturer() -> str:
     for line in output.splitlines():
         if "vendor_id" in line:
             return id_vendor[line.split(": ")[1].strip()]
+    error = "Failed to get CPU manufacturer"
+    raise RuntimeError(error)
 
 
 def get_boot_drive_id(disk: str) -> str:
@@ -167,9 +176,8 @@ def get_boot_drive_id(disk: str) -> str:
     return output.splitlines()[1]
 
 
-def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], encrypt: bool) -> None:
+def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], *, encrypt: bool) -> None:
     """Create a NixOS hardware file."""
-
     cpu_manufacturer = get_cpu_manufacturer()
 
     devices = ""
@@ -193,7 +201,15 @@ def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], encrypt: bool)
         '  imports = [ (modulesPath + "/installer/scan/not-detected.nix") ];\n\n'
         "  boot = {\n"
         "    initrd = {\n"
-        '      availableKernelModules = [ \n        "ahci"\n        "ehci_pci"\n        "nvme"\n        "sd_mod"\n        "usb_storage"\n        "usbhid"\n        "xhci_pci"\n      ];\n'
+        "      availableKernelModules = [ \n"
+        '        "ahci"\n'
+        '        "ehci_pci"\n'
+        '        "nvme"\n'
+        '        "sd_mod"\n'
+        '        "usb_storage"\n'
+        '        "usbhid"\n'
+        '        "xhci_pci"\n'
+        "      ];\n"
         "      kernelModules = [ ];\n"
         f" {devices}"
         "    };\n"
@@ -207,11 +223,18 @@ def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], encrypt: bool)
         '    "/nix" = {\n      device = "root_pool/nix";\n      fsType = "zfs";\n    };\n\n'
         '    "/boot" = {\n'
         f'      device = "/dev/disk/by-uuid/{get_boot_drive_id(disks[0])}";\n'
-        '      fsType = "vfat";\n      options = [\n        "fmask=0077"\n        "dmask=0077"\n      ];\n    };\n  };\n\n'
+        '      fsType = "vfat";\n'
+        "      options = [\n"
+        '        "fmask=0077"\n'
+        '        "dmask=0077"\n'
+        "      ];\n"
+        "    };\n"
+        "  };\n\n"
         "  swapDevices = [ ];\n\n"
         "  networking.useDHCP = lib.mkDefault true;\n\n"
         '  nixpkgs.hostPlatform = lib.mkDefault "x86_64-linux";\n'
-        f"  hardware.cpu.{cpu_manufacturer}.updateMicrocode = lib.mkDefault config.hardware.enableRedistributableFirmware;\n"
+        f"  hardware.cpu.{cpu_manufacturer}.updateMicrocode = lib.mkDefault "
+        "config.hardware.enableRedistributableFirmware;\n"
         f'  networking.hostId = "{host_id}";\n'
         "}\n"
     )
@@ -219,7 +242,7 @@ def create_nix_hardware_file(mnt_dir: str, disks: Sequence[str], encrypt: bool)
     Path(f"{mnt_dir}/etc/nixos/hardware-configuration.nix").write_text(nix_hardware)
 
 
-def install_nixos(mnt_dir: str, disks: Sequence[str], encrypt: bool) -> None:
+def install_nixos(mnt_dir: str, disks: Sequence[str], *, encrypt: bool) -> None:
     """Install NixOS."""
     bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/root {mnt_dir}")
     bash_wrapper(f"mount -o X-mount.mkdir -t zfs root_pool/home {mnt_dir}/home")
@@ -230,14 +253,16 @@ def install_nixos(mnt_dir: str, disks: Sequence[str], encrypt: bool) -> None:
         bash_wrapper(f"mkfs.vfat -n EFI {disk}-part1")
 
     # set up mirroring afterwards if more than one disk
-    boot_partition = f"mount -t vfat -o fmask=0077,dmask=0077,iocharset=iso8859-1,X-mount.mkdir {disks[0]}-part1 {mnt_dir}/boot"
+    boot_partition = (
+        f"mount -t vfat -o fmask=0077,dmask=0077,iocharset=iso8859-1,X-mount.mkdir {disks[0]}-part1 {mnt_dir}/boot"
+    )
     bash_wrapper(boot_partition)
 
     bash_wrapper(f"nixos-generate-config --root {mnt_dir}")
 
-    create_nix_hardware_file(mnt_dir, disks, encrypt)
+    create_nix_hardware_file(mnt_dir, disks, encrypt=encrypt)
 
-    run(("nixos-install", "--root", mnt_dir), check=True)  # noqa: S603
+    run(("nixos-install", "--root", mnt_dir), check=True)
 
 
 def installer(
@@ -247,27 +272,37 @@ def installer(
     encrypt_key: str | None,
 ) -> None:
     """Main."""
-    logging.info("Starting installation")
+    logger.info("Starting installation")
 
     for disk in disks:
         partition_disk(disk, swap_size, reserve)
 
         if encrypt_key:
             sleep(1)
-            for command in (
-                f'printf "{encrypt_key}" | cryptsetup luksFormat --type luks2 {disk}-part2 -',
-                f'printf "{encrypt_key}" | cryptsetup luksOpen {disk}-part2 luks-root-pool-{disk.split("/")[-1]}-part2 -',
-            ):
-                run(command, shell=True, check=True)
+            key_input = encrypt_key.encode()
+            run(
+                ("cryptsetup", "luksFormat", "--type", "luks2", f"{disk}-part2", "-"),
+                input=key_input,
+                check=True,
+            )
+            run(
+                (
+                    "cryptsetup",
+                    "luksOpen",
+                    f"{disk}-part2",
+                    f"luks-root-pool-{disk.split('/')[-1]}-part2",
+                    "-",
+                ),
+                input=key_input,
+                check=True,
+            )
 
     mnt_dir = "/tmp/nix_install"  # noqa: S108
 
     Path(mnt_dir).mkdir(parents=True, exist_ok=True)
 
     if encrypt_key:
-        pool_disks = [
-            f"/dev/mapper/luks-root-pool-{disk.split('/')[-1]}-part2" for disk in disks
-        ]
+        pool_disks = [f"/dev/mapper/luks-root-pool-{disk.split('/')[-1]}-part2" for disk in disks]
     else:
         pool_disks = [f"{disk}-part2" for disk in disks]
 
@@ -275,57 +310,73 @@ def installer(
 
     create_zfs_datasets()
 
-    install_nixos(mnt_dir, disks, encrypt_key)
+    install_nixos(mnt_dir, disks, encrypt=bool(encrypt_key))
 
-    logging.info("Installation complete")
+    logger.info("Installation complete")
 
 
 class Cursor:
-    def __init__(self):
+    """Track cursor position and constrain movement to screen bounds."""
+
+    def __init__(self) -> None:
+        """Initialize cursor position and screen dimensions."""
         self.x_position = 0
         self.y_position = 0
         self.height = 0
         self.width = 0
 
-    def set_height(self, height: int):
+    def set_height(self, height: int) -> None:
+        """Set the maximum screen height."""
         self.height = height
 
-    def set_width(self, width: int):
+    def set_width(self, width: int) -> None:
+        """Set the maximum screen width."""
         self.width = width
 
     def x_bounce_check(self, cursor: int) -> int:
+        """Clamp an x position to the screen width."""
         cursor = max(0, cursor)
         return min(self.width - 1, cursor)
 
     def y_bounce_check(self, cursor: int) -> int:
+        """Clamp a y position to the screen height."""
         cursor = max(0, cursor)
         return min(self.height - 1, cursor)
 
-    def set_x(self, x: int):
+    def set_x(self, x: int) -> None:
+        """Set the cursor x position."""
         self.x_position = self.x_bounce_check(x)
 
-    def set_y(self, y: int):
+    def set_y(self, y: int) -> None:
+        """Set the cursor y position."""
         self.y_position = self.y_bounce_check(y)
 
     def get_x(self) -> int:
+        """Get the cursor x position."""
         return self.x_position
 
     def get_y(self) -> int:
+        """Get the cursor y position."""
         return self.y_position
 
-    def move_up(self):
+    def move_up(self) -> None:
+        """Move the cursor up one row."""
         self.set_y(self.y_position - 1)
 
-    def move_down(self):
+    def move_down(self) -> None:
+        """Move the cursor down one row."""
         self.set_y(self.y_position + 1)
 
-    def move_left(self):
+    def move_left(self) -> None:
+        """Move the cursor left one column."""
         self.set_x(self.x_position - 1)
 
-    def move_right(self):
+    def move_right(self) -> None:
+        """Move the cursor right one column."""
         self.set_x(self.x_position + 1)
 
     def navigation(self, key: int) -> None:
+        """Move the cursor for a curses navigation key."""
         action = {
             curses.KEY_DOWN: self.move_down,
             curses.KEY_UP: self.move_up,
@@ -339,7 +390,8 @@ class Cursor:
 class State:
     """State class to store the state of the program."""
 
-    def __init__(self):
+    def __init__(self) -> None:
+        """Initialize installer menu state."""
         self.key = 0
         self.cursor = Cursor()
 
@@ -357,11 +409,9 @@ class State:
 
 
 def get_device(raw_device: str) -> dict[str, str]:
+    """Parse an lsblk key-value device row."""
     raw_device_components = raw_device.split(" ")
-    return {
-        thing.split("=")[0].lower(): thing.split("=")[1].strip('"')
-        for thing in raw_device_components
-    }
+    return {thing.split("=")[0].lower(): thing.split("=")[1].strip('"') for thing in raw_device_components}
 
 
 def get_devices() -> list[dict[str, str]]:
@@ -373,6 +423,7 @@ def get_devices() -> list[dict[str, str]]:
 
 def get_device_id_mapping() -> dict[str, set[str]]:
     """Get a list of device ids.
+
     Returns:
         list[str]: the list of device ids
     """
@@ -387,9 +438,8 @@ def get_device_id_mapping() -> dict[str, set[str]]:
     return device_id_mapping
 
 
-def calculate_device_menu_padding(
-    devices: list[dict[str, str]], column: str, padding: int = 0
-) -> int:
+def calculate_device_menu_padding(devices: list[dict[str, str]], column: str, padding: int = 0) -> int:
+    """Calculate the width needed for a device menu column."""
     return max(len(device[column]) for device in devices) + padding
 
 
@@ -401,6 +451,7 @@ def draw_device_ids(
     menu_width: list[int],
     device_ids: set[str],
 ) -> tuple[State, int]:
+    """Draw selectable device IDs for a device row."""
     for device_id in sorted(device_ids):
         row_number = row_number + 1
         if row_number == state.cursor.get_y() and state.cursor.get_x() in menu_width:
@@ -429,8 +480,9 @@ def draw_device_menu(
     state: State,
     menu_start_y: int = 0,
     menu_start_x: int = 0,
-) -> State:
-    """draw the device menu and handle user input
+) -> tuple[State, int]:
+    """Draw the device menu and handle user input.
+
     Args:
         std_screen (curses.window): the curses window to draw on
         devices (list[dict[str, str]]): the list of devices to draw
@@ -438,6 +490,7 @@ def draw_device_menu(
         state (State): the state object to update
         menu_start_y (int, optional): the y position to start drawing the menu. Defaults to 0.
         menu_start_x (int, optional): the x position to start drawing the menu. Defaults to 0.
+
     Returns:
         State: the updated state object
     """
@@ -448,7 +501,9 @@ def draw_device_menu(
     type_padding = calculate_device_menu_padding(devices, "type", padding)
     mountpoints_padding = calculate_device_menu_padding(devices, "mountpoints", padding)
 
-    device_header = f"{'Name':{name_padding}}{'Size':{size_padding}}{'Type':{type_padding}}{'Mountpoints':{mountpoints_padding}}"
+    device_header = (
+        f"{'Name':{name_padding}}{'Size':{size_padding}}{'Type':{type_padding}}{'Mountpoints':{mountpoints_padding}}"
+    )
 
     menu_width = range(menu_start_x, len(device_header) + menu_start_x)
 
@@ -481,8 +536,9 @@ def draw_device_menu(
 
 
 def debug_menu(std_screen: curses.window, key: int) -> None:
+    """Draw debug information for the current curses screen."""
     height, width = std_screen.getmaxyx()
-    width_height = "Width: {}, Height: {}".format(width, height)
+    width_height = f"Width: {width}, Height: {height}"
     std_screen.addstr(height - 4, 0, width_height, curses.color_pair(5))
 
     key_pressed = f"Last key pressed: {key}"[: width - 1]
@@ -490,7 +546,7 @@ def debug_menu(std_screen: curses.window, key: int) -> None:
         key_pressed = "No key press detected..."[: width - 1]
     std_screen.addstr(height - 3, 0, key_pressed)
 
-    for i in range(0, 8):
+    for i in range(8):
         std_screen.addstr(height - 2, i * 3, f"{i}██", curses.color_pair(i))
 
 
@@ -500,12 +556,11 @@ def status_bar(
     width: int,
     height: int,
 ) -> None:
+    """Draw the footer status bar."""
     std_screen.attron(curses.A_REVERSE)
     std_screen.attron(curses.color_pair(3))
 
-    status_bar = (
-        f"Press 'q' to exit | STATUS BAR | Pos: {cursor.get_x()}, {cursor.get_y()}"
-    )
+    status_bar = f"Press 'q' to exit | STATUS BAR | Pos: {cursor.get_x()}, {cursor.get_y()}"
     std_screen.addstr(height - 1, 0, status_bar)
     std_screen.addstr(height - 1, len(status_bar), " " * (width - len(status_bar) - 1))
 
@@ -514,13 +569,15 @@ def status_bar(
 
 
 def set_color() -> None:
+    """Initialize curses color pairs."""
     curses.start_color()
     curses.use_default_colors()
-    for i in range(0, curses.COLORS):
+    for i in range(curses.COLORS):
         curses.init_pair(i + 1, i, -1)
 
 
 def get_text_input(std_screen: curses.window, prompt: str, y: int, x: int) -> str:
+    """Read text input from a curses screen."""
     curses.echo()
     std_screen.addstr(y, x, prompt)
     input_str = ""
@@ -528,10 +585,10 @@ def get_text_input(std_screen: curses.window, prompt: str, y: int, x: int) -> st
         key = std_screen.getch()
         if key == ord("\n"):
             break
-        elif key == 27:  # ESC key
+        if key == ESCAPE_KEY:
             input_str = ""
             break
-        elif key in (curses.KEY_BACKSPACE, ord("\b"), 127):
+        if key in (curses.KEY_BACKSPACE, ord("\b"), 127):
             input_str = input_str[:-1]
             std_screen.addstr(y, x + len(prompt), input_str + " ")
         else:
@@ -546,6 +603,7 @@ def swap_size_input(
     state: State,
     swap_offset: int,
 ) -> State:
+    """Handle swap size input."""
     swap_size_text = "Swap size (GB): "
     std_screen.addstr(swap_offset, 0, f"{swap_size_text}{state.swap_size}")
     if state.key == ord("\n") and state.cursor.get_y() == swap_offset:
@@ -557,9 +615,7 @@ def swap_size_input(
             state.swap_size = int(swap_size_str)
             state.show_swap_input = False
         except ValueError:
-            std_screen.addstr(
-                swap_offset, 0, "Invalid input. Press any key to continue."
-            )
+            std_screen.addstr(swap_offset, 0, "Invalid input. Press any key to continue.")
             std_screen.getch()
             state.show_swap_input = False
 
@@ -571,22 +627,19 @@ def reserve_size_input(
     state: State,
     reserve_offset: int,
 ) -> State:
+    """Handle reserve size input."""
     reserve_size_text = "reserve size (GB): "
     std_screen.addstr(reserve_offset, 0, f"{reserve_size_text}{state.reserve_size}")
     if state.key == ord("\n") and state.cursor.get_y() == reserve_offset:
         state.show_reserve_input = True
 
     if state.show_reserve_input:
-        reserve_size_str = get_text_input(
-            std_screen, reserve_size_text, reserve_offset, 0
-        )
+        reserve_size_str = get_text_input(std_screen, reserve_size_text, reserve_offset, 0)
         try:
             state.reserve_size = int(reserve_size_str)
             state.show_reserve_input = False
         except ValueError:
-            std_screen.addstr(
-                reserve_offset, 0, "Invalid input. Press any key to continue."
-            )
+            std_screen.addstr(reserve_offset, 0, "Invalid input. Press any key to continue.")
             std_screen.getch()
             state.show_reserve_input = False
 
@@ -594,9 +647,11 @@ def reserve_size_input(
 
 
 def draw_menu(std_screen: curses.window) -> State:
-    """draw the menu and handle user input
+    """Draw the menu and handle user input.
+
     Args:
         std_screen (curses.window): the curses window to draw on
+
     Returns:
         State: the state object
     """
@@ -656,17 +711,18 @@ def draw_menu(std_screen: curses.window) -> State:
 
 
 def main() -> None:
+    """Run the installer menu and start installation."""
     configure_logger("DEBUG")
 
     state = curses.wrapper(draw_menu)
 
     encrypt_key = getenv("ENCRYPT_KEY")
 
-    logging.info("installing_nixos")
-    logging.info(f"disks: {state.selected_device_ids}")
-    logging.info(f"swap_size: {state.swap_size}")
-    logging.info(f"reserve: {state.reserve_size}")
-    logging.info(f"encrypted: {bool(encrypt_key)}")
+    logger.info("installing_nixos")
+    logger.info(f"disks: {state.selected_device_ids}")
+    logger.info(f"swap_size: {state.swap_size}")
+    logger.info(f"reserve: {state.reserve_size}")
+    logger.info(f"encrypted: {bool(encrypt_key)}")
 
     sleep(3)
 

python/orm/richie/__init__.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from python.orm.richie.audiobook import Audiobook, AudiobookAuthor, AudiobookSeries
 from python.orm.richie.base import RichieBase, TableBase, TableBaseBig, TableBaseSmall
 from python.orm.richie.contact import (
     Contact,
@@ -12,6 +13,9 @@ from python.orm.richie.contact import (
 )
 
 __all__ = [
+    "Audiobook",
+    "AudiobookAuthor",
+    "AudiobookSeries",
     "Contact",
     "ContactNeed",
     "ContactRelationship",

python/orm/richie/audiobook.py

@@ -0,0 +1,55 @@
+"""Audiobook catalog models."""
+
+from __future__ import annotations
+
+from sqlalchemy import ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from python.orm.richie.base import TableBase
+
+
+class AudiobookAuthor(TableBase):
+    """Canonical audiobook author."""
+
+    __tablename__ = "audiobook_author"
+    __table_args__ = (UniqueConstraint("name"),)
+
+    name: Mapped[str] = mapped_column(String, unique=True)
+
+    books: Mapped[list[Audiobook]] = relationship("Audiobook", back_populates="author")
+    series: Mapped[list[AudiobookSeries]] = relationship("AudiobookSeries", back_populates="author")
+
+
+class AudiobookSeries(TableBase):
+    """Canonical audiobook series."""
+
+    __tablename__ = "audiobook_series"
+    __table_args__ = (UniqueConstraint("author_id", "name"),)
+
+    name: Mapped[str] = mapped_column(String)
+    author_id: Mapped[int] = mapped_column(ForeignKey("main.audiobook_author.id", ondelete="CASCADE"))
+
+    author: Mapped[AudiobookAuthor] = relationship("AudiobookAuthor", back_populates="series")
+    books: Mapped[list[Audiobook]] = relationship("Audiobook", back_populates="series")
+
+
+class Audiobook(TableBase):
+    """Canonical audiobook title."""
+
+    __tablename__ = "audiobook"
+    __table_args__ = (
+        UniqueConstraint(
+            "author_id",
+            "series_id",
+            "title",
+            postgresql_nulls_not_distinct=True,
+        ),
+    )
+
+    title: Mapped[str] = mapped_column(String)
+    author_id: Mapped[int] = mapped_column(ForeignKey("main.audiobook_author.id", ondelete="CASCADE"))
+    series_id: Mapped[int | None] = mapped_column(ForeignKey("main.audiobook_series.id", ondelete="SET NULL"))
+    series_index: Mapped[float] = mapped_column(default=0.0)
+
+    author: Mapped[AudiobookAuthor] = relationship("AudiobookAuthor", back_populates="books")
+    series: Mapped[AudiobookSeries | None] = relationship("AudiobookSeries", back_populates="books")

python/tools/audiobook/__init__.py

@@ -0,0 +1 @@
+"""Audiobook tools."""

python/tools/audiobook/audible_convert.py

@@ -0,0 +1,471 @@
+"""Convert Audible AAX downloads into Audiobookshelf-friendly M4B files."""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import shutil
+import subprocess
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import asdict, dataclass
+from os import getenv
+from pathlib import Path  # noqa: TC003 This is required for the typer CLI
+from typing import TYPE_CHECKING, Annotated, Any
+from uuid import uuid7
+
+import typer
+
+from python.common import configure_logger
+from python.orm.common import get_postgres_engine
+from python.tools.audiobook.metadata_agent import (
+    AgentConfig,
+    StandardBookMetadata,
+    standard_book_metadata,
+    write_agent_log,
+)
+
+if TYPE_CHECKING:
+    from sqlalchemy.engine import Engine
+
+logger = logging.getLogger(__name__)
+
+SENSITIVE_COMMAND_ARGUMENTS = {"-activation_bytes"}
+BOOK_RANGE_PATTERN = re.compile(r"(?:^|-)books?-(?P<start>[1-9]\d*)-(?P<end>[1-9]\d*)(?:-|$)")
+
+
+@dataclass(frozen=True)
+class ConversionConfig:
+    """Runtime settings for one conversion command."""
+
+    resolved_output: Path
+    ollama_api_key: str
+    agent_config: AgentConfig
+    engine: Engine
+    activation_bytes: str | None
+    dry_run: bool
+    overwrite: bool
+    work_directory_name: str = ".audible_convert"
+    dry_run_directory_name: str = "dry-run"
+    temp_directory_name: str = "tmp"
+    log_directory_name: str = "logs"
+    review_directory_name: str = "review"
+
+
+@dataclass(frozen=True)
+class ConcurrentConversionResult:
+    """Result from running ffmpeg and metadata resolution together."""
+
+    metadata: StandardBookMetadata | None
+    conversion_error: Exception | None
+    metadata_error: Exception | None
+
+
+class CommandExecutionError(RuntimeError):
+    """Command failed without exposing sensitive arguments."""
+
+    def __init__(self, arguments: list[str], returncode: int) -> None:
+        """Create a redacted command failure."""
+        self.arguments = tuple(arguments)
+        self.returncode = returncode
+        command = " ".join(redact_command_arguments(arguments))
+        super().__init__(f"Command failed with exit code {returncode}: {command}")
+
+
+def main(
+    input_directory: Annotated[Path, typer.Argument(help="Directory audible-cli downloads AAX files into.")],
+    output_directory: Annotated[Path, typer.Argument(help="Audiobook output directory.")],
+    *,
+    dry_run: Annotated[
+        bool,
+        typer.Option("--dry-run", help="Print planned output files and write marker files without converting."),
+    ] = False,
+    overwrite: Annotated[bool, typer.Option("--overwrite", help="Overwrite existing M4B files.")] = False,
+) -> None:
+    """Convert AAX files from a download directory into M4B files."""
+    configure_logger()
+    resolved_input = input_directory.resolve(strict=True)
+    resolved_output = output_directory.resolve()
+    if not dry_run:
+        resolved_output.mkdir(parents=True, exist_ok=True)
+
+    ollama_api_key = getenv("OLLAMA_API_KEY")
+    if not ollama_api_key:
+        msg = "OLLAMA_API_KEY is required for audiobook metadata resolution"
+        raise RuntimeError(msg)
+
+    config = ConversionConfig(
+        resolved_output=resolved_output,
+        ollama_api_key=ollama_api_key,
+        agent_config=AgentConfig(),
+        engine=get_postgres_engine(name="RICHIE"),
+        activation_bytes=getenv("AUDIBLE_ACTIVATION_BYTES"),
+        dry_run=dry_run,
+        overwrite=overwrite,
+    )
+
+    aax_files = sorted(resolved_input.glob("*.aax"))
+    if not aax_files:
+        logger.info("No AAX files found in %s", resolved_input)
+        return
+    for aax_file in aax_files:
+        logger.info("Converting %s", aax_file)
+        convert_aax_file_with_agent(aax_file, config)
+
+
+def run_command(arguments: list[str], *, capture: bool = False) -> subprocess.CompletedProcess[str]:
+    """Run a command and return the completed process.
+
+    Args:
+        arguments: Command and arguments to run.
+        capture: Whether to capture stdout and stderr.
+
+    Returns:
+        The completed process.
+    """
+    logger.debug("%s", " ".join(redact_command_arguments(arguments)))
+    try:
+        return subprocess.run(arguments, check=True, capture_output=capture, text=True)
+    except subprocess.CalledProcessError as error:
+        raise CommandExecutionError(arguments, error.returncode) from error
+
+
+def redact_command_arguments(arguments: list[str]) -> list[str]:
+    """Return command arguments with sensitive values redacted."""
+    redacted = []
+    redact_next = False
+    for argument in arguments:
+        if redact_next:
+            redacted.append("<redacted>")
+            redact_next = False
+            continue
+
+        redacted.append(argument)
+        redact_next = argument in SENSITIVE_COMMAND_ARGUMENTS
+    return redacted
+
+
+def read_metadata(aax_file: Path) -> dict[str, str]:
+    """Read ffprobe format tags from an AAX file.
+
+    Args:
+        aax_file: AAX file to inspect.
+
+    Returns:
+        Lower-cased metadata tag names mapped to their values.
+    """
+    completed = run_command(
+        [
+            "ffprobe",
+            "-v",
+            "quiet",
+            "-print_format",
+            "json",
+            "-show_format",
+            str(aax_file),
+        ],
+        capture=True,
+    )
+    ffprobe_data: dict[str, Any] = json.loads(completed.stdout)
+    tags = ffprobe_data.get("format", {}).get("tags", {})
+    return {str(key).lower(): str(value) for key, value in tags.items()}
+
+
+def output_stem(metadata: StandardBookMetadata) -> str:
+    """Build the output stem for a book.
+
+    Args:
+        metadata: Book metadata.
+
+    Returns:
+        Output stem in author-series_01-title form.
+    """
+    index_slug = series_index_slug(metadata.series_index, metadata.title)
+    return f"{metadata.author}-{metadata.series}_{index_slug}-{metadata.title}"
+
+
+def series_index_slug(series_index: float, title: str = "") -> str:
+    """Return a filename-safe series index."""
+    if title_range := title_series_range_slug(series_index, title):
+        return title_range
+    index = float(series_index)
+    if index.is_integer():
+        return f"{int(index):02}"
+    return f"{int(index):02}.5"
+
+
+def title_series_range_slug(series_index: float, title: str) -> str | None:
+    """Return a series range slug found in an omnibus title."""
+    index = float(series_index)
+    if not index.is_integer():
+        return None
+    first_index = int(index)
+    for match in BOOK_RANGE_PATTERN.finditer(title):
+        start = int(match.group("start"))
+        end = int(match.group("end"))
+        if start == first_index and end > start:
+            return f"{start:02}-{end:02}"
+    return None
+
+
+def metadata_output_path(output_directory: Path, metadata: StandardBookMetadata) -> Path:
+    """Build the final M4B path from resolved metadata."""
+    stem = output_stem(metadata)
+    return output_directory / stem / f"{stem}.m4b"
+
+
+def convert_aax_file(
+    aax_file: Path,
+    destination: Path,
+    activation_bytes: str | None,
+    *,
+    overwrite: bool,
+) -> None:
+    """Convert an AAX file into an M4B file.
+
+    Args:
+        aax_file: Source AAX file.
+        destination: Destination M4B file.
+        activation_bytes: Optional Audible activation bytes for ffmpeg.
+        overwrite: Whether to overwrite an existing M4B.
+    """
+    if destination.exists() and not overwrite:
+        logger.info("Skipping existing file %s", destination)
+        return
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    arguments = ["ffmpeg", "-hide_banner", "-y" if overwrite else "-n"]
+    if activation_bytes:
+        arguments.extend(["-activation_bytes", activation_bytes])
+    arguments.extend(["-i", str(aax_file), "-map_metadata", "0", "-c", "copy", str(destination)])
+    run_command(arguments)
+
+
+def write_review_file(
+    *,
+    destination: Path | None,
+    ffprobe_metadata: dict[str, str],
+    log_file: Path,
+    metadata: StandardBookMetadata | None,
+    reason: str,
+    review_file: Path,
+    source: Path,
+    temp_file: Path | None,
+) -> None:
+    """Write a manual review file for an unresolved conversion."""
+    review_file.parent.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "destination": str(destination) if destination else None,
+        "ffprobe_metadata": ffprobe_metadata,
+        "metadata": asdict(metadata) if metadata else None,
+        "reason": reason,
+        "source": str(source),
+        "temp_file": str(temp_file) if temp_file else None,
+    }
+    review_file.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")
+    write_agent_log(log_file, "review_written", path=str(review_file), reason=reason)
+
+
+def cleanup_temp_output(temp_file: Path) -> None:
+    """Remove a run's temporary output directory."""
+    shutil.rmtree(temp_file.parent, ignore_errors=True)
+
+
+def dry_run_aax_file_with_agent(
+    aax_file: Path,
+    ffprobe_metadata: dict[str, str],
+    engine: Engine,
+    config: ConversionConfig,
+    log_file: Path,
+    review_file: Path,
+) -> None:
+    """Resolve and print the planned output path without converting."""
+    metadata = standard_book_metadata(
+        aax_file.name,
+        ffprobe_metadata,
+        engine,
+        log_file,
+        config.ollama_api_key,
+        config.agent_config,
+    )
+    destination = None if metadata.needs_review else metadata_output_path(config.resolved_output, metadata)
+    if metadata.needs_review:
+        write_review_file(
+            destination=destination,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=metadata,
+            reason="metadata_needs_review",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=None,
+        )
+        typer.echo(f"{aax_file} -> REVIEW {review_file}")
+    else:
+        stem = output_stem(metadata)
+        dry_run_file = (
+            config.resolved_output / config.work_directory_name / config.dry_run_directory_name / stem / f"{stem}.m4b"
+        )
+        dry_run_file.parent.mkdir(parents=True, exist_ok=True)
+        dry_run_file.write_text(f"{destination}\n", encoding="utf-8")
+        write_agent_log(
+            log_file,
+            "dry_run_file_written",
+            destination=str(destination),
+            path=str(dry_run_file),
+        )
+        typer.echo(f"{aax_file} -> {destination}")
+
+
+def convert_temp_file_and_resolve_metadata(
+    aax_file: Path,
+    temp_file: Path,
+    ffprobe_metadata: dict[str, str],
+    config: ConversionConfig,
+    log_file: Path,
+) -> ConcurrentConversionResult:
+    """Run ffmpeg and metadata resolution in parallel."""
+    conversion_error: Exception | None = None
+    metadata_error: Exception | None = None
+    metadata: StandardBookMetadata | None = None
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        conversion_future = executor.submit(
+            convert_aax_file,
+            aax_file,
+            temp_file,
+            config.activation_bytes,
+            overwrite=True,
+        )
+        metadata_future = executor.submit(
+            standard_book_metadata,
+            aax_file.name,
+            ffprobe_metadata,
+            config.engine,
+            log_file,
+            config.ollama_api_key,
+            config.agent_config,
+        )
+
+        conversion_error = conversion_future.exception()
+        if conversion_error is None:
+            conversion_future.result()
+
+        metadata_error = metadata_future.exception()
+        if metadata_error is None:
+            metadata = metadata_future.result()
+
+    return ConcurrentConversionResult(
+        metadata=metadata,
+        conversion_error=conversion_error,
+        metadata_error=metadata_error,
+    )
+
+
+def convert_aax_file_with_agent(aax_file: Path, config: ConversionConfig) -> None:
+    """Convert one AAX file using the metadata agent for the final path."""
+    run_id = uuid7().hex
+    log_file = config.resolved_output / config.work_directory_name / config.log_directory_name / f"{run_id}.jsonl"
+    review_file = config.resolved_output / config.work_directory_name / config.review_directory_name / f"{run_id}.json"
+    write_agent_log(log_file, "conversion_start", source=str(aax_file), dry_run=config.dry_run)
+    try:
+        ffprobe_metadata = read_metadata(aax_file)
+    except Exception as error:
+        logger.exception("ffprobe failed")
+        write_review_file(
+            destination=None,
+            ffprobe_metadata={},
+            log_file=log_file,
+            metadata=None,
+            reason=f"ffprobe_failed: {error}",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=None,
+        )
+        return
+
+    if config.dry_run:
+        dry_run_aax_file_with_agent(
+            aax_file,
+            ffprobe_metadata,
+            config.engine,
+            config,
+            log_file,
+            review_file,
+        )
+        return
+
+    temp_file = (
+        config.resolved_output / config.work_directory_name / config.temp_directory_name / run_id / "converted.m4b"
+    )
+    temp_file.parent.mkdir(parents=True, exist_ok=True)
+
+    result = convert_temp_file_and_resolve_metadata(aax_file, temp_file, ffprobe_metadata, config, log_file)
+
+    if result.conversion_error:
+        reason = f"ffmpeg_failed: {result.conversion_error}"
+        write_review_file(
+            destination=None,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=result.metadata,
+            reason=reason,
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file if temp_file.exists() else None,
+        )
+        return
+
+    if result.metadata_error:
+        write_review_file(
+            destination=None,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=None,
+            reason=f"metadata_failed: {result.metadata_error}",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file,
+        )
+        return
+
+    if result.metadata is None or result.metadata.needs_review:
+        write_review_file(
+            destination=None,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=result.metadata,
+            reason="metadata_needs_review",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file,
+        )
+        return
+
+    destination = metadata_output_path(config.resolved_output, result.metadata)
+    if destination.exists() and not config.overwrite:
+        write_agent_log(log_file, "destination_exists", destination=str(destination))
+        cleanup_temp_output(temp_file)
+        return
+
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        temp_file.replace(destination)
+    except Exception as error:  # noqa: BLE001
+        write_review_file(
+            destination=destination,
+            ffprobe_metadata=ffprobe_metadata,
+            log_file=log_file,
+            metadata=result.metadata,
+            reason=f"rename_failed: {error}",
+            review_file=review_file,
+            source=aax_file,
+            temp_file=temp_file if temp_file.exists() else None,
+        )
+    else:
+        cleanup_temp_output(temp_file)
+        write_agent_log(log_file, "conversion_complete", destination=str(destination))
+
+
+if __name__ == "__main__":
+    typer.run(main)

python/tools/audiobook/catalog.py

@@ -0,0 +1,176 @@
+"""Import audiobook catalog authors and series from CSV files."""
+
+from __future__ import annotations
+
+import csv
+import logging
+from pathlib import Path  # noqa: TC003 This is required for the typer CLI
+from typing import Annotated
+
+import typer
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from python.common import configure_logger
+from python.orm.common import get_postgres_engine
+from python.orm.richie import AudiobookAuthor, AudiobookSeries
+
+logger = logging.getLogger(__name__)
+
+AUTHOR_NAME_COLUMN = "author_name"
+ID_COLUMN = "id"
+NAME_COLUMN = "name"
+
+
+class CatalogImportError(ValueError):
+    """CSV catalog import failed validation."""
+
+
+def main(
+    authors_csv: Annotated[Path, typer.Argument(help="CSV with name and optional id.")],
+    series_csv: Annotated[Path, typer.Argument(help="CSV with name, author_name, and optional id.")],
+) -> None:
+    """Upsert audiobook authors and series from CSV files."""
+    configure_logger()
+    try:
+        engine = get_postgres_engine(name="RICHIE")
+        with Session(engine) as session:
+            author_count = upsert_authors_from_csv(session, authors_csv)
+            series_count = upsert_series_from_csv(session, series_csv)
+            session.commit()
+    except CatalogImportError as error:
+        typer.echo(str(error), err=True)
+        raise typer.Exit(code=1) from error
+
+    logger.info("Upserted %s authors and %s series", author_count, series_count)
+
+
+def upsert_authors_from_csv(session: Session, authors_csv: Path) -> int:
+    """Upsert authors from a CSV file."""
+    count = 0
+    for row_number, row in csv_rows(authors_csv):
+        name = required_csv_value(row, authors_csv, row_number, NAME_COLUMN)
+        upsert_author(session, name, csv_id(row, authors_csv, row_number))
+        count += 1
+    return count
+
+
+def upsert_series_from_csv(session: Session, series_csv: Path) -> int:
+    """Upsert series from a CSV file."""
+    count = 0
+    for row_number, row in csv_rows(series_csv):
+        series_name = required_csv_value(row, series_csv, row_number, NAME_COLUMN)
+        author_name = required_csv_value(row, series_csv, row_number, AUTHOR_NAME_COLUMN)
+        author = find_author_by_name(session, author_name)
+        if author is None:
+            msg = f"{series_csv}:{row_number}: author not found: {author_name}"
+            raise CatalogImportError(msg)
+        upsert_series(session, series_name, author, csv_id(row, series_csv, row_number))
+        count += 1
+    return count
+
+
+def upsert_author(session: Session, name: str, author_id: int | None) -> AudiobookAuthor:
+    """Upsert one author by id or exact name."""
+    if author_id is not None:
+        author = session.get(AudiobookAuthor, author_id)
+        if author is None:
+            author = AudiobookAuthor(id=author_id, name=name)
+            session.add(author)
+        else:
+            author.name = name
+        session.flush()
+        return author
+
+    author = find_author_by_name(session, name)
+    if author is None:
+        author = AudiobookAuthor(name=name)
+        session.add(author)
+        session.flush()
+    return author
+
+
+def upsert_series(
+    session: Session,
+    name: str,
+    author: AudiobookAuthor,
+    series_id: int | None,
+) -> AudiobookSeries:
+    """Upsert one series by id or exact author/name match."""
+    if series_id is not None:
+        series = session.get(AudiobookSeries, series_id)
+        if series is None:
+            series = AudiobookSeries(id=series_id, name=name, author=author)
+            session.add(series)
+        else:
+            series.name = name
+            series.author = author
+        session.flush()
+        return series
+
+    series = find_series_by_name_and_author(session, name, author.id)
+    if series is None:
+        series = AudiobookSeries(name=name, author=author)
+        session.add(series)
+        session.flush()
+    return series
+
+
+def find_author_by_name(session: Session, name: str) -> AudiobookAuthor | None:
+    """Find one author by exact name."""
+    return session.scalar(select(AudiobookAuthor).where(AudiobookAuthor.name == name))
+
+
+def find_series_by_name_and_author(
+    session: Session,
+    name: str,
+    author_id: int,
+) -> AudiobookSeries | None:
+    """Find one series by exact name and author."""
+    return session.scalar(
+        select(AudiobookSeries).where(
+            AudiobookSeries.name == name,
+            AudiobookSeries.author_id == author_id,
+        ),
+    )
+
+
+def csv_rows(csv_path: Path) -> list[tuple[int, dict[str, str | None]]]:
+    """Read a CSV file as numbered rows."""
+    with csv_path.open(newline="", encoding="utf-8") as file:
+        reader = csv.DictReader(file)
+        if reader.fieldnames is None:
+            msg = f"{csv_path}: missing CSV header"
+            raise CatalogImportError(msg)
+        return [(row_number, row) for row_number, row in enumerate(reader, start=2)]
+
+
+def required_csv_value(
+    row: dict[str, str | None],
+    csv_path: Path,
+    row_number: int,
+    column: str,
+) -> str:
+    """Read a required CSV value."""
+    value = row.get(column)
+    if value and value.strip():
+        return value.strip()
+    msg = f"{csv_path}:{row_number}: missing required column value: {column}"
+    raise CatalogImportError(msg)
+
+
+def csv_id(row: dict[str, str | None], csv_path: Path, row_number: int) -> int | None:
+    """Read an optional id field from a CSV row."""
+    value = row.get(ID_COLUMN)
+    if value is None or not value.strip():
+        return None
+    try:
+        return int(value)
+    except ValueError as error:
+        msg = f"{csv_path}:{row_number}: id must be an integer: {value}"
+        raise CatalogImportError(msg) from error
+    return None
+
+
+if __name__ == "__main__":
+    typer.run(main)

python/tools/audiobook/llm_tool_calling.py

@@ -0,0 +1,599 @@
+"""LLM tool calling support for audiobook metadata resolution."""
+
+from __future__ import annotations
+
+import json
+import re
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from sqlalchemy import or_, select
+
+from python.orm.richie import Audiobook, AudiobookAuthor, AudiobookSeries
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from sqlalchemy.orm import Session
+
+    from python.tools.audiobook.metadata_agent import AgentConfig
+
+CATALOG_SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:_[a-z0-9]+)*$")
+TITLE_SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
+
+LogWriter = Callable[..., None]
+
+
+class MetadataResolutionError(ValueError):
+    """Metadata resolution failed validation."""
+
+
+@dataclass(frozen=True)
+class EnsuredBook:
+    """Book row plus whether it was created."""
+
+    book: Audiobook
+    action: str
+
+
+class CatalogToolRegistry:
+    """Controlled catalog tools exposed to the metadata model."""
+
+    def __init__(
+        self,
+        session: Session,
+        log_path: Path,
+        config: AgentConfig,
+        write_log: LogWriter,
+    ) -> None:
+        """Create a registry bound to one database session and audit log."""
+        self.session = session
+        self.log_path = log_path
+        self.config = config
+        self.write_log = write_log
+        self.seen_author_ids: set[int] = set()
+        self.seen_series_ids: set[int] = set()
+        self.seen_book_ids: set[int] = set()
+        self.created_author_ids: set[int] = set()
+        self.created_series_ids: set[int] = set()
+        self.created_book_ids: set[int] = set()
+
+    def tool_schemas(self) -> list[dict[str, object]]:
+        """Return Ollama tool schemas."""
+        schemas = [
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_authors",
+                    "description": "Search canonical audiobook authors by slug or noisy source text.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"query": {"type": "string"}},
+                        "required": ["query"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_series",
+                    "description": "Search canonical audiobook series by slug or noisy source text.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                            "author_id": {"type": ["integer", "null"]},
+                        },
+                        "required": ["query"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "search_books",
+                    "description": "Search canonical audiobook titles with optional author and series filters.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                            "author_id": {"type": ["integer", "null"]},
+                            "series_id": {"type": ["integer", "null"]},
+                        },
+                        "required": ["query"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "ensure_author",
+                    "description": "Normalize an author name to a catalog slug, then return or create that author.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {"name": {"type": "string"}},
+                        "required": ["name"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "ensure_series",
+                    "description": "Normalize a series name to a catalog slug, then return or create it for an author.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "author_id": {"type": "integer"},
+                        },
+                        "required": ["name", "author_id"],
+                    },
+                },
+            },
+            {
+                "type": "function",
+                "function": {
+                    "name": "ensure_book",
+                    "description": "Normalize a title to a book slug, then return or create it for an author/series.",
+                    "parameters": {
+                        "type": "object",
+                        "properties": {
+                            "title": {"type": "string"},
+                            "author_id": {"type": "integer"},
+                            "series_id": {"type": ["integer", "null"]},
+                            "series_index": {"type": "number", "multipleOf": 0.5},
+                        },
+                        "required": ["title", "author_id", "series_id", "series_index"],
+                    },
+                },
+            },
+        ]
+        enabled_tool_names = set(self.config.tool_names)
+        return [schema for schema in schemas if schema["function"]["name"] in enabled_tool_names]
+
+    def run(self, name: str, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Run one catalog tool and audit the call."""
+        handlers = {
+            "search_authors": self.run_search_authors,
+            "search_series": self.run_search_series,
+            "search_books": self.run_search_books,
+            "ensure_author": self.run_ensure_author,
+            "ensure_series": self.run_ensure_series,
+            "ensure_book": self.run_ensure_book,
+        }
+        handler = handlers.get(name)
+        if handler is None:
+            self.write_log(self.log_path, "tool_error", tool=name, arguments=arguments, error="unknown_tool")
+            msg = f"Unknown audiobook metadata tool: {name}"
+            raise MetadataResolutionError(msg)
+        if name not in self.config.tool_names:
+            self.write_log(self.log_path, "tool_error", tool=name, arguments=arguments, error="tool_not_enabled")
+            msg = f"Audiobook metadata tool is not enabled: {name}"
+            raise MetadataResolutionError(msg)
+
+        started = time.perf_counter()
+        self.write_log(self.log_path, "tool_call", tool=name, arguments=arguments)
+        result = handler(arguments)
+        duration_ms = round((time.perf_counter() - started) * 1000, 3)
+        self.write_log(
+            self.log_path,
+            "tool_result",
+            tool=name,
+            duration_ms=duration_ms,
+            result_count=len(result),
+            preview=result[:3],
+        )
+        return result
+
+    def get_author(self, author_id: int) -> AudiobookAuthor | None:
+        """Return an author by id."""
+        return self.session.get(AudiobookAuthor, author_id)
+
+    def get_book(self, book_id: int) -> Audiobook | None:
+        """Return a book by id."""
+        return self.session.get(Audiobook, book_id)
+
+    def get_series(self, series_id: int) -> AudiobookSeries | None:
+        """Return a series by id."""
+        return self.session.get(AudiobookSeries, series_id)
+
+    def prune_unused_created_rows(self, *, author_id: int, book_id: int | None, series_id: int | None) -> None:
+        """Remove catalog rows created during this run but not used by final metadata."""
+        used_book_ids = {book_id} if book_id is not None else set()
+        for created_book_id in self.created_book_ids - used_book_ids:
+            if book := self.get_book(created_book_id):
+                self.session.delete(book)
+
+        self.session.flush()
+        used_series_ids = {series_id} if series_id is not None else set()
+        for created_series_id in self.created_series_ids - used_series_ids:
+            series = self.get_series(created_series_id)
+            if series and not series.books:
+                self.session.delete(series)
+
+        self.session.flush()
+        for created_author_id in self.created_author_ids - {author_id}:
+            author = self.get_author(created_author_id)
+            if author and not author.books and not author.series:
+                self.session.delete(author)
+
+    def run_search_authors(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Search authors from tool arguments and remember returned ids."""
+        query = required_string(arguments, "query")
+        statement = select(AudiobookAuthor).order_by(AudiobookAuthor.name).limit(self.config.max_tool_results)
+        if terms := query_terms(query):
+            statement = statement.where(or_(*(AudiobookAuthor.name.ilike(f"%{term}%") for term in terms)))
+
+        authors = self.session.scalars(statement).all()
+        self.seen_author_ids.update(author.id for author in authors)
+        return [{"id": author.id, "name": author.name} for author in authors]
+
+    def run_search_series(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Search series from tool arguments and remember returned ids."""
+        query = required_string(arguments, "query")
+        author_id = optional_int(arguments.get("author_id"), "author_id")
+        statement = select(AudiobookSeries).order_by(AudiobookSeries.name).limit(self.config.max_tool_results)
+        if terms := query_terms(query):
+            statement = statement.where(or_(*(AudiobookSeries.name.ilike(f"%{term}%") for term in terms)))
+        if author_id is not None:
+            statement = statement.where(AudiobookSeries.author_id == author_id)
+
+        series_rows = self.session.scalars(statement).all()
+        self.seen_series_ids.update(series.id for series in series_rows)
+        self.seen_author_ids.update(series.author_id for series in series_rows)
+        return [
+            {
+                "id": series.id,
+                "name": series.name,
+                "author_id": series.author_id,
+                "author": series.author.name,
+            }
+            for series in series_rows
+        ]
+
+    def run_search_books(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Search books from tool arguments and remember returned ids."""
+        query = required_string(arguments, "query")
+        author_id = optional_int(arguments.get("author_id"), "author_id")
+        series_id = optional_int(arguments.get("series_id"), "series_id")
+        statement = select(Audiobook).order_by(Audiobook.title).limit(self.config.max_tool_results)
+        if terms := query_terms(query):
+            statement = statement.where(or_(*(Audiobook.title.ilike(f"%{term}%") for term in terms)))
+        if author_id is not None:
+            statement = statement.where(Audiobook.author_id == author_id)
+        if series_id is not None:
+            statement = statement.where(Audiobook.series_id == series_id)
+
+        books = self.session.scalars(statement).all()
+        self.seen_book_ids.update(book.id for book in books)
+        self.seen_author_ids.update(book.author_id for book in books)
+        self.seen_series_ids.update(book.series_id for book in books if book.series_id is not None)
+        return [
+            {
+                "id": book.id,
+                "title": book.title,
+                "author_id": book.author_id,
+                "author": book.author.name,
+                "series_id": book.series_id,
+                "series": book.series.name if book.series else self.config.standalone_series,
+                "series_index": book.series_index,
+            }
+            for book in books
+        ]
+
+    def run_ensure_author(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Ensure an author from tool arguments and return a tool result."""
+        name = normalize_catalog_slug(required_string(arguments, "name"))
+        validate_catalog_slug(name, "author")
+        author = self.session.scalar(select(AudiobookAuthor).where(AudiobookAuthor.name == name))
+        action = "existing"
+        if author is None:
+            author = AudiobookAuthor(name=name)
+            self.session.add(author)
+            self.session.flush()
+            self.created_author_ids.add(author.id)
+            action = "created"
+
+        self.seen_author_ids.add(author.id)
+        return [{"id": author.id, "name": author.name, "action": action}]
+
+    def run_ensure_series(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Ensure a series from tool arguments and return a tool result."""
+        name = normalize_catalog_slug(required_string(arguments, "name"))
+        author_id = required_int(arguments, "author_id")
+        validate_catalog_slug(name, "series")
+        author = self.required_author(author_id)
+        series = self.find_series_by_catalog_slug(name, author.id)
+        action = "existing"
+        if series is None:
+            series = AudiobookSeries(name=name, author=author)
+            self.session.add(series)
+            self.session.flush()
+            self.created_series_ids.add(series.id)
+            action = "created"
+
+        self.seen_author_ids.add(author.id)
+        self.seen_series_ids.add(series.id)
+        return [self.series_result(series, action)]
+
+    def run_ensure_book(self, arguments: dict[str, object]) -> list[dict[str, object]]:
+        """Ensure a book from tool arguments and return a tool result."""
+        title = required_string(arguments, "title")
+        author_id = required_int(arguments, "author_id")
+        series_id = optional_int(arguments.get("series_id"), "series_id")
+        series_index = required_series_index(arguments, "series_index")
+        ensured = self.ensure_book(title, author_id, series_id, series_index)
+        return [self.book_result(ensured.book, ensured.action)]
+
+    def ensure_book(
+        self,
+        title: str,
+        author_id: int,
+        series_id: int | None,
+        series_index: float,
+    ) -> EnsuredBook:
+        """Return an existing book row, or create it after validating ownership."""
+        title = normalize_title_slug(title)
+        validate_title_slug(title)
+        author = self.required_author(author_id)
+        series = None
+        if series_id is None:
+            if series_index != 0:
+                msg = "standalone books must use series_index 0"
+                raise MetadataResolutionError(msg)
+        else:
+            series = self.required_series(series_id)
+            if series.author_id != author.id:
+                msg = f"series_id {series_id} does not belong to author_id {author_id}"
+                raise MetadataResolutionError(msg)
+            if series_index <= 0:
+                msg = "series books must use a positive series_index"
+                raise MetadataResolutionError(msg)
+
+        statement = select(Audiobook).where(
+            Audiobook.title == title,
+            Audiobook.author_id == author.id,
+        )
+        if series is None:
+            statement = statement.where(Audiobook.series_id.is_(None))
+        else:
+            statement = statement.where(Audiobook.series_id == series.id)
+        book = self.session.scalar(statement)
+        if book is None:
+            book = Audiobook(title=title, author=author, series=series, series_index=series_index)
+            self.session.add(book)
+            self.session.flush()
+            self.created_book_ids.add(book.id)
+            action = "created"
+        else:
+            action = "existing"
+
+        self.seen_book_ids.add(book.id)
+        self.seen_author_ids.add(author.id)
+        if book.series_id is not None:
+            self.seen_series_ids.add(book.series_id)
+        return EnsuredBook(book=book, action=action)
+
+    def required_author(self, author_id: int) -> AudiobookAuthor:
+        """Return an author or fail metadata resolution."""
+        author = self.get_author(author_id)
+        if author is None:
+            msg = f"author_id {author_id} does not exist"
+            raise MetadataResolutionError(msg)
+        return author
+
+    def required_series(self, series_id: int) -> AudiobookSeries:
+        """Return a series or fail metadata resolution."""
+        series = self.get_series(series_id)
+        if series is None:
+            msg = f"series_id {series_id} does not exist"
+            raise MetadataResolutionError(msg)
+        return series
+
+    def find_series_by_catalog_slug(self, name: str, author_id: int) -> AudiobookSeries | None:
+        """Return a series by exact slug or underscore-insensitive slug."""
+        exact = self.session.scalar(
+            select(AudiobookSeries).where(
+                AudiobookSeries.name == name,
+                AudiobookSeries.author_id == author_id,
+            ),
+        )
+        if exact is not None:
+            return exact
+
+        compact_name = compact_catalog_slug(name)
+        series_rows = self.session.scalars(
+            select(AudiobookSeries).where(AudiobookSeries.author_id == author_id).order_by(AudiobookSeries.name),
+        ).all()
+        for series in series_rows:
+            if compact_catalog_slug(series.name) == compact_name:
+                return series
+        return None
+
+    def series_result(self, series: AudiobookSeries, action: str) -> dict[str, object]:
+        """Build a normalized series tool result."""
+        return {
+            "id": series.id,
+            "name": series.name,
+            "author_id": series.author_id,
+            "author": series.author.name,
+            "action": action,
+        }
+
+    def book_result(self, book: Audiobook, action: str) -> dict[str, object]:
+        """Build a normalized book tool result."""
+        return {
+            "id": book.id,
+            "title": book.title,
+            "author_id": book.author_id,
+            "author": book.author.name,
+            "series_id": book.series_id,
+            "series": book.series.name if book.series else self.config.standalone_series,
+            "series_index": book.series_index,
+            "action": action,
+        }
+
+
+def run_tool_calls(
+    messages: list[dict[str, object]],
+    message: dict[str, object],
+    tool_calls: list[tuple[str, dict[str, object]]],
+    registry: CatalogToolRegistry,
+    log_path: Path,
+    write_log: LogWriter,
+) -> str | None:
+    """Run tool calls, append tool messages, and return fatal error text when stopped."""
+    messages.append(message)
+    for tool_name, arguments in tool_calls:
+        try:
+            tool_result = registry.run(tool_name, arguments)
+        except MetadataResolutionError as error:
+            if is_fatal_tool_error(error):
+                return str(error)
+            write_log(log_path, "tool_error", tool=tool_name, arguments=arguments, error=str(error))
+            messages.append(
+                {
+                    "role": "tool",
+                    "tool_name": tool_name,
+                    "content": json.dumps({"error": str(error)}, sort_keys=True),
+                },
+            )
+            continue
+        messages.append(
+            {
+                "role": "tool",
+                "tool_name": tool_name,
+                "content": json.dumps(tool_result, sort_keys=True),
+            },
+        )
+    return None
+
+
+def parse_tool_calls(message: dict[str, object]) -> list[tuple[str, dict[str, object]]]:
+    """Parse Ollama tool calls from a response message."""
+    raw_tool_calls = message.get("tool_calls") or []
+    if not isinstance(raw_tool_calls, list):
+        msg = "tool_calls must be a list"
+        raise MetadataResolutionError(msg)
+
+    tool_calls = []
+    for raw_call in raw_tool_calls:
+        if not isinstance(raw_call, dict):
+            msg = "tool call must be an object"
+            raise MetadataResolutionError(msg)
+        function = raw_call.get("function")
+        if not isinstance(function, dict):
+            msg = "tool call is missing function"
+            raise MetadataResolutionError(msg)
+        name = function.get("name")
+        if not isinstance(name, str) or not name:
+            msg = "tool call is missing function name"
+            raise MetadataResolutionError(msg)
+        arguments = parse_tool_arguments(function.get("arguments", {}))
+        tool_calls.append((name, arguments))
+    return tool_calls
+
+
+def parse_tool_arguments(raw_arguments: object) -> dict[str, object]:
+    """Parse tool call arguments returned by Ollama."""
+    if isinstance(raw_arguments, dict):
+        return {str(key): value for key, value in raw_arguments.items()}
+    if isinstance(raw_arguments, str):
+        parsed = json.loads(raw_arguments) if raw_arguments else {}
+        if isinstance(parsed, dict):
+            return {str(key): value for key, value in parsed.items()}
+    msg = "tool arguments must be an object"
+    raise MetadataResolutionError(msg)
+
+
+def validate_title_slug(title: str) -> None:
+    """Validate a canonical book title slug."""
+    if not TITLE_SLUG_PATTERN.fullmatch(title):
+        msg = f"title slug is invalid: {title}"
+        raise MetadataResolutionError(msg)
+
+
+def validate_catalog_slug(value: str, label: str) -> None:
+    """Validate a canonical catalog slug."""
+    if not CATALOG_SLUG_PATTERN.fullmatch(value):
+        msg = f"{label} slug is invalid: {value}"
+        raise MetadataResolutionError(msg)
+
+
+def normalize_catalog_slug(value: str) -> str:
+    """Normalize noisy catalog names into lower snake-case slugs."""
+    return re.sub(r"[^a-z0-9]+", "_", value.strip().casefold()).strip("_")
+
+
+def compact_catalog_slug(value: str) -> str:
+    """Return a catalog slug comparison key that ignores underscores."""
+    return normalize_catalog_slug(value).replace("_", "")
+
+
+def normalize_title_slug(value: str) -> str:
+    """Normalize noisy book titles into lower kebab-case slugs."""
+    return re.sub(r"[^a-z0-9]+", "-", value.strip().casefold()).strip("-")
+
+
+def is_fatal_tool_error(error: MetadataResolutionError) -> bool:
+    """Return whether a tool error should stop the agent immediately."""
+    message = str(error)
+    return message.startswith(
+        (
+            "Unknown audiobook metadata tool",
+            "Audiobook metadata tool is not enabled",
+        ),
+    )
+
+
+def query_terms(query: str) -> tuple[str, ...]:
+    """Return text variants useful for matching noisy audiobook metadata."""
+    normalized = query.strip().casefold()
+    underscore_slug = normalize_catalog_slug(normalized)
+    compact_slug = compact_catalog_slug(normalized)
+    hyphen_slug = normalize_title_slug(normalized)
+    return tuple(dict.fromkeys(term for term in (normalized, underscore_slug, compact_slug, hyphen_slug) if term))
+
+
+def required_string(data: dict[str, object], key: str) -> str:
+    """Read a required string field."""
+    value = data.get(key)
+    if not isinstance(value, str) or not value.strip():
+        msg = f"{key} must be a non-empty string"
+        raise MetadataResolutionError(msg)
+    return value.strip()
+
+
+def required_int(data: dict[str, object], key: str) -> int:
+    """Read a required integer field."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int):
+        msg = f"{key} must be an integer"
+        raise MetadataResolutionError(msg)
+    return value
+
+
+def required_series_index(data: dict[str, object], key: str) -> float:
+    """Read a required whole-number or half-number series index."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        msg = f"{key} must be a number"
+        raise MetadataResolutionError(msg)
+    series_index = float(value)
+    if not (series_index * 2).is_integer():
+        msg = f"{key} must be a whole number or .5 increment"
+        raise MetadataResolutionError(msg)
+    return series_index
+
+
+def optional_int(value: object, key: str) -> int | None:
+    """Read an optional integer field."""
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int):
+        msg = f"{key} must be an integer or null"
+        raise MetadataResolutionError(msg)
+    return value

python/tools/audiobook/metadata_agent.py

@@ -0,0 +1,575 @@
+"""Resolve audiobook metadata with a controlled Ollama tool loop."""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import asdict, dataclass, is_dataclass, replace
+from os import PathLike
+from typing import TYPE_CHECKING
+
+import httpx
+from sqlalchemy.orm import Session
+
+from python.common import utcnow
+from python.tools.audiobook.llm_tool_calling import (
+    CatalogToolRegistry,
+    MetadataResolutionError,
+    normalize_title_slug,
+    optional_int,
+    parse_tool_calls,
+    required_int,
+    required_series_index,
+    required_string,
+    run_tool_calls,
+    validate_catalog_slug,
+    validate_title_slug,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from sqlalchemy.engine import Engine
+
+    from python.orm.richie import AudiobookAuthor
+
+FENCED_JSON_PATTERN = re.compile(r"^```(?:json)?\s*(?P<json>.*?)\s*```$", re.IGNORECASE | re.DOTALL)
+
+
+@dataclass(frozen=True)
+class AgentConfig:
+    """Runtime settings for the audiobook metadata agent."""
+
+    model: str = "deepseek-v4-flash:cloud"
+    ollama_chat_url: str = "https://ollama.com/api/chat"
+    http_timeout_seconds: int = 300
+    max_agent_turns: int = 8
+    max_tool_results: int = 10
+    min_confidence: float = 0.85
+    invalid_final_retries: int = 1
+    standalone_series: str = "standalone"
+    tool_names: tuple[str, ...] = (
+        "search_authors",
+        "search_series",
+        "search_books",
+        "ensure_author",
+        "ensure_series",
+        "ensure_book",
+    )
+
+
+@dataclass(frozen=True)
+class StandardBookMetadata:
+    """Canonical metadata for the final audiobook path."""
+
+    author_id: int
+    author: str
+    book_id: int | None
+    title: str
+    series_id: int | None
+    series: str
+    series_index: float
+    confidence: float
+    needs_review: bool
+    evidence: list[str]
+
+
+@dataclass(frozen=True)
+class FinalMetadataFields:
+    """Raw model fields after schema validation."""
+
+    author_id: int
+    book_id: int | None
+    title: str
+    series_id: int | None
+    series_index: float
+    confidence: float
+    evidence: list[str]
+
+
+@dataclass(frozen=True)
+class ResolvedBookFields:
+    """Book fields after optional catalog book resolution."""
+
+    book_id: int | None
+    title: str
+    series_id: int | None
+    series_index: float
+
+
+@dataclass(frozen=True)
+class AgentStepResult:
+    """Outcome from one model response."""
+
+    metadata: StandardBookMetadata | None
+    invalid_final_count: int
+    should_continue: bool
+
+
+def standard_book_metadata(
+    aax_file_name: str,
+    aax_metadata_from_ffprobe: dict[str, str],
+    engine: Engine,
+    log_path: Path,
+    ollama_api_key: str,
+    config: AgentConfig,
+) -> StandardBookMetadata:
+    """Resolve canonical audiobook metadata with the configured Ollama Cloud model."""
+    with Session(engine) as session:
+        registry = CatalogToolRegistry(session, log_path, config, write_agent_log)
+        agent = AudiobookMetadataAgent(
+            registry=registry, log_path=log_path, ollama_api_key=ollama_api_key, config=config
+        )
+        metadata = agent.run(aax_file_name, aax_metadata_from_ffprobe)
+        if metadata.needs_review:
+            session.rollback()
+        else:
+            registry.prune_unused_created_rows(
+                author_id=metadata.author_id,
+                book_id=metadata.book_id,
+                series_id=metadata.series_id,
+            )
+            session.commit()
+        return metadata
+
+
+class AudiobookMetadataAgent:
+    """Ollama-backed metadata resolver with a fixed local tool registry."""
+
+    def __init__(
+        self,
+        *,
+        registry: CatalogToolRegistry,
+        log_path: Path,
+        ollama_api_key: str,
+        config: AgentConfig,
+    ) -> None:
+        """Create an Ollama metadata agent."""
+        self._registry = registry
+        self._log_path = log_path
+        self._ollama_api_key = ollama_api_key
+        self._config = config
+
+    def run(self, aax_file_name: str, aax_metadata_from_ffprobe: dict[str, str]) -> StandardBookMetadata:
+        """Resolve metadata for one AAX file."""
+        messages = [
+            {"role": "system", "content": system_prompt()},
+            {"role": "user", "content": user_prompt(aax_file_name, aax_metadata_from_ffprobe)},
+        ]
+        invalid_final_count = 0
+        result: StandardBookMetadata | None = None
+
+        for turn in range(1, self._config.max_agent_turns + 1):
+            step = self.run_step(messages, turn, invalid_final_count)
+            invalid_final_count = step.invalid_final_count
+            if step.should_continue:
+                continue
+            result = step.metadata
+            break
+
+        if result is None:
+            return self.force_final_response(messages)
+        return result
+
+    def run_step(
+        self,
+        messages: list[dict[str, object]],
+        turn: int,
+        invalid_final_count: int,
+    ) -> AgentStepResult:
+        """Run one model turn and return the next agent-loop action."""
+        data = self.chat(messages, turn)
+        message = data.get("message")
+        if not isinstance(message, dict):
+            return AgentStepResult(
+                metadata=review_metadata("Ollama response did not include a message", self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+
+        try:
+            tool_calls = parse_tool_calls(message)
+        except (json.JSONDecodeError, MetadataResolutionError) as error:
+            return AgentStepResult(
+                metadata=review_metadata(str(error), self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+        if tool_calls:
+            fatal_error = run_tool_calls(messages, message, tool_calls, self._registry, self._log_path, write_agent_log)
+            if fatal_error is not None:
+                return AgentStepResult(
+                    metadata=review_metadata(fatal_error, self._config),
+                    invalid_final_count=invalid_final_count,
+                    should_continue=False,
+                )
+            return AgentStepResult(metadata=None, invalid_final_count=invalid_final_count, should_continue=True)
+        return self.handle_final_message(messages, message, invalid_final_count)
+
+    def handle_final_message(
+        self,
+        messages: list[dict[str, object]],
+        message: dict[str, object],
+        invalid_final_count: int,
+    ) -> AgentStepResult:
+        """Validate a final model message or request one retry."""
+        content = message.get("content")
+        if not isinstance(content, str):
+            return AgentStepResult(
+                metadata=review_metadata("Ollama final response did not include string content", self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+
+        try:
+            resolved = self.validate_final(parse_final_json_content(content))
+        except (json.JSONDecodeError, MetadataResolutionError) as error:
+            return self.handle_invalid_final(messages, error, invalid_final_count)
+
+        write_agent_log(self._log_path, "final_metadata", metadata=resolved)
+        return AgentStepResult(metadata=resolved, invalid_final_count=invalid_final_count, should_continue=False)
+
+    def handle_invalid_final(
+        self,
+        messages: list[dict[str, object]],
+        error: json.JSONDecodeError | MetadataResolutionError,
+        invalid_final_count: int,
+    ) -> AgentStepResult:
+        """Log invalid final JSON and either retry or return review metadata."""
+        invalid_final_count += 1
+        write_agent_log(
+            self._log_path,
+            "final_validation_error",
+            error=str(error),
+            invalid_final_count=invalid_final_count,
+        )
+        if invalid_final_count > self._config.invalid_final_retries:
+            return AgentStepResult(
+                metadata=review_metadata(str(error), self._config),
+                invalid_final_count=invalid_final_count,
+                should_continue=False,
+            )
+        messages.append(
+            {
+                "role": "user",
+                "content": (
+                    "Your previous final answer was invalid. Return only valid JSON matching the required "
+                    f"schema. Validation error: {error}"
+                ),
+            },
+        )
+        return AgentStepResult(metadata=None, invalid_final_count=invalid_final_count, should_continue=True)
+
+    def force_final_response(self, messages: list[dict[str, object]]) -> StandardBookMetadata:
+        """Request a no-tool final answer after the normal turn limit."""
+        messages.append({"role": "user", "content": forced_final_prompt()})
+        write_agent_log(self._log_path, "forced_final_request", reason="max_turns")
+        data = self.chat(messages, self._config.max_agent_turns + 1, tools_enabled=False)
+        message = data.get("message")
+        if not isinstance(message, dict):
+            return review_metadata("Ollama forced final response did not include a message", self._config)
+        content = message.get("content")
+        if not isinstance(content, str):
+            return review_metadata("Ollama forced final response did not include string content", self._config)
+        try:
+            resolved = self.validate_final(parse_final_json_content(content))
+        except (json.JSONDecodeError, MetadataResolutionError) as error:
+            return review_metadata(f"Ollama forced final response was invalid: {error}", self._config)
+        write_agent_log(self._log_path, "final_metadata", metadata=resolved)
+        return resolved
+
+    def chat(self, messages: list[dict[str, object]], turn: int, *, tools_enabled: bool = True) -> dict[str, object]:
+        """Send one chat request to Ollama and log the request and response."""
+        payload = {
+            "model": self._config.model,
+            "messages": messages,
+            "stream": False,
+            "options": {"temperature": 0.1},
+        }
+        tool_names = []
+        if tools_enabled:
+            payload["tools"] = self._registry.tool_schemas()
+            tool_names = self._config.tool_names
+        write_agent_log(
+            self._log_path,
+            "model_request",
+            model=self._config.model,
+            turn=turn,
+            message_count=len(messages),
+            tool_names=tool_names,
+            tools_enabled=tools_enabled,
+        )
+        write_agent_log(
+            self._log_path,
+            "llm_messages_sent",
+            model=self._config.model,
+            turn=turn,
+            messages=messages,
+            tools_enabled=tools_enabled,
+        )
+        response = httpx.post(
+            self._config.ollama_chat_url,
+            headers={"Authorization": f"Bearer {self._ollama_api_key}"},
+            json=payload,
+            timeout=self._config.http_timeout_seconds,
+        )
+        response.raise_for_status()
+        raw_data = response.json()
+        if not isinstance(raw_data, dict):
+            return {}
+        data = {str(key): value for key, value in raw_data.items()}
+        message = data.get("message", {})
+        content = message.get("content") if isinstance(message, dict) else ""
+        write_agent_log(
+            self._log_path,
+            "llm_message_received",
+            model=self._config.model,
+            turn=turn,
+            message=message,
+        )
+        write_agent_log(
+            self._log_path,
+            "model_response",
+            model=self._config.model,
+            turn=turn,
+            has_tool_calls=bool(isinstance(message, dict) and message.get("tool_calls")),
+            content_chars=len(content) if isinstance(content, str) else 0,
+        )
+        return data
+
+    def validate_final(self, raw_metadata: object) -> StandardBookMetadata:
+        """Validate final model metadata against catalog rows."""
+        fields = parse_final_metadata_fields(raw_metadata)
+        fields = replace(fields, title=normalize_title_slug(fields.title))
+        author = self.validate_author(fields.author_id)
+        validate_title_slug(fields.title)
+        book_fields = self.resolve_book_fields(fields)
+        series = self.validate_series(fields.author_id, book_fields.series_id, book_fields.series_index)
+
+        return StandardBookMetadata(
+            author_id=fields.author_id,
+            author=author.name,
+            book_id=book_fields.book_id,
+            title=book_fields.title,
+            series_id=book_fields.series_id,
+            series=series,
+            series_index=book_fields.series_index,
+            confidence=fields.confidence,
+            needs_review=fields.confidence < self._config.min_confidence,
+            evidence=fields.evidence,
+        )
+
+    def validate_author(self, author_id: int) -> AudiobookAuthor:
+        """Validate that an author id was seen and exists."""
+        if author_id not in self._registry.seen_author_ids:
+            msg = f"author_id {author_id} was not returned by search_authors"
+            raise MetadataResolutionError(msg)
+        author = self._registry.get_author(author_id)
+        if author is None:
+            msg = f"author_id {author_id} does not exist"
+            raise MetadataResolutionError(msg)
+        validate_catalog_slug(author.name, "author")
+        return author
+
+    def resolve_book_fields(self, fields: FinalMetadataFields) -> ResolvedBookFields:
+        """Resolve final book fields from a seen book id or created book."""
+        if fields.book_id is None:
+            ensured = self._registry.ensure_book(
+                fields.title,
+                fields.author_id,
+                fields.series_id,
+                fields.series_index,
+            )
+            return ResolvedBookFields(
+                book_id=ensured.book.id,
+                title=ensured.book.title,
+                series_id=ensured.book.series_id,
+                series_index=ensured.book.series_index,
+            )
+
+        if fields.book_id not in self._registry.seen_book_ids:
+            msg = f"book_id {fields.book_id} was not returned by search_books"
+            raise MetadataResolutionError(msg)
+        book = self._registry.get_book(fields.book_id)
+        if book is None:
+            msg = f"book_id {fields.book_id} does not exist"
+            raise MetadataResolutionError(msg)
+        if book.author_id != fields.author_id:
+            msg = f"book_id {fields.book_id} does not belong to author_id {fields.author_id}"
+            raise MetadataResolutionError(msg)
+        return ResolvedBookFields(
+            book_id=fields.book_id,
+            title=book.title,
+            series_id=book.series_id,
+            series_index=book.series_index,
+        )
+
+    def validate_series(self, author_id: int, series_id: int | None, series_index: float) -> str:
+        """Validate final series fields and return the canonical series slug."""
+        if series_id is None:
+            if series_index != 0:
+                msg = "standalone books must use series_index 0"
+                raise MetadataResolutionError(msg)
+            return self._config.standalone_series
+
+        if series_id not in self._registry.seen_series_ids:
+            msg = f"series_id {series_id} was not returned by search_series"
+            raise MetadataResolutionError(msg)
+        series = self._registry.get_series(series_id)
+        if series is None:
+            msg = f"series_id {series_id} does not exist"
+            raise MetadataResolutionError(msg)
+        if series.author_id != author_id:
+            msg = f"series_id {series_id} does not belong to author_id {author_id}"
+            raise MetadataResolutionError(msg)
+        if series_index <= 0:
+            msg = "series books must use a positive series_index"
+            raise MetadataResolutionError(msg)
+        validate_catalog_slug(series.name, "series")
+        return series.name
+
+
+def write_agent_log(log_path: Path, event: str, **fields: object) -> None:
+    """Append one JSONL audit event."""
+    log_path.parent.mkdir(parents=True, exist_ok=True)
+    record = {
+        "created": utcnow().isoformat(),
+        "event": event,
+        **{key: json_log_value(value) for key, value in fields.items()},
+    }
+    with log_path.open("a", encoding="utf-8") as file:
+        file.write(json.dumps(record, sort_keys=True))
+        file.write("\n")
+
+
+def json_log_value(value: object) -> object:
+    """Return a JSON-serializable value for audit logs."""
+    if is_dataclass(value) and not isinstance(value, type):
+        return json_log_value(asdict(value))
+    if isinstance(value, dict):
+        return {str(key): json_log_value(item) for key, item in value.items()}
+    if isinstance(value, list | tuple):
+        return [json_log_value(item) for item in value]
+    if isinstance(value, set):
+        return [json_log_value(item) for item in sorted(value, key=str)]
+    if isinstance(value, PathLike):
+        return str(value)
+    return value
+
+
+def system_prompt() -> str:
+    """Return the stable system prompt."""
+    return """You standardize Audible audiobook metadata against a private catalog.
+
+Rules:
+- You must use the provided tools before returning final metadata.
+- Only use author_id, series_id, or book_id values returned by tools.
+- Return final metadata as JSON only. Do not wrap it in Markdown.
+- The final JSON object must contain author_id, book_id, title, series_id, series_index, confidence, and evidence.
+- title must be a canonical title slug using lower-case words separated by hyphens.
+- Use series_id null and series_index 0 for standalone books.
+- If you use a series_id, series_index must be a whole number or .5 value greater than 0.
+- Treat series slugs that differ only by underscores as the same series. Prefer the existing catalog row instead of
+  creating a new series.
+- Detect omnibus or box-set editions that contain multiple numbered novels, books, or novellas.
+- For an omnibus, make a best-effort range from the filename, tags, and catalog rows. Keep series_index as the
+  first covered book number and include the range in the title when the source title includes it, for example
+  books-1-3.
+- Be careful with omnibuses of novels or novellas later published as one book: keep the omnibus as the audiobook's
+  book record unless catalog rows clearly identify a better match.
+- Do not create publisher collections or author collections as series unless the book metadata clearly gives a
+  numbered series.
+- Series belong to authors. Use a series_id only when it belongs to the selected author_id.
+- Always search for the author before creating one. If no exact author slug exists, call ensure_author.
+- Always search for a series with author_id before creating one. If no exact series slug exists, call ensure_series.
+- Always search for a book before creating one. If no exact title slug exists, call ensure_book.
+- If a tool returns an error, correct your tool arguments or final metadata before continuing.
+- confidence must be a number from 0 to 1.
+- evidence must be a short list of strings explaining which filename, tags, and catalog rows support the answer."""
+
+
+def forced_final_prompt() -> str:
+    """Return the no-tools finalization prompt."""
+    return (
+        "Stop calling tools. Return final metadata as JSON only using the tool results already provided. "
+        "If search_books returned no matching rows but author and series are known, use book_id null and resolve "
+        "the title slug from the AAX filename and ffprobe tags. The validator will create the missing book. "
+        "Use only author_id and series_id values returned by earlier tool results."
+    )
+
+
+def user_prompt(aax_file_name: str, metadata: dict[str, str]) -> str:
+    """Build the user prompt from source metadata."""
+    return (
+        "Resolve this Audible audiobook.\n\n"
+        f"AAX file name: {aax_file_name}\n\n"
+        "ffprobe format tags:\n"
+        f"{json.dumps(metadata, indent=2, sort_keys=True)}"
+    )
+
+
+def parse_final_json_content(content: str) -> object:
+    """Parse final model content, accepting bare or fenced JSON."""
+    stripped = content.strip()
+    if match := FENCED_JSON_PATTERN.fullmatch(stripped):
+        stripped = match.group("json").strip()
+    return json.loads(stripped)
+
+
+def parse_final_metadata_fields(raw_metadata: object) -> FinalMetadataFields:
+    """Parse the model's final JSON object into typed fields."""
+    if not isinstance(raw_metadata, dict):
+        msg = "Final metadata must be a JSON object"
+        raise MetadataResolutionError(msg)
+    data = {str(key): value for key, value in raw_metadata.items()}
+    return FinalMetadataFields(
+        author_id=required_int(data, "author_id"),
+        book_id=optional_int(data.get("book_id"), "book_id"),
+        title=required_string(data, "title"),
+        series_id=optional_int(data.get("series_id"), "series_id"),
+        series_index=required_series_index(data, "series_index"),
+        confidence=required_float(data, "confidence"),
+        evidence=required_string_list(data, "evidence"),
+    )
+
+
+def review_metadata(reason: str, config: AgentConfig) -> StandardBookMetadata:
+    """Return a metadata result that must be reviewed manually."""
+    return StandardBookMetadata(
+        author_id=0,
+        author="unknown_author",
+        book_id=None,
+        title="unknown-title",
+        series_id=None,
+        series=config.standalone_series,
+        series_index=0,
+        confidence=0,
+        needs_review=True,
+        evidence=[reason],
+    )
+
+
+def required_float(data: dict[str, object], key: str) -> float:
+    """Read a required float field."""
+    value = data.get(key)
+    if isinstance(value, bool) or not isinstance(value, int | float):
+        msg = f"{key} must be a number"
+        raise MetadataResolutionError(msg)
+    confidence = float(value)
+    if confidence < 0 or confidence > 1:
+        msg = f"{key} must be between 0 and 1"
+        raise MetadataResolutionError(msg)
+    return confidence
+
+
+def required_string_list(data: dict[str, object], key: str) -> list[str]:
+    """Read a required list of strings."""
+    value = data.get(key)
+    if not isinstance(value, list) or not value or not all(isinstance(item, str) for item in value):
+        msg = f"{key} must be a non-empty list of strings"
+        raise MetadataResolutionError(msg)
+    strings = [item.strip() for item in value if item.strip()]
+    if not strings:
+        msg = f"{key} must include at least one non-empty string"
+        raise MetadataResolutionError(msg)
+    return strings

tests/test_audiobook_catalog.py

@@ -0,0 +1,126 @@
+"""test_audiobook_catalog."""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import create_engine, select
+from sqlalchemy.orm import sessionmaker
+
+from python.orm.richie import AudiobookAuthor, AudiobookSeries, RichieBase
+from python.tools.audiobook import catalog
+
+
+@pytest.fixture
+def audiobook_session():
+    engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
+    RichieBase.metadata.create_all(engine)
+    with sessionmaker(bind=engine, expire_on_commit=False, future=True)() as session:
+        yield session
+    engine.dispose()
+
+
+def test_upsert_catalog_csv_inserts_and_updates_authors_and_series(tmp_path, audiobook_session) -> None:
+    audiobook_session.add_all(
+        [
+            AudiobookAuthor(id=10, name="old_author"),
+            AudiobookAuthor(id=11, name="craig_alanson"),
+        ],
+    )
+    audiobook_session.commit()
+    authors_csv = tmp_path / "authors.csv"
+    series_csv = tmp_path / "series.csv"
+    authors_csv.write_text(
+        "name,id\n"
+        "glynn_stewart,\n"
+        "craig_alanson,\n"
+        "updated_author,10\n",
+        encoding="utf-8",
+    )
+    series_csv.write_text(
+        "name,author_name,id\n"
+        "starships_mage,glynn_stewart,\n"
+        "expeditionary_force,craig_alanson,\n",
+        encoding="utf-8",
+    )
+
+    author_count = catalog.upsert_authors_from_csv(audiobook_session, authors_csv)
+    series_count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
+    audiobook_session.commit()
+
+    authors = audiobook_session.scalars(select(AudiobookAuthor).order_by(AudiobookAuthor.id)).all()
+    series = audiobook_session.scalars(select(AudiobookSeries).order_by(AudiobookSeries.name)).all()
+    assert author_count == 3
+    assert series_count == 2
+    assert [(author.id, author.name) for author in authors] == [
+        (10, "updated_author"),
+        (11, "craig_alanson"),
+        (12, "glynn_stewart"),
+    ]
+    assert [(row.name, row.author.name) for row in series] == [
+        ("expeditionary_force", "craig_alanson"),
+        ("starships_mage", "glynn_stewart"),
+    ]
+
+
+def test_upsert_series_csv_updates_series_by_id(tmp_path, audiobook_session) -> None:
+    author = AudiobookAuthor(id=1, name="glynn_stewart")
+    audiobook_session.add_all(
+        [
+            author,
+            AudiobookSeries(id=7, name="old_series", author=author),
+        ],
+    )
+    audiobook_session.commit()
+    series_csv = tmp_path / "series.csv"
+    series_csv.write_text(
+        "name,author_name,id\n"
+        "starships_mage,glynn_stewart,7\n",
+        encoding="utf-8",
+    )
+
+    count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
+    audiobook_session.commit()
+
+    series = audiobook_session.get(AudiobookSeries, 7)
+    assert count == 1
+    assert series.name == "starships_mage"
+    assert series.author.name == "glynn_stewart"
+
+
+def test_upsert_csv_allows_missing_id_column(tmp_path, audiobook_session) -> None:
+    authors_csv = tmp_path / "authors.csv"
+    series_csv = tmp_path / "series.csv"
+    authors_csv.write_text(
+        "name\n"
+        "glynn_stewart\n",
+        encoding="utf-8",
+    )
+    series_csv.write_text(
+        "name,author_name\n"
+        "starships_mage,glynn_stewart\n",
+        encoding="utf-8",
+    )
+
+    author_count = catalog.upsert_authors_from_csv(audiobook_session, authors_csv)
+    series_count = catalog.upsert_series_from_csv(audiobook_session, series_csv)
+    audiobook_session.commit()
+
+    series = audiobook_session.scalar(select(AudiobookSeries))
+    assert author_count == 1
+    assert series_count == 1
+    assert series.name == "starships_mage"
+    assert series.author.name == "glynn_stewart"
+
+
+def test_upsert_series_csv_rejects_unknown_author(tmp_path, audiobook_session) -> None:
+    series_csv = tmp_path / "series.csv"
+    series_csv.write_text(
+        "name,author_name,id\n"
+        "starships_mage,glynn_stewart,\n",
+        encoding="utf-8",
+    )
+
+    with pytest.raises(catalog.CatalogImportError) as error:
+        catalog.upsert_series_from_csv(audiobook_session, series_csv)
+
+    assert "author not found: glynn_stewart" in str(error.value)