diff --git a/sort_ym/analyze.py b/sort_ym/analyze.py new file mode 100644 index 0000000..65ad920 --- /dev/null +++ b/sort_ym/analyze.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import hashlib +import json +from collections import Counter +from pathlib import Path + +ANALYSIS_CACHE_FILE = "lyrics_analysis.json" + +# Настроение - закрытый список: структурная часть схемы существует ради будущей +# фильтрации/агрегации ("плейлист по настроению"), а свободная строка в пилоте схлопывалась +# в составные значения вроде "aggressive/triumphant". Ирония/сарказм сюда намеренно не входят - +# это ось stance, смешение осей и порождало составные значения. +MOODS = [ + "melancholy", + "longing", + "nostalgia", + "tenderness", + "sensuality", + "euphoria", + "serenity", + "playfulness", + "defiance", + "rage", + "anxiety", + "despair", + "grief", + "resolve", +] +EMOTIONAL_ARCS = ["static", "descent", "uplift", "turn"] +POVS = ["confessional", "narrative", "abstract", "dialogue"] +REGISTERS = ["poetic", "conversational", "slang", "archaic"] +CONCRETENESS = ["concrete", "mixed", "abstract"] +STANCES = ["sincere", "ironic", "bitter"] + +# Темы не закрытый enum (специфичность вроде digital_aging ценна), но язык и формат +# фиксируются схемой: в пилоте без паттерна язык плавал между английским и русским внутри +# одного прогона. Паттерн на уровне JSON Schema делает кириллицу и составные значения +# механически невозможными на этапе генерации, а не отлавливаемыми постфактум. +THEME_PATTERN = "^[a-z][a-z0-9_]{2,29}$" + + +def _enum(values: list[str]) -> dict: + return {"type": "string", "enum": values} + + +RESPONSE_SCHEMA: dict = { + "type": "object", + "properties": { + "mood": { + "type": "array", + "items": _enum(MOODS), + "minItems": 1, + "maxItems": 3, + "uniqueItems": True, + }, + "themes": { + "type": "array", + "items": {"type": "string", "pattern": THEME_PATTERN}, + "minItems": 2, + "maxItems": 5, + }, + "emotional_arc": _enum(EMOTIONAL_ARCS), + "pov": _enum(POVS), + "register": _enum(REGISTERS), + "concreteness": _enum(CONCRETENESS), + "stance": _enum(STANCES), + "confidence": {"type": "number", "minimum": 0.0, "maximum": 1.0}, + "summary": {"type": "string"}, + "resonance": {"type": "string"}, + "key_line": {"type": "string"}, + }, + "required": [ + "mood", + "themes", + "emotional_arc", + "pov", + "register", + "concreteness", + "stance", + "confidence", + "summary", + "resonance", + "key_line", + ], + "additionalProperties": False, +} + + +def lyrics_hash(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _normalized(text: str) -> str: + return " ".join(text.split()).casefold() + + +def verify_key_line(key_line: str, lyrics: str) -> bool: + """Цитата ли key_line на самом деле - модель иногда пересказывает вместо цитирования.""" + if not key_line.strip(): + return False + return _normalized(key_line) in _normalized(lyrics) + + +def is_fresh(entry: dict | None, model: str, prompt_version: int, text_hash: str) -> bool: + if not entry or "error" in entry: + return False + return ( + entry.get("model") == model + and entry.get("prompt_version") == prompt_version + and entry.get("lyrics_hash") == text_hash + ) + + +def top_themes(cache: dict[str, dict], limit: int = 60) -> list[str]: + """Самые частые уже присвоенные темы - подмешиваются в промпт, чтобы словарь тем + сходился по всей библиотеке, а не расползался в синонимы.""" + counter: Counter[str] = Counter() + for entry in cache.values(): + counter.update(entry.get("themes", [])) + return [theme for theme, _ in sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))[:limit]] + + +def load_analysis(cache_dir: Path) -> dict[str, dict]: + cache_file = cache_dir / ANALYSIS_CACHE_FILE + if cache_file.exists(): + return json.loads(cache_file.read_text(encoding="utf-8")) + return {} diff --git a/tests/test_analyze.py b/tests/test_analyze.py new file mode 100644 index 0000000..117521e --- /dev/null +++ b/tests/test_analyze.py @@ -0,0 +1,87 @@ +import re +from pathlib import Path + +from sort_ym import analyze + + +def test_theme_pattern_accepts_english_snake_case(): + assert re.match(analyze.THEME_PATTERN, "digital_aging") + assert re.match(analyze.THEME_PATTERN, "modern_loneliness") + + +def test_theme_pattern_rejects_pilot_failure_modes(): + # Пилот без паттерна давал кириллицу и составные значения через слэш вперемешку + # с английским в одном прогоне - агрегировать такое невозможно. + assert not re.match(analyze.THEME_PATTERN, "одиночество") + assert not re.match(analyze.THEME_PATTERN, "эмоциональное выгорание") + assert not re.match(analyze.THEME_PATTERN, "Digital_Aging") + assert not re.match(analyze.THEME_PATTERN, "ethereal/hopeful") + assert not re.match(analyze.THEME_PATTERN, "ab") + + +def test_mood_enum_excludes_stance_axis(): + # Смешение осей в пилоте давало "aggressive/triumphant"; ирония/сарказм живут в stance. + assert "ironic" not in analyze.MOODS + assert "sarcastic" not in analyze.MOODS + assert "sincere" not in analyze.MOODS + assert len(analyze.MOODS) == len(set(analyze.MOODS)) + + +def test_response_schema_constrains_mood_and_themes(): + props = analyze.RESPONSE_SCHEMA["properties"] + assert props["mood"]["items"]["enum"] == analyze.MOODS + assert props["mood"]["maxItems"] == 3 + assert props["themes"]["items"]["pattern"] == analyze.THEME_PATTERN + assert set(analyze.RESPONSE_SCHEMA["required"]) == set(props) + + +def test_lyrics_hash_changes_with_text(): + assert analyze.lyrics_hash("текст") == analyze.lyrics_hash("текст") + assert analyze.lyrics_hash("текст") != analyze.lyrics_hash("другой текст") + + +def test_verify_key_line_accepts_verbatim_quote_ignoring_whitespace_and_case(): + text = "Первая строка\nА я иду, шагаю по Москве\nТретья строка" + + assert analyze.verify_key_line("А я иду, шагаю по Москве", text) + assert analyze.verify_key_line("а я иду, шагаю по москве", text) + + +def test_verify_key_line_rejects_paraphrase_and_empty(): + text = "А я иду, шагаю по Москве" + + assert not analyze.verify_key_line("Герой идёт по городу", text) + assert not analyze.verify_key_line("", text) + + +def test_is_fresh_requires_model_prompt_version_and_hash_match(): + entry = {"model": "m1", "prompt_version": 1, "lyrics_hash": "h1"} + + assert analyze.is_fresh(entry, "m1", 1, "h1") + assert not analyze.is_fresh(entry, "m2", 1, "h1") + assert not analyze.is_fresh(entry, "m1", 2, "h1") + assert not analyze.is_fresh(entry, "m1", 1, "h2") + assert not analyze.is_fresh(None, "m1", 1, "h1") + + +def test_is_fresh_false_for_error_marker(): + # Ошибочные записи перезапрашиваются при следующем прогоне - это и есть механизм + # "повторить упавшие треки" вместо отдельного флага. + entry = {"model": "m1", "prompt_version": 1, "lyrics_hash": "h1", "error": "TimeoutError: ..."} + + assert not analyze.is_fresh(entry, "m1", 1, "h1") + + +def test_top_themes_orders_by_frequency_and_limits(): + cache = { + "1": {"themes": ["loneliness", "city"]}, + "2": {"themes": ["loneliness", "memory"]}, + "3": {"themes": ["loneliness", "city"]}, + "4": {"error": "TimeoutError: ..."}, + } + + assert analyze.top_themes(cache, limit=2) == ["loneliness", "city"] + + +def test_load_analysis_returns_empty_dict_when_missing(tmp_path: Path): + assert analyze.load_analysis(tmp_path) == {}