Files
anti-plagiarism/services/worker-indexer/app/fragments.py
jze9 b2d2061a83 refactor(indexer): вынести разбиение на фрагменты в app.fragments + 5 тестов
_split_into_fragments жила в тяжёлой Celery-задаче (celery/sqlalchemy/minio) и
не тестировалась, хотя именно она определяет, ЧТО проверяется на плагиат.
Вынес в чистый app.fragments.split_into_fragments — тестируется изолированно:

- пустой/пробельный текст → [];
- текст короче 20 слов → отбрасывается порогом длины;
- короткий текст → один фрагмент на весь объём, start=0;
- скользящее окно: перекрытие ровно (window-step) слов, хвост==голова соседа;
- start/end — корректные символьные офсеты в исходный текст.

Поведение сохранено 1:1. Добавлен в mypy-гейт. Тестов всего: 66.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-08-11 20:17:33 +05:00

57 lines
2.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Разбиение текста документа на фрагменты для проверки плагиата — чистая логика.
Скользящее окно по словам с перекрытием. Вынесено из Celery-задачи, чтобы
границы фрагментов (от них зависит, что именно проверяется на плагиат) можно
было тестировать изолированно.
"""
from typing import Any
def split_into_fragments(
text: str,
window: int = 200,
overlap: int = 50,
) -> list[dict[str, Any]]:
"""Разбить текст на фрагменты скользящим окном с перекрытием.
Args:
text: Исходный текст
window: Размер окна в словах
overlap: Перекрытие между соседними фрагментами в словах
Returns:
Список словарей {"text": str, "start": int, "end": int}, где start/end —
позиции в символах. Фрагменты короче 20 слов отбрасываются.
"""
words = text.split()
if not words:
return []
fragments: list[dict[str, Any]] = []
step = window - overlap
# Позиция первого символа каждого слова (слова разделены одним пробелом)
char_positions = []
pos = 0
for word in words:
char_positions.append(pos)
pos += len(word) + 1 # +1 для пробела
for i in range(0, max(1, len(words) - window + 1), step):
chunk_words = words[i : i + window]
if len(chunk_words) < 20: # слишком короткий фрагмент — пропустить
continue
start_char = char_positions[i]
end_idx = min(i + window - 1, len(words) - 1)
end_char = char_positions[end_idx] + len(words[end_idx])
fragments.append({
"text": " ".join(chunk_words),
"start": start_char,
"end": end_char,
})
return fragments