chore: sync local updates
This commit is contained in:
@@ -1,23 +1,100 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import List, Literal, Optional, Set
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Query
|
||||
from fastapi import Depends, FastAPI, Header, HTTPException, Query
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
# Markdown 根目录：指向当前后端项目中的 `mengyanote` 文件夹
|
||||
MARKDOWN_ROOT = BASE_DIR / "mengyanote"
|
||||
# ignore.json 文件路径
|
||||
IGNORE_FILE = MARKDOWN_ROOT / "ignore.json"
|
||||
|
||||
|
||||
def _resolve_config_dir() -> Path:
|
||||
"""
|
||||
配置目录（config.json、ignore.json）：
|
||||
优先 MENGYANOTE_CONFIG_DIR；
|
||||
否则若存在 ./mengyanote/config（Docker 将主机 data 挂到 /app/mengyanote 时与笔记并列）则用之；
|
||||
否则 ./data/config（本地开发）。
|
||||
"""
|
||||
env = os.getenv("MENGYANOTE_CONFIG_DIR", "").strip()
|
||||
if env:
|
||||
return Path(env).resolve()
|
||||
beside = BASE_DIR / "mengyanote" / "config"
|
||||
if beside.is_dir():
|
||||
return beside.resolve()
|
||||
return (BASE_DIR / "data" / "config").resolve()
|
||||
|
||||
|
||||
CONFIG_DIR = _resolve_config_dir()
|
||||
CONFIG_FILE = CONFIG_DIR / "config.json"
|
||||
IGNORE_FILE = CONFIG_DIR / "ignore.json"
|
||||
|
||||
# 管理员令牌仅来自 config.json 的 admin_token；按文件 mtime 自动重读
|
||||
_admin_token_cache: Optional[str] = None
|
||||
_config_file_mtime: Optional[float] = None
|
||||
|
||||
|
||||
def _read_admin_token_from_config_file() -> str:
|
||||
"""从 config.json 读取 admin_token，缺省为 shumengya520。"""
|
||||
if not CONFIG_FILE.exists():
|
||||
return "shumengya520"
|
||||
try:
|
||||
with open(CONFIG_FILE, encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
raw = data.get("admin_token")
|
||||
if raw is None:
|
||||
return "shumengya520"
|
||||
s = str(raw).strip()
|
||||
return s if s else "shumengya520"
|
||||
except Exception:
|
||||
return "shumengya520"
|
||||
|
||||
|
||||
def get_admin_token() -> str:
|
||||
"""当前管理员令牌，仅来自 config.json（文件变更后自动生效）。"""
|
||||
global _admin_token_cache, _config_file_mtime
|
||||
try:
|
||||
mtime = CONFIG_FILE.stat().st_mtime if CONFIG_FILE.exists() else 0.0
|
||||
except OSError:
|
||||
mtime = 0.0
|
||||
if _admin_token_cache is None or mtime != _config_file_mtime:
|
||||
_config_file_mtime = mtime
|
||||
_admin_token_cache = _read_admin_token_from_config_file()
|
||||
return _admin_token_cache
|
||||
|
||||
|
||||
def _resolve_markdown_root() -> Path:
|
||||
"""
|
||||
笔记根目录：
|
||||
优先 MENGYANOTE_ROOT；
|
||||
否则若存在 ./mengyanote/mengyanote（主机 data 挂到 /app/mengyanote 时，笔记在子目录 mengyanote 下）则用之；
|
||||
否则若存在 ./mengyanote 则用之（镜像内 COPY 到 /app/mengyanote 的扁平结构）；
|
||||
否则 ./data/mengyanote（本地开发）。
|
||||
"""
|
||||
env = os.getenv("MENGYANOTE_ROOT", "").strip()
|
||||
if env:
|
||||
return Path(env).resolve()
|
||||
p1 = BASE_DIR / "mengyanote"
|
||||
if p1.is_dir():
|
||||
nested = p1 / "mengyanote"
|
||||
if nested.is_dir():
|
||||
return nested.resolve()
|
||||
return p1.resolve()
|
||||
return (BASE_DIR / "data" / "mengyanote").resolve()
|
||||
|
||||
|
||||
# Markdown 根目录（_resolve_markdown_root 已 resolve；供路径穿越校验复用，避免每次请求再 resolve）
|
||||
MARKDOWN_ROOT = _resolve_markdown_root()
|
||||
|
||||
|
||||
def load_ignore_list() -> Set[str]:
|
||||
"""从 ignore.json 加载需要忽略的文件夹列表"""
|
||||
"""从 data/config/ignore.json 加载需要忽略的文件夹列表（与笔记根目录无关）。"""
|
||||
if not IGNORE_FILE.exists():
|
||||
return set()
|
||||
|
||||
@@ -32,6 +109,70 @@ def load_ignore_list() -> Set[str]:
|
||||
# 加载忽略列表
|
||||
IGNORE_LIST = load_ignore_list()
|
||||
|
||||
# 目录树内存缓存：笔记库很大时全库 walk 成本高；短期 TTL 可吞掉重复请求（如前端 StrictMode 双请求）
|
||||
_tree_cache_nodes: Optional[List[DirectoryNode]] = None
|
||||
_tree_cache_until_monotonic: float = 0.0
|
||||
_TREE_CACHE_TTL_SEC = 5.0
|
||||
|
||||
|
||||
def _invalidate_directory_tree_cache() -> None:
|
||||
global _tree_cache_nodes, _tree_cache_until_monotonic
|
||||
_tree_cache_nodes = None
|
||||
_tree_cache_until_monotonic = 0.0
|
||||
|
||||
|
||||
def reload_ignore_list() -> None:
|
||||
"""重新从磁盘加载 IGNORE_LIST（写入 ignore.json 后调用）。"""
|
||||
global IGNORE_LIST
|
||||
IGNORE_LIST = load_ignore_list()
|
||||
_invalidate_directory_tree_cache()
|
||||
|
||||
|
||||
def persist_ignore_list() -> None:
|
||||
"""将当前 IGNORE_LIST 写入 data/config/ignore.json 并刷新内存。"""
|
||||
global IGNORE_LIST
|
||||
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
|
||||
sorted_list = sorted(IGNORE_LIST)
|
||||
with open(IGNORE_FILE, "w", encoding="utf-8") as f:
|
||||
json.dump({"ignore": sorted_list}, f, ensure_ascii=False, indent=4)
|
||||
reload_ignore_list() # 内部已刷新树缓存
|
||||
|
||||
|
||||
def validate_ignore_folder_name(name: str) -> str:
|
||||
"""校验并规范化忽略文件夹名称（仅顶层文件夹名，不含路径）。"""
|
||||
name = name.strip()
|
||||
if not name:
|
||||
raise HTTPException(status_code=400, detail="名称不能为空")
|
||||
if any(sep in name for sep in ("/", "\\")):
|
||||
raise HTTPException(status_code=400, detail="名称不能包含路径分隔符")
|
||||
if name in (".", ".."):
|
||||
raise HTTPException(status_code=400, detail="名称不合法")
|
||||
return name
|
||||
|
||||
|
||||
async def require_admin(
|
||||
x_admin_token: Optional[str] = Header(None, alias="X-Admin-Token"),
|
||||
) -> None:
|
||||
if not x_admin_token or x_admin_token != get_admin_token():
|
||||
raise HTTPException(status_code=401, detail="未授权或令牌无效")
|
||||
|
||||
|
||||
class AdminLoginBody(BaseModel):
|
||||
token: str
|
||||
|
||||
|
||||
class IgnoreListResponse(BaseModel):
|
||||
ignore: List[str]
|
||||
|
||||
|
||||
class AddIgnoreBody(BaseModel):
|
||||
name: str
|
||||
|
||||
|
||||
class UpdateIgnoreBody(BaseModel):
|
||||
old: str
|
||||
new: str
|
||||
|
||||
|
||||
class NodeType(str):
|
||||
FOLDER: Literal["folder"] = "folder"
|
||||
@@ -90,6 +231,14 @@ def should_skip(entry: Path) -> bool:
|
||||
if entry.is_dir() and name in IGNORE_LIST:
|
||||
return True
|
||||
|
||||
# 与「data 挂载为 mengyanote」布局并列的 config 目录不参与目录树
|
||||
if entry.is_dir() and name == "config":
|
||||
try:
|
||||
if entry.parent.resolve() == MARKDOWN_ROOT:
|
||||
return True
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
return False
|
||||
|
||||
|
||||
@@ -165,18 +314,19 @@ def resolve_markdown_path(relative_path: str) -> Path:
|
||||
"""将前端传入的相对路径安全地转换为磁盘路径，防止目录穿越。"""
|
||||
# 统一使用 / 分隔符
|
||||
safe_path = relative_path.replace("\\", "/").lstrip("/")
|
||||
candidate = MARKDOWN_ROOT / safe_path
|
||||
candidate_resolved = (MARKDOWN_ROOT / safe_path).resolve()
|
||||
try:
|
||||
candidate_resolved = candidate.resolve()
|
||||
except FileNotFoundError:
|
||||
candidate_resolved = candidate
|
||||
|
||||
if not str(candidate_resolved).startswith(str(MARKDOWN_ROOT.resolve())):
|
||||
candidate_resolved.relative_to(MARKDOWN_ROOT)
|
||||
except ValueError:
|
||||
raise HTTPException(status_code=400, detail="非法路径")
|
||||
|
||||
return candidate_resolved
|
||||
|
||||
|
||||
def _word_count_non_whitespace(text: str) -> int:
|
||||
"""与原先 replace 链语义一致：不计空格/换行/制表，且不做整串多次拷贝。"""
|
||||
return sum(1 for ch in text if ch not in " \n\r\t")
|
||||
|
||||
|
||||
@app.get("/api/tree", response_model=List[DirectoryNode])
|
||||
def get_directory_tree() -> List[DirectoryNode]:
|
||||
"""
|
||||
@@ -188,7 +338,13 @@ def get_directory_tree() -> List[DirectoryNode]:
|
||||
- type: 'folder' | 'file'
|
||||
- children: 子节点数组
|
||||
"""
|
||||
global _tree_cache_nodes, _tree_cache_until_monotonic
|
||||
now = time.monotonic()
|
||||
if _tree_cache_nodes is not None and now < _tree_cache_until_monotonic:
|
||||
return _tree_cache_nodes
|
||||
tree = build_directory_tree(MARKDOWN_ROOT)
|
||||
_tree_cache_nodes = tree
|
||||
_tree_cache_until_monotonic = now + _TREE_CACHE_TTL_SEC
|
||||
return tree
|
||||
|
||||
|
||||
@@ -215,7 +371,7 @@ def get_markdown_file(path: str = Query(..., description="相对于根目录的
|
||||
file_stat = file_path.stat()
|
||||
|
||||
# 计算字数（去除空格和换行符）
|
||||
word_count = len(content.replace(" ", "").replace("\n", "").replace("\r", "").replace("\t", ""))
|
||||
word_count = _word_count_non_whitespace(content)
|
||||
|
||||
# 获取文件大小（字节）
|
||||
file_size = file_stat.st_size
|
||||
@@ -243,10 +399,78 @@ def get_markdown_file(path: str = Query(..., description="相对于根目录的
|
||||
)
|
||||
|
||||
|
||||
@app.post("/api/admin/login")
|
||||
def admin_login(body: AdminLoginBody):
|
||||
"""验证管理员令牌（仅校验，不在响应中返回密钥）。"""
|
||||
if body.token != get_admin_token():
|
||||
raise HTTPException(status_code=401, detail="令牌错误")
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@app.get("/api/admin/ignore", response_model=IgnoreListResponse)
|
||||
def admin_list_ignore(_: None = Depends(require_admin)) -> IgnoreListResponse:
|
||||
"""列出 ignore.json 中的忽略文件夹名。"""
|
||||
return IgnoreListResponse(ignore=sorted(IGNORE_LIST))
|
||||
|
||||
|
||||
@app.post("/api/admin/ignore", response_model=IgnoreListResponse)
|
||||
def admin_add_ignore(body: AddIgnoreBody, _: None = Depends(require_admin)) -> IgnoreListResponse:
|
||||
"""添加忽略文件夹名。"""
|
||||
global IGNORE_LIST
|
||||
name = validate_ignore_folder_name(body.name)
|
||||
if name in IGNORE_LIST:
|
||||
raise HTTPException(status_code=409, detail="该文件夹已在忽略列表中")
|
||||
IGNORE_LIST.add(name)
|
||||
persist_ignore_list()
|
||||
return IgnoreListResponse(ignore=sorted(IGNORE_LIST))
|
||||
|
||||
|
||||
@app.put("/api/admin/ignore", response_model=IgnoreListResponse)
|
||||
def admin_update_ignore(body: UpdateIgnoreBody, _: None = Depends(require_admin)) -> IgnoreListResponse:
|
||||
"""重命名忽略列表中的一项。"""
|
||||
global IGNORE_LIST
|
||||
old = validate_ignore_folder_name(body.old)
|
||||
new = validate_ignore_folder_name(body.new)
|
||||
if old not in IGNORE_LIST:
|
||||
raise HTTPException(status_code=404, detail="未找到要修改的项")
|
||||
if new in IGNORE_LIST and new != old:
|
||||
raise HTTPException(status_code=409, detail="新名称已存在")
|
||||
IGNORE_LIST.discard(old)
|
||||
IGNORE_LIST.add(new)
|
||||
persist_ignore_list()
|
||||
return IgnoreListResponse(ignore=sorted(IGNORE_LIST))
|
||||
|
||||
|
||||
@app.delete("/api/admin/ignore", response_model=IgnoreListResponse)
|
||||
def admin_delete_ignore(
|
||||
name: str = Query(..., description="要移除的忽略文件夹名"),
|
||||
_: None = Depends(require_admin),
|
||||
) -> IgnoreListResponse:
|
||||
"""从忽略列表中删除一项。"""
|
||||
global IGNORE_LIST
|
||||
key = validate_ignore_folder_name(name)
|
||||
if key not in IGNORE_LIST:
|
||||
raise HTTPException(status_code=404, detail="未找到该项")
|
||||
IGNORE_LIST.discard(key)
|
||||
persist_ignore_list()
|
||||
return IgnoreListResponse(ignore=sorted(IGNORE_LIST))
|
||||
|
||||
|
||||
@app.get("/api/health")
|
||||
def health_check():
|
||||
"""简单健康检查接口。"""
|
||||
return {"status": "ok"}
|
||||
"""健康检查；附带笔记根路径与顶层条目数，便于排查「目录树为空」。"""
|
||||
root = MARKDOWN_ROOT
|
||||
exists = root.is_dir()
|
||||
try:
|
||||
n = len(list(root.iterdir())) if exists else 0
|
||||
except OSError:
|
||||
n = -1
|
||||
return {
|
||||
"status": "ok",
|
||||
"markdown_root": str(root),
|
||||
"markdown_root_exists": exists,
|
||||
"markdown_root_entry_count": n,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user