Python实现区块链核心逻辑：从交易记录到区块及链式结构开发

周予安

发布时间：2026-10-13 | 4人浏览过

原创

本文详细讲解使用Python从零构建区块链数据层的过程。首先定义包含交易集合、时间戳及哈希值的区块类，通过密封和链接机制确保数据完整性；随后扩展为区块链类，实现多区块的有序连接与整体校验。文中包含完整的Python代码示例及篡改测试，展示如何验证交易和区块的有效性，为后续共识机制开发奠定基础。

区块开发基础与结构定义

承接前文对交易类的整合，本节重点阐述区块的具体开发流程。

构建一个有效的区块，核心要素包括交易记录的集合、生成时间戳、当前区块的哈希值以及前一个区块的哈希引用。明确这些结构后，即可进行相应的代码实现。


import datetime
import hashlib
from Message import DaDaMessage, InvalidMessage
from Transaction import Transaction
class Block:
  #不定参数，*args,区块集合
  def __init__(self, *args):
    self.messagelist = [] # 存储多个交易记录
    self.timestamp = None # 存储多个记录最终锁定时间
    self.hash = None
    self.preHash = None
    if args:
      for arg in args:
        self.add_message(arg)

  def add_message(self, message): # 添加 交易信息
    # 区分第一条与后面多条，是否需要链接
    if len(self.messagelist) > 0:
      message.link(self.messagelist[-1]) # 链接最后一个
    message.seal() # 密封
    message.validate() # 校验
    self.messagelist.append(message)
  def link(self, block): # 区块链接
    self.preHash = block.hash
  def seal(self): # 密封当前hash
    self.timestamp = datetime.datetime.now()
    self.hash = self.hash_block()
  def hash_block(self): # 密封上一块哈希，时间戳，交易记录的最后一个

    return hashlib.sha512(
      (str(self.timestamp) + str(self.preHash) + str(self.messagelist[-1].hash)).encode("utf-8")).hexdigest()
  def validate(self): # 校验
    for i, message in enumerate(self.messagelist): # 每个交易记录校验一下
      message.validate() #每一条交易记录校验
      if i > 0 and message.prev_hash != self.messagelist[i - 1].hash:
        raise InvalidBlock("无效的block,交易记录被修改为在第{}条记录".format(i)+str(self))
    return str(self) + "Ok"
  def __repr__(self):
    return "money block= hash:{},prehash:{},len:{},time:{}".format(self.hash, self.preHash, len(self.messagelist),
                                    self.timestamp)

首先定义自定义异常类，用于处理区块验证失败的情况：


class InvalidBlock(Exception):
  def __init__(self,*args,**kwargs):
    super(Exception,self).__init__(*args,**kwargs)

接着编写测试模块，验证区块的基本功能：


if __name__=="__main__":
  t1 = Transaction("yicheng", "ddd1", 100)
  t2 = Transaction("yicheng", "ddd2", 200)
  t3 = Transaction("yicheng", "ddd3", 300)
  t4 = Transaction("yicheng", "ddd4", 400)
  m1 = DaDaMessage(t1)
  m2 = DaDaMessage(t2)
  m3 = DaDaMessage(t3)
  m4 = DaDaMessage(t4)
  try:
    block1 = Block(m1, m2, m3)
    block1.seal()
    #测试篡改数据
    #m3.data = "你妹的直播"
    #block1.messagelist[0] = m3
    print(block1.validate())
  except InvalidMessage as e: #交易记录被修改
    print(e)
  except InvalidBlock as e: #区块被修改
    print(e)

运行测试结果如下。为了便于观察，此处输出格式调整为MD5哈希值形式：

若尝试篡改区块信息，系统会抛出相应错误。具体报错内容取决于被修改的数据部分，结果可能有所不同：

区块链链式结构与完整实现

至此，交易记录与单个区块的开发已完成，在此基础上实现区块链结构将更为顺畅。具体实现代码如下：


from Block import InvalidBlock, Block
from Message import InvalidMessage, DaDaMessage
from Transaction import Transaction
# 区块链
class Dada_BlockCoin:
  def __init__(self):
    self.blocklist = [] # 装载所有区块
  def validate(self):#校验所有区块
    for i, block in enumerate(self.blocklist):
      try:
        block.validate()
      except InvalidBlockCoin as e:
        raise InvalidBlockCoin("区块校验错误，区块索引{}".format(i))
  def add_block(self, block): # 增加区块
    if len(self.blocklist) > 0:
      block.link(self.blocklist[-1]) #连接区块
    block.seal()#密封
    block.validate()#校验
    self.blocklist.append(block)#添加到区块链中
  def __repr__(self):
    return "Dada_BlockCoin:{}".format(len(self.blocklist))

定义区块链层面的自定义异常：


class InvalidBlockCoin(Exception):
  def __init__(self, *args, **kwargs):
    super(Exception, self).__init__(*args, **kwargs)

编写区块链的测试模块，验证多区块连接与校验逻辑：


if __name__ == "__main__":
  t1 = Transaction("yicheng", "ddd1", 100)
  t2 = Transaction("yicheng", "ddd2", 200)
  t3 = Transaction("yicheng", "ddd3", 300)
  t4 = Transaction("yicheng", "ddd4", 400)
  t5 = Transaction("yicheng", "ddd5", 500)
  t6 = Transaction("yicheng", "ddd6", 600)
  m1 = DaDaMessage(t1)
  m2 = DaDaMessage(t2)
  m3 = DaDaMessage(t3)
  m4 = DaDaMessage(t4)
  m5 = DaDaMessage(t5)
  m6 = DaDaMessage(t6)
  try:
    yin1 = Block(m1, m2)
    yin1.seal()
    yin2 = Block(m3, m4)
    yin2.seal()
    yin3 = Block(m5, m6)
    yin3.seal()
    # 篡改区块
    #yin3.messagelist.append(m1)
    coin = Dada_BlockCoin() # 区块链
    coin.add_block(yin1)
    coin.add_block(yin2)
    coin.add_block(yin3)
    coin.validate()
    print(coin)
  except InvalidMessage as e:
    print(e)
  except InvalidBlock as e:
    print(e)
  except InvalidBlockCoin as e:
    print(e)

测试运行结果如下：