admin/Nexus
1
0
Fork 0
Code Issues Pull Requests Actions 91 Packages Projects Releases Wiki Activity

feat: 推送页面 5 项迭代 — WS实时进度 + 分组选择 + 历史增强 + 文件操作 + 推送校验

Browse Source 
F1: broadcast_sync_progress() + batch_id + 前端WS逐台更新
F2: get_paginated() platform_id/node_id + /categories端点 + 筛选下拉框
F4: _sync_log_to_dict() 补全字段 + diff_summary捕获 + 可展开详情面板
F7: /local-file-ops端点 + 文件管理器删除/重命名
F8: /verify端点 md5sum对比 + 推送后校验UI

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-05-28 21:01:27 +08:00
parent 77b332c037
commit 82c297776a
8 changed files with 1308 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/changelog/2026-05-28-push-5-features.md
+61
View File
@@ -0,0 +1,61 @@
# 2026-05-28 推送页面 5 项迭代功能
## 变更摘要
1. F1: 推送进度实时追踪 — WebSocket 逐台推送进度更新
2. F2: 服务器分组选择 — 分类/平台筛选 + 选中当前筛选
3. F4: 推送历史增强 — 可展开详情面板（耗时/错误/diff）
4. F7: 文件管理器操作 — 解压文件删除/重命名
5. F8: 推送后校验 — md5sum 对比本地源与远程目标
## 动机
推送页面已完成 ZIP 上传 + rsync 推送基础功能，但存在 5 个体验缺陷：批量推送无实时反馈、2000+ 服务器无法分组选择、历史记录无详情、文件管理器只读、推送后无法校验文件一致性。
## 涉及文件
### 后端
- `server/api/websocket.py` — 新增 `broadcast_sync_progress()` 函数
- 参数: batch_id, server_id, server_name, status, completed, failed, total, error_message, duration_seconds
- 通过 `_dispatch_ws_message()` 发布到 Redis Pub/Sub + 本地 WebSocket
- `server/application/services/sync_engine_v2.py` — 推送引擎改造
- `sync_files()`: 新增 `batch_id` (uuid) 用于 WS 消息过滤
- `_sync_one()`: 计数器更新后调用 `broadcast_sync_progress()`
- `_sync_one()`: 捕获 `diff_summary = result["stdout"][:2000]`
- `_log_to_dict()`: 新增 files_transferred, files_skipped, bytes_transferred, diff_summary
- 返回 dict 新增 `batch_id`
- `server/api/servers.py` — 服务器列表 + 历史增强
- `list_servers()`: 新增 `platform_id`, `node_id` Query 参数
- 新增 `GET /api/servers/categories` 端点（分类+平台列表）
- `_sync_log_to_dict()`: 新增 files_skipped, bytes_transferred, diff_summary
- `server/infrastructure/database/server_repo.py` — `get_paginated()` 增加 `platform_id`, `node_id` 参数
- `server/api/schemas.py` — 新增 `LocalFileOperation`, `SyncVerify` 模型
- `server/api/sync_v2.py` — 新增端点
- `POST /api/sync/local-file-ops`: 删除/重命名上传临时目录中的文件
- `POST /api/sync/verify`: md5sum 对比本地源与远程目标
### 前端
- `web/app/push.html` — 全面改造
- F1: WebSocket 连接 `connectPushWS()` + `updateProgressFromWS()` 逐台更新
- F2: 分类/平台筛选下拉框 + `filterServers()` + "选中当前筛选" 按钮
- F4: 历史记录可点击展开详情（源/目标路径、耗时、错误、diff_summary）
- F7: 文件管理器 hover 显示操作按钮 + `deleteLocalFile()`/`renameLocalFile()`
- F8: "推送后校验" 复选框 + `doVerify()` + 校验结果面板
## 安全影响
- F1 (WS): WS 消息需 JWT 认证 + batch_id 过滤防混淆
- F7 (文件操作): `os.path.realpath()` + `startswith("/tmp/nexus_upload_")` 双重校验，新路径也必须在上传目录内
- F8 (校验): SSH 命令中文件名通过 `shlex.quote()` 防注入
- 所有新端点需 JWT 认证 + 审计日志
## 需要迁移/重启
- 是：后端 Python 需重启（supervisorctl restart nexus）
- 否：无数据库 schema 迁移
## 验证方式
1. F1: 推送 3+ 台服务器 → 观察 WS 逐台更新状态（不等全部完成）
2. F2: 分类筛选 → 选中筛选 → 服务器列表正确过滤
3. F4: 点击历史记录 → 展开详情 → 显示耗时/错误/diff
4. F7: 上传 ZIP → 文件管理器 → 删除/重命名文件 → 刷新确认
5. F8: 勾选校验 → 推送 → 校验结果正确显示匹配/缺失
## 进度条
☑实现 ☑WSL验证 ☐审计8步 ☐部署 ☐健康检查 ☐浏览器验证 ☐changelog
server/api/schemas.py
+20
View File
@@ -135,6 +135,26 @@ class SyncBrowse(BaseModel):
path: str = Field("/", min_length=1)
class SyncBrowseLocal(BaseModel):
"""Browse a local directory on the Nexus server (for upload file manager)."""
path: str = Field(..., min_length=1)
class LocalFileOperation(BaseModel):
"""Delete or rename a file/directory in the local upload staging area."""
operation: str = Field(..., pattern="^(delete|rename)$")
path: str = Field(..., min_length=1)
new_name: Optional[str] = None # required for rename
class SyncVerify(BaseModel):
"""Post-push verification: compare local source with remote target via md5sum."""
server_ids: List[int] = Field(..., min_length=1)
source_path: str = Field(..., min_length=1)
target_path: Optional[str] = None
max_files: int = Field(200, ge=1, le=500)
class FileOperation(BaseModel):
"""Remote file operation via SSH exec (delete / rename / mkdir)."""
server_id: int = Field(..., ge=1)
server/api/servers.py
+329 -5
View File
@@ -46,6 +46,69 @@ def _sudo_wrap(cmd: str, ssh_user: str) -> str:
cleanup = f"; sudo rm -f /etc/sudoers.d/{sudoers_tag} 2>/dev/null"
return setup + cmd + cleanup
# Translation table: install.sh ERROR lines (English) → Chinese
_INSTALL_ERROR_ZH: dict[str, str] = {
"Python 3.10+ required but not found and could not be installed automatically.":
"Python 版本低于 3.10，且无法自动安装",
"sudo requires a password. Configure passwordless sudo first:":
"sudo 需要密码，请先配置免密 sudo",
"Running as non-root and sudo is not available.":
"非 root 用户且无 sudo 权限",
"--url is required":
"缺少 --url 参数",
"--id is required (server ID from Nexus dashboard)":
"缺少 --id 参数（服务器 ID）",
"--key is required (API key from Nexus settings)":
"缺少 --key 参数（API Key）",
"venv creation failed even after installing python3-venv":
"虚拟环境创建失败（python3-venv 模块不可用）",
"Cannot download from":
"无法下载 agent.py",
"upgrade_ok":
"升级验证失败",
}
def _install_error_msg(stdout: str, stderr: str, exit_code: int) -> str:
"""Extract concise error reason from install.sh / upgrade output.
install.sh writes ``ERROR: ...`` lines to **stdout** (not stderr),
so extracting only stderr yields empty/unhelpful messages like
"安装失败 (exit 1)". This function finds the first ``ERROR:`` line
in stdout, translates it to Chinese, and appends the original English
for full context. Falls back to the last non-empty stdout line,
then stderr, and finally a generic exit-code message.
"""
# 1. Prefer ERROR: lines from stdout (install.sh convention)
for line in reversed(stdout.split("\n")):
line = line.strip()
if line.startswith("ERROR:"):
en = line.replace("ERROR: ", "")
zh = None
# Exact match first
if en in _INSTALL_ERROR_ZH:
zh = _INSTALL_ERROR_ZH[en]
else:
# Prefix match (e.g. "Cannot download from https://...")
for prefix, translation in _INSTALL_ERROR_ZH.items():
if en.startswith(prefix):
zh = translation
break
if zh:
return f"{zh}（{en[:200]}）"
return en[:300]
# 2. Last non-empty stdout line (may contain useful context)
for line in reversed(stdout.split("\n")):
line = line.strip()
if line:
return line[:300]
# 3. stderr
if stderr.strip():
return stderr.strip()[:300]
# 4. Generic
return f"exit {exit_code}"
logger = logging.getLogger("nexus.servers")
router = APIRouter(prefix="/api/servers", tags=["servers"])
@@ -58,6 +121,8 @@ REDIS_KEY_PREFIX = "heartbeat:"
@router.get("/", response_model=dict)
async def list_servers(
category: Optional[str] = Query(None, description="Filter by category"),
platform_id: Optional[int] = Query(None, description="Filter by platform ID"),
node_id: Optional[int] = Query(None, description="Filter by node ID"),
page: int = Query(1, ge=1, description="Page number"),
per_page: int = Query(50, ge=1, le=200, description="Items per page"),
admin: Admin = Depends(get_current_admin),
@@ -74,7 +139,7 @@ async def list_servers(
offset = (page - 1) * per_page
repo = ServerRepositoryImpl(db)
page_servers, total = await repo.get_paginated(category, offset, per_page)
page_servers, total = await repo.get_paginated(category, platform_id, node_id, offset, per_page)
redis = get_redis()
result = []
@@ -82,6 +147,10 @@ async def list_servers(
server_data = _server_to_dict(server)
# Overlay Redis heartbeat data (real-time)
# Redis is the source of truth for Agent status.
# If a server has/had an Agent (agent_version set), absence of a Redis
# heartbeat key means the Agent is offline (heartbeat TTL expired).
# Only fall back to MySQL is_online for servers that never had an Agent.
try:
heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
if heartbeat:
@@ -100,6 +169,10 @@ async def list_servers(
server_data["time_drift_seconds"] = 0.0
server_data["drift_level"] = heartbeat.get("drift_level", "ok")
server_data["_source"] = "redis"
elif server.agent_version:
# Had Agent but no heartbeat → Agent is offline
server_data["is_online"] = False
server_data["_source"] = "redis_miss"
except Exception as e:
logger.warning(f"Redis read failed for server {server.id}: {e}")
@@ -178,6 +251,33 @@ async def server_stats(
# ── Sync Logs ──
@router.get("/categories", response_model=dict)
async def server_categories(
admin: Admin = Depends(get_current_admin),
db: AsyncSession = Depends(get_db),
):
"""Return distinct categories with counts and platform list for server filtering."""
from sqlalchemy import func as sqlfunc
from server.domain.models import Platform
# Categories with server counts
cat_result = await db.execute(
select(Server.category, sqlfunc.count(Server.id))
.group_by(Server.category)
)
categories = []
for cat, cnt in cat_result.all():
categories.append({"name": cat or "uncategorized", "count": cnt})
# Platforms
plat_result = await db.execute(select(Platform).order_by(Platform.id))
platforms = []
for p in plat_result.scalars().all():
platforms.append({"id": p.id, "name": p.name, "category": p.category or ""})
return {"categories": categories, "platforms": platforms}
@router.get("/logs", response_model=dict)
async def get_all_sync_logs(
server_id: Optional[int] = Query(None, description="Filter by server ID"),
@@ -497,7 +597,7 @@ async def batch_install_agent(
return BatchAgentResultItem(
server_id=sid, server_name=server_name, success=ok,
stdout=r["stdout"][:2000] if ok else r["stdout"][:1000],
error="" if ok else (r.get("stderr", "")[:500] or f"安装失败 (exit {r['exit_code']}), 查看stdout"),
error="" if ok else _install_error_msg(r.get("stdout", ""), r.get("stderr", ""), r["exit_code"]),
)
except Exception as e:
return BatchAgentResultItem(server_id=sid, server_name=server_name, success=False, error=str(e)[:300])
@@ -581,7 +681,7 @@ async def batch_upgrade_agent(
return BatchAgentResultItem(
server_id=sid, server_name=server.name, success=ok,
stdout=r["stdout"][:500] if ok else "",
error=r["stderr"][:300] if not ok else "",
error=_install_error_msg(r.get("stdout", ""), r.get("stderr", ""), r["exit_code"]) if not ok else "",
)
except Exception as e:
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error=str(e)[:200])
@@ -604,6 +704,153 @@ async def batch_upgrade_agent(
return BatchAgentResult(results=results)
@router.post("/batch/detect-path", response_model=BatchAgentResult)
async def batch_detect_path(
request: Request,
payload: BatchAgentAction,
admin: Admin = Depends(get_current_admin),
service: ServerService = Depends(get_server_service),
db: AsyncSession = Depends(get_db),
):
"""SSH 到各服务器搜索 workerman.bat，自动设置 target_path。
递归搜索 /www/wwwroot 下的 workerman.bat 文件，找到后取其目录路径
作为服务器的 target_path 写入数据库。找到第一个即停止搜索。
"""
import asyncio
import os
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
sem = asyncio.Semaphore(5)
results = []
async def _detect_one(sid: int):
async with sem:
server_name = ""
try:
server = await service.get_server(sid)
if not server:
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error="服务器不存在")
server_name = server.name
ssh_user = (server.username or "root").strip()
# find 第一个 workerman.bat 即停止 (-quit)
cmd = "find /www/wwwroot -iname 'workerman.bat' -type f -print -quit 2>/dev/null"
cmd = _sudo_wrap(cmd, ssh_user)
r = await exec_ssh_command(server, cmd, timeout=30)
if r["exit_code"] != 0 and not r.get("stdout", "").strip():
return BatchAgentResultItem(
server_id=sid, server_name=server_name, success=False,
error=f"搜索失败 (exit {r['exit_code']})",
)
found = r["stdout"].strip()
if not found:
return BatchAgentResultItem(
server_id=sid, server_name=server_name, success=False,
error="未找到 workerman.bat",
)
# 取目录路径（可能有多个结果，取第一个）
first_match = found.split("\n")[0].strip()
target_dir = os.path.dirname(first_match)
# 更新服务器 target_path
server.target_path = target_dir
await db.commit()
return BatchAgentResultItem(
server_id=sid, server_name=server_name, success=True,
stdout=f"target_path → {target_dir}",
)
except Exception as e:
return BatchAgentResultItem(server_id=sid, server_name=server_name, success=False, error=str(e)[:300])
items = await asyncio.gather(*[_detect_one(sid) for sid in payload.server_ids])
results = list(items)
# Audit
ip_address = request.client.host if request.client else ""
audit_repo = AuditLogRepositoryImpl(db)
ok_count = sum(1 for r in results if r.success)
await audit_repo.create(AuditLog(
admin_username=admin.username,
action="batch_detect_path",
target_type="server",
target_id=0,
detail=f"批量检测路径: {ok_count}/{len(results)} 成功",
ip_address=ip_address,
))
return BatchAgentResult(results=results)
@router.post("/batch/uninstall-agent", response_model=BatchAgentResult)
async def batch_uninstall_agent(
request: Request,
payload: BatchAgentAction,
admin: Admin = Depends(get_current_admin),
service: ServerService = Depends(get_server_service),
db: AsyncSession = Depends(get_db),
):
"""Uninstall Nexus Agent on multiple servers via SSH (concurrent, max 5 parallel)."""
import asyncio
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
import shlex
base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
if not base_url:
raise HTTPException(status_code=400, detail="NEXUS_API_BASE_URL 未配置")
sem = asyncio.Semaphore(5)
results = []
async def _uninstall_one(sid: int):
async with sem:
server_name = ""
try:
server = await service.get_server(sid)
if not server:
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error="服务器不存在")
server_name = server.name
ssh_user = (server.username or "root").strip()
uninstall_cmd = (
f"TMP=$(mktemp) && curl -fsSL {shlex.quote(base_url)}/agent/uninstall.sh -o \"$TMP\" "
f"&& bash \"$TMP\" ; RC=$?; rm -f \"$TMP\"; exit $RC"
)
uninstall_cmd = _sudo_wrap(uninstall_cmd, ssh_user)
r = await exec_ssh_command(server, uninstall_cmd, timeout=60)
ok = r["exit_code"] == 0
if ok:
server.is_online = False
server.agent_version = None
await db.commit()
return BatchAgentResultItem(
server_id=sid, server_name=server_name, success=ok,
stdout=r["stdout"][:500] if ok else "",
error="" if ok else _install_error_msg(r.get("stdout", ""), r.get("stderr", ""), r["exit_code"]),
)
except Exception as e:
return BatchAgentResultItem(server_id=sid, server_name=server_name, success=False, error=str(e)[:300])
items = await asyncio.gather(*[_uninstall_one(sid) for sid in payload.server_ids])
results = list(items)
# Audit
ip_address = request.client.host if request.client else ""
audit_repo = AuditLogRepositoryImpl(db)
ok_count = sum(1 for r in results if r.success)
await audit_repo.create(AuditLog(
admin_username=admin.username,
action="batch_uninstall_agent",
target_type="server",
target_id=0,
detail=f"批量卸载 Agent: {ok_count}/{len(results)} 成功",
ip_address=ip_address,
))
return BatchAgentResult(results=results)
@router.get("/{id}", response_model=dict)
async def get_server(
id: int,
@@ -927,7 +1174,7 @@ async def install_agent_remote(
if result["exit_code"] != 0:
raise HTTPException(
status_code=400,
detail=f"安装脚本失败 (exit {result['exit_code']}): {result['stderr'][:500]}",
detail=f"安装失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}",
)
# Audit
@@ -951,6 +1198,80 @@ async def install_agent_remote(
}
@router.post("/{id}/uninstall-agent", response_model=dict)
async def uninstall_agent_remote(
request: Request,
id: int,
admin: Admin = Depends(get_current_admin),
service: ServerService = Depends(get_server_service),
db: AsyncSession = Depends(get_db),
):
"""Uninstall Nexus Agent on the managed server via SSH.
Stops the systemd service, removes the install directory and log file,
then clears the agent metadata (is_online, agent_version) in the database.
"""
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
import shlex
server = await service.get_server(id)
if not server:
raise HTTPException(status_code=404, detail="Server not found")
base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
if not base_url:
raise HTTPException(
status_code=400,
detail="NEXUS_API_BASE_URL 未配置，无法下载卸载脚本。",
)
ssh_user = (server.username or "root").strip()
# Build uninstall command — download uninstall.sh then execute
uninstall_cmd = (
f"TMP=$(mktemp) && curl -fsSL {shlex.quote(base_url)}/agent/uninstall.sh -o \"$TMP\" "
f"&& bash \"$TMP\" ; RC=$?; rm -f \"$TMP\"; exit $RC"
)
uninstall_cmd = _sudo_wrap(uninstall_cmd, ssh_user)
# Execute via SSH
try:
result = await exec_ssh_command(server, uninstall_cmd, timeout=60)
except Exception as e:
raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}")
if result["exit_code"] != 0:
raise HTTPException(
status_code=400,
detail=f"卸载失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}",
)
# Clear agent metadata in database
server.is_online = False
server.agent_version = None
await db.commit()
# Audit
ip_address = request.client.host if request.client else ""
audit_repo = AuditLogRepositoryImpl(db)
await audit_repo.create(AuditLog(
admin_username=admin.username,
action="uninstall_agent",
target_type="server",
target_id=id,
detail=f"Agent 远程卸载: {server.name} ({server.domain})",
ip_address=ip_address,
))
return {
"success": True,
"server_id": id,
"server_name": server.name,
"stdout": result["stdout"][:3000],
"exit_code": result["exit_code"],
}
@router.post("/{id}/agent-install-cmd", response_model=dict)
async def get_agent_install_cmd(
id: int,
@@ -1115,7 +1436,7 @@ async def upgrade_agent(
logger.warning(f"Rollback failed for server {id}: {rb_err}")
raise HTTPException(
status_code=400,
detail=f"升级失败 (exit {result['exit_code']}): {result['stderr'][:300]}，已回滚至旧版本",
detail=f"升级失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}，已回滚至旧版本",
)
# Audit
@@ -1212,8 +1533,11 @@ def _sync_log_to_dict(log, server_name: str = None) -> dict:
"sync_mode": log.sync_mode,
"files_total": log.files_total,
"files_transferred": log.files_transferred,
"files_skipped": log.files_skipped,
"bytes_transferred": log.bytes_transferred,
"duration_seconds": log.duration_seconds,
"error_message": log.error_message,
"diff_summary": log.diff_summary,
"started_at": str(log.started_at) if log.started_at else None,
"finished_at": str(log.finished_at) if log.finished_at else None,
}
server/api/sync_v2.py
+348 -2
View File
@@ -9,7 +9,7 @@ import logging
from fastapi import APIRouter, Depends, HTTPException, Request
from server.api.schemas import SyncFiles, SyncBrowse, SyncPreview, FileOperation
from server.api.schemas import SyncFiles, SyncBrowse, SyncBrowseLocal, SyncPreview, FileOperation, LocalFileOperation, SyncVerify
from server.api.dependencies import get_server_service
from server.application.services.sync_engine_v2 import SyncEngineV2
from server.infrastructure.database.server_repo import ServerRepositoryImpl
@@ -191,7 +191,115 @@ async def browse_directory(
return {"path": path, "entries": entries}
# ── Shared audit helper for sync routes ──
@router.post("/browse-local")
async def browse_local_directory(
payload: SyncBrowseLocal,
request: Request,
admin: Admin = Depends(get_current_admin),
):
"""Browse a local directory on the Nexus server (for upload file manager).
Only allows browsing under /tmp/nexus_upload_* directories.
Returns entries with name, is_dir, size.
"""
import os
path = payload.path.rstrip("/") or "/"
# Security: only allow browsing under /tmp/nexus_upload_*
real_path = os.path.realpath(path)
if not real_path.startswith("/tmp/nexus_upload_"):
raise HTTPException(status_code=403, detail="仅允许浏览上传临时目录")
if not os.path.isdir(real_path):
raise HTTPException(status_code=404, detail="目录不存在")
entries = []
try:
for entry in sorted(os.scandir(real_path), key=lambda e: (not e.is_dir(), e.name.lower())):
try:
stat = entry.stat()
entries.append({
"name": entry.name,
"is_dir": entry.is_dir(),
"size": stat.st_size if not entry.is_dir() else 0,
})
except OSError:
continue
except PermissionError:
raise HTTPException(status_code=403, detail="无权限访问该目录")
return {"path": real_path, "entries": entries}
@router.post("/local-file-ops")
async def local_file_operation(
payload: LocalFileOperation,
request: Request,
admin: Admin = Depends(get_current_admin),
):
"""Delete or rename a file/directory in the local upload staging area.
Only allows operations under /tmp/nexus_upload_* directories.
"""
import os
import shutil
path = payload.path.rstrip("/") or "/"
# Security: only allow operations under /tmp/nexus_upload_*
real_path = os.path.realpath(path)
if not real_path.startswith("/tmp/nexus_upload_"):
raise HTTPException(status_code=403, detail="仅允许操作上传临时目录中的文件")
if payload.operation == "delete":
if not os.path.exists(real_path):
raise HTTPException(status_code=404, detail="文件或目录不存在")
try:
if os.path.isdir(real_path):
shutil.rmtree(real_path)
else:
os.unlink(real_path)
except OSError as e:
raise HTTPException(status_code=500, detail=f"删除失败: {e}")
await _audit_sync(
"local_file_delete", "sync", 0,
f"删除: {path}", admin.username, request,
)
return {"success": True, "operation": "delete", "path": path}
elif payload.operation == "rename":
if not payload.new_name:
raise HTTPException(status_code=400, detail="new_name 必填")
# Sanitize: strip slashes and path traversal
safe_name = payload.new_name.replace("/", "_").replace("\\", "_").strip()
if not safe_name or safe_name == "." or safe_name == "..":
raise HTTPException(status_code=400, detail="无效的新名称")
if not os.path.exists(real_path):
raise HTTPException(status_code=404, detail="文件或目录不存在")
parent_dir = os.path.dirname(real_path)
new_path = os.path.join(parent_dir, safe_name)
# Security: new path must also be under upload dir
if not os.path.realpath(new_path).startswith("/tmp/nexus_upload_"):
raise HTTPException(status_code=403, detail="目标路径不在上传临时目录内")
if os.path.exists(new_path):
raise HTTPException(status_code=400, detail="目标名称已存在")
try:
os.rename(real_path, new_path)
except OSError as e:
raise HTTPException(status_code=500, detail=f"重命名失败: {e}")
await _audit_sync(
"local_file_rename", "sync", 0,
f"重命名: {path} → {safe_name}", admin.username, request,
)
return {"success": True, "operation": "rename", "path": path, "new_name": safe_name}
async def _audit_sync(action: str, target_type: str, target_id: int, detail: str, operator: str, request: Request):
"""Create audit log entry for sync operations"""
@@ -303,3 +411,241 @@ async def upload_file(
"filename": safe_filename,
"size": len(content),
}
# ── ZIP Upload & Extract (for push workflow) ──
MAX_ZIP_FILE_SIZE = 524_288_000 # 500 MB
@router.post("/upload-zip")
async def upload_zip(
request: Request,
admin: Admin = Depends(get_current_admin),
):
"""Upload a ZIP file to Nexus, extract it, return the source path for rsync push.
Multipart form fields:
- file: UploadFile (required, must be .zip)
Returns {"source_path": "/tmp/nexus_upload_xxx/", "file_count": N}
"""
import os
import uuid
import zipfile
import shutil
form = await request.form()
file = form.get("file")
if not file or not hasattr(file, "filename") or not file.filename:
raise HTTPException(status_code=400, detail="请选择 ZIP 文件")
filename = file.filename
if not filename.lower().endswith(".zip"):
raise HTTPException(status_code=400, detail="仅支持 .zip 格式")
# Read with size check
content = await file.read()
if len(content) > MAX_ZIP_FILE_SIZE:
raise HTTPException(status_code=400, detail=f"文件大小超过限制 ({MAX_ZIP_FILE_SIZE // (1024*1024)}MB)")
if len(content) == 0:
raise HTTPException(status_code=400, detail="文件为空")
# Create temp directory
upload_id = uuid.uuid4().hex[:12]
extract_dir = f"/tmp/nexus_upload_{upload_id}"
zip_path = f"{extract_dir}.zip"
try:
os.makedirs(extract_dir, exist_ok=True)
# Save ZIP
with open(zip_path, "wb") as f:
f.write(content)
# Validate and extract with zip slip protection
file_count = 0
with zipfile.ZipFile(zip_path, "r") as zf:
for info in zf.infolist():
# Zip slip: ensure no path traversal
real_path = os.path.realpath(os.path.join(extract_dir, info.filename))
if not real_path.startswith(os.path.realpath(extract_dir)):
raise HTTPException(status_code=400, detail=f"ZIP 包含非法路径: {info.filename}")
if info.is_dir():
continue
file_count += 1
zf.extractall(extract_dir)
# Clean up ZIP file
os.unlink(zip_path)
if file_count == 0:
shutil.rmtree(extract_dir, ignore_errors=True)
raise HTTPException(status_code=400, detail="ZIP 文件为空（无文件）")
# Build file listing (relative paths, capped at 500)
files = []
for root, dirs, fnames in os.walk(extract_dir):
for fn in fnames:
rel = os.path.relpath(os.path.join(root, fn), extract_dir)
files.append(rel)
if len(files) >= 500:
break
if len(files) >= 500:
break
files_truncated = file_count > len(files)
# Audit
await _audit_sync(
"upload_zip", "sync", 0,
f"上传 ZIP: {filename} ({len(content)} bytes) → {extract_dir} ({file_count} 个文件)",
admin.username, request,
)
return {
"success": True,
"source_path": extract_dir,
"file_count": file_count,
"files": files,
"files_truncated": files_truncated,
"filename": filename,
"size": len(content),
}
except HTTPException:
shutil.rmtree(extract_dir, ignore_errors=True)
try:
os.unlink(zip_path)
except OSError:
pass
raise
except zipfile.BadZipFile:
shutil.rmtree(extract_dir, ignore_errors=True)
try:
os.unlink(zip_path)
except OSError:
pass
raise HTTPException(status_code=400, detail="无效的 ZIP 文件")
except Exception as e:
shutil.rmtree(extract_dir, ignore_errors=True)
try:
os.unlink(zip_path)
except OSError:
pass
raise HTTPException(status_code=500, detail=f"解压失败: {e}")
# ── Post-Push Verification (md5sum comparison) ──
@router.post("/verify")
async def verify_sync(
payload: SyncVerify,
request: Request,
admin: Admin = Depends(get_current_admin),
):
"""Verify pushed files by comparing md5sums between source and target servers.
Walks the local source directory and computes md5 for each file,
then runs md5sum on each target server via SSH and compares.
Returns per-server results: matched / missing / mismatched file lists.
"""
import asyncio
import hashlib
import shlex
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
from server.infrastructure.database.server_repo import ServerRepositoryImpl
session = request.state.db
if not os.path.isdir(payload.source_path):
raise HTTPException(status_code=400, detail=f"源路径不存在: {payload.source_path}")
# Build local file manifest: {relative_path: md5hex}
local_files = {}
for root, dirs, fnames in os.walk(payload.source_path):
for fn in fnames:
full = os.path.join(root, fn)
rel = os.path.relpath(full, payload.source_path)
if len(local_files) >= payload.max_files:
break
try:
h = hashlib.md5()
with open(full, "rb") as f:
for chunk in iter(lambda: f.read(8192), b""):
h.update(chunk)
local_files[rel] = h.hexdigest()
except OSError:
continue
if len(local_files) >= payload.max_files:
break
if not local_files:
raise HTTPException(status_code=400, detail="源目录无文件可校验")
# Verify against each server concurrently
sem = asyncio.Semaphore(5)
results = {}
async def _verify_one(sid: int):
async with sem:
server = await ServerRepositoryImpl(session).get_by_id(sid)
if not server:
results[sid] = {"server_id": sid, "server_name": "", "error": "服务器不存在"}
return
dest = payload.target_path or server.target_path or "/tmp/sync"
# Run md5sum on remote server for the same relative paths
# Build a file list for md5sum to check
file_list = " ".join(shlex.quote(f) for f in local_files.keys())
cmd = f"cd {shlex.quote(dest)} && md5sum {file_list} 2>/dev/null"
try:
r = await exec_ssh_command(server, cmd, timeout=60)
except Exception as e:
results[sid] = {
"server_id": sid, "server_name": server.name,
"error": f"SSH 连接失败: {e}",
}
return
# Parse md5sum output: "hash filename"
remote_files = {}
if r["exit_code"] == 0:
for line in r["stdout"].strip().split("\n"):
parts = line.split(None, 1)
if len(parts) == 2:
remote_files[parts[1]] = parts[0]
matched = []
missing = []
mismatched = []
for rel, local_hash in local_files.items():
if rel not in remote_files:
missing.append(rel)
elif remote_files[rel] != local_hash:
mismatched.append(rel)
else:
matched.append(rel)
results[sid] = {
"server_id": sid,
"server_name": server.name,
"matched": len(matched),
"missing": len(missing),
"mismatched": len(mismatched),
"missing_files": missing[:50],
"mismatched_files": mismatched[:50],
}
await asyncio.gather(*[_verify_one(sid) for sid in payload.server_ids])
# Audit
await _audit_sync(
"sync_verify", "sync", 0,
f"推送校验: {payload.source_path} → {len(payload.server_ids)} 台服务器",
admin.username, request,
)
return {"results": results, "total_local_files": len(local_files)}
server/api/websocket.py
+25
View File
@@ -427,3 +427,28 @@ async def broadcast_system_event(event_type: str, message: str):
await _dispatch_ws_message(msg)
async def broadcast_sync_progress(
batch_id: str, server_id: int, server_name: str,
status: str, completed: int, failed: int, total: int,
error_message: str = None, duration_seconds: int = None,
):
"""Broadcast per-server push progress to all WebSocket clients.
Called from sync_engine_v2._sync_one() after each server completes.
Frontend uses batch_id to filter stale events from previous pushes.
"""
msg = {
"type": "sync_progress",
"batch_id": batch_id,
"server_id": server_id,
"server_name": server_name,
"status": status,
"completed": completed,
"failed": failed,
"total": total,
"error_message": error_message,
"duration_seconds": duration_seconds,
}
await _dispatch_ws_message(msg)
server/application/services/sync_engine_v2.py
+176 -80
View File
@@ -1,33 +1,35 @@
"""Nexus — Sync Engine v2 (Step 4 Upgrade)
"""Nexus — Sync Engine v2
S1: Existing rsync push retained as Sync file mode
S5: Parallel execution + result aggregation with Semaphore
Rsync push from Nexus server to target servers via SSH.
Supports: local source path / uploaded ZIP extraction.
Parallel execution with Semaphore concurrency control.
"""
import asyncio
import json
import logging
import os
import re
import shutil
import tempfile
import uuid
from datetime import datetime, timezone
from typing import List, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from server.domain.models import Server, SyncLog, AuditLog
from server.infrastructure.database.server_repo import ServerRepositoryImpl
from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl
from server.infrastructure.redis.client import get_redis
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command, ssh_pool
from server.infrastructure.database.crypto import decrypt_value
logger = logging.getLogger("nexus.sync_v2")
# S5: Semaphore for concurrency control
MAX_CONCURRENT = 10
RSYNC_TIMEOUT = 300
class SyncEngineV2:
"""Unified sync engine: file sync + preview, with parallel execution (S5)"""
"""Unified sync engine: rsync push from Nexus + preview + ZIP upload"""
def __init__(
self,
@@ -41,7 +43,7 @@ class SyncEngineV2:
self.audit_repo = audit_repo
self.retry_repo = retry_repo
# ── S1: Sync File Mode (rsync-based) ──
# ── Rsync push from Nexus to target servers ──
async def sync_files(
self,
@@ -54,12 +56,15 @@ class SyncEngineV2:
concurrency: int = 10,
trigger_type: str = "manual",
) -> dict:
"""S1: File sync using rsync over SSH"""
"""Push files from Nexus to target servers via rsync over SSH"""
if not os.path.isdir(source_path):
return {"total": 0, "completed": 0, "failed": 0, "results": {},
"error": f"源路径不存在: {source_path}"}
batch_id = uuid.uuid4().hex[:12]
concurrency = min(concurrency, MAX_CONCURRENT)
sem = asyncio.Semaphore(concurrency)
results = {}
# Build server list
servers = []
for sid in server_ids:
server = await self.server_repo.get_by_id(sid)
@@ -70,15 +75,16 @@ class SyncEngineV2:
completed = 0
failed = 0
_counter_lock = asyncio.Lock()
results = {}
async def _sync_one(server: Server):
nonlocal completed, failed
async with sem:
# Create sync log
dest = target_path or server.target_path or "/tmp/sync"
sync_log = SyncLog(
server_id=server.id,
source_path=source_path,
target_path=target_path or server.target_path or "/tmp/sync",
target_path=dest,
trigger_type=trigger_type,
operator=operator,
status="running",
@@ -87,18 +93,15 @@ class SyncEngineV2:
)
sync_log = await self.sync_log_repo.create(sync_log)
# Execute rsync
import shlex
rsync_cmd = f"rsync -az --delete {shlex.quote(source_path)}/ {shlex.quote(target_path or server.target_path or '/tmp/sync')}/"
result = await exec_ssh_command(server, rsync_cmd, timeout=300)
result = await _rsync_push(server, source_path, dest, sync_mode)
# Update sync log
sync_log.status = "success" if result["exit_code"] == 0 else "failed"
sync_log.finished_at = datetime.now(timezone.utc)
sync_log.duration_seconds = int(
(sync_log.finished_at - sync_log.started_at).total_seconds()
) if sync_log.started_at else 0
sync_log.error_message = result["stderr"][:1000] if result["exit_code"] != 0 else None
sync_log.diff_summary = result["stdout"][:2000] if result["exit_code"] == 0 else None
sync_log = await self.sync_log_repo.update(sync_log)
async with _counter_lock:
@@ -107,14 +110,28 @@ class SyncEngineV2:
else:
failed += 1
results[server.id] = sync_log
return sync_log
# Broadcast per-server progress via WebSocket
try:
from server.api.websocket import broadcast_sync_progress
await broadcast_sync_progress(
batch_id=batch_id,
server_id=server.id,
server_name=server.name,
status=sync_log.status,
completed=completed,
failed=failed,
total=total,
error_message=sync_log.error_message[:200] if sync_log.error_message else None,
duration_seconds=sync_log.duration_seconds,
)
except Exception as e:
logger.debug(f"WS broadcast failed for server {server.id}: {e}")
results[server.id] = sync_log
# S5: Parallel execution with concurrency control
tasks = [asyncio.create_task(_sync_one(s)) for s in servers]
await asyncio.gather(*tasks, return_exceptions=True)
# Audit
await self._audit(
"sync_files", "sync", 0,
f"文件推送: {source_path} → {total} 台服务器 ({completed} 成功, {failed} 失败)",
@@ -125,24 +142,10 @@ class SyncEngineV2:
"total": total,
"completed": completed,
"failed": failed,
"batch_id": batch_id,
"results": {sid: _log_to_dict(log) for sid, log in results.items()},
}
async def _audit(self, action: str, target_type: str, target_id: int, detail: str, operator: str = "admin"):
"""Create audit log entry"""
try:
audit_log = AuditLog(
admin_username=operator,
action=action,
target_type=target_type,
target_id=target_id,
detail=detail,
)
await self.audit_repo.create(audit_log)
except Exception as e:
logger.error(f"Audit log failed: {e}")
# ── Preview (dry-run) ──
async def preview_sync(
@@ -153,49 +156,26 @@ class SyncEngineV2:
sync_mode: str = "incremental",
verbose: bool = False,
) -> dict:
"""Run rsync --dry-run --stats against one server; no data is transferred.
Returns parsed stats and optional file list (verbose, capped at 200 lines).
"""
import re
"""Dry-run rsync --dry-run --stats against one server"""
server = await self.server_repo.get_by_id(server_id)
if not server:
return {"error": "Server not found", "exit_code": -1}
target = target_path or server.target_path or "/tmp/sync"
if not os.path.isdir(source_path):
return {"error": f"源路径不存在: {source_path}", "exit_code": -1}
# Build flags matching the actual sync mode
mode_flags = {
"incremental": "-az",
"full": "-az --delete",
"overwrite": "-az --inplace",
"checksum": "-az --checksum",
}
flags = mode_flags.get(sync_mode, "-az")
dry_flags = f"{flags} --dry-run --stats"
if verbose:
dry_flags += " -v"
dest = target_path or server.target_path or "/tmp/sync"
result = await _rsync_push(server, source_path, dest, sync_mode, dry_run=True, verbose=verbose)
cmd = (
f"rsync {dry_flags} "
f"{shlex.quote(source_path)}/ "
f"{shlex.quote(target)}/"
)
result = await exec_ssh_command(server, cmd, timeout=60)
stdout = result.get("stdout", "")
stderr = result.get("stderr", "")
exit_code = result.get("exit_code", -1)
# exit_code 23 = partial transfer (some files not transferred) is OK for dry-run
if exit_code not in (0, 23):
if result["exit_code"] not in (0, 23):
return {
"server_id": server_id,
"server_name": server.name,
"error": stderr[:500] or f"rsync exited with code {exit_code}",
"exit_code": exit_code,
"error": result["stderr"][:500] or f"rsync exited with code {result['exit_code']}",
"exit_code": result["exit_code"],
}
stdout = result.get("stdout", "")
stats = _parse_rsync_stats(stdout)
files: list[str] = []
@@ -220,26 +200,138 @@ class SyncEngineV2:
"server_id": server_id,
"server_name": server.name,
"source_path": source_path,
"target_path": target,
"target_path": dest,
"sync_mode": sync_mode,
"dry_run": True,
"stats": stats,
"files": files,
"files_truncated": files_truncated,
"exit_code": exit_code,
"exit_code": result["exit_code"],
}
async def _audit(self, action: str, target_type: str, target_id: int, detail: str, operator: str = "admin"):
try:
audit_log = AuditLog(
admin_username=operator,
action=action,
target_type=target_type,
target_id=target_id,
detail=detail,
)
await self.audit_repo.create(audit_log)
except Exception as e:
logger.error(f"Audit log failed: {e}")
# ── Rsync execution on Nexus host ──
async def _rsync_push(
server: Server,
source_path: str,
target_path: str,
sync_mode: str = "incremental",
dry_run: bool = False,
verbose: bool = False,
) -> dict:
"""Execute rsync on Nexus host, pushing to target server via SSH.
Handles both password auth (sshpass) and key auth (temp key file).
Returns {exit_code, stdout, stderr}.
"""
import shlex
ssh_user = server.username or "root"
ssh_host = server.domain
ssh_port = server.port or 22
dest = f"{ssh_user}@{ssh_host}:{target_path}"
# Build rsync args
args = ["rsync", "-az"]
if sync_mode == "full":
args.append("--delete")
elif sync_mode == "checksum":
args.append("--checksum")
elif sync_mode == "overwrite":
args.append("--inplace")
if dry_run:
args.extend(["--dry-run", "--stats"])
if verbose:
args.append("-v")
# SSH options: port + no strict host key checking (matches asyncssh pool behavior)
ssh_opts = f"ssh -o StrictHostKeyChecking=no -p {ssh_port}"
# Auth: prepare temp key file or sshpass
key_file = None
env_override = None
try:
if server.auth_method == "password" and server.password:
password = decrypt_value(server.password)
# sshpass -p requires the password as an argument (visible in /proc on Linux,
# but acceptable for internal ops tool; sshpass -e from env is slightly safer)
env_override = {"SSHPASS": password}
args = ["sshpass", "-e"] + args
ssh_opts += " -o PubkeyAuthentication=no"
elif server.ssh_key_configured and server.ssh_key_private:
key_content = decrypt_value(server.ssh_key_private)
key_file = tempfile.NamedTemporaryFile(
mode="w", suffix=".key", prefix=f"nexus_rsync_{server.id}_",
delete=False,
)
key_file.write(key_content)
key_file.close()
os.chmod(key_file.name, 0o600)
ssh_opts += f" -i {shlex.quote(key_file.name)}"
elif server.ssh_key_path:
ssh_opts += f" -i {shlex.quote(server.ssh_key_path)}"
else:
return {"exit_code": -1, "stdout": "", "stderr": f"服务器 {server.name} 无有效 SSH 凭据"}
args.extend(["-e", ssh_opts])
args.append(source_path.rstrip("/") + "/")
args.append(dest.rstrip("/") + "/")
# Execute rsync on Nexus host
proc = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**os.environ, **(env_override or {})},
)
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(), timeout=RSYNC_TIMEOUT + 30
)
except asyncio.TimeoutError:
proc.kill()
await proc.communicate()
return {"exit_code": -1, "stdout": "", "stderr": f"rsync 超时 ({RSYNC_TIMEOUT}s)"}
return {
"exit_code": proc.returncode or 0,
"stdout": stdout.decode("utf-8", errors="replace")[:10000],
"stderr": stderr.decode("utf-8", errors="replace")[:10000],
}
finally:
if key_file:
try:
os.unlink(key_file.name)
except OSError:
pass
def _parse_rsync_stats(output: str) -> dict:
"""Extract key numbers from rsync --stats output."""
import re
stats: dict = {}
patterns = {
"files_total": r"Number of files:\s+([\d,]+)",
"files_created": r"Number of created files:\s+([\d,]+)",
"files_deleted": r"Number of deleted files:\s+([\d,]+)",
"files_transferred": r"Number of regular files transferred:\s+([\d,]+)",
"total_size_bytes": r"Total file size:\s+([\d,]+)\s+bytes",
"files_total": r"Number of files:\s+([\d,]+)",
"files_created": r"Number of created files:\s+([\d,]+)",
"files_deleted": r"Number of deleted files:\s+([\d,]+)",
"files_transferred": r"Number of regular files transferred:\s+([\d,]+)",
"total_size_bytes": r"Total file size:\s+([\d,]+)\s+bytes",
"transfer_size_bytes": r"Total transferred file size:\s+([\d,]+)\s+bytes",
}
for key, pattern in patterns.items():
@@ -256,6 +348,10 @@ def _log_to_dict(log: SyncLog) -> dict:
return {
"id": log.id, "server_id": log.server_id,
"status": log.status, "sync_mode": log.sync_mode,
"files_transferred": log.files_transferred,
"files_skipped": log.files_skipped,
"bytes_transferred": log.bytes_transferred,
"diff_summary": log.diff_summary,
"error_message": log.error_message,
"started_at": str(log.started_at) if log.started_at else None,
"finished_at": str(log.finished_at) if log.finished_at else None,
server/infrastructure/database/server_repo.py
+14 -6
View File
@@ -27,6 +27,8 @@ class ServerRepositoryImpl:
async def get_paginated(
self,
category: Optional[str] = None,
platform_id: Optional[int] = None,
node_id: Optional[int] = None,
offset: int = 0,
limit: int = 50,
) -> Tuple[List[Server], int]:
@@ -34,20 +36,26 @@ class ServerRepositoryImpl:
Returns (servers, total_count) — DB-level pagination for 2000+ servers.
"""
# Base query
base_query = select(Server)
filters = []
if category:
base_query = base_query.where(Server.category == category)
filters.append(Server.category == category)
if platform_id:
filters.append(Server.platform_id == platform_id)
if node_id:
filters.append(Server.node_id == node_id)
# Count query
count_query = select(func.count(Server.id))
if category:
count_query = count_query.where(Server.category == category)
for f in filters:
count_query = count_query.where(f)
count_result = await self.session.execute(count_query)
total = count_result.scalar_one() or 0
# Paginated data query
data_query = base_query.order_by(Server.id).offset(offset).limit(limit)
data_query = select(Server)
for f in filters:
data_query = data_query.where(f)
data_query = data_query.order_by(Server.id).offset(offset).limit(limit)
result = await self.session.execute(data_query)
servers = list(result.scalars().all())
web/app/push.html
+335 -23
View File
@@ -8,12 +8,48 @@
<main class="flex-1 overflow-y-auto p-6 max-w-3xl mx-auto w-full">
<!-- Push Form -->
<div class="bg-slate-900 rounded-xl border border-slate-800 p-6 space-y-4">
<div><label class="block text-sm text-slate-300 mb-2">源路径</label><input id="srcPath" placeholder="/home/deploy/source" class="w-full px-4 py-3 bg-slate-800 border border-slate-700 rounded-xl text-white text-sm"></div>
<!-- Source: ZIP upload only -->
<div>
<label class="text-sm text-slate-300 mb-2 block">上传 ZIP</label>
<div id="sourceUpload" class="space-y-2">
<div class="flex items-center gap-3">
<input id="zipFile" type="file" accept=".zip" class="flex-1 text-sm text-slate-300 file:mr-3 file:py-2 file:px-4 file:rounded-lg file:border-0 file:text-sm file:bg-brand file:text-white hover:file:bg-brand-dark">
<button onclick="uploadZip()" id="uploadZipBtn" class="px-4 py-2.5 bg-brand hover:bg-brand-dark text-white text-sm rounded-lg transition whitespace-nowrap">上传并解压</button>
</div>
<div id="uploadProgress" class="hidden flex items-center gap-2 text-sm text-slate-400">
<div class="animate-spin w-4 h-4 border-2 border-brand border-t-transparent rounded-full"></div>
上传并解压中...
</div>
<div id="uploadResult" class="hidden space-y-2">
<div class="flex items-center gap-3 px-3 py-2 bg-green-900/20 border border-green-700/30 rounded-lg">
<span class="text-green-400 text-sm">✓</span>
<span id="uploadedInfo" class="text-xs text-slate-400"></span>
<button onclick="clearUpload()" class="text-xs text-slate-500 hover:text-slate-300 ml-auto">✕ 清除</button>
</div>
<!-- File Manager -->
<div id="fileManager" class="bg-slate-800/50 border border-slate-700/50 rounded-lg overflow-hidden">
<div class="flex items-center gap-2 px-3 py-2 bg-slate-800 border-b border-slate-700/50">
<span class="text-xs text-slate-400">📁</span>
<span id="fmBreadcrumb" class="text-xs text-slate-300 flex-1 truncate"></span>
<span id="fmCount" class="text-xs text-slate-500"></span>
</div>
<div id="fmEntries" class="max-h-64 overflow-y-auto"></div>
</div>
</div>
</div>
</div>
<div><label class="block text-sm text-slate-300 mb-2">目标路径</label><input id="destPath" placeholder="默认使用服务器配置的目标路径" class="w-full px-4 py-3 bg-slate-800 border border-slate-700 rounded-xl text-white text-sm"></div>
<div><label class="block text-sm text-slate-300 mb-2">目标服务器</label>
<div class="flex gap-2 mb-2">
<div class="flex flex-wrap items-center gap-2 mb-2">
<select id="filterCategory" onchange="filterServers()" class="px-3 py-1.5 bg-slate-800 border border-slate-700 rounded-lg text-xs text-slate-300">
<option value="">全部分类</option>
</select>
<select id="filterPlatform" onchange="filterServers()" class="px-3 py-1.5 bg-slate-800 border border-slate-700 rounded-lg text-xs text-slate-300">
<option value="">全部平台</option>
</select>
<button onclick="selectAllServers()" class="text-xs text-brand-light hover:underline">全选</button>
<button onclick="deselectAllServers()" class="text-xs text-slate-400 hover:underline">全不选</button>
<button onclick="selectFilteredServers()" class="text-xs text-brand-light hover:underline">选中当前筛选</button>
</div>
<select id="targetServers" multiple class="w-full px-4 py-3 bg-slate-800 border border-slate-700 rounded-xl text-white text-sm h-40"></select>
<div id="serverCount" class="text-slate-500 text-xs mt-1">已选择 0 台服务器</div>
@@ -75,6 +111,9 @@
<div><label class="block text-sm text-slate-300 mb-2">批次大小</label><input id="batchSize" type="number" value="50" min="1" max="200" class="w-full px-4 py-3 bg-slate-800 border border-slate-700 rounded-xl text-white text-sm"></div>
</div>
<button onclick="doPush()" id="pushBtn" class="w-full py-3 bg-brand hover:bg-brand-dark text-white font-semibold rounded-xl transition disabled:opacity-50 disabled:cursor-not-allowed">开始推送</button>
<label class="flex items-center gap-2 text-sm text-slate-400 -mt-2">
<input type="checkbox" id="verifyAfterPush" class="accent-brand"> 推送后校验（md5sum 对比）
</label>
</div>
<!-- Progress Section -->
@@ -101,6 +140,15 @@
<h2 class="text-white font-semibold mb-3">最近推送记录</h2>
<div id="pushHistory" class="space-y-2"><div class="text-slate-500 text-center py-6 text-sm">加载中...</div></div>
</div>
<!-- Verify Result Section -->
<div id="verifySection" class="hidden mt-6 bg-slate-900 rounded-xl border border-slate-800 p-6 space-y-4">
<div class="flex items-center justify-between">
<h2 class="text-white font-semibold">推送校验结果</h2>
<button onclick="document.getElementById('verifySection').classList.add('hidden')" class="text-xs text-slate-500 hover:text-slate-300">✕ 关闭</button>
</div>
<div id="verifyResults" class="space-y-2 max-h-96 overflow-y-auto"></div>
</div>
</main>
</div>
<script src="/app/api.js"></script>
@@ -109,18 +157,208 @@
initLayout('push');
let _serversCache=[];
let _pushInProgress=false;
let _uploadSourcePath=''; // set after ZIP upload
let _fmCurrentPath=''; // current browsed path in file manager
let _pushBatchId=''; // current push batch_id for WS filtering
let _pushWs=null; // WebSocket connection for push progress
// ── ZIP upload ──
// ── WebSocket for real-time push progress ──
function connectPushWS(){
if(_pushWs&&_pushWs.readyState<=1)return; // already connected/connecting
const token=localStorage.getItem('access_token');
if(!token)return;
const proto=location.protocol==='https:'?'wss:':'ws:';
const url=proto+'//'+location.host+'/ws/alerts?token='+encodeURIComponent(token);
try{
_pushWs=new WebSocket(url);
_pushWs.onmessage=function(ev){
try{
const msg=JSON.parse(ev.data);
if(msg.type==='sync_progress'&&msg.batch_id===_pushBatchId){
updateProgressFromWS(msg);
}
}catch(e){}
};
_pushWs.onclose=function(){
_pushWs=null;
// Reconnect if push is still in progress
if(_pushInProgress) setTimeout(connectPushWS,3000);
};
_pushWs.onerror=function(){_pushWs=null};
}catch(e){_pushWs=null}
}
function updateProgressFromWS(msg){
const sid=msg.server_id;
const iconEl=document.getElementById('srv_icon_'+sid);
const statusEl=document.getElementById('srv_status_'+sid);
const detailEl=document.getElementById('srv_detail_'+sid);
if(!iconEl||!statusEl)return;
if(msg.status==='success'){
iconEl.textContent='✓';iconEl.className='shrink-0 w-5 h-5 flex items-center justify-center text-green-400';
statusEl.textContent='成功';statusEl.className='text-xs text-green-400';
if(msg.duration_seconds!=null){detailEl.textContent=msg.duration_seconds+'s';detailEl.classList.remove('hidden');detailEl.className='text-xs text-green-400/70'}
}else{
iconEl.textContent='✗';iconEl.className='shrink-0 w-5 h-5 flex items-center justify-center text-red-400';
statusEl.textContent='失败';statusEl.className='text-xs text-red-400';
if(msg.error_message){detailEl.textContent=msg.error_message.substring(0,80);detailEl.classList.remove('hidden');detailEl.className='text-xs text-red-400/70'}
}
// Update progress bar and counters
const total=msg.total||0;
const done=msg.completed+msg.failed;
const pct=total>0?Math.round((done/total)*100):0;
document.getElementById('progressBar').style.width=pct+'%';
document.getElementById('progressStats').textContent=done+'/'+total;
document.getElementById('countSuccess').textContent=msg.completed;
document.getElementById('countFailed').textContent=msg.failed;
document.getElementById('countPending').textContent=Math.max(0,total-done);
}
async function uploadZip(){
const file=document.getElementById('zipFile').files[0];
if(!file){toast('warning','请选择 ZIP 文件');return}
if(!file.name.toLowerCase().endsWith('.zip')){toast('warning','仅支持 .zip 格式');return}
const form=new FormData();form.append('file',file);
const btn=document.getElementById('uploadZipBtn');
btn.disabled=true;btn.textContent='上传中...';
document.getElementById('uploadProgress').classList.remove('hidden');
document.getElementById('uploadResult').classList.add('hidden');
try{
const r=await apiFetch(API+'/sync/upload-zip',{method:'POST',body:form});
if(!r){toast('error','上传失败：认证错误');return}
const d=await r.json();
if(!r.ok){toast('error','上传失败: '+(d.detail||r.status));return}
_uploadSourcePath=d.source_path;
document.getElementById('uploadProgress').classList.add('hidden');
document.getElementById('uploadResult').classList.remove('hidden');
document.getElementById('uploadedInfo').textContent=`已解压 ${d.file_count} 个文件 (${fmtBytes(d.size)})`;
// Open file manager at root
browseLocal(d.source_path);
toast('success',`ZIP 解压完成: ${d.file_count} 个文件`);
}catch(e){
toast('error','上传异常: '+e.message);
document.getElementById('uploadProgress').classList.add('hidden');
}finally{
btn.disabled=false;btn.textContent='上传并解压';
}
}
function clearUpload(){
_uploadSourcePath='';
_fmCurrentPath='';
document.getElementById('zipFile').value='';
document.getElementById('uploadResult').classList.add('hidden');
}
// ── File Manager (browse local upload dir) ──
async function browseLocal(path){
_fmCurrentPath=path;
// Update breadcrumb
const rel=path.replace(_uploadSourcePath,'').replace(/^\//,'');
document.getElementById('fmBreadcrumb').textContent=rel?'📁 '+rel:'📁 /';
document.getElementById('fmEntries').innerHTML='<div class="px-3 py-2 text-xs text-slate-500">加载中...</div>';
try{
const r=await apiFetch(API+'/sync/browse-local',{method:'POST',headers:apiHeadersJSON(),body:JSON.stringify({path})});
if(!r){toast('error','浏览失败');return}
const d=await r.json();
if(!r.ok){toast('error','浏览失败: '+(d.detail||r.status));return}
renderFileManager(d);
}catch(e){
document.getElementById('fmEntries').innerHTML='<div class="px-3 py-2 text-xs text-red-400">加载失败</div>';
}
}
function renderFileManager(d){
const entries=d.entries||[];
const dirs=entries.filter(e=>e.is_dir);
const files=entries.filter(e=>!e.is_dir);
document.getElementById('fmCount').textContent=`${dirs.length} 目录, ${files.length} 文件`;
let html='';
// Parent dir link (if not at root)
if(d.path!==_uploadSourcePath){
const parent=d.path.substring(0,d.path.lastIndexOf('/'))||_uploadSourcePath;
html+=`<div class="flex items-center gap-2 px-3 py-1.5 hover:bg-slate-700/30 cursor-pointer text-xs" onclick="browseLocal('${esc(parent)}')">
<span class="text-slate-400">📁</span><span class="text-slate-300">..</span></div>`;
}
// Directories
for(const e of dirs){
const fullPath=d.path+'/'+e.name;
html+=`<div class="group flex items-center gap-2 px-3 py-1.5 hover:bg-slate-700/30 cursor-pointer text-xs" onclick="browseLocal('${esc(fullPath)}')">
<span class="text-brand-light">📁</span><span class="text-slate-200 flex-1 truncate">${esc(e.name)}</span>
<span class="hidden group-hover:flex items-center gap-1 shrink-0">
<button onclick="event.stopPropagation();renameLocalFile('${esc(fullPath)}','${esc(e.name)}')" class="text-slate-500 hover:text-slate-300 px-1" title="重命名">✏</button>
<button onclick="event.stopPropagation();deleteLocalFile('${esc(fullPath)}','${esc(e.name)}')" class="text-slate-500 hover:text-red-400 px-1" title="删除">🗑</button>
</span></div>`;
}
// Files
for(const e of files){
const fullPath=d.path+'/'+e.name;
html+=`<div class="group flex items-center gap-2 px-3 py-1.5 text-xs">
<span class="text-slate-500">📄</span><span class="text-slate-300 flex-1 truncate">${esc(e.name)}</span>
<span class="text-slate-600 shrink-0">${fmtBytes(e.size)}</span>
<span class="hidden group-hover:flex items-center gap-1 shrink-0">
<button onclick="renameLocalFile('${esc(fullPath)}','${esc(e.name)}')" class="text-slate-500 hover:text-slate-300 px-1" title="重命名">✏</button>
<button onclick="deleteLocalFile('${esc(fullPath)}','${esc(e.name)}')" class="text-slate-500 hover:text-red-400 px-1" title="删除">🗑</button>
</span></div>`;
}
if(!dirs.length&&!files.length){
html='<div class="px-3 py-2 text-xs text-slate-500">空目录</div>';
}
document.getElementById('fmEntries').innerHTML=html;
}
async function deleteLocalFile(path,name){
if(!confirm(`确认删除 "${name}"？`))return;
try{
const r=await apiFetch(API+'/sync/local-file-ops',{method:'POST',headers:apiHeadersJSON(),body:JSON.stringify({operation:'delete',path})});
if(!r){toast('error','删除失败');return}
const d=await r.json();
if(!r.ok){toast('error','删除失败: '+(d.detail||r.status));return}
toast('success','已删除: '+name);
browseLocal(_fmCurrentPath);
}catch(e){toast('error','删除异常: '+e.message)}
}
async function renameLocalFile(path,oldName){
const newName=prompt('输入新名称:',oldName);
if(!newName||newName===oldName)return;
try{
const r=await apiFetch(API+'/sync/local-file-ops',{method:'POST',headers:apiHeadersJSON(),body:JSON.stringify({operation:'rename',path,new_name:newName})});
if(!r){toast('error','重命名失败');return}
const d=await r.json();
if(!r.ok){toast('error','重命名失败: '+(d.detail||r.status));return}
toast('success',`已重命名: ${oldName} → ${newName}`);
browseLocal(_fmCurrentPath);
}catch(e){toast('error','重命名异常: '+e.message)}
}
async function loadServers(){
const r=await apiFetch(API+'/servers/');if(!r)return;
const data=await r.json();_serversCache=data.items||data;
const sel=document.getElementById('targetServers');
sel.innerHTML=_serversCache.map(s=>`<option value="${s.id}">${esc(s.name)} (${esc(s.domain)})</option>`).join('');
// Load categories/platforms for filter dropdowns
try{
const cr=await apiFetch(API+'/servers/categories');if(cr){
const cd=await cr.json();
const catSel=document.getElementById('filterCategory');
const platSel=document.getElementById('filterPlatform');
(cd.categories||[]).forEach(c=>{
const o=document.createElement('option');o.value=c.name;o.textContent=c.name+' ('+c.count+')';catSel.appendChild(o);
});
(cd.platforms||[]).forEach(p=>{
const o=document.createElement('option');o.value=p.id;o.textContent=p.name+(p.category?' ('+p.category+')':'');platSel.appendChild(o);
});
}
}catch(e){}
// Load server list
await filterServers();
// Pre-select servers from batch push (servers.html)
const batchIds=sessionStorage.getItem('batchPushIds');
if(batchIds){
try{
const ids=JSON.parse(batchIds);
const sel=document.getElementById('targetServers');
Array.from(sel.options).forEach(o=>{
if(ids.includes(parseInt(o.value)))o.selected=true;
});
@@ -131,6 +369,21 @@
updateServerCount();
}
async function filterServers(){
const cat=document.getElementById('filterCategory').value;
const plat=document.getElementById('filterPlatform').value;
let url=API+'/servers/?per_page=200';
if(cat)url+='&category='+encodeURIComponent(cat);
if(plat)url+='&platform_id='+plat;
const r=await apiFetch(url);if(!r)return;
const data=await r.json();_serversCache=data.items||data;
const sel=document.getElementById('targetServers');
// Preserve existing selections
const prevSelected=new Set(Array.from(sel.selectedOptions).map(o=>parseInt(o.value)));
sel.innerHTML=_serversCache.map(s=>`<option value="${s.id}"${prevSelected.has(s.id)?' selected':''}>${esc(s.name)} (${esc(s.domain)})</option>`).join('');
updateServerCount();
}
document.getElementById('targetServers').addEventListener('change',updateServerCount);
function updateServerCount(){
const sel=document.getElementById('targetServers').selectedOptions.length;
@@ -142,14 +395,17 @@
function deselectAllServers(){
const s=document.getElementById('targetServers');Array.from(s.options).forEach(o=>o.selected=false);updateServerCount();
}
function selectFilteredServers(){
const s=document.getElementById('targetServers');Array.from(s.options).forEach(o=>o.selected=true);updateServerCount();
}
async function doPush(){
if(_pushInProgress)return;
const opts=document.getElementById('targetServers').selectedOptions;
const ids=Array.from(opts).map(o=>parseInt(o.value));
if(!ids.length){toast('warning','请选择至少一台服务器');return}
const srcPath=document.getElementById('srcPath').value;
if(!srcPath){toast('warning','请输入源路径');return}
const srcPath=_uploadSourcePath;
if(!srcPath){toast('warning','请先上传 ZIP 文件');return}
const syncMode=document.querySelector('input[name="syncMode"]:checked')?.value||'incremental';
const body={
server_ids:ids,
@@ -161,22 +417,31 @@
};
_pushInProgress=true;
_pushBatchId=''; // reset, will be set from HTTP response
const btn=document.getElementById('pushBtn');btn.disabled=true;btn.textContent='推送中...';
// Show progress section with pending states
showProgress(ids);
// Connect WebSocket for real-time progress
connectPushWS();
try{
const r=await apiFetch(API+'/sync/files',{method:'POST',headers:apiHeadersJSON(),body:JSON.stringify(body)});
if(!r){toast('error','认证失败或请求被拒绝');resetProgress();return}
const d=await r.json();
if(d.batch_id)_pushBatchId=d.batch_id; // for late WS filtering
updateProgressFromResult(d);
toast(d.failed>0?'warning':'success',`推送完成: ${d.completed||0} 成功, ${d.failed||0} 失败`);
// Post-push verification if checked
if(document.getElementById('verifyAfterPush').checked&&d.completed>0){
doVerify(ids,_uploadSourcePath,document.getElementById('destPath').value||null);
}
}catch(e){
toast('error','推送请求失败: '+e.message);
resetProgress();
}finally{
_pushInProgress=false;btn.disabled=false;btn.textContent='开始推送';
if(_pushWs){try{_pushWs.close()}catch(e){}_pushWs=null}
loadHistory(); // Refresh history
}
}
@@ -243,24 +508,71 @@
document.getElementById('progressSection').classList.add('hidden');
}
// ── Post-Push Verify (md5sum) ──
async function doVerify(ids,sourcePath,targetPath){
const section=document.getElementById('verifySection');
section.classList.remove('hidden');
document.getElementById('verifyResults').innerHTML='<div class="text-slate-500 text-sm text-center py-4">校验中...</div>';
try{
const body={server_ids:ids,source_path:sourcePath,target_path:targetPath||null};
const r=await apiFetch(API+'/sync/verify',{method:'POST',headers:apiHeadersJSON(),body:JSON.stringify(body)});
if(!r){document.getElementById('verifyResults').innerHTML='<div class="text-red-400 text-sm">校验请求失败</div>';return}
const d=await r.json();
if(!r.ok){document.getElementById('verifyResults').innerHTML=`<div class="text-red-400 text-sm">${esc(d.detail||'校验失败')}</div>`;return}
const results=d.results||{};
document.getElementById('verifyResults').innerHTML=Object.values(results).map(v=>{
if(v.error)return `<div class="bg-slate-800/50 rounded-lg px-4 py-3 border-l-2 border-l-red-500">
<span class="text-red-400 text-sm">✗</span> <span class="text-sm text-slate-300">${esc(v.server_name||'#'+v.server_id)}</span>
<span class="text-xs text-red-400 ml-2">${esc(v.error)}</span></div>`;
const allOk=v.missing===0&&v.mismatched===0;
const border=allOk?'border-l-green-500':'border-l-amber-500';
let detail='';
if(v.missing>0)detail+=`<div class="text-xs text-amber-400 mt-1">缺失 ${v.missing} 文件: ${v.missing_files.slice(0,5).map(esc).join(', ')}${v.missing>5?' ...':''}</div>`;
if(v.mismatched>0)detail+=`<div class="text-xs text-red-400 mt-1">不一致 ${v.mismatched} 文件: ${v.mismatched_files.slice(0,5).map(esc).join(', ')}${v.mismatched>5?' ...':''}</div>`;
return `<div class="bg-slate-800/50 rounded-lg px-4 py-3 border-l-2 ${border}">
<span class="${allOk?'text-green-400':'text-amber-400'} text-sm">${allOk?'✓':'⚠'}</span>
<span class="text-sm text-slate-300">${esc(v.server_name||'#'+v.server_id)}</span>
<span class="text-xs text-slate-500 ml-2">${v.matched} 匹配${v.missing>0?`, ${v.missing} 缺失`:''}${v.mismatched>0?`, ${v.mismatched} 不一致`:''}</span>
${detail}</div>`;
}).join('');
}catch(e){
document.getElementById('verifyResults').innerHTML=`<div class="text-red-400 text-sm">校验异常: ${esc(e.message)}</div>`;
}
}
// ── Push History (recent sync logs) ──
async function loadHistory(){
try{
const r=await apiFetch(API+'/servers/logs?limit=10');if(!r)return;
const r=await apiFetch(API+'/servers/logs?limit=15');if(!r)return;
const res=await r.json();const logs=res.items||res;
document.getElementById('pushHistory').innerHTML=logs.length?logs.map(l=>`<div class="bg-slate-900 rounded-lg border border-slate-800 px-4 py-3 flex items-center justify-between">
<div class="flex items-center gap-3">
<span class="${l.status==='success'?'text-green-400':'text-red-400'} text-sm">${l.status==='success'?'✓':'✗'}</span>
<div>
<span class="text-sm text-slate-300">${esc(l.server_name||'#'+l.server_id)}</span>
<span class="text-slate-600 mx-1">·</span>
<span class="text-xs text-slate-500">${esc(l.operator||'system')}</span>
<span class="text-slate-600 mx-1">·</span>
<span class="text-xs text-slate-500">${esc(l.sync_mode||'file')}</span>
document.getElementById('pushHistory').innerHTML=logs.length?logs.map(l=>{
const ok=l.status==='success';
const border=ok?'border-l-2 border-l-green-500':'border-l-2 border-l-red-500';
let summary=`${esc(l.source_path||'')} → ${esc(l.target_path||'')}`;
if(l.duration_seconds!=null)summary+=` · ${l.duration_seconds}s`;
let detail='';
if(l.files_transferred!=null)detail+=`<div class="text-xs text-slate-400">传输: ${l.files_transferred} 文件</div>`;
if(l.bytes_transferred!=null&&l.bytes_transferred>0)detail+=`<div class="text-xs text-slate-400">大小: ${fmtBytes(l.bytes_transferred)}</div>`;
if(l.error_message)detail+=`<div class="text-xs text-red-400/80 break-all">${esc(l.error_message.substring(0,300))}</div>`;
if(l.diff_summary)detail+=`<details class="mt-1"><summary class="text-xs text-slate-500 cursor-pointer hover:text-slate-300">查看 diff</summary><pre class="text-xs text-slate-500 bg-slate-900 rounded p-2 mt-1 max-h-40 overflow-y-auto whitespace-pre-wrap">${esc(l.diff_summary.substring(0,2000))}</pre></details>`;
return `<div class="bg-slate-900 rounded-lg border border-slate-800 ${border} px-4 py-3 cursor-pointer hover:bg-slate-800/50 transition" onclick="this.querySelector('.hist-detail')?.classList.toggle('hidden')">
<div class="flex items-center justify-between">
<div class="flex items-center gap-3">
<span class="${ok?'text-green-400':'text-red-400'} text-sm">${ok?'✓':'✗'}</span>
<div>
<span class="text-sm text-slate-300">${esc(l.server_name||'#'+l.server_id)}</span>
<span class="text-slate-600 mx-1">·</span>
<span class="text-xs text-slate-500">${esc(l.operator||'system')}</span>
<span class="text-slate-600 mx-1">·</span>
<span class="text-xs text-slate-500">${esc(l.sync_mode||'file')}</span>
</div>
</div>
<div class="text-xs text-slate-500">${fmtTime(l.started_at)}</div>
</div>
</div>
<div class="text-xs text-slate-500">${fmtTime(l.started_at)}</div>
</div>`).join(''):'<div class="text-slate-500 text-center py-6 text-sm">暂无推送记录</div>';
<div class="text-xs text-slate-500 mt-1 truncate">${summary}</div>
<div class="hist-detail hidden mt-2 space-y-1 border-t border-slate-700/50 pt-2">${detail}</div>
</div>`;
}).join(''):'<div class="text-slate-500 text-center py-6 text-sm">暂无推送记录</div>';
}catch(e){
document.getElementById('pushHistory').innerHTML='<div class="text-slate-500 text-center py-6 text-sm">加载失败</div>';
}
@@ -283,8 +595,8 @@
const opts=document.getElementById('targetServers').selectedOptions;
const ids=Array.from(opts).map(o=>parseInt(o.value));
if(!ids.length){toast('warning','请先选择服务器');return}
const srcPath=document.getElementById('srcPath').value;
if(!srcPath){toast('warning','请输入源路径');return}
const srcPath=_uploadSourcePath;
if(!srcPath){toast('warning','请先上传 ZIP 文件');return}
const firstServerId=ids[0];
const syncMode=document.querySelector('input[name="syncMode"]:checked')?.value||'incremental';