api_server.py

import argparse
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote, urlparse

from blockchain import Blockchain


class ChainState:
    def __init__(self, chain: Blockchain) -> None:
        self.chain = chain
        self.lock = threading.Lock()


PROJECT_ROOT = Path(__file__).resolve().parent
WEB_ROOT = PROJECT_ROOT / "web"


def write_json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None:
    body = json.dumps(payload).encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", "application/json")
    handler.send_header("Content-Length", str(len(body)))
    handler.end_headers()
    handler.wfile.write(body)


def write_text(handler: BaseHTTPRequestHandler, status: int, content_type: str, body: str) -> None:
    encoded = body.encode("utf-8")
    handler.send_response(status)
    handler.send_header("Content-Type", f"{content_type}; charset=utf-8")
    handler.send_header("Content-Length", str(len(encoded)))
    handler.end_headers()
    handler.wfile.write(encoded)


def write_bytes(handler: BaseHTTPRequestHandler, status: int, content_type: str, body: bytes) -> None:
    handler.send_response(status)
    handler.send_header("Content-Type", content_type)
    handler.send_header("Content-Length", str(len(body)))
    handler.end_headers()
    handler.wfile.write(body)


def read_json_body(handler: BaseHTTPRequestHandler) -> dict:
    content_length = int(handler.headers.get("Content-Length", "0"))
    if content_length <= 0:
        return {}
    raw = handler.rfile.read(content_length)
    return json.loads(raw.decode("utf-8"))


def build_handler(state: ChainState):
    static_files = {
        "/": ("index.html", "text/html"),
        "/index.html": ("index.html", "text/html"),
        "/styles.css": ("styles.css", "text/css"),
        "/app.js": ("app.js", "application/javascript"),
    }

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, format: str, *args) -> None:
            return

        def do_GET(self) -> None:
            parsed = urlparse(self.path)
            path = parsed.path

            if path in static_files:
                filename, content_type = static_files[path]
                file_path = WEB_ROOT / filename
                if not file_path.exists():
                    write_json(self, 500, {"error": f"Missing web asset: {filename}"})
                    return
                write_bytes(self, 200, content_type, file_path.read_bytes())
                return

            if path == "/favicon.ico":
                write_bytes(self, 204, "image/x-icon", b"")
                return

            if path == "/health":
                write_json(self, 200, {"status": "ok"})
                return

            if path == "/chain":
                with state.lock:
                    write_json(self, 200, state.chain.to_dict())
                return

            if path == "/pending":
                with state.lock:
                    write_json(
                        self,
                        200,
                        {"pending_transactions": [tx.__dict__ for tx in state.chain.pending_transactions]},
                    )
                return

            if path.startswith("/balance/"):
                address = unquote(path.removeprefix("/balance/"))
                if not address.strip():
                    write_json(self, 400, {"error": "Address is required."})
                    return
                with state.lock:
                    balance = state.chain.get_balance(address)
                write_json(self, 200, {"address": address, "balance": balance})
                return

            # Route fallback for portfolio hosting setups that request custom paths.
            if "." not in path:
                index_path = WEB_ROOT / "index.html"
                if index_path.exists():
                    write_bytes(self, 200, "text/html", index_path.read_bytes())
                    return
            write_text(self, 404, "text/plain", "Not found")

        def do_POST(self) -> None:
            parsed = urlparse(self.path)
            path = parsed.path

            try:
                body = read_json_body(self)
            except json.JSONDecodeError:
                write_json(self, 400, {"error": "Body must be valid JSON."})
                return

            if path == "/transactions":
                try:
                    sender = str(body["sender"])
                    recipient = str(body["recipient"])
                    amount = float(body["amount"])
                except (KeyError, ValueError, TypeError):
                    write_json(self, 400, {"error": "Expected sender, recipient, amount."})
                    return
                try:
                    with state.lock:
                        state.chain.add_transaction(sender, recipient, amount)
                except ValueError as exc:
                    write_json(self, 400, {"error": str(exc)})
                    return
                write_json(self, 201, {"message": "Transaction added."})
                return

            if path == "/mine":
                miner = str(body.get("miner", "")).strip()
                if not miner:
                    write_json(self, 400, {"error": "Expected miner in body."})
                    return
                try:
                    with state.lock:
                        block = state.chain.mine_pending_transactions(miner)
                except ValueError as exc:
                    write_json(self, 400, {"error": str(exc)})
                    return
                write_json(
                    self,
                    201,
                    {
                        "message": "Block mined.",
                        "index": block.index,
                        "hash": block.hash,
                        "nonce": block.nonce,
                    },
                )
                return

            if path == "/save":
                filepath = str(body.get("path", "data/chain.json"))
                try:
                    with state.lock:
                        state.chain.save_to_file(filepath)
                except OSError as exc:
                    write_json(self, 500, {"error": str(exc)})
                    return
                write_json(self, 200, {"message": "Saved.", "path": filepath})
                return

            if path == "/load":
                filepath = str(body.get("path", "data/chain.json"))
                try:
                    new_chain = Blockchain.load_from_file(filepath)
                except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError) as exc:
                    write_json(self, 400, {"error": str(exc)})
                    return
                with state.lock:
                    state.chain = new_chain
                write_json(self, 200, {"message": "Loaded.", "path": filepath})
                return

            write_json(self, 404, {"error": "Not found"})

    return Handler


def main() -> None:
    parser = argparse.ArgumentParser(description="Simple local API server for the blockchain demo.")
    parser.add_argument("--host", type=str, default="127.0.0.1", help="Bind host (default: 127.0.0.1).")
    parser.add_argument("--port", type=int, default=8000, help="Bind port (default: 8000).")
    parser.add_argument("--difficulty", type=int, default=3, help="Proof-of-work difficulty (default: 3).")
    parser.add_argument("--load", type=str, help="Load chain state from JSON file on startup.")
    args = parser.parse_args()

    if args.difficulty < 1 or args.difficulty > 6:
        raise ValueError("Difficulty must be between 1 and 6.")

    if args.load:
        chain = Blockchain.load_from_file(args.load)
    else:
        chain = Blockchain(difficulty=args.difficulty)

    state = ChainState(chain=chain)
    server = ThreadingHTTPServer((args.host, args.port), build_handler(state))
    print(f"Server listening on http://{args.host}:{args.port}")
    server.serve_forever()


if __name__ == "__main__":
    main()