482 lines
13 KiB
TypeScript
482 lines
13 KiB
TypeScript
import {
|
|
collectXhsMetricsFromMcp,
|
|
type CollectionMcpConfig,
|
|
} from "./mcp-collection-client";
|
|
import { uid } from "./mvp-db";
|
|
import type { DatabaseClient, DatabaseStatement } from "./database";
|
|
|
|
type DistributionForCollection = {
|
|
id: string;
|
|
task_id: string;
|
|
publish_url: string | null;
|
|
ocr_status: string;
|
|
};
|
|
|
|
type ScheduledTask = {
|
|
id: string;
|
|
collection_start_date: string;
|
|
collection_days: string;
|
|
};
|
|
|
|
type CollectionSource = "automatic" | "catchup" | "manual";
|
|
|
|
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) {
|
|
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 shanghaiDateFromTimestamp(timestamp: number) {
|
|
return new Date(timestamp + 8 * 60 * 60 * 1000)
|
|
.toISOString()
|
|
.slice(0, 10);
|
|
}
|
|
|
|
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 * FROM distributions WHERE 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("采集任务创建失败");
|
|
if (run.status === "success") 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 } =
|
|
await collectXhsMetricsFromMcp(current.publish_url, 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 = ?,
|
|
status_description = ?,
|
|
completed_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?`,
|
|
)
|
|
.bind(
|
|
likes,
|
|
comments,
|
|
collects,
|
|
successDescription,
|
|
run.id,
|
|
),
|
|
db
|
|
.prepare(
|
|
`UPDATE distributions
|
|
SET latest_likes = ?,
|
|
latest_comments = ?,
|
|
latest_collects = ?,
|
|
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,
|
|
successDescription,
|
|
scheduleDay,
|
|
distributionId,
|
|
),
|
|
]);
|
|
return { skipped: false, likes, comments, collects };
|
|
} catch (error) {
|
|
const message =
|
|
error instanceof Error ? error.message : "公开数据采集失败";
|
|
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(
|
|
`UPDATE distributions
|
|
SET collection_status = 'failed',
|
|
collection_status_description = ?,
|
|
collection_updated_at = CURRENT_TIMESTAMP,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?`,
|
|
)
|
|
.bind(message, distributionId),
|
|
]);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
export async function runScheduledCollections(
|
|
db: DatabaseClient,
|
|
scheduledTimestamp: number,
|
|
mcpConfig: CollectionMcpConfig,
|
|
) {
|
|
return runDueScheduledCollections(
|
|
db,
|
|
scheduledTimestamp,
|
|
mcpConfig,
|
|
"automatic",
|
|
);
|
|
}
|
|
|
|
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, collection_start_date, collection_days
|
|
FROM tasks
|
|
WHERE collection_start_date IS NOT NULL
|
|
AND collection_start_date != ''
|
|
AND collection_days IS NOT NULL
|
|
AND collection_days != '[]'
|
|
AND (? IS NULL OR id = ?)`,
|
|
)
|
|
.bind(onlyTaskId ?? null, onlyTaskId ?? null)
|
|
.all<ScheduledTask>();
|
|
let dueTasks = 0;
|
|
let attempted = 0;
|
|
let succeeded = 0;
|
|
let failed = 0;
|
|
|
|
for (const task of tasks.results) {
|
|
const schedules = dueSchedules(
|
|
task.collection_start_date,
|
|
parseDays(task.collection_days),
|
|
timestamp,
|
|
);
|
|
for (const { scheduleDay, scheduledDate } of schedules) {
|
|
dueTasks += 1;
|
|
const distributions = await db
|
|
.prepare(
|
|
`SELECT d.id FROM distributions d
|
|
WHERE d.task_id = ?
|
|
AND d.publish_url IS NOT NULL
|
|
AND d.publish_url != ''
|
|
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(task.id, scheduledDate)
|
|
.all<{ id: string }>();
|
|
|
|
for (const distribution of distributions.results) {
|
|
attempted += 1;
|
|
try {
|
|
await collectDistributionMetrics(
|
|
db,
|
|
distribution.id,
|
|
scheduledDate,
|
|
scheduleDay,
|
|
source,
|
|
mcpConfig,
|
|
);
|
|
succeeded += 1;
|
|
} catch {
|
|
failed += 1;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
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,
|
|
};
|
|
}
|