anti-plagiarism/scripts/ops/detection_benchmark.py

#!/usr/bin/env python3
"""Замер качества детекции на реальном корпусе: что находим и что выдумываем.

Зачем: до сих пор о качестве проверки мы знали только «механизм жив» — находит
подброшенный фрагмент. Этого мало. Продукт продаёт процент заимствований, и
нужно знать две вещи: сколько списываний система пропускает (полнота) и как
часто обвиняет невиновных (точность). Без этих чисел непонятно, что улучшать —
пороги, глубину корпуса или алгоритм.

Как устроен замер. Берём документы из самого корпуса и делаем из них
«студенческие работы» четырёх видов:

  дословно      — фрагмент скопирован как есть (обязан находиться);
  лёгкий рерайт — выброшено каждое 10-е слово (обязан находиться);
  сильный рерайт— выброшено каждое 3-е слово (L1 не обязан; это работа L3/L4);
  оригинал      — текст, которого в корпусе нет (обязан НЕ находиться).

Считается путь L1 — тот же, что в index.extract_and_check: фрагментация,
winnowing по каждому фрагменту, поиск по отпечаткам с порогом
EXACT_FRAGMENT_THRESHOLD. Уровни L3/L4 сюда не входят намеренно: они требуют
GPU-воркера и на порядок медленнее, а мерить нужно в первую очередь основу.

Запуск (в контейнере worker-indexer):
    cd /home/user/anti-plagiarism
    docker compose -f docker-compose.prod.yml exec -T worker-indexer python - \\
        < scripts/ops/detection_benchmark.py
    ... python - --docs 40 < scripts/ops/detection_benchmark.py

Ничего не пишет в базу — только читает.
"""

import argparse
import random


def make_cases(text: str, words_taken: int) -> dict[str, str]:
    """Сделать из текста источника четыре «студенческие работы»."""
    words = text.split()
    start = len(words) // 3
    chunk = words[start : start + words_taken]

    return {
        "дословно": " ".join(chunk),
        "лёгкий рерайт": " ".join(w for i, w in enumerate(chunk) if i % 10),
        "сильный рерайт": " ".join(w for i, w in enumerate(chunk) if i % 3),
    }


def original_text(rnd: random.Random, words_taken: int) -> str:
    """Текст, которого заведомо нет в корпусе — проверка на ложные обвинения."""
    topics = [
        "вчера вечером я варил гречку с грибами и вспоминал поездку на озеро",
        "мой сосед купил подержанный велосипед и красит его в оранжевый цвет",
        "бабушка печёт пироги с капустой по субботам и зовёт всех соседей",
        "кот запрыгнул на подоконник и уронил горшок с геранью на пол",
        "в субботу мы чинили забор и обсуждали цены на доски и гвозди",
    ]
    out: list[str] = []
    while len(out) < words_taken:
        out.extend(rnd.choice(topics).split())
    return " ".join(out[:words_taken])


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--docs", type=int, default=25, help="сколько документов взять для замера")
    ap.add_argument("--words", type=int, default=400, help="сколько слов «списывать»")
    ap.add_argument("--seed", type=int, default=20260905, help="зерно случайности (для повторяемости)")
    args = ap.parse_args()

    from app.algorithms.winnowing import winnow
    from app.config import settings
    from app.db import db_session, get_minio
    from app.fragments import split_into_fragments
    from app.models import Document, Fingerprint
    from sqlalchemy import func, select
    from sqlalchemy import text as sql_text

    rnd = random.Random(args.seed)

    # Берём документы с реальной глубиной: по аннотации в 30 отпечатков мерить
    # нечего, и такой замер сказал бы больше о корпусе, чем об алгоритме
    with db_session() as s:
        rows = s.execute(sql_text("""
            SELECT d.id, d.source, d.minio_key, d.abstract
            FROM documents d
            JOIN (SELECT doc_id, count(*) c FROM fingerprints
                  GROUP BY doc_id HAVING count(*) > 800) f ON f.doc_id = d.id
            WHERE d.source <> 'user_submission'
            ORDER BY random() LIMIT :n
        """), {"n": args.docs}).all()

    print(f"документов в замере: {len(rows)}, «списываем» по {args.words} слов")
    print(f"порог срабатывания фрагмента: {settings.EXACT_FRAGMENT_THRESHOLD}%\n")

    def find_source(work_text: str) -> set[int]:
        """Прогнать «работу» через L1 и вернуть найденные документы-источники."""
        found: set[int] = set()
        frags = split_into_fragments(
            work_text,
            window=settings.FRAGMENT_WINDOW_WORDS,
            overlap=settings.FRAGMENT_OVERLAP_WORDS,
        )
        with db_session() as s:
            for frag in frags:
                fp = winnow(frag["text"])
                if not fp:
                    continue
                hit = s.execute(
                    select(Fingerprint.doc_id, func.count(Fingerprint.id).label("c"))
                    .join(Document, Document.id == Fingerprint.doc_id)
                    .where(
                        Fingerprint.hash_value.in_(list(fp)),
                        Document.source != "user_submission",
                    )
                    .group_by(Fingerprint.doc_id)
                    .order_by(func.count(Fingerprint.id).desc())
                    .limit(1)
                ).first()
                if hit and hit[1] / len(fp) * 100 >= settings.EXACT_FRAGMENT_THRESHOLD:
                    found.add(hit[0])
        return found

    stats: dict[str, dict[str, int]] = {}
    minio = get_minio()

    for doc_id, _source, minio_key, abstract in rows:
        # Полный текст, если он сохранён; иначе то, что есть в базе
        body = abstract or ""
        if minio_key:
            try:
                obj = minio.get_object(settings.MINIO_BUCKET_DOCS, minio_key)
                body = obj.read().decode("utf-8", errors="replace")
                obj.close()
                obj.release_conn()
            except Exception:
                pass
        if len(body.split()) < args.words * 2:
            continue

        for case_name, work in make_cases(body, args.words).items():
            found = find_source(work)
            bucket = stats.setdefault(case_name, {"всего": 0, "нашли верно": 0, "не нашли": 0})
            bucket["всего"] += 1
            if doc_id in found:
                bucket["нашли верно"] += 1
            else:
                bucket["не нашли"] += 1

        # Контроль на ложные обвинения
        found = find_source(original_text(rnd, args.words))
        bucket = stats.setdefault("оригинал", {"всего": 0, "ложно обвинён": 0, "чисто": 0})
        bucket["всего"] += 1
        if found:
            bucket["ложно обвинён"] += 1
        else:
            bucket["чисто"] += 1

    print(f"{'случай':16} {'всего':>6} {'найден':>8} {'пропущен':>10}  доля найденных")
    for case in ("дословно", "лёгкий рерайт", "сильный рерайт"):
        b = stats.get(case)
        if not b:
            continue
        share = b["нашли верно"] / b["всего"] * 100 if b["всего"] else 0
        print(f"{case:16} {b['всего']:>6} {b['нашли верно']:>8} {b['не нашли']:>10}  {share:5.1f}%")

    b = stats.get("оригинал")
    if b:
        fp_rate = b["ложно обвинён"] / b["всего"] * 100 if b["всего"] else 0
        print(f"\nоригинальные тексты: {b['всего']}, ложных обвинений {b['ложно обвинён']} "
              f"({fp_rate:.1f}%)")

    print("\nЧитать так: «дословно» и «лёгкий рерайт» — обязанность L1, доля должна быть "
          "близка к 100%. «Сильный рерайт» L1 брать не обязан, это работа L3/L4. "
          "Ложные обвинения должны быть нулевыми.")


if __name__ == "__main__":
    main()