用Python实现一个300行代码的区块链CLI工具
项目地址:https://github.com/simpleapples/py-blockchain-cli
最近区块链比较火,我也趁机学习了一丢丢,顺手撸了一个区块链 CLI
可以以 CLI 的方式开节点,连接节点,挖矿(节点间会自动同步)
并且可以以表格形式看到整条区块链的信息,如图

欢迎 star&fork !
用Python实现一个300行代码的区块链CLI工具
不错 你这视频用什么软件做的?
#!/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
核心特点:
- 完整的区块链数据结构(Block + Blockchain类)
- SHA256哈希和工作量证明机制
- 交易池和矿工奖励系统
- 链完整性验证
- 简洁的CLI界面
代码可以直接运行,不需要额外依赖。
LiceCap
$ 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 了

