1. Cơ chế Định tuyến URL trong Framework
Để xây dựng một framework web linh hoạt, việc ánh xạ URL đến các hàm xử lý cụ thể là bước đầu tiên quan trọng. Thay vì sử dụng câu lệnh điều kiện rẽ nhánh phức tạp, chúng ta có thể sử dụng từ điển (dictionary) để lưu trữ mối quan hệ này và một decorator để đăng ký route.
Tệp: dynamic/my_web.py
import time
import os
import re
VIEW_DIR = "./templates"
# Bảng định tuyến toàn cục
route_map = {}
def map_url(url_path):
"""Decorator để đăng ký URL vào bảng định tuyến"""
def register_handler(handler_func):
route_map[url_path] = handler_func
def wrapper(file_name):
return handler_func(file_name)
return wrapper
return register_handler
@map_url("/index.html")
def render_home(file_name):
"""Xử lý yêu cầu cho trang chủ"""
try:
target_file = file_name.replace(".py", ".html")
with open(VIEW_DIR + target_file, "r") as f:
content = f.read()
# Dữ liệu giả lập tạm thời
mock_data = "Đang chờ kết nối cơ sở dữ liệu..."
content = re.sub(r"\{%content%\}", mock_data, content)
return content
except Exception as e:
return str(e)
@map_url("/center.html")
def render_stock_info(file_name):
"""Xử lý yêu cầu cho trang danh sách cổ phiếu"""
try:
target_file = file_name.replace(".py", ".html")
with open(VIEW_DIR + target_file, "r") as f:
content = f.read()
mock_data = "Chưa có dữ liệu hiển thị,,,,~~~~(>_<)~~~~ "
content = re.sub(r"\{%content%\}", mock_data, content)
return content
except Exception as e:
return str(e)
def wsgi_handler(environ, start_response):
"""Hàm entry point chuẩn WSGI"""
status = '200 OK'
headers = [('Content-Type', 'text/html')]
start_response(status, headers)
path_info = environ['PATH_INFO']
try:
return route_map[path_info](path_info)
except Exception as e:
return "Lỗi định tuyến: %s" % e
2. Phân biệt URL Tĩnh, Động và Giả tĩnh
Trong phát triển web, cấu trúc URL ảnh hưởng đến hiệu suất và khả năng tối ưu hóa công cụ tìm kiếm (SEO). Chúng ta thường gặp ba loại URL chính:
URL Tĩnh (Static URL)
Ví dụ: domain.com/news/2023-05-18/110.html. Đây là các tệp tin vật lý thực sự tồn tại trên máy chủ.
- Ưu điểm: Tốc độ truy cập nhanh do không cần xử lý logic, thân thiện với bộ nhớ đệm trình duyệt.
- Nhược điểm: Khó quản lý khi số lượng trang lớn, khó cập nhật nội dung đồng loạt.
URL Động (Dynamic URL)
Ví dụ: domain.com/NewsMore.asp?id=5. URL này trỏ đến một script xử lý logic, không tồn tại dưới dạng tệp tin vật lý.
- Ưu điểm: Linh hoạt, dễ dàng quản lý nội dung qua cơ sở dữ liệu, phù hợp web lớn.
- Nhược điểm: Cấu trúc URL phức tạp, có thể ảnh hưởng nhẹ đến SEO nếu không cấu hình đúng (tuy nhiên các công cụ tìm kiếm hiện đại đã xử lý tốt).
URL Giả tĩnh (Pseudo-static URL)
Ví dụ: domain.com/course/74.html. Về bản chất vẫn là URL động nhưng được viết lại (rewrite) để trông giống URL tĩnh.
- Ưu điểm: Kết hợp được lợi ích về mặt hiển thị của URL tĩnh và sự linh hoạt của URL động.
- Nhược điểm: Cần cấu hình server hỗ trợ rewrite rule, tăng tải nhẹ cho server do phải xử lý quy tắc viết lại.
3. Cấu hình Server hỗ trợ URL Giả tĩnh
Để framework có thể xử lý các yêu cầu có đuôi `.html` như là các script động, chúng ta cần điều chỉnh logic trong server WSGI. Nếu yêu cầu không phải là `.html`, server sẽ trả về file tĩnh. Nếu là `.html`, server sẽ gọi ứng dụng web.
Tệp: web_server.py (Cập nhật)
import select
import time
import socket
import sys
import re
import multiprocessing
class HTTPService(object):
"""Máy chủ Web đơn giản hỗ trợ WSGI"""
def __init__(self, port, static_path, app):
self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self.server_socket.bind(("", port))
self.server_socket.listen(128)
self.static_path = static_path
self.app = app
def run_forever(self):
while True:
client_socket, client_addr = self.server_socket.accept()
client_socket.settimeout(3)
process = multiprocessing.Process(target=self.handle_client, args=(client_socket,))
process.start()
client_socket.close()
def handle_client(self, client_socket):
while True:
try:
request_data = client_socket.recv(1024).decode("utf-8")
except Exception as e:
client_socket.close()
return
if not request_data:
client_socket.close()
return
lines = request_data.splitlines()
# Phân tích dòng yêu cầu HTTP
match = re.match(r"([^/]*)([^ ]+)", lines[0])
if match:
file_path = match.group(2)
if file_path == "/":
file_path = "/index.html"
# Phân loại yêu cầu tĩnh hay động
if not file_path.endswith(".html"):
# Xử lý tài nguyên tĩnh
try:
with open(self.static_path + file_path, "rb") as f:
content = f.read()
header = "HTTP/1.1 200 OK\r\n"
header += "Content-Length: %d\r\n" % len(content)
header += "\r\n"
client_socket.send(header.encode('utf-8') + content)
except:
error_body = "404 Not Found"
header = "HTTP/1.1 404 Not Found\r\n"
header += "Content-Type: text/html; charset=utf-8\r\n"
header += "Content-Length: %d\r\n" % len(error_body)
header += "\r\n"
client_socket.send((header + error_body).encode('utf-8'))
else:
# Xử lý yêu cầu động (giả tĩnh)
env = dict()
env['PATH_INFO'] = file_path
response_body = self.app(env, self.set_headers)
status_line = "HTTP/1.1 {status}\r\n".format(status=self.headers[0])
response_header = status_line
response_header += "Content-Type: text/html; charset=utf-8\r\n"
response_header += "Content-Length: %d\r\n" % len(response_body.encode("utf-8"))
for key, val in self.headers[1]:
response_header += "{0}:{1}\r\n".format(key, val)
response_header += "\r\n"
client_socket.send((response_header + response_body).encode('utf-8'))
def set_headers(self, status, headers):
default_headers = [
("Date", time.time()),
("Server", "Custom Python Web Server")
]
self.headers = [status, default_headers + headers]
STATIC_ROOT = "./static"
DYNAMIC_ROOT = "./dynamic"
def main():
if len(sys.argv) == 3:
port = int(sys.argv[1])
module_app = sys.argv[2]
else:
print("Usage: python3 server.py 8080 module:app")
return
sys.path.append(DYNAMIC_ROOT)
match = re.match(r"([^:]*):(.*)", module_app)
if match:
module_name = match.group(1)
app_name = match.group(2)
module = __import__(module_name)
app = getattr(module, app_name)
server = HTTPService(port, STATIC_ROOT, app)
server.run_forever()
if __name__ == "__main__":
main()
Tệp: my_web.py (Điều chỉnh route)
import time
import os
import re
VIEW_DIR = "./templates"
route_map = {}
def map_url(url_path):
def register_handler(handler_func):
route_map[url_path] = handler_func
def wrapper(file_name):
return handler_func(file_name)
return wrapper
return register_handler
@map_url("/index.html")
def render_home(file_name):
try:
target_file = file_name.replace(".py", ".html")
with open(VIEW_DIR + target_file, "r") as f:
content = f.read()
mock_data = "Đang chờ kết nối cơ sở dữ liệu..."
content = re.sub(r"\{%content%\}", mock_data, content)
return content
except Exception as e:
return str(e)
@map_url("/center.html")
def render_stock_info(file_name):
try:
target_file = file_name.replace(".py", ".html")
with open(VIEW_DIR + target_file, "r") as f:
content = f.read()
mock_data = "Chưa có dữ liệu hiển thị,,,,~~~~(>_<)~~~~ "
content = re.sub(r"\{%content%\}", mock_data, content)
return content
except Exception as e:
return str(e)
def wsgi_handler(environ, start_response):
status = '200 OK'
headers = [('Content-Type', 'text/html')]
start_response(status, headers)
path_info = environ['PATH_INFO']
try:
return route_map[path_info](path_info)
except Exception as e:
return "Lỗi định tuyến: %s" % e
4. Thiết lập Cơ sở dữ liệu Chứng khoán
Để hiển thị dữ liệu thực tế, chúng ta cần chuẩn bị môi trường MySQL. Dưới đây là các bước khởi tạo database và bảng dữ liệu mẫu.
Các lệnh SQL
-- Tạo cơ sở dữ liệu
create database finance_db charset=utf8;
-- Chọn cơ sở dữ liệu
use finance_db;
-- Import dữ liệu từ file script (giả sử đã có file sql)
source stock_data.sql;
Cấu trúc bảng
Hệ thống gồm hai bảng chính: stock_info lưu thông tin cổ phiếu và watchlist lưu danh sách theo dõi.
mysql> desc watchlist;
+-----------+------------------+------+-----+---------+----------------+
| Field | Type | Null | Key | Default | Extra |
+-----------+------------------+------+-----+---------+----------------+
| id | int(10) unsigned | NO | PRI | NULL | auto_increment |
| note_info | varchar(200) | YES | | | |
| info_id | int(10) unsigned | YES | MUL | NULL | |
+-----------+------------------+------+-----+---------+----------------+
mysql> desc stock_info;
+----------+------------------+------+-----+---------+----------------+
| Field | Type | Null | Key | Default | Extra |
+----------+------------------+------+-----+---------+----------------+
| id | int(10) unsigned | NO | PRI | NULL | auto_increment |
| code | varchar(6) | NO | | NULL | |
| short | varchar(10) | NO | | NULL | |
| chg | varchar(10) | NO | | NULL | |
| turnover | varchar(255) | NO | | NULL | |
| price | decimal(10,2) | NO | | NULL | |
| highs | decimal(10,2) | NO | | NULL | |
| time | date | YES | | NULL | |
+----------+------------------+------+-----+---------+----------------+
5. Truy vấn và Hiển thị Dữ liệu từ MySQL
Sau khi có database, chúng ta cập nhật logic trong my_web.py để kết nối MySQL, truy vấn dữ liệu và динамically inject vào template HTML.
Tệp: my_web.py (Hoàn thiện)
import pymysql
import time
import os
import re
VIEW_DIR = "./templates"
route_map = {}
def map_url(url_path):
def register_handler(handler_func):
route_map[url_path] = handler_func
def wrapper(file_name):
return handler_func(file_name)
return wrapper
return register_handler
@map_url("/index.html")
def render_home(file_name):
try:
target_file = file_name.replace(".py", ".html")
with open(VIEW_DIR + target_file, "r") as f:
content = f.read()
# Kết nối và truy vấn dữ liệu
db_conn = pymysql.connect(host='localhost', port=3306, user='root', password='mysql', database='finance_db', charset='utf8')
cursor_obj = db_conn.cursor()
query = """select * from stock_info;"""
cursor_obj.execute(query)
result_data = cursor_obj.fetchall()
cursor_obj.close()
db_conn.close()
# Tạo bảng HTML từ dữ liệu
row_template = """
<tr>
<td>%d</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="Thêm" id="btnAdd" name="btnAdd" data-id="%s">
</td>
</tr>"""
html_rows = ""
for row in result_data:
html_rows += row_template % (row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7], row[1])
content = re.sub(r"\{%content%\}", html_rows, content)
return content
except Exception as e:
return str(e)
@map_url("/center.html")
def render_stock_info(file_name):
try:
target_file = file_name.replace(".py", ".html")
with open(VIEW_DIR + target_file, "r") as f:
content = f.read()
# Truy vấn dữ liệu đã theo dõi (Join 2 bảng)
db_conn = pymysql.connect(host='localhost', port=3306, user='root', password='mysql', database='finance_db', charset='utf8')
cursor_obj = db_conn.cursor()
query = """select i.code, i.short, i.chg, i.turnover, i.price, i.highs, j.note_info
from stock_info as i
inner join watchlist as j on i.id = j.info_id;"""
cursor_obj.execute(query)
result_data = cursor_obj.fetchall()
cursor_obj.close()
db_conn.close()
row_template = """
<tr>
<td>%s</td>
<td>%s</td>
<td>%s</td>
<td>%s</td>
<td>%s</td>
<td>%s</td>
<td>%s</td>
<td>
<a class="btn btn-default btn-xs" href="/update/%s.html"> Sửa </a>
</td>
<td>
<input type="button" value="Xóa" id="btnDel" name="btnDel" data-id="%s">
</td>
</tr>
"""
html_rows = ""
for row in result_data:
html_rows += row_template % (row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[0], row[0])
content = re.sub(r"\{%content%\}", html_rows, content)
return content
except Exception as e:
return str(e)
def wsgi_handler(environ, start_response):
status = '200 OK'
headers = [('Content-Type', 'text/html')]
start_response(status, headers)
path_info = environ['PATH_INFO']
try:
return route_map[path_info](path_info)
except Exception as e:
return "Lỗi định tuyến: %s" % e