Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 105 additions & 45 deletions bookbot/cli.py

Large diffs are not rendered by default.

3 changes: 2 additions & 1 deletion bookbot/config/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,10 @@
import os
from pathlib import Path

import toml
from pydantic import ValidationError

import toml

from .models import Config, Profile


Expand Down
24 changes: 14 additions & 10 deletions bookbot/convert/ffmpeg.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import subprocess
import tempfile
from pathlib import Path
from typing import Any, Dict, List, Optional
from typing import Any


def _safe_unlink(path: Path) -> None:
Expand Down Expand Up @@ -52,7 +52,7 @@ def _find_ffprobe(self) -> str:

raise RuntimeError("FFprobe not found. Please install FFmpeg.")

def probe_file(self, file_path: Path) -> Dict[str, Any]:
def probe_file(self, file_path: Path) -> dict[str, Any]:
"""Get detailed information about an audio file."""
cmd = [
self.ffprobe_path,
Expand All @@ -73,10 +73,12 @@ def probe_file(self, file_path: Path) -> Dict[str, Any]:
raise RuntimeError(f"FFprobe failed: {result.stderr}")

try:
probe_data: Dict[str, Any] = json.loads(result.stdout)
probe_data: dict[str, Any] = json.loads(result.stdout)
return probe_data
except json.JSONDecodeError as e:
raise RuntimeError(f"Failed to parse FFprobe output for {file_path}: {e}") from e
raise RuntimeError(
f"Failed to parse FFprobe output for {file_path}: {e}"
) from e

def get_duration(self, file_path: Path) -> float:
"""Get duration of an audio file in seconds."""
Expand All @@ -92,7 +94,7 @@ def get_duration(self, file_path: Path) -> float:

raise RuntimeError(f"Could not determine duration for {file_path}")

def analyze_loudness(self, file_path: Path) -> Dict[str, float]:
def analyze_loudness(self, file_path: Path) -> dict[str, float]:
"""Analyze loudness using EBU R128."""
cmd = [
self.ffmpeg_path,
Expand All @@ -107,7 +109,9 @@ def analyze_loudness(self, file_path: Path) -> Dict[str, float]:
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
except subprocess.TimeoutExpired as e:
raise RuntimeError(f"FFmpeg loudness analysis timed out for {file_path}") from e
raise RuntimeError(
f"FFmpeg loudness analysis timed out for {file_path}"
) from e

# FFmpeg outputs the JSON to stderr for this filter
stderr_lines = result.stderr.strip().split("\n")
Expand Down Expand Up @@ -201,9 +205,9 @@ def convert_to_aac(

def concatenate_files(
self,
input_files: List[Path],
input_files: list[Path],
output_path: Path,
chapters: Optional[List[Dict]] = None,
chapters: list[dict] | None = None,
) -> bool:
"""Concatenate multiple audio files into one."""
# Create a temporary file list for FFmpeg concat
Expand Down Expand Up @@ -247,7 +251,7 @@ def concatenate_files(
finally:
_safe_unlink(concat_file)

def add_chapters(self, file_path: Path, chapters: List[Dict]) -> bool:
def add_chapters(self, file_path: Path, chapters: list[dict]) -> bool:
"""Add chapter markers to an audio file."""
# Create chapters metadata file
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
Expand Down Expand Up @@ -353,7 +357,7 @@ def can_stream_copy(self, file_path: Path) -> bool:
except Exception:
return False

def set_metadata(self, file_path: Path, metadata: Dict[str, str]) -> bool:
def set_metadata(self, file_path: Path, metadata: dict[str, str]) -> bool:
"""Set metadata tags on an audio file."""
temp_output = file_path.with_suffix(".tmp.m4b")

Expand Down
19 changes: 9 additions & 10 deletions bookbot/convert/pipeline.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,8 @@
import uuid
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

import aiofiles
import aiofiles # type: ignore[import-untyped]
import aiohttp

from ..config.manager import ConfigManager
Expand All @@ -21,13 +20,13 @@ class ConversionPlan:
def __init__(self, plan_id: str):
self.plan_id = plan_id
self.created_at = datetime.now()
self.operations: List[ConversionOperation] = []
self.operations: list[ConversionOperation] = []

def add_operation(self, operation: "ConversionOperation") -> None:
"""Add a conversion operation to the plan."""
self.operations.append(operation)

def to_dict(self) -> Dict:
def to_dict(self) -> dict:
"""Convert plan to dictionary for serialization."""
return {
"plan_id": self.plan_id,
Expand All @@ -45,10 +44,10 @@ def __init__(
self.audiobook_set = audiobook_set
self.output_path = output_path
self.config = config
self.chapters: List[Dict] = []
self.temp_files: List[Path] = []
self.chapters: list[dict] = []
self.temp_files: list[Path] = []

def to_dict(self) -> Dict:
def to_dict(self) -> dict:
"""Convert operation to dictionary for serialization."""
return {
"source_path": str(self.audiobook_set.source_path),
Expand Down Expand Up @@ -121,7 +120,7 @@ async def _execute_operation(self, operation: ConversionOperation) -> None:
try:
# Step 1: Convert tracks to AAC if needed
aac_files = []
chapter_data = []
chapter_data: list[dict[str, object]] = []
current_time = 0.0

for track in sorted(
Expand Down Expand Up @@ -258,7 +257,7 @@ async def _apply_metadata(
metadata["language"] = identity.language

if identity.isbn_13 or identity.isbn_10:
metadata["isbn"] = identity.isbn_13 or identity.isbn_10
metadata["isbn"] = identity.isbn_13 or identity.isbn_10 or ""

metadata["genre"] = "Audiobook"
metadata["comment"] = "Converted by BookBot"
Expand All @@ -285,7 +284,7 @@ async def _embed_cover_art(
except Exception:
continue

async def _download_cover(self, url: str, temp_dir: Path) -> Optional[Path]:
async def _download_cover(self, url: str, temp_dir: Path) -> Path | None:
"""Download cover art from URL."""
try:
async with aiohttp.ClientSession() as session:
Expand Down
3 changes: 1 addition & 2 deletions bookbot/core/discovery.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

import re
from pathlib import Path
from typing import Optional

from mutagen import File as MutagenFile
from mutagen.id3 import ID3NoHeaderError
Expand Down Expand Up @@ -149,7 +148,7 @@ def _create_audiobook_set(

return audiobook_set

def _create_track_from_file(self, file_path: Path) -> Optional[Track]:
def _create_track_from_file(self, file_path: Path) -> Track | None:
"""Create a Track object from an audio file."""
stat = None
try:
Expand Down
4 changes: 2 additions & 2 deletions bookbot/core/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Exception hierarchy with structured error details."""

from typing import Any, Dict, Optional
from typing import Any


class BookBotError(Exception):
Expand All @@ -9,7 +9,7 @@ class BookBotError(Exception):
def __init__(
self,
message: str,
details: Optional[Dict[str, Any]] = None,
details: dict[str, Any] | None = None,
recoverable: bool = False,
):
super().__init__(message)
Expand Down
12 changes: 5 additions & 7 deletions bookbot/core/logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, Optional
from typing import Any


class StructuredLogger:
Expand Down Expand Up @@ -34,15 +34,13 @@ def __init__(self, name: str, log_dir: Path, level: int = logging.INFO):
# Console for errors only
console = logging.StreamHandler(sys.stderr)
console.setLevel(logging.ERROR)
console.setFormatter(
logging.Formatter("%(levelname)s: %(message)s")
)
console.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
self.logger.addHandler(console)

def _json_formatter(self) -> logging.Formatter:
class JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
data: Dict[str, Any] = {
data: dict[str, Any] = {
"ts": datetime.utcnow().isoformat(),
"level": record.levelname,
"msg": record.getMessage(),
Expand Down Expand Up @@ -71,10 +69,10 @@ def debug(self, msg: str, **kwargs: Any) -> None:
self.logger.debug(msg, extra={"extra_data": kwargs})


_loggers: Dict[str, StructuredLogger] = {}
_loggers: dict[str, StructuredLogger] = {}


def get_logger(name: str, log_dir: Optional[Path] = None) -> StructuredLogger:
def get_logger(name: str, log_dir: Path | None = None) -> StructuredLogger:
"""Get or create logger."""
if name not in _loggers:
if log_dir is None:
Expand Down
23 changes: 11 additions & 12 deletions bookbot/core/matching.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import re
import unicodedata
from dataclasses import dataclass
from typing import Dict, List, Optional, Set, Tuple

from rapidfuzz import fuzz

Expand All @@ -22,24 +21,24 @@ class MatchScore:
year_score: float
combined_score: float
confidence: str # 'high' | 'medium' | 'low'
reasons: List[str]
reasons: list[str]


class AdvancedMatcher:
"""Fuzzy matching for audiobook metadata."""

# Common author aliases (expandable via JSON file)
AUTHOR_ALIASES: Dict[str, Set[str]] = {
AUTHOR_ALIASES: dict[str, set[str]] = {
"j.k. rowling": {"joanne rowling", "robert galbraith"},
"stephen king": {"richard bachman"},
"iain banks": {"iain m. banks"},
}

# Articles to strip from titles
ARTICLES: Set[str] = {"the", "a", "an", "el", "la", "le", "der", "die"}
ARTICLES: set[str] = {"the", "a", "an", "el", "la", "le", "der", "die"}

# Series detection patterns (pattern, confidence)
SERIES_PATTERNS: List[Tuple[re.Pattern[str], float]] = [
SERIES_PATTERNS: list[tuple[re.Pattern[str], float]] = [
(re.compile(r"(.+?)\s+(?:book|vol(?:ume)?)\s+(\d+)", re.I), 0.95),
(re.compile(r"(.+?)\s+#(\d+)", re.I), 0.90),
(re.compile(r"(.+?)\s+part\s+(\d+)", re.I), 0.85),
Expand Down Expand Up @@ -122,7 +121,7 @@ def match_title(self, title1: str, title2: str) -> float:

return ratio * 0.4 + token_set * 0.6

def extract_series(self, title: str) -> Optional[Tuple[str, Optional[int], float]]:
def extract_series(self, title: str) -> tuple[str, int | None, float] | None:
"""Extract (series_name, book_number, confidence) from title."""
for pattern, confidence in self.SERIES_PATTERNS:
match = pattern.search(title)
Expand All @@ -138,13 +137,13 @@ def extract_series(self, title: str) -> Optional[Tuple[str, Optional[int], float
def calculate_match(
self,
query_title: str,
query_author: Optional[str],
query_series: Optional[str],
query_year: Optional[int],
query_author: str | None,
query_series: str | None,
query_year: int | None,
result_title: str,
result_authors: List[str],
result_series: Optional[str],
result_year: Optional[int],
result_authors: list[str],
result_series: str | None,
result_year: int | None,
) -> MatchScore:
"""Calculate comprehensive match score."""

Expand Down
3 changes: 2 additions & 1 deletion bookbot/core/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,8 @@ class MatchCandidate(BaseModel):
match_reasons: list[str] = Field(default_factory=list)

@field_validator("confidence_level", mode="before")
def set_confidence_level(cls, v, info: ValidationInfo) -> MatchConfidence:
@classmethod
def set_confidence_level(cls, v: Any, info: ValidationInfo) -> MatchConfidence:
if v is not None and v is not PydanticUndefined:
return MatchConfidence(v)

Expand Down
6 changes: 3 additions & 3 deletions bookbot/core/operations.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,8 @@
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any

from mutagen import File as MutagenFile
from mutagen import MutagenError

from ..config.manager import ConfigManager
Expand Down Expand Up @@ -324,7 +324,7 @@ def apply_tags(
print(f"Failed to apply tags to {track.src_path}: {e}")
return False

def _write_all_tags(self, audio_file: MutagenFile, new_tags: AudioTags) -> None:
def _write_all_tags(self, audio_file: Any, new_tags: AudioTags) -> None:
"""Write all tags, overwriting existing ones."""
tag_mapping = {
"title": new_tags.title,
Expand All @@ -342,7 +342,7 @@ def _write_all_tags(self, audio_file: MutagenFile, new_tags: AudioTags) -> None:
audio_file[key] = value

def _write_missing_tags(
self, audio_file: MutagenFile, new_tags: AudioTags, original_tags: AudioTags
self, audio_file: Any, new_tags: AudioTags, original_tags: AudioTags
) -> None:
"""Write tags only if they don't already exist."""
tag_mapping = {
Expand Down
9 changes: 4 additions & 5 deletions bookbot/core/templates.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
import string
import unicodedata
from pathlib import Path
from typing import Optional

from ..config.models import CasePolicy
from .models import AudiobookSet, ProviderIdentity, Track
Expand All @@ -29,7 +28,7 @@ def __init__(
def generate_folder_name(
self,
audiobook_set: AudiobookSet,
identity: Optional[ProviderIdentity] = None,
identity: ProviderIdentity | None = None,
template: str = "{AuthorLastFirst}/{Title} ({Year})",
) -> str:
"""Generate folder name from template."""
Expand All @@ -41,7 +40,7 @@ def generate_filename(
self,
track: Track,
audiobook_set: AudiobookSet,
identity: Optional[ProviderIdentity] = None,
identity: ProviderIdentity | None = None,
template: str = "{DiscPad}{TrackPad} - {Title}",
zero_padding_width: int = 0,
) -> str:
Expand All @@ -59,8 +58,8 @@ def generate_filename(
def _build_tokens(
self,
audiobook_set: AudiobookSet,
identity: Optional[ProviderIdentity] = None,
track: Optional[Track] = None,
identity: ProviderIdentity | None = None,
track: Track | None = None,
zero_padding_width: int = 0,
) -> dict[str, str]:
"""Build template tokens from metadata."""
Expand Down
Loading