src/split_groups.py"""Сплит train/val/test ПО ГРУППАМ (листам), а не по кропам.

Зачем: кропы одного листа похожи друг на друга (перо, бумага, сканер) и часто повторяют
значения. Если часть листа попала в train, а часть в val, модель «узнаёт» лист, а не учится
читать, и оценка получается завышенной (утечка).

Результат: split.json (списки групп, один на все модели курса) и train/val/test.jsonl.
"""
import argparse
import hashlib
import json
import random
import sys
from collections import Counter
from pathlib import Path

from patterns import label_pattern

SPLITS = ("train", "val", "test")


def check_ratios(ratios: dict) -> None:
    if set(ratios) != set(SPLITS):
        raise ValueError(f"ratios должен содержать ключи {SPLITS}, получено {sorted(ratios)}")
    if any(v < 0 for v in ratios.values()):
        raise ValueError("доли не могут быть отрицательными")
    if abs(sum(ratios.values()) - 1.0) > 1e-6:
        raise ValueError(f"доли должны давать в сумме 1, получено {sum(ratios.values())}")


def split_groups(strata: dict[str, str], ratios: dict[str, float], seed: int = 42) -> dict[str, list[str]]:
    """Делит группы на train/val/test, сохраняя пропорции страт.

    strata: {group_id: страта}; ratios: {"train": .., "val": .., "test": ..}.
    Внутри каждой страты группы перемешиваются отдельным генератором
    random.Random(f"{seed}:{страта}"), затем val получает round(n * доля_val) групп,
    test — round(n * доля_test), остальное — train.
    Возвращает {"train": [...], "val": [...], "test": [...]}, списки отсортированы.
    """
    check_ratios(ratios)
    result: dict[str, list[str]] = {name: [] for name in SPLITS}
    for stratum in sorted(set(strata.values())):
        groups = sorted(g for g, s in strata.items() if s == stratum)
        random.Random(f"{seed}:{stratum}").shuffle(groups)
        n = len(groups)
        n_val = round(n * ratios["val"])
        n_test = round(n * ratios["test"])
        result["val"] += groups[:n_val]
        result["test"] += groups[n_val:n_val + n_test]
        result["train"] += groups[n_val + n_test:]
    return {name: sorted(ids) for name, ids in result.items()}


# --- Дальше всё готово: чтение данных, страты, запись файлов ------------------------

def read_jsonl(path) -> list[dict]:
    with open(path, encoding="utf-8") as f:
        return [json.loads(line) for line in f]


def group_strata(records: list[dict], rare_threshold: float = 0.10) -> dict[str, str]:
    """Страта группы = какие редкие паттерны в ней встречаются (редкий: < 10 % value-кропов)."""
    counts = Counter(label_pattern(r["text"], r["task"]) for r in records if not r["is_empty"])
    total = sum(counts.values())
    rare = {p for p, c in counts.items() if c / total < rare_threshold}
    found: dict[str, set] = {}
    for r in records:
        found.setdefault(r["group_id"], set())
        if not r["is_empty"]:
            pattern = label_pattern(r["text"], r["task"])
            if pattern in rare:
                found[r["group_id"]].add(pattern)
    return {g: "+".join(sorted(s)) or "none" for g, s in found.items()}


def split_hash(split: dict) -> str:
    """Отпечаток разбиения: по нему позже проверим, что две модели учились на одном сплите."""
    payload = json.dumps({name: split[name] for name in SPLITS}, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()[:16]


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--src", default="data/processed", help="папка с all.jsonl")
    ap.add_argument("--out", default="data/processed")
    ap.add_argument("--val-ratio", type=float, default=0.15)
    ap.add_argument("--test-ratio", type=float, default=0.15)
    ap.add_argument("--seed", type=int, default=42)
    args = ap.parse_args()

    records = read_jsonl(Path(args.src) / "all.jsonl")
    ratios = {"train": round(1 - args.val_ratio - args.test_ratio, 6),
              "val": args.val_ratio, "test": args.test_ratio}
    strata = group_strata(records)
    split = split_groups(strata, ratios, args.seed)

    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    meta = {"version": 1, "seed": args.seed, "ratios": ratios,
            "strata": dict(sorted(Counter(strata.values()).items())),
            "hash": split_hash(split), **split}
    (out / "split.json").write_text(json.dumps(meta, indent=1, ensure_ascii=False), encoding="utf-8")

    owner = {g: name for name in SPLITS for g in split[name]}
    for name in SPLITS:
        with open(out / f"{name}.jsonl", "w", encoding="utf-8") as f:
            for r in records:
                if owner[r["group_id"]] == name:
                    f.write(json.dumps(r, ensure_ascii=False) + "\n")

    print(f"split.json: hash={meta['hash']}, seed={args.seed}")
    for name in SPLITS:
        crops = [r for r in records if owner[r["group_id"]] == name]
        value = sum(not r["is_empty"] for r in crops)
        print(f"  {name:<5} групп {len(split[name]):>3}  кропов {len(crops):>4}  "
              f"(value {value}, empty {len(crops) - value})")
    print(f"страты (групп): {meta['strata']}")


if __name__ == "__main__":
    sys.exit(main())

src/verify_split.py"""Проверка разбиения: нет пересечений групп, ничего не потеряно, test достаточно велик."""
import argparse
import json
import math
import sys
from pathlib import Path

from split_groups import SPLITS, read_jsonl, split_hash

Z = 1.96           # 95 %-й доверительный интервал
TARGET_RECALL = 0.99


def min_value_crops(target: float = TARGET_RECALL, z: float = Z) -> int:
    """Сколько value-кропов нужно, чтобы при 0 ошибок нижняя граница Уилсона достигла target.

    Нижняя граница при 0 ошибок на n объектах: n / (n + z^2).
    """
    return math.ceil(target * z * z / (1 - target))


def verify(data_dir) -> tuple[list[str], list[str], list[dict]]:
    """Возвращает (ошибки, предупреждения, строки таблицы)."""
    data_dir = Path(data_dir)
    errors: list[str] = []
    warnings: list[str] = []
    all_records = read_jsonl(data_dir / "all.jsonl")
    part = {name: read_jsonl(data_dir / f"{name}.jsonl") for name in SPLITS}
    groups = {name: {r["group_id"] for r in part[name]} for name in SPLITS}

    for i, a in enumerate(SPLITS):
        for b in SPLITS[i + 1:]:
            common = groups[a] & groups[b]
            if common:
                sample = ", ".join(sorted(common)[:3])
                errors.append(f"УТЕЧКА: {len(common)} групп есть и в {a}, и в {b} (например, {sample})")

    ids = [r["id"] for name in SPLITS for r in part[name]]
    if len(ids) != len(set(ids)):
        errors.append("есть кропы, попавшие в несколько сплитов")
    missing = {r["id"] for r in all_records} - set(ids)
    if missing:
        errors.append(f"потеряно кропов: {len(missing)} (есть в all.jsonl, нет ни в одном сплите)")

    split_path = data_dir / "split.json"
    if split_path.exists():
        meta = json.loads(split_path.read_text(encoding="utf-8"))
        for name in SPLITS:
            if set(meta[name]) != groups[name]:
                errors.append(f"split.json не совпадает с {name}.jsonl")
        if meta.get("hash") != split_hash(meta):
            errors.append("hash в split.json не соответствует спискам групп")
    else:
        errors.append("нет split.json")

    table = []
    for name in SPLITS:
        value = sum(not r["is_empty"] for r in part[name])
        table.append({"split": name, "groups": len(groups[name]), "crops": len(part[name]),
                      "value": value, "empty": len(part[name]) - value})
    need = min_value_crops()
    test_value = next(row["value"] for row in table if row["split"] == "test")
    if test_value < need:
        warnings.append(f"в test {test_value} value-кропов; для проверки recall ≥ {TARGET_RECALL} "
                        f"нужно хотя бы {need} (иначе даже 0 ошибок не докажет цель)")
    return errors, warnings, table


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data-dir", default="data/processed")
    args = ap.parse_args()
    errors, warnings, table = verify(args.data_dir)

    print(f"{'split':<6}{'groups':>8}{'crops':>8}{'value':>8}{'empty':>8}")
    for row in table:
        print(f"{row['split']:<6}{row['groups']:>8}{row['crops']:>8}{row['value']:>8}{row['empty']:>8}")
    for w in warnings:
        print("WARN:", w)
    if errors:
        for e in errors:
            print("ERROR:", e)
        print("verify_split: FAIL")
        sys.exit(1)
    test_value = next(row["value"] for row in table if row["split"] == "test")
    print(f"пересечения групп: нет; test value-кропов: {test_value} (нужно ≥ {min_value_crops()})")
    print("verify_split: OK")


if __name__ == "__main__":
    main()

src/baselines.py"""Простые базовые модели: с ними сравниваем всё остальное (в уроке 7 добавятся ещё)."""
from pathlib import Path

import numpy as np

from image_to_tensor import preprocess_array


def load_features(records: list[dict], image_root, pool: int = 2) -> np.ndarray:
    """Кропы → матрица [N, D]: препроцессинг урока 2 и усреднение блоками pool×pool."""
    rows = []
    for r in records:
        arr = preprocess_array(Path(image_root) / r["image"])[0]      # [48, 160]
        if pool > 1:
            h, w = arr.shape
            arr = arr.reshape(h // pool, pool, w // pool, pool).mean(axis=(1, 3))
        rows.append(arr.reshape(-1))
    return np.stack(rows).astype(np.float32)


class OneNN:
    """Один ближайший сосед по евклидову расстоянию между пикселями."""

    def fit(self, X: np.ndarray, y) -> "OneNN":
        self.X = np.asarray(X, dtype=np.float32)
        self.y = list(y)
        self._sq = (self.X ** 2).sum(axis=1)
        return self

    def nearest(self, Q: np.ndarray, chunk: int = 256) -> np.ndarray:
        """Индекс ближайшего обучающего объекта для каждой строки Q."""
        Q = np.asarray(Q, dtype=np.float32)
        out = []
        for i in range(0, len(Q), chunk):
            part = Q[i:i + chunk]
            # |q - x|^2 = |q|^2 - 2 q·x + |x|^2; |q|^2 для argmin не нужен
            dist = self._sq[None, :] - 2.0 * part @ self.X.T
            out.append(dist.argmin(axis=1))
        return np.concatenate(out)

    def predict(self, Q: np.ndarray) -> list:
        return [self.y[i] for i in self.nearest(Q)]

experiments/leakage_demo.py"""Демо утечки: одна и та же модель (1-NN по пикселям), два способа сделать val.

LEAKY — val выбран случайно среди кропов (листы перемешаны между train и val).
SHEET — val из split.json: листы val модель не видела.
Размеры train и val в обоих случаях одинаковы, меняется только способ деления.
"""
import argparse
import json
import random
import sys
from collections import Counter
from pathlib import Path

import numpy as np

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from baselines import OneNN, load_features  # noqa: E402
from split_groups import read_jsonl  # noqa: E402


def evaluate(train: list[dict], val: list[dict], image_root, pool: int) -> dict:
    X_train = load_features(train, image_root, pool)
    X_val = load_features(val, image_root, pool)
    model = OneNN().fit(X_train, [r["text"] for r in train])
    idx = model.nearest(X_val)
    pred = [model.y[i] for i in idx]
    exact = float(np.mean([p == r["text"] for p, r in zip(pred, val)]))
    same_sheet = float(np.mean([train[i]["group_id"] == r["group_id"] for i, r in zip(idx, val)]))
    seen = {r["text"] for r in train}
    ceiling = float(np.mean([r["text"] in seen for r in val]))
    return {"exact": exact, "same_sheet": same_sheet, "ceiling": ceiling}


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--data-dir", default="data/processed")
    ap.add_argument("--seed", type=int, default=0)
    ap.add_argument("--pool", type=int, default=2, help="усреднение пикселей блоками (ускоряет)")
    args = ap.parse_args()

    data_dir = Path(args.data_dir)
    manifest = json.loads((data_dir / "manifest.json").read_text(encoding="utf-8"))
    image_root = manifest["image_root"]
    train = [r for r in read_jsonl(data_dir / "train.jsonl") if not r["is_empty"]]
    val = [r for r in read_jsonl(data_dir / "val.jsonl") if not r["is_empty"]]

    # честный вариант: val из split.json
    honest = evaluate(train, val, image_root, args.pool)

    # утечка: тот же пул кропов, но val выбран случайно и того же размера
    pool_records = train + val
    order = list(range(len(pool_records)))
    random.Random(args.seed).shuffle(order)
    leaky_val = [pool_records[i] for i in order[:len(val)]]
    leaky_train = [pool_records[i] for i in order[len(val):]]
    leaky = evaluate(leaky_train, leaky_val, image_root, args.pool)

    top = Counter(r["text"] for r in train).most_common(1)[0]
    chance = float(np.mean([r["text"] == top[0] for r in val]))
    print(f"train {len(train)} кропов, val {len(val)} кропов (только value)")
    print(f"наивный ответ «всегда {top[0]!r}»: exact {chance:.3f}\n")
    print(f"{'':<8}{'exact':>8}{'сосед с того же листа':>26}{'потолок*':>11}")
    print(f"{'LEAKY':<8}{leaky['exact']:>8.3f}{leaky['same_sheet']:>26.3f}{leaky['ceiling']:>11.3f}")
    print(f"{'SHEET':<8}{honest['exact']:>8.3f}{honest['same_sheet']:>26.3f}{honest['ceiling']:>11.3f}")
    print("* потолок: доля val-кропов, значение которых вообще встречается в train")


if __name__ == "__main__":
    main()

experiments/make_leaky_split.py"""Намеренно портит разбиение: делит по КРОПАМ, а не по группам (для проверки verify_split)."""
import argparse
import json
import random
import shutil
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from split_groups import SPLITS, read_jsonl, split_hash  # noqa: E402


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--src", default="data/processed")
    ap.add_argument("--out", default="data/processed-leaky")
    ap.add_argument("--seed", type=int, default=0)
    args = ap.parse_args()

    src, out = Path(args.src), Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    shutil.copy(src / "manifest.json", out / "manifest.json")
    shutil.copy(src / "all.jsonl", out / "all.jsonl")

    records = read_jsonl(src / "all.jsonl")
    random.Random(args.seed).shuffle(records)                 # перемешиваем кропы, а не листы
    n = len(records)
    cuts = {"train": records[: int(n * 0.70)],
            "val": records[int(n * 0.70): int(n * 0.85)],
            "test": records[int(n * 0.85):]}
    meta = {"version": 1, "seed": args.seed}
    for name in SPLITS:
        with open(out / f"{name}.jsonl", "w", encoding="utf-8") as f:
            for r in cuts[name]:
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
        meta[name] = sorted({r["group_id"] for r in cuts[name]})
    meta["hash"] = split_hash(meta)
    (out / "split.json").write_text(json.dumps(meta, indent=1, ensure_ascii=False), encoding="utf-8")
    print(f"испорченный сплит записан в {out}")


if __name__ == "__main__":
    main()

tests/test_split_groups.pyimport random
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from split_groups import split_groups  # noqa: E402

RATIOS = {"train": 0.70, "val": 0.15, "test": 0.15}


def make_strata(n_rare: int = 100, n_common: int = 150) -> dict[str, str]:
    strata = {f"g{i:03d}": "rare" for i in range(n_rare)}
    strata.update({f"g{n_rare + i:03d}": "common" for i in range(n_common)})
    return strata


def test_partition_no_overlap_nothing_lost():
    strata = make_strata()
    result = split_groups(strata, RATIOS, seed=42)
    assert set(result) == {"train", "val", "test"}
    all_ids = result["train"] + result["val"] + result["test"]
    assert len(all_ids) == len(set(all_ids)), "группа попала в несколько сплитов"
    assert set(all_ids) == set(strata), "часть групп потеряна или появилась лишняя"


def test_deterministic_and_independent_of_input_order():
    strata = make_strata(30, 40)
    shuffled_items = list(strata.items())
    random.Random(1).shuffle(shuffled_items)
    shuffled = dict(shuffled_items)
    assert split_groups(strata, RATIOS, seed=7) == split_groups(strata, RATIOS, seed=7)
    assert split_groups(strata, RATIOS, seed=7) == split_groups(shuffled, RATIOS, seed=7)


def test_different_seed_gives_different_split():
    strata = make_strata()
    assert split_groups(strata, RATIOS, seed=1) != split_groups(strata, RATIOS, seed=2)


def test_overall_proportions():
    result = split_groups(make_strata(), RATIOS, seed=42)   # 250 групп
    assert abs(len(result["val"]) - 37.5) <= 3
    assert abs(len(result["test"]) - 37.5) <= 3
    assert abs(len(result["train"]) - 175) <= 6


def test_stratified_each_stratum_keeps_proportions():
    strata = make_strata(100, 150)
    result = split_groups(strata, RATIOS, seed=42)
    for stratum, n in (("rare", 100), ("common", 150)):
        for name, ratio in RATIOS.items():
            count = sum(1 for g in result[name] if strata[g] == stratum)
            assert abs(count - n * ratio) <= 1, f"{stratum}/{name}: {count} вместо ≈{n * ratio:.1f}"


def test_tiny_stratum_goes_to_train():
    strata = make_strata(1, 20)       # в страте rare одна группа: round(0.15) = 0
    result = split_groups(strata, RATIOS, seed=42)
    assert "g000" in result["train"]


def test_lists_are_sorted():
    result = split_groups(make_strata(), RATIOS, seed=42)
    for ids in result.values():
        assert ids == sorted(ids)


def test_zero_test_ratio_is_allowed():
    ratios = {"train": 0.85, "val": 0.15, "test": 0.0}
    result = split_groups(make_strata(), ratios, seed=42)
    assert result["test"] == []
    assert len(result["train"]) + len(result["val"]) == 250


def test_bad_ratios_raise():
    strata = make_strata(5, 5)
    with pytest.raises(ValueError):
        split_groups(strata, {"train": 0.5, "val": 0.2, "test": 0.2}, seed=1)   # сумма ≠ 1
    with pytest.raises(ValueError):
        split_groups(strata, {"train": 0.8, "val": 0.2}, seed=1)                # нет test
    with pytest.raises(ValueError):
        split_groups(strata, {"train": 1.2, "val": -0.1, "test": -0.1}, seed=1) # отрицательные

tests/test_baselines.pyimport sys
import tempfile
from pathlib import Path

import numpy as np
from PIL import Image

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from baselines import OneNN, load_features  # noqa: E402


def test_one_nn_picks_nearest_point():
    X = np.array([[0.0, 0.0], [10.0, 10.0], [0.0, 10.0]], dtype=np.float32)
    model = OneNN().fit(X, ["a", "b", "c"])
    Q = np.array([[1.0, 1.0], [9.0, 9.0], [1.0, 9.0]], dtype=np.float32)
    assert model.predict(Q) == ["a", "b", "c"]


def test_one_nn_is_exact_on_training_points():
    rng = np.random.default_rng(0)
    X = rng.normal(size=(50, 8)).astype(np.float32)
    labels = [str(i) for i in range(50)]
    assert OneNN().fit(X, labels).predict(X) == labels


def test_chunking_does_not_change_result():
    rng = np.random.default_rng(1)
    X = rng.normal(size=(40, 6)).astype(np.float32)
    Q = rng.normal(size=(30, 6)).astype(np.float32)
    model = OneNN().fit(X, range(40))
    assert (model.nearest(Q, chunk=7) == model.nearest(Q, chunk=1000)).all()


def test_load_features_shape():
    with tempfile.TemporaryDirectory() as d:
        Image.new("L", (100, 40), 230).save(Path(d) / "a.png")
        records = [{"image": "a.png"}, {"image": "a.png"}]
        assert load_features(records, d, pool=2).shape == (2, 24 * 80)
        assert load_features(records, d, pool=1).shape == (2, 48 * 160)

tests/test_verify_split.pyimport json
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from split_groups import split_hash  # noqa: E402
from verify_split import min_value_crops, verify  # noqa: E402


def make_records() -> list[dict]:
    records = []
    for g in range(6):
        for n in range(4):
            empty = n % 2 == 1
            records.append({"id": f"g{g}_{n}", "image": f"g{g}/{n}.png", "text": "" if empty else "7,3",
                            "group_id": f"g{g}", "task": "measurement", "is_empty": empty})
    return records


def write_dir(d, parts: dict, split_lists: dict | None = None, tamper_hash: bool = False) -> None:
    d = Path(d)
    records = make_records()
    write = lambda name, rows: (d / name).write_text(
        "".join(json.dumps(r, ensure_ascii=False) + "\n" for r in rows), encoding="utf-8")
    write("all.jsonl", records)
    for name, groups in parts.items():
        write(f"{name}.jsonl", [r for r in records if r["group_id"] in groups])
    lists = split_lists or {name: sorted(groups) for name, groups in parts.items()}
    meta = dict(lists)
    meta["hash"] = "bad" if tamper_hash else split_hash(meta)
    (d / "split.json").write_text(json.dumps(meta), encoding="utf-8")


GOOD = {"train": {"g0", "g1", "g2", "g3"}, "val": {"g4"}, "test": {"g5"}}


def test_min_value_crops_matches_wilson_arithmetic():
    n = min_value_crops()
    assert n == 381
    assert n / (n + 1.96 ** 2) >= 0.99 > (n - 1) / (n - 1 + 1.96 ** 2)


def test_good_split_has_no_errors():
    with tempfile.TemporaryDirectory() as d:
        write_dir(d, GOOD)
        errors, warnings, table = verify(d)
        assert errors == []
        assert warnings, "test слишком мал: ожидали предупреждение"
        assert [row["split"] for row in table] == ["train", "val", "test"]


def test_group_in_two_splits_is_reported_as_leak():
    parts = {"train": {"g0", "g1", "g2", "g3"}, "val": {"g3", "g4"}, "test": {"g5"}}
    with tempfile.TemporaryDirectory() as d:
        write_dir(d, parts)
        errors, _, _ = verify(d)
        assert any("УТЕЧКА" in e and "train" in e and "val" in e for e in errors), errors


def test_lost_crops_are_reported():
    parts = {"train": {"g0", "g1", "g2"}, "val": {"g4"}, "test": {"g5"}}      # g3 нигде нет
    with tempfile.TemporaryDirectory() as d:
        write_dir(d, parts)
        errors, _, _ = verify(d)
        assert any("потеряно" in e for e in errors), errors


def test_tampered_split_json_is_reported():
    with tempfile.TemporaryDirectory() as d:
        write_dir(d, GOOD, tamper_hash=True)
        errors, _, _ = verify(d)
        assert any("hash" in e for e in errors), errors
    with tempfile.TemporaryDirectory() as d:
        lists = {"train": ["g0", "g1", "g2"], "val": ["g3", "g4"], "test": ["g5"]}   # не совпадает с jsonl
        write_dir(d, GOOD, split_lists=lists)
        errors, _, _ = verify(d)
        assert any("не совпадает" in e for e in errors), errors