feat(push): refactor B0-B6, /ws/sync channel, staging cleanup

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Your Name
2026-06-02 08:14:09 +08:00
parent 68fdcbae0a
commit 78798e3630
43 changed files with 2823 additions and 650 deletions
+96
View File
@@ -0,0 +1,96 @@
# 审计记录 — 推送页重构 B0–B6
## 审计信息
- **日期**: 2026-06-02
- **审计人**: Cursor Agent
- **触发原因**: 推送页重构 + M2 独立 WS + M9 清理 + 全量预览
## 审计范围(实际改动文件清单)
| 文件 | 状态 |
|------|------|
| server/api/websocket.py | ☑ 已审 |
| server/api/sync_v2.py | ☑ 已审 |
| server/api/schemas.py | ☑ 已审 |
| server/application/services/sync_engine_v2.py | ☑ 已审 |
| server/background/retry_runner.py | ☑ 已审 |
| server/background/upload_staging_cleanup.py | ☑ 已审 |
| server/infrastructure/database/sync_log_repo.py | ☑ 已审 |
| server/infrastructure/database/server_repo.py | ☑ 已审 |
| server/domain/models/__init__.py | ☑ 已审 |
| server/main.py | ☑ 已审 |
| frontend/src/pages/PushPage.vue | ☑ 已审 |
| frontend/src/composables/push/* | ☑ 已审 |
| frontend/src/components/push/* | ☑ 已审 |
| frontend/src/composables/useWebSocket.ts | ☑ 已审 |
| frontend/src/types/api.ts | ☑ 已审 |
| tests/test_sync_websocket.py | ☑ 已审 |
| tests/test_upload_staging_cleanup.py | ☑ 已审 |
## Step 3: 规则扫描H
- H1 鉴权:/ws/sync、/sync/* API 均 JWTWS token 校验同 alerts
- H2 注入:source_path `ensure_under_nexus_upload`target_path SSH 经现有引擎
- H3 凭据:无新增明文返回
- H4 并发:preview 并发 4sync gather 已有
- H5 错误:无静默吞掉 AttributeErrorsync_log update 已修)
- H6 资源:WS disconnect + timer clearstaging cleanup realpath 校验
- H7 跨层:API→Service→Repo 未破坏
- H8 审计:CUD 路径未减
## Step 4: Closure 表
| H | 判定 | 依据 |
|---|------|------|
| H1 | PASS | websocket sync_ws 复用 _verify_ws_tokensync_v2 Depends get_current_admin |
| H2 | PASS | upload_staging_cleanup realpath + prefixsync 路径既有校验 |
| H3 | PASS | 无 password 字段新增 |
| H4 | PASS | preview 限流 PREVIEW_CONCURRENCY=4 |
| H5 | PASS | retry_runner 成功判定修正;update() 存在 |
| H6 | PASS | onUnmounted dispose24h mtime 清理 |
| H7 | PASS | 未在 API 写 SQL |
| H8 | PASS | 推送/取消仍记审计(既有) |
## Step 5: 入口表
| 入口 | 鉴权 |
|------|------|
| POST /api/sync/files | JWT |
| POST /api/sync/preview | JWT |
| POST /api/sync/cancel | JWT |
| POST /api/sync/reconcile-stale-logs | JWT |
| POST /api/sync/diagnose | JWT |
| WS /ws/sync | JWT query |
| upload_staging_cleanup_loop | 内部任务 |
## Step 6: 输入→Sink
- ZIP/路径 → ensure_under_nexus_upload → rsync/SSH(既有池)
- WS batch_id → 前端过滤,不执行命令
## Step 7: 归类
```
websocket.py 8H 0FINDING
sync_engine_v2.py 8H 0FINDING
upload_staging_cleanup.py 6H 0FINDING
PushPage + composables 6H 0FINDING
──────────────────────────────────
总计 0FINDING
```
## Step 8: DoD
- ☑ 设计/计划/changelog 齐全
- ☑ pytest 新增用例通过(sync_ws、cleanup
- ☑ frontend build-only 通过
- ☑ 0 FINDING
## FINDING 列表
## 结论
☑ 审计通过,0 FINDING,可部署
@@ -0,0 +1,33 @@
# Changelog: 推送页 M2 / M9 / 全量预览
**日期**: 2026-06-02
## 摘要
推送进度改用独立 WebSocket `/ws/sync` 与 Redis 频道 `nexus:sync`;新增每小时清理过期 `/tmp/nexus_upload_*` 后台任务;预览支持全部选中服务器(并发 4,带进度条)。
## 动机
- M2:告警与推送进度解耦,避免多批次/多用户在同一 WS 上交叉
- M9:失败或放弃的上传临时目录不再长期占用磁盘
- 预览:取消 12 台上限,大批量推送前可看到完整预估
## 涉及文件
- `server/api/websocket.py``server/background/upload_staging_cleanup.py``server/main.py`
- `frontend/src/composables/push/usePushProgress.ts``usePushPreview.ts``types.ts`
- `frontend/src/components/push/PushPreviewDialog.vue``pages/PushPage.vue`
- `tests/test_sync_websocket.py``tests/test_upload_staging_cleanup.py`
## 迁移 / 重启
- **必须** `supervisorctl restart nexus`(新 WS 路由 + 后台任务)
- 前端需重新 build 并部署 `web/app/`
## 验证
```bash
ruff check server/api/websocket.py server/background/upload_staging_cleanup.py server/main.py
pytest tests/test_sync_websocket.py tests/test_upload_staging_cleanup.py -q
cd frontend && npm run build-only
```
@@ -0,0 +1,43 @@
# 推送页重构 Batch 0 + Batch 1
**日期**: 2026-06-02
## 变更摘要
- **B0(后端热修复)**:修复推送历史无法落库终态、rsync 统计未写入、重试误判成功、`max_retries` 默认值。
- **B1(前端 Critical**:合并进度 UI、修复全选/预览/WS 字段映射、取消状态、表单校验与重入保护。
## 动机
推送页存在 S1 级缺陷(`sync_log_repo.update` 缺失导致历史永远 `running`)及多处前后端契约不一致;按设计文档分批先交付 B0+B1。
## 涉及文件
- `server/infrastructure/database/sync_log_repo.py`
- `server/application/services/sync_engine_v2.py`
- `server/background/retry_runner.py`
- `server/domain/models/__init__.py`
- `frontend/src/pages/PushPage.vue`
- `docs/design/plans/2026-06-02-push-refactor.md`
## 迁移 / 重启
- 需重启 `nexus` Supervisor 进程以加载后端改动。
- 前端需重新 `npm run build` 并部署 `web/app/`
## 验证方式
```bash
ruff check server/infrastructure/database/sync_log_repo.py server/application/services/sync_engine_v2.py server/background/retry_runner.py
cd frontend && npm run build-only
```
- 推送多台服务器:进度列表应出现「同步中」→ 成功/失败,失败行显示 `error_message`
- 推送历史新记录 `status` 应为 `success`/`failed`,非长期 `running`
- 预览对话框显示文件数与传输大小。
## 后续批次
- B2:移除冗余 fallback、M8 `get_by_ids`、M9 临时目录清理
- B3:历史表列与 `PushItem` 类型对齐
- B4/B5:体验与 composable 拆分
@@ -0,0 +1,36 @@
# 推送页重构 Batch 2
**日期**: 2026-06-02
## 变更摘要
- 批量推送改为单次 `get_by_ids` 查询,消除 N+1。
- 手动 ZIP 推送全部成功后自动清理 `/tmp/nexus_upload_*` 临时目录(失败/取消保留供重试)。
- `PushItem` 类型与 `/api/servers/logs` 响应对齐;历史表增加模式、错误列,时间用 `started_at`
## 动机
B0/B1 修复功能契约后,B2 处理性能、磁盘与历史展示数据准确性。
## 涉及文件
- `server/infrastructure/database/server_repo.py`
- `server/application/services/sync_engine_v2.py`
- `frontend/src/types/api.ts`
- `frontend/src/pages/PushPage.vue`
## 迁移 / 重启
- 需重启 `nexus` 后端。
- 前端需重新构建部署。
## 验证方式
```bash
ruff check server/infrastructure/database/server_repo.py server/application/services/sync_engine_v2.py
cd frontend && npm run build-only
```
- 多机推送:DB 查询应为 1 次 servers 加载(非每台 1 次)。
- ZIP 全成功推送后 `/tmp/nexus_upload_<id>` 目录应被删除;若有失败机则目录保留。
- 推送历史表显示开始时间、同步模式、错误信息。
@@ -0,0 +1,36 @@
# 推送页重构 Batch 3
**日期**: 2026-06-02
## 变更摘要
- 推送历史支持按状态、同步模式、服务器筛选(对接已有 `/api/servers/logs` 查询参数)。
- 新增 `POST /api/sync/reconcile-stale-logs`,将超时仍为 `running` 的历史记录订正为 `failed`
- 上传 ZIP 后禁用「推送源路径」输入,避免与 `effectiveSourcePath()` 逻辑不一致。
## 动机
B0 修复后仍有历史脏数据;运营需要筛选失败记录并一键清理卡住的 `running` 行。
## 涉及文件
- `server/infrastructure/database/sync_log_repo.py`
- `server/api/sync_v2.py`
- `server/api/schemas.py`
- `frontend/src/pages/PushPage.vue`
## 迁移 / 重启
- 需重启 `nexus` 后端。
- 前端需重新构建部署。
## 验证方式
```bash
ruff check server/infrastructure/database/sync_log_repo.py server/api/sync_v2.py server/api/schemas.py
cd frontend && npm run build-only
```
- 推送历史:切换状态/模式/服务器筛选,表格应刷新。
- 点击「修复卡住记录」:返回订正条数,原 `running` 行变为失败并带说明文案。
- 上传 ZIP 后源路径输入框为禁用状态。
@@ -0,0 +1,31 @@
# 推送页重构 Batch 4
**日期**: 2026-06-02
## 变更摘要
- 服务器卡片在线状态改用 API 返回的 `is_online`(Redis 实时),并支持手动/60s 自动刷新列表。
- 推送预览改为对所选服务器并行 dry-run(最多 12 台,4 路并发),表格展示每台统计。
- 推送历史失败记录增加「诊断」按钮,调用 `POST /api/sync/diagnose` 展示 SSH/磁盘/路径检查结果。
## 动机
B1 修复单台预览与 WS 字段后,运营仍需多机预览对比、失败根因排查,以及准确的在线状态指示。
## 涉及文件
- `frontend/src/pages/PushPage.vue`
## 迁移 / 重启
- 仅前端变更,重新构建部署 `web/app/` 即可。
## 验证方式
```bash
cd frontend && npm run build-only
```
- 选多台服务器点「预览」:应出现每台文件数/传输量表格。
- 历史失败行点「诊断」:应展示 SSH、路径、磁盘项。
- 服务器绿点应与仪表盘在线状态一致;点刷新或等待 60s 会更新。
@@ -0,0 +1,38 @@
# Changelog: 推送页 B5 架构拆分
**日期**: 2026-06-02
## 摘要
将单体 `PushPage.vue`~1080 行)拆分为 composables + 8 个子组件,页面仅保留编排与生命周期,行为与 B4 一致。
## 动机
- 降低推送页维护成本,对齐 Files 页的 `useFilesPage` + `components/files/` 模式
- 为后续独立推送 WS 频道、预览策略等改动提供清晰边界
## 涉及文件
| 路径 | 说明 |
|------|------|
| `frontend/src/pages/PushPage.vue` | 重写为薄模板 |
| `frontend/src/composables/push/*` | 表单、服务器选择、进度 WS、预览、历史/诊断 |
| `frontend/src/components/push/*` | UI 子组件 |
| `docs/design/plans/2026-06-02-push-refactor.md` | B5 标记完成 |
## 迁移 / 重启
- 仅前端:需 `npm run build-only` 并部署 `web/app/` 静态资源
- 后端无变更,无需重启 `nexus`
## 验证
```bash
cd frontend && npm run build-only
```
本地已执行,构建成功(`PushPage-*.js` 约 33 kB gzip)。
## 回滚
`git revert` 本批提交或恢复 `PushPage.vue` 单体版本;删除 `composables/push/``components/push/` 目录。
@@ -0,0 +1,36 @@
# 推送页 B5 后增强 — 实施计划
**日期**: 2026-06-02
**设计**: `docs/design/specs/2026-06-02-push-post-b5-enhancements-design.md`
## 涉及文件
| 文件 | 变更 |
|------|------|
| `server/api/websocket.py` | `/ws/sync``nexus:sync``_dispatch_sync_message` |
| `server/background/upload_staging_cleanup.py` | 新建 |
| `server/main.py` | 注册清理任务、shutdown sync_manager |
| `frontend/src/composables/push/usePushProgress.ts` | WS URL |
| `frontend/src/composables/push/usePushPreview.ts` | 全量预览 |
| `frontend/src/components/push/PushPreviewDialog.vue` | 进度 UI |
| `tests/test_sync_websocket.py` | 新建 |
| `tests/test_upload_staging_cleanup.py` | 新建 |
## 步骤
1. 后端拆分 sync 广播与 Redis 频道
2. 注册 hourly cleanup 后台任务
3. 前端改 WS + 预览 UX
4. `ruff check` + `pytest` + `npm run build-only`
## 回滚
- Revert 上述文件;旧前端仍连 `/ws/alerts` 时推送进度不再更新(需同步回滚前后端)
## 验证
```bash
ruff check server/api/websocket.py server/background/upload_staging_cleanup.py server/main.py
pytest tests/test_sync_websocket.py tests/test_upload_staging_cleanup.py -q
cd frontend && npm run build-only
```
@@ -0,0 +1,45 @@
# 推送页重构 — 分批实施计划
**日期**: 2026-06-02
**设计**: `docs/design/specs/2026-06-02-push-refactor-design.md` (v2.0)
## 分批总览
| 批次 | 名称 | 范围 | 状态 |
|------|------|------|------|
| **B0** | 后端热修复 | S1/S2/S3/S4、C8 running 广播 | ✅ 已完成 |
| **B1** | 前端 Critical | C1C6、C7、C9、C10 | ✅ 已完成 |
| **B2** | 性能与数据 | M8/M9、M6、`PushItem`、历史表列 | ✅ 已完成 |
| **B3** | 历史筛选、reconcile API、ZIP 源路径禁用 | ✅ 已完成 |
| **B4** | 体验增强 | 多机预览、诊断、is_online、列表刷新 | ✅ 已完成 |
| **B5** | 架构拆分 | composables + 子组件 | ✅ 已完成 |
| **B6** | M2 `/ws/sync` + M9 清理 + 全量预览 | 见 post-b5 设计 | ✅ 已完成 |
## B0 — 涉及文件
- `server/infrastructure/database/sync_log_repo.py` — 新增 `update()`
- `server/application/services/sync_engine_v2.py``--stats`、解析统计、running 广播
- `server/background/retry_runner.py` — 成功判定逻辑
- `server/domain/models/__init__.py``max_retries` 默认 3
## B1 — 涉及文件
- `frontend/src/pages/PushPage.vue`
## B5 — 涉及文件
- `frontend/src/pages/PushPage.vue` — 薄编排层(~100 行)
- `frontend/src/composables/push/``types.ts``labels.ts``usePushForm``usePushServers``usePushProgress``usePushPreview``usePushLogs``usePushPage`
- `frontend/src/components/push/``PushZipUpload``PushToolbar``PushServerGrid``PushSyncModeCard``PushProgressList``PushPreviewDialog``PushDiagnoseDialog``PushHistoryTable`
## 验证
```bash
cd frontend && npm run build-only
ruff check server/infrastructure/database/sync_log_repo.py server/application/services/sync_engine_v2.py server/background/retry_runner.py
python -c "from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl; assert hasattr(SyncLogRepositoryImpl, 'update')"
```
## 回滚
- B0/B1 均为向后兼容改动,回滚对应文件 git revert 即可;无 DB 迁移(`max_retries` 仅影响新行默认值)。
@@ -0,0 +1,48 @@
# 推送页 B5 后增强 — 设计文档
**日期**: 2026-06-02
**依赖**: `2026-06-02-push-refactor-design.md`B0B5 已完成)
## 背景
B5 完成架构拆分后,设计文档中仍有三类未闭环项:
| 项 | 问题 |
|----|------|
| **M2** | 推送进度走 `/ws/alerts`,与告警共用频道,多用户/多批次时客户端需过滤 `batch_id` |
| **M9** | 失败/放弃的上传目录 `/tmp/nexus_upload_*` 无定期清理,仅全成功手动推送后删除 |
| **预览** | B4 限制最多预览 12 台,大批量推送前信息不完整 |
## 方案
### M2 — 独立 `/ws/sync` 频道
- 新增 WebSocket`GET /ws/sync?token=<JWT>`,认证与 `/ws/alerts` 相同
- 新增 Redis 频道 `nexus:sync``broadcast_sync_progress` 仅发往该频道 + `sync_manager` 本地连接
- 告警/恢复/系统事件仍走 `nexus:alerts`,互不影响
- 前端 `usePushProgress` 改为连接 `/ws/sync`
### M9 — 后台清理任务
- `upload_staging_cleanup_loop`primary worker 每小时执行
- 删除 `mtime` 超过 24h 且路径经 `realpath` 校验仍在 `UPLOAD_STAGING_PREFIX` 下的目录
-`sync_engine_v2._maybe_cleanup_upload_staging`(全成功即时删)互补
### 预览 — 全量并发预览
- 取消 12 台上限;对全部选中服务器按并发度 4 分批请求 `/sync/preview`
- 对话框显示进度条与 `已完成/总数`;超过 30 台显示耗时提示
## 安全与性能
- `/ws/sync` 仍须 JWT`/ws/` 已在 HTTP 中间件 public 列表(与 alerts 一致)
- 预览并发固定为 4,避免打满 SSH/rsync
- 清理任务仅删本机 `/tmp/nexus_upload_*``realpath` 防路径逃逸
## 验收标准
- [ ] 推送页 WS 连接 `/ws/sync`,进度与取消正常
- [ ] Dashboard 告警 WS 仍连 `/ws/alerts`,不受推送流量干扰
- [ ] 多机(>12)预览表格逐行填充,进度条正确
- [ ] 超过 24h 的 staging 目录在清理周期后被删除
- [ ] `pytest tests/test_sync_websocket.py tests/test_upload_staging_cleanup.py` 通过
@@ -0,0 +1,392 @@
# 推送页重构设计文档
**日期**: 2026-06-02
**版本**: v2.0(深度审计补充)
---
## 1. 已发现问题(逐行审计)
### 🔴 Critical(功能性 Bug
| # | 问题 | 位置 | 现象 |
|---|------|------|------|
| C1 | **双进度条同时显示** | L130172 | `v-if="pushing"` 的进度条 + WS 进度列表**同时渲染**,信息重复且相互矛盾 |
| C2 | **`mdi-loading` 不旋转** | L162 | `running` 状态图标使用 `mdi-loading`,该图标是静态的,无法表达"进行中"。正确写法是 `mdi-loading mdi-spin``mdi-progress-clock` |
| C3 | **服务器卡片状态从不更新(WS 路径)** | L100102 | WS 推送时 `pushStatus[s.id]` 永远不被赋值(只有 fallback 路径才写),服务器卡片上的 chip 始终为空 |
| C4 | **`selectAll` computed setter 为空** | L430433 | `v-model="selectAll"` 绑定的 computed 的 `set``() => {}`Vue 内部赋值被吞掉,checkbox 双向绑定失效,UI 和状态不同步 |
| C5 | **取消推送把状态标为 `'failed'`** | L395399 | `cancelPush` 将 pending/running 的 items 标记为 `status = 'failed'`,混淆了"失败"和"已取消",历史图表/统计数据被污染 |
| C6 | **`toggleAll` 忽略搜索过滤** | L482488 | 搜索时 `toggleAll` 操作的是 `servers.value`(全部),而非 `filteredServers`,导致搜索过滤后「全选」会选中不可见的服务器 |
### 🔴 深度审计追加 CriticalFrontend↔Backend 契约断裂)
| # | 问题 | 位置 | 现象 |
|---|------|------|------|
| C7 | **WS 字段名不匹配:`error_message` vs `detail`** | `PushPage.vue` L356/359/363 vs `websocket.py` L492 | 后端广播 `error_message`,前端读取 `msg.detail`**错误信息从不显示** |
| C8 | **后端从不广播 `running` 状态** | `sync_engine_v2.py` L198214 | `broadcast_sync_progress` 只在任务完成后调用,pending → 直接跳 success/failedloading 动画永远不激活 |
| C9 | **预览 API 响应结构完全错配** | `PushPage.vue` L182 vs `sync_engine_v2.py` L314325 | 后端返回 `{ stats: { files_total, ... }, files: string[] }`,前端期望 `{ files: number, size: string }`**预览对话框内容永远为空** |
| C10 | **`doPush` 无重入保护** | `PushPage.vue` L509 | 无 `if (pushing.value) return` 快速失败,快速双击可并发触发两次推送 |
### 🟡 Medium(架构 / 可维护性问题)
| # | 问题 | 位置 |
|---|------|------|
| M1 | **WS URL 构造重复实现** | L328336:内联了 `getWsUrl()`,不使用已有的 `buildWebSocketUrl` 工具,逻辑不一致 |
| M2 | **复用告警 WS`/ws/alerts`)传输推送进度** | L338–377:推送进度通过全局告警频道过滤 `batch_id`,理论上两个用户同时推送会串扰;WS 断开后 pending 状态永远不更新 |
| M3 | **预览只对第一台服务器执行** | L499`selectedIds.value.values().next().value`,多选时只预览第一台,误导性强 |
| M4 | **`_pushTimers` 混用 setTimeout/setInterval** | L261/L588:类型声明 `setTimeout[]`,实际 push 了 `setInterval` 返回值,`onUnmounted``clearTimeout` 无法清理 interval,导致定时器泄漏 |
| M5 | **`wsSocket` 是模块级 `let` 变量** | L326:组件多次挂载/卸载不会重置,可能引用旧 socket |
| M6 | **`PushItem` 类型定义与 API 响应脱节** | `types/api.ts` L176185:缺少 `sync_mode``error_message``duration_seconds``operator``trigger_type` 字段,后端实际都返回了 |
| M7 | **`doPush` fallback 路径逻辑矛盾** | L539–569:批量 API 只在网络层失败时触发 fallback,而此 fallback 又重新调相同的 `/sync/files` 但没接 WS,进度追踪完全失效;实际上此 fallback 几乎永远不会触发 |
| M8 | **后端 N+1 DB 查询** | `sync_engine_v2.py` L7578`for sid in server_ids: await repo.get_by_id(sid)`100 台服务器 = 100 次 DB 往返 |
| M9 | **无 `/tmp/nexus_upload_*` 清理机制** | `sync_v2.py` L949:ZIP 解压后临时目录从不自动删除,服务器 `/tmp` 会随时间积累 |
| M10 | **`setInterval` 完成检测可由 WS `onmessage` 替代** | L573587:用 500ms polling 检测推送完成,完全可以在 WS 消息里直接判断,消除不必要的定时器 |
| M11 | **单文件 633 行,无 composable/组件拆分** | 整个页面逻辑扁平 |
### 🟢 Minor(体验问题)
| # | 问题 |
|---|------|
| E1 | 目标路径为空时可以静默推送(无校验) |
| E2 | 推送历史表格缺少 `sync_mode``error_message``duration_seconds` 列 |
| E3 | WS 意外断线时(非 5min 超时),pending 项目无任何提示 |
| E4 | 取消推送后进度卡片仍可能显示取消按钮(条件判断未包含 `cancelled` |
| E5 | 上传 ZIP 后 `sourcePath` 输入框仍可编辑,但 `effectiveSourcePath()` 会忽略它,UI 与逻辑不一致 |
| E6 | 服务器在线状态显示 DB 静态值,非 Redis 实时心跳状态 |
---
## 2. 修复方案
### C1 — 合并两个进度区域
删除旧的 `v-if="pushing"` 进度条(含 `pushProgress` + `completedCount`)。
所有进度统一由 `wsProgressItems` 驱动,始终显示 per-server 粒度进度。
总进度 computed 由 `wsProgressItems` 派生:
```typescript
const doneCount = computed(() => wsProgressItems.value.filter(p => p.status === 'success' || p.status === 'failed' || p.status === 'cancelled').length)
const totalProgress = computed(() => wsProgressItems.value.length ? Math.round(doneCount.value / wsProgressItems.value.length * 100) : 0)
```
### C2 — 修复旋转图标
```html
<!-- 修复前 -->
<v-icon v-else-if="s.status === 'running'" color="blue">mdi-loading</v-icon>
<!-- 修复后 -->
<v-icon v-else-if="s.status === 'running'" color="blue" class="mdi-spin">mdi-loading</v-icon>
```
### C3 — WS 推送时同步更新服务器卡片
在 WS `onmessage` 处理中,同步写入 `pushStatus`
```typescript
if (msg.status === 'success') {
item.status = 'success'
pushStatus.value[msg.server_id] = '成功'
} else if (msg.status === 'failed') {
item.status = 'failed'
pushStatus.value[msg.server_id] = '失败'
}
```
### C4 — 修复 selectAll 双向绑定
`v-model="selectAll"` 改为 `:model-value="selectAll"` + `@update:model-value="toggleAll"`
同时修复 `selectAll` computed 使其只检查 `filteredServers`(与搜索对齐):
```typescript
const selectAll = computed(() =>
filteredServers.value.length > 0 &&
filteredServers.value.every(s => selectedIds.value.has(s.id))
)
```
### C5 — 新增 `cancelled` 状态
```typescript
type ServerProgressStatus = 'pending' | 'running' | 'success' | 'failed' | 'cancelled'
// cancelPush 中:
item.status = 'cancelled'
item.detail = '已取消'
```
模板增加对应图标和颜色:
```html
<v-icon v-else-if="s.status === 'cancelled'" color="grey">mdi-cancel</v-icon>
```
### C6 — toggleAll 只操作 filteredServers
```typescript
function toggleAll() {
const visible = filteredServers.value
const allSelected = visible.every(s => selectedIds.value.has(s.id))
const s = new Set(selectedIds.value)
for (const srv of visible) {
if (allSelected) s.delete(srv.id)
else s.add(srv.id)
}
selectedIds.value = s
}
```
### M1 — 统一 WS URL 构造
删除 `getWsUrl()`,改用 `buildWebSocketUrl`
```typescript
import { buildWebSocketUrl } from '@/utils/wsUrl'
function connectProgressWs() {
if (!auth.token) return
const url = buildWebSocketUrl('/ws/alerts', { token: auth.token })
// ...
}
```
### C7 — 修复 WS 字段名映射
```typescript
// onmessage 中统一用 error_message,不用 detail
if (msg.status === 'running') {
item.status = 'running'
item.detail = '同步中...'
} else if (msg.status === 'success') {
item.status = 'success'
item.detail = `完成${msg.duration_seconds ? `${msg.duration_seconds}s` : ''}`
pushStatus.value[msg.server_id] = '成功'
} else if (msg.status === 'failed') {
item.status = 'failed'
item.detail = msg.error_message || '失败' // ← 修复:error_message not detail
pushStatus.value[msg.server_id] = '失败'
} else if (msg.status === 'cancelled') {
item.status = 'cancelled'
item.detail = '已取消'
}
```
### C8 — 后端补发 `running` 广播
`sync_engine_v2.py``_sync_one` 创建 SyncLog 后立即广播 `running`
```python
# 在 sync_log = await self.sync_log_repo.create(sync_log) 之后
try:
await broadcast_sync_progress(
batch_id=batch_id,
server_id=server.id,
server_name=server.name,
status="running",
completed=completed,
failed=failed,
total=total,
)
except Exception as e:
logger.debug(f"WS running broadcast failed for server {server.id}: {e}")
```
### C9 — 修复预览对话框数据读取
```typescript
// PreviewData 类型对齐后端响应
interface PreviewData {
error?: string
stats?: {
files_total?: number
files_created?: number
files_deleted?: number
files_transferred?: number
total_size_bytes?: number
transfer_size_bytes?: number
}
files?: string[]
files_truncated?: boolean
server_name?: string
}
```
模板改为读取 `previewData.stats`
```html
<div v-if="previewData.stats" class="text-body-2">
{{ previewData.stats.files_total ?? 0 }} 个文件,
将传输 {{ previewData.stats.files_transferred ?? 0 }} 个,
{{ formatSize(previewData.stats.transfer_size_bytes ?? 0) }}
</div>
```
### C10 — doPush 重入保护
```typescript
async function doPush() {
if (pushing.value) return // ← 新增
if (!validateForm()) return
// ...
}
```
### M4 — 分离 timers
```typescript
const _pushIntervals: ReturnType<typeof setInterval>[] = []
const _pushTimeouts: ReturnType<typeof setTimeout>[] = []
// onUnmounted:
_pushIntervals.forEach(t => clearInterval(t))
_pushTimeouts.forEach(t => clearTimeout(t))
```
### M5 — wsSocket 改为 ref
```typescript
const wsSocket = ref<WebSocket | null>(null)
```
### E1 — 目标路径验证
```typescript
function validateForm(): boolean {
if (!effectiveSourcePath()) {
snackbar('请选择源路径或上传 ZIP 文件', 'warning')
return false
}
if (!targetPath.value.trim()) {
snackbar('请填写目标路径', 'warning')
return false
}
if (selectedIds.value.size === 0) {
snackbar('请选择至少一台服务器', 'warning')
return false
}
return true
}
```
### M8 — 后端批量查询替代 N+1
```python
# sync_engine_v2.py
servers = await self.server_repo.get_by_ids(server_ids) # 单次查询
```
需在 `ServerRepositoryImpl` 增加 `get_by_ids(ids: list[int])` 方法:
```python
async def get_by_ids(self, ids: list[int]) -> list[Server]:
from sqlalchemy import select
result = await self.session.execute(
select(Server).where(Server.id.in_(ids))
)
return list(result.scalars().all())
```
### M9 — 临时目录定期清理
`background/` 添加清理任务,或在 ZIP 推送完成后主动删除:
```python
# sync_engine_v2.py sync_files() 执行完后
if source_path.startswith("/tmp/nexus_upload_"):
import shutil
shutil.rmtree(source_path, ignore_errors=True)
logger.info(f"Cleaned up upload temp dir: {source_path}")
```
### M10 — 消除 setInterval,改为 onmessage 驱动完成检测
```typescript
// onmessage 中调用
function _checkPushComplete() {
const total = wsProgressItems.value.length
const done = wsProgressItems.value.filter(
p => p.status === 'success' || p.status === 'failed' || p.status === 'cancelled'
).length
if (total > 0 && done >= total) {
pushing.value = false
disconnectProgressWs()
loadLogs()
const failed = wsProgressItems.value.filter(p => p.status === 'failed').length
const cancelled = wsProgressItems.value.filter(p => p.status === 'cancelled').length
snackbar(`推送完成:${done - failed - cancelled} 成功,${failed} 失败,${cancelled} 取消`)
}
}
```
### E3 — WS 意外断线兜底
```typescript
socket.onclose = (ev) => {
wsSocket.value = null
// 非正常关闭(code !== 1000)且仍有 pending/running 项目时提示
if (ev.code !== 1000) {
wsProgressItems.value.forEach(p => {
if (p.status === 'pending' || p.status === 'running') {
p.status = 'failed'
p.detail = '连接中断'
}
})
snackbar('推送进度连接中断,部分服务器状态未知', 'warning')
pushing.value = false
loadLogs()
}
}
```
---
## 3. 拆分规划(组件/composable
```
frontend/src/
├── pages/
│ └── PushPage.vue (~120行,仅组合)
├── composables/push/
│ ├── usePushSessions.ts (WS 连接、进度追踪、取消)
│ ├── usePushForm.ts (源路径、目标路径、同步模式、校验)
│ └── usePushLogs.ts (历史记录分页加载)
└── components/push/
├── PushServerGrid.vue (服务器分组网格 + 搜索 + 全选)
├── PushProgressList.vue (进度列表 + 取消按钮)
└── PushHistoryTable.vue (历史记录表格)
```
---
## 4. 实施阶段
| 阶段 | 范围 | 涉及文件 | 优先级 |
|------|------|---------|--------|
| **R1** | C1C6:前端 Critical Bug | `PushPage.vue` | 最高 |
| **R2** | C7C10 + M1/M4/M5/M10 | `PushPage.vue``websocket.py``sync_engine_v2.py` | 高(含 Backend |
| **R3** | M6/M8/M9 + `PushItem` 类型补全 + 临时目录清理 | `types/api.ts``server_repo.py``sync_engine_v2.py` | 中 |
| **R4** | E1–E6 体验细节 + 推送历史表补列 | `PushPage.vue` | 中 |
| **R5** | 组件/composable 拆分(M11/M7 | 新建 `composables/push/``components/push/` | 低(架构优化) |
---
## 5. 验收标准
**功能正确性**
- [ ] 推送过程中只显示一个进度区域
- [ ] `running` 状态显示旋转动画(后端广播 + 前端渲染)
- [ ] 进度列表中失败服务器显示具体错误信息
- [ ] 预览对话框正确显示文件数量和传输大小
- [ ] 推送时服务器卡片 chip 实时更新(WS 路径)
- [ ] 取消推送后状态为「已取消」而非「失败」
**交互正确性**
- [ ] 全选 checkbox 与 Vue 双向绑定正常工作
- [ ] 搜索后全选仅选中可见服务器
- [ ] 快速双击推送按钮不触发重复推送
- [ ] 目标路径为空时阻止推送并提示
**资源管理**
- [ ] WS 意外断线时标记未完成项目并提示用户
- [ ] `onUnmounted` 正确清理所有 interval 和 timeout(无 setInterval 泄漏)
- [ ] ZIP 推送完成后临时目录自动清理
**性能**
- [ ] 批量推送 DB 查询从 N 次降到 1 次
@@ -0,0 +1,70 @@
<template>
<v-dialog :model-value="modelValue" max-width="520" @update:model-value="$emit('update:modelValue', $event)">
<v-card border>
<v-card-title>推送失败诊断 {{ targetName }}</v-card-title>
<v-card-text>
<v-progress-linear v-if="loading" indeterminate class="mb-3" />
<template v-else-if="result">
<v-list density="compact">
<v-list-item title="SSH 连接">
<template #append>
<v-icon :color="result.ssh_ok ? 'success' : 'error'">
{{ result.ssh_ok ? 'mdi-check' : 'mdi-close' }}
</v-icon>
</template>
</v-list-item>
<v-list-item v-if="result.disk_avail" :title="`磁盘可用: ${result.disk_avail}`" />
<v-list-item
v-if="result.disk_used_pct != null"
:title="`磁盘使用率: ${result.disk_used_pct}%`"
/>
<v-list-item title="目标路径存在">
<template #append>
<v-icon :color="result.path_exists ? 'success' : 'error'">
{{ result.path_exists ? 'mdi-check' : 'mdi-close' }}
</v-icon>
</template>
</v-list-item>
<v-list-item v-if="result.path_perms" :subtitle="result.path_perms" title="权限" />
<v-list-item title="目标路径可写">
<template #append>
<v-icon :color="result.path_writable ? 'success' : 'error'">
{{ result.path_writable ? 'mdi-check' : 'mdi-close' }}
</v-icon>
</template>
</v-list-item>
</v-list>
<v-alert
v-for="(err, i) in result.errors"
:key="i"
type="error"
density="compact"
variant="tonal"
class="mt-2"
>
{{ err }}
</v-alert>
</template>
</v-card-text>
<v-card-actions>
<v-spacer />
<v-btn variant="text" @click="$emit('update:modelValue', false)">关闭</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
</template>
<script setup lang="ts">
import type { DiagnoseResult } from '@/composables/push/types'
defineProps<{
modelValue: boolean
loading: boolean
result: DiagnoseResult | null
targetName?: string
}>()
defineEmits<{
'update:modelValue': [value: boolean]
}>()
</script>
@@ -0,0 +1,133 @@
<template>
<v-card elevation="0" border rounded="lg">
<v-card-title class="d-flex flex-wrap align-center ga-2">
<span>推送历史</span>
<v-spacer />
<v-select
:model-value="statusFilter"
:items="statusOptions"
item-title="label"
item-value="value"
label="状态"
variant="outlined"
density="compact"
hide-details
clearable
style="max-width: 120px"
@update:model-value="$emit('update:statusFilter', $event); $emit('filter-change')"
/>
<v-select
:model-value="syncModeFilter"
:items="syncModeOptions"
item-title="label"
item-value="value"
label="模式"
variant="outlined"
density="compact"
hide-details
clearable
style="max-width: 120px"
@update:model-value="$emit('update:syncModeFilter', $event); $emit('filter-change')"
/>
<v-select
:model-value="serverFilter"
:items="serverOptions"
item-title="title"
item-value="value"
label="服务器"
variant="outlined"
density="compact"
hide-details
clearable
style="max-width: 180px"
@update:model-value="$emit('update:serverFilter', $event); $emit('filter-change')"
/>
<v-btn
size="small"
variant="tonal"
prepend-icon="mdi-wrench"
:loading="reconciling"
@click="$emit('reconcile')"
>
修复卡住记录
</v-btn>
</v-card-title>
<v-data-table-server
:items="items"
:headers="headers"
:items-length="total"
:loading="loading"
:page="page"
:items-per-page="15"
hover
density="comfortable"
@update:page="$emit('update:page', $event); $emit('page-change')"
>
<template #item.status="{ item }">
<v-chip :color="logStatusColor(item.status)" size="x-small" variant="tonal" label border="sm">
{{ logStatusLabel(item.status) }}
</v-chip>
</template>
<template #item.sync_mode="{ item }">
<span class="text-caption">{{ syncModeLabel(item.sync_mode) }}</span>
</template>
<template #item.started_at="{ item }">
<span class="text-caption text-medium-emphasis">{{ formatLogTime(item.started_at) }}</span>
</template>
<template #item.error_message="{ item }">
<span
v-if="item.error_message"
class="text-caption text-error text-truncate d-inline-block"
style="max-width: 200px"
:title="item.error_message"
>{{ item.error_message }}</span>
<span v-else class="text-caption text-medium-emphasis"></span>
</template>
<template #item.actions="{ item }">
<v-btn
v-if="item.status === 'failed'"
size="x-small"
variant="tonal"
prepend-icon="mdi-stethoscope"
@click="$emit('diagnose', item)"
>
诊断
</v-btn>
</template>
<template #no-data>
<div class="text-center text-medium-emphasis py-6">暂无数据</div>
</template>
</v-data-table-server>
</v-card>
</template>
<script setup lang="ts">
import { formatLogTime, logStatusColor, logStatusLabel, syncModeLabel } from '@/composables/push/labels'
import type { PushItem } from '@/types/api'
defineProps<{
items: PushItem[]
headers: { title: string; key: string; width?: number; sortable?: boolean }[]
total: number
loading: boolean
page: number
statusFilter: string | null
syncModeFilter: string | null
serverFilter: number | null
statusOptions: { label: string; value: string }[]
syncModeOptions: { label: string; value: string }[]
serverOptions: { title: string; value: number }[]
reconciling: boolean
}>()
defineEmits<{
'update:page': [page: number]
'update:statusFilter': [value: string | null]
'update:syncModeFilter': [value: string | null]
'update:serverFilter': [value: number | null]
'filter-change': []
'page-change': []
reconcile: []
diagnose: [item: PushItem]
}>()
</script>
@@ -0,0 +1,91 @@
<template>
<v-dialog :model-value="modelValue" max-width="800" @update:model-value="$emit('update:modelValue', $event)">
<v-card border>
<v-card-title class="d-flex align-center">
推送预览
<v-spacer />
<span class="text-caption text-medium-emphasis">
<template v-if="loading">{{ previewDone }} / {{ previewTotal }} </template>
<template v-else>{{ results.length }} </template>
</span>
</v-card-title>
<v-card-text>
<v-progress-linear
v-if="loading"
:model-value="previewTotal ? Math.round((previewDone / previewTotal) * 100) : 0"
class="mb-3"
/>
<v-alert
v-if="largeBatch && !loading"
type="info"
density="compact"
variant="tonal"
class="mb-2"
>
已预览 {{ results.length }} 台服务器超过 {{ warnThreshold }} 台时推送耗时可能较长
</v-alert>
<div v-if="!loading" class="mb-2 text-body-2">
将推送到 <strong>{{ selectedCount }}</strong> 台服务器
</div>
<v-table v-if="results.length" density="compact">
<thead>
<tr>
<th>服务器</th>
<th>文件数</th>
<th>预计传输</th>
<th>大小</th>
<th>备注</th>
</tr>
</thead>
<tbody>
<tr v-for="row in results" :key="row.server_id">
<td>{{ row.server_name }}</td>
<td>{{ row.stats?.files_total ?? '—' }}</td>
<td>{{ row.stats?.files_transferred ?? '—' }}</td>
<td>{{ row.stats ? formatSize(row.stats.transfer_size_bytes ?? 0) : '—' }}</td>
<td>
<span v-if="row.error" class="text-error text-caption">{{ row.error }}</span>
<span
v-else-if="syncMode === 'full' && (row.stats?.files_deleted ?? 0) > 0"
class="text-warning text-caption"
>将删除 {{ row.stats?.files_deleted }} 个远端文件</span>
<span v-else class="text-caption text-medium-emphasis"></span>
</td>
</tr>
</tbody>
</v-table>
<v-alert v-if="errorCount > 0 && !loading" type="warning" density="compact" variant="tonal" class="mt-3">
{{ errorCount }} 台预览失败推送时这些服务器可能仍会失败
</v-alert>
</v-card-text>
<v-card-actions>
<v-spacer />
<v-btn variant="text" @click="$emit('update:modelValue', false)">关闭</v-btn>
<v-btn color="primary" variant="flat" :disabled="loading" @click="$emit('confirm-push')">确认推送</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
</template>
<script setup lang="ts">
import { formatSize } from '@/composables/push/labels'
import type { PreviewServerRow, SyncMode } from '@/composables/push/types'
defineProps<{
modelValue: boolean
loading: boolean
results: PreviewServerRow[]
selectedCount: number
previewDone: number
previewTotal: number
largeBatch: boolean
warnThreshold: number
errorCount: number
syncMode: SyncMode
}>()
defineEmits<{
'update:modelValue': [value: boolean]
'confirm-push': []
}>()
</script>
@@ -0,0 +1,69 @@
<template>
<v-card v-if="items.length > 0" class="mb-4" elevation="0" border rounded="lg">
<v-card-title class="d-flex align-center text-subtitle-1">
推送进度
<v-spacer />
<v-btn
v-if="batchId && items.some(s => s.status === 'running' || s.status === 'pending') && pushing"
size="small"
variant="tonal"
color="error"
class="mr-2"
:loading="cancelling"
@click="$emit('cancel')"
>
取消推送
</v-btn>
<v-chip size="small" variant="tonal" color="primary">
{{ items.filter(s => s.status === 'success').length }}/{{ items.length }} 完成
</v-chip>
</v-card-title>
<v-divider />
<v-card-text>
<v-progress-linear
v-if="pushing || totalProgress > 0"
:model-value="totalProgress"
color="primary"
height="8"
rounded
class="mb-3"
/>
<div v-if="items.length" class="text-caption text-medium-emphasis mb-2">
{{ doneCount }} / {{ items.length }} 台已结束
</div>
<v-list density="compact">
<v-list-item v-for="s in items" :key="s.id" :title="s.name">
<template #append>
<v-icon v-if="s.status === 'pending'" color="grey">mdi-clock-outline</v-icon>
<v-icon v-else-if="s.status === 'running'" color="blue" class="mdi-spin">mdi-loading</v-icon>
<v-icon v-else-if="s.status === 'success'" color="green">mdi-check-circle</v-icon>
<v-icon v-else-if="s.status === 'cancelled'" color="grey">mdi-cancel</v-icon>
<v-icon v-else color="red">mdi-alert-circle</v-icon>
</template>
<template v-if="s.detail" #subtitle>
<span
:class="s.status === 'failed' ? 'text-error' : 'text-medium-emphasis'"
>{{ s.detail }}</span>
</template>
</v-list-item>
</v-list>
</v-card-text>
</v-card>
</template>
<script setup lang="ts">
import type { ServerProgress } from '@/composables/push/types'
defineProps<{
items: ServerProgress[]
batchId: string
pushing: boolean
cancelling: boolean
doneCount: number
totalProgress: number
}>()
defineEmits<{
cancel: []
}>()
</script>
@@ -0,0 +1,110 @@
<template>
<v-card elevation="0" border rounded="lg" class="mb-4">
<v-card-title class="d-flex align-center">
<v-checkbox
:model-value="selectAll"
label="全选"
density="compact"
hide-details
class="mr-4"
@update:model-value="$emit('toggle-all')"
/>
<v-text-field
:model-value="serverSearch"
prepend-inner-icon="mdi-magnify"
label="搜索服务器..."
density="compact"
hide-details
variant="outlined"
rounded
style="max-width: 240px"
@update:model-value="$emit('update:serverSearch', $event)"
/>
<v-btn
icon="mdi-refresh"
variant="text"
size="small"
:loading="serversLoading"
title="刷新在线状态"
@click="$emit('refresh')"
/>
</v-card-title>
<v-divider />
<v-card-text class="pa-2">
<template v-if="serversByCategory.length === 0">
<div class="text-center text-medium-emphasis py-6">暂无服务器</div>
</template>
<template v-else>
<div v-for="group in serversByCategory" :key="group.category" class="mb-4">
<div class="d-flex align-center px-2 mb-2">
<v-btn
size="x-small"
variant="text"
:color="isCategoryAllSelected(group.category) ? 'primary' : 'default'"
@click="$emit('toggle-category', group.category)"
>
<v-icon size="16" class="mr-1">
{{ isCategoryAllSelected(group.category) ? 'mdi-checkbox-marked' : 'mdi-checkbox-blank-outline' }}
</v-icon>
{{ group.category }}
</v-btn>
<span class="text-caption text-medium-emphasis ml-1">
({{ group.servers.filter(s => selectedIds.has(s.id)).length }}/{{ group.servers.length }})
</span>
</div>
<v-row dense>
<v-col v-for="s in group.servers" :key="s.id" cols="6" sm="4" md="3" lg="2">
<v-card
:color="selectedIds.has(s.id) ? 'primary' : undefined"
:variant="selectedIds.has(s.id) ? 'tonal' : 'outlined'"
rounded="lg"
class="pa-3 cursor-pointer"
@click="$emit('toggle-server', s.id)"
>
<div class="d-flex align-center ga-2">
<v-icon :color="serverOnlineColor(s.is_online)" size="12">mdi-circle</v-icon>
<span class="text-body-2 font-weight-medium text-truncate">{{ s.name }}</span>
</div>
<div class="text-caption text-medium-emphasis text-truncate">{{ s.domain }}</div>
<div v-if="pushStatus[s.id]" class="text-caption mt-1">
<v-chip
:color="pushStatusChipColor(pushStatus[s.id])"
size="x-small"
variant="tonal"
label
border="sm"
>
{{ pushStatus[s.id] }}
</v-chip>
</div>
</v-card>
</v-col>
</v-row>
</div>
</template>
</v-card-text>
</v-card>
</template>
<script setup lang="ts">
import { pushStatusChipColor, serverOnlineColor } from '@/composables/push/labels'
import type { PushServerItem } from '@/composables/push/types'
defineProps<{
selectAll: boolean
serverSearch: string
serversLoading: boolean
serversByCategory: { category: string; servers: PushServerItem[] }[]
selectedIds: Set<number>
pushStatus: Record<number, string>
isCategoryAllSelected: (category: string) => boolean
}>()
defineEmits<{
'update:serverSearch': [value: string]
'toggle-all': []
refresh: []
'toggle-category': [category: string]
'toggle-server': [id: number]
}>()
</script>
@@ -0,0 +1,29 @@
<template>
<v-card elevation="0" border rounded="lg" class="mb-4">
<v-card-title class="text-subtitle-1">同步模式</v-card-title>
<v-card-text>
<v-radio-group :model-value="syncMode" inline density="compact" @update:model-value="$emit('update:syncMode', $event)">
<v-radio label="增量同步" value="incremental" />
<v-radio label="全量同步" value="full" />
<v-radio label="校验和同步" value="checksum" />
</v-radio-group>
<v-alert v-if="syncMode === 'full'" type="warning" density="compact" variant="tonal" class="mt-2">
全量同步会删除目标目录中不存在于源目录的文件
</v-alert>
<div class="text-caption text-medium-emphasis mt-1">{{ syncModeDesc }}</div>
</v-card-text>
</v-card>
</template>
<script setup lang="ts">
import type { SyncMode } from '@/composables/push/types'
defineProps<{
syncMode: SyncMode
syncModeDesc: string
}>()
defineEmits<{
'update:syncMode': [value: SyncMode]
}>()
</script>
@@ -0,0 +1,58 @@
<template>
<v-card elevation="0" border rounded="lg" class="mb-4 pa-4">
<v-row align="center" dense>
<v-col cols="12" sm="4">
<v-text-field
:model-value="sourcePath"
label="推送源路径"
variant="outlined"
density="compact"
prepend-inner-icon="mdi-folder"
:disabled="zipLocked"
:hint="zipLocked ? '已使用 ZIP 解压目录作为源路径' : undefined"
persistent-hint
@update:model-value="$emit('update:sourcePath', $event)"
/>
</v-col>
<v-col cols="12" sm="4">
<v-text-field
:model-value="targetPath"
label="目标路径"
variant="outlined"
density="compact"
prepend-inner-icon="mdi-folder-marker"
@update:model-value="$emit('update:targetPath', $event)"
/>
</v-col>
<v-col cols="12" sm="4" class="d-flex ga-2 justify-end">
<v-btn size="small" variant="tonal" prepend-icon="mdi-eye" @click="$emit('preview')">预览</v-btn>
<v-btn
size="small"
color="primary"
variant="flat"
prepend-icon="mdi-upload"
:loading="pushing"
@click="$emit('push')"
>
推送
</v-btn>
</v-col>
</v-row>
</v-card>
</template>
<script setup lang="ts">
defineProps<{
sourcePath: string
targetPath: string
zipLocked: boolean
pushing: boolean
}>()
defineEmits<{
'update:sourcePath': [value: string]
'update:targetPath': [value: string]
preview: []
push: []
}>()
</script>
@@ -0,0 +1,51 @@
<template>
<v-card elevation="0" border rounded="lg" class="mb-4">
<v-card-title class="text-subtitle-1">源文件</v-card-title>
<v-card-text>
<div v-if="!uploadedZip">
<div
class="border-dashed border rounded pa-6 text-center cursor-pointer"
:class="{ 'border-primary bg-primary-lighten-5': isDragging }"
@dragover.prevent="$emit('dragover')"
@dragleave="$emit('dragleave')"
@drop.prevent="$emit('drop', $event)"
@click="zipInputRef?.click()"
>
<v-icon size="48" color="grey">mdi-cloud-upload-outline</v-icon>
<div class="text-body-2 mt-2">拖拽 ZIP 文件到此处或点击选择</div>
<div class="text-caption text-medium-emphasis">支持 .zip 格式</div>
</div>
<input ref="zipInputRef" type="file" accept=".zip" hidden @change="$emit('file-select', $event)" />
</div>
<div v-else class="d-flex align-center">
<v-icon class="mr-2">mdi-folder-zip</v-icon>
<span>{{ uploadedZip.name }} ({{ uploadedZip.fileCount }} 个文件, {{ formatSize(uploadedZip.size) }})</span>
<v-spacer />
<v-btn size="small" variant="text" color="error" @click="$emit('clear')">清除</v-btn>
</div>
<v-progress-linear v-if="zipUploading" indeterminate class="mt-2" />
</v-card-text>
</v-card>
</template>
<script setup lang="ts">
import { ref } from 'vue'
import { formatSize } from '@/composables/push/labels'
import type { UploadedZip } from '@/composables/push/types'
defineProps<{
uploadedZip: UploadedZip | null
isDragging: boolean
zipUploading: boolean
}>()
defineEmits<{
dragover: []
dragleave: []
drop: [e: DragEvent]
'file-select': [e: Event]
clear: []
}>()
const zipInputRef = ref<HTMLInputElement | null>(null)
</script>
+3
View File
@@ -0,0 +1,3 @@
export * from './types'
export * from './labels'
export { usePushPage } from './usePushPage'
+55
View File
@@ -0,0 +1,55 @@
/** Display helpers for push UI */
export function formatSize(bytes: number): string {
if (bytes < 1024) return `${bytes} B`
if (bytes < 1048576) return `${(bytes / 1024).toFixed(1)} KB`
return `${(bytes / 1048576).toFixed(1)} MB`
}
export function formatLogTime(iso: string | null | undefined): string {
if (!iso) return '—'
const d = new Date(iso)
if (Number.isNaN(d.getTime())) return iso
return d.toLocaleString('zh-CN', { hour12: false })
}
export function logStatusLabel(status: string): string {
switch (status) {
case 'success': return '成功'
case 'failed': return '失败'
case 'running': return '进行中'
case 'cancelled': return '已取消'
default: return status
}
}
export function logStatusColor(status: string): string {
switch (status) {
case 'success': return 'success'
case 'failed': return 'error'
case 'cancelled': return 'grey'
case 'running': return 'info'
default: return 'warning'
}
}
export function syncModeLabel(mode: string | undefined): string {
switch (mode) {
case 'incremental': return '增量'
case 'full': return '全量'
case 'checksum': return '校验和'
case 'overwrite': return '覆盖'
default: return mode || '—'
}
}
export function pushStatusChipColor(status: string): string {
if (status === '成功') return 'success'
if (status === '失败') return 'error'
if (status === '已取消') return 'grey'
return 'warning'
}
export function serverOnlineColor(isOnline: boolean | undefined): string {
return isOnline ? 'success' : 'error'
}
+72
View File
@@ -0,0 +1,72 @@
/** Shared types for push page composables and components */
export type SyncMode = 'incremental' | 'full' | 'checksum'
export interface PushServerItem {
id: number
name: string
domain?: string
is_online?: boolean
category?: string
}
export type ServerProgressStatus = 'pending' | 'running' | 'success' | 'failed' | 'cancelled'
export interface ServerProgress {
id: number
name: string
status: ServerProgressStatus
detail?: string
}
export interface SyncProgressMsg {
type: string
batch_id?: string
server_id?: number
status?: string
error_message?: string | null
duration_seconds?: number | null
}
export interface UploadedZip {
name: string
sourcePath: string
fileCount: number
size: number
}
export interface PreviewStats {
files_total?: number
files_created?: number
files_deleted?: number
files_transferred?: number
total_size_bytes?: number
transfer_size_bytes?: number
}
export interface PreviewServerRow {
server_id: number
server_name: string
error?: string
stats?: PreviewStats
}
export interface DiagnoseResult {
server_id: number
server_name: string
ssh_ok: boolean
disk_avail: string | null
disk_used_pct: number | null
path_exists: boolean | null
path_perms: string | null
path_writable: boolean | null
errors: string[]
}
export const TERMINAL_PROGRESS_STATUSES: ServerProgressStatus[] = ['success', 'failed', 'cancelled']
/** Parallel preview requests per batch */
export const PREVIEW_CONCURRENCY = 4
/** Show a warning when previewing more than this many servers */
export const PREVIEW_WARN_THRESHOLD = 30
@@ -0,0 +1,101 @@
import { ref, computed } from 'vue'
import { http } from '@/api'
import { useSnackbar } from '@/composables/useSnackbar'
import type { SyncMode, UploadedZip } from './types'
export function usePushForm() {
const snackbar = useSnackbar()
const sourcePath = ref('')
const targetPath = ref('')
const syncMode = ref<SyncMode>('incremental')
const isDragging = ref(false)
const uploadedZip = ref<UploadedZip | null>(null)
const zipUploading = ref(false)
const syncModeDesc = computed(() => {
switch (syncMode.value) {
case 'incremental':
return '仅传输新增或修改的文件'
case 'full':
return '使目标目录与源目录完全一致,会删除不存在于源的文件'
case 'checksum':
return '通过文件内容校验和判断是否需要传输'
default:
return ''
}
})
function effectiveSourcePath(): string {
return uploadedZip.value?.sourcePath || sourcePath.value
}
function validateForm(selectedCount: number): boolean {
if (!effectiveSourcePath()) {
snackbar('请选择源路径或上传 ZIP 文件', 'warning')
return false
}
if (!targetPath.value.trim()) {
snackbar('请填写目标路径', 'warning')
return false
}
if (selectedCount === 0) {
snackbar('请选择至少一台服务器', 'warning')
return false
}
return true
}
async function handleZipDrop(e: DragEvent) {
isDragging.value = false
const file = e.dataTransfer?.files[0]
if (file?.name.endsWith('.zip')) await uploadZip(file)
}
function handleZipSelect(e: Event) {
const file = (e.target as HTMLInputElement).files?.[0]
if (file) uploadZip(file)
}
async function uploadZip(file: File) {
zipUploading.value = true
try {
const form = new FormData()
form.append('file', file)
const res = await http.upload<{ source_path: string; file_count: number; size: number }>(
'/sync/upload-zip',
form,
)
uploadedZip.value = {
name: file.name,
sourcePath: res.source_path,
fileCount: res.file_count,
size: res.size,
}
snackbar('ZIP 上传成功')
} catch (e: unknown) {
snackbar(e instanceof Error ? e.message : '上传失败', 'error')
} finally {
zipUploading.value = false
}
}
function clearZipUpload() {
uploadedZip.value = null
}
return {
sourcePath,
targetPath,
syncMode,
syncModeDesc,
isDragging,
uploadedZip,
zipUploading,
effectiveSourcePath,
validateForm,
handleZipDrop,
handleZipSelect,
clearZipUpload,
}
}
@@ -0,0 +1,146 @@
import { ref, computed } from 'vue'
import { http } from '@/api'
import { useSnackbar } from '@/composables/useSnackbar'
import type { PaginatedResponse, PushItem } from '@/types/api'
import type { usePushForm } from './usePushForm'
import type { usePushServers } from './usePushServers'
import type { DiagnoseResult } from './types'
type PushForm = ReturnType<typeof usePushForm>
type PushServers = ReturnType<typeof usePushServers>
const LOG_STATUS_OPTIONS = [
{ label: '成功', value: 'success' },
{ label: '失败', value: 'failed' },
{ label: '进行中', value: 'running' },
{ label: '已取消', value: 'cancelled' },
]
const LOG_SYNC_MODE_OPTIONS = [
{ label: '增量', value: 'incremental' },
{ label: '全量', value: 'full' },
{ label: '校验和', value: 'checksum' },
]
const LOG_HEADERS = [
{ title: '时间', key: 'started_at', width: 170 },
{ title: '服务器', key: 'server_name', width: 120 },
{ title: '状态', key: 'status', width: 88 },
{ title: '模式', key: 'sync_mode', width: 88 },
{ title: '源路径', key: 'source_path' },
{ title: '目标路径', key: 'target_path' },
{ title: '错误', key: 'error_message', width: 200 },
{ title: '操作', key: 'actions', width: 88, sortable: false },
]
export function usePushLogs(form: PushForm, servers: PushServers) {
const snackbar = useSnackbar()
const logs = ref<PushItem[]>([])
const logTotal = ref(0)
const logLoading = ref(false)
const logPage = ref(1)
const logStatusFilter = ref<string | null>(null)
const logSyncModeFilter = ref<string | null>(null)
const logServerFilter = ref<number | null>(null)
const reconcilingLogs = ref(false)
const showDiagnose = ref(false)
const diagnoseLoading = ref(false)
const diagnoseResult = ref<DiagnoseResult | null>(null)
const diagnoseTarget = ref<PushItem | null>(null)
const logServerOptions = computed(() =>
servers.servers.value.map(s => ({ title: s.name, value: s.id })),
)
function onLogFilterChange() {
logPage.value = 1
loadLogs()
}
async function loadLogs() {
logLoading.value = true
try {
const params: Record<string, string | number> = {
page: logPage.value,
per_page: 15,
}
if (logStatusFilter.value) params.status = logStatusFilter.value
if (logSyncModeFilter.value) params.sync_mode = logSyncModeFilter.value
if (logServerFilter.value != null) params.server_id = logServerFilter.value
const res = await http.get<PaginatedResponse<PushItem>>('/servers/logs', params)
logs.value = res.items || []
logTotal.value = res.total || 0
} catch {
logs.value = []
logTotal.value = 0
} finally {
logLoading.value = false
}
}
async function reconcileStaleLogs() {
reconcilingLogs.value = true
try {
const res = await http.post<{ reconciled: number; max_age_minutes: number }>(
'/sync/reconcile-stale-logs',
{ max_age_minutes: 120 },
)
const n = res.reconciled ?? 0
snackbar(n > 0 ? `已订正 ${n} 条卡住的推送记录` : '没有需要订正的记录')
await loadLogs()
} catch (e: unknown) {
snackbar(e instanceof Error ? e.message : '订正失败', 'error')
} finally {
reconcilingLogs.value = false
}
}
async function openDiagnose(item: PushItem) {
diagnoseTarget.value = item
diagnoseResult.value = null
showDiagnose.value = true
diagnoseLoading.value = true
try {
diagnoseResult.value = await http.post<DiagnoseResult>('/sync/diagnose', {
server_id: item.server_id,
target_path: item.target_path || form.targetPath.value || undefined,
})
} catch (e: unknown) {
snackbar(e instanceof Error ? e.message : '诊断失败', 'error')
showDiagnose.value = false
} finally {
diagnoseLoading.value = false
}
}
function closeDiagnose() {
showDiagnose.value = false
}
return {
logs,
logTotal,
logLoading,
logPage,
logStatusFilter,
logSyncModeFilter,
logServerFilter,
reconcilingLogs,
logStatusOptions: LOG_STATUS_OPTIONS,
logSyncModeOptions: LOG_SYNC_MODE_OPTIONS,
logHeaders: LOG_HEADERS,
logServerOptions,
showDiagnose,
diagnoseLoading,
diagnoseResult,
diagnoseTarget,
onLogFilterChange,
loadLogs,
reconcileStaleLogs,
openDiagnose,
closeDiagnose,
}
}
@@ -0,0 +1,40 @@
import { ref, reactive, onMounted, onUnmounted } from 'vue'
import { usePushForm } from './usePushForm'
import { usePushServers } from './usePushServers'
import { usePushProgress } from './usePushProgress'
import { usePushPreview } from './usePushPreview'
import { usePushLogs } from './usePushLogs'
/** Orchestrates push page composables and lifecycle */
export function usePushPage() {
const form = usePushForm()
const pushStatus = ref<Record<number, string>>({})
const servers = usePushServers(pushStatus)
const logs = usePushLogs(form, servers)
const progress = usePushProgress(form, servers, () => logs.loadLogs())
const preview = usePushPreview(form, servers)
function onConfirmAndPush() {
preview.confirmAndPush(() => progress.doPush())
}
onMounted(() => {
servers.loadServers()
logs.loadLogs()
servers.startAutoRefresh(() => progress.pushing.value)
})
onUnmounted(() => {
progress.dispose()
servers.stopAutoRefresh()
})
return {
form: reactive(form),
servers: reactive(servers),
progress: reactive(progress),
preview: reactive(preview),
logs: reactive(logs),
onConfirmAndPush,
}
}
@@ -0,0 +1,104 @@
import { ref, computed } from 'vue'
import { http } from '@/api'
import type { usePushForm } from './usePushForm'
import type { usePushServers } from './usePushServers'
import {
PREVIEW_CONCURRENCY,
PREVIEW_WARN_THRESHOLD,
type PreviewServerRow,
type PreviewStats,
} from './types'
type PushForm = ReturnType<typeof usePushForm>
type PushServers = ReturnType<typeof usePushServers>
export function usePushPreview(form: PushForm, servers: PushServers) {
const showPreview = ref(false)
const previewResults = ref<PreviewServerRow[]>([])
const previewLoading = ref(false)
const previewTotal = ref(0)
const previewDone = ref(0)
const previewLargeBatch = computed(() => previewTotal.value > PREVIEW_WARN_THRESHOLD)
const previewErrorCount = computed(() => previewResults.value.filter(r => r.error).length)
async function previewOneServer(serverId: number, serverName: string): Promise<PreviewServerRow> {
try {
const res = await http.post<{
server_id: number
server_name: string
error?: string
stats?: PreviewStats
}>('/sync/preview', {
source_path: form.effectiveSourcePath(),
target_path: form.targetPath.value,
server_id: serverId,
sync_mode: form.syncMode.value,
})
if (res.error) {
return { server_id: serverId, server_name: serverName, error: res.error }
}
return {
server_id: serverId,
server_name: res.server_name || serverName,
stats: res.stats,
}
} catch (e: unknown) {
return {
server_id: serverId,
server_name: serverName,
error: e instanceof Error ? e.message : '预览失败',
}
}
}
async function doPreview() {
if (!form.validateForm(servers.selectedIds.value.size)) return
const allIds = [...servers.selectedIds.value]
previewTotal.value = allIds.length
previewDone.value = 0
showPreview.value = true
previewLoading.value = true
previewResults.value = []
const queue = allIds.map(id => {
const srv = servers.servers.value.find(s => s.id === id)
return { id, name: srv?.name || `#${id}` }
})
for (let i = 0; i < queue.length; i += PREVIEW_CONCURRENCY) {
const batch = queue.slice(i, i + PREVIEW_CONCURRENCY)
const rows = await Promise.all(batch.map(s => previewOneServer(s.id, s.name)))
previewResults.value = [...previewResults.value, ...rows]
previewDone.value = previewResults.value.length
}
previewLoading.value = false
}
function closePreview() {
showPreview.value = false
}
function confirmAndPush(doPush: () => void) {
showPreview.value = false
doPush()
}
return {
showPreview,
previewResults,
previewLoading,
previewTotal,
previewDone,
previewLargeBatch,
previewWarnThreshold: PREVIEW_WARN_THRESHOLD,
previewErrorCount,
doPreview,
closePreview,
confirmAndPush,
}
}
@@ -0,0 +1,238 @@
import { ref, computed } from 'vue'
import { http } from '@/api'
import { useSnackbar } from '@/composables/useSnackbar'
import { useAuthStore } from '@/stores/auth'
import { buildWebSocketUrl } from '@/utils/wsUrl'
import type { usePushForm } from './usePushForm'
import type { usePushServers } from './usePushServers'
import {
TERMINAL_PROGRESS_STATUSES,
type ServerProgress,
type ServerProgressStatus,
type SyncProgressMsg,
} from './types'
type PushForm = ReturnType<typeof usePushForm>
type PushServers = ReturnType<typeof usePushServers>
export function usePushProgress(
form: PushForm,
servers: PushServers,
onPushFinished: () => void,
) {
const snackbar = useSnackbar()
const auth = useAuthStore()
const pushing = ref(false)
const cancelling = ref(false)
const wsProgressItems = ref<ServerProgress[]>([])
const pushBatchId = ref('')
const wsSocket = ref<WebSocket | null>(null)
const _pushTimeouts: ReturnType<typeof setTimeout>[] = []
const doneCount = computed(() =>
wsProgressItems.value.filter(p => TERMINAL_PROGRESS_STATUSES.includes(p.status)).length,
)
const totalProgress = computed(() => {
const n = wsProgressItems.value.length
return n ? Math.round((doneCount.value / n) * 100) : 0
})
function clearPushTimers() {
_pushTimeouts.forEach(t => clearTimeout(t))
_pushTimeouts.length = 0
}
function checkPushComplete() {
const total = wsProgressItems.value.length
if (!total || !pushing.value) return
const done = wsProgressItems.value.filter(p =>
TERMINAL_PROGRESS_STATUSES.includes(p.status),
).length
if (done < total) return
pushing.value = false
disconnectProgressWs()
onPushFinished()
const failed = wsProgressItems.value.filter(p => p.status === 'failed').length
const cancelled = wsProgressItems.value.filter(p => p.status === 'cancelled').length
const success = wsProgressItems.value.filter(p => p.status === 'success').length
snackbar(`推送完成:${success} 成功,${failed} 失败${cancelled ? `${cancelled} 取消` : ''}`)
}
function applySyncProgressMsg(msg: SyncProgressMsg) {
if (msg.type !== 'sync_progress' || msg.batch_id !== pushBatchId.value || msg.server_id == null) {
return
}
const idx = wsProgressItems.value.findIndex(p => p.id === msg.server_id)
if (idx === -1) return
const item = wsProgressItems.value[idx]
const st = msg.status
if (st === 'running') {
item.status = 'running'
item.detail = '同步中...'
} else if (st === 'success') {
item.status = 'success'
const sec = msg.duration_seconds
item.detail = sec != null ? `完成(${sec}s` : '完成'
servers.pushStatus.value[msg.server_id] = '成功'
} else if (st === 'failed') {
item.status = 'failed'
item.detail = msg.error_message || '失败'
servers.pushStatus.value[msg.server_id] = '失败'
} else if (st === 'cancelled') {
item.status = 'cancelled'
item.detail = msg.error_message || '已取消'
servers.pushStatus.value[msg.server_id] = '已取消'
}
checkPushComplete()
}
function connectProgressWs() {
if (!auth.token) return
disconnectProgressWs()
const url = buildWebSocketUrl('/ws/sync', { token: auth.token })
const socket = new WebSocket(url)
socket.onmessage = (event) => {
if (event.data === 'pong') return
try {
const msg = JSON.parse(event.data) as SyncProgressMsg & { type?: string }
if (msg.type === 'ping') {
socket.send('pong')
return
}
applySyncProgressMsg(msg)
} catch {
// Ignore malformed messages
}
}
socket.onclose = (ev) => {
wsSocket.value = null
if (ev.code !== 1000 && pushing.value) {
wsProgressItems.value.forEach(p => {
if (p.status === 'pending' || p.status === 'running') {
p.status = 'failed'
p.detail = '连接中断'
servers.pushStatus.value[p.id] = '失败'
}
})
pushing.value = false
snackbar('推送进度连接中断,部分服务器状态未知', 'warning')
onPushFinished()
}
}
wsSocket.value = socket
}
function disconnectProgressWs() {
if (wsSocket.value) {
wsSocket.value.close(1000)
wsSocket.value = null
}
}
async function cancelPush() {
if (!pushBatchId.value) return
cancelling.value = true
try {
await http.post('/sync/cancel', { batch_id: pushBatchId.value })
wsProgressItems.value.forEach(s => {
if (s.status === 'pending' || s.status === 'running') {
s.status = 'cancelled'
s.detail = '已取消'
servers.pushStatus.value[s.id] = '已取消'
}
})
pushing.value = false
disconnectProgressWs()
onPushFinished()
snackbar('推送已取消')
} catch (e: unknown) {
snackbar(e instanceof Error ? e.message : '取消失败', 'error')
} finally {
cancelling.value = false
}
}
async function doPush() {
if (pushing.value) return
if (!form.validateForm(servers.selectedIds.value.size)) return
clearPushTimers()
disconnectProgressWs()
pushBatchId.value = ''
servers.pushStatus.value = {}
const ids = [...servers.selectedIds.value]
wsProgressItems.value = ids.map(id => {
const srv = servers.servers.value.find(s => s.id === id)
return { id, name: srv?.name || `#${id}`, status: 'pending' as ServerProgressStatus }
})
pushing.value = true
try {
const res = await http.post<{ batch_id: string }>('/sync/files', {
source_path: form.effectiveSourcePath(),
target_path: form.targetPath.value,
server_ids: ids,
sync_mode: form.syncMode.value,
})
pushBatchId.value = res.batch_id || ''
if (!pushBatchId.value) {
throw new Error('未返回 batch_id,无法追踪进度')
}
connectProgressWs()
} catch (e: unknown) {
const msg = e instanceof Error ? e.message : '推送请求失败'
wsProgressItems.value.forEach(p => {
p.status = 'failed'
p.detail = msg
})
pushing.value = false
snackbar(msg, 'error')
return
}
const safetyTimer = setTimeout(() => {
if (!pushing.value) return
wsProgressItems.value.forEach(p => {
if (p.status === 'pending' || p.status === 'running') {
p.status = 'failed'
p.detail = '等待超时'
servers.pushStatus.value[p.id] = '失败'
}
})
pushing.value = false
disconnectProgressWs()
onPushFinished()
snackbar(`推送超时,${doneCount.value}/${wsProgressItems.value.length} 台已有结果`, 'warning')
}, 300_000)
_pushTimeouts.push(safetyTimer)
}
function dispose() {
disconnectProgressWs()
clearPushTimers()
}
return {
pushing,
cancelling,
wsProgressItems,
pushBatchId,
doneCount,
totalProgress,
doPush,
cancelPush,
dispose,
}
}
@@ -0,0 +1,115 @@
import { ref, computed, type Ref } from 'vue'
import { http } from '@/api'
import type { PushServerItem } from './types'
export function usePushServers(pushStatus: Ref<Record<number, string>>) {
const servers = ref<PushServerItem[]>([])
const serversLoading = ref(false)
const serverSearch = ref('')
const selectedIds = ref<Set<number>>(new Set())
let refreshTimer: ReturnType<typeof setInterval> | null = null
async function loadServers() {
serversLoading.value = true
try {
const res = await http.get<{ items: PushServerItem[] }>('/servers/', { per_page: 200 })
servers.value = res.items || []
} catch {
servers.value = []
} finally {
serversLoading.value = false
}
}
function startAutoRefresh(isPushing: () => boolean) {
stopAutoRefresh()
refreshTimer = setInterval(() => {
if (!isPushing()) loadServers()
}, 60_000)
}
function stopAutoRefresh() {
if (refreshTimer) {
clearInterval(refreshTimer)
refreshTimer = null
}
}
const filteredServers = computed(() => {
if (!serverSearch.value) return servers.value
const q = serverSearch.value.toLowerCase()
return servers.value.filter(
s => s.name.toLowerCase().includes(q) || (s.domain || '').toLowerCase().includes(q),
)
})
const selectAll = computed(() => {
const visible = filteredServers.value
return visible.length > 0 && visible.every(s => selectedIds.value.has(s.id))
})
const serversByCategory = computed(() => {
const groups: Record<string, PushServerItem[]> = {}
for (const s of filteredServers.value) {
const cat = s.category || '未分类'
if (!groups[cat]) groups[cat] = []
groups[cat].push(s)
}
return Object.entries(groups).map(([category, list]) => ({ category, servers: list }))
})
function isCategoryAllSelected(category: string): boolean {
const group = serversByCategory.value.find(g => g.category === category)
if (!group || group.servers.length === 0) return false
return group.servers.every(s => selectedIds.value.has(s.id))
}
function toggleCategory(category: string) {
const group = serversByCategory.value.find(g => g.category === category)
if (!group) return
const next = new Set(selectedIds.value)
const allSelected = group.servers.every(srv => next.has(srv.id))
for (const srv of group.servers) {
if (allSelected) next.delete(srv.id)
else next.add(srv.id)
}
selectedIds.value = next
}
function toggleServer(id: number) {
const next = new Set(selectedIds.value)
if (next.has(id)) next.delete(id)
else next.add(id)
selectedIds.value = next
}
function toggleAll() {
const visible = filteredServers.value
const allSelected = visible.length > 0 && visible.every(s => selectedIds.value.has(s.id))
const next = new Set(selectedIds.value)
for (const srv of visible) {
if (allSelected) next.delete(srv.id)
else next.add(srv.id)
}
selectedIds.value = next
}
return {
servers,
serversLoading,
serverSearch,
selectedIds,
pushStatus,
filteredServers,
selectAll,
serversByCategory,
loadServers,
startAutoRefresh,
stopAutoRefresh,
isCategoryAllSelected,
toggleCategory,
toggleServer,
toggleAll,
}
}
+1 -1
View File
@@ -8,7 +8,7 @@
* alert → { type: "alert", server_id, server_name, alert_type, alert_value }
* recovery → { type: "recovery", server_id, server_name, metric, value }
* system → { type: "system", event_type, message }
* sync_progress → { type: "sync_progress", batch_id, server_id, status, completed, failed, total }
* Push progress uses a separate channel: /ws/sync (see usePushProgress)
* ping → { type: "ping" } → client must respond with text "pong"
*/
import { ref, computed } from 'vue'
+91 -618
View File
@@ -1,633 +1,106 @@
<template>
<v-container fluid class="pa-6">
<!-- ZIP Upload -->
<v-card elevation="0" border rounded="lg" class="mb-4">
<v-card-title class="text-subtitle-1">源文件</v-card-title>
<v-card-text>
<div v-if="!uploadedZip">
<div
class="border-dashed border rounded pa-6 text-center cursor-pointer"
:class="{ 'border-primary bg-primary-lighten-5': isDragging }"
@dragover.prevent="isDragging = true"
@dragleave="isDragging = false"
@drop.prevent="handleZipDrop"
@click="triggerZipInput"
>
<v-icon size="48" color="grey">mdi-cloud-upload-outline</v-icon>
<div class="text-body-2 mt-2">拖拽 ZIP 文件到此处或点击选择</div>
<div class="text-caption text-medium-emphasis">支持 .zip 格式</div>
</div>
<input ref="zipInput" type="file" accept=".zip" hidden @change="handleZipSelect" />
</div>
<div v-else class="d-flex align-center">
<v-icon class="mr-2">mdi-folder-zip</v-icon>
<span>{{ uploadedZip.name }} ({{ uploadedZip.fileCount }} 个文件, {{ formatSize(uploadedZip.size) }})</span>
<v-spacer />
<v-btn size="small" variant="text" color="error" @click="clearZipUpload">清除</v-btn>
</div>
<v-progress-linear v-if="zipUploading" indeterminate class="mt-2" />
</v-card-text>
</v-card>
<PushZipUpload
:uploaded-zip="form.uploadedZip"
:is-dragging="form.isDragging"
:zip-uploading="form.zipUploading"
@dragover="form.isDragging = true"
@dragleave="form.isDragging = false"
@drop="form.handleZipDrop"
@file-select="form.handleZipSelect"
@clear="form.clearZipUpload"
/>
<!-- Toolbar -->
<v-card elevation="0" border rounded="lg" class="mb-4 pa-4">
<v-row align="center" dense>
<v-col cols="12" sm="4">
<v-text-field v-model="sourcePath" label="推送源路径" variant="outlined" density="compact" prepend-inner-icon="mdi-folder" />
</v-col>
<v-col cols="12" sm="4">
<v-text-field v-model="targetPath" label="目标路径" variant="outlined" density="compact" prepend-inner-icon="mdi-folder-marker" />
</v-col>
<v-col cols="12" sm="4" class="d-flex ga-2 justify-end">
<v-btn size="small" variant="tonal" prepend-icon="mdi-eye" @click="doPreview">预览</v-btn>
<v-btn size="small" color="primary" variant="flat" prepend-icon="mdi-upload" @click="doPush" :loading="pushing">推送</v-btn>
</v-col>
</v-row>
</v-card>
<PushToolbar
v-model:source-path="form.sourcePath"
v-model:target-path="form.targetPath"
:zip-locked="!!form.uploadedZip"
:pushing="progress.pushing"
@preview="preview.doPreview"
@push="progress.doPush"
/>
<!-- Server Selection -->
<v-card elevation="0" border rounded="lg" class="mb-4">
<v-card-title class="d-flex align-center">
<v-checkbox v-model="selectAll" label="全选" density="compact" hide-details class="mr-4" @update:model-value="toggleAll" />
<v-text-field
v-model="serverSearch"
prepend-inner-icon="mdi-magnify"
label="搜索服务器..."
density="compact"
hide-details
variant="outlined"
rounded
style="max-width: 240px"
/>
</v-card-title>
<v-divider />
<v-card-text class="pa-2">
<template v-if="filteredServers.length === 0">
<div class="text-center text-medium-emphasis py-6">暂无服务器</div>
</template>
<template v-else>
<div v-for="group in serversByCategory" :key="group.category" class="mb-4">
<div class="d-flex align-center px-2 mb-2">
<v-btn
size="x-small"
variant="text"
:color="isCategoryAllSelected(group.category) ? 'primary' : 'default'"
@click="toggleCategory(group.category)"
>
<v-icon size="16" class="mr-1">
{{ isCategoryAllSelected(group.category) ? 'mdi-checkbox-marked' : 'mdi-checkbox-blank-outline' }}
</v-icon>
{{ group.category }}
</v-btn>
<span class="text-caption text-medium-emphasis ml-1">
({{ group.servers.filter(s => selectedIds.has(s.id)).length }}/{{ group.servers.length }})
</span>
</div>
<v-row dense>
<v-col cols="6" sm="4" md="3" lg="2" v-for="s in group.servers" :key="s.id">
<v-card
:color="selectedIds.has(s.id) ? 'primary' : undefined"
:variant="selectedIds.has(s.id) ? 'tonal' : 'outlined'"
rounded="lg"
class="pa-3 cursor-pointer"
@click="toggleServer(s.id)"
>
<div class="d-flex align-center ga-2">
<v-icon :color="s.status === 'online' ? 'success' : 'error'" size="12">mdi-circle</v-icon>
<span class="text-body-2 font-weight-medium text-truncate">{{ s.name }}</span>
</div>
<div class="text-caption text-medium-emphasis text-truncate">{{ s.domain }}</div>
<div v-if="pushStatus[s.id]" class="text-caption mt-1">
<v-chip :color="pushColor(pushStatus[s.id])" size="x-small" variant="tonal" label border="sm">{{ pushStatus[s.id] }}</v-chip>
</div>
</v-card>
</v-col>
</v-row>
</div>
</template>
</v-card-text>
</v-card>
<PushServerGrid
:select-all="servers.selectAll"
v-model:server-search="servers.serverSearch"
:servers-loading="servers.serversLoading"
:servers-by-category="servers.serversByCategory"
:selected-ids="servers.selectedIds"
:push-status="servers.pushStatus"
:is-category-all-selected="servers.isCategoryAllSelected"
@toggle-all="servers.toggleAll"
@refresh="servers.loadServers"
@toggle-category="servers.toggleCategory"
@toggle-server="servers.toggleServer"
/>
<!-- Sync Mode -->
<v-card elevation="0" border rounded="lg" class="mb-4">
<v-card-title class="text-subtitle-1">同步模式</v-card-title>
<v-card-text>
<v-radio-group v-model="syncMode" inline density="compact">
<v-radio label="增量同步" value="incremental" />
<v-radio label="全量同步" value="full" />
<v-radio label="校验和同步" value="checksum" />
</v-radio-group>
<v-alert v-if="syncMode === 'full'" type="warning" density="compact" variant="tonal" class="mt-2">
全量同步会删除目标目录中不存在于源目录的文件
</v-alert>
<div class="text-caption text-medium-emphasis mt-1">
{{ syncModeDesc }}
</div>
</v-card-text>
</v-card>
<PushSyncModeCard
v-model:sync-mode="form.syncMode"
:sync-mode-desc="form.syncModeDesc"
/>
<!-- Push Progress -->
<v-card v-if="pushing" elevation="0" border rounded="lg" class="mb-4 pa-4">
<div class="text-subtitle-2 mb-2">推送进度</div>
<v-progress-linear :model-value="pushProgress" color="primary" height="8" rounded />
<div class="text-caption text-medium-emphasis mt-1">{{ completedCount }} / {{ selectedIds.size }} 台完成</div>
</v-card>
<PushProgressList
:items="progress.wsProgressItems"
:batch-id="progress.pushBatchId"
:pushing="progress.pushing"
:cancelling="progress.cancelling"
:done-count="progress.doneCount"
:total-progress="progress.totalProgress"
@cancel="progress.cancelPush"
/>
<!-- WebSocket Per-Server Progress -->
<v-card v-if="wsProgressItems.length > 0" class="mb-4" elevation="0" border rounded="lg">
<v-card-title class="d-flex align-center text-subtitle-1">
推送进度
<v-spacer />
<v-btn
v-if="pushBatchId && wsProgressItems.some(s => s.status === 'running' || s.status === 'pending')"
size="small"
variant="tonal"
color="error"
class="mr-2"
:loading="cancelling"
@click="cancelPush"
>
取消推送
</v-btn>
<v-chip size="small" variant="tonal" color="primary">
{{ wsProgressItems.filter(s => s.status === 'success').length }}/{{ wsProgressItems.length }} 完成
</v-chip>
</v-card-title>
<v-divider />
<v-card-text>
<v-list density="compact">
<v-list-item v-for="s in wsProgressItems" :key="s.id" :title="s.name">
<template #append>
<v-icon v-if="s.status === 'pending'" color="grey">mdi-clock-outline</v-icon>
<v-icon v-else-if="s.status === 'running'" color="blue">mdi-loading</v-icon>
<v-icon v-else-if="s.status === 'success'" color="green">mdi-check-circle</v-icon>
<v-icon v-else color="red">mdi-alert-circle</v-icon>
</template>
<template v-if="s.detail" #subtitle>
<span :class="s.status === 'failed' ? 'text-error' : 'text-medium-emphasis'">{{ s.detail }}</span>
</template>
</v-list-item>
</v-list>
</v-card-text>
</v-card>
<PushPreviewDialog
v-model="preview.showPreview"
:loading="preview.previewLoading"
:results="preview.previewResults"
:selected-count="servers.selectedIds.size"
:preview-done="preview.previewDone"
:preview-total="preview.previewTotal"
:large-batch="preview.previewLargeBatch"
:warn-threshold="preview.previewWarnThreshold"
:error-count="preview.previewErrorCount"
:sync-mode="form.syncMode"
@confirm-push="onConfirmAndPush"
/>
<!-- Preview Dialog -->
<v-dialog v-model="showPreview" max-width="600">
<v-card border>
<v-card-title>推送预览</v-card-title>
<v-card-text>
<v-alert v-if="previewData.error" type="error" class="mb-2">{{ previewData.error }}</v-alert>
<div v-else>
<div class="mb-2">将推送到 <strong>{{ selectedIds.size }}</strong> 台服务器</div>
<div v-if="previewData.files" class="text-body-2">{{ previewData.files }} 个文件{{ previewData.size }}</div>
</div>
</v-card-text>
<v-card-actions>
<v-spacer />
<v-btn variant="text" @click="showPreview = false">关闭</v-btn>
<v-btn color="primary" variant="flat" @click="showPreview = false; doPush()">确认推送</v-btn>
</v-card-actions>
</v-card>
</v-dialog>
<PushDiagnoseDialog
v-model="logs.showDiagnose"
:loading="logs.diagnoseLoading"
:result="logs.diagnoseResult"
:target-name="logs.diagnoseTarget?.server_name"
/>
<!-- Push Log -->
<v-card elevation="0" border rounded="lg">
<v-card-title>推送历史</v-card-title>
<v-data-table-server
:items="logs"
:headers="logHeaders"
:items-length="logTotal"
:loading="logLoading"
:page="logPage"
:items-per-page="15"
hover
density="comfortable"
@update:page="logPage = $event; loadLogs()"
>
<template #item.status="{ item }">
<v-chip :color="item.status === 'success' ? 'success' : item.status === 'failed' ? 'error' : 'warning'" size="x-small" variant="tonal" label border="sm">
{{ item.status === 'success' ? '成功' : item.status === 'failed' ? '失败' : item.status }}
</v-chip>
</template>
<template #item.created_at="{ item }">
<span class="text-caption text-medium-emphasis">{{ item.created_at }}</span>
</template>
<template #no-data>
<div class="text-center text-medium-emphasis py-6">暂无数据</div>
</template>
</v-data-table-server>
</v-card>
<PushHistoryTable
:items="logs.logs"
:headers="logs.logHeaders"
:total="logs.logTotal"
:loading="logs.logLoading"
v-model:page="logs.logPage"
v-model:status-filter="logs.logStatusFilter"
v-model:sync-mode-filter="logs.logSyncModeFilter"
v-model:server-filter="logs.logServerFilter"
:status-options="logs.logStatusOptions"
:sync-mode-options="logs.logSyncModeOptions"
:server-options="logs.logServerOptions"
:reconciling="logs.reconcilingLogs"
@filter-change="logs.onLogFilterChange"
@page-change="logs.loadLogs"
@reconcile="logs.reconcileStaleLogs"
@diagnose="logs.openDiagnose"
/>
</v-container>
</template>
<script setup lang="ts">
import { ref, computed, onMounted, onUnmounted } from 'vue'
import { http } from '@/api'
import { useSnackbar } from '@/composables/useSnackbar'
import { useAuthStore } from '@/stores/auth'
import type { PaginatedResponse, PushItem } from '@/types/api'
const snackbar = useSnackbar()
const auth = useAuthStore()
// ── Server list with category ──
interface ServerItem {
id: number
name: string
domain?: string
status?: string
category?: string
}
const servers = ref<ServerItem[]>([])
async function loadServers() {
try {
const res = await http.get<{ items: ServerItem[] }>('/servers/', { per_page: 200 })
servers.value = res.items || []
} catch {
servers.value = []
}
}
// ── State ──
const sourcePath = ref('')
const targetPath = ref('')
const serverSearch = ref('')
const selectedIds = ref<Set<number>>(new Set())
const pushing = ref(false)
const pushProgress = ref(0)
const completedCount = ref(0)
const _pushTimers: ReturnType<typeof setTimeout>[] = []
const pushStatus = ref<Record<number, string>>({})
// ── Sync Mode ──
const syncMode = ref<'incremental' | 'full' | 'checksum'>('incremental')
const syncModeDesc = computed(() => {
switch (syncMode.value) {
case 'incremental': return '仅传输新增或修改的文件'
case 'full': return '使目标目录与源目录完全一致,会删除不存在于源的文件'
case 'checksum': return '通过文件内容校验和判断是否需要传输'
default: return ''
}
})
// ── ZIP Upload ──
const isDragging = ref(false)
const uploadedZip = ref<{ name: string; sourcePath: string; fileCount: number; size: number } | null>(null)
const zipUploading = ref(false)
const zipInput = ref<HTMLInputElement | null>(null)
function triggerZipInput() {
zipInput.value?.click()
}
async function handleZipDrop(e: DragEvent) {
isDragging.value = false
const file = e.dataTransfer?.files[0]
if (file && file.name.endsWith('.zip')) await uploadZip(file)
}
function handleZipSelect(e: Event) {
const file = (e.target as HTMLInputElement).files?.[0]
if (file) uploadZip(file)
}
async function uploadZip(file: File) {
zipUploading.value = true
try {
const form = new FormData()
form.append('file', file)
const res = await http.upload<{ source_path: string; file_count: number; size: number }>('/sync/upload-zip', form)
uploadedZip.value = { name: file.name, sourcePath: res.source_path, fileCount: res.file_count, size: res.size }
snackbar('ZIP 上传成功')
} catch (e: any) { snackbar(e.message || '上传失败', 'error') }
finally { zipUploading.value = false }
}
function clearZipUpload() { uploadedZip.value = null }
function formatSize(bytes: number) {
if (bytes < 1024) return bytes + ' B'
if (bytes < 1048576) return (bytes / 1024).toFixed(1) + ' KB'
return (bytes / 1048576).toFixed(1) + ' MB'
}
// ── WebSocket Progress ──
interface ServerProgress {
id: number
name: string
status: 'pending' | 'running' | 'success' | 'failed'
detail?: string
}
const wsProgressItems = ref<ServerProgress[]>([])
const pushBatchId = ref('')
let wsSocket: WebSocket | null = null
function getWsUrl(): string {
const base = import.meta.env.VITE_API_BASE || ''
if (base) {
const wsBase = base.replace(/^http/, 'ws')
return `${wsBase}/ws/alerts`
}
const proto = location.protocol === 'https:' ? 'wss:' : 'ws:'
return `${proto}//${location.host}/ws/alerts`
}
function connectProgressWs() {
if (!auth.token) return
const url = `${getWsUrl()}?token=${auth.token}`
const socket = new WebSocket(url)
socket.onmessage = (event) => {
if (event.data === 'pong') return
try {
const msg = JSON.parse(event.data)
if (msg.type === 'ping') {
socket.send('pong')
return
}
if (msg.type === 'sync_progress' && msg.batch_id === pushBatchId.value) {
const idx = wsProgressItems.value.findIndex(p => p.id === msg.server_id)
if (idx !== -1) {
const item = wsProgressItems.value[idx]
if (msg.status === 'running') {
item.status = 'running'
item.detail = msg.detail || '同步中...'
} else if (msg.status === 'success') {
item.status = 'success'
item.detail = msg.detail || '完成'
} else if (msg.status === 'failed') {
item.status = 'failed'
item.detail = msg.detail || '失败'
}
}
}
} catch {
// Ignore malformed messages
}
}
socket.onclose = () => {
wsSocket = null
}
wsSocket = socket
}
function disconnectProgressWs() {
if (wsSocket) {
wsSocket.close()
wsSocket = null
}
}
// ── Push Cancel ──
const cancelling = ref(false)
async function cancelPush() {
if (!pushBatchId.value) return
cancelling.value = true
try {
await http.post('/sync/cancel', { batch_id: pushBatchId.value })
// Mark all pending/running as cancelled
wsProgressItems.value.forEach(s => {
if (s.status === 'pending' || s.status === 'running') {
s.status = 'failed'
s.detail = '已取消'
}
})
snackbar('推送已取消')
} catch (e: unknown) {
const msg = e instanceof Error ? e.message : '取消失败'
snackbar(msg, 'error')
} finally {
cancelling.value = false
}
}
// Preview
const showPreview = ref(false)
interface PreviewData { error?: string; files?: number; size?: string }
const previewData = ref<PreviewData>({})
// Logs
const logs = ref<PushItem[]>([])
const logTotal = ref(0)
const logLoading = ref(false)
const logPage = ref(1)
const logHeaders = [
{ title: '时间', key: 'created_at', width: 180 },
{ title: '服务器', key: 'server_name' },
{ title: '状态', key: 'status', width: 100 },
{ title: '源路径', key: 'source_path' },
{ title: '目标路径', key: 'target_path' },
]
// ── Computed ──
const selectAll = computed({
get: () => servers.value.length > 0 && selectedIds.value.size === servers.value.length,
set: () => {},
})
const filteredServers = computed(() => {
if (!serverSearch.value) return servers.value
const q = serverSearch.value.toLowerCase()
return servers.value.filter(s => s.name.toLowerCase().includes(q) || (s.domain || '').toLowerCase().includes(q))
})
// ── Server Grouping by Category ──
const serversByCategory = computed(() => {
const groups: Record<string, ServerItem[]> = {}
for (const s of filteredServers.value) {
const cat = s.category || '未分类'
if (!groups[cat]) groups[cat] = []
groups[cat].push(s)
}
return Object.entries(groups).map(([category, servers]) => ({ category, servers }))
})
function isCategoryAllSelected(category: string): boolean {
const group = serversByCategory.value.find(g => g.category === category)
if (!group || group.servers.length === 0) return false
return group.servers.every(s => selectedIds.value.has(s.id))
}
function toggleCategory(category: string) {
const group = serversByCategory.value.find(g => g.category === category)
if (!group) return
const s = new Set(selectedIds.value)
const allSelected = group.servers.every(srv => s.has(srv.id))
for (const srv of group.servers) {
if (allSelected) s.delete(srv.id)
else s.add(srv.id)
}
selectedIds.value = s
}
// ── Actions ──
/** Effective source path: uploaded ZIP path takes priority over manual input */
function effectiveSourcePath(): string {
return uploadedZip.value?.sourcePath || sourcePath.value
}
function toggleServer(id: number) {
const s = new Set(selectedIds.value)
if (s.has(id)) s.delete(id); else s.add(id)
selectedIds.value = s
}
function toggleAll() {
if (selectedIds.value.size === servers.value.length) {
selectedIds.value = new Set()
} else {
selectedIds.value = new Set(servers.value.map(s => s.id))
}
}
async function doPreview() {
if (!effectiveSourcePath() || selectedIds.value.size === 0) {
snackbar('请选择服务器和源路径(或上传 ZIP 文件)', 'warning')
return
}
try {
previewData.value = await http.post('/sync/preview', {
source_path: effectiveSourcePath(),
target_path: targetPath.value,
server_id: selectedIds.value.values().next().value,
sync_mode: syncMode.value,
})
showPreview.value = true
} catch (e: any) {
previewData.value = { error: e.message }
showPreview.value = true
}
}
async function doPush() {
if (!effectiveSourcePath() || selectedIds.value.size === 0) {
snackbar('请选择服务器和源路径(或上传 ZIP 文件)', 'warning')
return
}
pushing.value = true
pushProgress.value = 0
completedCount.value = 0
pushStatus.value = {}
// Initialize WebSocket progress tracking
const ids = [...selectedIds.value]
wsProgressItems.value = ids.map(id => {
const srv = servers.value.find(s => s.id === id)
return { id, name: srv?.name || `#${id}`, status: 'pending' as const }
})
// Fire batch push and capture batch_id
try {
const res = await http.post<{ batch_id: string }>('/sync/files', {
source_path: effectiveSourcePath(),
target_path: targetPath.value,
server_ids: ids,
sync_mode: syncMode.value,
})
pushBatchId.value = res.batch_id || ''
if (pushBatchId.value) {
connectProgressWs()
}
} catch (e: any) {
// Batch push failed — fall back to per-server push (parallel, not sequential)
const total = ids.length
const results = await Promise.allSettled(
ids.map(async (sid) => {
try {
await http.post('/sync/files', {
source_path: effectiveSourcePath(),
target_path: targetPath.value,
server_ids: [sid],
sync_mode: syncMode.value,
})
pushStatus.value[sid] = '成功'
const idx = wsProgressItems.value.findIndex(p => p.id === sid)
if (idx !== -1) wsProgressItems.value[idx].status = 'success'
} catch (innerErr) {
pushStatus.value[sid] = '失败'
const idx = wsProgressItems.value.findIndex(p => p.id === sid)
if (idx !== -1) {
wsProgressItems.value[idx].status = 'failed'
wsProgressItems.value[idx].detail = (innerErr instanceof Error ? innerErr.message : '推送失败')
}
}
completedCount.value++
pushProgress.value = Math.round((completedCount.value / total) * 100)
})
)
pushing.value = false
loadLogs()
const succeeded = results.filter(r => r.status === 'fulfilled').length
snackbar(`推送完成:${succeeded}/${total} 台成功`)
return
}
// Monitor WS progress completion
const checkInterval = setInterval(() => {
const done = wsProgressItems.value.filter(p => p.status === 'success' || p.status === 'failed').length
completedCount.value = done
pushProgress.value = Math.round((done / wsProgressItems.value.length) * 100)
if (done >= wsProgressItems.value.length) {
clearInterval(checkInterval)
pushing.value = false
disconnectProgressWs()
loadLogs()
const failed = wsProgressItems.value.filter(p => p.status === 'failed').length
const success = wsProgressItems.value.filter(p => p.status === 'success').length
snackbar(`推送完成:${success} 成功,${failed} 失败`)
}
}, 500)
_pushTimers.push(checkInterval as unknown as ReturnType<typeof setTimeout>)
// Safety timeout: stop waiting after 5 minutes
const safetyTimer = setTimeout(() => {
if (pushing.value) {
clearInterval(checkInterval)
pushing.value = false
disconnectProgressWs()
loadLogs()
const done = wsProgressItems.value.filter(p => p.status === 'success' || p.status === 'failed').length
snackbar(`推送超时,${done}/${wsProgressItems.value.length} 台已完成`, 'warning')
}
}, 300000)
_pushTimers.push(safetyTimer)
}
async function loadLogs() {
logLoading.value = true
try {
const res = await http.get<PaginatedResponse<PushItem>>('/servers/logs', { page: logPage.value, per_page: 15 })
logs.value = res.items || []
logTotal.value = res.total || 0
} catch {
logs.value = []
} finally {
logLoading.value = false
}
}
function pushColor(status: string) {
return status === '成功' ? 'success' : status === '失败' ? 'error' : 'warning'
}
onMounted(() => {
loadServers()
loadLogs()
})
onUnmounted(() => {
disconnectProgressWs()
_pushTimers.forEach(t => clearTimeout(t))
_pushTimers.length = 0
})
import PushDiagnoseDialog from '@/components/push/PushDiagnoseDialog.vue'
import PushHistoryTable from '@/components/push/PushHistoryTable.vue'
import PushPreviewDialog from '@/components/push/PushPreviewDialog.vue'
import PushProgressList from '@/components/push/PushProgressList.vue'
import PushServerGrid from '@/components/push/PushServerGrid.vue'
import PushSyncModeCard from '@/components/push/PushSyncModeCard.vue'
import PushToolbar from '@/components/push/PushToolbar.vue'
import PushZipUpload from '@/components/push/PushZipUpload.vue'
import { usePushPage } from '@/composables/push'
const { form, servers, progress, preview, logs, onConfirmAndPush } = usePushPage()
</script>
+9 -4
View File
@@ -172,16 +172,21 @@ export interface ScriptItem {
updated_at: string | null
}
/** Push history item from /api/sync/push-history or /api/sync/history */
/** Push history item from GET /api/servers/logs */
export interface PushItem {
id: number
batch_id: string
server_id: number
server_name: string
server_name: string | null
source_path: string
target_path: string
status: string
created_at: string | null
sync_mode?: string
trigger_type?: string
operator?: string | null
duration_seconds?: number
error_message?: string | null
started_at: string | null
finished_at?: string | null
}
/** Schedule item from /api/schedules/ */
+5
View File
@@ -276,6 +276,11 @@ class SyncCancel(BaseModel):
batch_id: str = Field(..., min_length=1, max_length=32)
class ReconcileStaleLogs(BaseModel):
"""Mark long-stuck sync_logs rows (status=running) as failed."""
max_age_minutes: int = Field(120, ge=5, le=10080)
class SyncVerify(BaseModel):
"""Post-push verification: compare local source with remote target via md5sum."""
server_ids: List[int] = Field(..., min_length=1)
+28
View File
@@ -19,6 +19,7 @@ from server.api.schemas import (
LocalFilePreview,
SyncVerify,
SyncCancel,
ReconcileStaleLogs,
ValidateSourcePath,
SyncDiagnose,
FileSyncDiff,
@@ -487,6 +488,33 @@ async def cancel_sync(
return {"cancelled": True, "batch_id": payload.batch_id}
@router.post("/reconcile-stale-logs")
async def reconcile_stale_sync_logs(
payload: ReconcileStaleLogs,
request: Request,
admin: Admin = Depends(get_current_admin),
):
"""Mark sync_logs stuck in ``running`` longer than *max_age_minutes* as failed.
Repairs rows left by the historical missing ``sync_log_repo.update()`` bug.
"""
from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
repo = SyncLogRepositoryImpl(request.state.db)
count = await repo.reconcile_stale_running(max_age_minutes=payload.max_age_minutes)
await _audit_sync(
"sync_reconcile_stale",
"sync",
0,
f"订正卡住推送记录: {count} 条(>{payload.max_age_minutes} 分钟仍为 running",
admin.username,
request,
)
return {"reconciled": count, "max_age_minutes": payload.max_age_minutes}
# ── H5: Validate Source Path ──
_FORBIDDEN_PATH_PREFIXES = ("/etc", "/root", "/home", "/var", "/boot", "/dev", "/proc", "/sys", "/run", "/srv")
+69 -15
View File
@@ -25,8 +25,9 @@ logger = logging.getLogger("nexus.websocket")
router = APIRouter()
# Redis Pub/Sub channel name
# Redis Pub/Sub channel names (alerts vs push progress — separate to avoid cross-traffic)
REDIS_CHANNEL = "nexus:alerts"
REDIS_SYNC_CHANNEL = "nexus:sync"
class ConnectionManager:
@@ -173,8 +174,9 @@ class ConnectionManager:
self._heartbeat_task = None
# Global connection manager for this worker
# Global connection managers for this worker
manager = ConnectionManager()
sync_manager = ConnectionManager()
# ── Redis Pub/Sub Layer 2 ──
@@ -191,10 +193,14 @@ async def start_redis_subscriber():
from server.infrastructure.redis.client import get_redis
redis = get_redis()
_redis_subscriber = redis.pubsub()
await _redis_subscriber.subscribe(REDIS_CHANNEL)
await _redis_subscriber.subscribe(REDIS_CHANNEL, REDIS_SYNC_CHANNEL)
_pubsub_task = asyncio.create_task(_redis_listen_loop(), name="ws_redis_sub")
logger.info(f"Redis Pub/Sub subscriber started on channel: {REDIS_CHANNEL}")
logger.info(
"Redis Pub/Sub subscriber started on channels: %s, %s",
REDIS_CHANNEL,
REDIS_SYNC_CHANNEL,
)
except Exception as e:
logger.error(f"Redis Pub/Sub subscriber failed to start: {e}")
# Non-fatal — WebSocket still works locally
@@ -214,7 +220,7 @@ async def stop_redis_subscriber():
if _redis_subscriber:
try:
await _redis_subscriber.unsubscribe(REDIS_CHANNEL)
await _redis_subscriber.unsubscribe(REDIS_CHANNEL, REDIS_SYNC_CHANNEL)
await _redis_subscriber.aclose()
except Exception:
logger.debug("Failed to unsubscribe Redis Pub/Sub during shutdown", exc_info=True)
@@ -232,9 +238,15 @@ async def _redis_listen_loop():
)
if message and message.get("type") == "message":
data = message.get("data")
channel = message.get("channel")
if isinstance(channel, bytes):
channel = channel.decode()
if isinstance(data, str):
msg = json.loads(data)
await manager.broadcast_local(msg)
if channel == REDIS_SYNC_CHANNEL:
await sync_manager.broadcast_local(msg)
else:
await manager.broadcast_local(msg)
except asyncio.CancelledError:
break
except Exception as e:
@@ -242,18 +254,27 @@ async def _redis_listen_loop():
await asyncio.sleep(1) # Back off on error
async def _publish_to_redis(message: dict) -> bool:
async def _publish_to_redis(message: dict, channel: str = REDIS_CHANNEL) -> bool:
"""Publish message to Redis channel (Layer 2) — other workers will receive it"""
try:
from server.infrastructure.redis.client import get_redis
redis = get_redis()
await redis.publish(REDIS_CHANNEL, json.dumps(message))
await redis.publish(channel, json.dumps(message))
return True
except Exception as e:
logger.error(f"Redis Pub/Sub publish failed: {e}")
logger.error(f"Redis Pub/Sub publish failed on {channel}: {e}")
return False
async def _dispatch_sync_message(message: dict) -> None:
"""Deliver sync progress to /ws/sync clients only (separate from alert channel)."""
if _pubsub_task is not None and _redis_subscriber is not None:
if not await _publish_to_redis(message, REDIS_SYNC_CHANNEL):
await sync_manager.broadcast_local(message)
else:
await sync_manager.broadcast_local(message)
async def _dispatch_ws_message(message: dict) -> None:
"""Deliver alert to local WebSocket clients without duplicate delivery.
@@ -307,6 +328,41 @@ async def alert_ws(
manager.disconnect(client_id)
@router.websocket("/ws/sync")
async def sync_progress_ws(
websocket: WebSocket,
token: Optional[str] = Query(None, description="JWT access token"),
):
"""WebSocket endpoint dedicated to file-push progress (sync_progress only).
Separated from /ws/alerts so concurrent pushes and alert traffic do not share a channel.
"""
if not token:
await websocket.close(code=4001, reason="Missing JWT token")
return
admin = await _verify_ws_token(token)
if not admin:
await websocket.close(code=4001, reason="Invalid or expired JWT token")
return
client_id = f"sync:{admin.id}:{id(websocket)}"
await sync_manager.connect(client_id, websocket)
try:
while True:
data = await websocket.receive_text()
if data == "pong":
sync_manager.update_pong(client_id)
elif data == "ping":
await websocket.send_json({"type": "pong"})
except WebSocketDisconnect:
sync_manager.disconnect(client_id)
except Exception as e:
logger.warning(f"Sync WebSocket error for {client_id}: {e}")
sync_manager.disconnect(client_id)
async def _verify_ws_token(token: str):
"""Verify JWT token for WebSocket connection (same logic as get_current_admin)
@@ -473,12 +529,10 @@ async def broadcast_sync_progress(
error_message: str = None, duration_seconds: int = None,
retry_job_id: int = None, cancelled: int = 0,
):
"""Broadcast per-server push progress to all WebSocket clients.
"""Broadcast per-server push progress to /ws/sync subscribers.
Called from sync_engine_v2._sync_one() after each server completes.
Frontend uses batch_id to filter stale events from previous pushes.
retry_job_id is set when a failed push auto-creates a retry job.
cancelled is the count of servers skipped due to user cancellation.
Called from sync_engine_v2._sync_one() after each server state change.
Frontend Push page connects to /ws/sync and filters by batch_id.
"""
msg = {
"type": "sync_progress",
@@ -494,5 +548,5 @@ async def broadcast_sync_progress(
"retry_job_id": retry_job_id,
"cancelled": cancelled,
}
await _dispatch_ws_message(msg)
await _dispatch_sync_message(msg)
+50 -7
View File
@@ -20,7 +20,7 @@ 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.database.crypto import decrypt_value
from server.utils.posix_paths import normalize_remote_abs_path, to_posix
from server.utils.posix_paths import UPLOAD_STAGING_PREFIX, normalize_remote_abs_path, to_posix
def _nexus_source_path(path: str) -> str:
@@ -71,11 +71,7 @@ class SyncEngineV2:
concurrency = min(concurrency, MAX_CONCURRENT)
sem = asyncio.Semaphore(concurrency)
servers = []
for sid in server_ids:
server = await self.server_repo.get_by_id(sid)
if server:
servers.append(server)
servers = await self.server_repo.get_by_ids(server_ids)
total = len(servers)
completed = 0
@@ -142,6 +138,20 @@ class SyncEngineV2:
)
sync_log = await self.sync_log_repo.create(sync_log)
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="running",
completed=completed,
failed=failed,
total=total,
)
except Exception as e:
logger.debug(f"WS running broadcast failed for server {server.id}: {e}")
result = await _rsync_push(server, source_path, dest, sync_mode)
sync_log.status = "success" if result["exit_code"] == 0 else "failed"
@@ -150,7 +160,14 @@ class SyncEngineV2:
(sync_log.finished_at - sync_log.started_at).total_seconds()
) if sync_log.started_at else 0
sync_log.error_message = (result.get("stderr") or "")[:1000] if result["exit_code"] != 0 else None
sync_log.diff_summary = result["stdout"][:2000] if result["exit_code"] == 0 else None
if result["exit_code"] == 0:
stats = _parse_rsync_stats(result.get("stdout") or "")
sync_log.files_total = int(stats.get("files_total") or 0)
sync_log.files_transferred = int(stats.get("files_transferred") or 0)
sync_log.bytes_transferred = int(stats.get("transfer_size_bytes") or 0)
sync_log.diff_summary = (result.get("stdout") or "")[:2000]
else:
sync_log.diff_summary = None
sync_log = await self.sync_log_repo.update(sync_log)
# Auto-create retry job for failed pushes (dedup: one pending per server+path)
@@ -252,10 +269,13 @@ class SyncEngineV2:
d["retry_job_id"] = retry_jid
result_dicts[sid] = d
_maybe_cleanup_upload_staging(source_path, trigger_type, failed, cancelled)
return {
"total": total,
"completed": completed,
"failed": failed,
"cancelled": cancelled,
"batch_id": batch_id,
"results": result_dicts,
}
@@ -371,6 +391,8 @@ async def _rsync_push(
args.append("--inplace")
if dry_run:
args.extend(["--dry-run", "--stats"])
else:
args.append("--stats")
if verbose:
args.append("-v")
@@ -440,6 +462,27 @@ async def _rsync_push(
pass
def _maybe_cleanup_upload_staging(
source_path: str,
trigger_type: str,
failed: int,
cancelled: int,
) -> None:
"""Remove ZIP extract dir after a fully successful manual push (retry needs the tree if failed)."""
if trigger_type != "manual" or failed > 0 or cancelled > 0:
return
normalized = _nexus_source_path(source_path)
if not normalized.startswith(UPLOAD_STAGING_PREFIX):
return
staging_dir = normalized.rstrip("/")
if not os.path.isdir(staging_dir):
return
import shutil
shutil.rmtree(staging_dir, ignore_errors=True)
logger.info("Cleaned upload staging dir after push: %s", staging_dir)
def _parse_rsync_stats(output: str) -> dict:
"""Extract key numbers from rsync --stats output."""
stats: dict = {}
+19 -1
View File
@@ -64,7 +64,25 @@ async def retry_runner_loop():
job.retry_count += 1
if sync_result["failed"] == 0:
sync_error = sync_result.get("error")
sync_total = int(sync_result.get("total") or 0)
sync_completed = int(sync_result.get("completed") or 0)
if sync_error or sync_total == 0:
job.last_error = (sync_error or "推送源不可用或无目标服务器")[:500]
if job.retry_count >= job.max_retries:
job.status = "failed"
logger.warning(
f"Retry job {job.id}: invalid sync context, max retries reached"
)
else:
backoff = RETRY_BACKOFF_BASE * (2 ** min(job.retry_count - 1, 6))
job.next_retry_at = now + timedelta(seconds=backoff)
logger.warning(
f"Retry job {job.id}: sync error — {job.last_error}, "
f"next retry in {backoff}s"
)
elif sync_result["failed"] == 0 and sync_completed > 0:
job.status = "completed"
logger.info(f"Retry job {job.id}: success after {job.retry_count} attempts")
elif job.retry_count >= job.max_retries:
@@ -0,0 +1,60 @@
"""Periodic cleanup of stale /tmp/nexus_upload_* ZIP extract directories.
Removes dirs older than MAX_AGE_HOURS that were left behind after failed/cancelled
pushes or abandoned uploads. Successful manual pushes are already cleaned in sync_engine_v2.
"""
from __future__ import annotations
import asyncio
import logging
import os
import shutil
import time
import glob as glob_module
from server.utils.posix_paths import UPLOAD_STAGING_PREFIX
logger = logging.getLogger("nexus.upload_staging_cleanup")
LOOP_INTERVAL_SECONDS = 3600
MAX_AGE_SECONDS = 24 * 3600
def _cleanup_stale_upload_dirs() -> int:
"""Remove upload staging dirs not modified within MAX_AGE_SECONDS. Returns count removed."""
now = time.time()
removed = 0
pattern = f"{UPLOAD_STAGING_PREFIX}*"
for path in glob_module.glob(pattern):
if not os.path.isdir(path):
continue
try:
real = os.path.realpath(path)
if not real.startswith(UPLOAD_STAGING_PREFIX):
logger.warning("Skip upload cleanup (path escape): %s", path)
continue
age = now - os.path.getmtime(real)
if age < MAX_AGE_SECONDS:
continue
shutil.rmtree(real, ignore_errors=True)
if not os.path.exists(real):
removed += 1
logger.info("Removed stale upload staging dir (age %.0fh): %s", age / 3600, real)
except OSError as exc:
logger.warning("Upload staging cleanup skip %s: %s", path, exc)
return removed
async def upload_staging_cleanup_loop() -> None:
"""Run hourly on primary worker."""
while True:
try:
await asyncio.sleep(LOOP_INTERVAL_SECONDS)
count = await asyncio.to_thread(_cleanup_stale_upload_dirs)
if count:
logger.info("Upload staging cleanup cycle removed %d director(ies)", count)
except asyncio.CancelledError:
raise
except Exception:
logger.exception("Upload staging cleanup loop error")
+1 -1
View File
@@ -261,7 +261,7 @@ class PushRetryJob(Base):
target_path = Column(String(500), nullable=False)
status = Column(String(20), default="pending")
retry_count = Column(Integer, default=0)
max_retries = Column(Integer, default=100)
max_retries = Column(Integer, default=3)
next_retry_at = Column(DateTime)
last_error = Column(Text)
created_at = Column(DateTime, default=_utcnow)
@@ -31,6 +31,14 @@ class ServerRepositoryImpl:
result = await self.session.execute(select(Server).where(Server.id == id))
return result.scalar_one_or_none()
async def get_by_ids(self, ids: List[int]) -> List[Server]:
"""Fetch multiple servers in one query; order matches *ids* (skips missing)."""
if not ids:
return []
result = await self.session.execute(select(Server).where(Server.id.in_(ids)))
by_id = {s.id: s for s in result.scalars().all()}
return [by_id[i] for i in ids if i in by_id]
async def get_all(self) -> List[Server]:
result = await self.session.execute(select(Server).order_by(Server.id))
return list(result.scalars().all())
@@ -1,11 +1,16 @@
"""Nexus — SyncLog Repository (Async SQLAlchemy implementation)"""
from datetime import datetime, timedelta, timezone
from typing import Optional, List, Tuple
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from server.domain.models import SyncLog, Server
_STALE_RUNNING_MESSAGE = (
"推送状态异常:长时间未完成(已自动订正,可能因历史缺陷未落库终态)"
)
class SyncLogRepositoryImpl:
"""Async SQLAlchemy implementation of SyncLog data access"""
@@ -66,6 +71,41 @@ class SyncLogRepositoryImpl:
await self.session.commit()
return log
async def update(self, log: SyncLog) -> SyncLog:
"""Persist in-memory mutations on an existing SyncLog (used after rsync completes)."""
self.session.add(log)
await self.session.commit()
await self.session.refresh(log)
return log
async def reconcile_stale_running(self, max_age_minutes: int = 120) -> int:
"""Mark ``running`` logs older than *max_age_minutes* as ``failed``."""
cutoff = datetime.now(timezone.utc) - timedelta(minutes=max_age_minutes)
result = await self.session.execute(
select(SyncLog).where(
SyncLog.status == "running",
SyncLog.started_at < cutoff,
)
)
stale = list(result.scalars().all())
if not stale:
return 0
now = datetime.now(timezone.utc)
for log in stale:
log.status = "failed"
if log.finished_at is None:
log.finished_at = now
if not log.error_message:
log.error_message = _STALE_RUNNING_MESSAGE
if log.duration_seconds == 0 and log.started_at:
log.duration_seconds = max(
0, int((now - log.started_at).total_seconds())
)
await self.session.commit()
return len(stale)
# Allowed fields for update_status — prevents arbitrary field injection via **kwargs
SYNC_LOG_UPDATABLE_FIELDS = {"finished_at", "duration_seconds", "error_message", "diff_summary", "status"}
+15 -3
View File
@@ -76,6 +76,7 @@ from server.background.self_monitor import self_monitor_loop
from server.background.schedule_runner import schedule_runner_loop
from server.background.retry_runner import retry_runner_loop
from server.background.ip_allowlist_refresh import ip_allowlist_refresh_loop
from server.background.upload_staging_cleanup import upload_staging_cleanup_loop
logger = logging.getLogger("nexus")
@@ -263,8 +264,17 @@ async def lifespan(app: FastAPI):
task_schedule = asyncio.create_task(schedule_runner_loop(), name="schedule_runner")
task_retry = asyncio.create_task(retry_runner_loop(), name="retry_runner")
task_ip_refresh = asyncio.create_task(ip_allowlist_refresh_loop(), name="ip_allowlist_refresh")
task_upload_cleanup = asyncio.create_task(
upload_staging_cleanup_loop(), name="upload_staging_cleanup"
)
_background_tasks.extend([
task_flush, task_script_flush, task_monitor, task_schedule, task_retry, task_ip_refresh,
task_flush,
task_script_flush,
task_monitor,
task_schedule,
task_retry,
task_ip_refresh,
task_upload_cleanup,
])
logger.info("Primary worker — background tasks launched")
else:
@@ -277,15 +287,17 @@ async def lifespan(app: FastAPI):
logger.info(
f"{settings.SYSTEM_NAME} v6.0.0 started — "
f"background tasks: heartbeat_flush(10min), script_execution_flush(60s), "
f"self_monitor(30s), schedule_runner(60s), retry_runner(5min)"
f"self_monitor(30s), schedule_runner(60s), retry_runner(5min), "
f"upload_staging_cleanup(1h)"
)
yield
# ── Shutdown ──
# Cancel WebSocket heartbeat task (H11)
from server.api.websocket import manager as ws_manager
from server.api.websocket import manager as ws_manager, sync_manager as ws_sync_manager
ws_manager.cancel_heartbeat()
ws_sync_manager.cancel_heartbeat()
for task in _background_tasks:
task.cancel()
+44
View File
@@ -0,0 +1,44 @@
"""Tests for dedicated push-progress WebSocket channel."""
import pytest
def test_sync_progress_uses_sync_dispatch(monkeypatch):
"""broadcast_sync_progress must not publish to the alert channel."""
from server.api import websocket as ws
calls: list[tuple[str, dict]] = []
async def capture_sync(msg):
calls.append(("sync", msg))
async def capture_alert(msg):
calls.append(("alert", msg))
monkeypatch.setattr(ws, "_dispatch_sync_message", capture_sync)
monkeypatch.setattr(ws, "_dispatch_ws_message", capture_alert)
import asyncio
asyncio.run(
ws.broadcast_sync_progress(
batch_id="b1",
server_id=1,
server_name="srv",
status="running",
completed=0,
failed=0,
total=1,
),
)
assert len(calls) == 1
assert calls[0][0] == "sync"
assert calls[0][1]["type"] == "sync_progress"
assert calls[0][1]["batch_id"] == "b1"
def test_ws_sync_path_is_public():
from server.api.auth_jwt import _is_public_path
assert _is_public_path("/ws/sync") is True
+64
View File
@@ -0,0 +1,64 @@
"""Tests for /tmp/nexus_upload_* periodic cleanup."""
import os
import time
from unittest.mock import patch
import pytest
from server.background.upload_staging_cleanup import _cleanup_stale_upload_dirs
from server.utils.posix_paths import UPLOAD_STAGING_PREFIX
@pytest.fixture
def staging_dirs(tmp_path, monkeypatch):
"""Point UPLOAD_STAGING_PREFIX at tmp_path for isolated tests."""
prefix = str(tmp_path / "nexus_upload_")
monkeypatch.setattr("server.background.upload_staging_cleanup.UPLOAD_STAGING_PREFIX", prefix)
monkeypatch.setattr("server.utils.posix_paths.UPLOAD_STAGING_PREFIX", prefix)
return prefix
def test_cleanup_removes_old_dir(staging_dirs):
old_dir = staging_dirs + "old123"
os.makedirs(old_dir)
old_time = time.time() - (25 * 3600)
os.utime(old_dir, (old_time, old_time))
fresh_dir = staging_dirs + "fresh456"
os.makedirs(fresh_dir)
removed = _cleanup_stale_upload_dirs()
assert removed == 1
assert not os.path.isdir(old_dir)
assert os.path.isdir(fresh_dir)
def test_cleanup_skips_non_directory(staging_dirs):
file_path = staging_dirs + "notadir"
with open(file_path, "w", encoding="utf-8") as f:
f.write("x")
assert _cleanup_stale_upload_dirs() == 0
assert os.path.isfile(file_path)
def test_cleanup_skips_path_outside_prefix(staging_dirs, monkeypatch):
"""realpath must stay under prefix — symlink escape is skipped."""
safe = staging_dirs + "safe"
os.makedirs(safe)
old_time = time.time() - (25 * 3600)
os.utime(safe, (old_time, old_time))
outside = os.path.join(os.path.dirname(staging_dirs.rstrip("_")), "outside_escape")
os.makedirs(outside, exist_ok=True)
link = staging_dirs + "evil"
try:
os.symlink(outside, link)
except OSError:
pytest.skip("symlink not supported on this platform")
with patch("server.background.upload_staging_cleanup.logger"):
removed = _cleanup_stale_upload_dirs()
assert removed >= 1
assert not os.path.isdir(safe)
assert os.path.isdir(outside)