python實(shí)現(xiàn)簡單區(qū)塊鏈

import hashlib import json from time import time from typing import Any, Dict, List, Optional from urllib.parse import urlparse from uuid import uuid4 import requests from flask import Flask, jsonify, request import mathclass Blockchain(object):def __init__(self):self.current_transactions = []self.chain = []# 創(chuàng)建“創(chuàng)世塊”self.new_block(previous_hash=1, proof=100)def new_block(self, proof, previous_hash=None):"""生成新塊:param proof: <int> 工作量證明，它是一個(gè)工作算法對(duì)應(yīng)的一個(gè)值:param previous_hash: (Optional) <str> 前一個(gè)區(qū)塊的hash值:return: <dict> 返回一個(gè)新的塊，這個(gè)塊block是一個(gè)字典結(jié)構(gòu)"""block = {# 新block對(duì)應(yīng)的index'index': len(self.chain) + 1,# 時(shí)間戳，記錄區(qū)塊創(chuàng)建的時(shí)間'timestamp': time(),# 記錄當(dāng)前的交易記錄，即通過new_transactions創(chuàng)建的交易，記錄在這個(gè)新的block里'transactions': self.current_transactions,# 工作量證明'proof': proof,# 前一個(gè)block對(duì)應(yīng)的hash值'previous_hash': previous_hash or self.hash(self.chain[-1]),}# 重置當(dāng)前的交易，用于記錄下一次交易self.current_transactions = []# 將新生成的block添加到block列表中self.chain.append(block)# 返回新創(chuàng)建的blcokreturn blockdef new_transaction(self, sender, recipient, amount):"""生成新交易信息，信息將加入到下一個(gè)待挖的區(qū)塊中:param sender: <str> 發(fā)送者的地址:param recipient: <str> 接受者的地址:param amount: <int> 交易額度:return: <int> 返回新的Block的Id值，新產(chǎn)生的交易將會(huì)被記錄在新的Block中"""# 實(shí)現(xiàn)很簡單，向交易列表中添加一個(gè)字典，這個(gè)字典中記錄交易雙發(fā)的信息self.current_transactions.append({'sender': sender,'recipient': recipient,'amount': amount,})# 返回最后一個(gè)區(qū)塊的index加上1，即對(duì)應(yīng)到新的區(qū)塊上return self.last_block['index'] + 1@staticmethoddef hash(block):"""生成塊的 SHA-256 hash值:param block: <dict> Block:return: <str>"""# 首先將block字典結(jié)構(gòu)轉(zhuǎn)換成json字符串，通過sort_keys指定按key拍好序。block_string = json.dumps(block, sort_keys=True).encode()# 調(diào)用sha256函數(shù)求取摘要return hashlib.sha256(block_string).hexdigest()@propertydef last_block(self):return self.chain[-1]def proof_of_work(self, last_proof):"""簡單的工作量證明:- 查找一個(gè)數(shù) p 使得 hash(p+last_proof) 以'abc'開頭- last_proof 是上一個(gè)塊的證明, p是當(dāng)前的證明:param last_proof: <int>:return: <int>"""proof = 0# 定義一個(gè)死循環(huán)，直到valid_proof驗(yàn)證通過while self.valid_proof(last_proof, proof) is False:proof += math.pireturn proof@staticmethoddef valid_proof(last_proof, proof):"""驗(yàn)證證明: 是否hash(last_proof, proof)以'abc'開頭?:param last_proof: <int> 前一個(gè)證明:param proof: <int> 當(dāng)前證明:return: <bool> """guess = f'{last_proof}{proof}'.encode()guess_hash = hashlib.sha256(guess).hexdigest()return guess_hash[:3] == "abc"#實(shí)例化一個(gè)Flask節(jié)點(diǎn) app = Flask(__name__)# 為當(dāng)前節(jié)點(diǎn)生成一個(gè)全局唯一的地址，使用uuid4方法 node_identifier = str(uuid4()).replace('-', '')#初始化區(qū)塊鏈 blockchain = Blockchain()# 告訴服務(wù)器去挖掘新的區(qū)塊 @app.route('/mine', methods=['GET']) def mine():#獲取區(qū)塊鏈最后一個(gè)blocklast_block = blockchain.last_block#取出最后一個(gè)block的proof工作量證明last_proof = last_block['proof']# 運(yùn)行工作量的證明和驗(yàn)證算法，得到proof。proof = blockchain.proof_of_work(last_proof)# 給工作量證明的節(jié)點(diǎn)提供獎(jiǎng)勵(lì).# 發(fā)送者為 "0" 表明是新挖出的幣# 接收者是我們自己的節(jié)點(diǎn)，即上面生成的node_identifier。實(shí)際中這個(gè)值可以用用戶的賬號(hào)。blockchain.new_transaction(sender="0",recipient=node_identifier,amount=1,)# 產(chǎn)生一個(gè)新的區(qū)塊，并添加到區(qū)塊鏈中block = blockchain.new_block(proof)#構(gòu)造返回響應(yīng)信息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)建一個(gè)交易并添加到區(qū)塊，POST接口可以給接口發(fā)送交易數(shù)據(jù) @app.route('/transactions/new', methods=['POST']) def new_transaction():#獲取請(qǐng)求的參數(shù)，得到參數(shù)的json格式數(shù)據(jù)values = request.get_json()print('request parameters:%s'%(values))#檢查請(qǐng)求的參數(shù)是否合法，包含sender,recipient,amount幾個(gè)字段required = ['sender', 'recipient', 'amount']if not all(k in values for k in required):return 'Missing values', 400# 使用blockchain的new_transaction方法創(chuàng)建新的交易index = blockchain.new_transaction(values['sender'], values['recipient'], values['amount'])#構(gòu)建response信息response = {'message': f'Transaction will be added to Block {index}'}#返回響應(yīng)信息return jsonify(response), 201#返回整個(gè)區(qū)塊鏈，GET接口 @app.route('/chain', methods=['GET']) def full_chain():response = {'chain': blockchain.chain,'length': len(blockchain.chain),}return jsonify(response), 200if __name__ == '__main__':#服務(wù)器運(yùn)行在5000端口上app.run(host='localhost', port=5000)

總結(jié)

以上是生活随笔為你收集整理的python实现简单区块链的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。