482 lines
16 KiB
Python
482 lines
16 KiB
Python
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 Depends, FastAPI, Header, HTTPException, Query
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from pydantic import BaseModel
|
||
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
|
||
|
||
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]:
|
||
"""从 data/config/ignore.json 加载需要忽略的文件夹列表（与笔记根目录无关）。"""
|
||
if not IGNORE_FILE.exists():
|
||
return set()
|
||
|
||
try:
|
||
with open(IGNORE_FILE, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
return set(data.get('ignore', []))
|
||
except Exception:
|
||
return set()
|
||
|
||
|
||
# 加载忽略列表
|
||
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"
|
||
FILE: Literal["file"] = "file"
|
||
|
||
|
||
class DirectoryNode(BaseModel):
|
||
name: str
|
||
path: str # 相对于 MARKDOWN_ROOT 的路径，使用 / 作为分隔符
|
||
type: Literal["folder", "file"]
|
||
children: Optional[List["DirectoryNode"]] = None
|
||
|
||
|
||
DirectoryNode.update_forward_refs()
|
||
|
||
|
||
class FileContent(BaseModel):
|
||
path: str
|
||
content: str
|
||
word_count: int = 0
|
||
file_size: int = 0 # 文件大小，字节
|
||
created_time: str = ""
|
||
modified_time: str = ""
|
||
|
||
|
||
app = FastAPI(title="MengyaNote Backend", version="1.0.0")
|
||
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
# 现在允许任意来源方便本地和静态托管访问，如 http://localhost:9090
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
def is_markdown_file(path: Path) -> bool:
|
||
return path.is_file() and path.suffix.lower() == ".md"
|
||
|
||
|
||
def should_skip(entry: Path) -> bool:
|
||
"""判断是否应该跳过该文件或文件夹"""
|
||
name = entry.name
|
||
|
||
# 跳过隐藏文件/文件夹
|
||
if name.startswith("."):
|
||
return True
|
||
|
||
# 跳过 ignore.json 文件本身
|
||
if name == "ignore.json":
|
||
return True
|
||
|
||
# 跳过 ignore.json 中配置的文件夹
|
||
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
|
||
|
||
|
||
def build_directory_tree(root: Path) -> List[DirectoryNode]:
|
||
"""从文件系统构建目录树，结构尽量与原先 JSON 保持一致。"""
|
||
if not root.exists() or not root.is_dir():
|
||
return []
|
||
|
||
def walk(current: Path, rel: Path) -> DirectoryNode:
|
||
name = current.name
|
||
rel_path_str = rel.as_posix() if rel.as_posix() != "." else ""
|
||
|
||
if current.is_dir():
|
||
children_nodes: List[DirectoryNode] = []
|
||
for child in sorted(current.iterdir(), key=lambda p: (p.is_file(), p.name.lower())):
|
||
if should_skip(child):
|
||
continue
|
||
child_rel = rel / child.name
|
||
# 只收录 Markdown 文件和非空目录
|
||
if child.is_dir():
|
||
node = walk(child, child_rel)
|
||
# 如果目录下完全没有 md 文件/子目录，可以选择丢弃
|
||
if node.children:
|
||
children_nodes.append(node)
|
||
elif is_markdown_file(child):
|
||
children_nodes.append(
|
||
DirectoryNode(
|
||
name=child.name,
|
||
path=child_rel.as_posix(),
|
||
type="file",
|
||
children=None,
|
||
)
|
||
)
|
||
|
||
return DirectoryNode(
|
||
name=name,
|
||
path=rel_path_str or name,
|
||
type="folder",
|
||
children=children_nodes,
|
||
)
|
||
else:
|
||
# 单独文件的情况一般不会作为根调用
|
||
return DirectoryNode(
|
||
name=name,
|
||
path=rel_path_str or name,
|
||
type="file",
|
||
children=None,
|
||
)
|
||
|
||
nodes: List[DirectoryNode] = []
|
||
for child in sorted(MARKDOWN_ROOT.iterdir(), key=lambda p: (p.is_file(), p.name.lower())):
|
||
if should_skip(child):
|
||
continue
|
||
rel = Path(child.name)
|
||
if child.is_dir():
|
||
node = walk(child, rel)
|
||
if node.children:
|
||
nodes.append(node)
|
||
elif is_markdown_file(child):
|
||
nodes.append(
|
||
DirectoryNode(
|
||
name=child.name,
|
||
path=rel.as_posix(),
|
||
type="file",
|
||
children=None,
|
||
)
|
||
)
|
||
|
||
return nodes
|
||
|
||
|
||
def resolve_markdown_path(relative_path: str) -> Path:
|
||
"""将前端传入的相对路径安全地转换为磁盘路径，防止目录穿越。"""
|
||
# 统一使用 / 分隔符
|
||
safe_path = relative_path.replace("\\", "/").lstrip("/")
|
||
candidate_resolved = (MARKDOWN_ROOT / safe_path).resolve()
|
||
try:
|
||
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]:
|
||
"""
|
||
获取 Markdown 目录树。
|
||
|
||
返回结构与原来的 directoryTree.json 尽量保持兼容：
|
||
- name: 文件或文件夹名
|
||
- path: 相对路径（使用 /）
|
||
- 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
|
||
|
||
|
||
@app.get("/api/file", response_model=FileContent)
|
||
def get_markdown_file(path: str = Query(..., description="相对于根目录的 Markdown 路径")) -> FileContent:
|
||
"""
|
||
获取指定 Markdown 文件内容。
|
||
|
||
Query 参数:
|
||
- path: 例如 'AI/大语言模型的API 调用.md'
|
||
"""
|
||
file_path = resolve_markdown_path(path)
|
||
|
||
if not file_path.exists() or not file_path.is_file() or not is_markdown_file(file_path):
|
||
raise HTTPException(status_code=404, detail="文件不存在")
|
||
|
||
try:
|
||
content = file_path.read_text(encoding="utf-8")
|
||
except UnicodeDecodeError:
|
||
# 回退编码
|
||
content = file_path.read_text(encoding="utf-8", errors="ignore")
|
||
|
||
# 获取文件统计信息
|
||
file_stat = file_path.stat()
|
||
|
||
# 计算字数（去除空格和换行符）
|
||
word_count = _word_count_non_whitespace(content)
|
||
|
||
# 获取文件大小（字节）
|
||
file_size = file_stat.st_size
|
||
|
||
# 获取创建时间和修改时间
|
||
try:
|
||
# Windows 上 st_ctime 是创建时间，Linux 上是元数据更改时间
|
||
created_time = datetime.fromtimestamp(file_stat.st_ctime).strftime("%Y年%m月%d日 %H:%M:%S")
|
||
except:
|
||
created_time = "未知"
|
||
|
||
try:
|
||
modified_time = datetime.fromtimestamp(file_stat.st_mtime).strftime("%Y年%m月%d日 %H:%M:%S")
|
||
except:
|
||
modified_time = "未知"
|
||
|
||
rel = file_path.relative_to(MARKDOWN_ROOT).as_posix()
|
||
return FileContent(
|
||
path=rel,
|
||
content=content,
|
||
word_count=word_count,
|
||
file_size=file_size,
|
||
created_time=created_time,
|
||
modified_time=modified_time
|
||
)
|
||
|
||
|
||
@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():
|
||
"""健康检查；附带笔记根路径与顶层条目数，便于排查「目录树为空」。"""
|
||
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__":
|
||
import uvicorn
|
||
|
||
uvicorn.run("main:app", host="0.0.0.0", port=int(os.getenv("PORT", 8000)), reload=True)
|
||
|
||
|