江晨
c33fe31f49
- dsh-scheduler: shell 请求无 args 字段，改 stdin 传脚本 + JS 字面量持久化（Unicode 安全） - dsh-usage: 余额查询同样改 stdin 机制 - 记录 CollectedOutput.text / sandboxPolicy danger-full-access / exitCode / workdir 等关键经验
235 lines
10 KiB
JavaScript
235 lines
10 KiB
JavaScript
// dsh-scheduler 插件 — Host 半（最终版，pkg-10）
|
||
// 定时任务调度器：
|
||
// - 调度表达式：cron 5字段（分 时 日 月 周）或 every:N{s|m|h}
|
||
// - 持久化：%DSH_HOME%\storages\dsh-scheduler.json（node 子进程，JS 字符串字面量嵌入，Unicode 安全）
|
||
// - 每秒 tick 检查到期任务，cmd /c 执行命令行
|
||
// - 工具：sched_add / sched_list / sched_remove / sched_run_now / sched_toggle / sched_diag
|
||
//
|
||
// 关键技术点（踩坑记录）：
|
||
// 1. ShellExecRequest 没有 args 字段！脚本走 stdin（command:'node' + stdin:脚本），
|
||
// 数据用 env 或 JS 字符串字面量嵌入（btoa 曾引发中文乱码，改字面量后 UTF-8 正确）。
|
||
// 2. res.stdout 是 CollectedOutput 对象（{text, truncated, spillPath}），用 .text 取值。
|
||
// 3. 宿主 shell 默认沙箱 workspace-write 本机不可用 → 显式 sandboxPolicy: danger-full-access。
|
||
// 4. 结果字段是 res.exitCode（不是 res.code）；工作目录字段是 workdir。
|
||
// 5. 任务命令执行：command 用完整字符串 'cmd /c <命令行>'。
|
||
return {
|
||
inject: ['shell', 'timer'],
|
||
async apply(ctx) {
|
||
const sp = ctx.get('sandboxPolicy')
|
||
const fullAccess = sp ? sp.resolve({ mode: 'danger-full-access' }) : { mode: 'danger-full-access' }
|
||
|
||
const readScript = [
|
||
"const fs=require('node:fs');const os=require('node:os');const path=require('node:path');",
|
||
"const home=process.env.DSH_HOME||path.join(os.homedir(),'.dsh');",
|
||
"const f=path.join(home,'storages','dsh-scheduler.json');",
|
||
"try{console.log(fs.readFileSync(f,'utf8'))}catch(e){console.log('[]')}",
|
||
].join('\n')
|
||
|
||
let tasks = []
|
||
const out = (o) => (o && typeof o === 'object' && 'text' in o) ? o.text : String(o || '')
|
||
const load = async () => {
|
||
try {
|
||
const spec = ctx.shell.resolve({ command: 'node', stdin: readScript, sandboxPolicy: fullAccess })
|
||
const res = await ctx.shell.run(spec)
|
||
const parsed = JSON.parse(out(res.stdout).trim())
|
||
tasks = Array.isArray(parsed) ? parsed : []
|
||
} catch (e) { tasks = [] }
|
||
}
|
||
const save = async () => {
|
||
try {
|
||
const lit = JSON.stringify(JSON.stringify(tasks))
|
||
const script = [
|
||
"const fs=require('node:fs');const os=require('node:os');const path=require('node:path');",
|
||
"const home=process.env.DSH_HOME||path.join(os.homedir(),'.dsh');",
|
||
"const f=path.join(home,'storages','dsh-scheduler.json');",
|
||
'const d=' + lit + ';',
|
||
"fs.mkdirSync(path.dirname(f),{recursive:true});fs.writeFileSync(f,d);",
|
||
].join('\n')
|
||
const spec = ctx.shell.resolve({ command: 'node', stdin: script, sandboxPolicy: fullAccess })
|
||
const res = await ctx.shell.run(spec)
|
||
if (res.exitCode !== 0) console.error('scheduler save exit=' + res.exitCode + ' stderr=' + out(res.stderr).slice(0, 300))
|
||
} catch (e) { console.error('scheduler save failed: ' + String((e && e.message) || e)) }
|
||
}
|
||
|
||
const fieldMatch = (field, v) => {
|
||
if (field === '*' || field === '') return true
|
||
const toks = String(field).split(',')
|
||
for (const tok of toks) {
|
||
const m = tok.match(/^(\d+|\*)(?:-(\d+))?(?:\/(\d+))?$/)
|
||
if (!m) continue
|
||
const step = m[3] ? parseInt(m[3], 10) : 1
|
||
if (m[1] === '*') { if (v % step === 0) return true; continue }
|
||
const start = parseInt(m[1], 10)
|
||
const end = m[2] ? parseInt(m[2], 10) : start
|
||
if (v >= start && v <= end && (v - start) % step === 0) return true
|
||
}
|
||
return false
|
||
}
|
||
const cronMatch = (expr, d) => {
|
||
const p = String(expr).trim().split(/\s+/)
|
||
if (p.length !== 5) return false
|
||
return fieldMatch(p[0], d.getMinutes()) && fieldMatch(p[1], d.getHours()) &&
|
||
fieldMatch(p[2], d.getDate()) && fieldMatch(p[3], d.getMonth() + 1) && fieldMatch(p[4], d.getDay())
|
||
}
|
||
const nextCron = (expr, from) => {
|
||
const d = new Date(from)
|
||
d.setSeconds(0, 0)
|
||
for (let i = 0; i < 60 * 24 * 7; i++) {
|
||
if (cronMatch(expr, d)) return d.getTime()
|
||
d.setMinutes(d.getMinutes() + 1)
|
||
}
|
||
return null
|
||
}
|
||
const parseSchedule = (schedule) => {
|
||
const m = String(schedule).trim().match(/^every:(\d+)(s|m|h)$/)
|
||
if (m) {
|
||
const n = parseInt(m[1], 10)
|
||
const unit = m[2]
|
||
const ms = unit === 's' ? n * 1000 : unit === 'm' ? n * 60000 : n * 3600000
|
||
return { type: 'every', everyMs: ms, cronExpr: null }
|
||
}
|
||
const parts = String(schedule).trim().split(/\s+/)
|
||
if (parts.length === 5) return { type: 'cron', everyMs: null, cronExpr: parts.join(' ') }
|
||
return null
|
||
}
|
||
|
||
const runTask = async (t) => {
|
||
try {
|
||
const spec = ctx.shell.resolve({
|
||
command: 'cmd /c ' + t.command,
|
||
workdir: t.cwd || 'D:\\project',
|
||
timeoutMs: t.timeoutMs || 60000,
|
||
sandboxPolicy: fullAccess,
|
||
})
|
||
const res = await ctx.shell.run(spec)
|
||
t.lastStatus = res.exitCode === 0 ? 'ok' : 'error'
|
||
t.lastOutput = out(res.stdout).slice(0, 400)
|
||
t.lastError = res.exitCode === 0 ? null : out(res.stderr).slice(0, 400)
|
||
} catch (e) {
|
||
t.lastStatus = 'error'
|
||
t.lastError = String((e && e.message) || e).slice(0, 400)
|
||
}
|
||
save()
|
||
}
|
||
|
||
await load()
|
||
for (const t of tasks) {
|
||
const sch = parseSchedule(t.schedule)
|
||
if (sch) { t.scheduleType = sch.type; t.everyMs = sch.everyMs; t.cronExpr = sch.cronExpr }
|
||
if (t.nextRun === undefined || t.nextRun === null) {
|
||
t.nextRun = t.scheduleType === 'cron' ? nextCron(t.cronExpr, Date.now()) : Date.now()
|
||
}
|
||
}
|
||
save()
|
||
|
||
let ticking = false
|
||
ctx.timer.interval(async () => {
|
||
if (ticking) return
|
||
ticking = true
|
||
try {
|
||
const now = Date.now()
|
||
let changed = false
|
||
for (const t of tasks) {
|
||
if (!t.enabled) continue
|
||
let due = false
|
||
if (t.scheduleType === 'every') {
|
||
due = (t.lastRun === null || now - t.lastRun >= t.everyMs)
|
||
if (due) t.nextRun = now + t.everyMs
|
||
} else if (t.scheduleType === 'cron') {
|
||
if (t.nextRun === null || t.nextRun === undefined) t.nextRun = nextCron(t.cronExpr, now)
|
||
if (t.nextRun !== null && now >= t.nextRun) {
|
||
due = true
|
||
t.nextRun = nextCron(t.cronExpr, now + 60000)
|
||
}
|
||
}
|
||
if (due) {
|
||
t.lastRun = now
|
||
t.runs = (t.runs || 0) + 1
|
||
t.lastStatus = 'running'
|
||
changed = true
|
||
runTask(t)
|
||
}
|
||
}
|
||
if (changed) save()
|
||
} finally { ticking = false }
|
||
}, 1000)
|
||
|
||
const tool = (name, description, parameters, execute) => harness.registerTool(ctx, harness.defineTool({
|
||
name, description, parameters,
|
||
output: { schema: { type: 'string' }, render(_a, v) { return [{ type: 'text', text: String(v) }] } },
|
||
async execute(args) {
|
||
try { return await execute(args) } catch (e) { return '错误: ' + String((e && e.message) || e) }
|
||
},
|
||
}))
|
||
|
||
tool('sched_add', '添加一个定时任务。schedule 支持 cron 5字段（分 时 日 月 周，如 */10 * * * * 表示每10分钟）或 every:N{s|m|h}（如 every:30s、every:5m、every:1h）。command 为要执行的命令行（cmd 语法），默认工作目录 D:\\project，默认超时60秒。', {
|
||
name: { type: 'string', required: true, description: '任务名称' },
|
||
schedule: { type: 'string', required: true, description: '调度表达式：cron 5字段或 every:N{s|m|h}' },
|
||
command: { type: 'string', required: true, description: '要执行的命令行' },
|
||
}, async (args) => {
|
||
const sch = parseSchedule(args.schedule)
|
||
if (!sch) return '无效的调度表达式: ' + args.schedule
|
||
const t = {
|
||
id: 't' + Date.now().toString(36) + Math.random().toString(36).slice(2, 6),
|
||
name: args.name,
|
||
schedule: args.schedule,
|
||
scheduleType: sch.type,
|
||
everyMs: sch.everyMs,
|
||
cronExpr: sch.cronExpr,
|
||
command: args.command,
|
||
cwd: 'D:\\project',
|
||
enabled: true,
|
||
lastRun: null, runs: 0, lastStatus: null, lastOutput: null, lastError: null,
|
||
nextRun: sch.type === 'every' ? Date.now() : nextCron(sch.cronExpr, Date.now()),
|
||
createdAt: Date.now(),
|
||
}
|
||
tasks.push(t)
|
||
save()
|
||
return '已添加任务 ' + t.id + '（' + t.name + '），下次运行: ' + (t.nextRun ? new Date(t.nextRun).toLocaleString() : '未排定')
|
||
})
|
||
|
||
tool('sched_list', '列出所有定时任务及其状态。', {}, async () => {
|
||
return JSON.stringify(tasks.map((t) => ({
|
||
id: t.id, name: t.name, schedule: t.schedule, enabled: t.enabled,
|
||
runs: t.runs, lastStatus: t.lastStatus,
|
||
lastRun: t.lastRun ? new Date(t.lastRun).toISOString() : null,
|
||
nextRun: t.nextRun ? new Date(t.nextRun).toISOString() : null,
|
||
command: t.command,
|
||
lastOutput: t.lastOutput || null, lastError: t.lastError || null,
|
||
})), null, 2)
|
||
})
|
||
|
||
tool('sched_remove', '按 id 删除一个定时任务。', {
|
||
id: { type: 'string', required: true, description: '任务 id（sched_list 查看）' },
|
||
}, async (args) => {
|
||
const before = tasks.length
|
||
tasks = tasks.filter((t) => t.id !== args.id)
|
||
if (tasks.length === before) return '未找到任务 ' + args.id
|
||
save()
|
||
return '已删除任务 ' + args.id
|
||
})
|
||
|
||
tool('sched_run_now', '立即执行一个定时任务（不等调度）。', {
|
||
id: { type: 'string', required: true, description: '任务 id' },
|
||
}, async (args) => {
|
||
const t = tasks.find((x) => x.id === args.id)
|
||
if (!t) return '未找到任务 ' + args.id
|
||
t.lastRun = Date.now(); t.runs = (t.runs || 0) + 1; t.lastStatus = 'running'
|
||
save()
|
||
runTask(t)
|
||
return '任务 ' + args.id + '（' + t.name + '）已触发执行'
|
||
})
|
||
|
||
tool('sched_toggle', '启用或停用一个定时任务。', {
|
||
id: { type: 'string', required: true, description: '任务 id' },
|
||
enabled: { type: 'boolean', required: true, description: 'true 启用 / false 停用' },
|
||
}, async (args) => {
|
||
const t = tasks.find((x) => x.id === args.id)
|
||
if (!t) return '未找到任务 ' + args.id
|
||
t.enabled = !!args.enabled
|
||
save()
|
||
return '任务 ' + args.id + ' 已' + (t.enabled ? '启用' : '停用')
|
||
})
|
||
},
|
||
}
|