anti-plagiarism/services/worker-gpu/app/scoring.py

"""Чистая доменная логика скоринга плагиата — без Celery/БД/сети.

Объединяет совпадения всех уровней (1 winnowing, 2 minhash, 3+4 semantic+llm),
дедуплицирует их, размечает корректно процитированные фрагменты и считает
итоговые проценты, которые видит студент. Вынесено из Celery-задачи, чтобы
логику можно было тестировать изолированно.
"""

import re
from typing import Any

# Кавычки (открывающая/закрывающая) — рус./англ. варианты
_OPEN_QUOTES = '«"„'
_CLOSE_QUOTES = '»"“”'

# Скобочная конструкция с 4-значным годом внутри — [Иванов, 2023], (Smith, 2020),
# [Иванов и др., 2023]. Совпадает и с нашим же форматом ГОСТ 7.0.5.
_CITATION_RE = re.compile(r"[\[(][^\[\]()]{0,80}(?:19|20)\d{2}[^\[\]()]{0,20}[\])]")

# Насколько далеко после фрагмента искать ссылку на источник
_CITATION_LOOKAHEAD = 150


def is_cited(full_text: str, position_start: int, position_end: int) -> bool:
    """Эвристика: похож ли фрагмент на корректно процитированный, а не на голый плагиат.

    Фрагмент считается процитированным, если:
    - обрамлён кавычками («…», "…") сразу по границам, ИЛИ
    - сразу за ним (в пределах ~150 символов) идёт скобочная ссылка с годом —
      [Иванов, 2023], (Smith, 2020) и т.п.

    Эвристика, не 100%-но точная — сигнал для сортировки/отчёта, не финальный
    вердикт. Для фрагментов без надёжной позиции в тексте (например, уровень 2
    MinHash — совпадение на уровне всего документа) возвращает False: их нельзя
    локализовать, чтобы проверить окружение.

    Args:
        full_text: полный текст проверяемого документа
        position_start: начало фрагмента (символ)
        position_end: конец фрагмента (символ)

    Returns:
        True, если похоже на корректную цитату
    """
    if not full_text or position_start < 0 or position_end > len(full_text):
        return False

    before = full_text[max(0, position_start - 3) : position_start]
    after = full_text[position_end : position_end + 3]
    if any(c in before for c in _OPEN_QUOTES) and any(c in after for c in _CLOSE_QUOTES):
        return True

    tail = full_text[position_end : position_end + _CITATION_LOOKAHEAD]
    return bool(_CITATION_RE.search(tail))


def aggregate_results(
    level1_matches: list[dict[str, Any]],
    level2_matches: list[dict[str, Any]],
    semantic_matches: list[dict[str, Any]],
    total_fragments: int,
    full_text: str = "",
) -> dict[str, Any]:
    """Свести совпадения уровней в итог проверки.

    Дедупликация — по паре (source_title, position_start): одно и то же совпадение
    источника в одной позиции не дублируется. `overall_similarity` — доля УНИКАЛЬНЫХ
    помеченных позиций от всех фрагментов (фрагмент, совпавший с несколькими
    источниками, не раздувает процент выше 100). `uncited_similarity` — та же доля,
    но БЕЗ фрагментов, похожих на корректную цитату (см. is_cited) — это ближе к
    тому, что коммерческие системы называют «% некорректных заимствований».

    Args:
        level1_matches: совпадения уровня 1 (Winnowing, точные)
        level2_matches: совпадения уровня 2 (MinHash, нечёткие)
        semantic_matches: совпадения уровней 3-4 (FAISS + LLM-парафраз)
        total_fragments: всего проверенных фрагментов документа
        full_text: полный текст проверяемого документа — нужен для is_cited;
            пустая строка → ни один фрагмент не размечается как цитата

    Returns:
        dict с полями overall_similarity, uncited_similarity, matches (с полем
        "cited" в каждом), total_fragments, flagged_fragments, cited_fragments,
        uncited_fragments, by_method.
    """
    all_matches = level1_matches + level2_matches + semantic_matches

    seen: set[str] = set()
    unique_matches: list[dict[str, Any]] = []
    for m in all_matches:
        key = f"{m.get('source_title', '')}:{m.get('position_start', '')}"
        if key not in seen:
            seen.add(key)
            m = {**m, "cited": is_cited(full_text, m.get("position_start", -1), m.get("position_end", -1))}
            unique_matches.append(m)

    flagged_positions = {m.get("position_start") for m in unique_matches}
    uncited_positions = {m.get("position_start") for m in unique_matches if not m["cited"]}

    def _pct(positions: set) -> float:
        pct = (len(positions) / total_fragments * 100) if total_fragments > 0 else 0.0
        return round(min(pct, 100.0), 2)

    return {
        "overall_similarity": _pct(flagged_positions),
        "uncited_similarity": _pct(uncited_positions),
        "matches": unique_matches,
        "total_fragments": total_fragments,
        "flagged_fragments": len(flagged_positions),
        "cited_fragments": len(flagged_positions) - len(uncited_positions),
        "uncited_fragments": len(uncited_positions),
        "by_method": {
            "exact": len(level1_matches),
            "fuzzy": len(level2_matches),
            "semantic_llm": len(semantic_matches),
        },
    }