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 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)