koc-loop/app/api/partner-batch-workbook/route.ts

import { getRuntimeEnv } from "../../../lib/runtime-env";
import { runInBackground } from "../../../lib/background";
import type { DatabaseStatement } from "../../../lib/database";
import {
  downloadFeishuMedia,
  type FeishuBindings,
} from "../../../lib/feishu-client";
import {
  ensureSchema,
  getRawDb,
  getUploadBucket,
  hashText,
  uid,
} from "../../../lib/mvp-db";
import {
  PARTNER_BATCH_MAX_BYTES,
  buildPartnerBatchWorkbookColumns,
  parsePartnerBatchWorkbook,
  resolvePartnerWorkbookOrigin,
} from "../../../lib/partner-batch-workbook";
import {
  accountFromPublishLink,
} from "../../../lib/partner-utils";
import { extractPublishUrl } from "../../../lib/publish-url";
import {
  partnerOptions,
  withPartnerCors,
} from "../../../lib/partner-cors";
import {
  buildRecoveryWorkbook,
  type RecoveryWorkbookImage,
  type RecoveryWorkbookRow,
} from "../../../lib/recovery-workbook";
import {
  resolveCollectionMcpConfig,
  type CollectionMcpBindings,
} from "../../../lib/mcp-collection-client";
import { enrichDistributionAccount } from "../../../lib/account-enrichment-service";
import {
  createCollectionRunTasks,
  parseCollectionDays,
  runDueScheduledCollections,
} from "../../../lib/collection-service";
import { normalizeWorkbookImage } from "../../../lib/workbook-image";

const env = getRuntimeEnv();

type StoredAsset = {
  index: number;
  key: string;
  fileToken?: string;
  width?: number | null;
  height?: number | null;
};

type BatchRow = {
  distribution_id: string;
  title: string;
  body: string;
  source_row: number | null;
  image_assets: string;
  video_assets: string;
  publish_url: string | null;
  publish_screenshot_key: string | null;
  screenshot_key: string | null;
  partner_id: string;
  account_id: string | null;
  claimant_name: string;
};

type TaskRow = {
  id: string;
  name: string;
  task_type: string;
  collection_start_date: string | null;
  collection_days: string;
  platform: "小红书" | "抖音";
  content_format: "image_text" | "video";
};

type BatchAccess = {
  task: TaskRow;
  rows: BatchRow[];
};

function textValue(value: string | null, maxLength = 100) {
  return String(value ?? "").trim().slice(0, maxLength);
}

function safeFileName(value: string) {
  return value.replace(/[\\/:*?"<>|]/g, " ").trim().slice(0, 60) || "领取笔记";
}

function exactArrayBuffer(bytes: Uint8Array) {
  const copy = new Uint8Array(bytes.byteLength);
  copy.set(bytes);
  return copy.buffer;
}

function contentTypeFromObject(object: { writeHttpMetadata(headers: Headers): void }) {
  const headers = new Headers();
  object.writeHttpMetadata(headers);
  return headers.get("Content-Type") || "application/octet-stream";
}

function parseAssets(
  value: string,
  prefixes = ["content-assets/", "task-assets/"],
) {
  try {
    const assets = JSON.parse(value || "[]") as StoredAsset[];
    return Array.isArray(assets)
      ? assets
          .filter(
            (asset) =>
              Number.isInteger(Number(asset.index)) &&
              Number(asset.index) > 0 &&
              typeof asset.key === "string" &&
              prefixes.some((prefix) => asset.key.startsWith(prefix)),
          )
          .map((asset) => ({ ...asset, index: Number(asset.index) }))
          .sort((left, right) => left.index - right.index)
      : [];
  } catch {
    return [];
  }
}

async function loadImage(
  key: string,
  description: string,
  fileToken?: string,
  width?: number | null,
  height?: number | null,
  compactSource = false,
) {
  const bucket = getUploadBucket();
  let object = await bucket.get(key);
  if (!object && fileToken) {
    const media = await downloadFeishuMedia(
      fileToken,
      env as unknown as FeishuBindings,
    );
    await bucket.put(key, media.bytes, {
      httpMetadata: { contentType: media.contentType },
      customMetadata: { source: "feishu-api" },
    });
    object = await bucket.get(key);
  }
  if (!object) return null;
  return normalizeWorkbookImage(
    {
      bytes: new Uint8Array(await object.arrayBuffer()),
      contentType: contentTypeFromObject(object),
      description,
      width,
      height,
    } satisfies RecoveryWorkbookImage,
    compactSource
      ? { maxDimension: 1600, outputFormat: "jpeg", jpegQuality: 82 }
      : undefined,
  );
}

async function findAccess(
  taskToken: string,
  claimToken: string,
  delegationToken: string,
): Promise<BatchAccess | null> {
  const db = getRawDb();
  const task = delegationToken
    ? await db
        .prepare(
          `SELECT t.id, t.name, t.task_type, t.collection_start_date, t.collection_days,
                  t.platform, t.content_format
           FROM tasks t
           JOIN delegation_bundles b ON b.task_id = t.id
           WHERE b.share_token = ? AND b.status = 'active'`,
        )
        .bind(delegationToken)
        .first<TaskRow>()
    : await db
        .prepare(
          `SELECT t.id, t.name, t.task_type, t.collection_start_date, t.collection_days,
                  t.platform, t.content_format
           FROM tasks t
           JOIN claims cl ON cl.task_id = t.id
           WHERE t.share_token = ? AND cl.claim_token = ?`,
        )
        .bind(taskToken, claimToken)
        .first<TaskRow>();
  if (!task) return null;
  const rows = delegationToken
    ? await db
        .prepare(
          `SELECT d.id AS distribution_id, d.partner_id, d.account_id,
                  d.publish_url, d.publish_screenshot_key, d.screenshot_key,
                  c.title, c.body, c.source_row, c.image_assets, c.video_assets,
                  cl.claimant_name
           FROM distributions d
           JOIN contents c ON c.id = d.content_id
           JOIN delegation_bundles b ON b.id = d.delegation_bundle_id
           JOIN claims cl ON cl.id = d.claim_id
           WHERE b.share_token = ? AND b.task_id = ? AND b.status = 'active'
           ORDER BY d.claimed_at, d.id`,
        )
        .bind(delegationToken, task.id)
        .all<BatchRow>()
    : await db
        .prepare(
          `SELECT d.id AS distribution_id, d.partner_id, d.account_id,
                  d.publish_url, d.publish_screenshot_key, d.screenshot_key,
                  c.title, c.body, c.source_row, c.image_assets, c.video_assets,
                  cl.claimant_name
           FROM distributions d
           JOIN contents c ON c.id = d.content_id
           JOIN claims cl ON cl.id = d.claim_id
           WHERE cl.claim_token = ? AND cl.task_id = ?
           ORDER BY d.claimed_at, d.id`,
        )
        .bind(claimToken, task.id)
        .all<BatchRow>();
  return { task, rows: rows.results };
}

async function handleGet(request: Request) {
  try {
    await ensureSchema();
    const url = new URL(request.url);
    const taskToken = textValue(url.searchParams.get("task"));
    const claimToken = textValue(url.searchParams.get("claim"));
    const delegationToken = textValue(url.searchParams.get("share"));
    if (!delegationToken && (!taskToken || !claimToken)) {
      return Response.json({ error: "领取凭证不完整" }, { status: 400 });
    }
    const access = await findAccess(taskToken, claimToken, delegationToken);
    if (!access) return Response.json({ error: "领取凭证无效" }, { status: 403 });
    if (access.task.task_type !== "content_publish") {
      return Response.json({ error: "当前任务不支持笔记批量回填" }, { status: 400 });
    }
    const maxSourceImages = Math.max(
      0,
      ...access.rows.map((row) => parseAssets(row.image_assets).length),
    );
    const maxSourceVideos = Math.max(
      0,
      ...access.rows.map(
        (row) => parseAssets(row.video_assets, ["content-videos/"]).length,
      ),
    );
    const columns = buildPartnerBatchWorkbookColumns({
      contentFormat: access.task.content_format,
      maxSourceImages,
      maxSourceVideos,
    });
    const downloadOrigin = resolvePartnerWorkbookOrigin(
      request,
      env.APP_ORIGIN,
    );
    const workbookRows: RecoveryWorkbookRow[] = [];
    for (let rowIndex = 0; rowIndex < access.rows.length; rowIndex += 1) {
      const row = access.rows[rowIndex];
      const images: RecoveryWorkbookRow["images"] = [];
      const hyperlinks: NonNullable<RecoveryWorkbookRow["hyperlinks"]> = [];
      const assets =
        columns.sourceImageCount > 0 ? parseAssets(row.image_assets) : [];
      for (let index = 0; index < assets.length; index += 1) {
        const asset = assets[index];
        const image = await loadImage(
          asset.key,
          `${row.title} 原图${asset.index}`,
          asset.fileToken,
          asset.width,
          asset.height,
          true,
        );
        if (image) {
          images.push({
            column: columns.sourceImageStartColumn + index,
            image,
            maxWidth: 160,
            maxHeight: 118,
          });
        }
      }
      if (row.publish_screenshot_key) {
        const image = await loadImage(
          row.publish_screenshot_key,
          `${row.title} 笔记截图`,
        );
        if (image) {
          images.push({ column: columns.publishScreenshotColumn, image });
        }
      }
      if (row.screenshot_key) {
        const image = await loadImage(
          row.screenshot_key,
          `${row.title} 数据分析截图`,
        );
        if (image) {
          images.push({ column: columns.creatorScreenshotColumn, image });
        }
      }
      const videoAssets = parseAssets(row.video_assets, ["content-videos/"]);
      for (let index = 0; index < videoAssets.length; index += 1) {
        const params = new URLSearchParams({
          distribution: row.distribution_id,
          index: String(videoAssets[index].index),
          kind: "video",
          download: "1",
        });
        if (delegationToken) params.set("share", delegationToken);
        else {
          params.set("task", taskToken);
          params.set("claim", claimToken);
        }
        hyperlinks.push({
          column: columns.sourceVideoStartColumn + index,
          url: `${downloadOrigin}/api/partner-image?${params}`,
        });
      }
      workbookRows.push({
        cells: [
          rowIndex + 1,
          row.title,
          row.body,
          ...Array.from({ length: columns.sourceImageCount }, () => ""),
          ...Array.from(
            { length: columns.sourceVideoCount },
            (_, index) => (index < videoAssets.length ? `下载视频${index + 1}` : ""),
          ),
          row.publish_url || "",
          "",
          "",
          row.distribution_id,
          row.publish_screenshot_key || "",
          row.screenshot_key || "",
        ],
        images,
        hyperlinks: [
          ...hyperlinks,
          ...(row.publish_url
            ? [{ column: columns.publishUrlColumn, url: row.publish_url }]
            : []),
        ],
      });
    }
    const workbook = buildRecoveryWorkbook({
      sheetName: "批量回填",
      headers: columns.headers,
      columnWidths: columns.columnWidths,
      rows: workbookRows,
      hiddenColumns: [
        columns.systemColumn,
        columns.systemColumn + 1,
        columns.systemColumn + 2,
      ],
    });
    const fileName = `${safeFileName(access.task.name)}-批量回填.xlsx`;
    return new Response(exactArrayBuffer(workbook), {
      headers: {
        "Cache-Control": "private, no-store",
        "Content-Type":
          "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        "Content-Disposition": `attachment; filename="koc-batch.xlsx"; filename*=UTF-8''${encodeURIComponent(fileName)}`,
        "X-Content-Type-Options": "nosniff",
      },
    });
  } catch (error) {
    return Response.json(
      { error: error instanceof Error ? error.message : "导出失败" },
      { status: 500 },
    );
  }
}

function extensionForImage(image: { fileName: string; contentType: string }) {
  const fromName = image.fileName.split(".").at(-1)?.replace(/[^a-zA-Z0-9]/g, "");
  if (fromName) return fromName.toLowerCase() === "jpeg" ? "jpg" : fromName;
  return image.contentType.includes("png")
    ? "png"
    : image.contentType.includes("webp")
      ? "webp"
      : image.contentType.includes("gif")
        ? "gif"
        : "jpg";
}

async function storeImportedImage(
  kind: "publish" | "creator",
  distributionId: string,
  image: { bytes: Uint8Array; contentType: string; fileName: string },
) {
  const prefix = kind === "publish" ? "publish-evidence" : "creator-center";
  const key = `${prefix}/${distributionId}/${uid("sheet")}.${extensionForImage(image)}`;
  await getUploadBucket().put(key, image.bytes, {
    httpMetadata: { contentType: image.contentType },
    customMetadata: { source: "partner-batch-workbook" },
  });
  return key;
}

async function isDifferentFromStoredImage(
  existingKey: string | null,
  image: { bytes: Uint8Array },
) {
  if (!existingKey) return true;
  const stored = await getUploadBucket().get(existingKey);
  if (!stored) return true;
  const storedBytes = new Uint8Array(await stored.arrayBuffer());
  if (storedBytes.byteLength !== image.bytes.byteLength) return true;
  for (let index = 0; index < storedBytes.byteLength; index += 1) {
    if (storedBytes[index] !== image.bytes[index]) return true;
  }
  return false;
}

async function handlePost(request: Request) {
  try {
    await ensureSchema();
    const form = await request.formData();
    const file = form.get("file");
    if (!(file instanceof File) || file.size === 0) {
      return Response.json({ error: "请选择填写完成的Excel表" }, { status: 400 });
    }
    if (file.size > PARTNER_BATCH_MAX_BYTES) {
      return Response.json({ error: "批量回填表不能超过80MB" }, { status: 400 });
    }
    if (!/\.xlsx$/i.test(file.name)) {
      return Response.json({ error: "仅支持系统导出的 .xlsx 表格" }, { status: 400 });
    }
    const taskToken = textValue(String(form.get("taskToken") ?? ""));
    const claimToken = textValue(String(form.get("claimToken") ?? ""));
    const delegationToken = textValue(String(form.get("delegationToken") ?? ""));
    const access = await findAccess(taskToken, claimToken, delegationToken);
    if (!access) return Response.json({ error: "领取凭证无效" }, { status: 403 });
    if (access.task.task_type !== "content_publish") {
      return Response.json({ error: "当前任务不支持笔记批量回填" }, { status: 400 });
    }
    const importedRows = parsePartnerBatchWorkbook(await file.arrayBuffer());
    const assignmentById = new Map(
      access.rows.map((row) => [row.distribution_id, row]),
    );
    const seen = new Set<string>();
    const errors: string[] = [];
    const prepared = importedRows.map((row) => {
      const assignment = assignmentById.get(row.distributionId);
      if (!assignment) {
        errors.push(`第${row.spreadsheetRow}行不属于当前领取批次，请重新导出表格`);
      } else if (seen.has(row.distributionId)) {
        errors.push(`第${row.spreadsheetRow}行笔记重复`);
      } else if (row.title && row.title !== assignment.title) {
        errors.push(`第${row.spreadsheetRow}行标题已被修改，请重新导出表格`);
      }
      seen.add(row.distributionId);
      const publishUrl = row.publishUrl
        ? extractPublishUrl(row.publishUrl, access.task.platform)
        : assignment?.publish_url || "";
      if (row.publishUrl && !publishUrl) {
        errors.push(`第${row.spreadsheetRow}行发布链接不是有效的${access.task.platform}作品链接`);
      }
      const hasPublishScreenshot = Boolean(
        assignment?.publish_screenshot_key || row.publishScreenshot,
      );
      if (publishUrl && !hasPublishScreenshot) {
        errors.push(`第${row.spreadsheetRow}行填写了发布链接，请同时插入笔记截图`);
      }
      if (row.publishScreenshot && !publishUrl) {
        errors.push(`第${row.spreadsheetRow}行插入了笔记截图，请同时填写发布链接`);
      }
      if (row.creatorScreenshot && !publishUrl) {
        errors.push(`第${row.spreadsheetRow}行需先回填发布链接，再补数据分析截图`);
      }
      return {
        imported: row,
        assignment,
        publishUrl,
      };
    });
    if (errors.length > 0) {
      return Response.json(
        { error: errors.slice(0, 8).join("；"), errors },
        { status: 400 },
      );
    }

    const db = getRawDb();
    let updatedRows = 0;
    let publishedCount = 0;
    let analysisScreenshotCount = 0;
    let noteScreenshotCount = 0;
    let publishUrlChangedCount = 0;
    const enrichments: Array<{ id: string; url: string; nickname: string }> = [];
    for (const item of prepared) {
      const assignment = item.assignment!;
      let publishScreenshotKey = assignment.publish_screenshot_key;
      let creatorScreenshotKey = assignment.screenshot_key;
      const hasNewPublishScreenshot = item.imported.publishScreenshot
        ? await isDifferentFromStoredImage(
            assignment.publish_screenshot_key,
            item.imported.publishScreenshot,
          )
        : false;
      const hasNewCreatorScreenshot = item.imported.creatorScreenshot
        ? await isDifferentFromStoredImage(
            assignment.screenshot_key,
            item.imported.creatorScreenshot,
          )
        : false;
      if (hasNewPublishScreenshot && item.imported.publishScreenshot) {
        publishScreenshotKey = await storeImportedImage(
          "publish",
          assignment.distribution_id,
          item.imported.publishScreenshot,
        );
        noteScreenshotCount += 1;
      }
      if (hasNewCreatorScreenshot && item.imported.creatorScreenshot) {
        creatorScreenshotKey = await storeImportedImage(
          "creator",
          assignment.distribution_id,
          item.imported.creatorScreenshot,
        );
        analysisScreenshotCount += 1;
      }
      const statements: DatabaseStatement[] = [];
      if (publishScreenshotKey !== assignment.publish_screenshot_key) {
        statements.push(
          db
            .prepare(
              `UPDATE distributions SET publish_screenshot_key = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
            )
            .bind(publishScreenshotKey, assignment.distribution_id),
        );
      }
      if (creatorScreenshotKey !== assignment.screenshot_key) {
        statements.push(
          db
            .prepare(
              `UPDATE distributions SET screenshot_key = ?,
                 ocr_status = CASE WHEN exposure IS NOT NULL AND views IS NOT NULL THEN ocr_status ELSE 'uploaded' END,
                 updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
            )
            .bind(creatorScreenshotKey, assignment.distribution_id),
        );
      }
      if (item.publishUrl && item.publishUrl !== assignment.publish_url) {
        const isReplacement = Boolean(assignment.publish_url);
        const account = accountFromPublishLink(
          item.publishUrl,
          access.task.platform,
        );
        if (!account) {
          return Response.json(
            { error: `第${item.imported.spreadsheetRow}行发布链接格式不正确` },
            { status: 400 },
          );
        }
        const matchedAccount = await db
          .prepare(
            `SELECT id FROM accounts
             WHERE platform = ? AND platform_uid = ?
             LIMIT 1`,
          )
          .bind(account.platform, account.platformUid)
          .first<{ id: string }>();
        const accountId =
          matchedAccount?.id ||
          `account-${hashText(`${account.platform}:${account.platformUid}`)}`;
        statements.push(
          ...(isReplacement
            ? [
                db
                  .prepare(
                    `UPDATE distributions SET
                      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 = 'pending',
                      collection_status_description = '批量回填已更新链接，等待重新采集',
                      collection_updated_at = NULL, last_collection_day = NULL,
                      updated_at = CURRENT_TIMESTAMP
                     WHERE id = ?`,
                  )
                  .bind(assignment.distribution_id),
                db
                  .prepare("DELETE FROM collection_runs WHERE distribution_id = ?")
                  .bind(assignment.distribution_id),
              ]
            : []),
          db
            .prepare(
              `INSERT INTO accounts
                (id, platform, platform_uid, nickname, profile_url,
                 current_contact, post_count)
               VALUES (?, ?, ?, ?, ?, ?, 1)
               ON CONFLICT(platform, platform_uid) DO UPDATE SET
                 nickname = excluded.nickname,
                 profile_url = excluded.profile_url,
                 current_contact = CASE
                   WHEN excluded.current_contact != ''
                     THEN excluded.current_contact
                   ELSE accounts.current_contact
                 END,
                 last_seen_at = CURRENT_TIMESTAMP`,
            )
            .bind(
              accountId,
              account.platform,
              account.platformUid,
              account.nickname,
              account.profileUrl,
              assignment.claimant_name,
            ),
          db
            .prepare(
              `UPDATE distributions SET account_id = ?, publish_url = ?,
                 publish_time = COALESCE(publish_time, CURRENT_TIMESTAMP),
                 status = 'published', updated_at = CURRENT_TIMESTAMP
               WHERE id = ?`,
            )
            .bind(accountId, item.publishUrl, assignment.distribution_id),
        );
        statements.push(
          db
            .prepare(
              `UPDATE accounts SET post_count = (
                 SELECT COUNT(*) FROM distributions WHERE account_id = ?
               ) WHERE id = ?`,
            )
            .bind(accountId, accountId),
        );
        if (assignment.account_id && assignment.account_id !== accountId) {
          statements.push(
            db
              .prepare(
                `UPDATE accounts SET post_count = (
                   SELECT COUNT(*) FROM distributions WHERE account_id = ?
                 ) WHERE id = ?`,
              )
              .bind(assignment.account_id, assignment.account_id),
          );
        }
        if (isReplacement) publishUrlChangedCount += 1;
        if (!assignment.publish_url) {
          statements.push(
            db
              .prepare(
                `UPDATE partners SET completed_total = completed_total + 1 WHERE id = ?`,
              )
              .bind(assignment.partner_id),
          );
          publishedCount += 1;
        }
        enrichments.push({
          id: assignment.distribution_id,
          url: item.publishUrl,
          nickname: account.nickname,
        });
      }
      if (statements.length > 0) {
        await db.batch(statements);
        updatedRows += 1;
      }
    }
    if (publishedCount > 0 || publishUrlChangedCount > 0) {
      const collectionDays = parseCollectionDays(
        access.task.collection_days,
      );
      const isScheduled = Boolean(
        access.task.collection_start_date &&
          collectionDays.length > 0,
      );
      await db
        .prepare(
          `UPDATE distributions SET collection_status = 'scheduled',
             collection_status_description = ?
           WHERE task_id = ? AND publish_url IS NOT NULL AND publish_url != ''`,
        )
        .bind(
          isScheduled
            ? `已安排${collectionDays.length}个采集日，每日09:00执行`
            : "默认发布后7天自动采集，每日09:00执行",
          access.task.id,
        )
        .run();
      if (isScheduled && access.task.collection_start_date) {
        await createCollectionRunTasks(
          db,
          access.task.id,
          access.task.collection_start_date,
          collectionDays,
        );
      }
      // 未设置计划时调度器同样兜底：默认自发布日起 7 天自动采集。
      runInBackground(
        runDueScheduledCollections(
          db,
          Date.now(),
          resolveCollectionMcpConfig(
            env as unknown as CollectionMcpBindings,
          ),
          "catchup",
          access.task.id,
        ).catch(() => undefined),
        "collection catchup after batch publish update",
      );
    }
    for (const enrichment of enrichments) {
      runInBackground(
        enrichDistributionAccount(
          db,
          enrichment.id,
          enrichment.url,
          enrichment.nickname,
          resolveCollectionMcpConfig(env as unknown as CollectionMcpBindings),
        ).catch(() => undefined),
        "batch distribution account enrichment",
      );
    }
    return Response.json({
      imported: true,
      updatedRows,
      publishedCount,
      publishUrlChangedCount,
      noteScreenshotCount,
      analysisScreenshotCount,
      skippedRows: importedRows.length - updatedRows,
    });
  } catch (error) {
    return Response.json(
      { error: error instanceof Error ? error.message : "批量回填失败" },
      { status: 500 },
    );
  }
}

export async function GET(request: Request) {
  return withPartnerCors(request, await handleGet(request));
}

export async function POST(request: Request) {
  return withPartnerCors(request, await handlePost(request));
}

export async function OPTIONS(request: Request) {
  return partnerOptions(request);
}