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>
This commit is contained in:
56
services/worker-indexer/app/fragments.py
Normal file
56
services/worker-indexer/app/fragments.py
Normal file
@@ -0,0 +1,56 @@
|
||||
"""Разбиение текста документа на фрагменты для проверки плагиата — чистая логика.
|
||||
|
||||
Скользящее окно по словам с перекрытием. Вынесено из 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
|
||||
Reference in New Issue
Block a user