(*)api根据采集任务表初始化采集定时任务编辑任务更新任务

2 years ago · 5189dbe8da
8 changed files with 124 additions and 40 deletions
--- a/api/app/lib/controllers/metadataAcquisition/initJob.js
+++ b/api/app/lib/controllers/metadataAcquisition/initJob.js
@ -0,0 +1,28 @@
+const moment = require('moment')
+// let TEST = false
+TEST = true
+
+module.exports = async function (app, task) {
+
+    let job = null
+    try {
+        job = app.fs.scheduleInit(
+            {
+                interval: task.cron,
+                immediate: TEST,
+                proRun: !TEST,
+            },
+            async () => {
+                try {
+                    console.log(task.taskName, moment().format('YYYY-MM-DD HH:mm:ss'));
+                } catch (error) {
+                    app.fs.logger.error(`sechedule: taskJobs, error: ${error}`)
+                }
+            });
+
+    } catch (error) {
+        app.fs.logger.error(`sechedule: taskJobs, error: ${error}`);
+    }
+
+    return job;
+}
--- a/api/app/lib/controllers/metadataAcquisition/task.js
+++ b/api/app/lib/controllers/metadataAcquisition/task.js
@ -1,5 +1,6 @@
 'use strict';
-
+const schedule = require('node-schedule')
+const moment = require('moment');
 // 新增采集任务
 function addAcquisitionTask(opts) {
    return async function (ctx, next) {
@ -74,9 +75,34 @@ function getAcquisitionTask(opts) {
    }
 }

+const TEST = true;
+async function createJob(app, task) {
+    let job = null
+    try {
+        job = app.fs.scheduleInit(
+            {
+                interval: task.cron,
+                immediate: TEST,
+                proRun: !TEST,
+            },
+            async () => {
+                try {
+                    console.log(task.taskName, moment().format('YYYY-MM-DD HH:mm:ss'));
+                } catch (error) {
+                    app.fs.logger.error(`sechedule: taskJobs, error: ${error}`)
+                }
+            });
+
+    } catch (error) {
+        app.fs.logger.error(`sechedule: taskJobs, error: ${error}`);
+    }
+
+    return { job }
+}
+

 // 修改采集任务
-function editAcquisitionTask(opts) {
+function editAcquisitionTask(opts, app) {
    return async function (ctx, next) {

        try {
@ -93,6 +119,19 @@ function editAcquisitionTask(opts) {
                    body,
                    { where: { id: id, } }
                )
+                const task = await models.AcquisitionTask.findOne({ where: { id: id } });
+
+                //编辑采集任务 更新采集定时任务状态
+                if ((task.enabled && (body.cron || body.enabled))) {
+                    const job = await require(`./initJob.js`)(app, task)
+                    if (app.fs.schedule['taskJobs'][id]) schedule.cancelJob(app.fs.schedule['taskJobs'][id])
+                    app.fs.schedule['taskJobs'][id] = job
+                } else if (!task.enabled) {
+                    if (app.fs.schedule['taskJobs'][id]) {
+                        schedule.cancelJob(app.fs.schedule['taskJobs'][id])
+                    }
+                }
+
                ctx.status = 204;
                ctx.body = { message: '修改采集任务成功' }
            }
@ -131,4 +170,4 @@ module.exports = {
    getAcquisitionTask,
    editAcquisitionTask,
    deleteAcquisitionTask
-}
+}
--- a/api/app/lib/routes/metadataAcquisition/task.js
+++ b/api/app/lib/routes/metadataAcquisition/task.js
@ -13,7 +13,7 @@ module.exports = function (app, router, opts, AuthCode) {

    // 修改采集任务信息
    app.fs.api.logAttr['PUT/acq/task/:id'] = { content: '修改采集任务信息', visible: true };
-    router.put('/acq/task/:id', task.editAcquisitionTask(opts))
+    router.put('/acq/task/:id', task.editAcquisitionTask(opts, app))

    // 删除采集任务信息
    app.fs.api.logAttr['DEL/acq/task/:id'] = { content: '删除采集任务信息', visible: true };
--- a/api/app/lib/schedule/acqTask.js
+++ b/api/app/lib/schedule/acqTask.js
@ -0,0 +1,41 @@
+const moment = require('moment')
+const rimraf = require('rimraf');
+const fs = require("fs");
+const path = require("path")
+
+let TEST = false
+// TEST = true
+
+module.exports = async function (app, opts) {
+
+    const jobs = {}
+    try {
+        const { models } = app.fs.dc
+        const tasks = await models.AcquisitionTask.findAll({
+            where: {
+                enabled: true
+            }
+        })
+        tasks.map(s => {
+            const job = app.fs.scheduleInit(
+                {
+                    interval: s.cron,
+                    immediate: TEST,
+                    proRun: !TEST,
+                },
+                async () => {
+                    try {
+                        console.log(s.taskName, moment().format('YYYY-MM-DD HH:mm:ss'));
+                    } catch (error) {
+                        app.fs.logger.error(`sechedule: taskJobs, error: ${error}`)
+                    }
+                });
+            jobs[s.id] = job;
+        })
+
+    } catch (error) {
+        app.fs.logger.error(`sechedule: acqTask, error: ${error}`);
+    }
+
+    return { taskJobs: jobs }
+}
--- a/api/app/lib/schedule/index.js
+++ b/api/app/lib/schedule/index.js
@ -21,9 +21,9 @@ module.exports = async function (app, opts) {

   app.fs.scheduleInit = scheduleInit

-   fs.readdirSync(__dirname).forEach((filename) => {
+   fs.readdirSync(__dirname).forEach(async (filename) => {
      if (!['index.js'].some(f => filename == f)) {
-         const scheduleList = require(`./${filename}`)(app, opts)
+         const scheduleList = await require(`./${filename}`)(app, opts)
         for (let k of Object.keys(scheduleList)) {
            console.info(`定时任务 ${k} 启动`);
         }
--- a/web/client/src/sections/metadataAcquisition/components/steps/postgre/stepThree.js
+++ b/web/client/src/sections/metadataAcquisition/components/steps/postgre/stepThree.js
@ -10,29 +10,16 @@ import {
    ProFormDependency
 } from '@ant-design/pro-form';
 import { BellOutlined } from '@ant-design/icons'
-import { parseHumanReadable } from 'cron-js-parser';
+var parser = require('cron-parser');
+import moment from 'moment';
 import '../../style.less';
 function StepThree(props) {
    const { prev, next, dataSourceFilter, editData } = props;
    const formRef = React.createRef();
    const initialValues = editData ? editData : {}
-    const testCron = () => {
+    const testCron = (value) => {
        try {
-            let lang = 'zh_CN' //english
-            let obj = {
-                atSeconds: [1, 5, 10],
-                runEveryXMins: {
-                    startAt: 10,
-                    every: 10
-                },
-                runEveryHourInRange: {
-                    from: 2,
-                    to: 20
-                },
-                isEveryDay: true,
-                atYears: [2020, 2022]
-            };
-            console.log(parseHumanReadable("0 15 10 ? * *", obj, lang))
+            var interval = parser.parseExpression(value);
        } catch (err) {
            console.log('Error: ' + err.message);
        }
@ -40,21 +27,8 @@ function StepThree(props) {

    const checkCron = async (rule, value) => {
        try {
-            let lang = 'zh_CN' //english
-            let obj = {
-                atSeconds: [1, 5, 10],
-                runEveryXMins: {
-                    startAt: 10,
-                    every: 10
-                },
-                runEveryHourInRange: {
-                    from: 2,
-                    to: 20
-                },
-                isEveryDay: true,
-                atYears: [2020, 2022]
-            };
-            console.log(parseHumanReadable(value, obj, lang))
+            var interval = parser.parseExpression(value);
+            console.log('Date: ', moment(interval.next().toString()).format('YYYY-MM-DD HH:mm:ss'));
            return Promise.resolve();
        } catch (err) {
            return Promise.reject(new Error('请输入正确的cron表达式!'));
@ -119,7 +93,7 @@ function StepThree(props) {
                ]}
                name="cron"
                label="请输入cron表达式"
-                addonAfter={<Button type='primary' onClick={testCron}>测试</Button>}
+            // addonAfter={<Button type='primary' onClick={testCron}>测试</Button>}
            />

            <ProFormCheckbox
--- a/web/client/src/sections/metadataAcquisition/containers/acquisitionTask.js
+++ b/web/client/src/sections/metadataAcquisition/containers/acquisitionTask.js
@ -203,7 +203,8 @@ function AcquisitionTask(props) {
                // showQuickJumper: true,
                // current: currentPage,
                // pageSize: pageSize || 10,
-                // pageSizeOptions: [10, 20, 50],
+                defaultPageSize: 10,
+                pageSizeOptions: [10, 20, 50],
                // showTotal: (total) => {
                //     return <span style={{ fontSize: 15 }}>{`共${Math.ceil(total / pageSize)}页，${total}项`}</span>
                // },
--- a/web/package.json
+++ b/web/package.json
@ -74,6 +74,7 @@
        "antd": "^4.24.5",
        "args": "^5.0.1",
        "cron-js-parser": "^1.0.17",
+        "cron-parser": "^4.8.1",
        "cross-env": "^7.0.3",
        "crypto-js": "^4.1.1",
        "echarts": "^5.4.1",