87f4d4066d
KOC 详情弹窗（合作方/数据回收/工作台三处入口）： - 新增 KocDetailModal：基本信息 / 绑定KOC账号画像 / 参与任务记录 / 履约漏斗 - 工作台最近分发、回收队列账号名、截图回收合作方名、合作方页名称、分发详情头部合作方均可点开 - 分发表行点击仍打开分发详情（stopPropagation 隔离） 任务中心单入口改版（对齐"单任务双阶段"模型）： - 任务卡发布/回收面板整块可点，全卡仅保留一个「查看明细」，回收面板重复按钮移除 - 详情页自述身份：动态 label（任务详情·分发与回收一体/仅分发）+ 阶段徽章（含回收·截止/未开通回收） - 内容分发表新增阶段筛选 chips（全部/待发布/已发布待数据/待收截图，未开通回收不显示截图项），筛选落空给专用空态文案 - 数据回收页入口文案改为「选择一个任务进入回收队列」，全站不再区分发布/回收两种任务类型 资源池 KOC 默认加入合作方 + 采集兜底： - partners 新增 released_account_id（mysql/0023）：显式解绑/被顶替的账号不再被默认合作方同步回填 - 任务未设置采集计划时，默认按发布时间起连续采集 7 天，等待文案显示「等待D1 · 默认7天」
619 lines
18 KiB
TypeScript
619 lines
18 KiB
TypeScript
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;
|
||
task_type: string | null;
|
||
collection_start_date: string | null;
|
||
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 | null | undefined) {
|
||
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 parseCollectionDays(value: string | null | undefined) {
|
||
return parseDays(value);
|
||
}
|
||
|
||
// 数据回收（自动采集）兜底：任务未设置采集日时默认连续采集 7 天。
|
||
export const DEFAULT_COLLECTION_DAYS = [1, 2, 3, 4, 5, 6, 7];
|
||
|
||
export function effectiveCollectionDays(value: string | null | undefined) {
|
||
const days = parseDays(value);
|
||
return days.length > 0 ? days : [...DEFAULT_COLLECTION_DAYS];
|
||
}
|
||
|
||
export function shanghaiDateFromTimestamp(timestamp: number) {
|
||
return new Date(timestamp + 8 * 60 * 60 * 1000)
|
||
.toISOString()
|
||
.slice(0, 10);
|
||
}
|
||
|
||
// publish_time 等列以 UTC 文本存储（CURRENT_TIMESTAMP），统一换算成上海日期。
|
||
function shanghaiDateFromSqlTimestamp(value: string | null | undefined) {
|
||
if (!value) return null;
|
||
const normalized = value.includes("T")
|
||
? value
|
||
: `${value.replace(" ", "T")}Z`;
|
||
const epoch = Date.parse(normalized);
|
||
if (Number.isNaN(epoch)) return null;
|
||
return shanghaiDateFromTimestamp(epoch);
|
||
}
|
||
|
||
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",
|
||
);
|
||
}
|
||
|
||
async function distributionIdsNeedingCollection(
|
||
db: DatabaseClient,
|
||
taskId: string,
|
||
scheduledDate: string,
|
||
distributionId?: string,
|
||
) {
|
||
const scoped = distributionId ? "AND d.id = ?" : "";
|
||
const { results } = await db
|
||
.prepare(
|
||
`SELECT d.id FROM distributions d
|
||
WHERE d.task_id = ?
|
||
AND d.publish_url IS NOT NULL
|
||
AND d.publish_url != ''
|
||
${scoped}
|
||
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(...(distributionId ? [taskId, distributionId, scheduledDate] : [taskId, scheduledDate]))
|
||
.all<{ id: string }>();
|
||
return results.map((row) => row.id);
|
||
}
|
||
|
||
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, task_type, collection_start_date, collection_days
|
||
FROM tasks
|
||
WHERE COALESCE(task_type, 'content_publish') != 'screenshot_collect'
|
||
AND (? IS NULL OR id = ?)`,
|
||
)
|
||
.bind(onlyTaskId ?? null, onlyTaskId ?? null)
|
||
.all<ScheduledTask>();
|
||
let dueTasks = 0;
|
||
let attempted = 0;
|
||
let succeeded = 0;
|
||
let failed = 0;
|
||
|
||
const collectDue = async (
|
||
taskId: string,
|
||
schedules: { scheduleDay: number; scheduledDate: string }[],
|
||
distributionId?: string,
|
||
) => {
|
||
for (const { scheduleDay, scheduledDate } of schedules) {
|
||
dueTasks += 1;
|
||
const ids = await distributionIdsNeedingCollection(
|
||
db,
|
||
taskId,
|
||
scheduledDate,
|
||
distributionId,
|
||
);
|
||
for (const id of ids) {
|
||
attempted += 1;
|
||
try {
|
||
await collectDistributionMetrics(
|
||
db,
|
||
id,
|
||
scheduledDate,
|
||
scheduleDay,
|
||
source,
|
||
mcpConfig,
|
||
);
|
||
succeeded += 1;
|
||
} catch {
|
||
failed += 1;
|
||
}
|
||
}
|
||
}
|
||
};
|
||
|
||
for (const task of tasks.results) {
|
||
const parsedDays = parseDays(task.collection_days);
|
||
if (task.collection_start_date && parsedDays.length > 0) {
|
||
await collectDue(
|
||
task.id,
|
||
dueSchedules(task.collection_start_date, parsedDays, timestamp),
|
||
);
|
||
continue;
|
||
}
|
||
// 未设置采集计划：默认按每篇作品的发布时间起连续采集 7 天。
|
||
const days = effectiveCollectionDays(task.collection_days);
|
||
const published = await db
|
||
.prepare(
|
||
`SELECT d.id, COALESCE(d.publish_time, d.claimed_at) AS publish_anchor
|
||
FROM distributions d
|
||
WHERE d.task_id = ?
|
||
AND d.publish_url IS NOT NULL
|
||
AND d.publish_url != ''`,
|
||
)
|
||
.bind(task.id)
|
||
.all<{ id: string; publish_anchor: string | null }>();
|
||
for (const distribution of published.results) {
|
||
const anchorDate = shanghaiDateFromSqlTimestamp(
|
||
distribution.publish_anchor,
|
||
);
|
||
if (!anchorDate) continue;
|
||
const windowEndDate = dateForScheduleDay(
|
||
anchorDate,
|
||
days[days.length - 1],
|
||
);
|
||
// 默认窗口已过（发布超过 7 天且从未配置计划）不再补采历史天，
|
||
// 避免 D1 快照与真实发布时间脱节；确需补采时由管理员设置明确计划。
|
||
if (!windowEndDate || currentDate > windowEndDate) continue;
|
||
await collectDue(
|
||
task.id,
|
||
dueSchedules(anchorDate, days, timestamp),
|
||
distribution.id,
|
||
);
|
||
}
|
||
}
|
||
|
||
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,
|
||
};
|
||
}
|