631 lines
21 KiB
JavaScript
631 lines
21 KiB
JavaScript
import fs from 'fs';
|
||
import path from 'path';
|
||
import { fileURLToPath } from 'url';
|
||
import { genId, timestamp } from './utils.mjs';
|
||
import { checkLogin } from './login.mjs';
|
||
import { searchProducts } from './search.mjs';
|
||
import { coarseFilter, fineFilter } from './filter.mjs';
|
||
import { buildAnalysis, renderAnalysisMarkdown } from './analyzer.mjs';
|
||
import { sendNotifications } from './notifier.mjs';
|
||
|
||
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
||
const DATA_DIR = path.join(__dirname, '..', 'data');
|
||
const TASKS_FILE = path.join(DATA_DIR, 'tasks.json');
|
||
|
||
const STAGES = ['login', 'searching', 'coarse_filter', 'fine_filter'];
|
||
|
||
export class TaskManager {
|
||
constructor(broadcast) {
|
||
this.tasks = new Map();
|
||
this.broadcast = broadcast;
|
||
this._loadFromDisk();
|
||
this._startScheduler();
|
||
}
|
||
|
||
_buildTaskConfig(config = {}) {
|
||
const existingSchedule = config.schedule || {};
|
||
const scheduleEnabled = config.scheduleEnabled ?? existingSchedule.enabled ?? false;
|
||
const scheduleIntervalMinutes = config.scheduleIntervalMinutes ?? existingSchedule.intervalMinutes ?? 0;
|
||
return {
|
||
queries: Array.isArray(config.queries) ? config.queries.map(q => String(q).trim()).filter(Boolean) : [],
|
||
priceMin: config.priceMin === '' ? null : (config.priceMin ?? null),
|
||
priceMax: config.priceMax === '' ? null : (config.priceMax ?? null),
|
||
region: config.region || '',
|
||
personalSeller: !!config.personalSeller,
|
||
customRequirements: config.customRequirements || '',
|
||
coarsePrompt: config.coarsePrompt || '',
|
||
finePrompt: config.finePrompt || '',
|
||
maxPages: Math.max(1, Math.min(20, Number(config.maxPages || 3))),
|
||
maxItems: Math.max(1, Math.min(500, Number(config.maxItems || 60))),
|
||
autoAnalyze: config.autoAnalyze !== false,
|
||
autoNotify: config.autoNotify !== false,
|
||
schedule: {
|
||
enabled: !!scheduleEnabled,
|
||
intervalMinutes: Math.max(0, Number(scheduleIntervalMinutes || 0) || 0),
|
||
nextRunAt: config.scheduleNextRunAt ?? existingSchedule.nextRunAt ?? null,
|
||
lastRunAt: config.scheduleLastRunAt ?? existingSchedule.lastRunAt ?? null,
|
||
},
|
||
};
|
||
}
|
||
|
||
createTask(config) {
|
||
const id = genId();
|
||
const task = {
|
||
id,
|
||
name: config.name || `任务-${id}`,
|
||
config: this._buildTaskConfig(config),
|
||
stage: 'pending',
|
||
running: false,
|
||
stopped: false,
|
||
products: { raw: [], coarseFiltered: [], fineFiltered: [] },
|
||
hermesAnalysis: null,
|
||
chatSessions: [],
|
||
chatSessionsFull: [],
|
||
logs: [],
|
||
createdAt: Date.now(),
|
||
updatedAt: Date.now(),
|
||
};
|
||
this._normalizeSchedule(task);
|
||
|
||
this.tasks.set(id, task);
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:created', this._serialize(task));
|
||
return task;
|
||
}
|
||
|
||
duplicateTask(sourceId) {
|
||
const source = this.tasks.get(sourceId);
|
||
if (!source) return null;
|
||
|
||
const id = genId();
|
||
const task = {
|
||
id,
|
||
name: `${source.name} (副本)`,
|
||
config: JSON.parse(JSON.stringify(source.config || {})),
|
||
stage: 'pending',
|
||
running: false,
|
||
stopped: false,
|
||
products: { raw: [], coarseFiltered: [], fineFiltered: [] },
|
||
hermesAnalysis: null,
|
||
chatSessions: [],
|
||
chatSessionsFull: [],
|
||
logs: [],
|
||
createdAt: Date.now(),
|
||
updatedAt: Date.now(),
|
||
};
|
||
this._normalizeSchedule(task);
|
||
|
||
this.tasks.set(id, task);
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:created', this._serialize(task));
|
||
return task;
|
||
}
|
||
|
||
updateTask(id, patch = {}) {
|
||
const task = this.tasks.get(id);
|
||
if (!task) return null;
|
||
if (task.running) return null;
|
||
|
||
task.name = patch.name || task.name;
|
||
task.config = { ...task.config, ...this._buildTaskConfig({ ...task.config, ...patch }) };
|
||
task.updatedAt = Date.now();
|
||
this._normalizeSchedule(task);
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:updated', this._serialize(task));
|
||
return task;
|
||
}
|
||
|
||
_resetTaskProducts(task) {
|
||
task.products = { raw: [], coarseFiltered: [], fineFiltered: [] };
|
||
task.hermesAnalysis = null;
|
||
task.chatSessions = [];
|
||
task.chatSessionsFull = [];
|
||
task.stage = 'pending';
|
||
task.stopped = false;
|
||
}
|
||
|
||
async startTask(id, configUpdates = {}) {
|
||
const task = this.tasks.get(id);
|
||
if (!task || task.running) return null;
|
||
|
||
if (Object.keys(configUpdates || {}).length) {
|
||
const { reset, scheduled, ...patch } = configUpdates;
|
||
if (Object.keys(patch).length) task.config = { ...task.config, ...this._buildTaskConfig({ ...task.config, ...patch }) };
|
||
if (reset) this._resetTaskProducts(task);
|
||
if (scheduled) {
|
||
task.config.schedule.lastRunAt = Date.now();
|
||
this._normalizeSchedule(task, true);
|
||
}
|
||
}
|
||
|
||
task.running = true;
|
||
task.stopped = false;
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:started', { id });
|
||
|
||
const emitLog = (msg) => {
|
||
const entry = { time: timestamp(), message: msg };
|
||
task.logs.push(entry);
|
||
this._emit(id, 'task:log', entry);
|
||
};
|
||
|
||
const shouldStop = () => task.stopped;
|
||
|
||
try {
|
||
await this._runPipeline(task, emitLog, shouldStop);
|
||
if (!task.stopped) {
|
||
await this._runAutoAnalysis(task, emitLog);
|
||
task.stage = 'completed';
|
||
}
|
||
} catch (err) {
|
||
emitLog(`❌ 任务异常: ${err.message}`);
|
||
} finally {
|
||
task.running = false;
|
||
if (task.stopped) {
|
||
task.stage = 'stopped';
|
||
}
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:finished', this._serialize(task));
|
||
}
|
||
}
|
||
|
||
stopTask(id) {
|
||
const task = this.tasks.get(id);
|
||
if (!task) return;
|
||
task.stopped = true;
|
||
task.stage = 'stopping';
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:stopping', { id });
|
||
}
|
||
|
||
deleteTask(id) {
|
||
const task = this.tasks.get(id);
|
||
if (task) {
|
||
task.stopped = true;
|
||
this.tasks.delete(id);
|
||
this._persistToDisk();
|
||
}
|
||
}
|
||
|
||
clearTaskCache(id, { mode = 'all', minScore = 75 } = {}) {
|
||
const task = this.tasks.get(id);
|
||
if (!task || task.running) return null;
|
||
|
||
const before = {
|
||
raw: task.products?.raw?.length || 0,
|
||
coarseFiltered: task.products?.coarseFiltered?.length || 0,
|
||
fineFiltered: task.products?.fineFiltered?.length || 0,
|
||
hermesRecommended: task.hermesAnalysis?.recommendedCount || 0,
|
||
};
|
||
|
||
let kept = [];
|
||
const keepMode = mode === 'hermes-recommend' || mode === 'hermes-pass' ? mode : 'all';
|
||
if (keepMode !== 'all') {
|
||
const fine = task.products?.fineFiltered || [];
|
||
kept = fine.filter(item => {
|
||
const analysis = item.hermesAnalysis;
|
||
if (!analysis) return false;
|
||
if (keepMode === 'hermes-recommend') return analysis.verdict === 'recommend';
|
||
return (analysis.verdict === 'recommend' || analysis.verdict === 'caution') && Number(analysis.score || 0) >= Number(minScore || 0);
|
||
});
|
||
}
|
||
|
||
task.products = keepMode === 'all'
|
||
? { raw: [], coarseFiltered: [], fineFiltered: [] }
|
||
: { raw: kept, coarseFiltered: kept, fineFiltered: kept };
|
||
task.hermesAnalysis = null;
|
||
task.chatSessions = [];
|
||
task.chatSessionsFull = [];
|
||
task.stage = keepMode === 'all' ? 'pending' : (kept.length ? 'completed' : 'pending');
|
||
task.stopped = false;
|
||
task.updatedAt = Date.now();
|
||
task.logs.push({
|
||
time: timestamp(),
|
||
message: keepMode === 'all'
|
||
? `🧹 已清除本任务缓存：采集 ${before.raw} / 粗筛 ${before.coarseFiltered} / 细筛 ${before.fineFiltered}`
|
||
: `🧹 已清理缓存并保留 ${kept.length} 个优质结果（模式：${keepMode}，最低分：${minScore}）`,
|
||
});
|
||
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:updated', this._serialize(task));
|
||
return {
|
||
ok: true,
|
||
mode: keepMode,
|
||
minScore: Number(minScore || 0),
|
||
before,
|
||
after: {
|
||
raw: task.products.raw.length,
|
||
coarseFiltered: task.products.coarseFiltered.length,
|
||
fineFiltered: task.products.fineFiltered.length,
|
||
},
|
||
kept: kept.length,
|
||
task: this._serialize(task),
|
||
};
|
||
}
|
||
|
||
getTask(id) {
|
||
const task = this.tasks.get(id);
|
||
return task ? this._serialize(task) : null;
|
||
}
|
||
|
||
getTaskFull(id) {
|
||
return this.tasks.get(id) || null;
|
||
}
|
||
|
||
analyzeTask(id, options = {}) {
|
||
const task = this.tasks.get(id);
|
||
if (!task) return null;
|
||
const analysis = buildAnalysis(task, options);
|
||
this.setTaskAnalysis(id, analysis);
|
||
return analysis;
|
||
}
|
||
|
||
setTaskAnalysis(id, analysis) {
|
||
const task = this.tasks.get(id);
|
||
if (!task || !analysis) return null;
|
||
const map = new Map((analysis.items || []).map(item => [item.id, item]));
|
||
task.products.fineFiltered = (task.products.fineFiltered || []).map(item => map.get(item.id) || item);
|
||
task.hermesAnalysis = analysis.summary;
|
||
task.updatedAt = Date.now();
|
||
this._persistToDisk();
|
||
this._emit(id, 'task:updated', this._serialize(task));
|
||
return task;
|
||
}
|
||
|
||
async _runAutoAnalysis(task, emitLog) {
|
||
if (task.config?.autoAnalyze === false) {
|
||
emitLog('🦐 已关闭自动 Hermes 分析：可在详情页手动执行分析');
|
||
return null;
|
||
}
|
||
if (!task.products?.fineFiltered?.length) {
|
||
emitLog('🦐 没有细筛候选，跳过自动 Hermes 分析');
|
||
return null;
|
||
}
|
||
|
||
emitLog(`🦐 自动执行 Hermes 分析：${task.products.fineFiltered.length} 个候选`);
|
||
const analysis = this.analyzeTask(task.id, { limit: Number(task.config.maxItems || 50) });
|
||
if (!analysis?.summary) {
|
||
emitLog('⚠️ Hermes 分析未生成结果');
|
||
return null;
|
||
}
|
||
|
||
try {
|
||
const report = renderAnalysisMarkdown(this.getTaskFull(task.id), analysis.summary);
|
||
analysis.summary.reportPath = report.path;
|
||
this.setTaskAnalysis(task.id, analysis);
|
||
emitLog(`📝 Hermes 分析报告已生成：${report.path}`);
|
||
} catch (err) {
|
||
emitLog(`⚠️ Hermes 报告生成失败：${err.message}`);
|
||
}
|
||
|
||
if (task.config?.autoNotify !== false) {
|
||
try {
|
||
const notification = await sendNotifications(this.getTaskFull(task.id), analysis.summary);
|
||
if (notification?.skipped) {
|
||
emitLog(`🔕 自动通知跳过：${notification.reason || '未达到通知条件'}`);
|
||
} else if (notification?.ok) {
|
||
emitLog('📣 自动通知已发送');
|
||
} else {
|
||
const errors = (notification?.results || []).filter(r => !r.ok).map(r => `${r.name || r.id}: ${r.error}`).join('；');
|
||
emitLog(`⚠️ 自动通知部分失败：${errors || '未知错误'}`);
|
||
}
|
||
} catch (err) {
|
||
emitLog(`⚠️ 自动通知失败：${err.message}`);
|
||
}
|
||
} else {
|
||
emitLog('🔕 已关闭自动通知');
|
||
}
|
||
return analysis;
|
||
}
|
||
|
||
getAllTasks() {
|
||
return [...this.tasks.values()].map(t => this._serialize(t));
|
||
}
|
||
|
||
_resolveResumeStage(task) {
|
||
if (task.products.fineFiltered.length > 0) return 'fine_filter';
|
||
if (task.products.coarseFiltered.length > 0) return 'fine_filter';
|
||
if (task.products.raw.length > 0) return 'coarse_filter';
|
||
return null;
|
||
}
|
||
|
||
async _runPipeline(task, emitLog, shouldStop) {
|
||
const resumeStage = this._resolveResumeStage(task);
|
||
const stageOrder = ['login', 'searching', 'coarse_filter', 'fine_filter'];
|
||
const resumeIdx = resumeStage ? stageOrder.indexOf(resumeStage) : -1;
|
||
|
||
if (resumeStage) {
|
||
emitLog(`♻️ 检测到历史数据，从「${resumeStage}」阶段恢复`);
|
||
emitLog(` 已有: ${task.products.raw.length}采集 / ${task.products.coarseFiltered.length}粗筛 / ${task.products.fineFiltered.length}候选`);
|
||
}
|
||
|
||
// Stage 1: Login (always)
|
||
task.stage = 'login';
|
||
this._emitStage(task);
|
||
emitLog('========== 阶段1: 登录验证 ==========');
|
||
|
||
const loggedIn = await checkLogin(emitLog);
|
||
if (!loggedIn || shouldStop()) return;
|
||
|
||
const filterRequirements = this._buildFilterRequirements(task.config);
|
||
|
||
// Stage 2: Search
|
||
if (resumeIdx >= stageOrder.indexOf('searching')) {
|
||
emitLog(`⏭️ 跳过搜索采集（已有 ${task.products.raw.length} 个商品）`);
|
||
} else {
|
||
task.stage = 'searching';
|
||
this._emitStage(task);
|
||
emitLog('========== 阶段2: 搜索采集 ==========');
|
||
|
||
const filters = {
|
||
priceMin: task.config.priceMin,
|
||
priceMax: task.config.priceMax,
|
||
region: task.config.region,
|
||
personalSeller: task.config.personalSeller,
|
||
};
|
||
|
||
for (const query of task.config.queries) {
|
||
if (shouldStop()) break;
|
||
const products = await searchProducts(query, filters, task.config.maxPages, emitLog, shouldStop);
|
||
const deduped = products.filter(p => !task.products.raw.some(e => e.id === p.id));
|
||
const remaining = Math.max(0, (task.config.maxItems || 60) - task.products.raw.length);
|
||
const accepted = deduped.slice(0, remaining);
|
||
task.products.raw.push(...accepted);
|
||
this._emitProducts(task);
|
||
if (task.products.raw.length >= (task.config.maxItems || 60)) {
|
||
emitLog(`📦 已达到单次最大商品寻找量 ${task.config.maxItems || 60}，停止继续搜索`);
|
||
break;
|
||
}
|
||
}
|
||
|
||
emitLog(`📊 总计采集 ${task.products.raw.length} 个商品（去重后）`);
|
||
this._persistToDisk();
|
||
}
|
||
if (shouldStop() || task.products.raw.length === 0) return;
|
||
|
||
// Stage 3: Coarse Filter
|
||
if (resumeIdx >= stageOrder.indexOf('coarse_filter')) {
|
||
emitLog(`⏭️ 跳过粗筛（已有 ${task.products.coarseFiltered.length} 个通过）`);
|
||
} else {
|
||
task.stage = 'coarse_filter';
|
||
this._emitStage(task);
|
||
emitLog('========== 阶段3: 粗筛（标题筛选） ==========');
|
||
|
||
const coarseRequirements = this._buildCoarseRequirements(task.config);
|
||
task.products.coarseFiltered = await coarseFilter(
|
||
task.products.raw, coarseRequirements, emitLog, shouldStop, task.config.coarsePrompt
|
||
);
|
||
this._emitProducts(task);
|
||
this._persistToDisk();
|
||
}
|
||
if (shouldStop() || task.products.coarseFiltered.length === 0) return;
|
||
|
||
// Stage 4: Fine Filter
|
||
if (resumeIdx >= stageOrder.indexOf('fine_filter')) {
|
||
emitLog(`⏭️ 跳过细筛（已有 ${task.products.fineFiltered.length} 个通过）`);
|
||
} else {
|
||
task.stage = 'fine_filter';
|
||
this._emitStage(task);
|
||
emitLog('========== 阶段4: 细筛（详情页筛选） ==========');
|
||
|
||
task.products.fineFiltered = await fineFilter(
|
||
task.products.coarseFiltered.slice(0, task.config.maxItems || 60), filterRequirements, emitLog, shouldStop, task.config.finePrompt
|
||
);
|
||
this._emitProducts(task);
|
||
this._persistToDisk();
|
||
}
|
||
if (shouldStop()) return;
|
||
|
||
task.chatSessions = [];
|
||
task.chatSessionsFull = [];
|
||
this._emit(task.id, 'task:chats', task.chatSessions);
|
||
this._persistToDisk();
|
||
emitLog(`🦐 采集和本地预筛完成：保留 ${task.products.fineFiltered.length} 个候选，准备自动执行 Hermes 分析`);
|
||
emitLog('========== 采集流程结束 ==========');
|
||
}
|
||
|
||
_buildCoarseRequirements(config) {
|
||
const parts = [];
|
||
if (config.queries.length > 0) {
|
||
parts.push(`目标商品: ${config.queries.join(', ')}`);
|
||
}
|
||
if (config.customRequirements) {
|
||
parts.push(`补充需求: ${config.customRequirements}`);
|
||
}
|
||
return parts.join('\n');
|
||
}
|
||
|
||
_buildFilterRequirements(config) {
|
||
const parts = [];
|
||
if (config.queries.length > 0) {
|
||
parts.push(`搜索关键词: ${config.queries.join(', ')}`);
|
||
}
|
||
if (config.priceMin != null || config.priceMax != null) {
|
||
parts.push(`价格范围: ${config.priceMin ?? '不限'} - ${config.priceMax ?? '不限'}`);
|
||
}
|
||
if (config.region) {
|
||
parts.push(`地区偏好: ${config.region}`);
|
||
}
|
||
if (config.personalSeller) {
|
||
parts.push('只要个人卖家');
|
||
}
|
||
if (config.customRequirements) {
|
||
parts.push(`商品需求: ${config.customRequirements}`);
|
||
}
|
||
return parts.join('\n');
|
||
}
|
||
|
||
_serialize(task) {
|
||
return {
|
||
id: task.id,
|
||
name: task.name,
|
||
config: task.config,
|
||
stage: task.stage,
|
||
running: task.running,
|
||
stopped: task.stopped,
|
||
products: {
|
||
rawCount: task.products.raw.length,
|
||
coarseCount: task.products.coarseFiltered.length,
|
||
fineCount: task.products.fineFiltered.length,
|
||
raw: task.products.raw.slice(0, 100),
|
||
coarseFiltered: task.products.coarseFiltered.slice(0, 50),
|
||
fineFiltered: task.products.fineFiltered,
|
||
},
|
||
chatSessions: task.chatSessions || [],
|
||
hermesAnalysis: task.hermesAnalysis || null,
|
||
logs: task.logs.slice(-200),
|
||
createdAt: task.createdAt,
|
||
updatedAt: task.updatedAt,
|
||
};
|
||
}
|
||
|
||
_serializeFull(task) {
|
||
return {
|
||
id: task.id,
|
||
name: task.name,
|
||
config: task.config,
|
||
stage: task.stage,
|
||
products: {
|
||
raw: task.products.raw,
|
||
coarseFiltered: task.products.coarseFiltered,
|
||
fineFiltered: task.products.fineFiltered,
|
||
},
|
||
chatSessions: task.chatSessions || [],
|
||
chatSessionsFull: task.chatSessionsFull || [],
|
||
hermesAnalysis: task.hermesAnalysis || null,
|
||
logs: task.logs.slice(-500),
|
||
createdAt: task.createdAt,
|
||
updatedAt: task.updatedAt,
|
||
};
|
||
}
|
||
|
||
_persistToDisk() {
|
||
try {
|
||
fs.mkdirSync(DATA_DIR, { recursive: true });
|
||
const data = [...this.tasks.values()].map(t => this._serializeFull(t));
|
||
fs.writeFileSync(TASKS_FILE, JSON.stringify(data, null, 2), 'utf-8');
|
||
} catch (err) {
|
||
console.error('持久化任务失败:', err.message);
|
||
}
|
||
}
|
||
|
||
_loadFromDisk() {
|
||
try {
|
||
if (!fs.existsSync(TASKS_FILE)) return;
|
||
const raw = fs.readFileSync(TASKS_FILE, 'utf-8');
|
||
const data = JSON.parse(raw);
|
||
for (const t of data) {
|
||
let chatSessionsFull = t.chatSessionsFull || [];
|
||
|
||
if (chatSessionsFull.length === 0 && (t.chatSessions || []).length > 0) {
|
||
const fineMap = new Map((t.products?.fineFiltered || []).map(p => [p.id, p]));
|
||
chatSessionsFull = t.chatSessions.map(s => {
|
||
const product = fineMap.get(s.id) || {
|
||
id: s.id,
|
||
title: s.productTitle || '',
|
||
price: s.productPrice || '',
|
||
description: s.productDescription || '',
|
||
sellerName: s.sellerName || '',
|
||
chatUrl: '',
|
||
};
|
||
const messages = (s.messages || []).map(m => ({
|
||
...m,
|
||
fingerprint: `${m.role}:${m.content?.trim()}`,
|
||
}));
|
||
const chatHistory = messages.map(m => ({
|
||
role: m.role === 'self' ? 'assistant' : 'user',
|
||
content: m.content,
|
||
}));
|
||
return {
|
||
id: s.id,
|
||
product,
|
||
chatContext: null,
|
||
status: s.status || 'waiting',
|
||
goalReason: s.goalReason || '',
|
||
messages,
|
||
chatHistory,
|
||
lastChecked: s.lastChecked || Date.now(),
|
||
backoffLevel: 0,
|
||
};
|
||
});
|
||
console.log(` 🔄 任务 ${t.id}: 从旧格式迁移了 ${chatSessionsFull.length} 个聊天会话`);
|
||
}
|
||
|
||
const task = {
|
||
id: t.id,
|
||
name: t.name,
|
||
config: this._buildTaskConfig(t.config || {}),
|
||
stage: (t.stage === 'completed' || t.stage === 'stopped') ? t.stage : 'stopped',
|
||
running: false,
|
||
stopped: true,
|
||
products: {
|
||
raw: t.products?.raw || [],
|
||
coarseFiltered: t.products?.coarseFiltered || [],
|
||
fineFiltered: t.products?.fineFiltered || [],
|
||
},
|
||
chatSessions: t.chatSessions || [],
|
||
chatSessionsFull,
|
||
hermesAnalysis: t.hermesAnalysis || null,
|
||
logs: t.logs || [],
|
||
createdAt: t.createdAt || Date.now(),
|
||
updatedAt: t.updatedAt || t.createdAt || Date.now(),
|
||
};
|
||
this._normalizeSchedule(task);
|
||
this.tasks.set(task.id, task);
|
||
}
|
||
console.log(`📂 从磁盘恢复了 ${data.length} 个任务`);
|
||
} catch (err) {
|
||
console.error('加载持久化任务失败:', err.message);
|
||
}
|
||
}
|
||
|
||
_normalizeSchedule(task, afterRun = false) {
|
||
if (!task.config.schedule) {
|
||
task.config.schedule = { enabled: false, intervalMinutes: 0, nextRunAt: null, lastRunAt: null };
|
||
}
|
||
const s = task.config.schedule;
|
||
s.intervalMinutes = Number(s.intervalMinutes || 0) || 0;
|
||
s.enabled = !!s.enabled && s.intervalMinutes > 0;
|
||
if (!s.enabled) {
|
||
s.nextRunAt = null;
|
||
return;
|
||
}
|
||
const base = afterRun ? Date.now() : (Number(s.nextRunAt) || Date.now());
|
||
if (!s.nextRunAt || Number(s.nextRunAt) <= Date.now()) {
|
||
s.nextRunAt = base + s.intervalMinutes * 60 * 1000;
|
||
}
|
||
}
|
||
|
||
_startScheduler() {
|
||
if (process.env.XIANHU_DISABLE_SCHEDULER === '1') return;
|
||
this.schedulerTimer = setInterval(() => this._tickSchedules(), 30000);
|
||
}
|
||
|
||
_tickSchedules() {
|
||
const now = Date.now();
|
||
for (const task of this.tasks.values()) {
|
||
const sched = task.config?.schedule;
|
||
if (!sched?.enabled || task.running || !sched.nextRunAt || sched.nextRunAt > now) continue;
|
||
task.logs.push({ time: timestamp(), message: `⏰ 定时任务触发，开始后台采集` });
|
||
this.startTask(task.id, { reset: true, scheduled: true });
|
||
}
|
||
}
|
||
|
||
_emit(taskId, event, data) {
|
||
this.broadcast({ event, taskId, data });
|
||
}
|
||
|
||
_emitStage(task) {
|
||
this._emit(task.id, 'task:stage', { id: task.id, stage: task.stage });
|
||
}
|
||
|
||
_emitProducts(task) {
|
||
this._emit(task.id, 'task:products', {
|
||
id: task.id,
|
||
rawCount: task.products.raw.length,
|
||
coarseCount: task.products.coarseFiltered.length,
|
||
fineCount: task.products.fineFiltered.length,
|
||
});
|
||
}
|
||
}
|