koc-loop/app/api/action/route.ts

import { getRuntimeEnv } from "../../../lib/runtime-env";
import { runInBackground } from "../../../lib/background";

const env = getRuntimeEnv();
import {
  backfillAccountProfiles,
  enrichDistributionAccount,
} from "../../../lib/account-enrichment-service";
import {
  ensureSchema,
  getDashboardData,
  getRawDb,
  uid,
} from "../../../lib/mvp-db";
import {
  collectDistributionMetrics,
  createCollectionRunTasks,
  retryFailedCollections,
  runDueScheduledCollections,
  shanghaiDateFromTimestamp,
} from "../../../lib/collection-service";
import {
  resolveCollectionMcpConfig,
  type CollectionMcpBindings,
} from "../../../lib/mcp-collection-client";
import { adminForbidden, isAdminRequest } from "../../../lib/admin-auth";
import {
  FeishuSourceError,
  readFeishuSource,
  type FeishuBindings,
} from "../../../lib/feishu-client";
import {
  createDistributionTask,
  createScreenshotTask,
} from "../../../lib/task-service";
import {
  DistributionReleaseError,
  releaseUnfinishedDistribution,
} from "../../../lib/distribution-release-service";
import { isManagerRequest } from "../../../lib/user-auth";
import { extractPublishUrl } from "../../../lib/publish-url";

type ActionBody = {
  action?: string;
  [key: string]: unknown;
};

function numberValue(value: unknown, fallback = 0) {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? parsed : fallback;
}

export async function POST(request: Request) {
  if (!(await isAdminRequest(request))) return adminForbidden();
  try {
    await ensureSchema();
    const body = (await request.json()) as ActionBody;
    const db = getRawDb();

    if (body.action === "inspect_feishu") {
      const source = await readFeishuSource(
        String(body.feishuUrl ?? "").trim(),
        env as unknown as FeishuBindings,
      );
      return Response.json({
        sheetId: source.sheetId,
        sheetName: source.sheetName,
        syncedAt: source.syncedAt,
        rowCount: source.rows.length,
        imageCount: source.rows.reduce(
          (total, row) => total + row.images.length,
          0,
        ),
        videoCount: source.rows.reduce(
          (total, row) => total + row.videos.length,
          0,
        ),
        columns: source.columns,
        preview: source.rows.slice(0, 3),
      });
    }

    if (body.action === "create_task") {
      const name = String(body.name ?? "").trim();
      const brand = String(body.brand ?? "").trim();
      const dueAt = String(body.dueAt ?? "").trim();
      const platform = body.platform === "抖音" ? "抖音" : "小红书";
      const contentFormat = body.contentFormat === "video" ? "video" : "image_text";
      if (!name || !brand || !dueAt) {
        return Response.json(
          { error: "请补全任务名称、品牌和截止日期" },
          { status: 400 },
        );
      }
      await createDistributionTask(
        {
          feishuUrl: String(body.feishuUrl ?? "").trim(),
          name,
          brand,
          dueAt,
          platform,
          contentFormat,
        },
        env as unknown as FeishuBindings,
      );
    } else if (body.action === "create_screenshot_task") {
      await createScreenshotTask({
        name: String(body.name ?? "").trim(),
        brand: String(body.brand ?? "").trim(),
        dueAt: String(body.dueAt ?? "").trim(),
        keyword: String(body.keyword ?? "").trim(),
        instructions: String(body.instructions ?? "").trim(),
        quantity: numberValue(body.quantity),
        exampleImageKey: String(body.exampleImageKey ?? "").trim(),
      });
    } else if (body.action === "claim") {
      const partnerId = String(body.partnerId ?? "");
      const taskId = String(body.taskId ?? "");
      const quantity = Math.max(1, Math.min(50, numberValue(body.quantity, 1)));
      const available = await db
        .prepare(
          `SELECT id FROM contents
           WHERE task_id = ? AND status = 'available'
           ORDER BY created_at, id
           LIMIT ${quantity}`,
        )
        .bind(taskId)
        .all<{ id: string }>();
      if (available.results.length === 0) {
        return Response.json({ error: "当前没有可领取内容" }, { status: 409 });
      }
      const statements = available.results.flatMap((content) => [
        db
          .prepare(
            `INSERT INTO distributions
              (id, task_id, content_id, partner_id, status)
             VALUES (?, ?, ?, ?, 'claimed')`,
          )
          .bind(uid("dist"), taskId, content.id, partnerId),
        db
          .prepare("UPDATE contents SET status = 'allocated' WHERE id = ?")
          .bind(content.id),
      ]);
      statements.push(
        db
          .prepare(
            `UPDATE tasks SET claimed_quantity = claimed_quantity + ?
             WHERE id = ?`,
          )
          .bind(available.results.length, taskId),
        db
          .prepare(
            `UPDATE partners SET claimed_total = claimed_total + ?
             WHERE id = ?`,
          )
          .bind(available.results.length, partnerId),
      );
      await db.batch(statements);
    } else if (body.action === "release_distribution") {
      if (!(await isManagerRequest(request))) return adminForbidden();
      await releaseUnfinishedDistribution(
        db,
        String(body.distributionId ?? "").trim(),
      );
    } else if (body.action === "update_distribution_publish_url") {
      if (!(await isManagerRequest(request))) return adminForbidden();
      const distributionId = String(body.distributionId ?? "").trim();
      if (!distributionId) {
        return Response.json(
          { error: "作品记录不存在" },
          { status: 400 },
        );
      }
      const current = await db
        .prepare(
          `SELECT d.id, d.task_id, d.partner_id, d.publish_url,
                  t.task_type, t.platform, t.collection_start_date, t.collection_days,
                  COALESCE(a.nickname, '待识别账号') AS account_nickname
           FROM distributions d
           JOIN tasks t ON t.id = d.task_id
           LEFT JOIN accounts a ON a.id = d.account_id
           WHERE d.id = ?`,
        )
        .bind(distributionId)
        .first<{
          id: string;
          task_id: string;
          partner_id: string;
          publish_url: string | null;
          task_type?: string | null;
          platform: string;
          collection_start_date: string | null;
          collection_days: string;
          account_nickname: string;
        }>();
      if (!current) {
        return Response.json({ error: "作品记录不存在" }, { status: 404 });
      }
      if (current.task_type === "screenshot_collect") {
        return Response.json(
          { error: "截图回收任务不需要填写发布链接" },
          { status: 400 },
        );
      }
      const platform = current.platform === "抖音" ? "抖音" : "小红书";
      const publishUrl = extractPublishUrl(
        String(body.publishUrl ?? "").trim(),
        platform,
      );
      if (!publishUrl) {
        return Response.json(
          { error: `请填写包含${platform}作品链接的发布内容` },
          { status: 400 },
        );
      }
      if (current.publish_url === publishUrl) {
        return Response.json(await getDashboardData());
      }
      let collectionDays: number[] = [];
      try {
        const parsed = JSON.parse(current.collection_days || "[]");
        if (Array.isArray(parsed)) {
          collectionDays = [...new Set(parsed.map(Number))]
            .filter(
              (day) =>
                Number.isInteger(day) && day >= 1 && day <= 7,
            )
            .sort((a, b) => a - b);
        }
      } catch {
        collectionDays = [];
      }
      const isScheduled = Boolean(
        current.collection_start_date && collectionDays.length > 0,
      );
      const statements = [
        db
          .prepare(
            `UPDATE distributions SET
              publish_url = ?,
              publish_time = CURRENT_TIMESTAMP,
              status = 'published',
              d2_likes = NULL,
              d2_comments = NULL,
              d2_collects = NULL,
              d5_likes = NULL,
              d5_comments = NULL,
              d5_collects = NULL,
              d7_likes = NULL,
              d7_comments = NULL,
              d7_collects = NULL,
              latest_likes = NULL,
              latest_comments = NULL,
              latest_collects = NULL,
              latest_shares = NULL,
              collection_status = ?,
              collection_status_description = ?,
              collection_updated_at = NULL,
              last_collection_day = NULL,
              updated_at = CURRENT_TIMESTAMP
             WHERE id = ?`,
          )
          .bind(
            publishUrl,
            isScheduled ? "scheduled" : "pending",
            isScheduled
              ? `管理员已更新链接，等待${collectionDays.length}个采集日`
              : "管理员已更新链接，等待设置采集计划",
            distributionId,
          ),
        db
          .prepare("DELETE FROM collection_runs WHERE distribution_id = ?")
          .bind(distributionId),
      ];
      if (!current.publish_url) {
        statements.push(
          db
            .prepare(
              `UPDATE partners SET completed_total = completed_total + 1
               WHERE id = ?`,
            )
            .bind(current.partner_id),
        );
      }
      await db.batch(statements);
      if (isScheduled && current.collection_start_date) {
        await createCollectionRunTasks(
          db,
          current.task_id,
          current.collection_start_date,
          collectionDays,
        );
        runInBackground(
          runDueScheduledCollections(
            db,
            Date.now(),
            resolveCollectionMcpConfig(
              env as unknown as CollectionMcpBindings,
            ),
            "catchup",
            current.task_id,
          ).catch(() => undefined),
          "collection catchup after publish URL update",
        );
      }
      runInBackground(
        enrichDistributionAccount(
          db,
          distributionId,
          publishUrl,
          current.account_nickname,
          resolveCollectionMcpConfig(
            env as unknown as CollectionMcpBindings,
          ),
        ).catch(() => undefined),
        "account enrichment after publish URL update",
      );
    } else if (body.action === "save_collection_schedule") {
      const taskId = String(body.taskId ?? "").trim();
      const startDate = String(body.startDate ?? "").trim();
      const days = Array.isArray(body.days)
        ? [...new Set(body.days.map(Number))]
            .filter(
              (day) =>
                Number.isInteger(day) && day >= 1 && day <= 7,
            )
            .sort((a, b) => a - b)
        : [];
      if (!taskId || !/^\d{4}-\d{2}-\d{2}$/.test(startDate)) {
        return Response.json(
          { error: "请选择有效的开始采集日期" },
          { status: 400 },
        );
      }
      if (days.length === 0) {
        return Response.json(
          { error: "请至少选择一个自动采集日" },
          { status: 400 },
        );
      }
      const task = await db
        .prepare("SELECT id, task_type FROM tasks WHERE id = ?")
        .bind(taskId)
        .first<{ id: string; task_type?: string | null }>();
      if (!task) {
        return Response.json({ error: "任务不存在" }, { status: 404 });
      }
      if (task.task_type === "screenshot_collect") {
        return Response.json(
          { error: "截图回收任务不需要设置数据采集计划" },
          { status: 400 },
        );
      }
      await db.batch([
        db
          .prepare(
            `UPDATE tasks
             SET collection_start_date = ?,
                 collection_days = ?,
                 collection_schedule_updated_at = CURRENT_TIMESTAMP
             WHERE id = ?`,
          )
          .bind(startDate, JSON.stringify(days), taskId),
        db
          .prepare(
            `UPDATE distributions
             SET collection_status = CASE
                   WHEN latest_likes IS NULL THEN 'scheduled'
                   ELSE collection_status
                 END,
                 collection_status_description = CASE
                   WHEN latest_likes IS NULL THEN ?
                   ELSE collection_status_description
                 END,
                 updated_at = CURRENT_TIMESTAMP
             WHERE task_id = ?
               AND publish_url IS NOT NULL
               AND publish_url != ''`,
          )
          .bind(
            `已安排${days.length}个采集日，每日09:00执行`,
            taskId,
          ),
      ]);
      await createCollectionRunTasks(db, taskId, startDate, days);
      const catchup = runDueScheduledCollections(
        db,
        Date.now(),
        resolveCollectionMcpConfig(
          env as unknown as CollectionMcpBindings,
        ),
        "catchup",
        taskId,
      ).catch(() => undefined);
      runInBackground(catchup, "collection catchup");
    } else if (body.action === "run_due_collections") {
      await runDueScheduledCollections(
        db,
        Date.now(),
        resolveCollectionMcpConfig(
          env as unknown as CollectionMcpBindings,
        ),
        "catchup",
      );
    } else if (
      body.action === "collect_now" ||
      body.action === "collect"
    ) {
      const distributionId = String(body.distributionId ?? "");
      const rawDay =
        body.day === null || body.day === undefined
          ? null
          : numberValue(body.day);
      const day =
        rawDay !== null && Number.isInteger(rawDay) && rawDay >= 1 && rawDay <= 7
          ? rawDay
          : null;
      if (body.action === "collect" && day === null) {
        return Response.json({ error: "采集周期无效" }, { status: 400 });
      }
      const distribution = await db
        .prepare(
          `SELECT t.task_type
           FROM distributions d
           JOIN tasks t ON t.id = d.task_id
           WHERE d.id = ?`,
        )
        .bind(distributionId)
        .first<{ task_type?: string | null }>();
      if (!distribution) {
        return Response.json({ error: "笔记记录不存在" }, { status: 404 });
      }
      if (distribution.task_type === "screenshot_collect") {
        return Response.json(
          { error: "截图回收任务不支持公开数据采集" },
          { status: 400 },
        );
      }
      await collectDistributionMetrics(
        db,
        distributionId,
        shanghaiDateFromTimestamp(Date.now()),
        day,
        "manual",
        resolveCollectionMcpConfig(
          env as unknown as CollectionMcpBindings,
        ),
      );
    } else if (body.action === "retry_failed_collections") {
      const taskId = String(body.taskId ?? "").trim();
      if (!taskId) {
        return Response.json({ error: "请选择需要补采的任务" }, { status: 400 });
      }
      const task = await db
        .prepare("SELECT task_type FROM tasks WHERE id = ?")
        .bind(taskId)
        .first<{ task_type?: string | null }>();
      if (!task) {
        return Response.json({ error: "任务不存在" }, { status: 404 });
      }
      if (task.task_type === "screenshot_collect") {
        return Response.json(
          { error: "截图回收任务没有需要补采的公开数据" },
          { status: 400 },
        );
      }
      await retryFailedCollections(
        db,
        taskId,
        resolveCollectionMcpConfig(
          env as unknown as CollectionMcpBindings,
        ),
      );
    } else if (body.action === "backfill_account_profiles") {
      const backfill = backfillAccountProfiles(
        db,
        resolveCollectionMcpConfig(
          env as unknown as CollectionMcpBindings,
        ),
        Math.max(1, Math.min(25, numberValue(body.limit, 10))),
      ).catch(() => undefined);
      runInBackground(backfill, "account profile backfill");
    } else if (body.action === "set_public_account_ids") {
      const items = Array.isArray(body.items) ? body.items.slice(0, 50) : [];
      const normalized = items
        .map((item) => {
          const record =
            item && typeof item === "object"
              ? (item as Record<string, unknown>)
              : {};
          return {
            accountId: String(record.accountId ?? "").trim().slice(0, 80),
            publicAccountId: String(record.publicAccountId ?? "")
              .trim()
              .slice(0, 80),
          };
        })
        .filter(
          (item) =>
            item.accountId &&
            /^[\p{L}\p{N}._-]{2,80}$/u.test(item.publicAccountId),
        );
      if (normalized.length === 0) {
        return Response.json(
          { error: "没有可回填的账号号值" },
          { status: 400 },
        );
      }
      await db.batch(
        normalized.map((item) =>
          db
            .prepare(
              `UPDATE accounts
               SET public_account_id = ?,
                   last_seen_at = CURRENT_TIMESTAMP
               WHERE id = ?`,
            )
            .bind(item.publicAccountId, item.accountId),
        ),
      );
    } else if (body.action === "manual_metrics") {
      const distributionId = String(body.distributionId ?? "");
      const exposure = Math.max(0, numberValue(body.exposure));
      const views = Math.max(0, numberValue(body.views));
      await db
        .prepare(
          `UPDATE distributions SET
            exposure = ?,
            views = ?,
            ocr_status = 'manual',
            status = CASE WHEN d7_likes IS NOT NULL THEN 'complete' ELSE status END,
            updated_at = CURRENT_TIMESTAMP
           WHERE id = ?`,
        )
        .bind(exposure, views, distributionId)
        .run();
    } else {
      return Response.json({ error: "不支持的操作" }, { status: 400 });
    }

    return Response.json(await getDashboardData());
  } catch (error) {
    return Response.json(
      { error: error instanceof Error ? error.message : "操作失败" },
      {
        status:
          error instanceof FeishuSourceError ||
          error instanceof DistributionReleaseError
            ? error.status
            : 500,
      },
    );
  }
}