104 lines
4.5 KiB
Python
104 lines
4.5 KiB
Python
"""Misc 路由:/、/healthz、/v1/changelog、企微域名校验文件、/v1/me。"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from core import __version__
|
|
from core.paths import ROOT
|
|
|
|
from ..auth import get_user_profile
|
|
from ..common import BRAND, INSTANCE
|
|
|
|
# 用户版更新日志:仓库根 CHANGELOG.md(面向用户的口径,区别于工程笔记 PROGRESS.md)。
|
|
# 按文件 mtime 缓存解析结果 —— 热改文件即生效,不用重启。条目头格式 `## <版本> — <日期>`。
|
|
_CHANGELOG_PATH = ROOT / "CHANGELOG.md"
|
|
_CHANGELOG_HEAD = re.compile(r"^##\s+v?(\d[\w.\-]*)\s*(?:[—–-]+\s*(\S+))?\s*$")
|
|
_changelog_cache: dict[str, Any] = {"mtime": None, "entries": []}
|
|
|
|
|
|
def _parse_changelog(text: str) -> list[dict[str, Any]]:
|
|
"""CHANGELOG.md → [{version, date, body_md}](文件序,即新在前);首个 `## ` 前的导语丢弃。"""
|
|
entries: list[dict[str, Any]] = []
|
|
body: list[str] = []
|
|
for line in text.splitlines():
|
|
m = _CHANGELOG_HEAD.match(line)
|
|
if m:
|
|
body = []
|
|
entries.append({"version": m.group(1), "date": m.group(2), "body_md": body})
|
|
elif entries:
|
|
body.append(line)
|
|
for e in entries:
|
|
e["body_md"] = "\n".join(e["body_md"]).strip()
|
|
return entries
|
|
|
|
|
|
def register_misc_routes(app, *, require_user) -> None:
|
|
@app.get("/", include_in_schema=False)
|
|
def root():
|
|
# 本地 dev SPA;Swagger UI 仍在 /docs
|
|
return RedirectResponse(url="/static/dev.html", status_code=302)
|
|
|
|
@app.get("/healthz", tags=["misc"])
|
|
def healthz():
|
|
# instance:蓝绿部署时验证打到的是哪个实例(blue/green);单实例为 null
|
|
# brand:对外品牌名(ZCBOT_BRAND_NAME),前端 boot 拉取覆盖页面标题 / 顶栏 / 登录卡
|
|
return {"status": "ok", "version": __version__, "instance": INSTANCE or None,
|
|
"brand": BRAND}
|
|
|
|
@app.get("/v1/changelog", tags=["misc"])
|
|
def changelog(limit: int = 20):
|
|
"""用户版更新日志(仓库根 CHANGELOG.md 解析,新在前)。
|
|
|
|
公开端点(无鉴权):内容本就是写给用户看的,登录页脚也展示版本号。
|
|
只记用户可感知的变化,当前版本可能没有专属条目(纯内部改动不记)。
|
|
前端:点左栏底部版本号弹层展示,配 localStorage 红点提示新版本。
|
|
"""
|
|
limit = max(1, min(limit, 100))
|
|
try:
|
|
mtime = _CHANGELOG_PATH.stat().st_mtime
|
|
except OSError:
|
|
return {"current": __version__, "entries": []}
|
|
if _changelog_cache["mtime"] != mtime:
|
|
_changelog_cache["entries"] = _parse_changelog(
|
|
_CHANGELOG_PATH.read_text(encoding="utf-8"))
|
|
_changelog_cache["mtime"] = mtime
|
|
return {"current": __version__, "entries": _changelog_cache["entries"][:limit]}
|
|
|
|
@app.get("/WW_verify_{token}.txt", include_in_schema=False)
|
|
def wecom_domain_verify(token: str):
|
|
"""企业微信「网页授权可信域名」归属校验文件(公开,无需登录)。
|
|
把企业微信下载的 WW_verify_<token>.txt 放到 ZCBOT_WECOM_VERIFY_DIR
|
|
(默认 repo 根)下,本路由按文件名在域名根 serve。"""
|
|
from fastapi.responses import PlainTextResponse
|
|
if not token.isalnum(): # 防路径穿越
|
|
raise HTTPException(404, "not found")
|
|
vdir = Path(os.getenv("ZCBOT_WECOM_VERIFY_DIR", "").strip() or str(ROOT))
|
|
fpath = vdir / f"WW_verify_{token}.txt"
|
|
if not fpath.is_file():
|
|
raise HTTPException(404, "verify file not found")
|
|
return PlainTextResponse(fpath.read_text(encoding="utf-8"))
|
|
|
|
@app.get("/v1/me", tags=["misc"])
|
|
def me(user_id: UUID = Depends(require_user)):
|
|
"""当前登录用户身份(JWT → user_id → DB role)。
|
|
|
|
前端 dev SPA 用 localStorage 里的 token 恢复会话时调一次,据 role=='admin'
|
|
决定显不显"管理"入口(/static/admin.html)。role 走 DB 查,改完即时生效。
|
|
name / user_name 是平台登录注入的档案(0016),可能为 null(邮箱密码 / 历史行)。
|
|
"""
|
|
prof = get_user_profile(user_id) or {}
|
|
return {
|
|
"user_id": str(user_id),
|
|
"role": prof.get("role", "user"),
|
|
"name": prof.get("name"),
|
|
"user_name": prof.get("user_name"),
|
|
"email": prof.get("email"),
|
|
}
|