Flask 区域块简单原理实现

看了好几个从零开始建区块链的文章，

只有以下这几个URL实在，

于是，花了一晚上，

妥妥的自已手打一次，

感觉棒棒哒~~：）

https://www.jianshu.com/p/ecfb2a9040a3

https://www.jianshu.com/p/b39c361687c1

https://www.jianshu.com/p/1e247faf8a15

ChgBlockChain.py

# coding=utf-8

import hashlib
import json
import datetime

from typing import Optional, Dict, Any, List
from urllib.parse import urlparse
import requests


class ComplexEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        else:
            return json.JSONEncoder.default(self, obj)


class ChgBlockChain(object):
    def __init__(self):
        self.chain = []
        self.current_transactions = []
        self.nodes = set()
        self.new_block(proof=100, prev_hash=1)

    def new_block(self,
                  proof: int,
                  prev_hash: Optional[str]
                  )->Dict[str, Any]:
        block = {
            "index": len(self.chain) + 1,
            "timestamp": datetime.datetime.now(),
            "transactions": self.current_transactions,
            "proof": proof,
            "prev_hash": prev_hash or self.hash(self.chain[-1])
        }

        self.current_transactions = []
        self.chain.append(block)

        return block

    def new_transaction(self, sender: str, recipient: str, amount: int)->int:
        self.current_transactions.append({
            "sender": sender,
            "recipient": recipient,
            "amount": amount
        })

        return self.last_block["index"] + 1

    @staticmethod
    def hash(block: Dict[str, Any])->str:
        block_str = json.dumps(block, sort_keys=True,  indent=4, cls=ComplexEncoder).encode("utf-8")

        return hashlib.sha256(block_str).hexdigest()

    @property
    def last_block(self)->Dict[str, Any]:

        return self.chain[-1]

    def proof_of_work(self, last_block)->int:
        last_proof = last_block["proof"]
        # last_hash = self.hash(last_block)

        proof = 0
        while self.valid_proof(last_proof, proof) is False:
            proof += 1

        return proof

    @staticmethod
    def valid_proof(last_proof: int, proof: int)->bool:
        guess = f'{last_proof}{proof}'.encode("utf-8")
        guess_hash = hashlib.sha256(guess).hexdigest()

        return guess_hash[:4] == "0000"

    def register_node(self, addr: str)->None:
        parsed_url = urlparse(addr)
        if parsed_url.netloc:
            self.nodes.add(parsed_url.netloc)
        elif parsed_url.path:
            self.nodes.add(parsed_url.path)
        else:
            raise ValueError("url invalid!")

    def valid_chain(self, chain: List[Dict[str, Any]])->bool:
        last_block = chain[0]
        current_index = 1

        while current_index < len(chain):
            block = chain[current_index]
            if block["prev_hash"] != self.hash(last_block):
                return False
            if not self.valid_proof(last_block["proof"],
                                    block["proof"]):
                return False
            last_block = block
            current_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_chain(chain):
                    max_length = length
                    new_chain = chain

        if new_chain:
            return True
        else:
            return False

           

ChgCoinBlockNode.py

# coding=utf-8

from uuid import uuid4
from flask import Flask, jsonify, request

from ChgBlockChain import ChgBlockChain

chgCoin = ChgBlockChain()
node_id = str(uuid4()).replace("-", "")
print("current node wallet address: ", node_id)

app = Flask(__name__)
app.config['DEBUG'] = True
app.config['JSONIFY_PRETTYPRINT_REGULAR'] = True


@app.route("/")
def index_page():
    return "welcome to ChgCoin..."


@app.route("/chain")
def index_chain():
    response = {
        "chain": chgCoin.chain,
        "length": len(chgCoin.chain)
    }
    return jsonify(response), 200


@app.route("/mine")
def index_mine():
    last_block = chgCoin.last_block
    proof = chgCoin.proof_of_work(last_block)

    chgCoin.new_transaction(
        sender=0,
        recipient=node_id,
        amount=12.5
    )
    block = chgCoin.new_block(proof, chgCoin.hash(last_block))
    response = {
        "message": "new block created...",
        "index": block["index"],
        "transactions": block["transactions"],
        "proof": block["proof"],
        "hash": chgCoin.hash(block),
        "prev_hash": block["prev_hash"]
    }
    return jsonify(response), 200


@app.route("/new_transactions", methods=["POST"])
def index_new_transactions():
    values = request.get_json()
    required = ["sender", "recipient", "amount"]

    if not all(key in values for key in required):
        return "data is invalid", 400
    index = chgCoin.new_transaction(values["sender"],
                                    values["recipient"],
                                    values["amount"])
    response = {"message": f"transaction added to block{index}"}

    return jsonify(response), 201


@app.route("/new_node", methods=["POST"])
def index_new_node():
    values = request.get_json()
    nodes = values.get("nodes")

    if nodes is None:
        return "It's empty node"
    for node in nodes:
        chgCoin.register_node(node)

    response = {
        "message": "new node added to block",
        "nodes": list(chgCoin.nodes)
    }

    return jsonify(response), 201


@app.route("/node_refresh")
def index_node_refresh():
    replaced = chgCoin.resolve_conflicts()
    print(replaced)

    if replaced:
        response = {
            "message": "block chain is replaced by longest chain",
            "new chain": chgCoin.chain
        }
    else:
        response = {
            "message": "block chain is no need replaced ",
            "chain": chgCoin.chain
        }
    return jsonify(response), 200


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