resumable ollama analysis run with per-track writes

sort_ym/analyze.py

@@ -5,7 +5,9 @@ import json
 import urllib.request
 from collections import Counter
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
+from typing import Callable
 
 ANALYSIS_CACHE_FILE = "lyrics_analysis.json"
 
@@ -194,3 +196,74 @@ def call_ollama(settings: OllamaSettings, prompt: str) -> dict:
     with urllib.request.urlopen(request, timeout=settings.timeout) as response:
         body = json.loads(response.read().decode("utf-8"))
     return json.loads(body["message"]["content"])
+
+
+def _atomic_write_json(path: Path, data: dict) -> None:
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+    tmp.replace(path)
+
+
+def _track_meta(tracks_cache: dict[str, dict]) -> dict[str, dict]:
+    """Числовой id -> название и исполнители. Ключ tracks_cache составной ("id:album_id"),
+    а кэши текста и разбора живут на числовом id, как и cache/lyrics_lang.json."""
+    return {str(t["id"]): {"title": t["title"], "artists": t["artists"]} for t in tracks_cache.values()}
+
+
+def analyze_tracks(
+    tracks_cache: dict[str, dict],
+    lyrics_cache: dict[str, str | None],
+    cache_dir: Path,
+    settings: OllamaSettings,
+    limit: int | None = None,
+    call: Callable[[OllamaSettings, str], dict] = call_ollama,
+) -> dict[str, dict]:
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    cache_file = cache_dir / ANALYSIS_CACHE_FILE
+    cache = load_analysis(cache_dir)
+    meta = _track_meta(tracks_cache)
+
+    pending = []
+    for tid, text in lyrics_cache.items():
+        if not text:
+            continue
+        text_hash = lyrics_hash(text)
+        if is_fresh(cache.get(tid), settings.model, settings.prompt_version, text_hash):
+            continue
+        pending.append((tid, text, text_hash))
+
+    if limit is not None:
+        pending = pending[:limit]
+
+    if not pending:
+        print("Все тексты уже разобраны.")
+        return cache
+
+    print(f"Разбор текстов: {len(pending)} треков, модель {settings.model}")
+    failed = 0
+    for i, (tid, text, text_hash) in enumerate(pending, 1):
+        info = meta.get(tid, {"title": tid, "artists": []})
+        prompt = build_prompt(info["title"], info["artists"], text, top_themes(cache))
+        stamp = {
+            "model": settings.model,
+            "prompt_version": settings.prompt_version,
+            "lyrics_hash": text_hash,
+            "analyzed_at": datetime.now().isoformat(timespec="seconds"),
+        }
+        try:
+            data = call(settings, prompt)
+        # Широкий except намеренно: трёхчасовой прогон не должен падать целиком из-за
+        # одного зависшего/битого ответа локальной модели. KeyboardInterrupt и SystemExit
+        # наследуются от BaseException и сюда не попадают - Ctrl+C прерывает прогон.
+        except Exception as e:
+            cache[tid] = {**stamp, "error": f"{type(e).__name__}: {e}"}
+            failed += 1
+            print(f"  [{i}/{len(pending)}] {info['title']} - ошибка: {type(e).__name__}")
+        else:
+            cache[tid] = {**data, **stamp, "key_line_verified": verify_key_line(data["key_line"], text)}
+            print(f"  [{i}/{len(pending)}] {info['title']} - {', '.join(data['mood'])}")
+        _atomic_write_json(cache_file, cache)
+
+    if failed:
+        print(f"  не разобрано: {failed} - повторный запуск analyze попробует их снова")
+    return cache

tests/test_analyze.py

@@ -169,3 +169,157 @@ def test_build_prompt_omits_known_themes_section_when_empty():
     prompt = analyze.build_prompt("Тоска", ["Артист"], "Текст песни", [])
 
     assert "Уже использованные темы" not in prompt
+
+
+GOOD_RESPONSE = {
+    "mood": ["melancholy", "longing"],
+    "themes": ["modern_loneliness", "city"],
+    "emotional_arc": "descent",
+    "pov": "confessional",
+    "register": "conversational",
+    "concreteness": "abstract",
+    "stance": "sincere",
+    "confidence": 0.9,
+    "summary": "Герой бродит по городу.",
+    "resonance": "Узнаваемое чувство одиночества.",
+    "key_line": "А я иду, шагаю по Москве",
+}
+
+TRACKS = {
+    "100:10": {"id": 100, "album_id": 10, "title": "Тоска", "artists": ["Артист"]},
+    "200:20": {"id": 200, "album_id": 20, "title": "Вторая", "artists": ["Другой"]},
+}
+LYRICS = {"100": "Первая строка\nА я иду, шагаю по Москве", "200": "Совсем другой текст"}
+
+
+def test_analyze_stores_metadata_and_verifies_key_line(tmp_path: Path):
+    settings = _settings()
+
+    result = analyze.analyze_tracks(
+        TRACKS, {"100": LYRICS["100"]}, tmp_path, settings, call=lambda s, p: dict(GOOD_RESPONSE)
+    )
+
+    entry = result["100"]
+    assert entry["model"] == settings.model
+    assert entry["prompt_version"] == 1
+    assert entry["lyrics_hash"] == analyze.lyrics_hash(LYRICS["100"])
+    assert entry["key_line_verified"] is True
+    assert entry["mood"] == ["melancholy", "longing"]
+    assert "analyzed_at" in entry
+    assert analyze.load_analysis(tmp_path) == result
+
+
+def test_analyze_marks_paraphrased_key_line_as_unverified(tmp_path: Path):
+    response = dict(GOOD_RESPONSE, key_line="Герой идёт по городу")
+
+    result = analyze.analyze_tracks(TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=lambda s, p: response)
+
+    assert result["100"]["key_line_verified"] is False
+
+
+def test_analyze_writes_after_every_track(tmp_path: Path):
+    # Прогон на 266 треках идёт ~3 часа: падение на середине не должно стоить больше
+    # одного трека, поэтому запись после каждого, а не пачками.
+    def call(settings, prompt):
+        if "Вторая" in prompt:
+            raise KeyboardInterrupt("пользователь прервал прогон")
+        return dict(GOOD_RESPONSE)
+
+    try:
+        analyze.analyze_tracks(TRACKS, LYRICS, tmp_path, _settings(), call=call)
+    except KeyboardInterrupt:
+        pass
+
+    persisted = analyze.load_analysis(tmp_path)
+    assert "100" in persisted, "первый трек должен быть на диске несмотря на обрыв на втором"
+
+
+def test_analyze_records_error_marker_and_continues(tmp_path: Path):
+    def call(settings, prompt):
+        if "Тоска" in prompt:
+            raise TimeoutError("модель зависла")
+        return dict(GOOD_RESPONSE)
+
+    result = analyze.analyze_tracks(TRACKS, LYRICS, tmp_path, _settings(), call=call)
+
+    assert "TimeoutError" in result["100"]["error"]
+    assert result["200"]["summary"] == "Герой бродит по городу.", "таймаут одного трека не роняет батч"
+
+
+def test_analyze_retries_error_entries_on_next_run(tmp_path: Path):
+    def failing(settings, prompt):
+        raise TimeoutError("модель зависла")
+
+    analyze.analyze_tracks(TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=failing)
+    result = analyze.analyze_tracks(
+        TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=lambda s, p: dict(GOOD_RESPONSE)
+    )
+
+    assert "error" not in result["100"]
+    assert result["100"]["summary"] == "Герой бродит по городу."
+
+
+def test_analyze_skips_fresh_entries(tmp_path: Path):
+    def boom(settings, prompt):
+        raise AssertionError("не должен переанализировать свежую запись")
+
+    analyze.analyze_tracks(TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=lambda s, p: dict(GOOD_RESPONSE))
+    result = analyze.analyze_tracks(TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=boom)
+
+    assert result["100"]["summary"] == "Герой бродит по городу."
+
+
+def test_analyze_reanalyzes_when_lyrics_text_changed(tmp_path: Path):
+    analyze.analyze_tracks(TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=lambda s, p: dict(GOOD_RESPONSE))
+
+    updated = dict(GOOD_RESPONSE, summary="Исправленный текст, другой разбор.")
+    result = analyze.analyze_tracks(TRACKS, {"100": "Исправленный текст песни"}, tmp_path, _settings(), call=lambda s, p: updated)
+
+    assert result["100"]["summary"] == "Исправленный текст, другой разбор."
+    assert result["100"]["lyrics_hash"] == analyze.lyrics_hash("Исправленный текст песни")
+
+
+def test_analyze_reanalyzes_when_prompt_version_bumped(tmp_path: Path):
+    analyze.analyze_tracks(TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(), call=lambda s, p: dict(GOOD_RESPONSE))
+
+    updated = dict(GOOD_RESPONSE, summary="Разбор по новому промпту.")
+    result = analyze.analyze_tracks(
+        TRACKS, {"100": LYRICS["100"]}, tmp_path, _settings(prompt_version=2), call=lambda s, p: updated
+    )
+
+    assert result["100"]["summary"] == "Разбор по новому промпту."
+    assert result["100"]["prompt_version"] == 2
+
+
+def test_analyze_skips_tracks_without_text(tmp_path: Path):
+    def boom(settings, prompt):
+        raise AssertionError("нечего анализировать у трека без текста")
+
+    result = analyze.analyze_tracks(TRACKS, {"100": None, "200": ""}, tmp_path, _settings(), call=boom)
+
+    assert result == {}
+
+
+def test_analyze_limit_processes_only_first_n(tmp_path: Path):
+    calls = []
+
+    def call(settings, prompt):
+        calls.append(prompt)
+        return dict(GOOD_RESPONSE)
+
+    analyze.analyze_tracks(TRACKS, LYRICS, tmp_path, _settings(), limit=1, call=call)
+
+    assert len(calls) == 1
+
+
+def test_analyze_feeds_known_themes_into_later_prompts(tmp_path: Path):
+    prompts = []
+
+    def call(settings, prompt):
+        prompts.append(prompt)
+        return dict(GOOD_RESPONSE)
+
+    analyze.analyze_tracks(TRACKS, LYRICS, tmp_path, _settings(), call=call)
+
+    assert "modern_loneliness" not in prompts[0], "первому треку подмешивать нечего"
+    assert "modern_loneliness" in prompts[1], "темы первого трека должны попасть в промпт второго"