看了好几个从零开始建区块链的文章，
只有以下这几个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)
