/** * 对爬虫任务列表目录进行监控 * 发现新任务 * 删除已完成的任务文件 * 内存中保存所有任务,及其状态 * 返回当前任务状态 */ import common from './common.mjs'; import fs from 'node:fs'; import { readdir, readFile } from 'node:fs/promises'; import path from 'node:path'; import cron from 'node-cron'; class TaskMoniter { constructor(task_list_dir) { this.check_time_gap = 10; //检测间隔时间,单位:秒 this.checking = false; this.task_dir = task_list_dir; //监控目录:任务列表保存目录 this.tasks = {}; //内存中的任务列表 this.taskStatus = { //当前任务状态 total: 0, //总任务数 waiting: 0, //等待执行的任务数 running: 0, //正在执行的任务数 done: 0, //已完成的任务数 failed: 0 //执行失败的任务数 }; this.statusCode = { waiting: 'waiting', running: 'running', done: 'done', failed: 'failed', }; } getStatus() { return this.taskStatus; } getNewTask() { let task = null; for (const item of this.tasks) { if (item.status == this.statusCode.waiting) { task = item; break; } } return task; } setTaskRunning(task_id) { if (typeof(this.tasks[task_id]) == 'undefined') { return false; } this.tasks[task_id].status = this.statusCode.running; this.taskStatus[this.statusCode.running] ++; return true; } setTaskDone(task_id) { if (typeof(this.tasks[task_id]) == 'undefined') { return false; } this.tasks[task_id].status = this.statusCode.done; this.taskStatus[this.statusCode.done] ++; return true; } setTaskFailed(task_id) { if (typeof(this.tasks[task_id]) == 'undefined') { return false; } this.tasks[task_id].status = this.statusCode.failed; this.taskStatus[this.statusCode.failed] ++; return true; } async parseTaskFile(filename, filepath) { let task = {}; try { task.id = filename.replace('.task', ''); task.status = this.statusCode.waiting; task.url = await readFile(filepath, { encoding: 'utf8' }); if (task.url) { task.url = task.url.replace(/[\r\n]/g, ''); } }catch(error) { console.error('Get task file content failed: %s', error); } return task; } addTask(task) { if (typeof(this.tasks[task.id]) != 'undefined') { return false; } this.tasks[task.id] = task; this.taskStatus[task.status] ++; this.taskStatus.total ++; return true; } async checkTasks() { if (this.checking == true) { return; } try { console.log('[%s] TaskMoniter auto check...', common.getTimeString()); this.checking = true; const dirPath = path.resolve(this.task_dir); const files = await readdir(dirPath); let task = null; for (const file of files) { if (file.indexOf('.task') === -1) {continue;} //ignore not *.task files task = await this.parseTaskFile(file, `${dirPath}/${file}`); this.addTask(task); } this.checking = false; }catch(error) { this.checking = false; console.error('Check tasks failed: %s', error); } } run() { //开始监控任务目录,把所有任务缓存到内存 console.log('[%s] TaskMoniter started.', common.getTimeString()); //auto run const _self = this; const task_check_time = this.check_time_gap; const task_auto_run = cron.schedule(`*/${task_check_time} * * * * *`, async () => { await _self.checkTasks(); console.log('Status', _self.getStatus()); }, { scheduled: false }); task_auto_run.start(); console.log('[%s] TaskMoniter auto check started.', common.getTimeString()); } } export default TaskMoniter;