增加新建订阅

This commit is contained in:
whyour 2022-05-10 14:38:06 +08:00
parent 85d8565241
commit ebb6290468
13 changed files with 489 additions and 31 deletions

View File

@ -3,8 +3,10 @@ import { DataTypes, Model, ModelDefined } from 'sequelize';
import { SimpleIntervalSchedule } from 'toad-scheduler';
export class Subscription {
id?: number;
name?: string;
type?: 'public-repo' | 'private-repo' | 'file';
schedule_type?: 'crontab' | 'interval';
schedule?: string | SimpleIntervalSchedule;
url?: string;
whitelist?: string;
@ -14,11 +16,15 @@ export class Subscription {
status?: SubscriptionStatus;
pull_type?: 'ssh-key' | 'user-pwd';
pull_option?:
| { private_key: string; key_alias: string }
| { private_key: string }
| { username: string; password: string };
pid?: string;
pid?: number;
isDisabled?: 1 | 0;
log_path?: string;
alias: string;
constructor(options: Subscription) {
this.id = options.id;
this.name = options.name;
this.type = options.type;
this.schedule = options.schedule;
@ -31,6 +37,10 @@ export class Subscription {
this.pull_type = options.pull_type;
this.pull_option = options.pull_option;
this.pid = options.pid;
this.isDisabled = options.isDisabled;
this.log_path = options.log_path;
this.schedule_type = options.schedule_type;
this.alias = options.alias;
}
}
@ -44,7 +54,7 @@ export enum SubscriptionStatus {
interface SubscriptionInstance
extends Model<Subscription, Subscription>,
Subscription {}
export const CrontabModel = sequelize.define<SubscriptionInstance>(
export const SubscriptionModel = sequelize.define<SubscriptionInstance>(
'Subscription',
{
name: {
@ -67,5 +77,9 @@ export const CrontabModel = sequelize.define<SubscriptionInstance>(
pull_type: DataTypes.STRING,
pull_option: DataTypes.JSON,
pid: DataTypes.NUMBER,
isDisabled: DataTypes.NUMBER,
log_path: DataTypes.STRING,
schedule_type: DataTypes.STRING,
alias: DataTypes.STRING,
},
);

View File

@ -0,0 +1,254 @@
import { Service, Inject } from 'typedi';
import winston from 'winston';
import config from '../config';
import {
Subscription,
SubscriptionModel,
SubscriptionStatus,
} from '../data/subscription';
import { exec, execSync, spawn } from 'child_process';
import fs from 'fs';
import cron_parser from 'cron-parser';
import { getFileContentByName, concurrentRun, fileExist } from '../config/util';
import { promises, existsSync } from 'fs';
import { promisify } from 'util';
import { Op } from 'sequelize';
import path from 'path';
@Service()
export default class SubscriptionService {
constructor(@Inject('logger') private logger: winston.Logger) {}
public async create(payload: Subscription): Promise<Subscription> {
const tab = new Subscription(payload);
const doc = await this.insert(tab);
return doc;
}
public async insert(payload: Subscription): Promise<Subscription> {
return await SubscriptionModel.create(payload, { returning: true });
}
public async update(payload: Subscription): Promise<Subscription> {
const newDoc = await this.updateDb(payload);
return newDoc;
}
public async updateDb(payload: Subscription): Promise<Subscription> {
await SubscriptionModel.update(payload, { where: { id: payload.id } });
return await this.getDb({ id: payload.id });
}
public async status({
ids,
status,
pid,
log_path,
last_running_time = 0,
last_execution_time = 0,
}: {
ids: number[];
status: SubscriptionStatus;
pid: number;
log_path: string;
last_running_time: number;
last_execution_time: number;
}) {
const options: any = {
status,
pid,
log_path,
last_execution_time,
};
if (last_running_time > 0) {
options.last_running_time = last_running_time;
}
return await SubscriptionModel.update(
{ ...options },
{ where: { id: ids } },
);
}
public async remove(ids: number[]) {
await SubscriptionModel.destroy({ where: { id: ids } });
}
public async getDb(query: any): Promise<Subscription> {
const doc: any = await SubscriptionModel.findOne({ where: { ...query } });
return doc && (doc.get({ plain: true }) as Subscription);
}
public async run(ids: number[]) {
await SubscriptionModel.update(
{ status: SubscriptionStatus.queued },
{ where: { id: ids } },
);
concurrentRun(
ids.map((id) => async () => await this.runSingle(id)),
10,
);
}
public async stop(ids: number[]) {
const docs = await SubscriptionModel.findAll({ where: { id: ids } });
for (const doc of docs) {
if (doc.pid) {
try {
process.kill(-doc.pid);
} catch (error) {
this.logger.silly(error);
}
}
const err = await this.killTask(doc.command);
const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
const logFileExist = doc.log_path && (await fileExist(absolutePath));
if (logFileExist) {
const str = err ? `\n${err}` : '';
fs.appendFileSync(
`${absolutePath}`,
`${str}\n## 执行结束... ${new Date()
.toLocaleString('zh', { hour12: false })
.replace(' 24:', ' 00:')} `,
);
}
}
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: undefined },
{ where: { id: ids } },
);
}
public async killTask(name: string) {
let taskCommand = `ps -ef | grep "${name}" | grep -v grep | awk '{print $1}'`;
const execAsync = promisify(exec);
try {
let pid = (await execAsync(taskCommand)).stdout;
if (pid) {
pid = (await execAsync(`pstree -p ${pid}`)).stdout;
} else {
return;
}
let pids = pid.match(/\(\d+/g);
const killLogs = [];
if (pids && pids.length > 0) {
// node 执行脚本时还会有10个子进程但是ps -ef中不存在所以截取前三个
for (const id of pids) {
const c = `kill -9 ${id.slice(1)}`;
try {
const { stdout, stderr } = await execAsync(c);
if (stderr) {
killLogs.push(stderr);
}
if (stdout) {
killLogs.push(stdout);
}
} catch (error: any) {
killLogs.push(error.message);
}
}
}
return killLogs.length > 0 ? JSON.stringify(killLogs) : '';
} catch (e) {
return JSON.stringify(e);
}
}
private async runSingle(cronId: number): Promise<number> {
return new Promise(async (resolve: any) => {
const cron = await this.getDb({ id: cronId });
if (cron.status !== SubscriptionStatus.queued) {
resolve();
return;
}
let { id, log_path } = cron;
const absolutePath = path.resolve(config.logPath, `${log_path}`);
const logFileExist = log_path && (await fileExist(absolutePath));
this.logger.silly('Running job');
this.logger.silly('ID: ' + id);
this.logger.silly('Original command: ');
let cmdStr = '';
const cp = spawn(cmdStr, { shell: '/bin/bash' });
await SubscriptionModel.update(
{ status: SubscriptionStatus.running, pid: cp.pid },
{ where: { id } },
);
cp.stderr.on('data', (data) => {
if (logFileExist) {
fs.appendFileSync(`${absolutePath}`, `${data}`);
}
});
cp.on('error', (err) => {
if (logFileExist) {
fs.appendFileSync(`${absolutePath}`, `${JSON.stringify(err)}`);
}
});
cp.on('exit', async (code, signal) => {
this.logger.info(`${''} pid: ${cp.pid} exit ${code} signal ${signal}`);
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: undefined },
{ where: { id } },
);
resolve();
});
cp.on('close', async (code) => {
this.logger.info(`${''} pid: ${cp.pid} closed ${code}`);
await SubscriptionModel.update(
{ status: SubscriptionStatus.idle, pid: undefined },
{ where: { id } },
);
resolve();
});
});
}
public async disabled(ids: number[]) {
await SubscriptionModel.update({ isDisabled: 1 }, { where: { id: ids } });
}
public async enabled(ids: number[]) {
await SubscriptionModel.update({ isDisabled: 0 }, { where: { id: ids } });
}
public async log(id: number) {
const doc = await this.getDb({ id });
if (!doc) {
return '';
}
const absolutePath = path.resolve(config.logPath, `${doc.log_path}`);
const logFileExist = doc.log_path && (await fileExist(absolutePath));
if (logFileExist) {
return getFileContentByName(`${absolutePath}`);
}
}
public async logs(id: number) {
const doc = await this.getDb({ id });
if (!doc) {
return [];
}
if (doc.log_path) {
const relativeDir = path.dirname(`${doc.log_path}`);
const dir = path.resolve(config.logPath, relativeDir);
if (existsSync(dir)) {
let files = await promises.readdir(dir);
return files
.map((x) => ({
filename: x,
directory: relativeDir.replace(config.logPath, ''),
time: fs.statSync(`${dir}/${x}`).mtime.getTime(),
}))
.sort((a, b) => b.time - a.time);
}
}
}
}

View File

@ -84,7 +84,8 @@ run_normal() {
fi
fi
log_time=$(date "+%Y-%m-%d-%H-%M-%S")
local time=$(date)
log_time=$(date -d "$time" "+%Y-%m-%d-%H-%M-%S")
log_dir_tmp="${file_param##*/}"
if [[ $file_param =~ "/" ]]; then
if [[ $file_param == /* ]]; then
@ -102,8 +103,8 @@ run_normal() {
[[ "$show_log" == "true" ]] && cmd=""
make_dir "$dir_log/$log_dir"
local begin_time=$(date '+%Y-%m-%d %H:%M:%S')
local begin_timestamp=$(date "+%s")
local begin_time=$(date -d "$time" "+%Y-%m-%d %H:%M:%S")
local begin_timestamp=$(date -d "$time" "+%s")
eval echo -e "\#\# 开始执行... $begin_time\\\n" $cmd
[[ -f $task_error_log_path ]] && eval cat $task_error_log_path $cmd
@ -153,7 +154,8 @@ run_concurrent() {
[[ ! -z $cookieStr ]] && export ${env_param}=${cookieStr}
define_program "$file_param"
log_time=$(date "+%Y-%m-%d-%H-%M-%S")
local time=$(date)
log_time=$(date -d "$time" "+%Y-%m-%d-%H-%M-%S")
log_dir_tmp="${file_param##*/}"
if [[ $file_param =~ "/" ]]; then
if [[ $file_param == /* ]]; then
@ -171,8 +173,8 @@ run_concurrent() {
[[ "$show_log" == "true" ]] && cmd=""
make_dir "$dir_log/$log_dir"
local begin_time=$(date '+%Y-%m-%d %H:%M:%S')
local begin_timestamp=$(date "+%s")
local begin_time=$(date -d "$time" "+%Y-%m-%d %H:%M:%S")
local begin_timestamp=$(date -d "$time" "+%s")
eval echo -e "\#\# 开始执行... $begin_time\\\n" $cmd
[[ -f $task_error_log_path ]] && eval cat $task_error_log_path $cmd
@ -222,7 +224,8 @@ run_designated() {
fi
define_program "$file_param"
log_time=$(date "+%Y-%m-%d-%H-%M-%S")
local time=$(date)
log_time=$(date -d "$time" "+%Y-%m-%d-%H-%M-%S")
log_dir_tmp="${file_param##*/}"
if [[ $file_param =~ "/" ]]; then
if [[ $file_param == /* ]]; then
@ -240,8 +243,8 @@ run_designated() {
[[ "$show_log" == "true" ]] && cmd=""
make_dir "$dir_log/$log_dir"
local begin_time=$(date '+%Y-%m-%d %H:%M:%S')
local begin_timestamp=$(date "+%s")
local begin_time=$(date -d "$time" "+%Y-%m-%d %H:%M:%S")
local begin_timestamp=$(date -d "$time" "+%s")
local envs=$(eval echo "\$${env_param}")
local array=($(echo $envs | sed 's/&/ /g'))
@ -285,7 +288,8 @@ run_designated() {
run_else() {
local file_param="$1"
define_program "$file_param"
log_time=$(date "+%Y-%m-%d-%H-%M-%S")
local time=$(date)
log_time=$(date -d "$time" "+%Y-%m-%d-%H-%M-%S")
log_dir_tmp="${file_param##*/}"
if [[ $file_param =~ "/" ]]; then
if [[ $file_param == /* ]]; then
@ -303,8 +307,8 @@ run_else() {
[[ "$show_log" == "true" ]] && cmd=""
make_dir "$dir_log/$log_dir"
local begin_time=$(date '+%Y-%m-%d %H:%M:%S')
local begin_timestamp=$(date "+%s")
local begin_time=$(date -d "$time" "+%Y-%m-%d %H:%M:%S")
local begin_timestamp=$(date -d "$time" "+%s")
eval echo -e "\#\# 开始执行... $begin_time\\\n" $cmd
[[ -f $task_error_log_path ]] && eval cat $task_error_log_path $cmd

View File

@ -9,6 +9,11 @@
url('../assets/fonts/SourceCodePro-Regular.ttf') format('truetype');
}
.ant-modal-body {
max-height: calc(85vh - 110px);
overflow-y: auto;
}
.log-modal {
.ant-modal {
padding-bottom: 0 !important;

View File

@ -49,6 +49,7 @@ const CronModal = ({
title={cron ? '编辑任务' : '新建任务'}
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()
@ -166,6 +167,7 @@ const CronLabelModal = ({
title="批量修改标签"
visible={visible}
footer={buttons}
centered
forceRender
onCancel={() => handleCancel(false)}
confirmLoading={loading}

View File

@ -71,6 +71,7 @@ const DependenceModal = ({
title={dependence ? '编辑依赖' : '新建依赖'}
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()

View File

@ -41,6 +41,7 @@ const EditNameModal = ({
title="修改环境变量名称"
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()

View File

@ -61,6 +61,7 @@ const EnvModal = ({
title={env ? '编辑变量' : '新建变量'}
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()

View File

@ -56,6 +56,7 @@ const EditScriptNameModal = ({
title="新建文件"
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()

View File

@ -43,6 +43,7 @@ const SaveModal = ({
title="保存文件"
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()

View File

@ -43,6 +43,7 @@ const SettingModal = ({
title="运行设置"
visible={visible}
forceRender
centered
onCancel={() => handleCancel()}
>
<Form

View File

@ -43,6 +43,7 @@ const AppModal = ({
title={app ? '编辑应用' : '新建应用'}
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()

View File

@ -1,10 +1,11 @@
import React, { useEffect, useState } from 'react';
import { Modal, message, Input, Form, Button } from 'antd';
import { Modal, message, InputNumber, Form, Radio, Select, Input } from 'antd';
import { request } from '@/utils/http';
import config from '@/utils/config';
import cron_parser from 'cron-parser';
import EditableTagGroup from '@/components/tag';
const { Option } = Select;
const SubscriptionModal = ({
subscription,
handleCancel,
@ -16,6 +17,9 @@ const SubscriptionModal = ({
}) => {
const [form] = Form.useForm();
const [loading, setLoading] = useState(false);
const [type, setType] = useState('public-repo');
const [scheduleType, setScheduleType] = useState('crontab');
const [pullType, setPullType] = useState<'ssh-key' | 'user-pwd'>('ssh-key');
const handleOk = async (values: any) => {
setLoading(true);
@ -45,6 +49,96 @@ const SubscriptionModal = ({
}
};
const typeChange = (e) => {
setType(e.target.value);
};
const scheduleTypeChange = (e) => {
setScheduleType(e.target.value);
form.setFieldsValue({ schedule: '' });
};
const pullTypeChange = (e) => {
setPullType(e.target.value);
};
const IntervalSelect = ({
value,
onChange,
}: {
value?: any;
onChange?: (param: any) => void;
}) => {
const [intervalType, setIntervalType] = useState('days');
const [intervalNumber, setIntervalNumber] = useState<number>();
const intervalTypeChange = (e) => {
setIntervalType(e.target.value);
onChange?.({ [e.target.value]: intervalNumber });
};
const numberChange = (value: number) => {
setIntervalNumber(value);
onChange?.({ [intervalType]: value });
};
useEffect(() => {
if (value) {
const key = Object.keys(value)[0];
if (key) {
setIntervalType(key);
setIntervalNumber(value[key]);
}
}
}, [value]);
return (
<Input.Group compact>
<InputNumber
addonBefore="每"
precision={0}
min={1}
defaultValue={intervalNumber}
style={{ width: 'calc(100% - 58px)' }}
onChange={numberChange}
/>
<Select defaultValue={intervalType} onChange={intervalTypeChange}>
<Option value="days"></Option>
<Option value="hours"></Option>
<Option value="minutes"></Option>
<Option value="seconds"></Option>
</Select>
</Input.Group>
);
};
const PullOptions = ({
value,
onChange,
type,
}: {
value?: any;
type: 'ssh-key' | 'user-pwd';
onChange?: (param: any) => void;
}) => {
return type === 'ssh-key' ? (
<Form.Item name="private_key" label="私钥" rules={[{ required: true }]}>
<Input.TextArea rows={4} autoSize={true} placeholder="请输入私钥" />
</Form.Item>
) : (
<>
<Form.Item name="username" label="用户名" rules={[{ required: true }]}>
<Input placeholder="请输入认证用户名" />
</Form.Item>
<Form.Item
name="password"
tooltip="Github已不支持密码认证请使用Token方式"
label="密码/Token"
rules={[{ required: true }]}
>
<Input placeholder="请输入密码或者Token" />
</Form.Item>
</>
);
};
useEffect(() => {
form.resetFields();
}, [subscription, visible]);
@ -54,6 +148,7 @@ const SubscriptionModal = ({
title={subscription ? '编辑订阅' : '新建订阅'}
visible={visible}
forceRender
centered
onOk={() => {
form
.validateFields()
@ -69,23 +164,25 @@ const SubscriptionModal = ({
>
<Form
form={form}
layout="vertical"
name="form_in_modal"
initialValues={subscription}
layout="vertical"
>
<Form.Item name="name" label="名称">
<Input placeholder="请输入订阅名称" />
<Form.Item name="name" label="别名">
<Input placeholder="请输入订阅别名" />
</Form.Item>
<Form.Item name="url" label="链接" rules={[{ required: true }]}>
<Input placeholder="请输入订阅链接" />
</Form.Item>
<Form.Item
name="command"
label="命令"
rules={[{ required: true, whitespace: true }]}
name="schedule_type"
label="定时类型"
initialValue={'crontab'}
>
<Input.TextArea
rows={4}
autoSize={true}
placeholder="请输入要执行的命令"
/>
<Radio.Group onChange={scheduleTypeChange}>
<Radio value="crontab">crontab</Radio>
<Radio value="interval">interval</Radio>
</Radio.Group>
</Form.Item>
<Form.Item
name="schedule"
@ -94,7 +191,11 @@ const SubscriptionModal = ({
{ required: true },
{
validator: (rule, value) => {
if (!value || cron_parser.parseExpression(value).hasNext()) {
if (
scheduleType === 'interval' ||
!value ||
cron_parser.parseExpression(value).hasNext()
) {
return Promise.resolve();
} else {
return Promise.reject('Subscription表达式格式有误');
@ -103,11 +204,82 @@ const SubscriptionModal = ({
},
]}
>
<Input placeholder="秒(可选) 分 时 天 月 周" />
{scheduleType === 'interval' ? (
<IntervalSelect />
) : (
<Input placeholder="秒(可选) 分 时 天 月 周" />
)}
</Form.Item>
<Form.Item name="labels" label="标签">
<EditableTagGroup />
<Form.Item name="type" label="类型" initialValue={'public-repo'}>
<Radio.Group onChange={typeChange}>
<Radio value="public-repo"></Radio>
<Radio value="private-repo"></Radio>
<Radio value="file"></Radio>
</Radio.Group>
</Form.Item>
{type !== 'file' && (
<>
<Form.Item
name="command"
label="白名单"
rules={[{ whitespace: true }]}
tooltip="多个关键词竖线分割,支持正则表达式"
>
<Input.TextArea
rows={4}
autoSize={true}
placeholder="请输入脚本筛选白名单关键词,多个关键词竖线分割"
/>
</Form.Item>
<Form.Item
name="command"
label="黑名单"
rules={[{ whitespace: true }]}
tooltip="多个关键词竖线分割,支持正则表达式"
>
<Input.TextArea
rows={4}
autoSize={true}
placeholder="请输入脚本筛选黑名单关键词,多个关键词竖线分割"
/>
</Form.Item>
<Form.Item
name="command"
label="依赖文件"
rules={[{ whitespace: true }]}
tooltip="多个关键词竖线分割,支持正则表达式"
>
<Input.TextArea
rows={4}
autoSize={true}
placeholder="请输入脚本依赖文件关键词,多个关键词竖线分割"
/>
</Form.Item>
{type === 'private-repo' && (
<>
<Form.Item
name="pull_type"
label="拉取方式"
initialValue={'ssh-key'}
>
<Radio.Group onChange={pullTypeChange}>
<Radio value="ssh-key"></Radio>
<Radio value="user-pwd">/Token</Radio>
</Radio.Group>
</Form.Item>
<PullOptions type={pullType} />
</>
)}
<Form.Item
name="alias"
label="唯一值"
rules={[{ required: true }]}
tooltip="唯一值用于日志目录和私钥别名"
>
<Input placeholder="自动生成" disabled />
</Form.Item>
</>
)}
</Form>
</Modal>
);