13_小项目实战

1.新浪图集爬虫(闭包+进程池+线程池+多协程)

import os
import re
import urllib.request
import time
from multiprocessing import Pool  # 进程池
from concurrent.futures import ThreadPoolExecutor  # 线程池

import gevent  # 协程
from gevent import monkey  # 猴子补丁

monkey.patch_all(thread=False)  # thread=False处理多线程时monkey会阻塞主线程的继续执行
IMG_URL = "http://slide.games.sina.com.cn/"
MAX_POOL_PROCESS = os.cpu_count() + 1  # 设置进程池中的最大进程数
MAX_POOL_THREADING = 5  # 设置每个子进程中线程池的最大线程数
PLATE_LIST = ["游戏壁纸", "原画欣赏", "游戏趣图", "精美周边", "动漫图片", "游戏画面"]  # 图集分类板块列表
LOCAL_PATH = os.getcwd()  # 图片保存路径


def themes():
    """主进程爬取主题图集分类"""
    req = urllib.request.urlopen(IMG_URL)
    req = str(req.read())

    # 取主题正则
    theme_re = r"<li> <a href="(.*?)" target="_blank" class="game_hover">"
    com = re.compile(theme_re)
    req_iterator = com.finditer(req)

    # 清洗数据-->迭代器的前两个成员无效url,拿到前两个值后直接丢弃
    try:
        next(req_iterator)
        next(req_iterator)
    except StopIteration:
        print("取主题正则已过期")
        return
    except Exception as e:
        print("其他错误:%s" % e)
        return

    def themes_pool():
        """根据主题对象个数创建进程池"""
        if req_iterator:
            # 定义一个进程池,网站有6个主题所以最大进程数设置为6
            po = Pool(MAX_POOL_PROCESS)
            for j, i_url in enumerate(req_iterator, 1):
                i_url = i_url.group(1)
                # 每次循环将会用空闲出来的子进程去调用目标,j是当前网站的主题板块序号
                po.apply_async(pages, (i_url, j))

            po.close()  # 关闭进程池,关闭后po不再接收新的请求
            po.join()  # 等待po中所有子进程执行完成,必须放在close语句之后

    # 返回包含迭代器(清洗后URL主题分类的链接)的闭包
    return themes_pool


def pages(i_url, j):
    """根据进程池中的每个子进程爬取页面图集分类,根据每个子进程创建线程池

    :param i_url: 第几个图集分类的URL链接
    :param j: 当前是第几个图集分类
    :return: None
    """
    req_pages = urllib.request.urlopen(i_url)
    req_pages = str(req_pages.read())

    # 取页码正则
    page_num_re = r"<!--  -->.*?href=".*?page=(d+).*?".*?<!--  -->"
    com = re.compile(page_num_re)
    page_num_iterator = com.finditer(req_pages)
    page_num = int(next(page_num_iterator).group(1))

    # 取页面url
    pages_re = r"</a><a  style=.*?href="(.*?)" title=".*?">.*?<!--  -->"
    com = re.compile(pages_re)
    pages_iterator = com.finditer(req_pages)
    pages_url = IMG_URL + next(pages_iterator).group(1)

    # 创建线程池并制定最大线程数max_workers
    executor = ThreadPoolExecutor(max_workers=MAX_POOL_THREADING)
    for i in range(1, page_num + 1):
        new_pages_url = pages_url + "&page=" + str(i) + "&dpc=1"
        executor.submit(pictures, new_pages_url, j, i)  # 异步提交任务
        # pictures(new_pages_url, j, i)  # 单线程执行
    executor.shutdown(True)  # 等待池内所有任务执行完毕回收完资源后才继续


def pictures(new_pages_url, j, i):
    """每个子进程中的线程池爬取分类中每页的图片,根据每个子线程用协程实现下载图片

    :param new_pages_url: 当前页面的图片URL链接
    :param j: 第几个图集分类
    :param i: 当前图集分类的第几页
    :return: None
    """
    page_url = urllib.request.urlopen(new_pages_url)
    page_str = str(page_url.read())
    img_re = r"img src="(.*?)" class="
    com = re.compile(img_re)
    img_iterator = com.findall(page_str)
    k = 1  # 下载每一页图片时每张图片名的自增序号,如: 1.jpg,2.jpg
    for img_url in img_iterator:
        page_path = LOCAL_PATH + "/新浪图集/%s%s/网站第%s页/" % (str(j), PLATE_LIST[j - 1], str(i))
        img_path = page_path + str(k) + ".jpg"
        if not os.path.exists(page_path):
            os.makedirs(page_path)

        # 启动协程-当出现延时时自动切换到下一个协程函数继续下载
        gevent.joinall([
            gevent.spawn(downloader, img_path, img_url)
        ])
        k += 1


def downloader(img_path, img_url):
    """协程实现-并发下载

    :param img_path: 写入图片的路径
    :param img_url: 单张图片的URL链接
    :return: None
    """
    # 读取每张图片内容
    req_rb = urllib.request.urlopen(img_url)
    img_content = req_rb.read()
    # 将每张图片内容写入到img_path路径
    with open(img_path, "wb") as f:
        f.write(img_content)


def count_num(start_time):
    """统计抓取的图片张数和爬虫效率"""
    file_count = 0
    for root, dirs, filenames in os.walk(os.getcwd() + "/新浪图集/"):
        for file in filenames:
            if file.endswith(".jpg"):
                file_count += 1
    end_time = time.time()  # 记录结束时间
    execution_time = end_time - start_time
    print("耗时%.2f秒共计抓取图片%d张" % (execution_time, file_count))
    print("平均每秒抓取%.f张图片" % (file_count / execution_time))


def main():
    start_time = time.time()  # 记录开始时间
    try:
        # 1.抓取网站主题返回包含迭代器(清洗后URL主题分类的链接)的闭包
        run = themes()
        # 2.以返回的闭包作为新的入口开启爬虫任务
        run()
    except Exception as e:
        print("Error: %s" % e)
    finally:
        # 3.报告任务结果和爬虫效率
        count_num(start_time)


if __name__ == "__main__":
    main()

2.电子词典项目-分析

    1.功能说明:
        1.用户能够登录注册
            登录凭借用户名和密码即可
            注册 要求用户名和密码 要求用户名不能重复
        2.用户信息需要长期保存
            使用mysql 或者 mongo
        3.要求能够满足多用户同时登录操作的情况
        4.用户开启客户端即进入 登录 注册 退出 界面
        5.用户登录后即进入用户的  查词  查看历史记录  退出界面
            查词: 英英词典  可以循环查询  查一个词会反馈给客户端词义
            历史记录: 查看自己曾经的查词记录返回给用户查过哪些词,用户名,什么时间查的
            退出: 退到上一级界面
        6.单词词义从单词本获取
            每个单词占一行
            单词和解释之间有空格
            单词按顺序排列

    2.思路:
        socket   pymysql   pymongo
        服务器: 注册  登录  查词  历史记录
        客户端: 图形界面打印,提出请求,接受反馈,反馈展示
        技术点: 并发   数据库操作,数据表建立

    3.确定用什么技术
        socket --> tcp
        并发 --> 多进程
        数据库 --> mysql

    4.建立数据表
        数据库: dict
        数据库表:   user   hist  words
        表字段设计:
            user: id    name    passwd
            hist: id    word    time  user_id  # 外键方案
                   id    name    word  time  # 非外键方案
            words: id   word    interpret

    5.框架搭建
        服务端:
            main(): 创建套接字 ---> 父子进程 --->子进程等待客户端请求recv ---> 父进程等待其他客户端链接accept

            child(): 子进程函数接受请求根据请求调用功能函数 quit()
            login(): 操作数据库比对用户密码回复结果
            register(): 插入数据库
            query(): 查询数据库将结果给客户端插入历史记录
            history(): 查询数据库将结果给客户端
        客户端:
            main(): 创建套接字发送链接请求 --->一级界面
                input --->发送请求

            login(): 登录--->进入二级界面
            register(): 注册

            query(): 发送请求  接受反馈  打印
            history(): 发送请求  接受反馈  打印

    6.实现具体功能

3.电子词典项目-实现

    1.项目目录结构

~/Desktop/python3/04_电子词典 $ tree
.
├── client.py
├── dict.txt
├── dict_insert.py
└── server.py

    2.数据库准备

create database dict charset=utf8;
use dict;

create table words(
    id int auto_increment,
    word varchar(32) not null,
    interpret text,
    primary key (id)
);

create table user(
    id int auto_increment primary key,
    name varchar(32) not null,
    passwd varchar(16) not null
    );

create table hist(
    id int auto_increment primary key,
     name varchar(32) not null,
     word varchar(32),
     time varchar(64) not null
    );

    3.插入数据

# dict.txt文件网盘链接: https://pan.baidu.com/s/1dJRwO8__lY9moFkebWdDyg  密码: ehsk
# 下载完命令行解压后放到程序的同级路径下,解压命令: tar -zxvf 05_dict.tar.gz -C ./
import pymysql
import re

f = open('dict.txt')
db = pymysql.connect('localhost', 'root', '123456', 'dict')
cursor = db.cursor()

for line in f:
    l = re.split('[ ]+', line)
    sql = "insert into words (word,interpret) values ('%s','%s')" % (l[0], ' '.join(l[1:]))
    try:
        cursor.execute(sql)
        db.commit()
    except:
        db.rollback()
f.close()

    4.电子词典-服务端

#!/usr/bin/env/ python3
# -*- coding: utf-8 -*-

"""
name: 唐雪成
data: 2020年08月06日21:26:23
email: 979937561@qq.com
MOLULES: python3.7.7 mysql8.0.19 pymysql
This is dict project for temp
"""

import os
import sys
import time
from socket import *
import signal

import pymysql

DICT_TEXT = "./dict.txt"
HOST = "0.0.0.0"
PORT = 7890
ADDR = (HOST, PORT)


def do_child(client_socket, db):
    """实现子进程函数为客服服务"""
    # print("子进程中的连接来自", client_socket.getpeername())  # 测试连接
    # 循环接收请求
    while True:
        data = client_socket.recv(128).decode("utf-8")
        print("Request: ", data)
        if data[0] == "R":  # 注册register
            do_register(client_socket, db, data)
        elif data[0] == "L":  # 登录login
            do_login(client_socket, db, data)
        elif data[0] == "E":  # 退出
            print("%s执行了退出操作" % str(client_socket.getpeername()))
            # 关闭客户端套接字
            client_socket.close()
            # 子进程退出
            sys.exit(0)
        elif data[0] == "Q":  # 查词
            do_query(client_socket, db, data)
        elif data[0] == "H":  # 历史记录
            do_history(client_socket, db, data)


def do_register(client_socket, db, data):
    """实现注册功能"""
    print("%s执行了注册操作" % str(client_socket.getpeername()))
    # 1.提取用户发送的用户名和密码
    temp_list = data.split(" ")  # 切割'R name passwd'字符串
    name = temp_list[1]
    passwd = temp_list[2]

    # 2.查询数据表user中是否已经存在此用户名
    cursor = db.cursor()
    sql = "select * from user where name=%s;"
    cursor.execute(sql, [name])
    res = cursor.fetchone()
    # 2.1用户名已存在则发送信息给用户提示用户名已存在不能使用此用户名注册
    if res is not None:
        client_socket.send(b"EXISTS")
        return
    # 2.2当用户名不存在则向数据库中插入用户信息并提示用户注册成功
    try:
        # 向数据表user中插入用户信息
        sql = "insert into user (name, passwd) values (%s, %s);"
        cursor.execute(sql, [name, passwd])  # 执行插入语句
        db.commit()  # 提交
        client_socket.send(b"OK")  # 向客户端发送注册成功的消息
    except Exception:
        client_socket.send(b"FALL")  # 向客户端发送注册失败的消息
        db.rollback()  # 回滚
        return
    else:
        print("注册成功")


def do_login(client_socket, db, data):
    """实现登录功能"""
    print("%s执行了登录操作" % str(client_socket.getpeername()))
    # 1.提取用户发送的用户名和密码
    temp_list = data.split(" ")  # 切割'L name passwd'字符串
    name = temp_list[1]
    passwd = temp_list[2]

    # 2.查询数据表user中是否存在该用户和密码
    cursor = db.cursor()
    sql = "select * from user where name=%s and passwd=%s;"
    cursor.execute(sql, [name, passwd])
    res = cursor.fetchone()
    if res is None:
        client_socket.send(b"FALL")
    else:
        client_socket.send(b"OK")


def do_query(client_socket, db, data):
    """实现查词功能"""
    print("%s执行了查词操作" % str(client_socket.getpeername()))
    # 1.提取用户发送的用户名和单词
    temp_list = data.split(" ")  # 切割'Q name word'字符串
    # print(temp_list)
    name = temp_list[1]
    word = temp_list[2]
    cursor = db.cursor()

    def insert_history():
        """实现插入查询历史记录功能"""
        temp_time = time.ctime()
        try:
            # 向数据表hist表中插入数据
            sql = "insert into hist (name, word, time) values (%s, %s, %s)"
            cursor.execute(sql, [name, word, temp_time])
            db.commit()
        except Exception:
            db.rollback()
            return

    # 2.1查询本地文本是否存在该单词
    try:
        f = open(DICT_TEXT, "rb")
    except Exception:
        client_socket.send(b"FALL")
        return
    while True:
        line = f.readline().decode("utf-8")
        temp_list = line.split(" ")
        if temp_list[0] > word:  # 词典是按序排列的,大于word的单词后面一定没有要查询的word
            # client_socket.send(b"FALL")
            f.close()
            break  # 本地没有查到直接跳出循环开始从数据库中查询
        if temp_list[0] == word:
            client_socket.send(b"OK")
            time.sleep(0.1)
            line = "数据来源[%s] : %s" % (DICT_TEXT, line)
            client_socket.send(line.encode("utf-8"))
            insert_history()  # 查词成功后向数据库中插入历史记录
            f.close()
            return  # 本地找到单词后直接返回结束函数

    # 2.2查询数据表words中是否存在该单词
    sql = "select * from words where word=%s;"
    cursor.execute(sql, [word])
    res = cursor.fetchone()
    if res is None:
        client_socket.send(b"FALL")
    else:
        client_socket.send(b"OK")
        time.sleep(0.1)
        res = "数据来源[MySQL]: %s	%s" % (res[1], res[2])
        client_socket.send(res.encode("utf-8"))
        insert_history()  # 查词成功后向数据库中插入历史记录
        time.sleep(0.1)


def do_history(client_socket, db, data):
    """实现查询历史记录功能"""
    print("%s执行了历史记录操作" % str(client_socket.getpeername()))
    # 1.提取用户发送的用户名
    temp_list = data.split(" ")  # 切割'H name'字符串
    print(temp_list)
    name = temp_list[1]

    # 2.查询数据表hist中对应用户的历史记录
    cursor = db.cursor()
    sql = "select * from hist where name=%s;"
    cursor.execute(sql, [name])
    res = cursor.fetchall()
    if not res:
        client_socket.send(b"FALL")
    else:
        client_socket.send(b"OK")
    # 3.循环发送历史记录
    for i in res:
        time.sleep(0.1)
        msg = "%s	%s	%s" % (i[1], i[2], i[3])
        client_socket.send(msg.encode("utf-8"))
    time.sleep(0.1)
    client_socket.send(b"##")  # 发送结束标识


def main():
    """完成主逻辑控制"""
    # 连接数据库
    db = pymysql.connect("localhost", "root", "123456", "dict")

    # 创建tcp流式套接字
    tcp_server_socket = socket()
    # 设置端口重用
    tcp_server_socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    # 绑定地址
    tcp_server_socket.bind(ADDR)
    # 开启监听,创建监听队列
    tcp_server_socket.listen(128)
    # 忽略子进程退出,避免僵尸进程
    signal.signal(signal.SIGCHLD, signal.SIG_IGN)

    while True:
        try:
            client_socket, client_addr = tcp_server_socket.accept()
            print("父进程中的连接来自", client_addr)
        except KeyboardInterrupt:
            os._exit(0)
        except Exception:
            continue

        # 创建子进程
        pid = os.fork()
        if pid < 0:
            print("创建子进程失败")
            client_socket.close()
        elif pid == 0:
            tcp_server_socket.close()
            do_child(client_socket, db)
        else:
            client_socket.close()
            continue


if __name__ == "__main__":
    main()

    5.电子词典-客户端

#!/usr/bin/env/ python3
# -*- coding: utf-8 -*-

"""
name: 唐雪成
data: 2020年08月06日21:26:23
email: 979937561@qq.com
MOLULES: python3.7.7 mysql8.0.19 pymysql
This is dict project for temp
"""

import sys
from socket import *
import getpass  # 隐式输入模块


def do_register(tcp_client_socket):
    """实现注册功能"""
    while True:
        name = input("请输入用户名:")
        if name is "":
            return
        passwd = getpass.getpass()
        passwd1 = getpass.getpass("确认密码: ")
        if (" " in name) or (" " in passwd):
            print("用户名或密码不能有空格")
            continue
        if passwd != passwd1:
            print("密码不一致!")
            continue

        # 向服务器发送注册请求
        msg = "R {} {}".format(name, passwd)
        tcp_client_socket.send(msg.encode("utf-8"))
        data = tcp_client_socket.recv(128).decode("utf-8")
        if data == "OK":
            return 0
        elif data == "EXISTS":
            print("用户名已存在")
            return 1
        else:
            return 1


def do_login(tcp_client_socket):
    """实现登录功能"""
    while True:
        name = input("请输入用户名:")
        if name is "":
            return
        passwd = getpass.getpass()
        if (" " in name) or (" " in passwd):
            print("用户名或密码不能有空格")
            continue

        # 向服务器发送登录请求
        msg = "L {} {}".format(name, passwd)
        tcp_client_socket.send(msg.encode("utf-8"))
        data = tcp_client_socket.recv(128).decode("utf-8")
        if data == "OK":
            return name
        else:
            print("用户名或密码不正确")
            return 1


def login(tcp_client_socket, name):
    """实现二级界面功能"""
    print("进入二级界面")
    while True:
        print("""
            ===========查询界面===========
            --1:查词  2:历史记录  3:退出--
            =============================
        """)
        try:
            cmd = int(input("输入选项: "))
        except Exception:
            print("输入的指令错误")
            continue
        if cmd not in [1, 2, 3]:
            print("请输入正确的选项")
            sys.stdin.flush()  # 清除标准输入的缓存
            continue
        elif cmd == 1:
            # 查词功能
            do_query(tcp_client_socket, name)
        elif cmd == 2:
            # 历史记录功能
            do_history(tcp_client_socket, name)
        elif cmd == 3:
            # 退出功能
            return


def do_query(tcp_client_socket, name):
    """实现查词功能"""
    while True:
        word = input("输入要查询的单词(输入##退出): ")
        if word == "##":
            break

        # 向服务器发送用户名和要查询的单词
        msg = "Q {} {}".format(name, word)
        tcp_client_socket.send(msg.encode("utf-8"))
        data = tcp_client_socket.recv(128).decode("utf-8")
        if data == "OK":
            data = tcp_client_socket.recv(2048).decode("utf-8")
            # 找到单词后打印结果
            print(data)
        else:
            print("sorry,本地文件和数据库中都没有此单词")


def do_history(tcp_client_socket, name):
    """实现查看历史记录功能"""
    msg = "H {}".format(name)
    tcp_client_socket.send(msg.encode("utf-8"))
    data = tcp_client_socket.recv(128).decode("utf-8")
    if data == "OK":
        while True:
            data = tcp_client_socket.recv(1024).decode("utf-8")
            if data == "##":
                break
            print(data)
    else:
        print("sorry,没有历史记录")


def main():
    if len(sys.argv) < 3:
        print("argv is error: python3 127.0.0.1 7890")
        return
    HOST = sys.argv[1]
    PORT = int(sys.argv[2])

    # 创建tcp流式套接字
    tcp_client_socket = socket()
    # 连接服务器
    tcp_client_socket.connect((HOST, PORT))

    while True:
        print("""
            ===========欢迎界面===========
            --1:注册  2:用户登录  3:退出--
            =============================
        """)
        try:
            cmd = int(input("输入选项: "))
        except Exception:
            print("输入的指令错误")
            continue
        if cmd not in [1, 2, 3]:
            print("请输入正确的选项")
            sys.stdin.flush()  # 清除标准输入的缓存
            continue
        elif cmd == 1:
            # 调用注册功能
            if do_register(tcp_client_socket) == 0:
                print("注册成功!")
            else:
                print("注册失败!")
        elif cmd == 2:
            # 调用登录功能
            name = do_login(tcp_client_socket)
            if name != 1:
                print("登录成功!")
                login(tcp_client_socket, name)  # 调用二级登录界面功能
            else:
                print("登录失败!")
        elif cmd == 3:
            # 调用退出功能
            tcp_client_socket.send(b"E")
            sys.exit("欢迎再次使用!")


if __name__ == "__main__":
    main()

    6.完整项目网盘链接: https://pan.baidu.com/s/1yykcL9K4VRMsp-_zk-BwYg  密码: 0as4

4.http动态服务器-多线程

    1.项目目录结构:

~/Desktop/Python/05_http动态服务器 $ tree
.
├── __pycache__
│   └── web_framework.cpython-37.pyc
├── http_server.py
├── static
│   ├── abc.html
│   └── index.html
├── web_framework.py
└── wsgiPy

    2.http_server.py文件代码

#!/usr/bin/env/ python3
# -*- coding: utf-8 -*-

"""
name: 唐雪成
data: 2020年08月08日22:09:13
email: 979937561@qq.com
MOLULES: python3.7.7 mysql8.0.19 pymysql
功能: 完成httpServer部分
"""

from socket import *
import sys
import re
from threading import Thread


class HttpServer(object):
    """处理http请求"""

    def __init__(self, app):
        """仅仅封装属性"""
        # 创建tcp套接字
        self.sockfd = socket()
        # 设置端口重用
        self.sockfd.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
        # 将框架应用设置为对象属性
        self.app = app

    def bind(self, addr):
        """将绑定操作封装为方法"""
        # 绑定
        self.sockfd.bind(addr)

    def start(self):
        # 监听
        self.sockfd.listen(128)
        while True:
            # 等待用户连接
            c, addr = self.sockfd.accept()
            print("%s用户连接" % str(addr))
            # 创建线程指定线程函数为客户端服务
            handler_thread = Thread(target=self.handle_client, args=(c,))
            # 启动线程对象
            handler_thread.start()

    def handle_client(self, c):
        """客户端处理函数"""
        # 1.接收浏览器请求
        request_data = c.recv(2048)
        request_lines = request_data.splitlines()
        # 2.获取请求行 'GET /index HTTP/1.1'
        request_line = request_lines[0].decode("utf-8")
        print("=====>1<=====", request_line)
        method = re.match(r"(w+)s+/S*", request_line).group(1)
        filename = re.match(r"w+s+(/S*)", request_line).group(1)

        # 3.将要传递的数据封装到字典中传递给应用程序
        env = {"MEHTOD": method, "PATH_INFO": filename}
        # 4.获取框架的响应体信息 body
        response_body = self.app(env, self.set_headers)
        # 5.拼接完整的响应信息 head + 空行 + body
        response = "%s%s%s" % (self.response_headers, "
", response_body)
        # 6.向客户端发送response信息
        c.send(response.encode("utf-8"))  # 写法二: c.send(bytes(response, "utf-8"))
        # 7.关闭为客服端服务的套接字
        c.close()

    def set_headers(self, status, headers):
        """在app调用该方法时希望得到框架的响应码和响应头
        status: 框架返回的响应码,例如 "200 OK"
        headers: 框架返回的响应头信息,例如 [("Content-Type","text/html"),(...)]
        """
        response_headers = "HTTP/1.1" + status + "
"  # 封装服务器自己特有的响应头信息
        for header in headers:
            response_headers += "%s: %s
" % header
        # 将含有框架响应头和服务器响应头的响应头信息封装为对象的属性
        self.response_headers = response_headers


def main():
    """完成http_server对象属性的添加和创建"""
    # 1.选择要使用网站的某个应用
    if len(sys.argv) < 2:
        sys.exit("运行方式: python3 http_server web_framework.py:app")

    # 2.获取到这个网站下的应用到本地
    module_name, app_name = sys.argv[1].split(":")
    sys.path.insert(1, ".")  # 将当前路径插入到环境变量
    m = __import__(module_name)  # 动态加载框架模块
    app = getattr(m, app_name)

    # 3.将该应用变为http_server对象的属性
    http_server = HttpServer(app)
    http_server.bind(("0.0.0.0", 7890))
    print("Listen on port 7890...")
    http_server.start()


if __name__ == "__main__":
    main()

    2.web_framework.py文件代码

#!/usr/bin/env/ python3
# -*- coding: utf-8 -*-

"""
name: 唐雪成
data: 2020年08月08日22:09:13
email: 979937561@qq.com
MOLULES: python3.7.7 mysql8.0.19 pymysql
功能: 完成后端请求处理服务代码
说明: 模拟web框架的基本原理
"""

import time

# 设置静态文件夹位置
HTML_ROOT_DIR = "static"
# 存放Python方法文件夹
PYTHON_DIR = ""


class Application(object):
    def __init__(self, urls):
        self.urls = urls

    def __call__(self, env, set_headers):
        # /static/index.html: 表示要获取静态文件
        # /time: 表示用Python方法处理请求
        path = env.get("PATH_INFO", "/")
        if path.startswith("/static"):
            # 1.静态网页请求
            file_name = path[7:]
            try:
                fd = open(HTML_ROOT_DIR + file_name, "rb")
            except IOError:
                # 没有找到网页时返回404
                status = "404 Not Fount"
                headers = []
                set_headers(status, headers)
                return "<h1>===Sorry Not Fount The Page===</h1>"
            else:
                file_data = fd.read()
                status = "200 OK"
                headers = [("Content-Type", "text/html"), ("charset", "utf-8")]
                set_headers(status, headers)
                fd.close()
                return file_data.decode("utf-8")
        else:
            # 2.动态网页请求
            for url, hander in self.urls:
                if path == url:
                    return hander(env, set_headers)

            # 没有找到网页时返回404
            status = "404 Not Fount"
            headers = []
            set_headers(status, headers)
            return "<h1>===Sorry url not found===</h1>"


def show_time(env, set_headers):
    status = "200 OK"
    headers = [("Content-Type", "text/html"), ("charset", "utf-8")]
    set_headers(status, headers)
    return "<h1>===%s===</h1>" % time.ctime()


def say_hello(env, set_headers):
    status = "200 OK"
    headers = [("Content-Type", "text/html"), ("charset", "utf-8")]
    set_headers(status, headers)
    return "<h1>===hello world!===</h1>"


def say_bye(env, set_headers):
    status = "200 OK"
    headers = [("Content-Type", "text/html"), ("charset", "utf-8")]
    set_headers(status, headers)
    return "<h1>===Good Bye!===</h1>"


# 存放url的路由映射
urls = {
    ("/time", show_time),
    ("/hello", say_hello),
    ("/bye", say_bye)
}

app = Application(urls)

    3.完整项目网盘链接: https://pan.baidu.com/s/1xbTqyqUZ9zvbzWswEDmi6g  密码: jel0