使用 Tron(TRX) 收款

今年年初的时候，我写过如何用 Stripe 收款。但是 Stripe 的注册门槛确实要高一些，这一点就足够劝退很多人了。

退而求其次，用加密货币来收款也是可以的。区块链上所有的交易都是公开的，因此我们可以通过这一点来验证用户的付款情况。

想要验证用户付款，包括但不限于如下方式：

• 每笔交易都生成一对密钥，对应一个付款地址。密钥管理会非常麻烦
• 有些交易可以添加一个描述字段，用于区分。像极了给对公账户转账时的OCR/Reference
• 尾数标记，略微改变付款金额，比如同样需要付款1 USDT，可以一个用户需要付款1.000001另外一个需要付款1.000002

前几天了解到 HD 钱包（高清钱包 分层确定性钱包 Hierarchical Deterministic Wallet），发现这种技术非常有趣。它允许通过单个种子源生成一整套密钥对（即公钥和私钥），像树形结构一样。有一个根，就可以计算出树上任意一个节点的公私钥。

那这就刚刚好，每一笔交易我都给树上增加一个叶子节点，只要有根和路径，就可以计算出这个节点的公私钥。

这里我选择 Tron的TRX，主要原因还是转账手续费比较低😂

钱包路径

从树根走到某一个叶子，肯定要一个特定的路径的。对于 tron来说，一个典型的路径是这样的

m/44'/195'/0'/0/0

• m/44表示BIP44标准
• 195表示 TRON，每种加密货币都分配有一个唯一的编号
• 0'表示钱包的第一个账户，创建多个账户那就是1‘ 2’
• 0 链接层级 一般0表示接收 1表示找零
• 0 特定地址索引，改变这个数字就会生成新的地址

由于我们想每一笔交易都生成不同的地址，那么我们可以这样设想

• m/44'/195'/0'/0/0 这个路径用于保存所有的余额
• m/44'/195'/1'/0/(0-10000) 每一笔交易请求都通过1钱包的不同索引生成不同地址
• 每一笔交易确认后，都计算出这个地址的私钥，然后把钱转到0账户中

库

Python 中有很多库可以处理加密货币。比如 python-hdwallet 实现了一百多种货币的HD钱包；我们本次的目的非常简单，只使用 TRON，那么就用 tronpy 好了

生成助记词

from tronpy.hdwallet import generate_mnemonic
m = generate_mnemonic(12, "english")
print(m)

连接区块链网络

TRON 有如下几个网络：

• mainnet 主网，里面的钱是具有经济价值的，意味着可以兑换法币
• shasta、nile 测试网，里面的钱没经济价值，完全测试用

由于不同国家和地区对于加密货币的法律政策限制不同，同时我并不想真正的去花钱，那么测试网就是再好不过的了。

from tronpy import Tron

MNEMONIC = "slogan reduce clap umbrella liquid crumble outer exchange quiz promote owner buffalo"
client = Tron(network="nile")
addr = client.generate_address_from_mnemonic(MNEMONIC, "", "m/44'/195'/0'/0/0")
print(addr["base58check_address"])

输出如下

TERbSCnpFp4FG6SLpUHpFRBRT69ZhHbLFp

你看我们这里使用了 account path，这个0'/0/0就是我们最终集资的钱包

为每笔交易生成收款地址

我们使用 1'账户，然后需要在数据库中记录下索引，每笔交易都要生成不同的索引，自增就可以了，要记住这个索引，否则就没办法生成私钥了

addr = client.generate_address_from_mnemonic(MNEMONIC, "", "m/44'/195'/1'/0/1")
print(addr["base58check_address"])

输出如下，这个地址就是用户要打钱的地方

TD2g6siz3NP4wcjP5iAueW9tottaAgP98a

模拟用户

模拟用户付钱，那么就要先弄到一个有数字货币的钱包地址。可是这又是测试网，没办法真正兑换法币的。不慌，有网站可以免费领用于测试的币，如： https://nileex.io/join/getJoinPage

同样生成一个地址，记住地址和私钥，然后页面提交就好了。

为了方便实用，我们可以封装一个简单的命令行小程序

付款

client.trx.transfer(from_, to, amount).build().inspect().sign(priv).broadcast()
print(t.wait())

amount是金额，1_000_000 是 1 TRX

priv 是私钥，如果通过助记词和路径生成key，可以使用如下代码

def mnemonic_to_private_key():
    seed = seed_from_mnemonic(mnemonic, passphrase="")
    private_key = key_from_seed(seed, account_path="m/44'/195'/0'/0/0")
    return PrivateKey(private_key)

验证付款

简单的轮询即可

# 获取地址信息
client.get_account("TD2g6siz3NP4wcjP5iAueW9tottaAgP98a")
获取地址余额，返回值类型是 Decimal 避免出现 0.1 + 0.2 = 0.30000000000000004 的悲剧
client.get_account_balance("TD2g6siz3NP4wcjP5iAueW9tottaAgP98a")

验证成功后转钱到集中地址

和付款一样，把 To 改成自己的地址就好。如果你想也可以弄成交易所给的地址、或者本地钱包软件给的地址。

在转账失败时，应该把金额减少 1.1 TRX用于付矿工费

使用 Telegram Bot处理

理解了以上所有步骤之后，简单封装一下就可以了，下面是一组示例代码，使用 apscheduler 轮询获取付款状态。简单起见没有引入数据库，正常要用起来的话，还是要把一些关键的信息，如付款地址、路径等写入到数据库中的。

import logging

import telebot
from apscheduler.schedulers.background import BackgroundScheduler
from telebot.types import Message
from tronpy import Tron
from tronpy.exceptions import TransactionError, ValidationError
from tronpy.hdwallet import key_from_seed, seed_from_mnemonic
from tronpy.keys import PrivateKey

logging.basicConfig(level=logging.INFO)
bot = telebot.TeleBot("12345")

index = 0
# address: {user_id: 123, path: "m/44'/195'/1'/0/0"
unpaid_map = {}

mnemonic = "slogan reduce clap umbrella liquid crumble outer exchange quiz promote owner buffalo"


class TronTrx:
    def __init__(self):
        self.client = Tron(network="nile")
        self.central = "m/44'/195'/0'/0/0"

    def mnemonic_to_private_key(self, path):
        seed = seed_from_mnemonic(mnemonic, passphrase="")
        private_key = key_from_seed(seed, account_path=path)
        return PrivateKey(private_key)

    def central_wallet(self):
        return self.client.generate_address_from_mnemonic(mnemonic, "", self.central)["base58check_address"]

    def generate_address(self, user_id):
        global index
        path = f"m/44'/195'/1'/0/{index}"
        logging.info("Generating address for user %s with path %s", user_id, path)
        addr = self.client.generate_address_from_mnemonic(mnemonic, "", path)["base58check_address"]
        unpaid_map[addr] = {"user_id": user_id, "path": path}
        index += 1
        return addr

    def transfer(self, from_: str, to: str, amount: int):
        logging.info("Transfer %s TRX from %s to %s", amount, from_, to)
        from_path = unpaid_map[from_]["path"]
        try:
            (
                self.client.trx.transfer(from_, to, amount)
                .build()
                .sign(self.mnemonic_to_private_key(from_path))
                .broadcast()
            )
        except (TransactionError, ValidationError):
            logging.warning("Balance not enough, try to transfer %s TRX", amount - 1_100_000)
            (
                self.client.trx.transfer(from_, to, amount - 1_100_000)
                .build()
                .sign(self.mnemonic_to_private_key(from_path))
                .broadcast()
            )

    def verify_payment(self):
        logging.info("Checking %s unpaid payment", len(unpaid_map))
        for addr, meta in unpaid_map.items():
            try:
                balance = self.client.get_account_balance(addr)
            except:
                balance = 0
            if balance:
                logging.info("addr %s has %s TRX", addr, balance)
                self.transfer(addr, self.central_wallet(), int(balance * 1_000_000))
                bot.send_message(meta["user_id"], f"You have paid {balance} TRX")
                del unpaid_map.copy()[addr]


trx = TronTrx()


@bot.message_handler(commands=["pay"])
def send_address(message: Message):
    bot.send_message(message.chat.id, trx.generate_address(message.chat.id))


@bot.message_handler(commands=["start"])
def send_start(message: Message):
    bot.send_message(message.chat.id, "hello")


if __name__ == "__main__":
    scheduler = BackgroundScheduler()
    scheduler.add_job(trx.verify_payment, "interval", seconds=10, max_instances=1)
    scheduler.start()
    bot.infinity_polling()