portrait-queue-worker.mjs 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. // 静态 import 而非 createRequire + path.join:打包器不跟踪运行时拼接的路径,编译后会找不到模块。
  2. import portraitTags from '../../../mcp/src/tools/qiwei-portrait-tags-run.js';
  3. const { processPortraitQueue } = portraitTags;
  4. export class PortraitQueueWorker {
  5. constructor({ intervalMs = 60_000, limit = 5, onState = () => {}, processor = processPortraitQueue } = {}) {
  6. this.intervalMs = intervalMs;
  7. this.limit = limit;
  8. this.onState = onState;
  9. this.processor = processor;
  10. this.running = false;
  11. this.timer = null;
  12. this.inFlight = null;
  13. }
  14. start() {
  15. if (this.running) return;
  16. this.running = true;
  17. this.onState({ portraitQueue: { status: 'running', lastError: '' } });
  18. void this.runOnce();
  19. this.timer = setInterval(() => void this.runOnce(), this.intervalMs);
  20. this.timer.unref?.();
  21. }
  22. async runOnce() {
  23. if (!this.running) return { skipped: true, reason: 'stopped' };
  24. if (this.inFlight) return { skipped: true, reason: 'in-flight' };
  25. this.inFlight = (async () => {
  26. try {
  27. const result = await this.processor(this.limit);
  28. this.onState({
  29. portraitQueue: {
  30. status: 'running',
  31. lastRunAt: new Date().toISOString(),
  32. processed: Number(result?.processed || 0),
  33. remaining: Number(result?.remaining || 0),
  34. lastError: '',
  35. },
  36. });
  37. return result;
  38. } catch (error) {
  39. this.onState({
  40. portraitQueue: {
  41. status: 'error',
  42. lastRunAt: new Date().toISOString(),
  43. lastError: error.message,
  44. },
  45. });
  46. return { processed: 0, error: error.message };
  47. } finally {
  48. this.inFlight = null;
  49. }
  50. })();
  51. return this.inFlight;
  52. }
  53. async stop() {
  54. this.running = false;
  55. if (this.timer) clearInterval(this.timer);
  56. this.timer = null;
  57. if (this.inFlight) await this.inFlight;
  58. this.onState({ portraitQueue: { status: 'stopped', stoppedAt: new Date().toISOString() } });
  59. }
  60. }