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, 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 { 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() : 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(); 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() : 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(); 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 = []; 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(); 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) && access.task.collection_start_date && access.task.collection_days !== "[]" ) { let collectionDays: number[] = []; try { const parsed = JSON.parse(access.task.collection_days); if (Array.isArray(parsed)) collectionDays = parsed.map(Number); } catch { collectionDays = []; } if (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( `已安排${collectionDays.length}个采集日,每日09:00执行`, access.task.id, ) .run(); await createCollectionRunTasks( db, access.task.id, access.task.collection_start_date, collectionDays, ); 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); }