1196 lines
61 KiB
Python
1196 lines
61 KiB
Python
import base64
|
|
import hashlib
|
|
import hmac
|
|
import io
|
|
import json
|
|
import random
|
|
import secrets
|
|
import sqlite3
|
|
import time
|
|
import traceback
|
|
from datetime import date, datetime
|
|
from datetime import timedelta
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, unquote, urlparse
|
|
|
|
from openpyxl import Workbook, load_workbook
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
from reportlab.lib.pagesizes import A4, landscape
|
|
from reportlab.pdfbase import pdfmetrics
|
|
from reportlab.pdfbase.ttfonts import TTFont
|
|
from reportlab.pdfgen import canvas
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
DATA = ROOT / "data"
|
|
DB_PATH = DATA / "local.db"
|
|
SECRET = "local-dev-secret"
|
|
ALPHABET = "23456789ABCDEFGHJKLMNPQRSTUVWXYZ"
|
|
CAPTCHAS: dict[str, tuple[str, float]] = {}
|
|
SMS_CODES: dict[str, tuple[str, float]] = {}
|
|
LOG_RETENTION_DAYS = 180
|
|
HIGH_RISK_LOG_RETENTION_DAYS = 365
|
|
DESIGN_WIDTH = 1024
|
|
DESIGN_HEIGHT = 759
|
|
PDF_BASE_RESOLUTION = 150.0
|
|
|
|
COL_NAME = "\u59d3\u540d"
|
|
COL_PHONE = "\u624b\u673a\u53f7"
|
|
COL_PROJECT = "\u9879\u76ee\u4ee3\u7801"
|
|
COL_ISSUE_DATE = "\u53d1\u8bc1\u65e5\u671f"
|
|
COLS = [COL_NAME, COL_PHONE, COL_PROJECT, COL_ISSUE_DATE]
|
|
DISCLAIMER = "\u672c\u8bc1\u4e66\u4ec5\u7528\u4e8e\u8bc1\u660e\u5b66\u5458\u5b8c\u6210\u76f8\u5173\u57f9\u8bad\u8bfe\u7a0b/\u9636\u6bb5\u5b66\u4e60\uff0c\u4e0d\u4ee3\u8868\u56fd\u5bb6\u5b66\u5386\u3001\u5b66\u4f4d\u3001\u804c\u4e1a\u8d44\u683c\u6216\u804c\u4e1a\u6280\u80fd\u7b49\u7ea7\u8ba4\u8bc1\u3002"
|
|
|
|
ACTION_LABELS = {
|
|
"create_project": "\u65b0\u589e\u9879\u76ee",
|
|
"update_project": "\u4fee\u6539\u9879\u76ee",
|
|
"create_learner": "\u65b0\u589e\u5b66\u5458",
|
|
"update_learner": "\u4fee\u6539\u5b66\u5458",
|
|
"delete_learner": "\u5220\u9664\u5b66\u5458",
|
|
"create_certificate": "\u65b0\u589e\u8bc1\u4e66",
|
|
"void_certificate": "\u4f5c\u5e9f\u8bc1\u4e66",
|
|
"regenerate_public_token": "\u91cd\u7f6e\u8bc1\u4e66\u94fe\u63a5",
|
|
"upload_import_file": "\u4e0a\u4f20\u5bfc\u5165\u6587\u4ef6",
|
|
"confirm_import_batch": "\u786e\u8ba4\u5bfc\u5165",
|
|
"delete_import_batch": "\u5220\u9664\u5bfc\u5165\u6279\u6b21",
|
|
"public_search_certificate": "\u516c\u5f00\u67e5\u8be2\u8bc1\u4e66",
|
|
"public_download_certificate": "\u516c\u5f00\u4e0b\u8f7d\u8bc1\u4e66",
|
|
}
|
|
OBJECT_LABELS = {
|
|
"project": "\u9879\u76ee",
|
|
"project_course": "\u9879\u76ee",
|
|
"learner": "\u5b66\u5458",
|
|
"certificate": "\u8bc1\u4e66",
|
|
"import_batch": "\u5bfc\u5165\u6279\u6b21",
|
|
"public_access": "\u516c\u5f00\u8bbf\u95ee",
|
|
}
|
|
HIGH_RISK_ACTIONS = {"void_certificate", "delete_import_batch", "regenerate_public_token", "delete_learner"}
|
|
MEDIUM_RISK_ACTIONS = {"update_project", "update_learner", "confirm_import_batch", "create_certificate"}
|
|
|
|
|
|
def db() -> sqlite3.Connection:
|
|
DATA.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def init_db() -> None:
|
|
with db() as conn:
|
|
conn.executescript(
|
|
"""
|
|
create table if not exists admins(id integer primary key, username text unique, password_hash text);
|
|
create table if not exists projects(
|
|
id integer primary key, code text unique, name text,
|
|
default_certificate_name text, default_course_name text, default_stage_name text, default_issuer_name text,
|
|
status text
|
|
);
|
|
create table if not exists learners(id integer primary key, phone text unique, current_name text, student_no text, status text, remark text, created_at text, updated_at text);
|
|
create table if not exists certificates(
|
|
id integer primary key, learner_id integer, project_code text, certificate_no text unique, certificate_name text,
|
|
course_name text, stage_name text, issue_date text, issuer_name text, status text, pdf_status text,
|
|
public_token text, qr_token text, remark text, created_at text, updated_at text
|
|
);
|
|
create table if not exists import_batches(id integer primary key, filename text, status text, total_rows integer, valid_rows integer, failed_rows integer, error_report_path text, created_at text, updated_at text);
|
|
create table if not exists logs(id integer primary key, admin_user_id integer, action text, object_type text, object_id text, detail_json text, created_at text);
|
|
"""
|
|
)
|
|
if not conn.execute("select 1 from admins where username='admin'").fetchone():
|
|
conn.execute("insert into admins(username,password_hash) values(?,?)", ("admin", password_hash("Admin123!")))
|
|
ensure_project_columns(conn)
|
|
for code, name in [("DBY", "\u5927\u672c\u8425"), ("QSX", "\u4e03\u4e09\u7ebf")]:
|
|
if not conn.execute("select 1 from projects where code=?", (code,)).fetchone():
|
|
conn.execute(
|
|
"insert into projects(code,name,default_certificate_name,default_course_name,default_stage_name,default_issuer_name,status) values(?,?,?,?,?,?,?)",
|
|
(code, name, name, name, None, "\u672c\u516c\u53f8", "active"),
|
|
)
|
|
conn.execute("update projects set default_certificate_name=name, default_course_name=coalesce(default_course_name,name), default_issuer_name=coalesce(default_issuer_name,?)", ("\u672c\u516c\u53f8",))
|
|
cleanup_expired_logs(conn)
|
|
|
|
|
|
def ensure_project_columns(conn: sqlite3.Connection) -> None:
|
|
columns = {row["name"] for row in conn.execute("pragma table_info(projects)").fetchall()}
|
|
for name in ["default_certificate_name", "default_course_name", "default_stage_name", "default_issuer_name"]:
|
|
if name not in columns:
|
|
conn.execute(f"alter table projects add column {name} text")
|
|
|
|
|
|
def password_hash(password: str) -> str:
|
|
salt = secrets.token_hex(8)
|
|
digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 120_000).hex()
|
|
return f"{salt}${digest}"
|
|
|
|
|
|
def check_password(password: str, stored: str) -> bool:
|
|
salt, digest = stored.split("$", 1)
|
|
actual = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 120_000).hex()
|
|
return hmac.compare_digest(actual, digest)
|
|
|
|
|
|
def make_token(admin_id: int) -> str:
|
|
payload = f"{admin_id}:{int(time.time()) + 86400}:{secrets.token_hex(8)}"
|
|
sig = hmac.new(SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest()
|
|
return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode()
|
|
|
|
|
|
def token_admin_id(token: str) -> int | None:
|
|
try:
|
|
raw = base64.urlsafe_b64decode(token.encode()).decode()
|
|
admin_id, exp, nonce, sig = raw.split(":", 3)
|
|
payload = f"{admin_id}:{exp}:{nonce}"
|
|
expected = hmac.new(SECRET.encode(), payload.encode(), hashlib.sha256).hexdigest()
|
|
if hmac.compare_digest(expected, sig) and int(exp) > time.time():
|
|
return int(admin_id)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def now() -> str:
|
|
return datetime.now().isoformat(timespec="seconds")
|
|
|
|
|
|
def log(conn: sqlite3.Connection, admin_id: int | None, action: str, object_type: str, object_id: object = None, detail: dict | None = None) -> None:
|
|
conn.execute(
|
|
"insert into logs(admin_user_id,action,object_type,object_id,detail_json,created_at) values(?,?,?,?,?,?)",
|
|
(admin_id, action, object_type, str(object_id) if object_id is not None else None, json.dumps(detail or {}, ensure_ascii=False), now()),
|
|
)
|
|
|
|
|
|
def mask_phone(phone: object) -> str:
|
|
raw = "".join(ch for ch in str(phone or "") if ch.isdigit())
|
|
if len(raw) >= 7:
|
|
return raw[:3] + "****" + raw[-4:]
|
|
return raw or "-"
|
|
|
|
|
|
def log_risk(action: str) -> str:
|
|
if action in HIGH_RISK_ACTIONS:
|
|
return "high"
|
|
if action in MEDIUM_RISK_ACTIONS:
|
|
return "medium"
|
|
return "low"
|
|
|
|
|
|
def log_risk_text(risk: str) -> str:
|
|
return {"high": "\u9ad8\u98ce\u9669", "medium": "\u4e2d\u98ce\u9669", "low": "\u4f4e\u98ce\u9669"}.get(risk, risk)
|
|
|
|
|
|
def diff_detail(before: sqlite3.Row | None, after: dict, fields: list[str]) -> dict[str, dict[str, object]]:
|
|
changes: dict[str, dict[str, object]] = {}
|
|
for field in fields:
|
|
old_value = before[field] if before else None
|
|
new_value = after.get(field, old_value)
|
|
if old_value != new_value:
|
|
changes[field] = {"before": old_value, "after": new_value}
|
|
return changes
|
|
|
|
|
|
def log_summary(action: str, object_type: str, object_id: object, detail: dict[str, object]) -> str:
|
|
label = ACTION_LABELS.get(action, action)
|
|
if action in {"create_project", "update_project"}:
|
|
return f"{label}\uff1a{detail.get('name') or detail.get('code') or object_id}"
|
|
if action in {"create_learner", "update_learner", "delete_learner"}:
|
|
phone = detail.get("phone")
|
|
return f"{label}\uff1a{detail.get('name') or object_id}" + (f"\uff08{phone}\uff09" if phone else "")
|
|
if action in {"create_certificate", "void_certificate", "regenerate_public_token"}:
|
|
learner_name = detail.get("learner_name")
|
|
return f"{label}\uff1a{detail.get('certificate_no') or object_id}" + (f"\uff08{learner_name}\uff09" if learner_name else "")
|
|
if action in {"upload_import_file", "delete_import_batch"}:
|
|
return f"{label}\uff1a{detail.get('filename') or object_id}"
|
|
if action == "confirm_import_batch":
|
|
count = detail.get("imported_rows", detail.get("valid_rows", 0))
|
|
return f"{label}\uff1a\u6279\u6b21 {object_id}\uff0c\u5bfc\u5165 {count} \u884c"
|
|
if action == "public_search_certificate":
|
|
mode = "\u8bc1\u4e66\u7f16\u53f7" if detail.get("mode") == "certificate_no" else "\u624b\u673a\u53f7"
|
|
return f"{label}\uff1a{detail.get('name') or '-'} \u7528{mode}\u67e5\u8be2\uff0c\u5339\u914d {detail.get('matched_count', 0)} \u6761"
|
|
if action == "public_download_certificate":
|
|
return f"{label}\uff1a{detail.get('certificate_no') or object_id}\uff08{detail.get('learner_name') or '-'}\uff09"
|
|
return f"{label}\uff1a{OBJECT_LABELS.get(object_type, object_type)} {object_id or ''}".strip()
|
|
|
|
|
|
def format_log_row(row: sqlite3.Row) -> dict:
|
|
detail: dict[str, object] = {}
|
|
if row["detail_json"]:
|
|
try:
|
|
detail = json.loads(row["detail_json"])
|
|
except Exception:
|
|
detail = {"raw": row["detail_json"]}
|
|
risk = log_risk(row["action"])
|
|
payload = row_dict(row)
|
|
payload.update({
|
|
"action_label": ACTION_LABELS.get(row["action"], row["action"]),
|
|
"object_label": OBJECT_LABELS.get(row["object_type"], row["object_type"]),
|
|
"risk": risk,
|
|
"risk_label": log_risk_text(risk),
|
|
"summary": log_summary(row["action"], row["object_type"], row["object_id"], detail),
|
|
})
|
|
return payload
|
|
|
|
|
|
def filtered_operation_logs(conn: sqlite3.Connection, query: dict[str, list[str]]) -> list[dict]:
|
|
action = (query.get("action") or [""])[0]
|
|
object_type = (query.get("object_type") or [""])[0]
|
|
risk = (query.get("risk") or [""])[0]
|
|
keyword = (query.get("keyword") or [""])[0].strip().lower()
|
|
items = [format_log_row(row) for row in conn.execute("select * from logs order by id desc limit 500").fetchall()]
|
|
if action:
|
|
items = [item for item in items if item["action"] == action]
|
|
if object_type:
|
|
items = [item for item in items if item["object_type"] == object_type]
|
|
if risk:
|
|
items = [item for item in items if item["risk"] == risk]
|
|
if keyword:
|
|
items = [
|
|
item for item in items
|
|
if keyword in " ".join(str(item.get(key, "")) for key in ["object_id", "summary", "detail_json", "action_label", "object_label"]).lower()
|
|
]
|
|
return items
|
|
|
|
|
|
def list_operation_logs(conn: sqlite3.Connection, query: dict[str, list[str]]) -> dict[str, object]:
|
|
items = filtered_operation_logs(conn, query)
|
|
page = max(1, int((query.get("page") or ["1"])[0] or "1"))
|
|
page_size = max(1, min(100, int((query.get("page_size") or ["20"])[0] or "20")))
|
|
total = len(items)
|
|
total_pages = max(1, (total + page_size - 1) // page_size)
|
|
page = min(page, total_pages)
|
|
start = (page - 1) * page_size
|
|
return {
|
|
"items": items[start:start + page_size],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"total_pages": total_pages,
|
|
}
|
|
|
|
|
|
def cleanup_expired_logs(conn: sqlite3.Connection) -> int:
|
|
normal_before = (datetime.now() - timedelta(days=LOG_RETENTION_DAYS)).isoformat(timespec="seconds")
|
|
high_before = (datetime.now() - timedelta(days=HIGH_RISK_LOG_RETENTION_DAYS)).isoformat(timespec="seconds")
|
|
high_actions = tuple(HIGH_RISK_ACTIONS)
|
|
normal_sql = f"delete from logs where created_at < ? and action not in ({','.join('?' for _ in high_actions)})"
|
|
cur = conn.execute(normal_sql, (normal_before, *high_actions))
|
|
removed = cur.rowcount if cur.rowcount != -1 else 0
|
|
high_sql = f"delete from logs where created_at < ? and action in ({','.join('?' for _ in high_actions)})"
|
|
cur = conn.execute(high_sql, (high_before, *high_actions))
|
|
removed += cur.rowcount if cur.rowcount != -1 else 0
|
|
return removed
|
|
|
|
|
|
def row_dict(row: sqlite3.Row | None) -> dict | None:
|
|
if row is None:
|
|
return None
|
|
return {key: row[key] for key in row.keys()}
|
|
|
|
|
|
def rows(conn: sqlite3.Connection, sql: str, args: tuple = ()) -> list[dict]:
|
|
return [row_dict(row) for row in conn.execute(sql, args).fetchall()]
|
|
|
|
|
|
def digest_int(text: str) -> int:
|
|
return int.from_bytes(hashlib.sha256(text.encode()).digest()[:8], "big")
|
|
|
|
|
|
def base_code(value: int, length: int) -> str:
|
|
chars = []
|
|
while value:
|
|
value, rem = divmod(value, len(ALPHABET))
|
|
chars.append(ALPHABET[rem])
|
|
return ("".join(reversed(chars)) or ALPHABET[0])[-length:].rjust(length, ALPHABET[0])
|
|
|
|
|
|
def certificate_no(cert_id: int, project: str, issue_date: str) -> str:
|
|
year = issue_date[:4] if issue_date else str(date.today().year)
|
|
seed = f"{SECRET}:{year}:{project}:{cert_id}"
|
|
short = base_code(digest_int(seed), 7)
|
|
check = base_code(digest_int(f"check:{seed}:{short}"), 2)
|
|
return f"PX{year}-{project}-{short}-{check}"
|
|
|
|
|
|
def json_bytes(value) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False, default=str).encode("utf-8")
|
|
|
|
|
|
def make_captcha() -> dict:
|
|
code = "".join(secrets.choice("ABCDEFGHJKLMNPQRSTUVWXYZ23456789") for _ in range(4))
|
|
captcha_id = secrets.token_urlsafe(12)
|
|
CAPTCHAS[captcha_id] = (code, time.time() + 300)
|
|
|
|
# 增大图片尺寸，提高可读性
|
|
width, height = 180, 56
|
|
image = Image.new("RGB", (width, height), "#f0f7f7")
|
|
draw = ImageDraw.Draw(image)
|
|
|
|
# 使用更大的字体
|
|
font = pil_font(28, "hei", True)
|
|
|
|
# 绘制背景干扰线
|
|
for _ in range(6):
|
|
x1, y1 = random.randint(0, width), random.randint(0, height)
|
|
x2, y2 = random.randint(0, width), random.randint(0, height)
|
|
draw.line((x1, y1, x2, y2), fill="#b8d4d4", width=1)
|
|
|
|
# 绘制干扰点
|
|
for _ in range(30):
|
|
x, y = random.randint(0, width), random.randint(0, height)
|
|
draw.point((x, y), fill="#a0c8c8")
|
|
|
|
# 绘制验证码字符，带随机偏移
|
|
colors = ["#1a5c5c", "#2d7a7a", "#1f6b6b", "#3a8a8a", "#0e4a4a"]
|
|
for i, ch in enumerate(code):
|
|
x = 22 + i * 38 + random.randint(-3, 3)
|
|
y = 10 + random.randint(-5, 5)
|
|
color = secrets.choice(colors)
|
|
draw.text((x, y), ch, fill=color, font=font)
|
|
|
|
# 添加边框
|
|
draw.rectangle((0, 0, width - 1, height - 1), outline="#c8dede", width=1)
|
|
|
|
buf = io.BytesIO()
|
|
image.save(buf, "PNG")
|
|
return {"captcha_id": captcha_id, "image": "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode()}
|
|
|
|
|
|
def verify_captcha(captcha_id: str, code: str) -> bool:
|
|
record = CAPTCHAS.pop(captcha_id, None)
|
|
return bool(record and record[1] > time.time() and record[0].upper() == code.strip().upper())
|
|
|
|
|
|
def generate_sms_code(phone: str) -> str:
|
|
"""生成短信验证码"""
|
|
# 清理过期的验证码
|
|
now = time.time()
|
|
expired = [sms_id for sms_id, (_, expires_at) in SMS_CODES.items() if expires_at < now]
|
|
for sms_id in expired:
|
|
SMS_CODES.pop(sms_id, None)
|
|
|
|
code = "".join(secrets.choice("0123456789") for _ in range(6))
|
|
sms_id = secrets.token_urlsafe(16)
|
|
SMS_CODES[sms_id] = (code, now + 300) # 5分钟有效期
|
|
|
|
# 在本地测试环境下，将验证码记录到日志
|
|
print(f"[短信验证码] 手机号: {phone}, 验证码: {code}, 有效期: 300秒")
|
|
|
|
return sms_id
|
|
|
|
|
|
def verify_sms_code(sms_id: str, code: str) -> bool:
|
|
"""验证短信验证码"""
|
|
# 清理过期的验证码
|
|
now = time.time()
|
|
expired = [sid for sid, (_, expires_at) in SMS_CODES.items() if expires_at < now]
|
|
for sid in expired:
|
|
SMS_CODES.pop(sid, None)
|
|
|
|
record = SMS_CODES.pop(sms_id, None)
|
|
if not record:
|
|
return False
|
|
expected, expires_at = record
|
|
if expires_at < now:
|
|
return False
|
|
return expected == code.strip()
|
|
|
|
|
|
def parse_multipart(body: bytes, content_type: str) -> tuple[str, bytes]:
|
|
boundary = content_type.split("boundary=", 1)[1].encode()
|
|
parts = body.split(b"--" + boundary)
|
|
for part in parts:
|
|
if b"filename=" not in part:
|
|
continue
|
|
head, data = part.split(b"\r\n\r\n", 1)
|
|
filename = head.split(b'filename="', 1)[1].split(b'"', 1)[0].decode(errors="ignore")
|
|
return filename, data.rsplit(b"\r\n", 1)[0]
|
|
raise ValueError("file not found")
|
|
|
|
|
|
def public_payload(conn: sqlite3.Connection, cert: sqlite3.Row) -> dict:
|
|
learner = conn.execute("select * from learners where id=?", (cert["learner_id"],)).fetchone()
|
|
return {
|
|
"certificate_no": cert["certificate_no"],
|
|
"learner_name": learner["current_name"] if learner else "",
|
|
"certificate_name": cert["certificate_name"],
|
|
"project_code": cert["project_code"],
|
|
"course_name": cert["course_name"],
|
|
"stage_name": cert["stage_name"],
|
|
"issue_date": cert["issue_date"],
|
|
"issuer_name": cert["issuer_name"],
|
|
"status": cert["status"],
|
|
"pdf_status": cert["pdf_status"],
|
|
"public_token": cert["public_token"],
|
|
"public_url": f"/cert/{cert['public_token']}",
|
|
"download_url": f"/api/public/certificates/token/{cert['public_token']}/download" if cert["status"] == "valid" else None,
|
|
"can_download_pdf": cert["status"] == "valid",
|
|
"disclaimer": DISCLAIMER,
|
|
}
|
|
|
|
|
|
def font_name() -> str:
|
|
for path in [
|
|
r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc",
|
|
"/System/Library/Fonts/PingFang.ttc",
|
|
"/System/Library/Fonts/Supplemental/Songti.ttc",
|
|
"/System/Library/Fonts/STHeiti Light.ttc",
|
|
"/System/Library/Fonts/Supplemental/Heiti.ttc",
|
|
]:
|
|
if Path(path).exists():
|
|
try:
|
|
pdfmetrics.registerFont(TTFont("LocalCJK", path))
|
|
return "LocalCJK"
|
|
except Exception:
|
|
pass
|
|
return "Helvetica"
|
|
|
|
|
|
def pil_font(size: int, kind: str = "song", bold: bool = False, scale: float = 1.0) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
|
|
choices = {
|
|
"kai": [
|
|
r"C:\Windows\Fonts\simkai.ttf", r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc",
|
|
"/usr/share/fonts/opentype/noto/NotoSerifCJK-Regular.ttc",
|
|
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
|
|
"/System/Library/Fonts/Supplemental/Kaiti.ttc",
|
|
("/System/Library/Fonts/Hiragino Sans GB.ttc", 2),
|
|
"/System/Library/Fonts/PingFang.ttc",
|
|
("/System/Library/Fonts/Supplemental/Songti.ttc", 1),
|
|
],
|
|
"hei": [
|
|
r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\msyh.ttc",
|
|
"/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
|
|
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
|
|
("/System/Library/Fonts/STHeiti Medium.ttc", 1),
|
|
("/System/Library/Fonts/Hiragino Sans GB.ttc", 2),
|
|
"/System/Library/Fonts/Supplemental/Heiti.ttc",
|
|
"/System/Library/Fonts/PingFang.ttc",
|
|
],
|
|
"song": [
|
|
r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc",
|
|
"/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
|
|
"/usr/share/fonts/opentype/noto/NotoSerifCJK-Regular.ttc",
|
|
("/System/Library/Fonts/Supplemental/Songti.ttc", 3),
|
|
("/System/Library/Fonts/Supplemental/Songti.ttc", 6),
|
|
("/System/Library/Fonts/Hiragino Sans GB.ttc", 0),
|
|
"/System/Library/Fonts/PingFang.ttc",
|
|
("/System/Library/Fonts/STHeiti Light.ttc", 1),
|
|
],
|
|
}
|
|
paths = choices.get(kind, choices["song"])
|
|
if bold and kind == "song":
|
|
paths = [
|
|
r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf",
|
|
"/usr/share/fonts/opentype/noto/NotoSansCJK-Bold.ttc",
|
|
("/System/Library/Fonts/STHeiti Medium.ttc", 1),
|
|
("/System/Library/Fonts/Hiragino Sans GB.ttc", 2),
|
|
"/System/Library/Fonts/Supplemental/Heiti.ttc",
|
|
] + paths
|
|
for item in paths:
|
|
path, index = item if isinstance(item, tuple) else (item, 0)
|
|
if Path(path).exists():
|
|
try:
|
|
return ImageFont.truetype(path, max(1, round(size * scale)), index=index)
|
|
except Exception:
|
|
pass
|
|
return ImageFont.load_default()
|
|
|
|
|
|
def date_parts(value: str) -> tuple[str, str, str]:
|
|
raw = str(value or "").strip()
|
|
for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
|
|
try:
|
|
d = datetime.strptime(raw[:10], fmt)
|
|
return str(d.year), str(d.month), str(d.day)
|
|
except ValueError:
|
|
pass
|
|
if len(raw) >= 10 and raw[0:4].isdigit():
|
|
return raw[0:4], raw[5:7].lstrip("0") or raw[5:7], raw[8:10].lstrip("0") or raw[8:10]
|
|
today = date.today()
|
|
return str(today.year), str(today.month), str(today.day)
|
|
|
|
|
|
def draw_fit(draw: ImageDraw.ImageDraw, pos: tuple[int, int], text: str, font: ImageFont.ImageFont, fill: str, max_width: int) -> None:
|
|
font_obj = font
|
|
size = getattr(font, "size", 24)
|
|
while draw.textlength(text, font=font_obj) > max_width and size > 12:
|
|
size -= 1
|
|
font_obj = pil_font(size, "song", True)
|
|
draw.text(pos, text, fill=fill, font=font_obj)
|
|
|
|
|
|
def draw_inline(
|
|
draw: ImageDraw.ImageDraw,
|
|
x: float,
|
|
y: float,
|
|
text: str,
|
|
size: int,
|
|
gap: int,
|
|
kind: str = "song",
|
|
bold: bool = False,
|
|
scale: float = 1.0,
|
|
x_scale: float = 1.0,
|
|
) -> float:
|
|
font = pil_font(size, kind, bold, scale)
|
|
draw.text((x, y), text, fill="#111111", font=font, anchor="ls")
|
|
if bold:
|
|
draw.text((x + 0.45 * x_scale, y), text, fill="#111111", font=font, anchor="ls")
|
|
return x + draw.textlength(text, font=font) + gap * x_scale
|
|
|
|
|
|
def draw_course_name(draw: ImageDraw.ImageDraw, text: str, x: float, y: int) -> int:
|
|
max_width = 900 - x
|
|
text_font = pil_font(27, "kai", True)
|
|
if draw.textlength(text, font=text_font) <= max_width:
|
|
draw.text((x, y), text, fill="#111111", font=text_font)
|
|
draw.text((x + 0.45, y), text, fill="#111111", font=text_font)
|
|
return y
|
|
|
|
start_y = y + 34
|
|
lines = split_by_width(draw, text, text_font, 690)[:2]
|
|
for index, line in enumerate(lines):
|
|
line_y = start_y + index * 34
|
|
draw.text((185, line_y), line, fill="#111111", font=text_font)
|
|
draw.text((185.45, line_y), line, fill="#111111", font=text_font)
|
|
return start_y + (len(lines) - 1) * 34
|
|
|
|
|
|
def draw_inline_flow(
|
|
draw: ImageDraw.ImageDraw,
|
|
segments: list[tuple[str, str, int]],
|
|
start_x: float,
|
|
start_y: float,
|
|
scale: float,
|
|
x_scale: float,
|
|
y_scale: float,
|
|
) -> None:
|
|
x = start_x
|
|
y = start_y
|
|
line_start = 185 * x_scale
|
|
line_height = 42 * y_scale
|
|
for text, style, gap_before in segments:
|
|
x += gap_before * x_scale
|
|
text_font = pil_font(27, "kai", True, scale) if style == "course" else pil_font(24, "song", False, scale)
|
|
for char in text:
|
|
max_right = (900 if y == start_y else 730) * x_scale
|
|
width = draw.textlength(char, font=text_font)
|
|
if x + width > max_right and x > line_start:
|
|
x = line_start
|
|
y += line_height
|
|
draw.text((x, y), char, fill="#111111", font=text_font, anchor="ls")
|
|
if style == "course":
|
|
draw.text((x + 0.45 * x_scale, y), char, fill="#111111", font=text_font, anchor="ls")
|
|
x += width
|
|
|
|
|
|
def split_by_width(draw: ImageDraw.ImageDraw, text: str, text_font: ImageFont.ImageFont, max_width: int) -> list[str]:
|
|
lines: list[str] = []
|
|
line = ""
|
|
for char in text:
|
|
next_line = line + char
|
|
if line and draw.textlength(next_line, font=text_font) > max_width:
|
|
lines.append(line)
|
|
line = char
|
|
else:
|
|
line = next_line
|
|
if line:
|
|
lines.append(line)
|
|
return lines
|
|
|
|
|
|
def scaled_box(box: tuple[int, int, int, int], x_scale: float, y_scale: float) -> tuple[int, int, int, int]:
|
|
x, y, w, h = box
|
|
return round(x * x_scale), round(y * y_scale), round(w * x_scale), round(h * y_scale)
|
|
|
|
|
|
def to_xyxy(box: tuple[int, int, int, int]) -> tuple[int, int, int, int]:
|
|
x, y, w, h = box
|
|
return x, y, x + w, y + h
|
|
|
|
|
|
def paper_patch(image: Image.Image, box: tuple[int, int, int, int], x_scale: float, y_scale: float) -> None:
|
|
x, y, w, h = scaled_box(box, x_scale, y_scale)
|
|
patch = Image.new("RGB", (w, h), "#fbf7ef")
|
|
source = image.crop((round(72 * x_scale), y, round(160 * x_scale), y + h))
|
|
for tx in range(0, w, source.width):
|
|
patch.paste(source.crop((0, 0, min(source.width, w - tx), h)), (tx, 0))
|
|
cover = Image.new("RGB", (w, h), "#fbf7ef")
|
|
patch = Image.blend(patch, cover, 0.38)
|
|
image.paste(patch, (x, y), soft_edge_mask(w, h, round(16 * x_scale), round(10 * y_scale)))
|
|
|
|
|
|
def soft_edge_mask(width: int, height: int, feather_x: int, feather_y: int) -> Image.Image:
|
|
feather_x = max(1, feather_x)
|
|
feather_y = max(1, feather_y)
|
|
mask = Image.new("L", (width, height), 255)
|
|
pixels = mask.load()
|
|
for y in range(height):
|
|
vertical = min(1.0, y / feather_y, (height - 1 - y) / feather_y)
|
|
for x in range(width):
|
|
horizontal = min(1.0, x / feather_x, (width - 1 - x) / feather_x)
|
|
pixels[x, y] = max(0, min(255, round(255 * min(horizontal, vertical))))
|
|
return mask
|
|
|
|
|
|
def draw_issue_date_numbers(
|
|
draw: ImageDraw.ImageDraw,
|
|
issue_year: str,
|
|
issue_month: str,
|
|
issue_day: str,
|
|
scale: float,
|
|
x_scale: float,
|
|
y_scale: float,
|
|
) -> None:
|
|
text_font = pil_font(13, "song", True, scale)
|
|
y = 688 * y_scale
|
|
for text, x in [(issue_year, 500), (issue_month, 535), (issue_day, 566)]:
|
|
draw.text((x * x_scale, y), text, fill="#111111", font=text_font, anchor="mm")
|
|
|
|
|
|
def certificate_template_path() -> Path:
|
|
assets_dir = ROOT / "static" / "assets"
|
|
for name in ["certificate-template.png", "certificate-template.jpg"]:
|
|
path = assets_dir / name
|
|
if path.exists():
|
|
return path
|
|
raise FileNotFoundError("Certificate template image not found")
|
|
|
|
|
|
def pdf_resolution(image: Image.Image) -> float:
|
|
return PDF_BASE_RESOLUTION * (image.width / DESIGN_WIDTH)
|
|
|
|
|
|
def render_certificate_image(conn: sqlite3.Connection, cert: sqlite3.Row) -> Image.Image:
|
|
template = certificate_template_path()
|
|
image = Image.open(template).convert("RGB")
|
|
base_image = image.copy()
|
|
draw = ImageDraw.Draw(image)
|
|
x_scale = image.width / DESIGN_WIDTH
|
|
y_scale = image.height / DESIGN_HEIGHT
|
|
text_scale = (x_scale + y_scale) / 2
|
|
learner = conn.execute("select * from learners where id=?", (cert["learner_id"],)).fetchone()
|
|
learner_name = (learner["current_name"] if learner else "") or ""
|
|
issue_year, issue_month, issue_day = date_parts(cert["issue_date"])
|
|
|
|
paper_patch(image, (150, 455, 760, 119), x_scale, y_scale)
|
|
|
|
draw.text(
|
|
(512 * x_scale, 414 * y_scale),
|
|
learner_name,
|
|
fill="#111111",
|
|
font=pil_font(32, "kai", True, text_scale),
|
|
anchor="mm",
|
|
)
|
|
|
|
x = 185.0 * x_scale
|
|
y = 494 * y_scale
|
|
x = draw_inline(draw, x, y, "\u5728", 24, 18, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, issue_year, 24, 8, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, "\u5e74", 24, 12, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, issue_month, 24, 6, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, "\u6708", 24, 12, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, issue_day, 24, 4, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, "\u65e5\u81f3", 24, 12, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, issue_year, 24, 8, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, "\u5e74", 24, 12, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, issue_month, 24, 6, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, "\u6708", 24, 12, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, issue_day, 24, 4, scale=text_scale, x_scale=x_scale)
|
|
x = draw_inline(draw, x, y, "\u65e5\u5b8c\u6210\u4e86", 24, 8, scale=text_scale, x_scale=x_scale)
|
|
course_text = f"\u201c{cert['course_name'] or cert['certificate_name'] or cert['project_code']}\u201d"
|
|
stage_input = (cert["stage_name"] or "").strip()
|
|
stage = f"{stage_input}\u8bfe\u7a0b\u7684\u4e13\u4e1a\u5b66\u4e60\u3002" if stage_input else "\u521d\u7ea7\u8bfe\u7a0b\u7684\u4e13\u4e1a\u5b66\u4e60\u3002"
|
|
draw_inline_flow(draw, [(course_text, "course", 0), (stage, "normal", 14)], x, y, text_scale, x_scale, y_scale)
|
|
|
|
draw_issue_date_numbers(draw, issue_year, issue_month, issue_day, text_scale, x_scale, y_scale)
|
|
draw.text(
|
|
(105 * x_scale, 76 * y_scale),
|
|
f"\u8bc1\u4e66\u7f16\u53f7\uff1a{cert['certificate_no']}",
|
|
fill="#111827",
|
|
font=pil_font(14, "hei", True, text_scale),
|
|
)
|
|
mentor_box = scaled_box((720, 535, 230, 94), x_scale, y_scale)
|
|
image.paste(base_image.crop(to_xyxy(mentor_box)), mentor_box[:2])
|
|
return image
|
|
|
|
|
|
def build_pdf(conn: sqlite3.Connection, cert: sqlite3.Row) -> Path:
|
|
pdf_dir = DATA / "pdf-cache"
|
|
pdf_dir.mkdir(exist_ok=True)
|
|
path = pdf_dir / f"{cert['certificate_no']}.pdf"
|
|
template = certificate_template_path()
|
|
renderer_mtime = Path(__file__).stat().st_mtime
|
|
if path.exists() and path.stat().st_mtime >= max(template.stat().st_mtime, renderer_mtime):
|
|
return path
|
|
image = render_certificate_image(conn, cert)
|
|
image.save(path, "PDF", resolution=pdf_resolution(image))
|
|
conn.execute("update certificates set pdf_status=? where id=?", ("generated", cert["id"]))
|
|
return path
|
|
|
|
|
|
def validate_import(conn: sqlite3.Connection, file_path: Path, filename: str) -> dict:
|
|
wb = load_workbook(file_path, data_only=True)
|
|
ws = wb.active
|
|
header = [cell.value for cell in ws[1]]
|
|
valid_rows = []
|
|
failed = 0
|
|
for row in ws.iter_rows(min_row=2, values_only=True):
|
|
if not any(row):
|
|
continue
|
|
item = dict(zip(header, row))
|
|
missing = [c for c in COLS if not item.get(c)]
|
|
project = conn.execute("select 1 from projects where code=? and status='active'", (str(item.get(COL_PROJECT, "")).upper(),)).fetchone()
|
|
if missing or not project:
|
|
failed += 1
|
|
else:
|
|
valid_rows.append(item)
|
|
cur = conn.execute(
|
|
"insert into import_batches(filename,status,total_rows,valid_rows,failed_rows,error_report_path,created_at,updated_at) values(?,?,?,?,?,?,?,?)",
|
|
(filename, "validated", len(valid_rows) + failed, len(valid_rows), failed, None, now(), now()),
|
|
)
|
|
batch_id = cur.lastrowid
|
|
(DATA / f"batch-{batch_id}.json").write_text(json.dumps(valid_rows, ensure_ascii=False, default=str), encoding="utf-8")
|
|
return row_dict(conn.execute("select * from import_batches where id=?", (batch_id,)).fetchone())
|
|
|
|
|
|
def import_batch(conn: sqlite3.Connection, batch_id: int, admin_id: int):
|
|
batch = conn.execute("select * from import_batches where id=?", (batch_id,)).fetchone()
|
|
if not batch:
|
|
raise ValueError("Import batch not found")
|
|
if batch["status"] == "imported":
|
|
return row_dict(batch)
|
|
if batch["status"] != "validated":
|
|
raise ValueError("Import batch is not ready")
|
|
|
|
rows_data = json.loads((DATA / f"batch-{batch_id}.json").read_text(encoding="utf-8"))
|
|
ok_rows = 0
|
|
failed_rows = 0
|
|
for item in rows_data:
|
|
phone = str(item[COL_PHONE])
|
|
learner = conn.execute("select * from learners where phone=?", (phone,)).fetchone()
|
|
if learner:
|
|
learner_id = learner["id"]
|
|
conn.execute("update learners set current_name=?,updated_at=? where id=?", (item[COL_NAME], now(), learner_id))
|
|
else:
|
|
cur = conn.execute(
|
|
"insert into learners(phone,current_name,status,created_at,updated_at) values(?,?,?,?,?)",
|
|
(phone, item[COL_NAME], "active", now(), now()),
|
|
)
|
|
learner_id = cur.lastrowid
|
|
issue = str(item[COL_ISSUE_DATE])[:10].replace("/", "-")
|
|
project = conn.execute("select * from projects where code=? and status='active'", (str(item[COL_PROJECT]).upper(),)).fetchone()
|
|
if not project:
|
|
failed_rows += 1
|
|
continue
|
|
duplicate = conn.execute(
|
|
"""
|
|
select 1 from certificates
|
|
where learner_id=?
|
|
and project_code=?
|
|
and certificate_name=?
|
|
and coalesce(course_name,'')=?
|
|
and coalesce(stage_name,'')=?
|
|
and issue_date=?
|
|
""",
|
|
(
|
|
learner_id,
|
|
project["code"],
|
|
project["name"],
|
|
project["default_course_name"] or "",
|
|
project["default_stage_name"] or "",
|
|
issue,
|
|
),
|
|
).fetchone()
|
|
if duplicate:
|
|
ok_rows += 1
|
|
continue
|
|
cur = conn.execute(
|
|
"insert into certificates(learner_id,project_code,certificate_no,certificate_name,course_name,stage_name,issue_date,issuer_name,status,pdf_status,public_token,qr_token,remark,created_at,updated_at) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(learner_id, project["code"], "PENDING", project["name"], project["default_course_name"], project["default_stage_name"], issue, project["default_issuer_name"], "valid", "not_generated", secrets.token_urlsafe(24), secrets.token_urlsafe(24), None, now(), now()),
|
|
)
|
|
cert_id = cur.lastrowid
|
|
conn.execute("update certificates set certificate_no=? where id=?", (certificate_no(cert_id, str(item[COL_PROJECT]).upper(), issue), cert_id))
|
|
ok_rows += 1
|
|
next_status = "imported" if ok_rows else "failed"
|
|
conn.execute("update import_batches set status=?,failed_rows=?,updated_at=? where id=?", (next_status, failed_rows or batch["failed_rows"], now(), batch_id))
|
|
log(conn, admin_id, "confirm_import_batch", "import_batch", batch_id, {"filename": batch["filename"], "imported_rows": ok_rows, "failed_rows": failed_rows})
|
|
return row_dict(conn.execute("select * from import_batches where id=?", (batch_id,)).fetchone())
|
|
|
|
|
|
def export_certs(conn: sqlite3.Connection) -> bytes:
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.append([COL_NAME, COL_PHONE, "\u8bc1\u4e66\u7f16\u53f7", "\u8bc1\u4e66\u5bf9\u5916\u76f4\u8fbe\u94fe\u63a5", COL_PROJECT, "\u8bc1\u4e66\u540d\u79f0", COL_ISSUE_DATE, "\u8bc1\u4e66\u72b6\u6001"])
|
|
certs = conn.execute("select c.*,l.current_name,l.phone from certificates c join learners l on l.id=c.learner_id order by c.id desc").fetchall()
|
|
for row in certs:
|
|
phone = row["phone"] or ""
|
|
ws.append([row["current_name"], phone[:3] + "****" + phone[-4:] if len(phone) >= 7 else phone, row["certificate_no"], f"http://localhost:8000/cert/{row['public_token']}", row["project_code"], row["certificate_name"], row["issue_date"], row["status"]])
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
def export_logs(items: list[dict]) -> bytes:
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "\u64cd\u4f5c\u65e5\u5fd7"
|
|
ws.append(["\u65f6\u95f4", "\u64cd\u4f5c\u4ebaID", "\u64cd\u4f5c", "\u5bf9\u8c61", "\u5bf9\u8c61ID", "\u98ce\u9669\u7b49\u7ea7", "\u6458\u8981", "\u8be6\u60c5"])
|
|
for item in items:
|
|
ws.append([
|
|
item.get("created_at"),
|
|
item.get("admin_user_id"),
|
|
item.get("action_label"),
|
|
item.get("object_label"),
|
|
item.get("object_id"),
|
|
item.get("risk_label"),
|
|
item.get("summary"),
|
|
item.get("detail_json"),
|
|
])
|
|
for column in "ABCDEFGH":
|
|
ws.column_dimensions[column].width = 22 if column != "H" else 60
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return buf.getvalue()
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
self.route()
|
|
|
|
def do_POST(self):
|
|
self.route()
|
|
|
|
def do_PUT(self):
|
|
self.route()
|
|
|
|
def do_DELETE(self):
|
|
self.route()
|
|
|
|
def route(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path
|
|
try:
|
|
if path.startswith("/api/"):
|
|
return self.handle_api(path, parse_qs(parsed.query))
|
|
if path in ["/", "/admin", "/admin/login", "/query"] or path.startswith("/cert/") or path.startswith("/verify/"):
|
|
return self.send_file(ROOT / "static" / "index.html", "text/html; charset=utf-8")
|
|
target = ROOT / "static" / path.lstrip("/")
|
|
if target.exists():
|
|
return self.send_file(target, "application/octet-stream")
|
|
return self.error(HTTPStatus.NOT_FOUND, "Not found")
|
|
except Exception as exc:
|
|
(DATA / "last-error.log").write_text(traceback.format_exc(), encoding="utf-8")
|
|
return self.error(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc))
|
|
|
|
def handle_api(self, path: str, query: dict):
|
|
body = self.read_body()
|
|
admin_id = self.admin_id()
|
|
with db() as conn:
|
|
if path == "/api/health":
|
|
return self.json({"status": "ok"})
|
|
if path == "/api/admin/auth/login":
|
|
data = json.loads(body or b"{}")
|
|
admin = conn.execute("select * from admins where username=?", (data.get("username"),)).fetchone()
|
|
if not admin or not check_password(data.get("password", ""), admin["password_hash"]):
|
|
return self.error(401, "\u8d26\u53f7\u6216\u5bc6\u7801\u9519\u8bef\uff0c\u8bf7\u91cd\u8bd5")
|
|
return self.json({"access_token": make_token(admin["id"]), "token_type": "bearer", "username": admin["username"], "role_code": "system_admin"})
|
|
if path == "/api/public/captcha":
|
|
return self.json(make_captcha())
|
|
if path == "/api/public/sms/send":
|
|
data = json.loads(body or b"{}")
|
|
phone = data.get("phone", "").strip()
|
|
if not phone:
|
|
return self.error(400, "\u8bf7\u8f93\u5165\u624b\u673a\u53f7")
|
|
if not verify_captcha(data.get("captcha_id", ""), data.get("captcha_code", "")):
|
|
return self.error(400, "\u56fe\u5f62\u9a8c\u8bc1\u7801\u9519\u8bef\u6216\u5df2\u8fc7\u671f\uff0c\u8bf7\u91cd\u65b0\u8f93\u5165")
|
|
sms_id = generate_sms_code(phone)
|
|
return self.json({"sms_id": sms_id, "message": "\u9a8c\u8bc1\u7801\u5df2\u53d1\u9001"})
|
|
if path == "/api/public/certificates/search":
|
|
data = json.loads(body or b"{}")
|
|
if not verify_captcha(data.get("captcha_id", ""), data.get("captcha_code", "")):
|
|
return self.error(400, "\u9a8c\u8bc1\u7801\u9519\u8bef\u6216\u5df2\u8fc7\u671f\uff0c\u8bf7\u91cd\u65b0\u8f93\u5165")
|
|
name = data.get("name", "").strip()
|
|
query_certificate_no = data.get("certificate_no", "").strip().upper()
|
|
phone = data.get("phone", "").strip()
|
|
if not name or (not query_certificate_no and not phone):
|
|
return self.error(400, "\u8bf7\u8f93\u5165\u59d3\u540d\uff0c\u5e76\u586b\u5199\u8bc1\u4e66\u7f16\u53f7\u6216\u624b\u673a\u53f7\u5176\u4e2d\u4e00\u9879")
|
|
if query_certificate_no:
|
|
result = rows(conn,
|
|
"select c.* from certificates c join learners l on l.id=c.learner_id where c.certificate_no=? and l.current_name=?",
|
|
(query_certificate_no, name),
|
|
)
|
|
else:
|
|
result = rows(conn,
|
|
"select c.* from certificates c join learners l on l.id=c.learner_id where l.phone=? and l.current_name=? order by c.issue_date desc,c.id desc",
|
|
(phone, name),
|
|
)
|
|
if not result:
|
|
log(conn, None, "public_search_certificate", "public_access", None, {"mode": "certificate_no" if query_certificate_no else "phone", "name": name, "certificate_no": query_certificate_no or None, "phone": mask_phone(phone), "matched_count": 0, "result": "not_found"})
|
|
return self.error(404, "\u672a\u67e5\u8be2\u5230\u5339\u914d\u7684\u6709\u6548\u8bc1\u4e66\u4fe1\u606f\uff0c\u8bf7\u6838\u5bf9\u540e\u91cd\u8bd5")
|
|
log(conn, None, "public_search_certificate", "public_access", result[0]["certificate_no"], {"mode": "certificate_no" if query_certificate_no else "phone", "name": name, "certificate_no": query_certificate_no or None, "phone": mask_phone(phone), "matched_count": len(result), "result": "matched"})
|
|
return self.json({"items": [public_payload(conn, item) for item in result]})
|
|
if path == "/api/public/certificates/search-by-sms":
|
|
data = json.loads(body or b"{}")
|
|
phone = data.get("phone", "").strip()
|
|
sms_id = data.get("sms_id", "").strip()
|
|
sms_code = data.get("sms_code", "").strip()
|
|
if not phone or not sms_id or not sms_code:
|
|
return self.error(400, "\u8bf7\u8f93\u5165\u624b\u673a\u53f7\u548c\u77ed\u4fe1\u9a8c\u8bc1\u7801")
|
|
if not verify_sms_code(sms_id, sms_code):
|
|
return self.error(400, "\u9a8c\u8bc1\u7801\u9519\u8bef\u6216\u5df2\u8fc7\u671f\uff0c\u8bf7\u91cd\u65b0\u83b7\u53d6")
|
|
result = rows(conn,
|
|
"select c.* from certificates c join learners l on l.id=c.learner_id where l.phone=? order by c.issue_date desc,c.id desc",
|
|
(phone,),
|
|
)
|
|
if not result:
|
|
log(conn, None, "public_search_certificate_sms", "public_access", None, {"mode": "phone_sms", "phone": mask_phone(phone), "matched_count": 0, "result": "not_found"})
|
|
return self.error(404, "\u672a\u67e5\u8be2\u5230\u5339\u914d\u7684\u8bc1\u4e66\u4fe1\u606f\uff0c\u8bf7\u6838\u5bf9\u624b\u673a\u53f7\u540e\u91cd\u8bd5")
|
|
log(conn, None, "public_search_certificate_sms", "public_access", result[0]["certificate_no"], {"mode": "phone_sms", "phone": mask_phone(phone), "matched_count": len(result), "result": "matched"})
|
|
return self.json({"items": [public_payload(conn, item) for item in result]})
|
|
if path.startswith("/api/public/certificates/token/"):
|
|
token = unquote(path.split("/")[5])
|
|
cert = conn.execute("select * from certificates where public_token=? or qr_token=?", (token, token)).fetchone()
|
|
if not cert:
|
|
return self.error(404, "\u94fe\u63a5\u65e0\u6548\u6216\u5df2\u5931\u6548")
|
|
if path.endswith("/download"):
|
|
if cert["status"] != "valid":
|
|
return self.error(403, "\u8bc1\u4e66\u5df2\u4f5c\u5e9f\uff0c\u4e0d\u652f\u6301\u4e0b\u8f7d\u6b63\u5e38 PDF")
|
|
learner = conn.execute("select * from learners where id=?", (cert["learner_id"],)).fetchone()
|
|
log(conn, None, "public_download_certificate", "certificate", cert["id"], {"certificate_no": cert["certificate_no"], "learner_name": learner["current_name"] if learner else "", "project_code": cert["project_code"]})
|
|
return self.send_file(build_pdf(conn, cert), "application/pdf", f"{cert['certificate_no']}.pdf")
|
|
return self.json(public_payload(conn, cert))
|
|
|
|
if not admin_id:
|
|
return self.error(401, "Not authenticated")
|
|
|
|
if path == "/api/admin/dashboard/summary":
|
|
return self.json({
|
|
"learner_count": conn.execute("select count(*) from learners where status!='deleted'").fetchone()[0],
|
|
"certificate_count": conn.execute("select count(*) from certificates").fetchone()[0],
|
|
"valid_certificate_count": conn.execute("select count(*) from certificates where status='valid'").fetchone()[0],
|
|
"voided_certificate_count": conn.execute("select count(*) from certificates where status='voided'").fetchone()[0],
|
|
"recent_import_count": conn.execute("select count(*) from import_batches").fetchone()[0],
|
|
})
|
|
|
|
if path == "/api/admin/projects" and self.command == "GET":
|
|
return self.json(rows(conn, "select * from projects order by code"))
|
|
if path == "/api/admin/projects" and self.command == "POST":
|
|
data = json.loads(body)
|
|
conn.execute(
|
|
"insert into projects(code,name,default_certificate_name,default_course_name,default_stage_name,default_issuer_name,status) values(?,?,?,?,?,?,?)",
|
|
(
|
|
data["code"].strip().upper(),
|
|
data["name"].strip(),
|
|
data["name"].strip(),
|
|
data.get("default_course_name"),
|
|
data.get("default_stage_name"),
|
|
data.get("default_issuer_name") or "\u672c\u516c\u53f8",
|
|
"active",
|
|
),
|
|
)
|
|
log(conn, admin_id, "create_project", "project", data["code"], {"code": data["code"].strip().upper(), "name": data["name"].strip(), "status": "active"})
|
|
return self.json({"ok": True}, 201)
|
|
if path.startswith("/api/admin/projects/") and self.command == "PUT":
|
|
project_id = int(path.rsplit("/", 1)[1])
|
|
data = json.loads(body)
|
|
project = conn.execute("select * from projects where id=?", (project_id,)).fetchone()
|
|
if not project:
|
|
return self.error(404, "Project not found")
|
|
next_data = {
|
|
"name": data.get("name", project["name"]),
|
|
"default_course_name": data.get("default_course_name", project["default_course_name"]),
|
|
"default_stage_name": data.get("default_stage_name", project["default_stage_name"]),
|
|
"default_issuer_name": data.get("default_issuer_name", project["default_issuer_name"]),
|
|
"status": data.get("status", project["status"]),
|
|
}
|
|
conn.execute(
|
|
"update projects set name=?,default_certificate_name=?,default_course_name=?,default_stage_name=?,default_issuer_name=?,status=? where id=?",
|
|
(
|
|
next_data["name"],
|
|
next_data["name"],
|
|
next_data["default_course_name"],
|
|
next_data["default_stage_name"],
|
|
next_data["default_issuer_name"],
|
|
next_data["status"],
|
|
project_id,
|
|
),
|
|
)
|
|
log(conn, admin_id, "update_project", "project", project_id, {"code": project["code"], "name": next_data["name"], "changes": diff_detail(project, next_data, ["name", "default_course_name", "default_stage_name", "default_issuer_name", "status"])})
|
|
return self.json(row_dict(conn.execute("select * from projects where id=?", (project_id,)).fetchone()))
|
|
|
|
if path == "/api/admin/learners" and self.command == "GET":
|
|
keyword = (query.get("keyword") or [""])[0]
|
|
if keyword:
|
|
return self.json(rows(conn, "select * from learners where status!='deleted' and (current_name like ? or phone like ? or student_no like ?) order by id desc", (f"%{keyword}%", f"%{keyword}%", f"%{keyword}%")))
|
|
return self.json(rows(conn, "select * from learners where status!='deleted' order by id desc limit 200"))
|
|
if path == "/api/admin/learners" and self.command == "POST":
|
|
data = json.loads(body)
|
|
cur = conn.execute(
|
|
"insert into learners(phone,current_name,student_no,status,remark,created_at,updated_at) values(?,?,?,?,?,?,?)",
|
|
(data["phone"].strip(), data["current_name"].strip(), data.get("student_no"), "active", data.get("remark"), now(), now()),
|
|
)
|
|
log(conn, admin_id, "create_learner", "learner", cur.lastrowid, {"name": data["current_name"].strip(), "phone": mask_phone(data["phone"]), "status": "active"})
|
|
return self.json(row_dict(conn.execute("select * from learners where id=?", (cur.lastrowid,)).fetchone()), 201)
|
|
if path.startswith("/api/admin/learners/"):
|
|
learner_id = int(path.rsplit("/", 1)[1])
|
|
if self.command == "GET":
|
|
learner = row_dict(conn.execute("select * from learners where id=?", (learner_id,)).fetchone())
|
|
return self.json(learner) if learner else self.error(404, "Learner not found")
|
|
if self.command == "PUT":
|
|
data = json.loads(body)
|
|
learner = conn.execute("select * from learners where id=?", (learner_id,)).fetchone()
|
|
if not learner:
|
|
return self.error(404, "Learner not found")
|
|
next_data = {
|
|
"phone": data.get("phone", learner["phone"]),
|
|
"current_name": data.get("current_name", learner["current_name"]),
|
|
"student_no": data.get("student_no", learner["student_no"]),
|
|
"status": data.get("status", learner["status"]),
|
|
"remark": data.get("remark", learner["remark"]),
|
|
}
|
|
conn.execute(
|
|
"update learners set phone=?,current_name=?,student_no=?,status=?,remark=?,updated_at=? where id=?",
|
|
(next_data["phone"], next_data["current_name"], next_data["student_no"], next_data["status"], next_data["remark"], now(), learner_id),
|
|
)
|
|
changes = diff_detail(learner, next_data, ["phone", "current_name", "student_no", "status", "remark"])
|
|
if "phone" in changes:
|
|
changes["phone"] = {"before": mask_phone(changes["phone"]["before"]), "after": mask_phone(changes["phone"]["after"])}
|
|
log(conn, admin_id, "update_learner", "learner", learner_id, {"name": next_data["current_name"], "phone": mask_phone(next_data["phone"]), "changes": changes})
|
|
return self.json(row_dict(conn.execute("select * from learners where id=?", (learner_id,)).fetchone()))
|
|
if self.command == "DELETE":
|
|
learner = conn.execute("select * from learners where id=?", (learner_id,)).fetchone()
|
|
conn.execute("update learners set status='deleted',updated_at=? where id=?", (now(), learner_id))
|
|
log(conn, admin_id, "delete_learner", "learner", learner_id, {"name": learner["current_name"] if learner else learner_id, "phone": mask_phone(learner["phone"] if learner else "")})
|
|
return self.json({"ok": True})
|
|
|
|
if path == "/api/admin/certificates" and self.command == "GET":
|
|
return self.json(rows(conn, "select c.*,l.current_name learner_name from certificates c left join learners l on l.id=c.learner_id order by c.id desc limit 200"))
|
|
if path == "/api/admin/certificates" and self.command == "POST":
|
|
data = json.loads(body)
|
|
project = conn.execute("select * from projects where code=? and status='active'", (data["project_code"].strip().upper(),)).fetchone()
|
|
if not project:
|
|
return self.error(400, "Project code is inactive or missing")
|
|
cur = conn.execute(
|
|
"insert into certificates(learner_id,project_code,certificate_no,certificate_name,course_name,stage_name,issue_date,issuer_name,status,pdf_status,public_token,qr_token,remark,created_at,updated_at) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(data["learner_id"], project["code"], "PENDING", project["name"], data.get("course_name") or project["default_course_name"], data.get("stage_name") or project["default_stage_name"], data["issue_date"], data.get("issuer_name") or project["default_issuer_name"], "valid", "not_generated", secrets.token_urlsafe(24), secrets.token_urlsafe(24), data.get("remark"), now(), now()),
|
|
)
|
|
no = certificate_no(cur.lastrowid, project["code"], data["issue_date"])
|
|
conn.execute("update certificates set certificate_no=? where id=?", (no, cur.lastrowid))
|
|
learner = conn.execute("select * from learners where id=?", (data["learner_id"],)).fetchone()
|
|
log(conn, admin_id, "create_certificate", "certificate", cur.lastrowid, {"certificate_no": no, "learner_name": learner["current_name"] if learner else "", "project_code": project["code"], "issue_date": data["issue_date"]})
|
|
return self.json(row_dict(conn.execute("select * from certificates where id=?", (cur.lastrowid,)).fetchone()), 201)
|
|
if path.startswith("/api/admin/certificates/"):
|
|
parts = path.split("/")
|
|
cert_id = int(parts[4])
|
|
cert = conn.execute("select * from certificates where id=?", (cert_id,)).fetchone()
|
|
if not cert:
|
|
return self.error(404, "Certificate not found")
|
|
if self.command == "GET" and len(parts) == 5:
|
|
payload = row_dict(cert)
|
|
payload["public_url"] = f"http://127.0.0.1:8000/cert/{cert['public_token']}"
|
|
payload["verify_url"] = f"http://127.0.0.1:8000/verify/{cert['qr_token']}"
|
|
return self.json(payload)
|
|
if path.endswith("/preview"):
|
|
return self.json(public_payload(conn, cert))
|
|
if path.endswith("/download"):
|
|
if cert["status"] != "valid":
|
|
return self.error(403, "\u8bc1\u4e66\u5df2\u4f5c\u5e9f\uff0c\u4e0d\u652f\u6301\u4e0b\u8f7d\u6b63\u5e38 PDF")
|
|
return self.send_file(build_pdf(conn, cert), "application/pdf", f"{cert['certificate_no']}.pdf")
|
|
if path.endswith("/void"):
|
|
conn.execute("update certificates set status='voided',pdf_status='not_generated' where id=?", (cert_id,))
|
|
learner = conn.execute("select * from learners where id=?", (cert["learner_id"],)).fetchone()
|
|
log(conn, admin_id, "void_certificate", "certificate", cert_id, {"certificate_no": cert["certificate_no"], "learner_name": learner["current_name"] if learner else "", "previous_status": cert["status"], "status": "voided"})
|
|
return self.json(row_dict(conn.execute("select * from certificates where id=?", (cert_id,)).fetchone()))
|
|
if path.endswith("/token/regenerate"):
|
|
conn.execute("update certificates set public_token=? where id=?", (secrets.token_urlsafe(24), cert_id))
|
|
learner = conn.execute("select * from learners where id=?", (cert["learner_id"],)).fetchone()
|
|
log(conn, admin_id, "regenerate_public_token", "certificate", cert_id, {"certificate_no": cert["certificate_no"], "learner_name": learner["current_name"] if learner else ""})
|
|
return self.json(row_dict(conn.execute("select * from certificates where id=?", (cert_id,)).fetchone()))
|
|
|
|
if path == "/api/admin/import-batches/template":
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.append(COLS)
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
return self.bytes(buf.getvalue(), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "certificate-import-template.xlsx")
|
|
if path == "/api/admin/import-batches" and self.command == "GET":
|
|
return self.json(rows(conn, "select * from import_batches order by id desc"))
|
|
if path == "/api/admin/import-batches" and self.command == "POST":
|
|
filename, content = parse_multipart(body, self.headers.get("Content-Type", ""))
|
|
uploads = DATA / "uploads"
|
|
uploads.mkdir(exist_ok=True)
|
|
file_path = uploads / filename
|
|
file_path.write_bytes(content)
|
|
batch = validate_import(conn, file_path, filename)
|
|
log(conn, admin_id, "upload_import_file", "import_batch", batch["id"], {"filename": filename, "total_rows": batch["total_rows"], "valid_rows": batch["valid_rows"], "failed_rows": batch["failed_rows"], "status": batch["status"]})
|
|
return self.json(batch, 201)
|
|
if path.endswith("/confirm") and path.startswith("/api/admin/import-batches/"):
|
|
batch_id = int(path.split("/")[4])
|
|
return self.json(import_batch(conn, batch_id, admin_id))
|
|
if path.endswith("/file") and path.startswith("/api/admin/import-batches/"):
|
|
batch_id = int(path.split("/")[4])
|
|
batch = conn.execute("select * from import_batches where id=?", (batch_id,)).fetchone()
|
|
if not batch:
|
|
return self.error(404, "Import batch not found")
|
|
file_path = DATA / "uploads" / batch["filename"]
|
|
if not file_path.exists():
|
|
return self.error(404, "\u539f\u59cb\u6587\u4ef6\u4e0d\u5b58\u5728")
|
|
return self.send_file(file_path, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", batch["filename"])
|
|
if path.startswith("/api/admin/import-batches/") and self.command == "DELETE":
|
|
batch_id = int(path.rsplit("/", 1)[1])
|
|
batch = conn.execute("select * from import_batches where id=?", (batch_id,)).fetchone()
|
|
if not batch:
|
|
return self.error(404, "Import batch not found")
|
|
for file_path in [DATA / "uploads" / batch["filename"], DATA / f"batch-{batch_id}.json"]:
|
|
if file_path.exists():
|
|
file_path.unlink()
|
|
conn.execute("delete from import_batches where id=?", (batch_id,))
|
|
log(conn, admin_id, "delete_import_batch", "import_batch", batch_id, {"filename": batch["filename"], "status": batch["status"], "total_rows": batch["total_rows"]})
|
|
return self.json({"ok": True})
|
|
if path == "/api/admin/exports/certificates":
|
|
return self.bytes(export_certs(conn), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "certificates-export.xlsx")
|
|
if path == "/api/admin/logs/export":
|
|
return self.bytes(export_logs(filtered_operation_logs(conn, query)), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "operation-logs.xlsx")
|
|
if path == "/api/admin/logs/cleanup" and self.command == "POST":
|
|
removed = cleanup_expired_logs(conn)
|
|
return self.json({"removed": removed, "normal_retention_days": LOG_RETENTION_DAYS, "high_risk_retention_days": HIGH_RISK_LOG_RETENTION_DAYS})
|
|
if path == "/api/admin/logs":
|
|
cleanup_expired_logs(conn)
|
|
return self.json(list_operation_logs(conn, query))
|
|
return self.error(404, "Not found")
|
|
|
|
def admin_id(self) -> int | None:
|
|
auth = self.headers.get("Authorization", "")
|
|
return token_admin_id(auth[7:]) if auth.startswith("Bearer ") else None
|
|
|
|
def read_body(self) -> bytes:
|
|
return self.rfile.read(int(self.headers.get("Content-Length", "0") or "0"))
|
|
|
|
def json(self, value, status_code=200):
|
|
self.bytes(json_bytes(value), "application/json; charset=utf-8", status_code=status_code)
|
|
|
|
def bytes(self, data: bytes, content_type: str, filename: str | None = None, status_code=200):
|
|
self.send_response(status_code)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Cache-Control", "no-store")
|
|
if filename:
|
|
self.send_header("Content-Disposition", f'attachment; filename="{filename}"')
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def send_file(self, path: Path, content_type: str, filename: str | None = None):
|
|
self.bytes(path.read_bytes(), content_type, filename)
|
|
|
|
def error(self, status_code, message):
|
|
self.bytes(json_bytes({"detail": str(message)}), "application/json; charset=utf-8", status_code=status_code)
|
|
|
|
|
|
def main():
|
|
init_db()
|
|
server = ThreadingHTTPServer(("127.0.0.1", 8000), Handler)
|
|
print("Local app running at http://127.0.0.1:8000")
|
|
print("Admin: admin / Admin123!")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|