koc-loop/lib/collection-service.ts

import {
  collectMetricsFromMcp,
  type CollectionMcpConfig,
} from "./mcp-collection-client";
import { uid } from "./mvp-db";
import type { DatabaseClient, DatabaseStatement } from "./database";
import { refreshTaskStatus } from "./task-service";

type DistributionForCollection = {
  id: string;
  task_id: string;
  publish_url: string | null;
  ocr_status: string;
  platform: string;
};

type ScheduledTask = {
  id: string;
  collection_start_date: string;
  collection_days: string;
};

type CollectionSource = "automatic" | "catchup" | "manual";

function userFacingCollectionError(error: unknown) {
  const message = error instanceof Error ? error.message : String(error ?? "");
  const normalized = message.toLowerCase();
  if (
    /cookie|登录|登陆|授权|access.?token|未登录|未授权|401|403/.test(
      normalized,
    )
  ) {
    return "采集 Cookie 已过期或无权限";
  }
  if (
    /链接|link|url|404|not found|不存在|删除|失效|无法识别.*笔记/.test(
      normalized,
    )
  ) {
    return "笔记链接失效或不可访问";
  }
  if (
    /超时|timeout|fetch failed|network|502|503|暂时不可用|响应空/.test(
      normalized,
    )
  ) {
    return "采集服务暂时不可用";
  }
  return "笔记链接失效或采集 Cookie 过期";
}

function utcDay(value: string) {
  const match = value.match(/^(\d{4})-(\d{2})-(\d{2})$/);
  if (!match) return null;
  return Date.UTC(Number(match[1]), Number(match[2]) - 1, Number(match[3]));
}

function dateForScheduleDay(startDate: string, scheduleDay: number) {
  const start = utcDay(startDate);
  if (start === null) return null;
  return new Date(start + (scheduleDay - 1) * 86_400_000)
    .toISOString()
    .slice(0, 10);
}

function parseDays(value: string) {
  try {
    const parsed = JSON.parse(value) as unknown;
    if (!Array.isArray(parsed)) return [];
    return [...new Set(parsed.map(Number))]
      .filter((day) => Number.isInteger(day) && day >= 1 && day <= 7)
      .sort((a, b) => a - b);
  } catch {
    return [];
  }
}

export function shanghaiDateFromTimestamp(timestamp: number) {
  return new Date(timestamp + 8 * 60 * 60 * 1000)
    .toISOString()
    .slice(0, 10);
}

function shanghaiHourFromTimestamp(timestamp: number) {
  return Number(
    new Date(timestamp + 8 * 60 * 60 * 1000)
      .toISOString()
      .slice(11, 13),
  );
}

export function isCollectionScheduleDue(
  scheduledDate: string,
  timestamp: number,
) {
  const currentDate = shanghaiDateFromTimestamp(timestamp);
  const currentHour = shanghaiHourFromTimestamp(timestamp);
  return (
    scheduledDate < currentDate ||
    (scheduledDate === currentDate && currentHour >= 9)
  );
}

function dueSchedules(
  startDate: string,
  days: number[],
  timestamp: number,
) {
  return days
    .map((scheduleDay) => ({
      scheduleDay,
      scheduledDate: dateForScheduleDay(startDate, scheduleDay),
    }))
    .filter(
      (
        value,
      ): value is { scheduleDay: number; scheduledDate: string } =>
        Boolean(
          value.scheduledDate &&
            isCollectionScheduleDue(value.scheduledDate, timestamp),
        ),
    );
}

export async function createCollectionRunTasks(
  db: DatabaseClient,
  taskId: string,
  startDate: string,
  days: number[],
) {
  const normalizedDays = [...new Set(days)]
    .filter((day) => Number.isInteger(day) && day >= 1 && day <= 7)
    .sort((a, b) => a - b);
  const distributions = await db
    .prepare(
      `SELECT id FROM distributions
       WHERE task_id = ?
         AND publish_url IS NOT NULL
         AND publish_url != ''`,
    )
    .bind(taskId)
    .all<{ id: string }>();

  await db
    .prepare(
      `DELETE FROM collection_runs
       WHERE task_id = ?
         AND status = 'pending'`,
    )
    .bind(taskId)
    .run();

  const statements: DatabaseStatement[] = [];
  for (const distribution of distributions.results) {
    for (const scheduleDay of normalizedDays) {
      const scheduledDate = dateForScheduleDay(startDate, scheduleDay);
      if (!scheduledDate) continue;
      statements.push(
        db
          .prepare(
            `INSERT OR IGNORE INTO collection_runs
              (id, task_id, distribution_id, scheduled_date, schedule_day,
               scheduled_at, status, status_description)
             VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)`,
          )
          .bind(
            uid("run"),
            taskId,
            distribution.id,
            scheduledDate,
            scheduleDay,
            `${scheduledDate}T09:00:00+08:00`,
            `等待第${scheduleDay}天 09:00自动采集`,
          ),
      );
    }
  }
  for (let index = 0; index < statements.length; index += 100) {
    await db.batch(statements.slice(index, index + 100));
  }
  return {
    distributions: distributions.results.length,
    days: normalizedDays.length,
    tasks: statements.length,
  };
}

export async function collectDistributionMetrics(
  db: DatabaseClient,
  distributionId: string,
  scheduledDate: string,
  scheduleDay: number | null,
  source: CollectionSource,
  mcpConfig: CollectionMcpConfig,
) {
  const current = await db
    .prepare(
      `SELECT d.*, t.platform
       FROM distributions d
       JOIN tasks t ON t.id = d.task_id
       WHERE d.id = ?`,
    )
    .bind(distributionId)
    .first<DistributionForCollection>();
  if (!current) throw new Error("分发记录不存在");
  if (!current.publish_url) throw new Error("作品尚未回填发布链接");

  const scheduledAt = `${scheduledDate}T09:00:00+08:00`;
  const runId = uid("run");
  await db
    .prepare(
      `INSERT OR IGNORE INTO collection_runs
        (id, task_id, distribution_id, scheduled_date, schedule_day,
         scheduled_at, status, status_description)
       VALUES (?, ?, ?, ?, ?, ?, 'pending', ?)`,
    )
    .bind(
      runId,
      current.task_id,
      distributionId,
      scheduledDate,
      scheduleDay,
      scheduledAt,
      source === "automatic"
        ? "等待自动采集"
        : source === "catchup"
          ? "等待自动追采"
          : "等待手动采集",
    )
    .run();

  const run = await db
    .prepare(
      `SELECT id, status FROM collection_runs
       WHERE distribution_id = ? AND scheduled_date = ?`,
    )
    .bind(distributionId, scheduledDate)
    .first<{ id: string; status: string }>();
  if (!run) throw new Error("采集任务创建失败");
  // Scheduled jobs should remain idempotent, but an operator clicking
  // “立即采集” is explicitly asking for a fresh snapshot. Reusing the same
  // daily run lets us correct stale or previously mis-mapped platform data.
  if (run.status === "success" && source !== "manual") {
    return { skipped: true };
  }

  const collectingDescription =
    source === "automatic"
      ? `第${scheduleDay ?? "—"}天自动采集中`
      : source === "catchup"
        ? `第${scheduleDay ?? "—"}天自动追采中`
        : "正在手动采集";
  await db.batch([
    db
      .prepare(
        `UPDATE collection_runs
         SET schedule_day = COALESCE(schedule_day, ?),
             scheduled_at = ?,
             status = 'collecting',
             status_description = ?,
             started_at = CURRENT_TIMESTAMP
         WHERE id = ?`,
      )
      .bind(scheduleDay, scheduledAt, collectingDescription, run.id),
    db
      .prepare(
        `UPDATE distributions
         SET collection_status = 'collecting',
             collection_status_description = ?,
             updated_at = CURRENT_TIMESTAMP
         WHERE id = ?`,
      )
      .bind(collectingDescription, distributionId),
  ]);

  try {
    const { likes, comments, collects, shares } =
      await collectMetricsFromMcp(
        current.publish_url,
        current.platform === "抖音" ? "抖音" : "小红书",
        mcpConfig,
      );
    const dayWeight = scheduleDay ?? 1;
    const successDescription =
      source === "automatic"
        ? `成功 · 第${dayWeight}天 09:00自动采集`
        : source === "catchup"
          ? `成功 · 第${dayWeight}天自动追采`
          : "成功 · 手动采集";

    await db.batch([
      db
        .prepare(
          `UPDATE collection_runs
           SET status = 'success',
               likes = ?,
               comments = ?,
               collects = ?,
               shares = ?,
               status_description = ?,
               completed_at = CURRENT_TIMESTAMP
           WHERE id = ?`,
        )
        .bind(
          likes,
          comments,
          collects,
          shares,
          successDescription,
          run.id,
        ),
      db
        .prepare(
          `UPDATE distributions
           SET latest_likes = ?,
               latest_comments = ?,
               latest_collects = ?,
               latest_shares = ?,
               collection_status = 'success',
               collection_status_description = ?,
               collection_updated_at = CURRENT_TIMESTAMP,
               last_collection_day = ?,
               status = CASE
                 WHEN exposure IS NOT NULL AND views IS NOT NULL THEN 'complete'
                 ELSE 'collecting'
               END,
               updated_at = CURRENT_TIMESTAMP
           WHERE id = ?`,
        )
        .bind(
          likes,
          comments,
          collects,
          shares,
          successDescription,
          scheduleDay,
          distributionId,
        ),
    ]);
    await refreshTaskStatus(current.task_id, db).catch(() => undefined);
    return { skipped: false, likes, comments, collects, shares };
  } catch (error) {
    const detail = error instanceof Error ? error.message : String(error);
    const message = userFacingCollectionError(error);
    console.error("[KOC LOOP] collection failed", { detail, distributionId });
    await db.batch([
      db
        .prepare(
          `UPDATE collection_runs
           SET status = 'failed',
               status_description = ?,
               completed_at = CURRENT_TIMESTAMP
           WHERE id = ?`,
        )
        .bind(message, run.id),
      db
        .prepare(
          // 已有历史数据（此前采集成功过）的分行失败时不降级为 failed：
          // 保留 success 状态与数据时间，仅在描述里说明本次刷新失败。
          `UPDATE distributions
           SET collection_status = CASE
             WHEN latest_likes IS NULL THEN 'failed'
             ELSE 'success'
           END,
               collection_status_description = CASE
             WHEN latest_likes IS NULL THEN ?
             ELSE ?
           END,
               collection_updated_at = CASE
             WHEN latest_likes IS NULL THEN CURRENT_TIMESTAMP
             ELSE collection_updated_at
           END,
               updated_at = CURRENT_TIMESTAMP
           WHERE id = ?`,
        )
        .bind(message, `${message}，沿用上次采集数据`, distributionId),
    ]);
    throw error;
  }
}

export async function runScheduledCollections(
  db: DatabaseClient,
  scheduledTimestamp: number,
  mcpConfig: CollectionMcpConfig,
) {
  return runDueScheduledCollections(
    db,
    scheduledTimestamp,
    mcpConfig,
    "automatic",
  );
}

export async function runDueScheduledCollections(
  db: DatabaseClient,
  timestamp: number,
  mcpConfig: CollectionMcpConfig,
  source: Extract<CollectionSource, "automatic" | "catchup"> = "catchup",
  onlyTaskId?: string,
) {
  const currentDate = shanghaiDateFromTimestamp(timestamp);
  const tasks = await db
    .prepare(
      `SELECT id, collection_start_date, collection_days
       FROM tasks
       WHERE collection_start_date IS NOT NULL
         AND collection_start_date != ''
         AND collection_days IS NOT NULL
         AND collection_days != '[]'
         AND (? IS NULL OR id = ?)`,
    )
    .bind(onlyTaskId ?? null, onlyTaskId ?? null)
    .all<ScheduledTask>();
  let dueTasks = 0;
  let attempted = 0;
  let succeeded = 0;
  let failed = 0;

  for (const task of tasks.results) {
    const schedules = dueSchedules(
      task.collection_start_date,
      parseDays(task.collection_days),
      timestamp,
    );
    for (const { scheduleDay, scheduledDate } of schedules) {
      dueTasks += 1;
      const distributions = await db
        .prepare(
          `SELECT d.id FROM distributions d
           WHERE d.task_id = ?
             AND d.publish_url IS NOT NULL
             AND d.publish_url != ''
             AND NOT EXISTS (
               SELECT 1 FROM collection_runs r
               WHERE r.distribution_id = d.id
                 AND r.scheduled_date = ?
                 AND (
                   r.status = 'success'
                   OR (
                     r.status = 'collecting'
                     AND r.started_at >= datetime('now', '-30 minutes')
                   )
                   OR (
                     r.status = 'failed'
                     AND r.completed_at >= datetime('now', '-15 minutes')
                   )
                 )
             )`,
        )
        .bind(task.id, scheduledDate)
        .all<{ id: string }>();

      for (const distribution of distributions.results) {
        attempted += 1;
        try {
          await collectDistributionMetrics(
            db,
            distribution.id,
            scheduledDate,
            scheduleDay,
            source,
            mcpConfig,
          );
          succeeded += 1;
        } catch {
          failed += 1;
        }
      }
    }
  }

  return { currentDate, dueTasks, attempted, succeeded, failed };
}

export async function retryFailedCollections(
  db: DatabaseClient,
  taskId: string,
  mcpConfig: CollectionMcpConfig,
) {
  const failedDistributions = await db
    .prepare(
      `SELECT
         d.id,
         COALESCE(
           (
             SELECT r.scheduled_date
             FROM collection_runs r
             WHERE r.distribution_id = d.id
               AND r.status = 'failed'
             ORDER BY COALESCE(r.completed_at, r.created_at) DESC
             LIMIT 1
           ),
           ?
         ) AS scheduled_date,
         (
           SELECT r.schedule_day
           FROM collection_runs r
           WHERE r.distribution_id = d.id
             AND r.status = 'failed'
           ORDER BY COALESCE(r.completed_at, r.created_at) DESC
           LIMIT 1
         ) AS schedule_day
       FROM distributions d
       WHERE d.task_id = ?
         AND d.publish_url IS NOT NULL
         AND d.publish_url != ''
         AND d.collection_status = 'failed'
       ORDER BY COALESCE(d.collection_updated_at, d.updated_at)
       LIMIT 50`,
    )
    .bind(shanghaiDateFromTimestamp(Date.now()), taskId)
    .all<{
      id: string;
      scheduled_date: string;
      schedule_day: number | null;
    }>();

  let succeeded = 0;
  let failed = 0;
  for (const distribution of failedDistributions.results) {
    try {
      await collectDistributionMetrics(
        db,
        distribution.id,
        distribution.scheduled_date,
        distribution.schedule_day,
        "manual",
        mcpConfig,
      );
      succeeded += 1;
    } catch {
      failed += 1;
    }
  }
  return {
    attempted: failedDistributions.results.length,
    succeeded,
    failed,
  };
}