增加手机号短信验证查询逻辑

local_app/server.py

@@ -3,6 +3,7 @@ import hashlib
 import hmac
 import io
 import json
+import random
 import secrets
 import sqlite3
 import time
@@ -27,6 +28,7 @@ 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
 
@@ -312,11 +314,37 @@ 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)
-    image = Image.new("RGB", (132, 44), "#f8fafc")
+    
+    # 增大图片尺寸，提高可读性
+    width, height = 180, 56
+    image = Image.new("RGB", (width, height), "#f0f7f7")
     draw = ImageDraw.Draw(image)
-    font = ImageFont.load_default()
+    
+    # 使用更大的字体
+    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):
-        draw.text((18 + i * 26, 14), ch, fill="#111827", font=font)
+        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()}
@@ -327,6 +355,41 @@ def verify_captcha(captcha_id: str, code: str) -> bool:
     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)
@@ -361,7 +424,13 @@ def public_payload(conn: sqlite3.Connection, cert: sqlite3.Row) -> dict:
 
 
 def font_name() -> str:
-    for path in [r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc"]:
+    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))
@@ -373,13 +442,32 @@ def font_name() -> str:
 
 def pil_font(size: int, kind: str = "song", bold: bool = False) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
     choices = {
-        "kai": [r"C:\Windows\Fonts\simkai.ttf", r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc"],
-        "hei": [r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\msyh.ttc"],
-        "song": [r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc"],
+        "kai": [
+            r"C:\Windows\Fonts\simkai.ttf", r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc",
+            "/System/Library/Fonts/Supplemental/Kaiti.ttc",
+            "/System/Library/Fonts/PingFang.ttc",
+            "/System/Library/Fonts/Supplemental/Songti.ttc",
+        ],
+        "hei": [
+            r"C:\Windows\Fonts\simhei.ttf", r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\msyh.ttc",
+            "/System/Library/Fonts/STHeiti Medium.ttc",
+            "/System/Library/Fonts/Supplemental/Heiti.ttc",
+            "/System/Library/Fonts/PingFang.ttc",
+        ],
+        "song": [
+            r"C:\Windows\Fonts\msyh.ttc", r"C:\Windows\Fonts\simsun.ttc",
+            "/System/Library/Fonts/Supplemental/Songti.ttc",
+            "/System/Library/Fonts/PingFang.ttc",
+            "/System/Library/Fonts/STHeiti Light.ttc",
+        ],
     }
     paths = choices.get(kind, choices["song"])
     if bold and kind == "song":
-        paths = [r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf"] + paths
+        paths = [
+            r"C:\Windows\Fonts\msyhbd.ttc", r"C:\Windows\Fonts\simhei.ttf",
+            "/System/Library/Fonts/STHeiti Medium.ttc",
+            "/System/Library/Fonts/Supplemental/Heiti.ttc",
+        ] + paths
     for item in paths:
         if Path(item).exists():
             try:
@@ -513,7 +601,8 @@ def render_certificate_image(conn: sqlite3.Connection, cert: sqlite3.Row) -> Ima
     x = draw_inline(draw, x, y, issue_day, 24, 4, "kai", True)
     x = draw_inline(draw, x, y, "\u65e5\u5b8c\u6210\u4e86", 24, 8)
     course_text = f"\u201c{cert['course_name'] or cert['certificate_name'] or cert['project_code']}\u201d"
-    stage = cert["stage_name"] or "\u521d\u7ea7\u8bfe\u7a0b\u7684\u4e13\u4e1a\u5b66\u4e60\u3002"
+    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)
 
     issue_label = f"\u53d1\u8bc1\u65e5\u671f\uff1a{issue_year} \u5e74 {issue_month} \u6708 {issue_day} \u65e5"
@@ -705,6 +794,15 @@ class Handler(BaseHTTPRequestHandler):
                 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", "")):
@@ -729,6 +827,24 @@ class Handler(BaseHTTPRequestHandler):
                     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()