import { env } from "cloudflare:workers"; import { getRequestExecutionContext } from "vinext/shims/request-context"; import { enrichDistributionAccount } from "../../../lib/account-enrichment-service"; import { createCollectionRunTasks } from "../../../lib/collection-service"; import { resolveCollectionMcpConfig, type CollectionMcpBindings, } from "../../../lib/mcp-collection-client"; import { ensureSchema, getRawDb, hashText, uid, } from "../../../lib/mvp-db"; import { accountFromPublishLink, extractXhsPublishUrl, } from "../../../lib/partner-utils"; import { parseClaimantIdentifier } from "../../../lib/claimant-identifier"; import { partnerOptions, withPartnerCors, } from "../../../lib/partner-cors"; export const runtime = "edge"; type PartnerBody = { action?: string; taskToken?: string; claimToken?: string; delegationToken?: string; distributionId?: string; distributionIds?: string[]; delegationLabel?: string; delegationBundleId?: string; claimantIdentifier?: string; claimantName?: string; quantity?: number; publishUrl?: string; exposure?: number | string; views?: number | string; }; type ImageAsset = { index: number; width: number | null; height: number | null; }; function textValue(value: unknown, maxLength = 200) { return String(value ?? "").trim().slice(0, maxLength); } function quantityValue(value: unknown) { const parsed = Number(value); return Number.isFinite(parsed) ? Math.max(1, Math.min(50, Math.floor(parsed))) : 1; } function creatorMetricValue(value: unknown) { const normalized = String(value ?? "").trim(); if (!/^\d{1,12}$/.test(normalized)) return null; const parsed = Number(normalized); return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : null; } function idList(value: unknown) { if (!Array.isArray(value)) return []; return [ ...new Set( value .map((item) => textValue(item, 80)) .filter(Boolean) .slice(0, 50), ), ]; } function publicImageAssets(value: unknown): ImageAsset[] { try { const assets = JSON.parse(String(value ?? "[]")) as Array< Partial & { key?: string } >; if (!Array.isArray(assets)) return []; return assets .map((asset) => ({ index: Number(asset.index), width: typeof asset.width === "number" && Number.isFinite(asset.width) ? asset.width : null, height: typeof asset.height === "number" && Number.isFinite(asset.height) ? asset.height : null, })) .filter((asset) => Number.isInteger(asset.index) && asset.index > 0); } catch { return []; } } async function findTask(taskToken: string) { return getRawDb() .prepare( `SELECT id, name, brand, quantity, claimed_quantity, due_at, status FROM tasks WHERE share_token = ?`, ) .bind(taskToken) .first<{ id: string; name: string; brand: string; quantity: number; claimed_quantity: number; due_at: string; status: string; }>(); } async function findDelegationAccess(delegationToken: string) { return getRawDb() .prepare( `SELECT t.id, t.name, t.brand, t.quantity, t.claimed_quantity, t.due_at, t.status, b.id AS bundle_id, b.label AS bundle_label, b.quantity AS bundle_quantity, b.created_at AS bundle_created_at FROM delegation_bundles b JOIN tasks t ON t.id = b.task_id WHERE b.share_token = ? AND b.status = 'active'`, ) .bind(delegationToken) .first<{ id: string; name: string; brand: string; quantity: number; claimed_quantity: number; due_at: string; status: string; bundle_id: string; bundle_label: string; bundle_quantity: number; bundle_created_at: string; }>(); } async function findAccessibleAssignment( taskId: string, distributionId: string, claimToken: string, delegationToken: string, ) { const db = getRawDb(); const select = `SELECT d.id, d.partner_id, d.account_id, d.publish_url, d.publish_screenshot_key, d.screenshot_key`; if (delegationToken) { return db .prepare( `${select} FROM distributions d JOIN delegation_bundles b ON b.id = d.delegation_bundle_id WHERE d.id = ? AND b.share_token = ? AND b.task_id = ? AND b.status = 'active'`, ) .bind(distributionId, delegationToken, taskId) .first<{ id: string; partner_id: string; account_id: string | null; publish_url: string | null; publish_screenshot_key: string | null; screenshot_key: string | null; }>(); } if (!claimToken) return null; return db .prepare( `${select} FROM distributions d JOIN claims c ON c.id = d.claim_id WHERE d.id = ? AND c.claim_token = ? AND c.task_id = ?`, ) .bind(distributionId, claimToken, taskId) .first<{ id: string; partner_id: string; account_id: string | null; publish_url: string | null; publish_screenshot_key: string | null; screenshot_key: string | null; }>(); } async function handleGet(request: Request) { try { await ensureSchema(); const url = new URL(request.url); const taskToken = textValue(url.searchParams.get("task"), 80); const claimToken = textValue(url.searchParams.get("claim"), 80); const delegationToken = textValue(url.searchParams.get("share"), 80); const delegationAccess = delegationToken ? await findDelegationAccess(delegationToken) : null; const task = delegationAccess ?? (await findTask(taskToken)); if (!task) { return Response.json( { error: delegationToken ? "分享链接无效、已撤销或已失效" : "任务链接无效或已失效", }, { status: 404 }, ); } const available = await getRawDb() .prepare( `SELECT COUNT(*) AS count FROM contents WHERE task_id = ? AND status = 'available'`, ) .bind(task.id) .first<{ count: number }>(); let claim: Record | null = null; let delegation: Record | null = null; if (claimToken && !delegationToken) { const claimRow = await getRawDb() .prepare( `SELECT id, claimant_name, quantity, created_at FROM claims WHERE claim_token = ? AND task_id = ?`, ) .bind(claimToken, task.id) .first<{ id: string; claimant_name: string; quantity: number; created_at: string; }>(); if (!claimRow) { return Response.json({ error: "领取凭证无效" }, { status: 403 }); } const assignments = await getRawDb() .prepare( `SELECT d.id, d.status, d.publish_url, d.publish_time, d.publish_screenshot_key, d.screenshot_key AS creator_screenshot_key, d.ocr_status, d.exposure, d.views, c.title, c.body, c.source_row, c.image_assets, a.nickname AS account_nickname, b.id AS delegation_bundle_id, b.label AS delegation_label FROM distributions d JOIN contents c ON c.id = d.content_id LEFT JOIN accounts a ON a.id = d.account_id LEFT JOIN delegation_bundles b ON b.id = d.delegation_bundle_id AND b.status = 'active' WHERE d.claim_id = ? ORDER BY d.claimed_at, d.id`, ) .bind(claimRow.id) .all(); const delegations = await getRawDb() .prepare( `SELECT b.id, b.label, b.share_token, b.quantity, b.status, b.created_at, b.revoked_at, SUM(CASE WHEN d.publish_url IS NOT NULL AND d.publish_url != '' THEN 1 ELSE 0 END) AS completed_count, SUM(CASE WHEN d.screenshot_key IS NOT NULL AND d.exposure IS NOT NULL AND d.views IS NOT NULL THEN 1 ELSE 0 END) AS creator_completed_count FROM delegation_bundles b LEFT JOIN distributions d ON d.delegation_bundle_id = b.id WHERE b.claim_id = ? GROUP BY b.id, b.label, b.share_token, b.quantity, b.status, b.created_at, b.revoked_at ORDER BY b.created_at DESC, b.id DESC`, ) .bind(claimRow.id) .all(); claim = { id: claimRow.id, claimantName: claimRow.claimant_name, quantity: claimRow.quantity, createdAt: claimRow.created_at, assignments: assignments.results.map((assignment) => ({ ...assignment, images: publicImageAssets(assignment.image_assets), image_assets: undefined, })), delegations: delegations.results, }; } else if (delegationAccess) { const assignments = await getRawDb() .prepare( `SELECT d.id, d.status, d.publish_url, d.publish_time, d.publish_screenshot_key, d.screenshot_key AS creator_screenshot_key, d.ocr_status, d.exposure, d.views, c.title, c.body, c.source_row, c.image_assets, a.nickname AS account_nickname FROM distributions d JOIN contents c ON c.id = d.content_id LEFT JOIN accounts a ON a.id = d.account_id WHERE d.delegation_bundle_id = ? ORDER BY d.claimed_at, d.id`, ) .bind(delegationAccess.bundle_id) .all(); delegation = { id: delegationAccess.bundle_id, label: "转派发布包", quantity: delegationAccess.bundle_quantity, createdAt: delegationAccess.bundle_created_at, assignments: assignments.results.map((assignment) => ({ ...assignment, images: publicImageAssets(assignment.image_assets), image_assets: undefined, })), }; } return Response.json({ task: delegationAccess ? { name: task.name, brand: task.brand, dueAt: task.due_at, status: task.status, } : { name: task.name, brand: task.brand, quantity: task.quantity, claimedQuantity: task.claimed_quantity, dueAt: task.due_at, status: task.status, availableQuantity: available?.count ?? 0, }, claim, delegation, }); } catch (error) { return Response.json( { error: error instanceof Error ? error.message : "暂时无法打开任务" }, { status: 500 }, ); } } async function handlePost(request: Request) { try { await ensureSchema(); const body = (await request.json()) as PartnerBody; const taskToken = textValue(body.taskToken, 80); const delegationToken = textValue(body.delegationToken, 80); const delegationAccess = delegationToken ? await findDelegationAccess(delegationToken) : null; const task = delegationAccess ?? (await findTask(taskToken)); if (!task) { return Response.json( { error: delegationToken ? "分享链接无效或已撤销" : "任务链接无效" }, { status: 404 }, ); } if ( delegationToken && body.action !== "submit" && body.action !== "submit_creator_metrics" ) { return Response.json( { error: "分享链接只能用于查看和回填包内笔记" }, { status: 403 }, ); } const db = getRawDb(); if (body.action === "recover") { const identifierInput = textValue( body.claimantIdentifier ?? body.claimantName, 64, ); const claimantIdentifier = parseClaimantIdentifier(identifierInput); if (!claimantIdentifier && identifierInput.length < 2) { return Response.json( { error: "请输入领取时使用的正确微信号或手机号" }, { status: 400 }, ); } const legacyPartnerId = `partner-ext-${hashText( `${task.id}:${identifierInput.toLowerCase()}`, )}`; const partnerId = claimantIdentifier ? `partner-ext-${hashText( `claimant:${claimantIdentifier.canonical}`, )}` : legacyPartnerId; const recovered = await db .prepare( `SELECT c.claim_token, c.quantity, c.created_at, COUNT(d.id) AS note_count, SUM(CASE WHEN d.publish_url IS NOT NULL AND d.publish_url != '' THEN 1 ELSE 0 END) AS completed_count, MIN(co.title) AS first_title FROM claims c LEFT JOIN distributions d ON d.claim_id = c.id LEFT JOIN contents co ON co.id = d.content_id WHERE c.task_id = ? AND c.partner_id IN (?, ?) GROUP BY c.id, c.claim_token, c.quantity, c.created_at ORDER BY c.created_at DESC, c.id DESC LIMIT 20`, ) .bind(task.id, partnerId, legacyPartnerId) .all<{ claim_token: string; quantity: number; created_at: string; note_count: number; completed_count: number; first_title: string | null; }>(); if (recovered.results.length === 0) { return Response.json( { error: "没有找到领取记录,请确认微信号或手机号与领取时完全一致" }, { status: 404 }, ); } return Response.json({ claims: recovered.results.map((claim) => ({ claimToken: claim.claim_token, quantity: claim.note_count || claim.quantity, completedCount: claim.completed_count || 0, createdAt: claim.created_at, firstTitle: claim.first_title || "领取的笔记", })), }); } if (body.action === "claim") { if (task.status !== "active") { return Response.json({ error: "任务已结束,无法继续领取" }, { status: 409 }); } const identifierInput = textValue( body.claimantIdentifier ?? body.claimantName, 64, ); const claimantIdentifier = parseClaimantIdentifier(identifierInput); const quantity = quantityValue(body.quantity); if (!claimantIdentifier) { return Response.json( { error: "请输入正确的微信号或手机号" }, { status: 400 }, ); } const available = await db .prepare( `SELECT id FROM contents WHERE task_id = ? AND status = 'available' ORDER BY COALESCE(source_row, 999999), created_at, id LIMIT ?`, ) .bind(task.id, quantity) .all<{ id: string }>(); if (available.results.length === 0) { return Response.json({ error: "当前任务已领完" }, { status: 409 }); } const partnerId = `partner-ext-${hashText( `claimant:${claimantIdentifier.canonical}`, )}`; const claimantName = claimantIdentifier.display; const claimId = uid("claim"); const claimToken = crypto.randomUUID().replaceAll("-", ""); const statements: D1PreparedStatement[] = [ db .prepare( `INSERT INTO partners (id, name, wecom_name, owner, claimed_total, completed_total) VALUES (?, ?, ?, '外部KOC', 0, 0) ON CONFLICT(id) DO UPDATE SET name = excluded.name, wecom_name = excluded.wecom_name`, ) .bind(partnerId, claimantName, claimantName), db .prepare( `INSERT INTO claims (id, task_id, partner_id, claimant_name, claim_token, quantity) VALUES (?, ?, ?, ?, ?, ?)`, ) .bind( claimId, task.id, partnerId, claimantName, claimToken, available.results.length, ), ]; for (const content of available.results) { statements.push( db .prepare( `INSERT INTO distributions (id, task_id, content_id, partner_id, claim_id, status) VALUES (?, ?, ?, ?, ?, 'claimed')`, ) .bind(uid("dist"), task.id, content.id, partnerId, claimId), db .prepare( `UPDATE contents SET status = 'allocated' WHERE id = ? AND status = 'available'`, ) .bind(content.id), ); } statements.push( db .prepare( `UPDATE tasks SET claimed_quantity = claimed_quantity + ? WHERE id = ?`, ) .bind(available.results.length, task.id), db .prepare( `UPDATE partners SET claimed_total = claimed_total + ? WHERE id = ?`, ) .bind(available.results.length, partnerId), ); await db.batch(statements); return Response.json({ claimToken, claimedCount: available.results.length, }); } if (body.action === "create_delegation") { const claimToken = textValue(body.claimToken, 80); const label = textValue(body.delegationLabel, 24); const distributionIds = idList(body.distributionIds); if (!label) { return Response.json( { error: "请填写接收人备注,例如 A 或小王" }, { status: 400 }, ); } if (distributionIds.length === 0) { return Response.json( { error: "请至少选择一篇待发布笔记" }, { status: 400 }, ); } const claimRow = await db .prepare( `SELECT id, partner_id FROM claims WHERE claim_token = ? AND task_id = ?`, ) .bind(claimToken, task.id) .first<{ id: string; partner_id: string }>(); if (!claimRow) { return Response.json({ error: "领取凭证无效" }, { status: 403 }); } const placeholders = distributionIds.map(() => "?").join(", "); const selected = await db .prepare( `SELECT id, publish_url, delegation_bundle_id FROM distributions WHERE claim_id = ? AND id IN (${placeholders})`, ) .bind(claimRow.id, ...distributionIds) .all<{ id: string; publish_url: string | null; delegation_bundle_id: string | null; }>(); if (selected.results.length !== distributionIds.length) { return Response.json( { error: "部分笔记不属于当前领取批次,请刷新后重试" }, { status: 403 }, ); } if (selected.results.some((item) => item.publish_url)) { return Response.json( { error: "已发布的笔记不能再次转派" }, { status: 409 }, ); } if (selected.results.some((item) => item.delegation_bundle_id)) { return Response.json( { error: "部分笔记已经转派,请刷新后重新选择" }, { status: 409 }, ); } const bundleId = uid("delegate"); const shareToken = crypto.randomUUID().replaceAll("-", ""); const statements: D1PreparedStatement[] = [ db .prepare( `INSERT INTO delegation_bundles (id, task_id, claim_id, partner_id, label, share_token, quantity) VALUES (?, ?, ?, ?, ?, ?, ?)`, ) .bind( bundleId, task.id, claimRow.id, claimRow.partner_id, label, shareToken, distributionIds.length, ), ]; for (const distributionId of distributionIds) { statements.push( db .prepare( `UPDATE distributions SET delegation_bundle_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND claim_id = ? AND publish_url IS NULL AND delegation_bundle_id IS NULL`, ) .bind(bundleId, distributionId, claimRow.id), ); } await db.batch(statements); const bound = await db .prepare( `SELECT COUNT(*) AS count FROM distributions WHERE delegation_bundle_id = ?`, ) .bind(bundleId) .first<{ count: number }>(); if ((bound?.count ?? 0) !== distributionIds.length) { await db.batch([ db .prepare( `UPDATE distributions SET delegation_bundle_id = NULL, updated_at = CURRENT_TIMESTAMP WHERE delegation_bundle_id = ?`, ) .bind(bundleId), db .prepare("DELETE FROM delegation_bundles WHERE id = ?") .bind(bundleId), ]); return Response.json( { error: "部分笔记刚刚已被转派,请刷新后重新选择" }, { status: 409 }, ); } return Response.json({ delegation: { id: bundleId, label, shareToken, quantity: distributionIds.length, }, }); } if (body.action === "revoke_delegation") { const claimToken = textValue(body.claimToken, 80); const bundleId = textValue(body.delegationBundleId, 80); const bundle = await db .prepare( `SELECT b.id FROM delegation_bundles b JOIN claims c ON c.id = b.claim_id WHERE b.id = ? AND b.task_id = ? AND b.status = 'active' AND c.claim_token = ?`, ) .bind(bundleId, task.id, claimToken) .first<{ id: string }>(); if (!bundle) { return Response.json( { error: "没有找到可撤销的转派记录" }, { status: 404 }, ); } const published = await db .prepare( `SELECT COUNT(*) AS count FROM distributions WHERE delegation_bundle_id = ? AND publish_url IS NOT NULL AND publish_url != ''`, ) .bind(bundle.id) .first<{ count: number }>(); if ((published?.count ?? 0) > 0) { return Response.json( { error: "该分享包已有笔记发布,需保留链接继续完成第7天数据回收" }, { status: 409 }, ); } await db.batch([ db .prepare( `UPDATE distributions SET delegation_bundle_id = NULL, updated_at = CURRENT_TIMESTAMP WHERE delegation_bundle_id = ?`, ) .bind(bundle.id), db .prepare( `UPDATE delegation_bundles SET status = 'revoked', revoked_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ?`, ) .bind(bundle.id), ]); return Response.json({ revoked: true }); } if (body.action === "submit") { const claimToken = textValue(body.claimToken, 80); const distributionId = textValue(body.distributionId, 80); const publishInput = textValue(body.publishUrl, 5000); if (!publishInput) { return Response.json( { error: "请填写发布链接" }, { status: 400 }, ); } const publishUrl = extractXhsPublishUrl(publishInput); if (!publishUrl) { return Response.json( { error: "请粘贴包含小红书长链或短链的分享内容" }, { status: 400 }, ); } const account = accountFromPublishLink(publishUrl); if (!account) { return Response.json({ error: "发布链接格式不正确" }, { status: 400 }); } const assignment = await findAccessibleAssignment( task.id, distributionId, claimToken, delegationToken, ); if (!assignment) { return Response.json({ error: "笔记与领取凭证不匹配" }, { status: 403 }); } if (!assignment.publish_screenshot_key) { return Response.json({ error: "请先上传发布截图" }, { status: 400 }); } const reuseExistingAccount = assignment.publish_url === publishUrl && assignment.account_id; const accountId = reuseExistingAccount || `account-${hashText(`${account.platform}:${account.platformUid}`)}`; const statements: D1PreparedStatement[] = []; if (!reuseExistingAccount) { statements.push( db .prepare( `INSERT INTO accounts (id, platform, platform_uid, nickname, profile_url, post_count) VALUES (?, ?, ?, ?, ?, 1) ON CONFLICT(platform, platform_uid) DO UPDATE SET nickname = excluded.nickname, profile_url = excluded.profile_url, post_count = accounts.post_count + ?, last_seen_at = CURRENT_TIMESTAMP`, ) .bind( accountId, account.platform, account.platformUid, account.nickname, account.profileUrl, assignment.publish_url ? 0 : 1, ), ); } statements.push( db .prepare( `UPDATE distributions SET account_id = ?, publish_url = ?, publish_time = CURRENT_TIMESTAMP, status = 'published', updated_at = CURRENT_TIMESTAMP WHERE id = ?`, ) .bind(accountId, publishUrl, assignment.id), ); if (!assignment.publish_url) { statements.push( db .prepare( `UPDATE partners SET completed_total = completed_total + 1 WHERE id = ?`, ) .bind(assignment.partner_id), ); } await db.batch(statements); const collectionSchedule = await db .prepare( `SELECT collection_start_date, collection_days FROM tasks WHERE id = ?`, ) .bind(task.id) .first<{ collection_start_date: string | null; collection_days: string; }>(); if ( collectionSchedule?.collection_start_date && collectionSchedule.collection_days !== "[]" ) { let collectionDays: number[] = []; try { const parsed = JSON.parse(collectionSchedule.collection_days); if (Array.isArray(parsed)) collectionDays = parsed.map(Number); } catch { collectionDays = []; } if (collectionDays.length > 0) { await createCollectionRunTasks( db, task.id, collectionSchedule.collection_start_date, collectionDays, ); } } if (account.platform === "小红书") { const enrichment = enrichDistributionAccount( db, assignment.id, publishUrl, account.nickname, resolveCollectionMcpConfig( env as unknown as CollectionMcpBindings, ), ).catch(() => undefined); const executionContext = getRequestExecutionContext(); if (executionContext) { executionContext.waitUntil(enrichment); } else { await enrichment; } } return Response.json({ ok: true }); } if (body.action === "submit_creator_metrics") { const claimToken = textValue(body.claimToken, 80); const distributionId = textValue(body.distributionId, 80); const exposure = creatorMetricValue(body.exposure); const views = creatorMetricValue(body.views); if (exposure === null || views === null) { return Response.json( { error: "请填写正确的曝光量和阅读量" }, { status: 400 }, ); } const assignment = await findAccessibleAssignment( task.id, distributionId, claimToken, delegationToken, ); if (!assignment) { return Response.json({ error: "笔记与领取凭证不匹配" }, { status: 403 }); } if (!assignment.publish_url) { return Response.json( { error: "请先回填这篇笔记的发布信息" }, { status: 409 }, ); } if (!assignment.screenshot_key) { return Response.json( { error: "请先上传创作者中心截图" }, { status: 400 }, ); } 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, assignment.id) .run(); return Response.json({ submitted: true }); } return Response.json({ error: "不支持的操作" }, { status: 400 }); } catch (error) { return Response.json( { error: error instanceof Error ? error.message : "操作失败" }, { status: 500 }, ); } } export async function GET(request: Request) { const response = await handleGet(request); response.headers.set("Cache-Control", "private, no-store"); response.headers.set("Referrer-Policy", "no-referrer"); return withPartnerCors(request, response); } export async function POST(request: Request) { const response = await handlePost(request); response.headers.set("Cache-Control", "private, no-store"); return withPartnerCors(request, response); } export async function OPTIONS(request: Request) { return partnerOptions(request); }