feat: 新增 /admin 只读授权管理后台

- admin.py：HTTP Basic Auth + 服务端渲染单页，概览统计 / 用户与当前授权 /
  最近使用记录 / 会话令牌 / 扫码场景五个区块
- ADMIN_PASSWORD 为空时返回 503 而非放行，避免漏配密码导致全库数据公开
- 不引模板引擎与 CDN，所有入库文本经 html.escape；令牌只显示首尾便于比对
- config.py / .env.example 新增 ADMIN_USER / ADMIN_PASSWORD
- wechat.py 挂载 admin.router；CODEBUDDY.md 补充文档

.env.example

@@ -30,3 +30,8 @@ SCENE_TTL_SECONDS=300
 SESSION_TTL_HOURS=24
 # 时间授权 usage_logs 节流窗口（秒）
 USAGE_LOG_THROTTLE_SECONDS=60
+
+# 管理后台 /admin 的 HTTP Basic Auth 凭据
+# ADMIN_PASSWORD 留空时 /admin 返回 503（不鉴权放行），务必填一个长随机串
+ADMIN_USER=admin
+ADMIN_PASSWORD=

CODEBUDDY.md

@@ -4,7 +4,7 @@ This file provides guidance to CodeBuddy Code when working with code in this rep
 
 ## 项目概述
 
-微信公众号扫码授权服务，基于 FastAPI，为 Windows MFC 桌面程序提供微信扫码授权与使用扣减。已实现微信服务器验证与消息/事件接收（`/wechat`）、扫码授权（`/auth`）、使用扣减（`/usage`）以及首次关注赠送 7 天免费授权。MySQL 已接入（aiomysql 连接池，5 张表）；Redis 仅在配置层声明，尚未使用。
+微信公众号扫码授权服务，基于 FastAPI，为 Windows MFC 桌面程序提供微信扫码授权与使用扣减。已实现微信服务器验证与消息/事件接收（`/wechat`）、扫码授权（`/auth`）、使用扣减（`/usage`）、只读管理后台（`/admin`）以及首次关注赠送 7 天免费授权。MySQL 已接入（aiomysql 连接池，5 张表）；Redis 仅在配置层声明，尚未使用。
 
 ## 常用命令
 
@@ -38,6 +38,7 @@ wechat_api.py   # 微信开放接口：access_token 内存缓存 + 临时二维
 wechat.py       # FastAPI app 本体：lifespan + /wechat 路由 + 签名校验 + XML 解析/构造
 auth.py         # /auth/* 路由 + 扫码授权业务逻辑（含 pending 激活）
 usage.py        # /usage/consume 路由：会话校验与使用扣减
+admin.py        # /admin 只读管理后台：Basic Auth + 服务端渲染 HTML
 run_local.py    # 开发启动入口（reload=True）
 run_server.py   # 生产启动入口（reload=False, workers=1）
 sql/schema.sql  # 建表脚本（5 张表，含索引与外键）
@@ -45,7 +46,7 @@ sql/schema.sql  # 建表脚本（5 张表，含索引与外键）
 
 - **配置**：所有敏感值经 `config.py` 从 `.env` 读取，`.env` 已被 `.gitignore` 排除。`.env.example` 是字段模板。新增配置项需同时更新这两处。
 - **应用入口**：两个启动脚本均以 `"wechat:app"` 字符串形式加载 `wechat.py` 中的 `app`，因此模块名/对象名不可随意重命名。`lifespan` 在启动时初始化 MySQL 连接池——**MySQL 不可达或库表不存在时应用会直接启动失败**。
-- **路由挂载**：`wechat.py` 通过 `include_router` 挂载 `auth.router` 与 `usage.router`。新增 MFC 侧接口应新建独立模块的 router，而不是塞进 `wechat.py`。
+- **路由挂载**：`wechat.py` 通过 `include_router` 挂载 `auth.router`、`usage.router` 与 `admin.router`。新增 MFC 侧接口应新建独立模块的 router，而不是塞进 `wechat.py`。
 - **微信交互协议**：
   - 所有请求先经 `verify_signature()`（token+timestamp+nonce 字典序拼接后 SHA1 比对）校验，失败返回 403。
   - GET 校验通过后原样返回 `echostr`。
@@ -63,6 +64,17 @@ sql/schema.sql  # 建表脚本（5 张表，含索引与外键）
 - **usage_logs**：积分授权每次调用都写（计费凭证）；时间授权按 `(user_id, device_id)` 在 `USAGE_LOG_THROTTLE_SECONDS`（默认 60 秒）窗口内节流。
 - **失败语义**：`/usage/consume` 一律返回 HTTP 200，用 `{"ok": false, "reason": "expired | exhausted | invalid_token"}` 表达失败。
 
+## 管理后台（/admin）
+
+只读单页，用于查看用户与授权现状：概览统计、用户+当前授权、最近使用记录、会话令牌、扫码场景。
+
+- **鉴权**：HTTP Basic Auth，凭据取自 `.env` 的 `ADMIN_USER` / `ADMIN_PASSWORD`，用 `secrets.compare_digest` 做定时安全比较。
+- **`ADMIN_PASSWORD` 为空时 `/admin` 返回 503 而非放行**——不要把它当成可选项，否则等于把全库用户数据公开。
+- **只提供 GET**，没有任何写操作（改授权/加积分/封号一律不做）。
+- **不引模板引擎**：HTML 由 f-string 拼装，所有入库字段经 `html.escape()`；CSS 内联，不依赖任何 CDN（服务器出网不可靠）。
+- 页脚会显示数据库名；页头显示数据生成时间。不做自动刷新。
+- `admin.py` 被 `wechat.py` import，因此**不要在 `admin.py` 里反向 import `wechat`**（循环依赖）。
+
 ## 数据库
 
 5 张表（定义见 `sql/schema.sql`）：`users`、`authorizations`（授权，type 分 time/points）、`auth_scenes`（扫码场景，300 秒一次性）、`usage_logs`（使用日志）、`sessions`（会话令牌）。

admin.py

@@ -0,0 +1,382 @@
+"""
+授权管理后台（/admin）
+
+单页、只读、服务端渲染：概览统计 + 用户与当前授权 + 最近使用记录 + 会话与扫码场景。
+
+鉴权：HTTP Basic Auth，凭据来自 .env 的 ADMIN_USER / ADMIN_PASSWORD。
+      ADMIN_PASSWORD 为空时整体返回 503（而不是放行）——避免「忘了配密码 = 谁都能进」。
+      只提供 GET，不提供任何写操作。
+
+刻意不引入 Jinja2 / Bootstrap / 任何 CDN 资源：
+  - 服务器出网不可靠，页面不能依赖外部资源；
+  - 少一个依赖就少一份部署风险。
+所有来自数据库的文本都经 html.escape()，避免昵称之类的字段注入 HTML。
+
+注意：本模块被 wechat.py import 并挂载，反向 import wechat 会造成循环依赖。
+"""
+
+import html
+import logging
+import secrets
+from datetime import datetime
+
+import aiomysql
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.responses import HTMLResponse
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+
+import db
+from config import ADMIN_PASSWORD, ADMIN_USER, MYSQL_DB
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/admin", tags=["admin"])
+
+# realm 必须是 ASCII：Starlette 按 latin-1 编码响应头，中文会直接抛异常
+_basic = HTTPBasic(realm="CadOcr Admin")
+
+
+def require_admin(credentials: HTTPBasicCredentials = Depends(_basic)) -> None:
+    """校验 Basic Auth；不匹配一律 401 并带 WWW-Authenticate（浏览器才会弹密码框）"""
+    if not ADMIN_PASSWORD:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="管理后台未启用：请在 .env 里配置 ADMIN_PASSWORD",
+        )
+
+    # 比字节而不是比字符串：compare_digest 对含非 ASCII 的 str 会直接抛 TypeError
+    user_ok = secrets.compare_digest(
+        credentials.username.encode("utf-8"), ADMIN_USER.encode("utf-8")
+    )
+    pass_ok = secrets.compare_digest(
+        credentials.password.encode("utf-8"), ADMIN_PASSWORD.encode("utf-8")
+    )
+    if not (user_ok and pass_ok):
+        # 只记用户名，绝不记密码
+        logger.warning("管理后台鉴权失败 user=%s", credentials.username)
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Unauthorized",
+            headers={"WWW-Authenticate": 'Basic realm="CadOcr Admin"'},
+        )
+
+
+# ---------------------------------------------------------------------------
+# 查询
+# ---------------------------------------------------------------------------
+
+OVERVIEW_SQL = """
+SELECT
+  (SELECT COUNT(*) FROM users)                                     AS users_total,
+  (SELECT COUNT(*) FROM users WHERE has_claimed_free = 1)          AS users_free,
+  (SELECT COUNT(*) FROM users WHERE created_at >= CURDATE())       AS users_today,
+  (SELECT COUNT(*) FROM authorizations WHERE status = 'active')    AS auth_active,
+  (SELECT COUNT(*) FROM authorizations WHERE status = 'pending')   AS auth_pending,
+  (SELECT COUNT(*) FROM authorizations WHERE status = 'expired')   AS auth_expired,
+  (SELECT COUNT(*) FROM authorizations WHERE status = 'exhausted') AS auth_exhausted,
+  (SELECT COUNT(*) FROM authorizations WHERE type = 'time')        AS auth_time,
+  (SELECT COUNT(*) FROM authorizations WHERE type = 'points')      AS auth_points,
+  (SELECT COUNT(*) FROM usage_logs)                                AS usage_total,
+  (SELECT COUNT(*) FROM usage_logs WHERE used_at >= CURDATE())     AS usage_today,
+  (SELECT COUNT(*) FROM sessions WHERE expires_at > NOW())         AS sessions_valid
+"""
+
+# 互斥原则保证同一用户最多一条 active，所以这个 LEFT JOIN 不会让用户行翻倍
+USERS_SQL = """
+SELECT
+  u.id, u.openid, u.nickname, u.created_at, u.last_seen_at, u.has_claimed_free,
+  a.type, a.source, a.end_at, a.remaining_points, a.total_points,
+  TIMESTAMPDIFF(SECOND, NOW(), a.end_at) AS remain_seconds,
+  (SELECT COUNT(*) FROM authorizations x WHERE x.user_id = u.id) AS auth_count
+FROM users u
+LEFT JOIN authorizations a ON a.user_id = u.id AND a.status = 'active'
+ORDER BY u.id DESC
+LIMIT 200
+"""
+
+USAGE_SQL = """
+SELECT l.id, l.user_id, u.openid, l.device_id, l.cost_type, l.cost_points, l.used_at
+FROM usage_logs l
+LEFT JOIN users u ON u.id = l.user_id
+ORDER BY l.id DESC
+LIMIT 50
+"""
+
+SESSIONS_SQL = """
+SELECT s.id, s.user_id, s.token, u.openid, s.device_id, s.created_at, s.expires_at,
+       (s.expires_at > NOW()) AS still_valid
+FROM sessions s
+LEFT JOIN users u ON u.id = s.user_id
+ORDER BY s.id DESC
+LIMIT 50
+"""
+
+SCENES_SQL = """
+SELECT id, scene_str, device_id, status, user_id, created_at, expires_at, authorized_at
+FROM auth_scenes
+ORDER BY id DESC
+LIMIT 50
+"""
+
+
+# ---------------------------------------------------------------------------
+# 展示辅助
+# ---------------------------------------------------------------------------
+
+def _e(value) -> str:
+    """转义后用于 HTML 文本位置；None 显示为占位符"""
+    if value is None:
+        return "—"
+    return html.escape(str(value))
+
+
+def _dt(value) -> str:
+    """DATETIME → 字符串（aiomysql 返回的是 naive datetime，按原样显示即可）"""
+    if value is None:
+        return "—"
+    if isinstance(value, datetime):
+        return value.strftime("%Y-%m-%d %H:%M:%S")
+    return html.escape(str(value))
+
+
+def _token(value) -> str:
+    """会话令牌是活的凭据，页面只显示首尾便于比对，不给全量"""
+    if not value:
+        return "—"
+    s = str(value)
+    if len(s) <= 16:
+        return html.escape(s)
+    return html.escape(f"{s[:10]}…{s[-4:]}")
+
+
+def _human_duration(seconds) -> str:
+    if seconds is None:
+        return "—"
+    seconds = int(seconds)
+    if seconds <= 0:
+        return "已过期"
+    days, rem = divmod(seconds, 86400)
+    hours, rem = divmod(rem, 3600)
+    minutes = rem // 60
+    if days:
+        return f"{days} 天 {hours} 小时"
+    if hours:
+        return f"{hours} 小时 {minutes} 分"
+    return f"{minutes} 分"
+
+
+def _tag(text: str, kind: str) -> str:
+    return f'<span class="tag {kind}">{_e(text)}</span>'
+
+
+def _card(number, label: str) -> str:
+    return f'<div class="card"><div class="n">{_e(number)}</div><div class="l">{_e(label)}</div></div>'
+
+
+def _table(headers: list[str], rows: list[list[str]], empty_text: str) -> str:
+    if not rows:
+        return f'<div class="empty">{_e(empty_text)}</div>'
+    head = "".join(f"<th>{_e(h)}</th>" for h in headers)
+    body = "".join("<tr>" + "".join(f"<td>{c}</td>" for c in r) + "</tr>" for r in rows)
+    return f"<table><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>"
+
+
+def _render_overview(ov: dict) -> str:
+    cards = [
+        _card(ov["users_total"], "用户总数"),
+        _card(ov["users_today"], "今日新增用户"),
+        _card(ov["users_free"], "已领免费授权"),
+        _card(ov["auth_active"], "生效中授权"),
+        _card(ov["auth_pending"], "待生效授权"),
+        _card(ov["auth_expired"], "已过期授权"),
+        _card(ov["auth_exhausted"], "已耗尽授权"),
+        _card(ov["auth_time"], "时间授权"),
+        _card(ov["auth_points"], "积分授权"),
+        _card(ov["usage_total"], "使用总次数"),
+        _card(ov["usage_today"], "今日使用次数"),
+        _card(ov["sessions_valid"], "有效会话"),
+    ]
+    return '<div class="cards">' + "".join(cards) + "</div>"
+
+
+def _render_users(rows: list[dict]) -> str:
+    out = []
+    for r in rows:
+        if r["type"] is None:
+            current = _tag("无可用授权", "none")
+            expire = "—"
+        else:
+            kind = "时间" if r["type"] == "time" else "积分"
+            source = {"free": "免费", "purchase": "购买", "admin": "后台"}.get(r["source"], r["source"])
+            current = f"{_e(kind)} · {_e(source)} " + _tag("生效中", "active")
+            if r["type"] == "time":
+                expire = f"{_dt(r['end_at'])}<br><span class=\"dim\">剩 {_e(_human_duration(r['remain_seconds']))}</span>"
+            else:
+                expire = f"{_e(r['remaining_points'])} / {_e(r['total_points'])} 次"
+
+        out.append([
+            _e(r["id"]),
+            f'<code>{_e(r["openid"])}</code>',
+            _e(r["nickname"]),
+            _dt(r["created_at"]),
+            _dt(r["last_seen_at"]),
+            _tag("是", "active") if r["has_claimed_free"] else _tag("否", "expired"),
+            current,
+            expire,
+            _e(r["auth_count"]),
+        ])
+    return _table(
+        ["ID", "openid", "昵称", "首次关注", "最近活跃", "已领免费", "当前授权", "到期 / 余量", "授权条数"],
+        out,
+        "还没有任何用户",
+    )
+
+
+def _render_usage(rows: list[dict]) -> str:
+    out = []
+    for r in rows:
+        kind = "时间" if r["cost_type"] == "time" else "积分"
+        out.append([
+            _e(r["id"]),
+            _e(r["user_id"]),
+            f'<code>{_e(r["openid"])}</code>',
+            f'<code>{_e(r["device_id"])}</code>',
+            _e(kind),
+            _e(r["cost_points"]),
+            _dt(r["used_at"]),
+        ])
+    return _table(
+        ["ID", "用户ID", "openid", "设备", "计费类型", "扣减", "时间"],
+        out,
+        "还没有使用记录",
+    )
+
+
+def _render_sessions(rows: list[dict]) -> str:
+    out = []
+    for r in rows:
+        out.append([
+            _e(r["id"]),
+            _e(r["user_id"]),
+            f'<code>{_e(r["openid"])}</code>',
+            f'<code>{_e(r["device_id"])}</code>',
+            f'<code>{_token(r["token"])}</code>',
+            _dt(r["created_at"]),
+            _dt(r["expires_at"]),
+            _tag("有效", "active") if r["still_valid"] else _tag("已过期", "expired"),
+        ])
+    return _table(
+        ["ID", "用户ID", "openid", "设备", "令牌", "签发时间", "过期时间", "状态"],
+        out,
+        "还没有会话",
+    )
+
+
+def _render_scenes(rows: list[dict]) -> str:
+    kind_map = {"pending": "pending", "scanned": "pending", "authorized": "active", "expired": "expired"}
+    out = []
+    for r in rows:
+        out.append([
+            _e(r["id"]),
+            f'<code>{_e(r["scene_str"])}</code>',
+            f'<code>{_e(r["device_id"])}</code>',
+            _tag(r["status"], kind_map.get(r["status"], "expired")),
+            _e(r["user_id"]),
+            _dt(r["created_at"]),
+            _dt(r["expires_at"]),
+            _dt(r["authorized_at"]),
+        ])
+    return _table(
+        ["ID", "scene_str", "设备", "状态", "用户ID", "创建时间", "过期时间", "授权完成时间"],
+        out,
+        "还没有扫码场景",
+    )
+
+
+_CSS = """
+* { box-sizing: border-box; }
+body {
+  font-family: -apple-system, "Segoe UI", "Microsoft YaHei", sans-serif;
+  margin: 0; padding: 24px 28px 48px; background: #f5f6f8; color: #1f2328;
+  font-size: 14px; line-height: 1.5;
+}
+h1 { font-size: 20px; margin: 0 0 2px; }
+h2 { font-size: 15px; margin: 30px 0 8px; padding-bottom: 6px; border-bottom: 1px solid #e3e5e8; }
+.meta { color: #6b7280; font-size: 12px; margin-bottom: 4px; }
+.meta a { color: #2563eb; text-decoration: none; }
+.meta a:hover { text-decoration: underline; }
+.cards { display: flex; flex-wrap: wrap; gap: 10px; }
+.card { background: #fff; border: 1px solid #e3e5e8; border-radius: 6px; padding: 10px 16px; min-width: 104px; }
+.card .n { font-size: 22px; font-weight: 600; font-variant-numeric: tabular-nums; }
+.card .l { font-size: 12px; color: #6b7280; margin-top: 2px; white-space: nowrap; }
+table { border-collapse: collapse; width: 100%; background: #fff; font-size: 12.5px; }
+th, td { border: 1px solid #e3e5e8; padding: 5px 9px; text-align: left; vertical-align: top; }
+th { background: #f0f1f3; font-weight: 600; white-space: nowrap; }
+code { font-family: Consolas, "Courier New", monospace; font-size: 12px; word-break: break-all; }
+.dim { color: #9ca3af; }
+.tag { display: inline-block; padding: 1px 7px; border-radius: 3px; font-size: 11px; white-space: nowrap; }
+.tag.active { background: #dcfce7; color: #166534; }
+.tag.pending { background: #fef9c3; color: #854d0e; }
+.tag.expired, .tag.exhausted, .tag.cancelled { background: #f3f4f6; color: #6b7280; }
+.tag.none { background: #fee2e2; color: #991b1b; }
+.empty { color: #9ca3af; font-size: 13px; padding: 8px 2px; }
+footer { margin-top: 36px; color: #9ca3af; font-size: 12px; }
+"""
+
+
+# ---------------------------------------------------------------------------
+# 路由
+# ---------------------------------------------------------------------------
+
+@router.get("", response_class=HTMLResponse)
+async def admin_page(_: None = Depends(require_admin)) -> HTMLResponse:
+    """管理页：一次拉齐四个区块的数据后整页渲染"""
+    async with db.acquire() as conn:
+        async with conn.cursor(aiomysql.DictCursor) as cur:
+            await cur.execute(OVERVIEW_SQL)
+            overview = await cur.fetchone()
+
+            await cur.execute(USERS_SQL)
+            users = await cur.fetchall()
+
+            await cur.execute(USAGE_SQL)
+            usage = await cur.fetchall()
+
+            await cur.execute(SESSIONS_SQL)
+            sessions = await cur.fetchall()
+
+            await cur.execute(SCENES_SQL)
+            scenes = await cur.fetchall()
+
+    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    page = f"""<!doctype html>
+<html lang="zh-CN">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>CadOcr 授权管理</title>
+<style>{_CSS}</style>
+</head>
+<body>
+<h1>CadOcr 微信授权管理</h1>
+<div class="meta">数据生成时间 {_e(now)} · <a href="">刷新</a></div>
+
+<h2>概览</h2>
+{_render_overview(overview)}
+
+<h2>用户与当前授权（最多 200 条，按 ID 倒序）</h2>
+{_render_users(users)}
+
+<h2>最近使用记录（最多 50 条）</h2>
+{_render_usage(usage)}
+
+<h2>会话令牌（最多 50 条）</h2>
+{_render_sessions(sessions)}
+
+<h2>扫码场景（最多 50 条）</h2>
+{_render_scenes(scenes)}
+
+<footer>只读页面，不提供任何修改操作 · 数据库 {_e(MYSQL_DB)}</footer>
+</body>
+</html>"""
+    return HTMLResponse(content=page)

config.py

@@ -32,3 +32,8 @@ SCENE_TTL_SECONDS = int(os.getenv("SCENE_TTL_SECONDS", "300"))  # 二维码/scen
 SESSION_TTL_HOURS = int(os.getenv("SESSION_TTL_HOURS", "24"))   # session_token 有效期（小时）
 # 时间授权 usage_logs 节流窗口（秒）：同一用户+设备在此窗口内只记一条
 USAGE_LOG_THROTTLE_SECONDS = int(os.getenv("USAGE_LOG_THROTTLE_SECONDS", "60"))
+
+# 管理后台（/admin）的 HTTP Basic Auth 凭据。
+# ADMIN_PASSWORD 为空时 /admin 直接返回 503，绝不放行——避免「忘了配密码 = 谁都能进」。
+ADMIN_USER = os.getenv("ADMIN_USER", "admin")
+ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "")

wechat.py

@@ -4,7 +4,8 @@
 GET  /wechat  - 微信服务器验证（签名校验 + 返回 echostr）
 POST /wechat  - 接收微信推送的消息和事件（subscribe / SCAN 触发扫码授权）
 
-授权接口在 auth.py、使用扣减接口在 usage.py 中定义，通过 include_router 挂载。
+授权接口在 auth.py、使用扣减接口在 usage.py、管理后台在 admin.py 中定义，
+通过 include_router 挂载。
 """
 
 import hashlib
@@ -16,6 +17,7 @@ from contextlib import asynccontextmanager
 from fastapi import FastAPI, HTTPException, Query, Request
 from fastapi.responses import PlainTextResponse
 
+import admin
 import auth
 import db
 import usage
@@ -41,6 +43,7 @@ async def lifespan(app: FastAPI):
 app = FastAPI(title="WeChat API", version="0.3.0", lifespan=lifespan)
 app.include_router(auth.router)
 app.include_router(usage.router)
+app.include_router(admin.router)
 
 
 def verify_signature(signature: str, timestamp: str, nonce: str) -> bool: