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=https://www.cnblogs.com/"(.*?)\" 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=https://www.cnblogs.com/".*?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=https://www.cnblogs.com/"(.*?)\" 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=https://www.cnblogs.com/"(.*?)\" 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: [email protected] 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\t%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\t%s\t%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: [email protected] 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 =https://www.cnblogs.com/tangxuecheng/p/= "OK": return 0 elif data =https://www.cnblogs.com/tangxuecheng/p/= "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 =https://www.cnblogs.com/tangxuecheng/p/= "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 =https://www.cnblogs.com/tangxuecheng/p/= "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 =https://www.cnblogs.com/tangxuecheng/p/= "OK": while True: data = tcp_client_socket.recv(1024).decode("utf-8") if data =https://www.cnblogs.com/tangxuecheng/p/= "##": 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: [email protected] 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 = https://www.cnblogs.com/tangxuecheng/p/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, "\r\n", 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 + "\r\n" # 封裝服務器自己特有的回應頭資訊 for header in headers: response_headers += "%s: %s\r\n" % 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: [email protected] 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
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/5069.html
標籤：Python