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 '); } main().catch(error => { console.error(error.stack || error.message); process.exit(1); });