feat: 文件管理器 + Windows兼容 + SVG主题适配 (v0.6.0)
新增: - services/web_panel/routes/files.py — 全功能文件管理后端API (11个端点) - pages/files.html + files.js — 文件管理前端 (浏览/编辑/上传/删除/右键菜单) - 插件Picker API: window.open+postMessage唤出文件选择器 - 符号链接目录双层面包屑 (逻辑路径+物理路径) - 文件系统全访问+跨平台 (Linux/Windows/macOS) - permission_rules.yaml新增4个filemanager.*权限 修复: - SVG fill=currentColor 日夜模式自适应 - 特殊目录容错 (/dev/fd损坏符号链接/proc) - 面包屑每层级独立可点击+可编辑路径跳转 - ..行返回上级+data-is-dir补全 - rmlint→lstat回退 损坏符号链接不炸页面 文档: - 插件开发指南新增第八章(文件管理器集成) - 开发踩坑记录新增4条(特殊目录/双面包屑/SVG颜色/Windows) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
qinglong
@@ -0,0 +1,425 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
File Manager Backend API
|
||||
|
||||
Endpoints:
|
||||
GET {prefix}/api/files/list?path=&show_hidden=0 — list directory
|
||||
POST {prefix}/api/files/mkdir {path, name} — create directory
|
||||
POST {prefix}/api/files/touch {path, name} — create empty file
|
||||
POST {prefix}/api/files/delete {path} — delete file/dir
|
||||
POST {prefix}/api/files/rename {path, new_name} — rename
|
||||
POST {prefix}/api/files/upload (multipart) — upload file(s)
|
||||
GET {prefix}/api/files/download?path= — download file
|
||||
GET {prefix}/api/files/read?path= — read text file
|
||||
POST {prefix}/api/files/write {path, content} — write text file
|
||||
GET {prefix}/api/files/info?path= — file/dir stat
|
||||
|
||||
Root directory is restricted to configurable ROOTS (default: [project_root, data/]).
|
||||
Path-traversal is blocked.
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import time
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
from pathlib import Path
|
||||
from aiohttp import web
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Security: allowed root directories ──
|
||||
_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent
|
||||
_ALLOWED_ROOTS = []
|
||||
|
||||
if os.name == 'nt':
|
||||
# Windows: enumerate all available drive letters
|
||||
import string
|
||||
for letter in string.ascii_uppercase:
|
||||
drive = Path(letter + ":\\")
|
||||
if drive.exists():
|
||||
_ALLOWED_ROOTS.append(drive)
|
||||
else:
|
||||
# Linux / macOS / Android
|
||||
_ALLOWED_ROOTS = [
|
||||
Path("/"),
|
||||
Path("/media/sd"), # Android shared storage
|
||||
Path("/mnt"), # WSL mounts
|
||||
]
|
||||
_ALLOWED_ROOTS.append(_PROJECT_ROOT)
|
||||
# Deduplicate and keep only existing
|
||||
_seen = set()
|
||||
_filtered = []
|
||||
for p in _ALLOWED_ROOTS:
|
||||
rp = p.resolve() if p.exists() else None
|
||||
if rp and str(rp) not in _seen:
|
||||
_seen.add(str(rp))
|
||||
_filtered.append(rp)
|
||||
_ALLOWED_ROOTS = _filtered or [Path("/") if os.name != 'nt' else Path("C:\\")]
|
||||
|
||||
MAX_READ_SIZE = 1 * 1024 * 1024 # 1 MB for text read
|
||||
MAX_UPLOAD_SIZE = 50 * 1024 * 1024 # 50 MB per upload
|
||||
TEXT_EXTENSIONS = {
|
||||
'.txt','.py','.js','.ts','.html','.css','.json','.yaml','.yml',
|
||||
'.md','.ini','.cfg','.conf','.log','.sh','.bat','.env','.xml',
|
||||
'.toml','.csv','.sql','.Makefile','.gitignore','.dockerfile',
|
||||
'.c','.h','.cpp','.hpp','.java','.kt','.rs','.go','.rb','.php',
|
||||
'.swift','.r','.lua','.pl','.scala','.dart',
|
||||
}
|
||||
|
||||
# ── Helpers ──
|
||||
|
||||
def _normalize(path_str: str) -> Path:
|
||||
"""Normalize path (resolve .. and .) WITHOUT following symlinks.
|
||||
Returns a Path that may be a symlink itself."""
|
||||
if not path_str:
|
||||
return _ALLOWED_ROOTS[0]
|
||||
|
||||
# Use os.path.normpath which resolves .. and . without following symlinks
|
||||
normalized = os.path.normpath(path_str)
|
||||
return Path(normalized)
|
||||
|
||||
def _resolve(path_str: str) -> Path:
|
||||
"""Resolve and validate path against allowed roots.
|
||||
Returns the LOGICAL path (may still be a symlink, not resolved)."""
|
||||
candidate = _normalize(path_str)
|
||||
# Security: ensure it's within an allowed root
|
||||
resolved = candidate.resolve()
|
||||
for root in _ALLOWED_ROOTS:
|
||||
try:
|
||||
resolved.relative_to(root)
|
||||
return candidate # Return the logical path, not resolved
|
||||
except ValueError:
|
||||
continue
|
||||
raise ValueError(f"Path not within allowed roots: {path_str}")
|
||||
|
||||
def _ensure_exists(p: Path, must_exist: bool = True):
|
||||
if not p.exists():
|
||||
if must_exist:
|
||||
raise FileNotFoundError(str(p))
|
||||
return p
|
||||
|
||||
def _stat(p: Path) -> dict:
|
||||
"""Stat a path: follow symlinks normally; fall back to lstat for broken ones."""
|
||||
try:
|
||||
s = p.stat() # Follow symlinks — resolves dir/file correctly
|
||||
return {
|
||||
"name": p.name,
|
||||
"path": str(p),
|
||||
"size": s.st_size,
|
||||
"is_dir": p.is_dir(),
|
||||
"is_file": p.is_file(),
|
||||
"is_symlink": p.is_symlink(),
|
||||
"mtime": int(s.st_mtime),
|
||||
"mtime_str": time.strftime("%Y-%m-%d %H:%M", time.localtime(s.st_mtime)),
|
||||
"mode": stat.filemode(s.st_mode),
|
||||
"ext": p.suffix.lower() if p.is_file() else "",
|
||||
"readable": os.access(p, os.R_OK),
|
||||
"writable": os.access(p, os.W_OK),
|
||||
}
|
||||
except FileNotFoundError:
|
||||
# Broken symlink or missing target
|
||||
try:
|
||||
s = p.lstat()
|
||||
return {
|
||||
"name": p.name,
|
||||
"path": str(p),
|
||||
"size": s.st_size,
|
||||
"is_dir": False,
|
||||
"is_file": False,
|
||||
"is_symlink": True,
|
||||
"mtime": int(s.st_mtime),
|
||||
"mtime_str": time.strftime("%Y-%m-%d %H:%M", time.localtime(s.st_mtime)),
|
||||
"mode": stat.filemode(s.st_mode),
|
||||
"ext": "",
|
||||
"readable": False,
|
||||
"writable": False,
|
||||
"broken": True,
|
||||
}
|
||||
except Exception:
|
||||
pass # Give up entirely
|
||||
except (OSError, PermissionError):
|
||||
pass
|
||||
# Ultimate fallback
|
||||
return {
|
||||
"name": p.name, "path": str(p), "size": 0,
|
||||
"is_dir": False, "is_file": False, "is_symlink": p.is_symlink(),
|
||||
"mtime": 0, "mtime_str": "—", "mode": "?---------", "ext": "",
|
||||
"readable": False, "writable": False, "broken": True,
|
||||
}
|
||||
|
||||
# ── Route setup ──
|
||||
|
||||
def setup_file_routes(app, service_manager, prefix=''):
|
||||
"""Register all file-manager routes on the aiohttp app."""
|
||||
logger.info(f"📁 文件管理路由已注册 ({prefix}/api/files)")
|
||||
|
||||
# List directory
|
||||
async def list_dir(req):
|
||||
try:
|
||||
path = req.query.get("path", "")
|
||||
show_hidden = req.query.get("show_hidden", "0") == "1"
|
||||
p = _resolve(path)
|
||||
_ensure_exists(p)
|
||||
if not p.is_dir():
|
||||
return web.json_response({"error": "Not a directory"}, status=400)
|
||||
|
||||
items = []
|
||||
try:
|
||||
for entry in sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
|
||||
if not show_hidden and entry.name.startswith('.'):
|
||||
continue
|
||||
try:
|
||||
items.append(_stat(entry))
|
||||
except Exception:
|
||||
pass # Skip entries that can't be stat'd
|
||||
except PermissionError:
|
||||
return web.json_response({"error": "Permission denied"}, status=403)
|
||||
|
||||
# Breadcrumbs: split path by OS separator, every segment clickable
|
||||
def make_breadcrumbs(path_obj):
|
||||
raw = str(path_obj)
|
||||
if os.name == 'nt' and len(raw) >= 2 and raw[1] == ':':
|
||||
# Windows: C:\Users\... → crumbs: [C:\, Users, ...]
|
||||
crumbs = [{"label": raw[:2] + "\\", "path": raw[:2] + "\\"}]
|
||||
tail = raw[3:]
|
||||
else:
|
||||
crumbs = [{"label": "/", "path": "/"}]
|
||||
tail = raw.lstrip("/")
|
||||
parts = [x for x in tail.replace("\\", "/").split("/") if x]
|
||||
acc = crumbs[0]["path"].rstrip("\\/")
|
||||
for part in parts:
|
||||
acc += ("\\" if os.name == 'nt' and acc.endswith(":") else "") + "/" + part
|
||||
crumbs.append({"label": part, "path": acc.replace("\\", "/")})
|
||||
return crumbs
|
||||
|
||||
crumbs = make_breadcrumbs(p)
|
||||
# Logical parent: None at filesystem root (Unix: /, Windows: C:\)
|
||||
p_str = str(p)
|
||||
is_root = p_str == "/" or (os.name == 'nt' and len(p_str) == 3 and p_str[1] == ':')
|
||||
logical_parent = str(Path(p_str).parent) if not is_root else None
|
||||
resolved = p.resolve()
|
||||
resolved_crumbs = None
|
||||
if str(resolved) != str(p):
|
||||
resolved_crumbs = make_breadcrumbs(resolved)
|
||||
|
||||
return web.json_response({
|
||||
"current": str(p),
|
||||
"resolved": str(resolved) if str(resolved) != str(p) else None,
|
||||
"breadcrumbs": crumbs,
|
||||
"resolved_breadcrumbs": resolved_crumbs,
|
||||
"items": items,
|
||||
"parent": logical_parent,
|
||||
"allowed_roots": [str(r) for r in _ALLOWED_ROOTS],
|
||||
})
|
||||
except (FileNotFoundError, ValueError) as e:
|
||||
return web.json_response({"error": str(e)}, status=404)
|
||||
except Exception as e:
|
||||
logger.error(f"list_dir: {e}")
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Create directory
|
||||
async def mkdir(req):
|
||||
try:
|
||||
data = await req.json()
|
||||
p = _resolve(data.get("path", ""))
|
||||
name = data.get("name", "").strip()
|
||||
if not name or "/" in name or "\\" in name:
|
||||
return web.json_response({"error": "Invalid name"}, status=400)
|
||||
target = (p / name)
|
||||
target.mkdir(parents=False, exist_ok=False)
|
||||
logger.info(f"📁 创建目录: {target}")
|
||||
return web.json_response({"ok": True, "path": str(target)})
|
||||
except FileExistsError:
|
||||
return web.json_response({"error": "Already exists"}, status=409)
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Create empty file
|
||||
async def touch(req):
|
||||
try:
|
||||
data = await req.json()
|
||||
p = _resolve(data.get("path", ""))
|
||||
name = data.get("name", "").strip()
|
||||
if not name or "/" in name or "\\" in name:
|
||||
return web.json_response({"error": "Invalid name"}, status=400)
|
||||
target = (p / name)
|
||||
target.touch(exist_ok=False)
|
||||
logger.info(f"📄 创建文件: {target}")
|
||||
return web.json_response({"ok": True, "path": str(target)})
|
||||
except FileExistsError:
|
||||
return web.json_response({"error": "Already exists"}, status=409)
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Delete file or empty directory (recursive for non-empty dirs)
|
||||
async def delete(req):
|
||||
try:
|
||||
data = await req.json()
|
||||
p = _resolve(data.get("path", ""))
|
||||
_ensure_exists(p)
|
||||
# Safety: refuse to delete project root
|
||||
if p == _PROJECT_ROOT:
|
||||
return web.json_response({"error": "Cannot delete project root"}, status=403)
|
||||
if p.is_dir():
|
||||
shutil.rmtree(p)
|
||||
else:
|
||||
p.unlink()
|
||||
logger.info(f"🗑 删除: {p}")
|
||||
return web.json_response({"ok": True})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Rename
|
||||
async def rename(req):
|
||||
try:
|
||||
data = await req.json()
|
||||
p = _resolve(data.get("path", ""))
|
||||
new_name = data.get("new_name", "").strip()
|
||||
if not new_name or "/" in new_name or "\\" in new_name:
|
||||
return web.json_response({"error": "Invalid name"}, status=400)
|
||||
target = p.parent / new_name
|
||||
p.rename(target)
|
||||
logger.info(f"✏ 重命名: {p} → {target}")
|
||||
return web.json_response({"ok": True, "path": str(target)})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Upload
|
||||
async def upload(req):
|
||||
try:
|
||||
reader = await req.multipart()
|
||||
target_dir_str = req.query.get("path", "")
|
||||
target_dir = _resolve(target_dir_str)
|
||||
_ensure_exists(target_dir)
|
||||
if not target_dir.is_dir():
|
||||
return web.json_response({"error": "Target not a directory"}, status=400)
|
||||
|
||||
uploaded = []
|
||||
while True:
|
||||
part = await reader.next()
|
||||
if part is None:
|
||||
break
|
||||
if part.name == "file":
|
||||
fname = part.filename
|
||||
if not fname:
|
||||
continue
|
||||
# Sanitize filename
|
||||
fname = Path(fname).name
|
||||
dest = target_dir / fname
|
||||
size = 0
|
||||
with open(dest, 'wb') as f:
|
||||
while True:
|
||||
chunk = await part.read_chunk(65536)
|
||||
if not chunk:
|
||||
break
|
||||
size += len(chunk)
|
||||
if size > MAX_UPLOAD_SIZE:
|
||||
f.close()
|
||||
dest.unlink()
|
||||
return web.json_response({"error": f"File too large: {fname}"}, status=413)
|
||||
f.write(chunk)
|
||||
uploaded.append(_stat(dest))
|
||||
logger.info(f"📤 上传 {len(uploaded)} 文件到 {target_dir}")
|
||||
return web.json_response({"ok": True, "files": uploaded})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Download
|
||||
async def download(req):
|
||||
try:
|
||||
p = _resolve(req.query.get("path", ""))
|
||||
_ensure_exists(p)
|
||||
if not p.is_file():
|
||||
return web.json_response({"error": "Not a file"}, status=400)
|
||||
ct, _ = mimetypes.guess_type(str(p))
|
||||
return web.FileResponse(p, headers={
|
||||
"Content-Type": ct or "application/octet-stream",
|
||||
"Content-Disposition": f'attachment; filename="{p.name}"',
|
||||
})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Read text file
|
||||
async def read_file(req):
|
||||
try:
|
||||
p = _resolve(req.query.get("path", ""))
|
||||
_ensure_exists(p)
|
||||
if not p.is_file():
|
||||
return web.json_response({"error": "Not a file"}, status=400)
|
||||
if p.suffix.lower() not in TEXT_EXTENSIONS:
|
||||
return web.json_response({"error": f"Not a text file: {p.suffix}"}, status=415)
|
||||
if p.stat().st_size > MAX_READ_SIZE:
|
||||
return web.json_response({"error": "File too large to read"}, status=413)
|
||||
content = p.read_text(encoding="utf-8", errors="replace")
|
||||
return web.json_response({
|
||||
"path": str(p),
|
||||
"name": p.name,
|
||||
"size": len(content),
|
||||
"content": content,
|
||||
})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Write text file
|
||||
async def write_file(req):
|
||||
try:
|
||||
data = await req.json()
|
||||
p = _resolve(data.get("path", ""))
|
||||
content = data.get("content", "")
|
||||
p.write_text(content, encoding="utf-8")
|
||||
logger.info(f"💾 写入文件: {p} ({len(content)} bytes)")
|
||||
return web.json_response({"ok": True, "size": len(content)})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# File/dir info
|
||||
async def file_info(req):
|
||||
try:
|
||||
p = _resolve(req.query.get("path", ""))
|
||||
_ensure_exists(p)
|
||||
return web.json_response(_stat(p))
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# Picker API (for plugins) — returns selected path as JSON
|
||||
# GET {prefix}/api/files/picker?mode=file|dir&path=
|
||||
async def picker_api(req):
|
||||
try:
|
||||
mode = req.query.get("mode", "dir") # file | dir
|
||||
path = req.query.get("path", "")
|
||||
p = _resolve(path)
|
||||
_ensure_exists(p)
|
||||
items = []
|
||||
if p.is_dir():
|
||||
for entry in sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
|
||||
if entry.name.startswith('.'):
|
||||
continue
|
||||
if mode == "file" and not entry.is_file():
|
||||
continue
|
||||
if mode == "dir" and not entry.is_dir():
|
||||
continue
|
||||
items.append(_stat(entry))
|
||||
return web.json_response({
|
||||
"current": str(p),
|
||||
"mode": mode,
|
||||
"items": items,
|
||||
})
|
||||
except Exception as e:
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
# ── Register routes ──
|
||||
app.router.add_get(f'{prefix}/api/files/list', list_dir)
|
||||
app.router.add_post(f'{prefix}/api/files/mkdir', mkdir)
|
||||
app.router.add_post(f'{prefix}/api/files/touch', touch)
|
||||
app.router.add_post(f'{prefix}/api/files/delete', delete)
|
||||
app.router.add_post(f'{prefix}/api/files/rename', rename)
|
||||
app.router.add_post(f'{prefix}/api/files/upload', upload)
|
||||
app.router.add_get(f'{prefix}/api/files/download', download)
|
||||
app.router.add_get(f'{prefix}/api/files/read', read_file)
|
||||
app.router.add_post(f'{prefix}/api/files/write', write_file)
|
||||
app.router.add_get(f'{prefix}/api/files/info', file_info)
|
||||
app.router.add_get(f'{prefix}/api/files/picker', picker_api)
|
||||
Reference in New Issue
Block a user