Python實現簡單HTTP服務器(一)

一.返回固定內容

# coding:utf-8

import socket

from multiprocessing import Process


def handle_client(client_socket):
    """
    處理客戶端請求
    """
    request_data = client_socket.recv(1024)
    print("request data:", request_data)
    # 構造響應數據
    response_start_line = "HTTP/1.1 200 OK\r\n"
    response_headers = "Server: My server\r\n"
    response_body = "<h1>Python HTTP Test</h1>"
    response = response_start_line + response_headers + "\r\n" + response_body

    # 向客戶端返回響應數據
    client_socket.send(bytes(response, "utf-8"))

    # 關閉客戶端連接
    client_socket.close()


if __name__ == "__main__":
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(("", 8000))
    server_socket.listen(128)

    while True:
        client_socket, client_address = server_socket.accept()
        print("[%s, %s]用戶連接上了" % client_address)
        handle_client_process = Process(target=handle_client, args=(client_socket,))
        handle_client_process.start()
        client_socket.close()

運行程序，打開瀏覽器輸入：http://127.0.0.1:8000/，顯示如下：

二.返回靜態文件內容

# coding:utf-8

import socket
import re

from multiprocessing import Process

# 設置靜態文件根目錄
HTML_ROOT_DIR = "./html"


def handle_client(client_socket):
    """
    處理客戶端請求
    """
    # 獲取客戶端請求數據
    request_data = client_socket.recv(1024)
    print("request data:", request_data)
    request_lines = request_data.splitlines()
    # 解析請求報文
    request_start_line = request_lines[0]
    # 提取用戶請求的文件名
    file_name = re.match(r"\w+ +(/[^ ]*) ", request_start_line.decode("utf-8")).group(1)

    if "/" == file_name:
        file_name = "/index.html"

    # 打開文件，讀取內容
    try:
        file = open(HTML_ROOT_DIR + file_name, "rb")
    except IOError:
        response_start_line = "HTTP/1.1 404 Not Found\r\n"
        response_headers = "Server: My server\r\n"
        response_body = "The file is not found!"
    else:
        file_data = file.read()
        file.close()

        # 構造響應數據
        response_start_line = "HTTP/1.1 200 OK\r\n"
        response_headers = "Server: My server\r\n"
        response_body = file_data.decode("utf-8")

    response = response_start_line + response_headers + "\r\n" + response_body
    print("response data:", response)

    # 向客戶端返回響應數據
    client_socket.send(bytes(response, "utf-8"))

    # 關閉客戶端連接
    client_socket.close()


if __name__ == "__main__":
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind(("", 8000))
    server_socket.listen(128)

    while True:
        client_socket, client_address = server_socket.accept()
        print("[%s, %s]用戶連接上了" % client_address)
        handle_client_process = Process(target=handle_client, args=(client_socket,))
        handle_client_process.start()
        client_socket.close()

在程序所在目錄下新建文件夾(html)，里面放入HTML文件，運行程序，打開瀏覽器輸入：http://127.0.0.1:8000/，顯示如下：

改為面向對象的程序：

# coding:utf-8

import socket
import re

from multiprocessing import Process

# 設置靜態文件根目錄
HTML_ROOT_DIR = "./html"


class HTTPServer(object):
    def __init__(self):
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    def start(self):
        self.server_socket.listen(128)
        while True:
            client_socket, client_address = self.server_socket.accept()
            print("[%s, %s]用戶連接上了" % client_address)
            handle_client_process = Process(target=self.handle_client, args=(client_socket,))
            handle_client_process.start()
            client_socket.close()

    def handle_client(self, client_socket):
        """
        處理客戶端請求
        """
        # 獲取客戶端請求數據
        request_data = client_socket.recv(1024)
        print("request data:", request_data)
        request_lines = request_data.splitlines()
        # 解析請求報文
        request_start_line = request_lines[0]
        # 提取用戶請求的文件名
        file_name = re.match(r"\w+ +(/[^ ]*) ", request_start_line.decode("utf-8")).group(1)

        if "/" == file_name:
            file_name = "/index.html"

        # 打開文件，讀取內容
        try:
            file = open(HTML_ROOT_DIR + file_name, "rb")
        except IOError:
            response_start_line = "HTTP/1.1 404 Not Found\r\n"
            response_headers = "Server: My server\r\n"
            response_body = "The file is not found!"
        else:
            file_data = file.read()
            file.close()

            # 構造響應數據
            response_start_line = "HTTP/1.1 200 OK\r\n"
            response_headers = "Server: My server\r\n"
            response_body = file_data.decode("utf-8")

        response = response_start_line + response_headers + "\r\n" + response_body
        print("response data:", response)

        # 向客戶端返回響應數據
        client_socket.send(bytes(response, "utf-8"))

        # 關閉客戶端連接
        client_socket.close()

    def bind(self, port):
        self.server_socket.bind(("", port))


def main():
    http_server = HTTPServer()
    http_server.bind(8000)
    http_server.start()


if __name__ == "__main__":
    main()

三.返回動態內容(運用wsgi）

# coding:utf-8

import socket
import re
import sys

from multiprocessing import Process

# 設置靜態文件根目錄
HTML_ROOT_DIR = "./html"
# 設置動態文件根目錄
WSGI_PYTHON_DIR = "./wsgitest"


class HTTPServer(object):
    def __init__(self):
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    def start(self):
        self.server_socket.listen(128)
        while True:
            client_socket, client_address = self.server_socket.accept()
            print("[%s, %s]用戶連接上了" % client_address)
            handle_client_process = Process(target=self.handle_client, args=(client_socket,))
            handle_client_process.start()
            client_socket.close()

    def start_response(self, status, headers):
        response_headers = "HTTP/1.1 " + status + "\r\n"
        for header in headers:
            response_headers += "%s: %s\r\n" % header

        self.response_headers = response_headers

    def handle_client(self, client_socket):
        """
        處理客戶端請求
        """
        # 獲取客戶端請求數據
        request_data = client_socket.recv(1024)
        print("request data:", request_data)
        request_lines = request_data.splitlines()
        for line in request_lines:
            print(line)

        # 解析請求報文
        request_start_line = request_lines[0]
        # 提取用戶請求的文件名及請求方法
        file_name = re.match(r"\w+ +(/[^ ]*) ", request_start_line.decode("utf-8")).group(1)
        method = re.match(r"(\w+) +/[^ ]* ", request_start_line.decode("utf-8")).group(1)

        # 處理動態文件
        if file_name.endswith(".py"):
            try:
                m = __import__(file_name[1:-3])
            except Exception:
                self.response_headers = "HTTP/1.1 404 Not Found\r\n"
                response_body = "not found"
            else:
                env = {
                    "PATH_INFO": file_name,
                    "METHOD": method
                }
                response_body = m.application(env, self.start_response)

            response = self.response_headers + "\r\n" + response_body
        # 處理靜態文件
        else:
            if "/" == file_name:
                file_name = "/index.html"

            # 打開文件，讀取內容
            try:
                file = open(HTML_ROOT_DIR + file_name, "rb")
            except IOError:
                response_start_line = "HTTP/1.1 404 Not Found\r\n"
                response_headers = "Server: My server\r\n"
                response_body = "The file is not found!"
            else:
                file_data = file.read()
                file.close()

                # 構造響應數據
                response_start_line = "HTTP/1.1 200 OK\r\n"
                response_headers = "Server: My server\r\n"
                response_body = file_data.decode("utf-8")

            response = response_start_line + response_headers + "\r\n" + response_body
            print("response data:", response)

        # 向客戶端返回響應數據
        client_socket.send(bytes(response, "utf-8"))

        # 關閉客戶端連接
        client_socket.close()

    def bind(self, port):
        self.server_socket.bind(("", port))


def main():
    sys.path.insert(1, WSGI_PYTHON_DIR)
    http_server = HTTPServer()
    http_server.bind(8000)
    http_server.start()


if __name__ == "__main__":
    main()

在程序所在目錄下新建文件夾(wsgitest)，里面放入python文件(ctime.py)

# coding:utf-8

import time

def application(env, start_response):

    status = "200 OK"
    headers = [("Content-Type", "text/plain")]

    start_response(status, headers)

    return time.ctime()

運行程序，打開瀏覽器輸入:http://127.0.0.1:8000/ctime.py，顯示如下：

每刷新一次就執行相應python代碼。