在區塊鏈或數字貨幣領域，Python并不是主流的開發語言。但是如果 你的目的是研究區塊鏈技術的原理，或者需要在自己的筆記本上仿真一個 區塊鏈網絡并進行一些研究性的實驗，比如完成自己的畢業設計項目 或科研課題，那么Python就是合適的。在這個教程里，我們將學習如何 使用Python從零開發一個多節點的區塊鏈網絡，并基于這個仿真區塊鏈網絡， 開發一個去中心化的數據分享應用。

相關教程鏈接： 區塊鏈畢業論文 | 以太坊 | 比特幣 | EOS | Tendermint | Hyperledger Fabric | Omni/USDT | Ripple

1、Python仿真區塊鏈：用區塊分批保存交易

我們首先要把數據以JSON格式存入區塊鏈。JSON是一種常用的跨語言的 數據交換格式，例如一篇博客的JSON表示看起來就像這樣：

{ "author": "some_author_name", "content": "Some thoughts that author wants to share", "timestamp": "The time at which the content was created"}

在區塊鏈領域，我們經常使用 交易 來代替上面說到的數據。因此，為了 避免引起混亂并保持一致，在這個教程里我們將使用 交易 這個術語 來表示要存入區塊鏈的數據。

交易被分批打包進區塊，一個區塊可以包含一個或多個交易。包含交易的區塊會 定期生成并加入區塊鏈。因為會有很多區塊，所以每個區塊都應當有一個唯一 的ID。下面是我們的Python仿真區塊鏈的Block類定義代碼：

class Block: def __init__(self, index, transactions, timestamp): """ Constructor for the `Block` class. :param index: Unique ID of the block. :param transactions: List of transactions. :param timestamp: Time of generation of the block. """ self.index = index self.transactions = transactions self.timestamp = timestamp

2、Python仿真區塊鏈：為區塊添加抗篡改的數字指紋

區塊鏈的一個特點就是存儲在區塊中的交易不可篡改，為了實現這個特性， 首先需要能夠檢測出區塊數據被篡改。為此目的，我們需要使用密碼學中 的哈希(Hash)函數。

哈希函數可以把任意大小的輸入數據轉換為固定大小的輸出數據，也就是 數據的哈希，而且不同的輸入數據(基本上)會得到不同的輸出數據，因此 可以使用輸出的哈希作為輸入數據的標識。一個理想的哈希函數具有如下 特點：

• 應當易于計算
• 應當是確定性的，對于相同的輸入數據總是生成相同的哈希
• 應當具有均勻隨機性，輸入數據的一點變化也會導致輸出哈希的顯著改變

這樣我們就可以保證：

• 從哈希猜測出輸入數據是什么基本是不可能的，唯一的辦法是嘗試所有可能的組合
• 如果同時知道輸入和輸出，那么你可以通過簡單地重算來驗證哈希是否正確

顯然，從輸入數據推導出哈希很簡單，然而從哈希推導出輸入數據則是幾乎 不可能的，這一非對稱性值就是區塊鏈用來獲取期望的抗篡改能力的關鍵。

目前有很多流行的哈希函數，下面是一個使用SHA-256哈希函數的Python示例：

>>> from hashlib import sha256>>> data = b"Some variable length data">>> sha256(data).hexdigest()'b919fbbcae38e2bdaebb6c04ed4098e5c70563d2dc51e085f784c058ff208516'>>> sha256(data).hexdigest() # no matter how many times you run it, the result is going to be the same 256 character string'b919fbbcae38e2bdaebb6c04ed4098e5c70563d2dc51e085f784c058ff208516'>>> data = b"Some variable length data2" # Added one character at the end.'9fcaab521baf8e83f07512a7de7a0f567f6eef2688e8b9490694ada0a3ddeec8'

注意在上面的示例中，輸入數據的一點變化就得到完全不同的哈希！

在教程的Python仿真區塊鏈項目中，我們將把區塊哈希保存為區塊的一個字段， 用它作為區塊數據的數字指紋(Digital Fingerprint)，或者說簽名(Signature)。

下面是計算區塊哈希的Python實現代碼：

from hashlib import sha256import jsondef compute_hash(block): """ Returns the hash of the block instance by first converting it into JSON string. """ block_string = json.dumps(self.__dict__, sort_keys=True) return sha256(block_string.encode()).hexdigest()

注意：在大多數數字加密貨幣實現中，區塊中的每個交易也需要計算哈希并 利用一個樹形結構(merkle樹)來計算一組交易的根哈希。不過這對于區塊鏈 來說并不是必需的，因此我們暫時忽略這一特性。

3、Python仿真區塊鏈：將區塊一個個鏈接起來

好了，現在我們已經搞定區塊類Block的Python實現了，現在來看看如何 用Ptyhon實現區塊鏈結構。

區塊鏈就是區塊的集合，我們可以使用Python列表來保存所有的區塊。 不過這還不夠，因為如果有人故意用一個較早的區塊替換掉集合中的新區塊 還會導致數據被篡改。

我們需要一個辦法來保證對較早的區塊的修改會導致整條區塊鏈的無效。 比特幣使用的辦法是讓后面區塊的哈希依賴于前面較早的區塊。為將 區塊鏈接起來，我們需要在區塊結構中增加一個新的字段來保存前一個 區塊的哈希：previous_hash。

好了，如果每個區塊都通過previous_hash字段鏈接到前一個區塊，那么 第一個區塊怎么辦？在區塊鏈領域，第一個區塊被稱為創世區塊(Genesis Block)， 可以手工生成創世區塊或者使用一些特定的邏輯。現在讓我們為Block類 添加previous_hash字段并實現區塊鏈結構定義，下面是Blockchain類的 Python實現代碼：

from hashlib import sha256import jsonimport timeclass Block: def__init__(self, index, transactions, timestamp, previous_hash): """ Constructor for the `Block` class. :param index: Unique ID of the block. :param transactions: List of transactions. :param timestamp: Time of generation of the block. :param previous_hash: Hash of the previous block in the chain which this block is part of. """ self.index = index self.transactions = transactions self.timestamp = timestamp self.previous_hash = previous_hash # Adding the previous hash field def compute_hash(self): """ Returns the hash of the block instance by first converting it into JSON string. """ block_string = json.dumps(self.__dict__, sort_keys=True) # The string equivalent also considers the previous_hash field now return sha256(block_string.encode()).hexdigest()class Blockchain: def __init__(self): """ Constructor for the `Blockchain` class. """ self.chain = [] self.create_genesis_block() def create_genesis_block(self): """ A function to generate genesis block and appends it to the chain. The block has index 0, previous_hash as 0, and a valid hash. """ genesis_block = Block(0, [], time.time(), "0") genesis_block.hash = genesis_block.compute_hash() self.chain.append(genesis_block) @property def last_block(self): """ A quick pythonic way to retrieve the most recent block in the chain. Note that the chain will always consist of at least one block (i.e., genesis block) """ return self.chain[-1]

現在，如果任何較早的區塊被修改，那么：

• 該較早區塊的哈希會變化
• 這會導致與后面區塊的previous_hash字段記錄的內容不一致
• 由于計算區塊哈希的輸入數據包含了previous_hash字段的內容，因此下一個區塊的哈希也會變化

最終，從被替換掉的區塊開始的整條鏈都失效了，修復這一問題的唯一 辦法是重算整條鏈。

4、Python仿真區塊鏈：實現工作量證明算法

不過還有一個問題。如果我們修改了之前的區塊，如果重算后面的其他 區塊非常簡單的話，那么篡改區塊鏈也不是什么難事了。為了避免這一問題， 我們可以利用前面提到的哈希函數的非對稱性來加大區塊哈希計算工作的 難度和隨機性。我們要做的是：只接受符合特定約束條件的區塊哈希。 現在讓我們增加一個約束條件，要求區塊哈希的開始部分至少有n個0， 其中n是一個正整數。

我們知道，除非改變區塊數據的內容，否則區塊哈希不會變化，當然 我們也不希望修改已有的數據。那么我們該怎么做？很簡單！我們再增加 一些我們可以隨便修改的數據就是了。因此我們需要為Block類增加一個 新的字段nonce，我們可以通過改變這個字段的值來得到不同的區塊哈希， 直到滿足指定的約束條件，而這時的nonce值就是我們工作量的證明。

上面的這一過程就是比特幣使用的hashcash算法的簡化版本。約束條件 中指定的前導0的數量決定了我們的工作量證明算法的難度：前導0的數量 越多，就越難找到合適的nonce。

同時，由于哈希函數的非對稱性，工作量證明不容易計算，但是容易進行驗證。

下面是工作量證明算法(PoW：Proof of Work)的Python實現代碼：

class Blockchain: # difficulty of PoW algorithm difficulty = 2 """ Previous code contd.. """ def proof_of_work(self, block): """ Function that tries different values of the nonce to get a hash that satisfies our difficulty criteria. """ block.nonce = 0 computed_hash = block.compute_hash() while not computed_hash.startswith('0' * Blockchain.difficulty): block.nonce += 1 computed_hash = block.compute_hash() return computed_hash

需要指出的是，沒有簡單的邏輯可以快速找到滿足約束條件的nonce值，因此 只能進行暴力計算。

5、Python仿真區塊鏈：將區塊加入區塊鏈

要將區塊加入區塊鏈，我們首先需要驗證：

• 區塊中的數據沒有被篡改，所提供的工作量證明是正確的
• 交易的順序是正確的，previous_hash字段指向我們鏈上最新區塊的哈希

現在讓我們看一下將區塊上鏈的Python實現代碼：

class Blockchain: """ Previous code contd.. """ def add_block(self, block, proof): """ A function that adds the block to the chain after verification. Verification includes: * Checking if the proof is valid. * The previous_hash referred in the block and the hash of a latest block in the chain match. """ previous_hash = self.last_block.hash if previous_hash != block.previous_hash: return False if not Blockchain.is_valid_proof(block, proof): return False block.hash = proof self.chain.append(block) return True def is_valid_proof(self, block, block_hash): """ Check if block_hash is valid hash of block and satisfies the difficulty criteria. """ return (block_hash.startswith('0' * Blockchain.difficulty) and block_hash == block.compute_hash())

6、Python仿真區塊鏈：挖礦

交易一開始是保存在未確認交易池中的。將未確認交易放入區塊并計算工作量 證明的過程，就是廣為人知的挖礦。一旦找出了滿足指定約束條件的nonce，我們 就可以說挖出了一個可以上鏈的區塊。

在大多數數字加密貨幣中，包括比特幣，礦工都會得到加密貨幣獎勵，以回報 其為計算工作量證明所投入的算力。下面是我們的挖礦函數的Python實現代碼：

class Blockchain: def __init__(self): self.unconfirmed_transactions = [] # data yet to get into blockchain self.chain = [] self.create_genesis_block() """ Previous code contd... """ def add_new_transaction(self, transaction): self.unconfirmed_transactions.append(transaction) def mine(self): """ This function serves as an interface to add the pending transactions to the blockchain by adding them to the block and figuring out proof of work. """ if not self.unconfirmed_transactions: return False last_block = self.last_block new_block = Block(index=last_block.index + 1, transactions=self.unconfirmed_transactions, timestamp=time.time(), previous_hash=last_block.hash) proof = self.proof_of_work(new_block) self.add_block(new_block, proof) self.unconfirmed_transactions = [] return new_block.index

好了，我們就快要完成這個Python仿真區塊鏈項目了！

7、Python仿真區塊鏈：為節點添加API接口

現在該為我們的仿真區塊鏈節點添加API接口了，這樣應用程序就可以利用 這些API開發具體的應用。我們將使用流行的Python微框架Flask來創建 REST API。如果你以前使用過其他web框架，那么下面的代碼應當不難理解， 如果沒有接觸過web框架的話也別擔心，這里有一個非常棒的Flask教程：

from flask import Flask, requestimport requests# Initialize flask applicationapp = Flask(__name__)# Initialize a blockchain object.blockchain = Blockchain()

我們需要一個可以提交新交易的訪問端節點，這樣我們的應用就可以 利用這個API來將新數據添加到區塊鏈中。下面是節點的/new_transaction 訪問端節點的Python實現代碼：

# Flask's way of declaring end-points@app.route('/new_transaction', methods=['POST'])def new_transaction(): tx_data = request.get_json() required_fields = ["author", "content"] for field in required_fields: if not tx_data.get(field): return "Invalid transaction data", 404 tx_data["timestamp"] = time.time() blockchain.add_new_transaction(tx_data) return "Success", 201

另一個端節點/chain可以返回區塊鏈的數據。我們的應用將利用這個API 來查詢要顯示的數據。下面是這個端節點的Python實現代碼：

@app.route('/chain', methods=['GET'])def get_chain(): chain_data = [] for block in blockchain.chain: chain_data.append(block.__dict__) return json.dumps({"length": len(chain_data), "chain": chain_data})

挖礦很費CPU，因此我們不希望讓節點一直挖礦，而是提供一個 訪問端節點/mine來提供按需挖礦服務。 下面是Python實現代碼：

@app.route('/mine', methods=['GET'])def mine_unconfirmed_transactions(): result = blockchain.mine() if not result: return "No transactions to mine" return "Block #{} is mined.".format(result)@app.route('/pending_tx')def get_pending_tx(): return json.dumps(blockchain.unconfirmed_transactions)

這些REST訪問端節點可以用來操作我們的區塊鏈，比如提交一些交易，然后 通過挖礦確認這些交易等等。

8、Python仿真區塊鏈：實現最長鏈共識與去中心化計算

到目前為止，我們用Python從零實現的仿真區塊鏈是運行在一臺計算機上的。即使 我們已經利用哈希將區塊前后鏈接起來，并應用了工作量證明約束，我們 還是不能只信任單一的節點。我們需要實現分布式數據存儲，我們需要 多個節點來維護區塊鏈。因此，為了從單一節點轉向P2P網絡，讓我們先 創建一個機制來讓網絡上的節點彼此了解。

首先定義一個新的訪問端節點/register_node用來在網絡中注冊新節點。 下面是Python實現代碼：

# Contains the host addresses of other participating members of the networkpeers = set()# Endpoint to add new peers to the network@app.route('/register_node', methods=['POST'])def register_new_peers(): # The host address to the peer node node_address = request.get_json()["node_address"] if not node_address: return "Invalid data", 400 # Add the node to the peer list peers.add(node_address) # Return the blockchain to the newly registered node so that it can sync return get_chain()@app.route('/register_with', methods=['POST'])def register_with_existing_node(): """ Internally calls the `register_node` endpoint to register current node with the remote node specified in the request, and sync the blockchain as well with the remote node. """ node_address = request.get_json()["node_address"] if not node_address: return "Invalid data", 400 data = {"node_address": request.host_url} headers = {'Content-Type': "application/json"} # Make a request to register with remote node and obtain information response = requests.post(node_address + "/register_node", data=json.dumps(data), headers=headers) if response.status_code == 200: global blockchain global peers # update chain and the peers chain_dump = response.json()['chain'] blockchain = create_chain_from_dump(chain_dump) peers.update(response.json()['peers']) return "Registration successful", 200 else: # if something goes wrong, pass it on to the API response return response.content, response.status_codedef create_chain_from_dump(chain_dump): blockchain = Blockchain() for idx, block_data in enumerate(chain_dump): block = Block(block_data["index"], block_data["transactions"], block_data["timestamp"], block_data["previous_hash"]) proof = block_data['hash'] if idx > 0: added = blockchain.add_block(block, proof) if not added: raise Exception("The chain dump is tampered!!") else: # the block is a genesis block, no verification needed blockchain.chain.append(block) return blockchain

新加入網絡的節點可以利用/register_with endpoint端節點調用 register_with_existing_node方法進行注冊。這有助于解決以下問題：

• 要求遠端節點在其已知鄰節點中添加一個新的條目
• 使用遠端節點的數據初始化新節點上的區塊鏈
• 如果節點中途有下線，而可以重新從網絡同步區塊鏈

然而，當存在多個區塊鏈節點時有一個問題需要解決：不管有意或無意(例如網絡延遲)， 不同節點上的區塊鏈可能彼此不同。在這種情況下，節點之間需要就區塊鏈的版本 達成一致，以便維護整個系統的一致性。換句話說，我們需要達成共識。

當不同節點上的區塊鏈出現分化時，一個簡單的共識算法是選擇最長有效鏈。 這一方法背后的合理性在于，最長的鏈包含了最多的已經投入的工作量證明計算。 下面是最長鏈共識算法的Python實現代碼：

class Blockchain """ previous code continued... """ def check_chain_validity(cls, chain): """ A helper method to check if the entire blockchain is valid. """ result = True previous_hash = "0" # Iterate through every block for block in chain: block_hash = block.hash # remove the hash field to recompute the hash again # using `compute_hash` method. delattr(block, "hash") if not cls.is_valid_proof(block, block.hash) or previous_hash != block.previous_hash: result = False break block.hash, previous_hash = block_hash, block_hash return resultdef consensus(): """ Our simple consensus algorithm. If a longer valid chain is found, our chain is replaced with it. """ global blockchain longest_chain = None current_len = len(blockchain.chain) for node in peers: response = requests.get('{}/chain'.format(node)) length = response.json()['length'] chain = response.json()['chain'] if length > current_len and blockchain.check_chain_validity(chain): # Longer valid chain found! current_len = length longest_chain = chain if longest_chain: blockchain = longest_chain return True return False

現在，我們需要提供一個Python方法讓節點在挖出區塊時可以將這一 消息廣播給其他節點，這樣我們的仿真區塊鏈網絡中的每個參與者都可以 更新其本地區塊鏈，然后接著挖下一個區塊。收到區塊廣播的節點很簡單 就可以驗證工作量證明，然后將收到區塊加入到自己的本地鏈上。

下面是節點的/add_block訪問端節點的Python實現代碼：

# endpoint to add a block mined by someone else to# the node's chain. The node first verifies the block# and then adds it to the chain.@app.route('/add_block', methods=['POST'])def verify_and_add_block(): block_data = request.get_json() block = Block(block_data["index"], block_data["transactions"], block_data["timestamp"], block_data["previous_hash"]) proof = block_data['hash'] added = blockchain.add_block(block, proof) if not added: return "The block was discarded by the node", 400 return "Block added to the chain", 201def announce_new_block(block): """ A function to announce to the network once a block has been mined. Other blocks can simply verify the proof of work and add it to their respective chains. """ for peer in peers: url = "{}add_block".format(peer) requests.post(url, data=json.dumps(block.__dict__, sort_keys=True))

announce_new_block方法應當在區塊被挖出的時候調用，這樣其他節點 就可以更新自己本地保存的區塊鏈副本：

@app.route('/mine', methods=['GET'])def mine_unconfirmed_transactions(): result = blockchain.mine() if not result: return "No transactions to mine" else: # Making sure we have the longest chain before announcing to the network chain_length = len(blockchain.chain) consensus() if chain_length == len(blockchain.chain): # announce the recently mined block to the network announce_new_block(blockchain.last_block) return "Block #{} is mined.".format(blockchain.last_block.index

9、Python仿真區塊鏈：開發去中心化應用程序

好了，現在我們的仿真區塊鏈的節點軟件已經開發完了。現在需要開發 應用程序的用戶界面了。我們使用Jinja2模板來渲染網頁，并使用一些CSS來 讓網頁看起來美觀一些。

我們的應用需要連接到這個仿真區塊鏈網絡中的某個節點以便獲取數據或者 提交新數據。下面是應用的初始化部分的Python代碼：

import datetimeimport jsonimport requestsfrom flask import render_template, redirect, requestfrom app import app# Node in the blockchain network that our application will communicate with# to fetch and add data.CONNECTED_NODE_ADDRESS = "http://127.0.0.1:8000"posts = []

fetch_posts方法利用節點的/chain端節點獲取數據、解析數據并保存在本地：

def fetch_posts(): """ Function to fetch the chain from a blockchain node, parse the data, and store it locally. """ get_chain_address = "{}/chain".format(CONNECTED_NODE_ADDRESS) response = requests.get(get_chain_address) if response.status_code == 200: content = [] chain = json.loads(response.content) for block in chain["chain"]: for tx in block["transactions"]: tx["index"] = block["index"] tx["hash"] = block["previous_hash"] content.append(tx) global posts posts = sorted(content, key=lambda k: k['timestamp'], reverse=True)

應用程序使用一個HTML表單來接收用戶輸入然后利用一個POST請求 將交易添加到所連接節點的未確認交易池中。然后交易會被我們的 仿真區塊鏈網絡確認并最終當刷新網頁時被再次讀取：

@app.route('/submit', methods=['POST'])def submit_textarea(): """ Endpoint to create a new transaction via our application """ post_content = request.form["content"] author = request.form["author"] post_object = { 'author': author, 'content': post_content, } # Submit a transaction new_tx_address = "{}/new_transaction".format(CONNECTED_NODE_ADDRESS)

10、Python仿真區塊鏈：如何運行應用

終于完成了！你可以在github上查看這個Python仿真區塊鏈的 完整源代碼。

首先克隆項目倉庫：

$ git clone https://github.com/ezpod/python-blockchain-sim.git

安裝必要的Python項目依賴：

$ cd python_blockchain_app$ pip install -r requirements.txt

啟動我們的仿真區塊鏈節點：

$ export FLASK_APP=node_server.py$ flask run --port 8000

現在我們的一個仿真區塊鏈節點實例已經啟動并在8000端口監聽。

開啟另一個終端運行我們的去中心化應用：

$ python run_app.py

現在應用已經啟動，可以通過這個網址訪問： http://localhost:5000.

下圖展示了如何利用web界面向我們的仿真區塊鏈提交內容：

下圖展示了如何利用web界面啟動節點挖礦：

下圖展示了如何利用web界面重新同步區塊鏈數據：

11、Python仿真區塊鏈：運行多個節點

要運行包含多個節點的仿真區塊鏈網絡，可以使用register_with/端節點 在網絡中注冊新節點。

下面是一個多節點的示例場景，我們啟動了3個仿真節點，分別在8000、8001和 8002端口監聽：

# already running$ flask run --port 8000 spinning up new nodes$ flask run --port 8001 &$ flask run --port 8002 &

可以使用下面的cURL請求注冊在8001和8002端口監聽的兩個新節點：

$ curl -X POST http://127.0.0.1:8001/register_with -H 'Content-Type: application/json' -d '{"node_address": "http://127.0.0.1:8000"}'$ curl -X POST http://127.0.0.1:8002/register_with -H 'Content-Type: application/json' -d '{"node_address": "http://127.0.0.1:8000"}'

這樣在端口8000監聽的節點就可以知道還有8001和8002監聽的節點， 反之亦然。新加入的節點也會從原來的在網節點同步區塊鏈數據， 這樣新節點就可以參與后續的挖礦過程了。

要修改前端應用同步的區塊鏈節點，可以修改views.py文件中的 CONNECTED_NODE_ADDRESS字段。

一旦完成上面的操作，你就可以運行應用(python run_app.py)并 通過web界面創建交易了。當你挖礦后，網絡中的所有節點都會更新 自己的本地區塊鏈。也可以使用cURL或Postman利用/chain端節點查看 區塊鏈。例如：

$ curl -X GET http://localhost:8001/chain$ curl -X GET http://localhost:8002/chain

12、Python仿真區塊鏈：如何驗證交易

你可能注意到我們的基于仿真區塊鏈的去中心化應用還有一點瑕疵：任何人在任何時間都 可以提交任何內容。解決這一問題的一個辦法，就是使用非對稱密鑰技術創建用戶賬戶。 每一個新用戶都需要一個公鑰(對應賬戶名)和一個私鑰才能在我們的應用 中提交數據。私鑰用來創建數據的簽名，而公鑰用來驗證數據的簽名，下面是 其工作機制：

• 每一個提交的新交易都使用用戶的私鑰進行簽名。這個簽名與用戶信息一起 添加到交易數據中
• 在驗證階段，當挖礦時，我們可以使用公鑰和簽名驗證數據中生成的發送方 和簽名是否相符，以及驗證消息是否被修改。

13、Python仿真區塊鏈：教程總結

在這個教程中，我們學習了一個公鏈的基本概念，并利用Python實現了一個仿真區塊鏈 以及基于這個仿真區塊鏈的Flask應用。如果你走完了整個教程，相信應該可以使用Python 從零實現一個區塊鏈，并基于這個仿真的區塊鏈開發自己的去中心化應用，或者利用 這個仿真區塊鏈網絡進行一些研究性的實驗。教程中的區塊鏈實現沒有比特幣或以太坊那么復雜， 但相信會有助于理解區塊鏈技術的核心問題和解決思路。

---

原文鏈接：http://blog.hubwiz.com/2020/04/01/python-sim-blockchain/

總結

以上是生活随笔為你收集整理的python timestamp转string_Python仿真区块链【含源码】的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。