"""Розыгрыш, который может пересчитать любой.

Этот файл публикуется вместе с реестром до эфира. Зависимостей нет, нужен
только Python 3.10 или новее.

Проверить розыгрыш - одна команда:

    python draw.py registry.csv --round 34441012 --annulled annulled.csv

Скрипт сам возьмёт значение раунда у маяка drand с нескольких зеркал, сверит
их ответы между собой и посчитает победителей. Номер раунда для момента
времени:

    python draw.py --when 2026-11-15T21:00+07:00

Как это устроено

1. Реестр - CSV с колонками code, participant, name, phone: код участника,
   обезличенный номер участника, маска имени и последние 4 цифры телефона.
   Номер участника нужен, чтобы исключать остальные коды победителя, не
   раскрывая телефон. Маска и 4 цифры - чтобы человек узнал свой билет:
   по одним цифрам не узнать, они совпадают у разных людей. Имя необязательно,
   у не назвавшихся колонка пустая.
2. Аннулированные коды - отдельный CSV с колонками code, participant, reason.
   В розыгрыше они не участвуют, но вместе с реестром покрывают все номера
   подряд с первого. Если номер пропал или появился в обоих файлах, скрипт
   остановится: тихо убрать код из середины нельзя.
   Отрезанные последние номера нумерация не выдаст - их ловит последний
   номер, опубликованный заранее, и участник, не нашедший свои коды.
3. До эфира публикуются оба файла и их SHA-256. Изменить список после этого
   незаметно нельзя: любая правка меняет хеш.
4. Seed - значение randomness раунда сети drand quicknet. Номер раунда
   объявлен в правилах заранее, а значение появляется только в момент раунда:
   его не знает никто, включая участников сети drand и магазин.
5. Коды сортируются по номеру. Для места k позиция победителя - это
   HMAC-SHA256(ключ: seed, сообщение: "<хеш реестра>:<k>") как число по модулю
   числа оставшихся кодов. Все коды победителя выбывают, дальше - место k+1.
   Так получается очередь участников. Какой приз кому достанется, очередь не
   решает: в эфире призы раздаются по ней по порядку, и каждый уходит первому,
   кто вышел на связь. Не дозвонились - тот же приз переходит следующему.

Смещение от взятия по модулю при реестре до миллиона строк - меньше 2^-230,
на результат оно не влияет.

Что скрипт проверяет у маяка

- Все ответившие зеркала отдали одно и то же значение, и ответили хотя бы два.
  Среди зеркал - серверы самой сети drand и Cloudflare.
- randomness - это SHA-256 от подписи раунда, как и устроен drand.

BLS-подпись раунда стандартной библиотекой Python не проверить, поэтому здесь
она не проверяется. Её проверяет страница эфира: браузер сверяет подпись
с постоянным публичным ключом сети quicknet, и там этого достаточно, кто бы
ни принёс значение.

Если из вашей сети зеркала drand не открываются - так в России без VPN, -
возьмите значение раунда на странице эфира, где его подпись уже проверена
браузером, и передайте скрипту через --seed. Арифметику он пересчитает
без всякой сети.
"""
from __future__ import annotations

import argparse
import csv
import hashlib
import hmac
import io
import json
import math
import re
import sys
import urllib.error
import urllib.request
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

HEADER = ("code", "participant", "name", "phone")
ANNULLED_HEADER = ("code", "participant", "reason")
CODE = re.compile(r"^RF-(\d{6,})$")

#: Сеть drand quicknet: раунд каждые 3 секунды, каждый раунд проверяется сам
#: по себе, без цепочки предыдущих.
QUICKNET = "52db9ba70e0cc0f6eaf7803dd07447a1f5477735fd3f661792ba94600c84e971"
GENESIS = 1692803367
PERIOD = 3
MIRRORS = (
    "https://api.drand.sh",
    "https://api2.drand.sh",
    "https://api3.drand.sh",
    "https://drand.cloudflare.com",
)
MIN_MIRRORS = 2
#: Новокузнецк живёт по UTC+7 круглый год.
SHOP_TZ = timezone(timedelta(hours=7), "Новокузнецк")

_HEX = re.compile(r"^[0-9a-f]+$")

#: Ответ зеркала: код HTTP и тело.
Fetch = Callable[[str], tuple[int, bytes]]


class RegistryError(ValueError):
    """Файл реестра не в том формате или номера кодов не сходятся."""


class BeaconError(RuntimeError):
    """Значение маяка получить нельзя или ему нельзя верить."""


class BeaconNotReady(BeaconError):
    """Раунд ещё не наступил."""


@dataclass(frozen=True)
class Entry:
    code: str
    participant: str
    #: Маска имени: «Д***л». Пусто, если участник не назвался.
    name: str
    #: Последние 4 цифры телефона.
    phone: str


@dataclass(frozen=True)
class Annulled:
    code: str
    participant: str
    reason: str


@dataclass(frozen=True)
class Pick:
    place: int
    code: str
    participant: str
    #: Как победитель выглядит для всех: маска имени и 4 цифры телефона.
    name: str
    phone: str
    #: Сколько кодов было в игре и какая позиция выпала - для протокола.
    pool_size: int
    index: int


@dataclass(frozen=True)
class Beacon:
    round: int
    randomness: str
    signature: str
    #: Зеркала, которые ответили одинаково.
    mirrors: tuple[str, ...]


# --- Реестр ---


def _csv_bytes(header: tuple[str, ...], rows: list[tuple[str, ...]]) -> bytes:
    out = io.StringIO()
    writer = csv.writer(out, lineterminator="\n")
    writer.writerow(header)
    writer.writerows(rows)
    return out.getvalue().encode("utf-8")


def _csv_rows(data: bytes, header: tuple[str, ...], what: str) -> list[list[str]]:
    rows = list(csv.reader(io.StringIO(data.decode("utf-8"))))
    if not rows or tuple(rows[0]) != header:
        raise RegistryError(f"первая строка {what} должна быть: {','.join(header)}")

    result = []
    seen = set()
    for number, row in enumerate(rows[1:], start=2):
        if not row:
            continue
        if len(row) != len(header):
            raise RegistryError(f"{what}, строка {number}: ожидается {len(header)} колонки")
        if row[0] in seen:
            raise RegistryError(f"{what}, строка {number}: код {row[0]} повторяется")
        seen.add(row[0])
        result.append(row)
    return result


def registry_bytes(entries: list[Entry]) -> bytes:
    """Канонический файл реестра: сортировка по коду, LF, UTF-8.

    Хешируется ровно то, что публикуется. Пересохранение в Excel меняет
    перевод строк и хеш - поэтому файл отдаётся на скачивание как есть.
    """
    ordered = sorted(entries, key=lambda e: e.code)
    return _csv_bytes(HEADER, [(e.code, e.participant, e.name, e.phone) for e in ordered])


def annulled_bytes(entries: list[Annulled]) -> bytes:
    """Канонический файл аннулированных кодов - по тем же правилам, что реестр."""
    ordered = sorted(entries, key=lambda e: e.code)
    return _csv_bytes(ANNULLED_HEADER, [(e.code, e.participant, e.reason) for e in ordered])


def registry_sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def parse_registry(data: bytes) -> list[Entry]:
    return [Entry(*row) for row in _csv_rows(data, HEADER, "реестра")]


def parse_annulled(data: bytes) -> list[Annulled]:
    return [Annulled(*row) for row in _csv_rows(data, ANNULLED_HEADER, "списка аннулированных")]


def code_number(code: str) -> int:
    found = CODE.match(code)
    if not found:
        raise RegistryError(f"код {code!r} не в формате RF-000001")
    return int(found.group(1))


def check_numbering(active: list[Entry], annulled: list[Annulled]) -> int:
    """Номера действующих и аннулированных кодов вместе идут подряд с первого.

    Возвращает последний номер. Номер аннулированного кода не выдаётся
    повторно, поэтому пропуск в нумерации значит, что код убрали молча.
    """
    live = {code_number(e.code) for e in active}
    dead = {code_number(a.code) for a in annulled}

    both = sorted(live & dead)
    if both:
        raise RegistryError(f"код RF-{both[0]:06d} есть и в реестре, и среди аннулированных")

    numbers = live | dead
    if not numbers:
        return 0
    last = max(numbers)
    missing = sorted(set(range(1, last + 1)) - numbers)
    if missing:
        shown = ", ".join(f"RF-{n:06d}" for n in missing[:5])
        more = f" и ещё {len(missing) - 5}" if len(missing) > 5 else ""
        raise RegistryError(f"в нумерации пропущены коды: {shown}{more}")
    return last


# --- Маяк drand ---


def round_time(number: int) -> datetime:
    """Момент выхода раунда, UTC."""
    return datetime.fromtimestamp(GENESIS + (number - 1) * PERIOD, timezone.utc)


def round_at(moment: datetime) -> int:
    """Первый раунд, который выйдет не раньше `moment`."""
    if moment.tzinfo is None:
        raise ValueError("у момента времени должен быть часовой пояс, например +07:00")
    return max(math.ceil((moment.timestamp() - GENESIS) / PERIOD), 0) + 1


def _http_get(url: str) -> tuple[int, bytes]:
    request = urllib.request.Request(
        url, headers={"User-Agent": "rusflow-draw/1", "Accept": "application/json"}
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return response.status, response.read(64_000)
    except urllib.error.HTTPError as e:
        return e.code, e.read(4_000)


def _beacon_from(body: bytes, number: int, mirror: str) -> tuple[str, str]:
    try:
        data = json.loads(body)
        got = int(data["round"])
        randomness = str(data["randomness"]).lower()
        signature = str(data["signature"]).lower()
    except (ValueError, KeyError, TypeError) as e:
        raise BeaconError(f"{mirror}: ответ не разобрать") from e

    if got != number:
        raise BeaconError(f"{mirror}: вместо раунда {number} пришёл раунд {got}")
    if not (_HEX.match(randomness) and len(randomness) == 64
            and _HEX.match(signature) and len(signature) == 96):
        raise BeaconError(f"{mirror}: значения раунда не в формате сети quicknet")
    if hashlib.sha256(bytes.fromhex(signature)).hexdigest() != randomness:
        raise BeaconError(f"{mirror}: randomness не равен SHA-256 подписи - ответу верить нельзя")
    return randomness, signature


def relay_beacon(
    number: int,
    *,
    mirrors: tuple[str, ...] = MIRRORS,
    get: Fetch = _http_get,
) -> Beacon:
    """Значение раунда от первого зеркала, которое ответило.

    Это не то же самое, что fetch_beacon: там значение нужно нам самим, и мы
    требуем согласия нескольких зеркал. Здесь значение просто передаётся
    дальше, зрителю, а он проверяет подпись сам - и одного источника довольно.
    Спрашивать все четыре по очереди ради этого незачем: раунд живёт три
    секунды, а ожидание видно на странице.
    """
    not_ready = False
    problems: list[str] = []
    for mirror in mirrors:
        try:
            status, body = get(f"{mirror}/{QUICKNET}/public/{number}")
        except OSError as e:
            problems.append(f"{mirror} ({e})")
            continue
        if status == 425:
            not_ready = True
            continue
        if status != 200:
            problems.append(f"{mirror} (HTTP {status})")
            continue
        randomness, signature = _beacon_from(body, number, mirror)
        return Beacon(number, randomness, signature, (mirror,))

    if not_ready:
        at = round_time(number)
        raise BeaconNotReady(
            f"раунд {number} ещё не наступил: он выйдет {at:%Y-%m-%d %H:%M:%S} UTC"
        )
    raise BeaconError("ни одно зеркало не ответило: " + ("; ".join(problems) or "нет"))


def fetch_beacon(
    number: int,
    *,
    mirrors: tuple[str, ...] = MIRRORS,
    min_mirrors: int = MIN_MIRRORS,
    get: Fetch = _http_get,
) -> Beacon:
    """Значение раунда quicknet, одинаковое на всех ответивших зеркалах.

    Зеркало, до которого не достучаться, пропускается. Зеркало, отдавшее
    неверное значение, останавливает проверку: это уже не сбой сети.
    """
    answers: dict[str, tuple[str, str]] = {}
    unreachable: list[str] = []
    not_ready = False

    for mirror in mirrors:
        try:
            status, body = get(f"{mirror}/{QUICKNET}/public/{number}")
        except OSError as e:
            unreachable.append(f"{mirror} ({e})")
            continue
        if status == 425:
            not_ready = True
            continue
        if status != 200:
            unreachable.append(f"{mirror} (HTTP {status})")
            continue
        answers[mirror] = _beacon_from(body, number, mirror)

    if not answers and not_ready:
        at = round_time(number)
        raise BeaconNotReady(
            f"раунд {number} ещё не наступил: он выйдет {at:%Y-%m-%d %H:%M:%S} UTC "
            f"({at.astimezone(SHOP_TZ):%H:%M:%S} по Новокузнецку)"
        )
    if len(set(answers.values())) > 1:
        detail = "; ".join(f"{m}: {r[:16]}…" for m, (r, _) in answers.items())
        raise BeaconError(f"зеркала отдали разные значения раунда {number}: {detail}")
    if len(answers) < min_mirrors:
        problems = "; ".join(unreachable) or "нет"
        raise BeaconError(
            f"значение раунда подтвердили зеркал: {len(answers)}, нужно не меньше "
            f"{min_mirrors}. Недоступны: {problems}"
        )

    randomness, signature = next(iter(answers.values()))
    return Beacon(number, randomness, signature, tuple(answers))


# --- Розыгрыш ---


def draw(entries: list[Entry], registry_hash: str, seed: str, places: int) -> list[Pick]:
    """Выбирает `places` мест. Один участник - одно место."""
    pool = sorted(entries, key=lambda e: e.code)
    key = seed.strip().encode("utf-8")
    picks: list[Pick] = []

    for place in range(1, places + 1):
        if not pool:
            break
        message = f"{registry_hash}:{place}".encode("utf-8")
        digest = hmac.new(key, message, hashlib.sha256).digest()
        index = int.from_bytes(digest, "big") % len(pool)
        winner = pool[index]
        picks.append(Pick(place, winner.code, winner.participant,
                          winner.name, winner.phone, len(pool), index))
        pool = [e for e in pool if e.participant != winner.participant]

    return picks


def _moment(raw: str) -> datetime:
    text = raw.strip()
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    return datetime.fromisoformat(text)


def main(argv: list[str] | None = None, *, get: Fetch = _http_get) -> int:
    parser = argparse.ArgumentParser(description="Проверяемый розыгрыш по опубликованному реестру")
    parser.add_argument("registry", nargs="?", help="файл реестра CSV, как он опубликован")
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--round", type=int, help="номер раунда drand quicknet из правил")
    source.add_argument("--seed", help="значение randomness, если оно уже известно")
    parser.add_argument("--annulled", help="файл аннулированных кодов CSV - проверить нумерацию")
    parser.add_argument("--when", help="узнать номер раунда для момента, например 2026-11-15T21:00+07:00")
    parser.add_argument("--prizes", type=int, default=1)
    parser.add_argument("--places", type=int, default=0,
                        help="сколько мест очереди показать; 0 - призы и ещё пятеро следом")
    parser.add_argument("--min-mirrors", type=int, default=MIN_MIRRORS)
    args = parser.parse_args(argv)

    try:
        if args.when:
            number = round_at(_moment(args.when))
            at = round_time(number)
            print(f"Раунд quicknet для {args.when}: {number}")
            print(f"Выйдет {at:%Y-%m-%d %H:%M:%S} UTC ({at.astimezone(SHOP_TZ):%Y-%m-%d %H:%M:%S} по Новокузнецку)")
            if not args.registry:
                return 0

        if not args.registry:
            parser.error("укажите файл реестра")
        if args.round is None and args.seed is None:
            parser.error("укажите --round с номером раунда из правил или --seed")

        with open(args.registry, "rb") as f:
            data = f.read()
        digest = registry_sha256(data)
        entries = parse_registry(data)
        print(f"SHA-256 реестра: {digest}")
        print(f"Кодов в реестре: {len(entries)}")

        if args.annulled:
            with open(args.annulled, "rb") as f:
                dead_data = f.read()
            dead = parse_annulled(dead_data)
            last = check_numbering(entries, dead)
            print(f"SHA-256 аннулированных: {registry_sha256(dead_data)}")
            print(f"Нумерация: RF-000001…RF-{last:06d} без пропусков "
                  f"(действующих {len(entries)}, аннулированных {len(dead)})")

        if args.round is not None:
            beacon = fetch_beacon(args.round, min_mirrors=args.min_mirrors, get=get)
            at = round_time(beacon.round)
            print(f"\nРаунд drand quicknet {beacon.round}, "
                  f"{at:%Y-%m-%d %H:%M:%S} UTC ({at.astimezone(SHOP_TZ):%H:%M} по Новокузнецку)")
            print(f"Совпало на зеркалах: {', '.join(beacon.mirrors)}")
            print(f"Подпись: {beacon.signature}")
            seed = beacon.randomness
        else:
            seed = args.seed.strip()
        print(f"Seed: {seed}\n")

        # Очередь, а не призы: какой приз кому достался, решается в эфире.
        # Приз уходит первому, кто вышел на связь, дальше по очереди.
        places = args.places or args.prizes + 5
        print(f"Очередь победителей, первые {places}:\n")
        for pick in draw(entries, digest, seed, places):
            who = " ".join(filter(None, (pick.name, f"…{pick.phone}")))
            print(f"{pick.place:>4}: {pick.code}  участник {pick.participant}  {who}  "
                  f"(позиция {pick.index} из {pick.pool_size})")
        return 0
    except (RegistryError, BeaconError, ValueError, OSError) as e:
        print(f"Ошибка: {e}", file=sys.stderr)
        return 1


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