/* 时间：2021/01/16 功能：mini-web框架 目录: 一: 动静资源分离 二: 动态资源替换 三: 路由列表_元组列表 四: 路由列表_装饰器 五: 动态资源_SQL查询 六: 接口开发 七: 前后端分离 八: logging 九: demo */
一: 动静资源分离
文件 : web.py
import socket import threading import frame_work class HttpWebServer(object): def __init__(self, port): # 1 创建套接字 # AF_INET: ipv4; SOCK_STREAM: tcp tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # SOL_SOCKET: 当前套接字; SO_REUSEADDR: 复用端口选项; True: 确定复用 tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) # 端口复用 # 2 绑定端口 tcp_server_socket.bind(("", port)) # 3 设置监听 tcp_server_socket.listen(128) # 最大链接数 - 128 # 4 设置属性 self.tcp_server_socket = tcp_server_socket # 处理请求 : 客户端 @staticmethod def handle_client_request(new_socket): # 1 接收数据 # 1.1 获取数据 recv_data = new_socket.recv(4096) # 1.2 检测数据 - 长度 if len(recv_data) == 0: new_socket.close() return # 1.3 打印数据 recv_content = recv_data.decode("utf-8") print(recv_content) # 2 获取资源 # 2.1 获取路径 request_list = recv_content.split(" ", maxsplit=2) request_path = request_list[1] print(request_path) # 2.2 检测路径 - 根目录 if request_path == "/": request_path = "/index.html" # 3 处理请求 # 3_1 动态数据 : web应用程序 if request_path.endswith(".html"): # 1 传递处理 env = { "request_path":request_path } status, headers, response_body = frame_work.handle_request(env) print(status, headers, response_body) # 2 组装数据 response_line = "HTTP/1.1 %s \r\n" %status # 相应行 response_header = "" # 相应头 for header in headers: response_header += "%s: %s\r\n" %header # 遍历相应头 : 元组列表 response_data = (response_line + response_header + "\r\n" + response_body).encode("utf-8") # 3 发送数据 new_socket.send(response_data) new_socket.close() # 4 关闭连接 new_socket.close() # 关闭新套接字 - 终止通讯 # 3_2 静态请求 : web服务器处理 else: try: # 1 打开资源 with open("static" + request_path, "rb") as file: # 读取方式 : rb二进制 - 图片 file_data = file.read() # 读取数据 except Exception as e: # 2_1 异常处理 # 1 读取数据 with open("static/error.html", "rb") as file: file_data = file.read() # 读取数据 # 2 组装数据 response_line = "HTTP/1.1 404 Not Found\r\n" # 相应行 response_header = "Server: PWS/1.0\r\n" # 相应头 response_body = file_data # 相应体 response = (response_line + response_header + "\r\n").encode("utf-8") + \ response_body # 组装数据 # 3 发送数据 new_socket.send(response) # 发送数据 else: # 2_2 正常处理 # 1 组装数据 response_line = "HTTP/1.1 200 OK\r\n" # 相应行 response_header = "Server: PWS/1.0\r\n" # 相应头 response_body = file_data # 相应体 response = (response_line + response_header + "\r\n").encode("utf-8") + \ response_body # 组装数据 # 2 发送数据 new_socket.send(response) # 发送数据 finally: # 3 关闭连接 new_socket.close() # 关闭新套接字 - 终止通讯 def start(self): while True: new_socket, ip_port = self.tcp_server_socket.accept() # new_socket: 新套接字; ip_port: 客户端信息 sub_thread = threading.Thread(target=self.handle_client_request, args=(new_socket,)) # 创建子线程 # print(ip_port) sub_thread.setDaemon(True) # 线程守护 sub_thread.start() # 线程启动 if __name__ == '__main__': # 创建对象 : HttpWebServer web_server = HttpWebServer(8866) # 启动对象 web_server.start()
文件 : frame_work.py
import time # 请求处理: index def index(): status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 data = time.ctime() return status, response_header, data # 请求处理: 其他 def no_found(): status = "404 Not Found" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 data = "not found" return status, response_header, data # 资源处理: 动态处理 def handle_request(env): # 1 接收数据 request_path = env["request_path"] print("# 动态资源请求 : web服务器处理", request_path) # 2 判断请求 if request_path == "/index.html": result = index() return result else: result = no_found() return result
二: 动态资源替换
文件 : frame_work.py
import time # 请求处理: index def index(): # 1 读取文件 with open("template/index.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 模拟查库 data = time.ctime() # 模拟数据 response_body = file_data.replace("{%content%}", data) # 替换变量 : {%content%} # 3 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 return status, response_header, response_body # 请求处理: 其他 def no_found(): status = "404 Not Found" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 data = "not found" return status, response_header, data # 资源处理: 动态处理 def handle_request(env): # 1 接收数据 request_path = env["request_path"] print("# 动态资源请求 : web服务器处理", request_path) # 2 判断请求 if request_path == "/index.html": result = index() return result else: result = no_found() return result
三: 路由列表_元组列表
文件 : frame_work.py
import time # 请求处理: center def center(): # 1 读取文件 with open("template/center.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 模拟查库 data = time.ctime() # 模拟数据 response_body = file_data.replace("{%content%}", data) # 替换变量 : {%content%} # 3 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 return status, response_header, response_body # 请求处理: index def index(): # 1 读取文件 with open("template/index.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 模拟查库 data = time.ctime() # 模拟数据 response_body = file_data.replace("{%content%}", data) # 替换变量 : {%content%} # 3 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 return status, response_header, response_body # 请求处理: 其他 def no_found(): status = "404 Not Found" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 data = "not found" return status, response_header, data # 路由列表 route_list = [ ("/index.html", index), ("/center.html", center) ] # 资源处理: 动态处理 def handle_request(env): request_path = env["request_path"] print("# 动态资源请求 : web服务器处理", request_path) # 新的判断: 路由列表 for path, func in route_list: if request_path == path: result = func() return result else: result = no_found() return result
四: 路由列表_装饰器
文件 : frame_work.py
import time # 路由列表 route_list = [] # 装饰器: 参数 def route(path): def decorator(func): # 添加数据 : 路由列表 route_list.append((path, func)) # def inner(): # result = func() # return result # return inner return decorator # 请求处理: center @route("/center.html") def center(): # 1 读取文件 with open("template/center.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 模拟查库 data = time.ctime() # 模拟数据 response_body = file_data.replace("{%content%}", data) # 替换变量 : {%content%} # 3 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 return status, response_header, response_body # 请求处理: index @route("/index.html") def index(): # 1 读取文件 with open("template/index.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 模拟查库 data = time.ctime() # 模拟数据 response_body = file_data.replace("{%content%}", data) # 替换变量 : {%content%} # 3 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 return status, response_header, response_body # 请求处理: 其他 def no_found(): status = "404 Not Found" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 data = "not found" return status, response_header, data # 资源处理: 动态处理 def handle_request(env): request_path = env["request_path"] print("# 动态资源请求 : web服务器处理", request_path) # 新的判断: 路由列表 for path, func in route_list: if request_path == path: result = func() return result else: result = no_found() return result
五: 动态资源_SQL查询
文件 : frame_work.py
# 请求处理: index @route("/index.html") def index(): # 1 读取文件 with open("template/index.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 查库数据 # 2.1 连接数据 conn = pymysql.connect(host=mysql_url, # ip port=3306, # 端口 user="root", # 登录用户 password="mysql", # 登录密码 database="stock_db", # 数据库名 charset="utf8") # 通讯编码 # 2.2 获取游标 cursor = conn.cursor() # 2.3 查询数据 sql = "select * from info;" # 查询语句 cursor.execute(sql) # 执行语句 # 2.4 获出数据 : 全部 result = cursor.fetchall() # 2.5 收尾数据 cursor.close() # 关闭游标 conn.close() # 关闭连接 # 3 替换变量 data = "" for row in result: data += """ <tr> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s</td> <td>%s</td> <td><input type="button" value="添加" id="toAdd" name="toAdd" systemidvaule="000007"></td> </tr> """ %row response_body = file_data.replace("{%content%}", data) # 替换变量 : {%content%} # 4 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1")] # 响应头 return status, response_header, response_body
六: 接口开发
文件 : frame_work.py
# 请求处理: center @route("/center_data.html") def center_data(): # 1 读取文件 with open("template/index.html", "r", encoding="utf-8") as file: file_data = file.read() # 2 查库数据 # 2.1 连接数据 conn = pymysql.connect(host=mysql_url, # ip port=3306, # 端口 user="root", # 登录用户 password="mysql", # 登录密码 database="stock_db", # 数据库名 charset="utf8") # 通讯编码 # 2.2 获取游标 cursor = conn.cursor() # 2.3 查询数据 sql = '''select i.code, i.short, i.chg, i.turnover, i.price, i.highs, f.note_info from info i inner join focus f on i.id = f.info_id ''' cursor.execute(sql) # 执行语句 # 2.4 获出数据 : 全部 result = cursor.fetchall() # 2.5 收尾数据 cursor.close() # 关闭游标 conn.close() # 关闭连接 # 3 类型转换 # 3.1 类型转换: 元组 - 字典列表 center_data_list = [{"code":row[0], "short":row[1], "chg":row[2], "turnover": row[3], "price": str(row[4]), "highs": str(row[5]), "note_info": row[6] } for row in result] # 3.2 类型转换: 字典列表 - json # 参数: ensure_ascii=False 控制台显示中文 json_str = json.dumps(center_data_list, ensure_ascii=False) # 4 返回数据 status = "200 OK" # 状态信息 response_header = [("Server", "PWS/1.1"), ("Content-Type", "text/html;charset=utf-8")] # 响应头 return status, response_header, json_str
七: 前后端分离
$(document).ready(function(){ // 发送ajax请求，获取个人中心数据 $.get("center_data.html", function(data){ // 回调函数 : 通讯成功 // 获取标签 : table var $table = $(".table"); // 返回数据: json; data为js对象 for(var i = 0; i < data.length; i++){ var oCenterData = data[i]; // 封装标签: tr var oTr = '<tr>' + '<td>' + oCenterData.code + '</td>'+ '<td>' + oCenterData.short + '</td>'+ '<td>' + oCenterData.chg + '</td>'+ '<td>' + oCenterData.turnover + '</td>'+ '<td>' + oCenterData.price + '</td>'+ '<td>' + oCenterData.highs + '</td>'+ '<td>' + oCenterData.note_info + '</td>'+ '<td><a type="button" class="btn btn-default btn-xs" href="/update/000007.html"> <span class="glyphicon glyphicon-star" aria-hidden="true"></span> 修改 </a></td>' + '<td><input type="button" value="删除" id="toDel" name="toDel" systemidvaule="000007"></td>' + '</tr>' // 追加标签: tr $table.append(oTr) } }, "json");
八: logging
文件 : web.py
import logging logging.basicConfig(level=logging.DEBUG, format="%(asctime)s: %(filename)s_[lineno:%(lineno)d] %(levelname)s %(message)s: ", filename="log.txt", filemode="w")
文件 : frame_work.py
# 资源处理: 动态处理 def handle_request(env): request_path = env["request_path"] # 新的判断: 路由列表 for path, func in route_list: if request_path == path: logging.info("请求动态资源: " + request_path) result = func() return result else: result = no_found() logging.error("请求动态资源: 没有相关路由信息:" + request_path) return result
九: demo
压缩包: 链接