vidconf/backend/tests/test_chunking.py

"""Тесты чистой функции чанкинга транскрипта `chunk_transcript` (ТЗ §1.3).

Синтетические транскрипты (строки `[Имя MM:SS] текст`) без сети/БД/токенизатора
— подсчёт токенов в тестах подменяется фейковым callable (TDD).
"""

from core.summarization.chunking import chunk_transcript


def _word_count(text: str) -> int:
    """Фейковый подсчёт токенов для тестов: число слов (детерминированно и
    без сторонних зависимостей — реальный `QwenTokenCounter` тестируется
    отдельно)."""
    return len(text.split())


def _line(speaker: str, offset_s: float, text: str) -> str:
    """Собрать строку-фразу транскрипта в формате `[Имя MM:SS] текст`."""
    minutes, seconds = divmod(int(offset_s), 60)
    return f"[{speaker} {minutes:02d}:{seconds:02d}] {text}"


def test_empty_transcript_returns_empty_list() -> None:
    """Пустой транскрипт → пустой список чанков."""
    assert chunk_transcript("", _word_count) == []


def test_blank_transcript_returns_empty_list() -> None:
    """Транскрипт из одних пустых строк — тоже пустой вход."""
    assert chunk_transcript("\n\n  \n", _word_count) == []


def test_short_meeting_fits_in_single_chunk() -> None:
    """Короткая встреча (мало времени и токенов) → один чанк со всеми фразами."""
    lines = [
        _line("Аня", 0, "привет всем"),
        _line("Боря", 3, "привет как дела"),
        _line("Аня", 8, "всё хорошо начинаем"),
    ]
    transcript = "\n".join(lines)

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=20
    )

    assert chunks == [transcript]


def test_long_meeting_splits_into_several_chunks_respecting_token_limit() -> None:
    """Длинная встреча: несколько чанков, ни один не превышает max_chunk_tokens,
    все исходные фразы сохранены без потерь, порядок не нарушен."""
    lines = []
    for i in range(40):
        speaker = "Аня" if i % 2 == 0 else "Боря"
        text = " ".join(["слово"] * 20)  # по 20 токенов на фразу
        lines.append(_line(speaker, i * 5, text))  # фразы каждые 5 секунд
    transcript = "\n".join(lines)

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=100, target_chunk_minutes=20
    )

    assert len(chunks) > 1
    for chunk in chunks:
        assert _word_count(chunk) <= 100

    # все исходные фразы присутствуют ровно один раз, порядок сохранён
    reconstructed = [line for chunk in chunks for line in chunk.splitlines()]
    assert reconstructed == lines


def test_chunk_closes_on_target_minutes_even_with_tokens_to_spare() -> None:
    """Чанк закрывается по достижении target_chunk_minutes, даже если лимит
    токенов ещё далеко не исчерпан; граница — строго на границе фразы."""
    lines = [_line("Аня" if i % 2 == 0 else "Боря", i * 5, "слово") for i in range(20)]
    transcript = "\n".join(lines)

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=100_000, target_chunk_minutes=1
    )

    assert len(chunks) > 1
    # каждая строка каждого чанка — целая исходная фраза, ничего не обрезано
    all_lines_in_chunks = [line for chunk in chunks for line in chunk.splitlines()]
    assert all_lines_in_chunks == lines


def test_hh_mm_ss_label_offset_triggers_boundary_across_hour_gap() -> None:
    """Метка `[Имя ЧЧ:MM:SS]` парсится корректно: часовой разрыв между двумя
    фразами закрывает чанк по времени при малом target_chunk_minutes."""
    transcript = "\n".join(
        [
            "[Аня 00:00] начало встречи",
            "[Аня 1:00:05] через час — подведём итоги",
        ]
    )

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=1
    )

    assert len(chunks) == 2
    assert chunks[0] == "[Аня 00:00] начало встречи"
    assert chunks[1] == "[Аня 1:00:05] через час — подведём итоги"


def test_mm_ss_label_under_hour_kept_in_one_chunk_when_within_target() -> None:
    """Метка `[Имя MM:SS]` (до часа): фразы укладываются в target_chunk_minutes
    и попадают в один чанк."""
    transcript = "\n".join(
        [
            "[Аня 00:00] первая фраза",
            "[Боря 00:30] вторая фраза",
        ]
    )

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=20
    )

    assert chunks == [transcript]


def test_monologue_exceeding_token_limit_splits_by_sentences_with_repeated_label() -> None:
    """Монолог одной фразой (40 минут) больше max_chunk_tokens: делится по
    предложениям, каждая часть — отдельный чанк с повторённой меткой спикера,
    ни одна часть не превышает лимит, предложения не разорваны."""
    sentences = [f"Пункт номер {i} с некоторыми словами тут." for i in range(30)]
    monologue_text = " ".join(sentences)
    transcript = _line("Иван", 0, monologue_text)

    chunks = chunk_transcript(transcript, _word_count, max_chunk_tokens=20, target_chunk_minutes=20)

    assert len(chunks) > 1
    for chunk in chunks:
        assert chunk.startswith("[Иван 00:00]")
        assert _word_count(chunk) <= 20

    # ни одно предложение не потеряно и не разорвано пополам
    recombined_text = " ".join(chunk.removeprefix("[Иван 00:00] ") for chunk in chunks)
    for sentence in sentences:
        assert sentence in recombined_text


def test_monologue_alone_still_produces_single_chunk_when_within_limit() -> None:
    """Одна длинная фраза, которая тем не менее укладывается в лимит токенов,
    не должна делиться по предложениям — она атомарна."""
    transcript = _line("Иван", 0, "Это не очень длинный монолог. Всего два предложения.")

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=20
    )

    assert chunks == [transcript]


def test_normal_phrases_are_never_split_across_or_within_chunks() -> None:
    """В обычном случае (без аварийного монолога) каждая исходная строка-фраза
    целиком попадает ровно в один чанк — ни разрывов внутри строки, ни
    дублирования между чанками."""
    lines = [
        _line("Аня", 0, "первая фраза без разрывов"),
        _line("Боря", 2, "вторая фраза целиком"),
        _line("Аня", 500, "третья фраза после большой паузы"),
        _line("Боря", 505, "четвёртая фраза"),
    ]
    transcript = "\n".join(lines)

    chunks = chunk_transcript(
        transcript, _word_count, max_chunk_tokens=8000, target_chunk_minutes=5
    )

    all_lines_in_chunks = [line for chunk in chunks for line in chunk.splitlines()]
    assert all_lines_in_chunks == lines
    # каждая фраза встречается ровно в одном чанке
    assert len(all_lines_in_chunks) == len(lines)