""" Проверка оплаты на стороне продавца. Зачем отдельный файл. Кнопка оплаты на сайте возвращает хэш перевода из браузера, и верить ему нельзя: страницу можно поменять в отладчике и вызвать обработчик успеха, не заплатив ни монеты. Единственный надёжный способ это спросить ноду со своего сервера. Здесь ровно это и делается, со всеми проверками, которые легко забыть. Что проверяется и почему каждая проверка нужна. получатель совпадает с вашим адресом иначе засчитается перевод, отправленный кому-то другому сумма не меньше счёта иначе оплатят рубль вместо тысячи перевод подтверждён, а не висит в очереди из очереди перевод можно вытеснить другим с тем же номером, и товар уйдёт за деньги, которых не будет этот хэш ещё не засчитывался иначе одним переводом оплатят десять заказов Последнюю проверку эта библиотека сделать за вас не может: она не знает, что вы уже засчитали. Храните использованные хэши у себя и проверяйте перед выдачей товара. Функция check возвращает хэш, чтобы вам было что сохранить. Зависимостей нет, только стандартная библиотека. """ import json import urllib.error import urllib.request from dataclasses import dataclass DEFAULT_NODE = "https://nxt.stackworks.ru" UNITS_PER_NXT = 1_000_000 TIMEOUT = 15 class PaymentError(Exception): """Оплата не подтвердилась. Текст объясняет, чем именно.""" @dataclass class Payment: hash: str sender: str recipient: str amount: int # в базовых единицах, 1 NXT = 1 000 000 fee: int block_height: int confirmations: int @property def amount_nxt(self) -> float: """Для показа человеку. В расчётах пользуйтесь amount, там целые.""" return self.amount / UNITS_PER_NXT def to_base_units(nxt) -> int: """ NXT в базовые единицы без потерь. Через строку, а не через float: 0.1 в двоичной дроби не представим точно, и умножение на миллион даёт 99999.99999999999. На деньгах это недопустимо. """ text = str(nxt).strip().replace(",", ".") if "." not in text: return int(text) * UNITS_PER_NXT whole, frac = text.split(".", 1) if len(frac) > 6: raise ValueError("не больше шести знаков после запятой") return int(whole or 0) * UNITS_PER_NXT + int(frac.ljust(6, "0")) def fetch_tx(tx_hash: str, node: str = DEFAULT_NODE) -> dict: """Спросить ноду о переводе. Бросает PaymentError, если её не слышно.""" if not tx_hash or len(tx_hash) != 64: raise PaymentError("хэш перевода это 64 знака") url = f"{node.rstrip('/')}/tx/{tx_hash}" try: with urllib.request.urlopen(url, timeout=TIMEOUT) as resp: return json.loads(resp.read()) except urllib.error.HTTPError as e: if e.code == 404: raise PaymentError("такого перевода нода не знает") from e raise PaymentError(f"нода ответила {e.code}") from e except (urllib.error.URLError, ValueError) as e: raise PaymentError(f"нода недоступна: {e}") from e def check(tx_hash: str, recipient: str, amount, *, node: str = DEFAULT_NODE, min_confirmations: int = 1) -> Payment: """ Проверить оплату. Возвращает Payment или бросает PaymentError. amount принимает и строку «28.5», и целое число базовых единиц. Строку разбираем точно, без float. min_confirmations. Единица означает «перевод попал в блок». Блок выходит раз в шесть секунд, поэтому для мелких сумм этого достаточно. Для крупных разумно подождать нескольких блоков. """ want = amount if isinstance(amount, int) else to_base_units(amount) tx = fetch_tx(tx_hash, node) got_recipient = tx.get("recipient") if got_recipient != recipient: raise PaymentError( f"перевод ушёл не вам: получатель {got_recipient}, а ждали {recipient}") got_amount = int(tx.get("amount", 0)) if got_amount < want: raise PaymentError( f"оплачено {got_amount / UNITS_PER_NXT} NXT, а счёт на {want / UNITS_PER_NXT} NXT") if not tx.get("confirmed"): where = "в очереди" if tx.get("in_mempool") else "неизвестно где" raise PaymentError( f"перевод ещё не подтверждён, он {where}. Из очереди его можно вытеснить, " "поэтому засчитывать оплату рано") confirmations = int(tx.get("confirmations", 0)) if confirmations < min_confirmations: raise PaymentError( f"подтверждений {confirmations}, а нужно {min_confirmations}") return Payment( hash=tx_hash, sender=tx.get("sender_address", ""), recipient=got_recipient, amount=got_amount, fee=int(tx.get("fee", 0)), block_height=int(tx.get("block_height") or 0), confirmations=confirmations, ) def is_paid(tx_hash: str, recipient: str, amount, **kw) -> bool: """То же самое, но без подробностей. Удобно в условии if.""" try: check(tx_hash, recipient, amount, **kw) return True except PaymentError: return False def balance(address: str, node: str = DEFAULT_NODE) -> int: """Баланс адреса в базовых единицах.""" url = f"{node.rstrip('/')}/balance/{address}" try: with urllib.request.urlopen(url, timeout=TIMEOUT) as resp: return int(json.loads(resp.read())["balance"]) except (urllib.error.URLError, ValueError, KeyError) as e: raise PaymentError(f"не удалось узнать баланс: {e}") from e def incoming(address: str, *, limit: int = 50, node: str = DEFAULT_NODE) -> list: """ Входящие переводы на адрес, свежие первыми. Нужно тем, кто принимает оплату на один общий адрес и различает заказы по сумме. Отдаём только входящие: свои же исходящие в этом списке лишние и однажды кто-нибудь засчитает по ним оплату. """ url = f"{node.rstrip('/')}/address/{address.strip().lower()}/txs?limit={int(limit)}" try: with urllib.request.urlopen(url, timeout=TIMEOUT) as resp: data = json.loads(resp.read()) except (urllib.error.URLError, ValueError) as e: raise PaymentError(f"не удалось получить историю адреса: {e}") from e rows = data.get("txs") if isinstance(data, dict) else data want = address.strip().lower() return [t for t in (rows or []) if (t.get("recipient") or "").lower() == want] def find_payment(address: str, amount, *, seen: set = None, limit: int = 50, node: str = DEFAULT_NODE): """ Найти оплату на точную сумму. Возвращает Payment или None. amount, как и в check: целое число это БАЗОВЫЕ ЕДИНИЦЫ, строка «1000» это монеты. То есть find_payment(addr, 1000) ищет одну тысячную монеты, а не тысячу монет. Я сам на этом споткнулся при первой же проверке, поэтому пишу прямо здесь: если сумма у вас в монетах, передавайте строкой либо умножайте на 1 000 000. Так работает приём на один адрес: к сумме заказа добавляется уникальный хвост, и по нему платёж находится однозначно. Memo в переводах NextCoin нет, поэтому иначе заказы не различить. Сумму сверяем на точное совпадение, а не «не меньше». При поиске по сумме это принципиально: перевод на большую сумму принадлежит другому заказу, и засчитать его сюда значит обокрасть того, кто заплатил. seen это множество уже засчитанных хэшей. Передавать его обязательно, иначе один перевод закроет несколько заказов. Хранить его должны вы, у себя в базе: библиотека вашей базы не знает. Переводы из очереди пропускаем. Их можно вытеснить другим с тем же номером операции, и товар уйдёт за деньги, которых не будет. """ want = amount if isinstance(amount, int) else to_base_units(amount) seen = seen or set() for tx in incoming(address, limit=limit, node=node): if tx.get("hash") in seen: continue if int(tx.get("amount", 0)) != want: continue if not tx.get("confirmed", not tx.get("pending", False)): continue return Payment( hash=tx.get("hash", ""), sender=tx.get("sender_address", ""), recipient=tx.get("recipient", ""), amount=int(tx.get("amount", 0)), fee=int(tx.get("fee", 0)), block_height=int(tx.get("block_height") or 0), confirmations=int(tx.get("confirmations", 1)), ) return None if __name__ == "__main__": import sys if len(sys.argv) != 4: print("Проверка оплаты NextCoin.") print(f" {sys.argv[0]} <хэш> <ваш адрес> <сумма NXT>") raise SystemExit(2) try: p = check(sys.argv[1], sys.argv[2], sys.argv[3]) print(f"Оплачено. {p.amount_nxt} NXT от {p.sender}") print(f"Блок {p.block_height}, подтверждений {p.confirmations}") print("Не забудьте запомнить хэш, чтобы не засчитать его второй раз.") except PaymentError as e: print("Не оплачено:", e) raise SystemExit(1)