用Python实现一个300行代码的区块链CLI工具

项目地址:https://github.com/simpleapples/py-blockchain-cli

最近区块链比较火,我也趁机学习了一丢丢,顺手撸了一个区块链 CLI

可以以 CLI 的方式开节点,连接节点,挖矿(节点间会自动同步)

并且可以以表格形式看到整条区块链的信息,如图

欢迎 star&fork !


用Python实现一个300行代码的区块链CLI工具

8 回复

不错 你这视频用什么软件做的?


#!/usr/bin/env python3
"""
一个简单的区块链CLI工具 - 300行以内实现
支持:创建区块链、挖矿、查看链、验证完整性
"""

import hashlib
import json
import time
from typing import List, Dict, Any
import argparse
import sys

class Block:
    """区块类"""
    def __init__(self, index: int, timestamp: float, transactions: List[Dict], previous_hash: str):
        self.index = index
        self.timestamp = timestamp
        self.transactions = transactions
        self.previous_hash = previous_hash
        self.nonce = 0
        self.hash = self.calculate_hash()
    
    def calculate_hash(self) -> str:
        """计算区块哈希值"""
        block_string = json.dumps({
            "index": self.index,
            "timestamp": self.timestamp,
            "transactions": self.transactions,
            "previous_hash": self.previous_hash,
            "nonce": self.nonce
        }, sort_keys=True).encode()
        return hashlib.sha256(block_string).hexdigest()
    
    def mine_block(self, difficulty: int):
        """工作量证明挖矿"""
        target = "0" * difficulty
        while self.hash[:difficulty] != target:
            self.nonce += 1
            self.hash = self.calculate_hash()
    
    def to_dict(self) -> Dict:
        """转换为字典格式"""
        return {
            "index": self.index,
            "timestamp": self.timestamp,
            "transactions": self.transactions,
            "previous_hash": self.previous_hash,
            "hash": self.hash,
            "nonce": self.nonce
        }

class Blockchain:
    """区块链类"""
    def __init__(self, difficulty: int = 2):
        self.chain: List[Block] = []
        self.difficulty = difficulty
        self.pending_transactions: List[Dict] = []
        self.create_genesis_block()
    
    def create_genesis_block(self):
        """创建创世区块"""
        genesis_block = Block(0, time.time(), [{"from": "genesis", "to": "genesis", "amount": 0}], "0")
        genesis_block.mine_block(self.difficulty)
        self.chain.append(genesis_block)
    
    def get_latest_block(self) -> Block:
        """获取最新区块"""
        return self.chain[-1]
    
    def add_transaction(self, sender: str, recipient: str, amount: float):
        """添加交易到待处理池"""
        self.pending_transactions.append({
            "sender": sender,
            "recipient": recipient,
            "amount": amount,
            "timestamp": time.time()
        })
        return len(self.chain) + 1
    
    def mine_pending_transactions(self, miner_address: str):
        """挖矿并添加新区块"""
        if not self.pending_transactions:
            return False
        
        # 添加矿工奖励交易
        self.pending_transactions.append({
            "sender": "network",
            "recipient": miner_address,
            "amount": 1.0,
            "timestamp": time.time()
        })
        
        block = Block(
            len(self.chain),
            time.time(),
            self.pending_transactions,
            self.get_latest_block().hash
        )
        block.mine_block(self.difficulty)
        
        self.chain.append(block)
        self.pending_transactions = []
        return True
    
    def is_chain_valid(self) -> bool:
        """验证区块链完整性"""
        for i in range(1, len(self.chain)):
            current = self.chain[i]
            previous = self.chain[i-1]
            
            # 验证当前区块哈希
            if current.hash != current.calculate_hash():
                return False
            
            # 验证与前一个区块的连接
            if current.previous_hash != previous.hash:
                return False
            
            # 验证工作量证明
            if current.hash[:self.difficulty] != "0" * self.difficulty:
                return False
        
        return True
    
    def print_chain(self):
        """打印整个区块链"""
        for block in self.chain:
            print(f"\n区块 #{block.index}:")
            print(f"  哈希: {block.hash}")
            print(f"  前哈希: {block.previous_hash}")
            print(f"  时间戳: {time.ctime(block.timestamp)}")
            print(f"  随机数: {block.nonce}")
            print(f"  交易数: {len(block.transactions)}")
            for tx in block.transactions:
                print(f"    {tx['sender']} -> {tx['recipient']}: {tx['amount']}")

def main():
    parser = argparse.ArgumentParser(description="区块链CLI工具")
    subparsers = parser.add_subparsers(dest="command", help="可用命令")
    
    # 创建新区块链
    subparsers.add_parser("new", help="创建新区块链")
    
    # 添加交易
    tx_parser = subparsers.add_parser("add", help="添加交易")
    tx_parser.add_argument("sender", help="发送方地址")
    tx_parser.add_argument("recipient", help="接收方地址")
    tx_parser.add_argument("amount", type=float, help="交易金额")
    
    # 挖矿
    mine_parser = subparsers.add_parser("mine", help="挖矿")
    mine_parser.add_argument("miner", help="矿工地址")
    
    # 查看区块链
    subparsers.add_parser("view", help="查看区块链")
    
    # 验证区块链
    subparsers.add_parser("validate", help="验证区块链完整性")
    
    args = parser.parse_args()
    
    # 初始化区块链(简单起见使用内存存储)
    blockchain = Blockchain(difficulty=2)
    
    if args.command == "new":
        blockchain = Blockchain()
        print("✅ 新区块链创建成功!")
    
    elif args.command == "add":
        blockchain.add_transaction(args.sender, args.recipient, args.amount)
        print(f"✅ 交易已添加:{args.sender} -> {args.recipient}: {args.amount}")
    
    elif args.command == "mine":
        if blockchain.mine_pending_transactions(args.miner):
            print(f"✅ 挖矿成功!矿工 {args.miner} 获得1个币奖励")
            print(f"   新区块 #{len(blockchain.chain)-1} 已添加到链上")
        else:
            print("❌ 没有待处理的交易")
    
    elif args.command == "view":
        blockchain.print_chain()
        print(f"\n📊 区块链统计:")
        print(f"   区块数: {len(blockchain.chain)}")
        print(f"   待处理交易: {len(blockchain.pending_transactions)}")
        print(f"   难度: {blockchain.difficulty}")
    
    elif args.command == "validate":
        if blockchain.is_chain_valid():
            print("✅ 区块链验证通过!")
        else:
            print("❌ 区块链验证失败!")
    
    else:
        parser.print_help()

if __name__ == "__main__":
    main()

这个实现包含了区块链的核心功能:创世区块创建、交易添加、工作量证明挖矿、链完整性验证。代码结构清晰,完全在300行以内。

使用示例:

# 创建新区块链
python blockchain.py new

# 添加交易
python blockchain.py add Alice Bob 5.0

# 挖矿
python blockchain.py mine Miner1

# 查看区块链
python blockchain.py view

# 验证区块链
python blockchain.py validate

核心特点:

  1. 完整的区块链数据结构(Block + Blockchain类)
  2. SHA256哈希和工作量证明机制
  3. 交易池和矿工奖励系统
  4. 链完整性验证
  5. 简洁的CLI界面

代码可以直接运行,不需要额外依赖。

$ python main.py
Traceback (most recent call last):
File “main.py”, line 1, in <module>
from cli.interface import Interface
File “e:\py-blockchain-cli\cli<a target=”_blank" href=“http://interface.py” rel=“nofollow noopener”>interface.py", line 3, in <modul
e>
from cli.command import Command
File “e:\py-blockchain-cli\cli<a target=”_blank" href=“http://command.py” rel=“nofollow noopener”>command.py", line 13
print(f’Peer running at {host}:{port}’)
^
SyntaxError: invalid syntax

需要用 python3 运行哈

就是 py3

f’'语法是 3.6 的

不好意思 忘记说明是 3.6 了

回到顶部