agent-foundation/scripts/local_media_matrix_worker.js

const douyin = require('./local_douyin_worker');
const xiaohongshu = require('./local_xiaohongshu_worker');

const DEFAULT_BASE_URL = process.env.MEDIA_BASE_URL || 'http://localhost:8880';
const DEFAULT_WORKER_ID = process.env.MEDIA_WORKER_ID || 'local-media-matrix-worker';
const DEFAULT_WORKER_TOKEN = process.env.MEDIA_WORKER_TOKEN || 'local-worker-dev-token';
const DEFAULT_DOUYIN_ACCOUNT_ID = process.env.DOUYIN_ACCOUNT_ID || 'douyin_company';
const DEFAULT_XHS_ACCOUNT_ID = process.env.XHS_ACCOUNT_ID || 'xiaohongshu_company';

const PLATFORM_HANDLERS = {
  douyin: {
    accountId: DEFAULT_DOUYIN_ACCOUNT_ID,
    collectMetrics: douyin.collectMetrics,
    checkLogin: douyin.checkLogin
  },
  xiaohongshu: {
    accountId: DEFAULT_XHS_ACCOUNT_ID,
    collectMetrics: xiaohongshu.collectMetrics,
    checkLogin: xiaohongshu.checkLogin
  }
};

function normalizePlatform(platform) {
  const value = String(platform || '').trim().toLowerCase();
  if (value === 'xhs' || value === 'rednote') {
    return 'xiaohongshu';
  }
  return value;
}

function requestedPlatforms(task) {
  const payload = task.payload || {};
  const raw = Array.isArray(payload.platforms) && payload.platforms.length > 0
      ? payload.platforms
      : ['douyin', 'xiaohongshu'];
  return raw.map(normalizePlatform).filter(platform => PLATFORM_HANDLERS[platform]);
}

function accountIdFor(task, platform) {
  const payload = task.payload || {};
  const accountIds = payload.accountIds || {};
  if (accountIds && typeof accountIds === 'object' && accountIds[platform]) {
    return accountIds[platform];
  }
  return PLATFORM_HANDLERS[platform].accountId;
}

function payloadForPlatform(task, platform) {
  const payload = task.payload || {};
  const platformPayloads = payload.platformPayloads || {};
  return {
    ...payload,
    ...(platformPayloads[platform] || {}),
    platforms: undefined,
    accountIds: undefined,
    platformPayloads: undefined
  };
}

async function pollTasks() {
  const response = await fetch(`${DEFAULT_BASE_URL}/v1/media/workers/poll`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Worker-Token': DEFAULT_WORKER_TOKEN
    },
    body: JSON.stringify({
      workerId: DEFAULT_WORKER_ID,
      platforms: ['multi', 'matrix', 'all'],
      maxTasks: 3
    })
  });
  if (!response.ok) {
    throw new Error(`Poll failed: ${response.status} ${await response.text()}`);
  }
  return response.json();
}

async function submitResult(taskId, body) {
  const response = await fetch(`${DEFAULT_BASE_URL}/v1/media/workers/tasks/${taskId}/result`, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'X-Worker-Token': DEFAULT_WORKER_TOKEN
    },
    body: JSON.stringify(body)
  });
  if (!response.ok) {
    throw new Error(`Submit result failed: ${response.status} ${await response.text()}`);
  }
  return response.json();
}

async function handleTask(task) {
  const platforms = requestedPlatforms(task);
  if (platforms.length === 0) {
    return submitResult(task.id, {
      status: 'FAILED',
      errorMessage: 'No supported platforms requested.',
      result: { action: 'sync_metrics', platform: 'multi', platformResults: {} }
    });
  }

  const platformResults = {};
  const videos = [];
  const metrics = emptyMetrics();
  const errors = [];

  for (const platform of platforms) {
    const handler = PLATFORM_HANDLERS[platform];
    const platformTask = {
      ...task,
      platform,
      accountId: accountIdFor(task, platform),
      payload: payloadForPlatform(task, platform)
    };
    try {
      const response = await handler.collectMetrics(platformTask);
      const result = response.result || {};
      platformResults[platform] = {
        ...result,
        platform,
        accountId: platformTask.accountId,
        status: response.status
      };
      addMetrics(metrics, result.metrics || {});
      for (const video of result.videos || []) {
        videos.push({ ...video, platform, accountId: platformTask.accountId });
      }
      if (response.status !== 'SUCCESS') {
        errors.push(`${platform}: ${response.errorMessage || 'failed'}`);
      }
    } catch (error) {
      platformResults[platform] = {
        platform,
        accountId: platformTask.accountId,
        status: 'FAILED',
        errorMessage: error.message
      };
      errors.push(`${platform}: ${error.message}`);
    }
  }

  return submitResult(task.id, {
    status: errors.length === platforms.length ? 'FAILED' : 'SUCCESS',
    errorMessage: errors.length > 0 ? errors.join('; ') : null,
    result: {
      action: 'sync_metrics',
      platform: 'multi',
      platforms,
      metrics,
      videos,
      videoCount: videos.length,
      platformResults,
      extractionMode: 'multi_platform'
    }
  });
}

function emptyMetrics() {
  return { views: 0, likes: 0, comments: 0, collects: 0, shares: 0 };
}

function addMetrics(target, source) {
  target.views += Number(source.views || 0);
  target.likes += Number(source.likes || 0);
  target.comments += Number(source.comments || 0);
  target.collects += Number(source.collects || 0);
  target.shares += Number(source.shares || 0);
}

async function workerOnce() {
  const tasks = await pollTasks();
  if (!Array.isArray(tasks) || tasks.length === 0) {
    console.log('No pending media matrix tasks.');
    return;
  }
  for (const task of tasks) {
    const result = await handleTask(task);
    console.log(`Completed ${task.id}: ${result.status}`);
  }
}

function sleep(ms) {
  return new Promise(resolve => setTimeout(resolve, ms));
}

async function workerLoop() {
  const intervalMs = Number(process.env.MEDIA_WORKER_INTERVAL_MS || 30000);
  console.log(`Media matrix worker loop started. baseUrl=${DEFAULT_BASE_URL}, intervalMs=${intervalMs}`);
  while (true) {
    try {
      await workerOnce();
    } catch (error) {
      console.error(error.stack || error.message);
    }
    await sleep(intervalMs);
  }
}

async function main() {
  const mode = process.argv[2];
  if (mode === 'worker') {
    await workerOnce();
    return;
  }
  if (mode === 'worker-loop') {
    await workerLoop();
    return;
  }
  throw new Error('Usage: node scripts/local_media_matrix_worker.js <worker|worker-loop>');
}

main().catch(error => {
  console.error(error.stack || error.message);
  process.exit(1);
});