987 lines
32 KiB
Python
987 lines
32 KiB
Python
"""test_audible_convert."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from python.orm.richie import Audiobook, AudiobookAuthor, AudiobookSeries, RichieBase
|
|
from python.tools.audiobook import audible_convert, metadata_agent
|
|
from python.tools.audiobook.metadata_agent import StandardBookMetadata, standard_book_metadata
|
|
|
|
|
|
class FakeOllamaResponse:
|
|
def __init__(self, payload):
|
|
self._payload = payload
|
|
|
|
def raise_for_status(self):
|
|
return None
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class FakeFfprobeError(RuntimeError):
|
|
def __str__(self):
|
|
return "bad ffprobe"
|
|
|
|
|
|
@pytest.fixture
|
|
def audiobook_engine():
|
|
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:
|
|
session.add_all(
|
|
[
|
|
AudiobookAuthor(id=1, name="glynn_stewart"),
|
|
AudiobookAuthor(id=2, name="craig_alanson"),
|
|
AudiobookAuthor(id=4, name="dennis_e_taylor"),
|
|
AudiobookSeries(id=1, name="starships_mage", author_id=1),
|
|
AudiobookSeries(id=2, name="black_fleet_trilogy", author_id=1),
|
|
AudiobookSeries(id=3, name="expeditionary_force", author_id=2),
|
|
AudiobookSeries(id=4, name="bobiverse", author_id=4),
|
|
],
|
|
)
|
|
session.commit()
|
|
yield engine
|
|
engine.dispose()
|
|
|
|
|
|
def install_fake_ollama(monkeypatch, payloads):
|
|
calls = []
|
|
|
|
def fake_post(*args, **kwargs):
|
|
calls.append((args, kwargs))
|
|
return FakeOllamaResponse(payloads.pop(0))
|
|
|
|
monkeypatch.setattr(metadata_agent.httpx, "post", fake_post)
|
|
return calls
|
|
|
|
|
|
def conversion_config(output_directory, *, dry_run=False, overwrite=False):
|
|
return audible_convert.ConversionConfig(
|
|
resolved_output=output_directory,
|
|
ollama_api_key="test-key",
|
|
agent_config=metadata_agent.AgentConfig(),
|
|
engine=create_engine("sqlite+pysqlite:///:memory:"),
|
|
activation_bytes=None,
|
|
dry_run=dry_run,
|
|
overwrite=overwrite,
|
|
)
|
|
|
|
|
|
def sqlite_engine():
|
|
return create_engine("sqlite+pysqlite:///:memory:")
|
|
|
|
|
|
def tool_response(name, arguments):
|
|
return {
|
|
"message": {
|
|
"role": "assistant",
|
|
"content": "",
|
|
"tool_calls": [{"function": {"name": name, "arguments": arguments}}],
|
|
},
|
|
}
|
|
|
|
|
|
def final_response(metadata):
|
|
return {"message": {"role": "assistant", "content": json.dumps(metadata)}}
|
|
|
|
|
|
def fenced_final_response(metadata):
|
|
return {"message": {"role": "assistant", "content": f"```json\n{json.dumps(metadata)}\n```"}}
|
|
|
|
|
|
def test_output_stem_uses_catalog_slugs() -> None:
|
|
metadata = StandardBookMetadata(
|
|
author_id=1,
|
|
author="glynn_stewart",
|
|
book_id=None,
|
|
title="title-slug",
|
|
series_id=1,
|
|
series="starships_mage",
|
|
series_index=1,
|
|
confidence=0.96,
|
|
needs_review=False,
|
|
evidence=["test"],
|
|
)
|
|
|
|
assert audible_convert.output_stem(metadata) == "glynn_stewart-starships_mage_01-title-slug"
|
|
|
|
|
|
def test_convert_aax_file_runs_ffmpeg(tmp_path, monkeypatch) -> None:
|
|
"""test_convert_aax_file_runs_ffmpeg."""
|
|
commands = []
|
|
|
|
def fake_run_command(arguments, *, capture=False):
|
|
assert capture is False
|
|
commands.append(arguments)
|
|
return subprocess.CompletedProcess(arguments, 0, "", "")
|
|
|
|
source = tmp_path / "book.aax"
|
|
destination = tmp_path / "book" / "book.m4b"
|
|
monkeypatch.setattr(audible_convert, "run_command", fake_run_command)
|
|
|
|
audible_convert.convert_aax_file(source, destination, "abc123", overwrite=False)
|
|
|
|
assert commands == [
|
|
[
|
|
"ffmpeg",
|
|
"-hide_banner",
|
|
"-n",
|
|
"-activation_bytes",
|
|
"abc123",
|
|
"-i",
|
|
str(source),
|
|
"-map_metadata",
|
|
"0",
|
|
"-c",
|
|
"copy",
|
|
str(destination),
|
|
],
|
|
]
|
|
assert destination.parent.is_dir()
|
|
|
|
|
|
def test_run_command_redacts_activation_bytes_in_logs_and_errors(monkeypatch, caplog) -> None:
|
|
def fake_run(arguments, *, check, capture_output, text):
|
|
assert check is True
|
|
assert capture_output is False
|
|
assert text is True
|
|
raise subprocess.CalledProcessError(1, arguments)
|
|
|
|
monkeypatch.setattr(audible_convert.subprocess, "run", fake_run)
|
|
caplog.set_level("DEBUG", audible_convert.__name__)
|
|
|
|
with pytest.raises(audible_convert.CommandExecutionError) as error:
|
|
audible_convert.run_command(["ffmpeg", "-activation_bytes", "secret-token", "-i", "book.aax"])
|
|
|
|
assert "secret-token" not in caplog.text
|
|
assert "secret-token" not in str(error.value)
|
|
assert "<redacted>" in caplog.text
|
|
assert "<redacted>" in str(error.value)
|
|
|
|
|
|
def test_write_agent_log_serializes_metadata_as_json_object(tmp_path) -> None:
|
|
metadata = StandardBookMetadata(
|
|
author_id=1,
|
|
author="glynn_stewart",
|
|
book_id=None,
|
|
title="starship-mage",
|
|
series_id=1,
|
|
series="starships_mage",
|
|
series_index=1,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["test"],
|
|
)
|
|
log_file = tmp_path / "agent.jsonl"
|
|
|
|
metadata_agent.write_agent_log(log_file, "final_metadata", metadata=metadata, path=tmp_path)
|
|
|
|
record = json.loads(log_file.read_text(encoding="utf-8"))
|
|
assert record["event"] == "final_metadata"
|
|
assert record["metadata"]["author"] == "glynn_stewart"
|
|
assert record["metadata"]["title"] == "starship-mage"
|
|
assert record["path"] == str(tmp_path)
|
|
|
|
|
|
def test_system_prompt_instructs_agent_to_detect_omnibuses() -> None:
|
|
prompt = metadata_agent.system_prompt()
|
|
|
|
assert "Detect omnibus or box-set editions" in prompt
|
|
assert "books-1-3" in prompt
|
|
assert "Keep series_index as the" in prompt
|
|
|
|
|
|
def test_standard_book_metadata_accepts_valid_tool_output(tmp_path, monkeypatch, audiobook_engine) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Glynn Stewart"}),
|
|
tool_response("search_series", {"query": "starships_mage"}),
|
|
final_response(
|
|
{
|
|
"author_id": 1,
|
|
"book_id": None,
|
|
"title": "starship-mage",
|
|
"series_id": 1,
|
|
"series_index": 1,
|
|
"confidence": 0.95,
|
|
"evidence": ["filename and catalog match"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"Starship Mage.aax",
|
|
{"title": "Starship Mage", "artist": "Glynn Stewart"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata == StandardBookMetadata(
|
|
author_id=1,
|
|
author="glynn_stewart",
|
|
book_id=1,
|
|
title="starship-mage",
|
|
series_id=1,
|
|
series="starships_mage",
|
|
series_index=1,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["filename and catalog match"],
|
|
)
|
|
records = [
|
|
json.loads(line)
|
|
for line in (tmp_path / "agent.jsonl").read_text(encoding="utf-8").splitlines()
|
|
]
|
|
sent = [record for record in records if record["event"] == "llm_messages_sent"]
|
|
received = [record for record in records if record["event"] == "llm_message_received"]
|
|
assert sent[0]["messages"][0]["role"] == "system"
|
|
assert "Starship Mage" in sent[0]["messages"][1]["content"]
|
|
assert received[0]["message"]["tool_calls"][0]["function"]["name"] == "search_authors"
|
|
with Session(audiobook_engine) as session:
|
|
book = session.get(Audiobook, 1)
|
|
assert book.title == "starship-mage"
|
|
assert book.author.name == "glynn_stewart"
|
|
|
|
|
|
def test_standard_book_metadata_uses_agent_config(tmp_path, monkeypatch, audiobook_engine) -> None:
|
|
config = metadata_agent.AgentConfig(
|
|
model="custom-model",
|
|
ollama_chat_url="https://ollama.example.test/api/chat",
|
|
http_timeout_seconds=12,
|
|
max_agent_turns=1,
|
|
min_confidence=0.5,
|
|
tool_names=("search_authors",),
|
|
)
|
|
calls = install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Glynn Stewart"}),
|
|
final_response(
|
|
{
|
|
"author_id": 1,
|
|
"book_id": None,
|
|
"title": "standalone-book",
|
|
"series_id": None,
|
|
"series_index": 0,
|
|
"confidence": 0.5,
|
|
"evidence": ["custom config"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"Standalone Book.aax",
|
|
{"title": "Standalone Book", "artist": "Glynn Stewart"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=config,
|
|
)
|
|
|
|
first_request_url = calls[0][0][0]
|
|
first_request_options = calls[0][1]
|
|
tool_names = [
|
|
tool_schema["function"]["name"]
|
|
for tool_schema in first_request_options["json"]["tools"]
|
|
]
|
|
assert first_request_url == "https://ollama.example.test/api/chat"
|
|
assert first_request_options["timeout"] == 12
|
|
assert first_request_options["json"]["model"] == "custom-model"
|
|
assert tool_names == ["search_authors"]
|
|
assert metadata.needs_review is False
|
|
assert metadata.series == "standalone"
|
|
|
|
|
|
def test_standard_book_metadata_retries_invalid_json_then_needs_review(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Glynn Stewart"}),
|
|
tool_response("search_series", {"query": "Starship Mage"}),
|
|
{"message": {"role": "assistant", "content": "{"}},
|
|
{"message": {"role": "assistant", "content": "{"}},
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"Starship Mage.aax",
|
|
{"title": "Starship Mage"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata.needs_review is True
|
|
assert metadata.confidence == 0
|
|
|
|
|
|
def test_standard_book_metadata_accepts_fenced_final_json(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Dennis E. Taylor"}),
|
|
tool_response("search_series", {"query": "Bobiverse", "author_id": 4}),
|
|
tool_response("search_books", {"query": "All These Worlds", "author_id": 4, "series_id": 4}),
|
|
fenced_final_response(
|
|
{
|
|
"author_id": 4,
|
|
"book_id": None,
|
|
"title": "all-these-worlds",
|
|
"series_id": 4,
|
|
"series_index": 3,
|
|
"confidence": 0.95,
|
|
"evidence": ["fenced json from model"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"All These Worlds.aax",
|
|
{"title": "All These Worlds: Bobiverse, Book 3", "artist": "Dennis E. Taylor"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata.needs_review is False
|
|
assert metadata.author == "dennis_e_taylor"
|
|
assert metadata.series == "bobiverse"
|
|
assert metadata.title == "all-these-worlds"
|
|
|
|
|
|
def test_standard_book_metadata_recovers_from_tool_validation_error(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Cormac McCarthy"}),
|
|
tool_response("ensure_author", {"name": "Cormac McCarthy"}),
|
|
tool_response("ensure_series", {"name": "The Cormac McCarthy Collection", "author_id": 5}),
|
|
tool_response(
|
|
"ensure_book",
|
|
{
|
|
"title": "The Road",
|
|
"author_id": 5,
|
|
"series_id": 5,
|
|
"series_index": 0,
|
|
},
|
|
),
|
|
final_response(
|
|
{
|
|
"author_id": 5,
|
|
"book_id": None,
|
|
"title": "The Road",
|
|
"series_id": None,
|
|
"series_index": 0,
|
|
"confidence": 0.9,
|
|
"evidence": ["tool error showed this should be standalone"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
log_file = tmp_path / "agent.jsonl"
|
|
|
|
metadata = standard_book_metadata(
|
|
"The Road.aax",
|
|
{"title": "The Road", "artist": "Cormac McCarthy"},
|
|
audiobook_engine,
|
|
log_file,
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata == StandardBookMetadata(
|
|
author_id=5,
|
|
author="cormac_mccarthy",
|
|
book_id=1,
|
|
title="the-road",
|
|
series_id=None,
|
|
series="standalone",
|
|
series_index=0,
|
|
confidence=0.9,
|
|
needs_review=False,
|
|
evidence=["tool error showed this should be standalone"],
|
|
)
|
|
assert "series books must use a positive series_index" in log_file.read_text(encoding="utf-8")
|
|
with Session(audiobook_engine) as session:
|
|
assert session.get(AudiobookSeries, 5) is None
|
|
book = session.get(Audiobook, 1)
|
|
assert book.title == "the-road"
|
|
assert book.series_id is None
|
|
|
|
|
|
def test_standard_book_metadata_rejects_unknown_tool(tmp_path, monkeypatch, audiobook_engine) -> None:
|
|
log_file = tmp_path / "agent.jsonl"
|
|
install_fake_ollama(monkeypatch, [tool_response("drop_table", {})])
|
|
|
|
metadata = standard_book_metadata(
|
|
"Book.aax",
|
|
{"title": "Book"},
|
|
audiobook_engine,
|
|
log_file,
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata.needs_review is True
|
|
assert "Unknown audiobook metadata tool" in metadata.evidence[0]
|
|
assert "tool_error" in log_file.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_standard_book_metadata_rejects_ids_not_returned_by_tools(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Glynn Stewart"}),
|
|
tool_response("search_series", {"query": "Starship Mage"}),
|
|
final_response(
|
|
{
|
|
"author_id": 2,
|
|
"book_id": None,
|
|
"title": "expeditionary-force",
|
|
"series_id": 1,
|
|
"series_index": 1,
|
|
"confidence": 0.99,
|
|
"evidence": ["bad id"],
|
|
},
|
|
),
|
|
final_response(
|
|
{
|
|
"author_id": 2,
|
|
"book_id": None,
|
|
"title": "expeditionary-force",
|
|
"series_id": 1,
|
|
"series_index": 1,
|
|
"confidence": 0.99,
|
|
"evidence": ["bad id"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"Book.aax",
|
|
{"title": "Book"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata.needs_review is True
|
|
assert "author_id 2 was not returned" in metadata.evidence[0]
|
|
|
|
|
|
def test_standard_book_metadata_rejects_series_for_wrong_author(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Glynn Stewart"}),
|
|
tool_response("search_series", {"query": "expeditionary_force"}),
|
|
final_response(
|
|
{
|
|
"author_id": 1,
|
|
"book_id": None,
|
|
"title": "expeditionary-force",
|
|
"series_id": 3,
|
|
"series_index": 1,
|
|
"confidence": 0.99,
|
|
"evidence": ["wrong author"],
|
|
},
|
|
),
|
|
final_response(
|
|
{
|
|
"author_id": 1,
|
|
"book_id": None,
|
|
"title": "expeditionary-force",
|
|
"series_id": 3,
|
|
"series_index": 1,
|
|
"confidence": 0.99,
|
|
"evidence": ["wrong author"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"Book.aax",
|
|
{"title": "Book"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata.needs_review is True
|
|
assert "series_id 3 does not belong to author_id 1" in metadata.evidence[0]
|
|
|
|
|
|
def test_standard_book_metadata_forces_final_after_empty_book_searches(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
config = metadata_agent.AgentConfig(max_agent_turns=5)
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Dennis E. Taylor"}),
|
|
tool_response("search_series", {"query": "Bobiverse", "author_id": 4}),
|
|
tool_response("search_books", {"query": "We Are Legion We Are Bob", "author_id": 4, "series_id": 4}),
|
|
tool_response("search_books", {"query": "we are legion", "author_id": 4}),
|
|
tool_response("search_books", {"query": "We Are Legion"}),
|
|
final_response(
|
|
{
|
|
"author_id": 4,
|
|
"book_id": None,
|
|
"title": "we-are-legion-we-are-bob",
|
|
"series_id": 4,
|
|
"series_index": 1,
|
|
"confidence": 0.95,
|
|
"evidence": ["author and series tool results; title from ffprobe tags"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"We_Are_Legion_(We_Are_Bob)_Bobiverse_Book_1-LC_128_44100_stereo.aax",
|
|
{
|
|
"album": "We Are Legion (We Are Bob): Bobiverse, Book 1",
|
|
"artist": "Dennis E. Taylor",
|
|
"title": "We Are Legion (We Are Bob): Bobiverse, Book 1",
|
|
},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=config,
|
|
)
|
|
|
|
assert metadata == StandardBookMetadata(
|
|
author_id=4,
|
|
author="dennis_e_taylor",
|
|
book_id=1,
|
|
title="we-are-legion-we-are-bob",
|
|
series_id=4,
|
|
series="bobiverse",
|
|
series_index=1,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["author and series tool results; title from ffprobe tags"],
|
|
)
|
|
assert '"tools_enabled": false' in (tmp_path / "agent.jsonl").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_standard_book_metadata_can_create_missing_catalog_rows(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Martha Wells"}),
|
|
tool_response("ensure_author", {"name": "martha_wells"}),
|
|
tool_response("search_series", {"query": "Murderbot Diaries", "author_id": 5}),
|
|
tool_response("ensure_series", {"name": "murderbot_diaries", "author_id": 5}),
|
|
tool_response("search_books", {"query": "All Systems Red", "author_id": 5, "series_id": 5}),
|
|
final_response(
|
|
{
|
|
"author_id": 5,
|
|
"book_id": None,
|
|
"title": "all-systems-red",
|
|
"series_id": 5,
|
|
"series_index": 1,
|
|
"confidence": 0.96,
|
|
"evidence": ["created missing author and series; title from tags"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"All Systems Red.aax",
|
|
{"title": "All Systems Red", "artist": "Martha Wells"},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata == StandardBookMetadata(
|
|
author_id=5,
|
|
author="martha_wells",
|
|
book_id=1,
|
|
title="all-systems-red",
|
|
series_id=5,
|
|
series="murderbot_diaries",
|
|
series_index=1,
|
|
confidence=0.96,
|
|
needs_review=False,
|
|
evidence=["created missing author and series; title from tags"],
|
|
)
|
|
with Session(audiobook_engine) as session:
|
|
author = session.get(AudiobookAuthor, 5)
|
|
series = session.get(AudiobookSeries, 5)
|
|
book = session.get(Audiobook, 1)
|
|
assert author.name == "martha_wells"
|
|
assert series.name == "murderbot_diaries"
|
|
assert series.author_id == author.id
|
|
assert book.title == "all-systems-red"
|
|
assert book.author_id == author.id
|
|
assert book.series_id == series.id
|
|
|
|
|
|
def test_standard_book_metadata_normalizes_noisy_created_catalog_rows(
|
|
tmp_path,
|
|
monkeypatch,
|
|
audiobook_engine,
|
|
) -> None:
|
|
install_fake_ollama(
|
|
monkeypatch,
|
|
[
|
|
tool_response("search_authors", {"query": "Charles Lamb"}),
|
|
tool_response("ensure_author", {"name": "charles-lamb"}),
|
|
tool_response("search_series", {"query": "AL:ICE Series", "author_id": 5}),
|
|
tool_response("ensure_series", {"name": "AL:ICE Series", "author_id": 5}),
|
|
tool_response("search_books", {"query": "AL:ICE Space War", "author_id": 5, "series_id": 5}),
|
|
final_response(
|
|
{
|
|
"author_id": 5,
|
|
"book_id": None,
|
|
"title": "AL:ICE Space War",
|
|
"series_id": 5,
|
|
"series_index": 4,
|
|
"confidence": 0.95,
|
|
"evidence": ["created normalized author and series; title from tags"],
|
|
},
|
|
),
|
|
],
|
|
)
|
|
|
|
metadata = standard_book_metadata(
|
|
"ALICE_Space_War_ALICE_Series_Book_4-LC_64_22050_stereo.aax",
|
|
{
|
|
"album": "AL:ICE Space War: AL:ICE Series, Book 4",
|
|
"artist": "Charles Lamb",
|
|
"title": "AL:ICE Space War: AL:ICE Series, Book 4",
|
|
},
|
|
audiobook_engine,
|
|
tmp_path / "agent.jsonl",
|
|
"test-key",
|
|
config=metadata_agent.AgentConfig(),
|
|
)
|
|
|
|
assert metadata == StandardBookMetadata(
|
|
author_id=5,
|
|
author="charles_lamb",
|
|
book_id=1,
|
|
title="al-ice-space-war",
|
|
series_id=5,
|
|
series="al_ice_series",
|
|
series_index=4,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["created normalized author and series; title from tags"],
|
|
)
|
|
with Session(audiobook_engine) as session:
|
|
author = session.get(AudiobookAuthor, 5)
|
|
series = session.get(AudiobookSeries, 5)
|
|
book = session.get(Audiobook, 1)
|
|
assert author.name == "charles_lamb"
|
|
assert series.name == "al_ice_series"
|
|
assert series.author_id == author.id
|
|
assert book.title == "al-ice-space-war"
|
|
assert book.author_id == author.id
|
|
assert book.series_id == series.id
|
|
|
|
|
|
def test_convert_aax_file_with_agent_success_renames_temp_output(tmp_path, monkeypatch) -> None:
|
|
source = tmp_path / "book.aax"
|
|
output_directory = tmp_path / "audiobooks"
|
|
source.touch()
|
|
monkeypatch.setattr(audible_convert, "read_metadata", lambda _: {"title": "Starship Mage"})
|
|
monkeypatch.setattr(
|
|
audible_convert,
|
|
"standard_book_metadata",
|
|
lambda *_, **__: StandardBookMetadata(
|
|
author_id=1,
|
|
author="glynn_stewart",
|
|
book_id=None,
|
|
title="starship-mage",
|
|
series_id=1,
|
|
series="starships_mage",
|
|
series_index=1,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["test"],
|
|
),
|
|
)
|
|
|
|
def fake_convert(_source, destination, _activation_bytes, *, overwrite):
|
|
assert overwrite is True
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text("converted", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert)
|
|
|
|
audible_convert.convert_aax_file_with_agent(
|
|
source,
|
|
conversion_config(output_directory),
|
|
)
|
|
|
|
expected = output_directory / "glynn_stewart-starships_mage_01-starship-mage"
|
|
destination = expected / "glynn_stewart-starships_mage_01-starship-mage.m4b"
|
|
assert destination.read_text(encoding="utf-8") == "converted"
|
|
assert not list((output_directory / ".audible_convert" / "tmp").glob("*/converted.m4b"))
|
|
|
|
|
|
def test_ffprobe_failure_writes_review_without_converting(tmp_path, monkeypatch) -> None:
|
|
source = tmp_path / "book.aax"
|
|
output_directory = tmp_path / "audiobooks"
|
|
source.touch()
|
|
calls = []
|
|
|
|
def fake_read_metadata(_source):
|
|
raise FakeFfprobeError
|
|
|
|
def fake_convert(*args, **kwargs):
|
|
calls.append((args, kwargs))
|
|
|
|
monkeypatch.setattr(audible_convert, "read_metadata", fake_read_metadata)
|
|
monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert)
|
|
|
|
audible_convert.convert_aax_file_with_agent(source, conversion_config(output_directory))
|
|
|
|
review_files = list((output_directory / ".audible_convert" / "review").glob("*.json"))
|
|
assert calls == []
|
|
assert len(review_files) == 1
|
|
review = json.loads(review_files[0].read_text(encoding="utf-8"))
|
|
assert review["ffprobe_metadata"] == {}
|
|
assert review["reason"] == "ffprobe_failed: bad ffprobe"
|
|
assert review["temp_file"] is None
|
|
|
|
|
|
def test_low_confidence_metadata_keeps_temp_output_for_review(tmp_path, monkeypatch) -> None:
|
|
source = tmp_path / "book.aax"
|
|
output_directory = tmp_path / "audiobooks"
|
|
source.touch()
|
|
monkeypatch.setattr(audible_convert, "read_metadata", lambda _: {"title": "Unknown"})
|
|
monkeypatch.setattr(
|
|
audible_convert,
|
|
"standard_book_metadata",
|
|
lambda *_, **__: StandardBookMetadata(
|
|
author_id=0,
|
|
author="unknown_author",
|
|
book_id=None,
|
|
title="unknown-title",
|
|
series_id=None,
|
|
series="standalone",
|
|
series_index=0,
|
|
confidence=0.25,
|
|
needs_review=True,
|
|
evidence=["unclear"],
|
|
),
|
|
)
|
|
|
|
def fake_convert(_source, destination, _activation_bytes, *, overwrite):
|
|
assert overwrite is True
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text("converted", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert)
|
|
|
|
audible_convert.convert_aax_file_with_agent(
|
|
source,
|
|
conversion_config(output_directory),
|
|
)
|
|
|
|
temp_files = list((output_directory / ".audible_convert" / "tmp").glob("*/converted.m4b"))
|
|
review_files = list((output_directory / ".audible_convert" / "review").glob("*.json"))
|
|
assert len(temp_files) == 1
|
|
assert temp_files[0].read_text(encoding="utf-8") == "converted"
|
|
assert len(review_files) == 1
|
|
|
|
|
|
def test_existing_destination_skips_rename_and_removes_temp(tmp_path, monkeypatch) -> None:
|
|
source = tmp_path / "book.aax"
|
|
output_directory = tmp_path / "audiobooks"
|
|
source.touch()
|
|
final_file = (
|
|
output_directory
|
|
/ "glynn_stewart-starships_mage_01-starship-mage"
|
|
/ "glynn_stewart-starships_mage_01-starship-mage.m4b"
|
|
)
|
|
final_file.parent.mkdir(parents=True)
|
|
final_file.write_text("existing", encoding="utf-8")
|
|
monkeypatch.setattr(audible_convert, "read_metadata", lambda _: {"title": "Starship Mage"})
|
|
monkeypatch.setattr(
|
|
audible_convert,
|
|
"standard_book_metadata",
|
|
lambda *_, **__: StandardBookMetadata(
|
|
author_id=1,
|
|
author="glynn_stewart",
|
|
book_id=None,
|
|
title="starship-mage",
|
|
series_id=1,
|
|
series="starships_mage",
|
|
series_index=1,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["test"],
|
|
),
|
|
)
|
|
|
|
def fake_convert(_source, destination, _activation_bytes, *, overwrite):
|
|
assert overwrite is True
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_text("converted", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert)
|
|
|
|
audible_convert.convert_aax_file_with_agent(
|
|
source,
|
|
conversion_config(output_directory),
|
|
)
|
|
|
|
assert final_file.read_text(encoding="utf-8") == "existing"
|
|
assert not list((output_directory / ".audible_convert" / "tmp").glob("*/converted.m4b"))
|
|
|
|
|
|
def test_richie_exports_audiobook_models() -> None:
|
|
from python.orm.richie import Audiobook # noqa: PLC0415
|
|
|
|
assert Audiobook.__tablename__ == "audiobook"
|
|
|
|
|
|
def test_main_dry_run_prints_outputs_without_converting(tmp_path, monkeypatch, capsys) -> None:
|
|
input_directory = tmp_path / "raw"
|
|
output_directory = tmp_path / "audiobooks"
|
|
input_directory.mkdir()
|
|
source = input_directory / "book.aax"
|
|
source.touch()
|
|
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
|
|
monkeypatch.setattr(
|
|
audible_convert,
|
|
"read_metadata",
|
|
lambda _: {
|
|
"artist": "Charles Lamb",
|
|
"title": "Alice: Alice Series #1",
|
|
},
|
|
)
|
|
calls = []
|
|
|
|
def fake_convert(*args, **kwargs):
|
|
calls.append((args, kwargs))
|
|
|
|
monkeypatch.setattr(audible_convert, "convert_aax_file", fake_convert)
|
|
monkeypatch.setattr(
|
|
audible_convert,
|
|
"standard_book_metadata",
|
|
lambda *_, **__: StandardBookMetadata(
|
|
author_id=1,
|
|
author="charles_lamb",
|
|
book_id=None,
|
|
title="alice",
|
|
series_id=1,
|
|
series="alice",
|
|
series_index=1,
|
|
confidence=0.95,
|
|
needs_review=False,
|
|
evidence=["test"],
|
|
),
|
|
)
|
|
|
|
def fake_get_postgres_engine(*, name):
|
|
assert name == "RICHIE"
|
|
return create_engine("sqlite+pysqlite:///:memory:")
|
|
|
|
monkeypatch.setattr(audible_convert, "get_postgres_engine", fake_get_postgres_engine)
|
|
|
|
audible_convert.main(input_directory, output_directory, dry_run=True)
|
|
|
|
assert calls == []
|
|
assert capsys.readouterr().out == (
|
|
f"{source} -> {output_directory / 'charles_lamb-alice_01-alice' / 'charles_lamb-alice_01-alice.m4b'}\n"
|
|
)
|
|
dry_run_file = (
|
|
output_directory
|
|
/ ".audible_convert"
|
|
/ "dry-run"
|
|
/ "charles_lamb-alice_01-alice"
|
|
/ "charles_lamb-alice_01-alice.m4b"
|
|
)
|
|
assert dry_run_file.read_text(encoding="utf-8") == (
|
|
f"{output_directory / 'charles_lamb-alice_01-alice' / 'charles_lamb-alice_01-alice.m4b'}\n"
|
|
)
|
|
assert (output_directory / ".audible_convert" / "logs").is_dir()
|
|
|
|
|
|
def test_main_reads_activation_bytes_from_env(tmp_path, monkeypatch) -> None:
|
|
input_directory = tmp_path / "raw"
|
|
output_directory = tmp_path / "audiobooks"
|
|
input_directory.mkdir()
|
|
source = input_directory / "book.aax"
|
|
source.touch()
|
|
configs = []
|
|
|
|
def fake_convert(_source, config):
|
|
configs.append(config)
|
|
|
|
def fake_get_postgres_engine(*, name):
|
|
assert name == "RICHIE"
|
|
return sqlite_engine()
|
|
|
|
monkeypatch.setenv("OLLAMA_API_KEY", "test-key")
|
|
monkeypatch.setenv("AUDIBLE_ACTIVATION_BYTES", "activation-secret")
|
|
monkeypatch.setattr(audible_convert, "get_postgres_engine", fake_get_postgres_engine)
|
|
monkeypatch.setattr(audible_convert, "convert_aax_file_with_agent", fake_convert)
|
|
|
|
audible_convert.main(input_directory, output_directory)
|
|
|
|
assert configs == [
|
|
audible_convert.ConversionConfig(
|
|
resolved_output=output_directory,
|
|
ollama_api_key="test-key",
|
|
agent_config=configs[0].agent_config,
|
|
engine=configs[0].engine,
|
|
activation_bytes="activation-secret",
|
|
dry_run=False,
|
|
overwrite=False,
|
|
),
|
|
]
|