anti-plagiarism/services/worker-indexer/app/algorithms/winnowing.py

"""Алгоритм Winnowing для fingerprinting текстов.

Winnowing — алгоритм выбора минимального хэша в скользящем окне.
Используется для нахождения точных и частичных совпадений текстов.
"""

import xxhash


def get_ngrams(tokens: list[str], k: int = 5) -> list[str]:
    """
    Сформировать n-граммы из токенов.

    Args:
        tokens: Список слов
        k: Размер n-граммы

    Returns:
        Список n-грамм в виде строк
    """
    if len(tokens) < k:
        return []
    return [" ".join(tokens[i : i + k]) for i in range(len(tokens) - k + 1)]


def hash_ngram(ngram: str) -> int:
    """
    Хэшировать n-грамму через xxHash (быстро, детерминированно).

    Args:
        ngram: n-грамма для хэширования

    Returns:
        Знаковый 64-битный хэш (умещается в PostgreSQL BIGINT)
    """
    # xxh64 даёт unsigned 64-bit (до 2^64-1), а PostgreSQL BIGINT — signed
    # (макс 2^63-1). Приводим к диапазону signed int64.
    unsigned = xxhash.xxh64(ngram.encode("utf-8")).intdigest()
    return unsigned - (1 << 64) if unsigned >= (1 << 63) else unsigned


def winnow(text: str, k: int = 5, window: int = 4) -> set[int]:
    """
    Алгоритм Winnowing для построения fingerprint документа.

    Шаги:
    1. Токенизация (lowercase)
    2. Построение k-грамм
    3. Хэширование k-грамм
    4. Скользящее окно — выбор минимального хэша в каждой позиции

    Args:
        text: Исходный текст
        k: Размер k-граммы (n-gram)
        window: Размер скользящего окна

    Returns:
        Множество отобранных хэшей (fingerprint)
    """
    tokens = text.lower().split()

    if len(tokens) < k:
        return set()

    ngrams = get_ngrams(tokens, k)
    hashes = [hash_ngram(ng) for ng in ngrams]

    if not hashes:
        return set()

    # Скользящее окно — выбираем минимальный хэш в каждом окне
    fingerprint: set[int] = set()
    prev_min_idx = -1

    for i in range(len(hashes) - window + 1):
        window_hashes = hashes[i : i + window]
        min_val = min(window_hashes)
        min_idx = i + window_hashes.index(min_val)

        # Добавляем только если это новый минимум или позиция изменилась
        if min_idx != prev_min_idx:
            fingerprint.add(min_val)
            prev_min_idx = min_idx

    return fingerprint


def winnow_ordered(text: str, k: int = 5, window: int = 4) -> list[int]:
    """То же, что winnow, но с сохранением порядка появления отпечатков в тексте.

    Нужно там, где отпечатки приходится обрезать по лимиту: множество не хранит
    порядка, и `list(winnow(text))[:limit]` берёт произвольное подмножество —
    у длинного документа целые куски остаются без покрытия, и списывание из них
    не находится. Со списком в порядке текста обрезку можно делать равномерной
    (см. sample_evenly).

    Args:
        text: Исходный текст
        k: Размер k-граммы
        window: Размер скользящего окна

    Returns:
        Отпечатки в порядке появления в тексте, без повторов
    """
    tokens = text.lower().split()
    if len(tokens) < k:
        return []

    hashes = [hash_ngram(ng) for ng in get_ngrams(tokens, k)]
    if not hashes:
        return []

    ordered: list[int] = []
    seen: set[int] = set()
    prev_min_idx = -1

    for i in range(len(hashes) - window + 1):
        window_hashes = hashes[i : i + window]
        min_val = min(window_hashes)
        min_idx = i + window_hashes.index(min_val)

        if min_idx != prev_min_idx:
            if min_val not in seen:
                seen.add(min_val)
                ordered.append(min_val)
            prev_min_idx = min_idx

    return ordered


def sample_evenly(items: list[int], limit: int) -> list[int]:
    """Оставить не больше limit элементов, равномерно по всей длине списка.

    Берём каждый n-й элемент, а не первые limit штук: обрезка «с начала»
    оставила бы без отпечатков весь конец документа.

    Args:
        items: Отпечатки в порядке текста (winnow_ordered)
        limit: Максимум отпечатков; 0 или меньше — не ограничивать

    Returns:
        Подсписок длиной не больше limit, сохраняющий порядок
    """
    if limit <= 0 or len(items) <= limit:
        return items
    step = len(items) / limit
    return [items[int(i * step)] for i in range(limit)]


def jaccard_similarity(fp_a: set[int], fp_b: set[int]) -> float:
    """
    Коэффициент Жаккара для двух fingerprint'ов.

    Args:
        fp_a: Fingerprint документа A
        fp_b: Fingerprint документа B

    Returns:
        Коэффициент сходства от 0.0 до 1.0
    """
    if not fp_a or not fp_b:
        return 0.0

    intersection = len(fp_a & fp_b)
    union = len(fp_a | fp_b)

    return intersection / union if union > 0 else 0.0


def compute_similarity(text_a: str, text_b: str, k: int = 5, window: int = 4) -> float:
    """
    Вычислить схожесть двух текстов через Winnowing.

    Args:
        text_a: Первый текст
        text_b: Второй текст
        k: Размер k-граммы
        window: Размер окна Winnowing

    Returns:
        Схожесть от 0.0 до 1.0
    """
    fp_a = winnow(text_a, k=k, window=window)
    fp_b = winnow(text_b, k=k, window=window)
    return jaccard_similarity(fp_a, fp_b)