liumangmang
386 lines
14 KiB
Python
386 lines
14 KiB
Python
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from backend.app.exception_service import ExceptionService
|
|
from backend.app.library_service import LibraryService, LibraryTrackNotFoundError
|
|
from backend.app.task_store import TaskStore
|
|
|
|
|
|
class LibraryServiceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.root = Path(tempfile.mkdtemp())
|
|
self.output_dir = self.root / 'output'
|
|
self.output_dir.mkdir()
|
|
self.task_store = TaskStore(self.root / 'music_workshop.db')
|
|
self.preprocessor = _FakePreprocessor()
|
|
self.service = LibraryService(
|
|
self.task_store,
|
|
self.preprocessor,
|
|
read_tags=self.preprocessor.read_tags
|
|
)
|
|
|
|
def tearDown(self):
|
|
shutil.rmtree(self.root)
|
|
|
|
def test_empty_output_dir_returns_empty_summary_and_tracks(self):
|
|
summary = self.service.get_summary(str(self.output_dir))
|
|
page = self.service.get_tracks_page(str(self.output_dir))
|
|
|
|
self.assertEqual(summary['total_tracks'], 0)
|
|
self.assertEqual(summary['total_albums'], 0)
|
|
self.assertEqual(summary['total_artists'], 0)
|
|
self.assertEqual(summary['suspected_duplicates'], 0)
|
|
self.assertEqual(page['items'], [])
|
|
self.assertEqual(page['total'], 0)
|
|
|
|
def test_scans_metadata_audio_and_filters_tracks(self):
|
|
first_path = self._write_library_file('A/Artist A/Album A/01 - Echoes.flac', _timestamp(2024, 1, 1))
|
|
second_path = self._write_library_file('B/Artist B/Album B/03 - Neon.mp3', _timestamp(2024, 1, 2))
|
|
self.preprocessor.audio_props[str(first_path)] = {
|
|
'format': 'FLAC',
|
|
'codec': 'FLAC',
|
|
'bitrate': 980000,
|
|
'sample_rate': 96000,
|
|
'bit_depth': 24,
|
|
'channels': 2,
|
|
'duration_seconds': 301.4
|
|
}
|
|
self.preprocessor.tags[str(first_path)] = {
|
|
'title': 'Echoes',
|
|
'artist': 'Artist A',
|
|
'album': 'Album A',
|
|
'albumartist': 'Artist A',
|
|
'tracknumber': '1',
|
|
'discnumber': '1',
|
|
'date': '2024-01-01'
|
|
}
|
|
self.preprocessor.audio_props[str(second_path)] = {
|
|
'format': 'MP3',
|
|
'codec': 'MP3',
|
|
'bitrate': 320000,
|
|
'sample_rate': 44100,
|
|
'bit_depth': 16,
|
|
'channels': 2,
|
|
'duration_seconds': 240.1
|
|
}
|
|
self.preprocessor.tags[str(second_path)] = {
|
|
'title': 'Neon',
|
|
'artist': 'Artist B',
|
|
'album': 'Album B',
|
|
'albumartist': 'Artist B',
|
|
'tracknumber': '3',
|
|
'discnumber': '1',
|
|
'date': '2023-12-01'
|
|
}
|
|
task = self._create_completed_task()
|
|
self._insert_provenance_item(
|
|
task['task_id'],
|
|
library_file_path=str(first_path),
|
|
library_relative_path='A/Artist A/Album A/01 - Echoes.flac',
|
|
updated_at='2024-01-03T09:00:00Z',
|
|
match_source='musicbrainz',
|
|
match_confidence=94.5,
|
|
dedupe_status='unique'
|
|
)
|
|
|
|
page = self.service.get_tracks_page(
|
|
str(self.output_dir),
|
|
q='echo',
|
|
artist='Artist A',
|
|
album='Album A',
|
|
format='flac',
|
|
has_provenance=True
|
|
)
|
|
|
|
self.assertEqual(page['total'], 1)
|
|
track = page['items'][0]
|
|
self.assertEqual(track['filename'], '01 - Echoes.flac')
|
|
self.assertEqual(track['title'], 'Echoes')
|
|
self.assertEqual(track['artist'], 'Artist A')
|
|
self.assertEqual(track['album'], 'Album A')
|
|
self.assertEqual(track['format'], 'FLAC')
|
|
self.assertEqual(track['codec'], 'FLAC')
|
|
self.assertEqual(track['bit_depth'], 24)
|
|
self.assertEqual(track['sample_rate'], 96000)
|
|
self.assertEqual(track['ingest_provenance']['task_id'], task['task_id'])
|
|
self.assertEqual(track['ingest_provenance']['match_source'], 'musicbrainz')
|
|
self.assertEqual(track['ingest_provenance']['dedupe_status'], 'unique')
|
|
|
|
def test_default_sort_prefers_organized_at_and_falls_back_to_modified_at(self):
|
|
newest_path = self._write_library_file('N/Newest/Album/01 - Fresh.flac', _timestamp(2024, 1, 3))
|
|
organized_path = self._write_library_file('O/Organized/Album/01 - Sorted.flac', _timestamp(2024, 1, 1))
|
|
oldest_path = self._write_library_file('Z/Oldest/Album/01 - Archive.flac', _timestamp(2023, 12, 31))
|
|
|
|
for path, title, artist in (
|
|
(newest_path, 'Fresh', 'Newest'),
|
|
(organized_path, 'Sorted', 'Organized'),
|
|
(oldest_path, 'Archive', 'Oldest')
|
|
):
|
|
self.preprocessor.audio_props[str(path)] = {'format': 'FLAC', 'codec': 'FLAC'}
|
|
self.preprocessor.tags[str(path)] = {
|
|
'title': title,
|
|
'artist': artist,
|
|
'album': 'Album',
|
|
'albumartist': artist,
|
|
'tracknumber': '1',
|
|
'discnumber': '1'
|
|
}
|
|
|
|
task = self._create_completed_task()
|
|
self._insert_provenance_item(
|
|
task['task_id'],
|
|
library_file_path=str(organized_path),
|
|
library_relative_path='O/Organized/Album/01 - Sorted.flac',
|
|
updated_at='2024-01-02T12:00:00Z'
|
|
)
|
|
|
|
page = self.service.get_tracks_page(str(self.output_dir))
|
|
ordered_titles = [item['title'] for item in page['items']]
|
|
|
|
self.assertEqual(ordered_titles, ['Fresh', 'Sorted', 'Archive'])
|
|
|
|
def test_provenance_prefers_absolute_path_then_falls_back_to_relative_path(self):
|
|
exact_path = self._write_library_file('A/Artist/Album/01 - Exact.flac', _timestamp(2024, 1, 1))
|
|
fallback_path = self._write_library_file('B/Artist/Album/02 - Fallback.flac', _timestamp(2024, 1, 1))
|
|
|
|
for path, title in ((exact_path, 'Exact'), (fallback_path, 'Fallback')):
|
|
self.preprocessor.audio_props[str(path)] = {'format': 'FLAC', 'codec': 'FLAC'}
|
|
self.preprocessor.tags[str(path)] = {
|
|
'title': title,
|
|
'artist': 'Artist',
|
|
'album': 'Album',
|
|
'albumartist': 'Artist',
|
|
'tracknumber': '1',
|
|
'discnumber': '1'
|
|
}
|
|
|
|
old_task = self._create_completed_task()
|
|
self._insert_provenance_item(
|
|
old_task['task_id'],
|
|
library_file_path=str(exact_path),
|
|
library_relative_path='A/Artist/Album/01 - Exact.flac',
|
|
updated_at='2024-01-01T08:00:00Z',
|
|
match_source='exact-source'
|
|
)
|
|
newer_task = self._create_completed_task()
|
|
self._insert_provenance_item(
|
|
newer_task['task_id'],
|
|
library_file_path='/legacy/output/A/Artist/Album/01 - Exact.flac',
|
|
library_relative_path='A/Artist/Album/01 - Exact.flac',
|
|
updated_at='2024-01-03T08:00:00Z',
|
|
match_source='relative-source'
|
|
)
|
|
fallback_task = self._create_completed_task()
|
|
self._insert_provenance_item(
|
|
fallback_task['task_id'],
|
|
library_file_path='/legacy/output/B/Artist/Album/02 - Fallback.flac',
|
|
library_relative_path='B/Artist/Album/02 - Fallback.flac',
|
|
updated_at='2024-01-04T08:00:00Z',
|
|
match_source='fallback-source'
|
|
)
|
|
|
|
page = self.service.get_tracks_page(str(self.output_dir), sort_by='filename', sort_order='asc')
|
|
exact_track = next(item for item in page['items'] if item['title'] == 'Exact')
|
|
fallback_track = next(item for item in page['items'] if item['title'] == 'Fallback')
|
|
|
|
self.assertEqual(exact_track['ingest_provenance']['match_source'], 'exact-source')
|
|
self.assertEqual(fallback_track['ingest_provenance']['match_source'], 'fallback-source')
|
|
|
|
def test_summary_counts_suspected_duplicates_without_false_live_match(self):
|
|
duplicate_one = self._write_library_file('A/Artist/Album/01 - Song.flac', _timestamp(2024, 1, 1))
|
|
duplicate_two = self._write_library_file('A/Artist/Album/01 - Song Copy.flac', _timestamp(2024, 1, 2))
|
|
studio = self._write_library_file('S/Artist/Singles/2024 - Ballad/01 - Ballad.flac', _timestamp(2024, 1, 3))
|
|
live = self._write_library_file('S/Artist/Singles/2024 - Ballad Live/01 - Ballad Live.flac', _timestamp(2024, 1, 4))
|
|
|
|
for path in (duplicate_one, duplicate_two, studio, live):
|
|
self.preprocessor.audio_props[str(path)] = {
|
|
'format': 'FLAC',
|
|
'codec': 'FLAC',
|
|
'duration_seconds': 201
|
|
}
|
|
|
|
self.preprocessor.tags[str(duplicate_one)] = {
|
|
'title': 'Song',
|
|
'artist': 'Artist',
|
|
'album': 'Album',
|
|
'albumartist': 'Artist',
|
|
'tracknumber': '1',
|
|
'discnumber': '1',
|
|
'musicbrainzrecordingid': 'recording-1'
|
|
}
|
|
self.preprocessor.tags[str(duplicate_two)] = {
|
|
'title': 'Song',
|
|
'artist': 'Artist',
|
|
'album': 'Album',
|
|
'albumartist': 'Artist',
|
|
'tracknumber': '1',
|
|
'discnumber': '1',
|
|
'musicbrainzrecordingid': 'recording-1'
|
|
}
|
|
self.preprocessor.tags[str(studio)] = {
|
|
'title': 'Ballad',
|
|
'artist': 'Artist',
|
|
'albumartist': 'Artist'
|
|
}
|
|
self.preprocessor.tags[str(live)] = {
|
|
'title': 'Ballad (Live)',
|
|
'artist': 'Artist',
|
|
'albumartist': 'Artist'
|
|
}
|
|
|
|
summary = self.service.get_summary(str(self.output_dir))
|
|
|
|
self.assertEqual(summary['total_tracks'], 4)
|
|
self.assertEqual(summary['suspected_duplicates'], 1)
|
|
|
|
def test_move_track_to_exception_moves_file_and_creates_match_failed_item(self):
|
|
trash_dir = self.root / 'trash'
|
|
library_path = self._write_library_file('A/Artist/Album/01 - Song.flac', _timestamp(2024, 1, 5))
|
|
self.preprocessor.audio_props[str(library_path)] = {
|
|
'format': 'FLAC',
|
|
'codec': 'FLAC',
|
|
'duration_seconds': 180.25
|
|
}
|
|
self.preprocessor.tags[str(library_path)] = {
|
|
'title': 'Song',
|
|
'artist': 'Artist',
|
|
'album': 'Album',
|
|
'albumartist': 'Artist',
|
|
'tracknumber': '1'
|
|
}
|
|
self.preprocessor.fingerprints[str(library_path)] = {
|
|
'fingerprint': 'abc123',
|
|
'duration_seconds': 180.0
|
|
}
|
|
track = self.service.get_tracks_page(str(self.output_dir))['items'][0]
|
|
|
|
response = self.service.move_track_to_exception(
|
|
{
|
|
'input': str(self.root / 'input'),
|
|
'output': str(self.output_dir),
|
|
'trash': str(trash_dir)
|
|
},
|
|
track['track_id']
|
|
)
|
|
|
|
trash_path = Path(response['trash_file_path'])
|
|
self.assertFalse(library_path.exists())
|
|
self.assertTrue(trash_path.exists())
|
|
self.assertEqual(response['library_relative_path'], 'A/Artist/Album/01 - Song.flac')
|
|
self.assertEqual(self.service.get_tracks_page(str(self.output_dir))['total'], 0)
|
|
|
|
exception_service = ExceptionService(self.task_store)
|
|
page = exception_service.get_items('match_failed')
|
|
self.assertEqual(page['total'], 1)
|
|
exception_item = page['items'][0]
|
|
self.assertEqual(exception_item['exception_id'], response['exception_id'])
|
|
self.assertEqual(exception_item['exception_type'], 'match_failed')
|
|
self.assertEqual(exception_item['exception_reason_code'], 'manual_library_requeue')
|
|
self.assertEqual(exception_item['trash_file_path'], str(trash_path))
|
|
self.assertEqual(exception_item['library_file_path'], str(library_path))
|
|
self.assertIn('retry_match', exception_item['available_actions'])
|
|
|
|
source_item = self.task_store.get_exception_source_item(response['exception_id'])
|
|
self.assertEqual(source_item['current_file_path'], str(trash_path))
|
|
self.assertEqual(source_item['original_tags_json']['title'], 'Song')
|
|
self.assertEqual(source_item['audio_props_json']['codec'], 'FLAC')
|
|
self.assertEqual(source_item['acoustic_fingerprint'], 'abc123')
|
|
|
|
def test_move_track_to_exception_rejects_unknown_track_id(self):
|
|
with self.assertRaises(LibraryTrackNotFoundError):
|
|
self.service.move_track_to_exception(
|
|
{
|
|
'input': str(self.root / 'input'),
|
|
'output': str(self.output_dir),
|
|
'trash': str(self.root / 'trash')
|
|
},
|
|
'unknown-track-id'
|
|
)
|
|
|
|
def _write_library_file(self, relative_path: str, modified_at_timestamp: int) -> Path:
|
|
path = self.output_dir / relative_path
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b'audio-data')
|
|
os.utime(path, (float(modified_at_timestamp), float(modified_at_timestamp)))
|
|
return path
|
|
|
|
def _create_completed_task(self) -> dict:
|
|
task = self.task_store.create_task_if_idle(
|
|
{'input': '', 'output': str(self.output_dir), 'trash': ''}
|
|
)
|
|
self.task_store.update_task(
|
|
task['task_id'],
|
|
status='completed',
|
|
completed_at='2024-01-01T00:00:00Z'
|
|
)
|
|
return task
|
|
|
|
def _insert_provenance_item(
|
|
self,
|
|
task_id: str,
|
|
*,
|
|
library_file_path: str,
|
|
library_relative_path: str,
|
|
updated_at: str,
|
|
match_source: str | None = None,
|
|
match_confidence: float | None = None,
|
|
dedupe_status: str = 'unique'
|
|
):
|
|
item = self.task_store.insert_task_item(
|
|
task_id,
|
|
original_path=library_file_path,
|
|
current_file_path=library_file_path,
|
|
relative_path=library_relative_path,
|
|
filename=Path(library_file_path).name,
|
|
extension=Path(library_file_path).suffix.lower(),
|
|
size_bytes=123,
|
|
modified_at='2024-01-01T00:00:00Z',
|
|
local_cover=None,
|
|
local_lyric=None,
|
|
scan_status='queued',
|
|
scan_reason=None,
|
|
scan_message=None,
|
|
preprocess_status='completed',
|
|
match_status='matched',
|
|
match_source=match_source,
|
|
match_confidence=match_confidence,
|
|
dedupe_status=dedupe_status,
|
|
organize_status='organized',
|
|
library_relative_path=library_relative_path,
|
|
library_file_path=library_file_path
|
|
)
|
|
with self.task_store._connect() as connection:
|
|
connection.execute(
|
|
'UPDATE task_items SET updated_at = ? WHERE id = ?',
|
|
(updated_at, item['id'])
|
|
)
|
|
connection.commit()
|
|
|
|
|
|
class _FakePreprocessor:
|
|
def __init__(self):
|
|
self.audio_props: dict[str, dict] = {}
|
|
self.tags: dict[str, dict] = {}
|
|
self.fingerprints: dict[str, dict] = {}
|
|
|
|
def probe_audio(self, file_path: str) -> dict:
|
|
return self.audio_props.get(file_path, {})
|
|
|
|
def read_tags(self, file_path: str) -> dict:
|
|
return self.tags.get(file_path, {})
|
|
|
|
def calculate_fingerprint(self, file_path: str) -> dict:
|
|
return self.fingerprints.get(file_path, {})
|
|
|
|
|
|
def _timestamp(year: int, month: int, day: int) -> int:
|
|
return int(datetime(year, month, day, tzinfo=timezone.utc).timestamp())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|