rewrited logic structure
This commit is contained in:
1
group_system/http/__init__.py
Normal file
1
group_system/http/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
"""HTTP 路由和响应工具。"""
|
||||
358
group_system/http/api.py
Normal file
358
group_system/http/api.py
Normal file
@@ -0,0 +1,358 @@
|
||||
"""学生和教师 API 适配层。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import sqlite3
|
||||
import urllib.parse
|
||||
from http import HTTPStatus
|
||||
from typing import Any
|
||||
|
||||
from group_system.config import BASE_DIR, MAX_JSON_BYTES, MAX_UPLOAD_BYTES
|
||||
from group_system.http.responses import (
|
||||
clear_session_cookie,
|
||||
json_response,
|
||||
make_session_cookie,
|
||||
)
|
||||
from group_system.storage.store import GroupStore
|
||||
|
||||
LOG = logging.getLogger("group-system")
|
||||
|
||||
|
||||
class GroupRequestHandler:
|
||||
"""把 HTTP 请求转换成 GroupStore 调用，不直接访问数据库。"""
|
||||
|
||||
server_version = "StudentGroupSystem/1.0"
|
||||
|
||||
def __init__(self, handler, store: GroupStore):
|
||||
self.handler = handler
|
||||
self.store = store
|
||||
|
||||
def cookies(self) -> dict[str, str]:
|
||||
header = self.handler.headers.get("Cookie", "")
|
||||
result = {}
|
||||
for part in header.split(";"):
|
||||
if "=" in part:
|
||||
key, value = part.strip().split("=", 1)
|
||||
result[key] = urllib.parse.unquote(value)
|
||||
return result
|
||||
|
||||
def token(self) -> str | None:
|
||||
return self.cookies().get("GROUP_SESSION")
|
||||
|
||||
def session(self) -> dict[str, Any] | None:
|
||||
result = self.store.get_session(self.token())
|
||||
if result:
|
||||
result["_token"] = self.token()
|
||||
return result
|
||||
|
||||
def read_json(self) -> dict[str, Any]:
|
||||
length = int(self.handler.headers.get("Content-Length", "0") or 0)
|
||||
if length <= 0:
|
||||
return {}
|
||||
if length > MAX_JSON_BYTES:
|
||||
raise ValueError("请求体过大")
|
||||
raw = self.handler.rfile.read(length)
|
||||
return json.loads(raw.decode("utf-8"))
|
||||
|
||||
def require_teacher(self):
|
||||
session = self.session()
|
||||
if not session or session["role"] != "teacher":
|
||||
return None, json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "请先登录教师后台"})
|
||||
return session, None
|
||||
|
||||
def require_student(self):
|
||||
session = self.session()
|
||||
if not session or session["role"] != "student" or not session["student_id"]:
|
||||
return None, json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": "请先登录"})
|
||||
return session, None
|
||||
|
||||
def handle_api(self, action: str, method: str):
|
||||
try:
|
||||
return self.dispatch_api(action, method)
|
||||
except (ValueError, sqlite3.IntegrityError, sqlite3.OperationalError) as exc:
|
||||
LOG.warning("API error %s %s: %s", method, action, exc)
|
||||
return json_response(HTTPStatus.BAD_REQUEST, {"ok": False, "message": str(exc)})
|
||||
except Exception:
|
||||
LOG.exception("Unexpected API error")
|
||||
return json_response(HTTPStatus.INTERNAL_SERVER_ERROR, {"ok": False, "message": "服务器内部错误"})
|
||||
|
||||
def dispatch_api(self, action: str, method: str):
|
||||
if action == "login" and method == "POST":
|
||||
data = self.read_json()
|
||||
result, message = self.store.login_student(data.get("username", ""), data.get("password", ""))
|
||||
if not result:
|
||||
return json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": message})
|
||||
return json_response(
|
||||
HTTPStatus.OK,
|
||||
{"ok": True, "message": message, "user": result["student"]},
|
||||
[{"value": make_session_cookie(result["token"])}],
|
||||
)
|
||||
|
||||
if action == "teacher-login" and method == "POST":
|
||||
data = self.read_json()
|
||||
result, message = self.store.login_teacher(data.get("username", ""), data.get("password", ""))
|
||||
if not result:
|
||||
return json_response(HTTPStatus.UNAUTHORIZED, {"ok": False, "message": message})
|
||||
return json_response(
|
||||
HTTPStatus.OK,
|
||||
{"ok": True, "message": message, "user": result["teacher"]},
|
||||
[{"value": make_session_cookie(result["token"])}],
|
||||
)
|
||||
|
||||
if action == "logout" and method == "POST":
|
||||
self.store.destroy_session(self.token())
|
||||
return json_response(HTTPStatus.OK, {"ok": True, "message": "已退出"}, [{"value": clear_session_cookie()}])
|
||||
|
||||
if action == "state" and method == "GET":
|
||||
session = self.session()
|
||||
if not session:
|
||||
return json_response(HTTPStatus.OK, {"ok": True, "authenticated": False})
|
||||
active = self.store.active_class()
|
||||
common = {
|
||||
"ok": True,
|
||||
"authenticated": True,
|
||||
"role": session["role"],
|
||||
"active_class": active,
|
||||
"classes": self.store.class_list() if session["role"] == "teacher" else [],
|
||||
}
|
||||
if session["role"] == "teacher":
|
||||
selected = self.handler.path.split("class=", 1)[1].split("&", 1)[0] if "class=" in self.handler.path else active
|
||||
selected = urllib.parse.unquote(selected)
|
||||
if selected and selected != active:
|
||||
# Teacher can preview any class, but active_class remains explicit.
|
||||
return json_response(
|
||||
HTTPStatus.OK,
|
||||
{
|
||||
**common,
|
||||
"view_class": selected,
|
||||
"snapshot": self.store.class_snapshot(selected),
|
||||
},
|
||||
)
|
||||
return json_response(
|
||||
HTTPStatus.OK,
|
||||
{
|
||||
**common,
|
||||
"view_class": active,
|
||||
"teacher": {"username": session["username"]},
|
||||
"snapshot": self.store.class_snapshot(active),
|
||||
},
|
||||
)
|
||||
|
||||
student = self.store.get_student(session["student_id"])
|
||||
if student is None:
|
||||
return json_response(HTTPStatus.OK, {"ok": True, "authenticated": False})
|
||||
membership = self.store.student_membership(student["id"])
|
||||
snapshot = self.store.class_snapshot(student["class_name"])
|
||||
return json_response(
|
||||
HTTPStatus.OK,
|
||||
{
|
||||
**common,
|
||||
"user": {
|
||||
"id": student["id"],
|
||||
"name": student["name"],
|
||||
"student_no": student["student_no"],
|
||||
"class_name": student["class_name"],
|
||||
"must_change_password": bool(student["must_change_password"]),
|
||||
},
|
||||
"membership": membership,
|
||||
"snapshot": snapshot,
|
||||
},
|
||||
)
|
||||
|
||||
if action == "change-password" and method == "POST":
|
||||
session, error = self.require_student()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.change_student_password(
|
||||
session, data.get("old_password", ""), data.get("new_password", "")
|
||||
)
|
||||
status = HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST
|
||||
return json_response(status, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/change-password" and method == "POST":
|
||||
session, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.change_teacher_password(
|
||||
session, data.get("old_password", ""), data.get("new_password", "")
|
||||
)
|
||||
status = HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST
|
||||
return json_response(status, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/group-limit" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.set_group_limit(int(data.get("limit", 0)))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {
|
||||
"ok": ok, "message": message, "group_limit": self.store.group_limit()
|
||||
})
|
||||
|
||||
if action == "groups" and method == "POST":
|
||||
session, error = self.require_student()
|
||||
if error:
|
||||
return error
|
||||
student = self.store.get_student(session["student_id"])
|
||||
if student and student["must_change_password"]:
|
||||
return json_response(HTTPStatus.FORBIDDEN, {"ok": False, "message": "首次登录必须先修改密码"})
|
||||
data = self.read_json()
|
||||
ok, message, _ = self.store.create_group(session["student_id"], data.get("name", ""))
|
||||
return json_response(
|
||||
HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST,
|
||||
{"ok": ok, "message": message},
|
||||
)
|
||||
|
||||
if action == "join" and method == "POST":
|
||||
session, error = self.require_student()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.join_group(
|
||||
session["student_id"], int(data.get("group_id", 0)), bool(data.get("confirmed"))
|
||||
)
|
||||
return json_response(
|
||||
HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST,
|
||||
{"ok": ok, "message": message, "require_confirm": (not ok and "确认后才会执行" in message)},
|
||||
)
|
||||
|
||||
if action == "leave" and method == "POST":
|
||||
session, error = self.require_student()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.leave_group(session["student_id"], bool(data.get("confirmed")))
|
||||
return json_response(
|
||||
HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST,
|
||||
{"ok": ok, "message": message, "require_confirm": (not ok and "确认后才会执行" in message)},
|
||||
)
|
||||
|
||||
if action == "groups/remove-member" and method == "POST":
|
||||
session, error = self.require_student()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.leader_remove_member(
|
||||
session["student_id"], int(data.get("student_id", 0))
|
||||
)
|
||||
return json_response(
|
||||
HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST,
|
||||
{"ok": ok, "message": message},
|
||||
)
|
||||
|
||||
if action == "teacher/class" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.switch_class(str(data.get("class_name", "")))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/lock" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
class_name = str(data.get("class_name", self.store.active_class()))
|
||||
ok, message = self.store.set_locked(class_name, bool(data.get("locked")))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/create-group" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.teacher_create_group(
|
||||
str(data.get("class_name", self.store.active_class())), data.get("name", "")
|
||||
)
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/rename-group" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.teacher_rename_group(int(data.get("group_id", 0)), data.get("name", ""))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/set-leader" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
student_id = data.get("student_id")
|
||||
ok, message = self.store.teacher_set_leader(
|
||||
int(data.get("group_id", 0)), int(student_id) if student_id else None
|
||||
)
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/add-member" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.teacher_add_member(int(data.get("group_id", 0)), int(data.get("student_id", 0)))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/remove-member" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.teacher_remove_member(int(data.get("student_id", 0)))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/disband" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
data = self.read_json()
|
||||
ok, message = self.store.teacher_disband_group(int(data.get("group_id", 0)))
|
||||
return json_response(HTTPStatus.OK if ok else HTTPStatus.BAD_REQUEST, {"ok": ok, "message": message})
|
||||
|
||||
if action == "teacher/import" and method == "POST":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
length = int(self.handler.headers.get("Content-Length", "0") or 0)
|
||||
if length <= 0 or length > MAX_UPLOAD_BYTES:
|
||||
return json_response(HTTPStatus.BAD_REQUEST, {"ok": False, "message": "Excel 文件大小无效"})
|
||||
raw = self.handler.rfile.read(length)
|
||||
temp_path = BASE_DIR / ".upload-roster.xlsx"
|
||||
temp_path.write_bytes(raw)
|
||||
try:
|
||||
student_count, class_count = self.store.import_roster(temp_path)
|
||||
finally:
|
||||
temp_path.unlink(missing_ok=True)
|
||||
return json_response(
|
||||
HTTPStatus.OK,
|
||||
{"ok": True, "message": f"导入成功：{student_count} 名学生，{class_count} 个班。所有原分组已重置。"},
|
||||
)
|
||||
|
||||
if action == "teacher/export" and method == "GET":
|
||||
_, error = self.require_teacher()
|
||||
if error:
|
||||
return error
|
||||
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.handler.path).query)
|
||||
class_name = query.get("class", [None])[0]
|
||||
if class_name:
|
||||
class_name = urllib.parse.unquote(class_name)
|
||||
content, filename = self.store.export_groups(class_name)
|
||||
|
||||
def send(handler):
|
||||
handler.send_response(HTTPStatus.OK)
|
||||
handler.send_header("Content-Type", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
|
||||
quoted = urllib.parse.quote(filename)
|
||||
handler.send_header("Content-Disposition", f"attachment; filename*=UTF-8''{quoted}")
|
||||
handler.send_header("Content-Length", str(len(content)))
|
||||
handler.send_header("Cache-Control", "no-store")
|
||||
handler.end_headers()
|
||||
handler.wfile.write(content)
|
||||
|
||||
return send
|
||||
|
||||
return json_response(HTTPStatus.NOT_FOUND, {"ok": False, "message": "接口不存在"})
|
||||
41
group_system/http/responses.py
Normal file
41
group_system/http/responses.py
Normal file
@@ -0,0 +1,41 @@
|
||||
"""HTTP 响应和 Cookie 工具。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from http import HTTPStatus
|
||||
from http.server import BaseHTTPRequestHandler
|
||||
from typing import Any
|
||||
|
||||
from group_system.config import SESSION_DAYS
|
||||
|
||||
|
||||
def json_response(
|
||||
status: int | HTTPStatus,
|
||||
payload: Any,
|
||||
cookies: list[dict[str, Any]] | None = None,
|
||||
):
|
||||
"""Return a callable that serializes and sends a JSON response."""
|
||||
def send(handler: BaseHTTPRequestHandler):
|
||||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||||
handler.send_response(int(status))
|
||||
handler.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
handler.send_header("Cache-Control", "no-store")
|
||||
handler.send_header("Content-Length", str(len(body)))
|
||||
for cookie in cookies or []:
|
||||
handler.send_header("Set-Cookie", cookie["value"])
|
||||
handler.end_headers()
|
||||
handler.wfile.write(body)
|
||||
|
||||
return send
|
||||
|
||||
|
||||
def make_session_cookie(token: str) -> str:
|
||||
return (
|
||||
f"GROUP_SESSION={token}; Path=/; HttpOnly; SameSite=Lax; "
|
||||
f"Max-Age={SESSION_DAYS * 86400}"
|
||||
)
|
||||
|
||||
|
||||
def clear_session_cookie() -> str:
|
||||
return "GROUP_SESSION=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0"
|
||||
138
group_system/http/server.py
Normal file
138
group_system/http/server.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""HTTP 服务器、静态路由和应用启动流程。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import http.server
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
import urllib.parse
|
||||
from http import HTTPStatus
|
||||
from pathlib import Path
|
||||
|
||||
from group_system.config import STATIC_DIR, XLSX_PATH
|
||||
from group_system.http.api import GroupRequestHandler
|
||||
from group_system.http.static_files import send_text, serve_file
|
||||
from group_system.storage.store import GroupStore
|
||||
|
||||
LOG = logging.getLogger("group-system")
|
||||
|
||||
|
||||
class GroupHTTPServer(http.server.ThreadingHTTPServer):
|
||||
"""Carry the application store on the standard threading HTTP server."""
|
||||
|
||||
def __init__(self, address: tuple[str, int], store: GroupStore):
|
||||
super().__init__(address, RequestHandler)
|
||||
self.store = store
|
||||
|
||||
|
||||
class RequestHandler(http.server.BaseHTTPRequestHandler):
|
||||
@property
|
||||
def store(self) -> GroupStore:
|
||||
return self.server.store # type: ignore[attr-defined]
|
||||
|
||||
def do_GET(self) -> None:
|
||||
route(self)
|
||||
|
||||
def do_POST(self) -> None:
|
||||
route(self)
|
||||
|
||||
def do_HEAD(self) -> None:
|
||||
route(self)
|
||||
|
||||
def log_message(self, format_string: str, *args) -> None:
|
||||
LOG.info("%s - %s", self.address_string(), format_string % args)
|
||||
|
||||
|
||||
def route(handler: RequestHandler) -> None:
|
||||
path = urllib.parse.urlparse(handler.path).path
|
||||
if path.startswith("/api/"):
|
||||
request = GroupRequestHandler(handler, handler.store)
|
||||
result = request.handle_api(path[len("/api/"):], handler.command)
|
||||
if callable(result):
|
||||
result(handler)
|
||||
return
|
||||
|
||||
if path in ("/", "/index.html", "/student"):
|
||||
serve_file(handler, STATIC_DIR / "index.html")
|
||||
return
|
||||
if path in ("/teacher", "/admin"):
|
||||
serve_file(handler, STATIC_DIR / "teacher.html")
|
||||
return
|
||||
|
||||
static_root = STATIC_DIR.resolve()
|
||||
safe_path = (static_root / urllib.parse.unquote(path).lstrip("/")).resolve()
|
||||
try:
|
||||
safe_path.relative_to(static_root)
|
||||
except ValueError:
|
||||
send_text(handler, HTTPStatus.NOT_FOUND, "Not Found")
|
||||
return
|
||||
serve_file(handler, safe_path)
|
||||
|
||||
|
||||
def local_ipv4_addresses() -> list[str]:
|
||||
addresses: set[str] = set()
|
||||
try:
|
||||
host_name = socket.gethostname()
|
||||
for item in socket.getaddrinfo(host_name, None, socket.AF_INET):
|
||||
address = item[4][0]
|
||||
if address != "127.0.0.1":
|
||||
addresses.add(address)
|
||||
except socket.gaierror:
|
||||
pass
|
||||
|
||||
# A UDP connection does not send packets here, but reveals the default-route address.
|
||||
try:
|
||||
probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
try:
|
||||
probe.connect(("192.0.2.1", 9))
|
||||
address = probe.getsockname()[0]
|
||||
if address != "127.0.0.1":
|
||||
addresses.add(address)
|
||||
finally:
|
||||
probe.close()
|
||||
except OSError:
|
||||
pass
|
||||
return sorted(addresses)
|
||||
|
||||
|
||||
def bootstrap_store(store: GroupStore, xlsx_path: Path) -> None:
|
||||
"""Import the initial roster only when the database is still empty."""
|
||||
if any(store.class_list()):
|
||||
return
|
||||
if not xlsx_path.exists():
|
||||
LOG.warning("未找到 %s，请先通过教师后台上传学生名单", xlsx_path)
|
||||
return
|
||||
student_count, class_count = store.import_roster(xlsx_path)
|
||||
LOG.info("已导入 %s： %d 名学生 / %d 个班", xlsx_path, student_count, class_count)
|
||||
|
||||
|
||||
def run(
|
||||
store: GroupStore,
|
||||
xlsx_path: Path = XLSX_PATH,
|
||||
host: str | None = None,
|
||||
port: int | None = None,
|
||||
) -> None:
|
||||
bootstrap_store(store, xlsx_path)
|
||||
host = host or os.environ.get("HOST", "0.0.0.0")
|
||||
port = port if port is not None else int(os.environ.get("PORT", "8000"))
|
||||
server = GroupHTTPServer((host, port), store)
|
||||
LOG.info("学生分组系统已启动：http://0.0.0.0:%d", port)
|
||||
addresses = local_ipv4_addresses()
|
||||
if addresses:
|
||||
for address in addresses:
|
||||
LOG.info(
|
||||
"学生入口：http://%s:%d/ 教师入口：http://%s:%d/teacher",
|
||||
address,
|
||||
port,
|
||||
address,
|
||||
port,
|
||||
)
|
||||
else:
|
||||
LOG.warning("未能自动获取局域网 IP，请在教师机上查看网络设置；端口：%d", port)
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
LOG.info("服务停止")
|
||||
finally:
|
||||
server.server_close()
|
||||
43
group_system/http/static_files.py
Normal file
43
group_system/http/static_files.py
Normal file
@@ -0,0 +1,43 @@
|
||||
"""静态页面与资源文件服务。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
from http.server import BaseHTTPRequestHandler
|
||||
from pathlib import Path
|
||||
|
||||
from group_system.config import STATIC_DIR
|
||||
|
||||
CONTENT_TYPES = {
|
||||
".html": "text/html; charset=utf-8",
|
||||
".htm": "text/html; charset=utf-8",
|
||||
".css": "text/css; charset=utf-8",
|
||||
".js": "text/javascript; charset=utf-8",
|
||||
".mjs": "text/javascript; charset=utf-8",
|
||||
".svg": "image/svg+xml",
|
||||
".png": "image/png",
|
||||
}
|
||||
|
||||
|
||||
def send_text(handler: BaseHTTPRequestHandler, status: int, text: str) -> None:
|
||||
body = text.encode("utf-8")
|
||||
handler.send_response(status)
|
||||
handler.send_header("Content-Type", "text/plain; charset=utf-8")
|
||||
handler.send_header("Content-Length", str(len(body)))
|
||||
handler.end_headers()
|
||||
handler.wfile.write(body)
|
||||
|
||||
|
||||
def serve_file(handler: BaseHTTPRequestHandler, path: Path) -> None:
|
||||
path = path if path.is_absolute() else STATIC_DIR / path
|
||||
if not path.is_file():
|
||||
send_text(handler, HTTPStatus.NOT_FOUND, "Not Found")
|
||||
return
|
||||
content_type = CONTENT_TYPES.get(path.suffix.lower(), "application/octet-stream")
|
||||
content = path.read_bytes()
|
||||
handler.send_response(HTTPStatus.OK)
|
||||
handler.send_header("Content-Type", content_type)
|
||||
handler.send_header("Content-Length", str(len(content)))
|
||||
handler.send_header("Cache-Control", "no-cache")
|
||||
handler.end_headers()
|
||||
handler.wfile.write(content)
|
||||
Reference in New Issue
Block a user