1. 程式人生 > >python利用flas框架完成區塊鏈小demo(實現轉賬,節點更新)

python利用flas框架完成區塊鏈小demo(實現轉賬,節點更新)

還是迫於老闆的壓力下 從網上看了尹成老師關於對區塊鏈的教程 模仿用python寫了個小demo 這注釋也花了我不少時間

import hashlib # 資訊保安加密
import json 
import time 
from urllib.parse import urlparse #網路編碼解碼
from uuid import uuid4 # 生成唯一序列號
import requests #生成網路請求
from flask import Flask,jsonify,request #請求,網路請求
from typing import Any,Dict,List,Optional #資料結構

class DaDaCoinBlockChain:

    def __init__(self): #初始化
        self.current_transactions = []#交易列表
        self.chain=[]#區塊鏈管理多個區塊
        self.nodes=set()#儲存網路中的多個節點
        self.new_block(previous_hash="1",proof=100)#建立創世區塊

    def new_block(self,proof:int,#確定proof為int型別
                  previous_hash:Optional[str])->Dict[str,Any]: #建立一個區塊,返回字典資料型別
        block={
            "index":len(self.chain)+1,#索引
            "timestamp":time.time(),#時間戳
            "teainsactions":self.current_transactions,#當前的交易
            "proof":proof,#工作量證明,
            "previous_hash":previous_hash or self.hash(self.chain[-1])#前一塊的雜湊
        }
        self.current_transactions=[]#交易記錄加入區塊後被清空
        self.chain.append(block)#區塊加入區塊鏈
        return block

    def new_transactions(self,sender:str,recipient:str,amount:int)->int: #建立一個交易
        self.current_transactions.append({
            "sender":sender,#付款方
            "recipient":recipient,#收款方
            "amount":amount#交易金額
        })
        return self.last_block["index"]+1#索引標記交易的數量

    @property
    def last_block(self)->Dict[str,any]:#取得最後一個區塊
        return self.chain[-1]

    @staticmethod
    def hash(block:Dict[str,any])->str:# 雜湊加密傳遞一個字典返回字串
        blockstring=json.dumps(block,sort_keys=True).encode()#編碼
        return hashlib.sha3_256(blockstring).hexdigest()#取出編碼16進位制的雜湊

    def proof_of_work(self,last_proof:int)->int:#工作量證明,挖礦過程
        proof=0
        while self.valid_proof(last_proof,proof) is False: #驗證求解
            proof+=1
        return proof

    @staticmethod #第N個區塊依賴於N-1個區塊
    def valid_proof(last_proof:int,proof:int)->bool:#驗證證明
        guess=f'{last_proof*proof}'.encode()#二進位制編碼
        guess_hash=hashlib.sha3_256(guess).hexdigest()#取出雜湊值
        return guess_hash[-4:]=="0000" #驗證是否符合條件

    def register_node(self,addr:str)->None:#加入網路的其他節點,用於更新
        now_url=urlparse(addr)#解析網路
        self.nodes.add(now_url.netloc)#增加網路節點

    def valid_chain(self,chain:List[Dict[str,any]])->bool:#區塊鏈校驗
        #List[Dict[str,any]是一個列表,列表的每一個元素都是字典
        last_block=chain[0]#第一個區塊
        curr_index=1#當前的第一個索引
        while curr_index<len(chain):
            block=chain[curr_index]
            #雜湊校驗,校驗區塊鏈的連線
            if block["previous_hash"]!=self.hash():
                return False
            #工作量校驗,挖礦的工作量校驗
            if not self.valid_proof(last_block["proof"],block["proof"]):
                return False
            last_block=block#輪替迴圈
            curr_index+=1
        return True

    def resolve_conflicts(self)->bool:#共識演算法
        #網路中的多個節點,取出最長的
        neighbours=self.nodes#取得所有節點
        new_chain=None#新的區塊鏈
        max_length=len(self.chain)#當前的區塊鏈長度
        for node in neighbours:
            response=requests.get(f"http://{node}/chain")#訪問網路節點
            if response.status_code==200:
                length=response.json()["length"]#取出長度
                chain=response.json()["chain"]#取出區塊鏈
                #如果當前區塊鏈長度比較長並且經過校驗
                if length>max_length and self.valid_proof(chain):
                    max_length=length
                    new_chain=chain#儲存長度與區塊鏈
        if new_chain:
            self.chain=new_chain#替換區塊鏈
            return True
        return False

dadacoin = DaDaCoinBlockChain()#建立一個網路節點
node_id=str(uuid4()).replace("-","")#生成祕鑰
print("當前節點錢包地址",node_id)

app=Flask(__name__)#初始化Flask框架
@app.route("/")
def index_page():
    return "歡迎來到凹梗 你好"

@app.route("/chain")#檢視所有的區塊鏈
def index_chain():
    response={
        "chain":dadacoin.chain,#區塊鏈
        "length":len(dadacoin.chain) #區塊鏈的長度
    }
    return jsonify(response),200#展示區塊鏈

@app.route("/mine")#挖礦
def index_mine():
    last_block=dadacoin.last_block#取得最後一個區塊
    last_proof=last_block["proof"]#取得工作量證明
    proof=dadacoin.proof_of_work(last_proof)#挖礦計算

    #系統獎勵比特幣挖礦產生交易
    dadacoin.new_transactions(
        sender="0",#系統獎勵
        recipient=node_id,#當前錢包
        amount=10,
    )
    block=dadacoin.new_block(proof,None)#增加一個區塊
    response = {
        "message":"新的區塊建立",
        "index":block["index"],#建立索引
        "teainsactions":block["teainsactions"],#交易
        "proof":block["proof"],#工作量證明
        "previous_hash":block["previous_hash"]#上一塊的雜湊
    }
    return jsonify(response),200

@app.route("/new_trainsactions",methods=["POST"])#實現交易
def new_trainsactions():
    values=request.get_json#抓取網路傳輸的資訊
    required=["sender","recipient","amount"]
    if not all(key in values for key in required):
        return "資料不完整",400
    index=dadacoin.new_transactions(values["sender"],values["recipient"],values["amount"])

    response = {
        "message":f"交易加入到區塊{index}"
    }
    return jsonify(response),200

@app.route("/new_node",methods=["POST"])#新註冊節點
def new_node():
    values=requests.get_json()#獲取json字串
    nodes=values.get("nodes")#獲取所有的節點
    if nodes is None:
        return "沒有節點思密達",400
    for node in nodes:
        dadacoin.register_node(node)#增加網路節點
    response = {
        "message":f"網路節點已經被追加",
        "nodes":list(dadacoin.nodes),#檢視所有節點
    }
    return jsonify(response),200

@app.route("/node_refresh", methods=["POST"])  # 更新節點
def node_refresh():
    replaced=dadacoin.resolve_conflicts()#共識演算法進行替換
    if replaced:
        response={
            "message":"區塊鏈已經被替換為最長",
            "new-chain":dadacoin.chain,
        }
    else:
        response={
            "message":"當前區塊鏈已經是最長無需替換",
            "new-chain":dadacoin.chain
        }
    return jsonify(response),200

if __name__=="__main__":
    app.run("127.0.0.1",port=5000)