代碼

import hashlib import json import requests from textwrap import dedent from time import time from uuid import uuid4 from urllib.parse import urlparse from flask import Flask, jsonify, requestclass Blockchain(object):def __init__(self):...self.nodes = set()# 用 set 來儲存節(jié)點，避免重復添加節(jié)點....self.chain = []self.current_transactions = []# 創(chuàng)建創(chuàng)世區(qū)塊self.new_block(previous_hash=1, proof=100)def reister_node(self, address):"""在節(jié)點列表中添加一個新節(jié)點:param address::return:"""prsed_url = urlparse(address)self.nodes.add(prsed_url.netloc)def valid_chain(self, chain):"""確定一個給定的區(qū)塊鏈是否有效:param chain::return:"""last_block = chain[0]current_index = 1while current_index < len(chain):block = chain[current_index]print('{}'.format(last_block))print('{}'.format(block))print("\n______\n")# 檢查block的散列是否正確if block['previous_hash'] != self.hash(last_block):return False# 檢查工作證明是否正確if not self.valid_proof(last_block['proof'], block['proof']):return Falselast_block = blockcurrent_index += 1return Truedef ressolve_conflicts(self):"""共識算法:return:"""neighbours = self.nodesnew_chain = None# 尋找最長鏈條max_length = len(self.chain)# 獲取并驗證網絡中的所有節(jié)點的鏈for node in neighbours:response = requests.get('http://{}/chain'.format(node))if response.status_code == 200:length = response.json()['length']chain = response.json()['chain']# 檢查長度是否長，鏈是否有效if length > max_length and self.valid_chain(chain):max_length = lengthnew_chain = chain# 如果發(fā)現(xiàn)一個新的有效鏈比當前的長，就替換當前的鏈if new_chain:self.chain = new_chainreturn Truereturn Falsedef new_block(self, proof, previous_hash=None):"""創(chuàng)建一個新的塊并將其添加到鏈中:param proof: 由工作證明算法生成證明:param previous_hash: 前一個區(qū)塊的hash值:return: 新區(qū)塊"""block = {'index': len(self.chain) + 1,'timestamp': time(),'transactions': self.current_transactions,'proof': proof,'previous_hash': previous_hash or self.hash(self.chain[-1]),}# 重置當前交易記錄self.current_transactions = []self.chain.append(block)return blockdef new_transaction(self, sender, recipient, amount):# 將新事務添加到事務列表中"""Creates a new transaction to go into the next mined Block:param sender:發(fā)送方的地址:param recipient:收信人地址:param amount:數(shù)量:return:保存該事務的塊的索引"""self.current_transactions.append({'sender': sender,'recipient': recipient,'amount': amount,})return self.last_block['index'] + 1@staticmethoddef hash(block):"""給一個區(qū)塊生成 SHA-256 值:param block::return:"""# 必須確保這個字典（區(qū)塊）是經過排序的，否則將會得到不一致的散列block_string = json.dumps(block, sort_keys=True).encode()return hashlib.sha256(block_string).hexdigest()@propertydef last_block(self):# 返回鏈中的最后一個塊return self.chain[-1]def proof_of_work(self, last_proof):# 工作算法的簡單證明proof = 0while self.valid_proof(last_proof, proof) is False:proof += 1return proof@staticmethoddef valid_proof(last_proof, proof):# 驗證證明guess = ('{}{}'.format(last_proof, proof)).encode()guess_hash = hashlib.sha256(guess).hexdigest()return guess_hash[:4] == "0000"# 實例化節(jié)點 app = Flask(__name__)# 為該節(jié)點生成一個全局惟一的地址 node_identifier = str(uuid4()).replace('-', '')# 實例化Blockchain類 blockchain = Blockchain()# 進行挖礦請求 @app.route('/mine', methods=['GET']) def mine():# 運行工作算法的證明來獲得下一個證明。last_block = blockchain.last_blocklast_proof = last_block['proof']proof = blockchain.proof_of_work(last_proof)# 必須得到一份尋找證據(jù)的獎賞。blockchain.new_transaction(sender="0",recipient=node_identifier,amount=1,)# 通過將其添加到鏈中來構建新的塊previous_hash = blockchain.hash(last_block)block = blockchain.new_block(proof, previous_hash)response = {'message': "New Block Forged",'index': block['index'],'transactions': block['transactions'],'proof': block['proof'],'previous_hash': block['previous_hash'],}return jsonify(response), 200# 創(chuàng)建交易請求 @app.route('/transactions/new', methods=['POST']) def new_transactions():values = request.get_json()# 檢查所需要的字段是否位于POST的data中required = ['seder', 'recipient', 'amount']if not all(k in values for k in request):return 'Missing values', 400# 創(chuàng)建一個新的事物index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])response = {'message': 'Transaction will be added to Block {}'.format(index)}return jsonify(response), 201# 獲取所有快信息 @app.route('/chain', methods=['GET']) def full_chain():response = {'chain': blockchain.chain,'length': len(blockchain.chain),}return jsonify(response), 200# 添加節(jié)點 @app.route('/nodes/register', methods=['POST']) def register_nodes():values = request.get_json()nodes = values.get('nodes')if nodes is None:return "Error: Please supply a valid list of nodes", 400for node in nodes:blockchain.register_node(node)response = {'message': 'New nodes have been added','total_nodes': list(blockchain.nodes),}return jsonify(response), 201# 解決沖突 @app.route('/nodes/resolve', methods=['GET']) def consensus():replaced = blockchain.resolve_conflicts()if replaced:response = {'message': 'Our chain was replaced','new_chain': blockchain.chain}else:response = {'message': 'Our chain is authoritative','chain': blockchain.chain}return jsonify(response), 200if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)

類 BlockChain分析:

構造函數(shù)：

• 創(chuàng)建一個變量nodes來存儲所有的節(jié)點(保證節(jié)點之間互不相同)
• 創(chuàng)建一個變量chain來存儲鏈
• 調用new_block函數(shù)創(chuàng)建創(chuàng)世塊(設置初始hash值(創(chuàng)世塊的previous hash), 檢驗值設置為100)

hash函數(shù)(static method):

• 將整個塊的轉成一個string,，然后再通過這個string生成hash值，并轉成16進制
• 值得稱贊的是，這里要確保通過json將dict轉成string的過程中，需要保證keys的順序。只有這樣，才能確保整個hash映射到整個塊的時候，得到的結果具有唯一性

last_block函數(shù)(通過@property操作符變成了一個屬性)

• 會返回這個blockChain中最后一個塊

valid_proof證明函數(shù)

• 傳進來的兩個檢驗值，一個是last_proof 還有一個就是proof。
• 在直接放在一起之后，通過哈希映射之中的sha256映射(再轉16進制)
• 最后，通過判斷上面得到的最終的哈希值來判斷前4位是不是都是0
• 如果4個0開頭，那么就是合法，否者就是不合法的

proof_of_work函數(shù)

• 通過傳進來的那個函數(shù)中的參數(shù)(也就是前一個的proof值)
• 通過循環(huán)來逐步找到對應的當前proof使得上面的valid_proof函數(shù)返回的是一個true值

reister_node()函數(shù)

• 傳進來的參數(shù)為address,表示地址。
• 這個參數(shù)必須是網頁url，或者是對應的ip地址。
• 經過解析之后，節(jié)點的命名就是一個ip地址或者是網頁地址（網絡層面上跟ip地址等價）

valid_chain()函數(shù)

• 這里會傳進來一個chain 可以理解為一個列表之類的可遍歷的對象。
• 然后判斷后一個塊的previous_hash是不是真的就是前一個塊做了hash的結果
• 同時也需要判斷，這個檢驗值是否合法
• 直到這些都滿足之后，才算是ok的

new_block函數(shù)：

• 創(chuàng)建一個新的塊。語言層面上，其實就是一個字典。每個塊存儲的交易都是整個類一開始就公有的
• index設置為鏈長度+1
• ‘previous_hash’:previous_hash or self.hash(self.chain[-1]) 這個地方有點意思。
• 如果是創(chuàng)世塊，這里就會直接得到對應的結果

ressolve_conflicts函數(shù)：

• 先循環(huán)檢驗每個節(jié)點。來進行判斷
• 要求子節(jié)點存儲的長度要小于總長度
• 要求子節(jié)點存儲的鏈也必須是合法的
• 如果有合法的，并且子節(jié)點上的鏈長度更長那就復制給main服務器
• 如果發(fā)生過改變，就返回True
• else return false

總結

以上是生活随笔為你收集整理的区块链实现代码详细分析（Python）的全部內容，希望文章能夠幫你解決所遇到的問題。

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