Expand KOC LOOP MCP operations
This commit is contained in:
巫凤萍
464
lib/mcp-operations.ts
Normal file
464
lib/mcp-operations.ts
Normal file
@@ -0,0 +1,464 @@
|
||||
import { enrichDistributionAccount } from "./account-enrichment-service";
|
||||
import {
|
||||
collectDistributionMetrics,
|
||||
createCollectionRunTasks,
|
||||
retryFailedCollections,
|
||||
runDueScheduledCollections,
|
||||
shanghaiDateFromTimestamp,
|
||||
} from "./collection-service";
|
||||
import { issueMcpExportToken } from "./mcp-export-token";
|
||||
import {
|
||||
resolveCollectionMcpConfig,
|
||||
type CollectionMcpBindings,
|
||||
} from "./mcp-collection-client";
|
||||
import { ensureSchema, getRawDb } from "./mvp-db";
|
||||
import { extractXhsPublishUrl } from "./publish-url";
|
||||
import { buildClaimUrl } from "./task-service";
|
||||
|
||||
export type McpOperationBindings = CollectionMcpBindings & {
|
||||
KOC_PORTAL_URL?: string;
|
||||
};
|
||||
|
||||
type Pagination = { limit?: number; offset?: number };
|
||||
type ResourceFilters = Pagination & {
|
||||
query?: string;
|
||||
ipLocation?: string;
|
||||
cooperationSource?: string;
|
||||
platform?: string;
|
||||
};
|
||||
|
||||
function page(input: Pagination) {
|
||||
return {
|
||||
limit: Math.max(1, Math.min(200, Math.floor(input.limit ?? 50))),
|
||||
offset: Math.max(0, Math.floor(input.offset ?? 0)),
|
||||
};
|
||||
}
|
||||
|
||||
function like(value: string) {
|
||||
return `%${value.replace(/[\\%_]/g, "\\$&")}%`;
|
||||
}
|
||||
|
||||
function parseJsonArray(value: unknown) {
|
||||
try {
|
||||
const parsed = JSON.parse(String(value ?? "[]"));
|
||||
return Array.isArray(parsed) ? parsed : [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
async function taskExists(taskId: string) {
|
||||
const task = await getRawDb()
|
||||
.prepare("SELECT id FROM tasks WHERE id = ?")
|
||||
.bind(taskId)
|
||||
.first<{ id: string }>();
|
||||
if (!task) throw new Error("任务不存在");
|
||||
}
|
||||
|
||||
export async function taskList(
|
||||
input: Pagination & { query?: string; status?: string },
|
||||
portalUrl: string,
|
||||
) {
|
||||
await ensureSchema();
|
||||
const { limit, offset } = page(input);
|
||||
const conditions: string[] = [];
|
||||
const bindings: unknown[] = [];
|
||||
if (input.query?.trim()) {
|
||||
conditions.push("(t.name LIKE ? ESCAPE '\\' OR t.brand LIKE ? ESCAPE '\\')");
|
||||
const pattern = like(input.query.trim());
|
||||
bindings.push(pattern, pattern);
|
||||
}
|
||||
if (input.status?.trim() && input.status !== "all") {
|
||||
conditions.push("t.status = ?");
|
||||
bindings.push(input.status.trim());
|
||||
}
|
||||
const where = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
|
||||
const db = getRawDb();
|
||||
const [rows, count] = await Promise.all([
|
||||
db
|
||||
.prepare(
|
||||
`SELECT t.*,
|
||||
(SELECT COUNT(*) FROM contents c WHERE c.task_id = t.id) AS note_count,
|
||||
(SELECT COUNT(*) FROM distributions d WHERE d.task_id = t.id) AS claimed_count,
|
||||
(SELECT COUNT(*) FROM distributions d WHERE d.task_id = t.id AND COALESCE(d.publish_url, '') != '') AS published_count,
|
||||
(SELECT COUNT(*) FROM distributions d WHERE d.task_id = t.id AND d.exposure IS NOT NULL AND d.views IS NOT NULL) AS day7_count
|
||||
FROM tasks t ${where}
|
||||
ORDER BY t.created_at DESC LIMIT ? OFFSET ?`,
|
||||
)
|
||||
.bind(...bindings, limit, offset)
|
||||
.all<Record<string, unknown>>(),
|
||||
db
|
||||
.prepare(`SELECT COUNT(*) AS total FROM tasks t ${where}`)
|
||||
.bind(...bindings)
|
||||
.first<{ total: number }>(),
|
||||
]);
|
||||
return {
|
||||
total: count?.total ?? 0,
|
||||
limit,
|
||||
offset,
|
||||
tasks: rows.results.map((row) => ({
|
||||
...row,
|
||||
collection_days: parseJsonArray(row.collection_days),
|
||||
claim_url:
|
||||
portalUrl && row.share_token
|
||||
? buildClaimUrl(portalUrl, String(row.share_token))
|
||||
: null,
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
export async function taskGet(taskId: string, portalUrl: string) {
|
||||
await ensureSchema();
|
||||
const db = getRawDb();
|
||||
const task = await db
|
||||
.prepare("SELECT * FROM tasks WHERE id = ?")
|
||||
.bind(taskId)
|
||||
.first<Record<string, unknown>>();
|
||||
if (!task) throw new Error("任务不存在");
|
||||
const [notes, claims, runs] = await Promise.all([
|
||||
db
|
||||
.prepare(
|
||||
`SELECT c.id AS content_id, c.source_row, c.title, c.body, c.image_assets, c.status AS content_status,
|
||||
d.id AS distribution_id, d.status AS distribution_status, d.publish_url, d.publish_time,
|
||||
d.exposure, d.views, d.latest_likes, d.latest_comments, d.latest_collects,
|
||||
d.collection_status, d.collection_status_description, d.collection_updated_at,
|
||||
a.id AS account_id, a.nickname AS account_nickname, a.profile_url, a.followers,
|
||||
p.name AS cooperation_source, cl.claimant_name
|
||||
FROM contents c
|
||||
LEFT JOIN distributions d ON d.content_id = c.id
|
||||
LEFT JOIN accounts a ON a.id = d.account_id
|
||||
LEFT JOIN partners p ON p.id = d.partner_id
|
||||
LEFT JOIN claims cl ON cl.id = d.claim_id
|
||||
WHERE c.task_id = ?
|
||||
ORDER BY COALESCE(c.source_row, 999999), c.created_at, d.claimed_at`,
|
||||
)
|
||||
.bind(taskId)
|
||||
.all<Record<string, unknown>>(),
|
||||
db
|
||||
.prepare(
|
||||
`SELECT cl.*, p.name AS partner_name,
|
||||
(SELECT COUNT(*) FROM distributions d WHERE d.claim_id = cl.id AND COALESCE(d.publish_url, '') != '') AS published_count
|
||||
FROM claims cl JOIN partners p ON p.id = cl.partner_id
|
||||
WHERE cl.task_id = ? ORDER BY cl.created_at DESC`,
|
||||
)
|
||||
.bind(taskId)
|
||||
.all<Record<string, unknown>>(),
|
||||
db
|
||||
.prepare(
|
||||
`SELECT * FROM collection_runs WHERE task_id = ?
|
||||
ORDER BY scheduled_date DESC, created_at DESC LIMIT 500`,
|
||||
)
|
||||
.bind(taskId)
|
||||
.all<Record<string, unknown>>(),
|
||||
]);
|
||||
return {
|
||||
task: {
|
||||
...task,
|
||||
collection_days: parseJsonArray(task.collection_days),
|
||||
claim_url:
|
||||
portalUrl && task.share_token
|
||||
? buildClaimUrl(portalUrl, String(task.share_token))
|
||||
: null,
|
||||
},
|
||||
notes: notes.results.map((row) => ({
|
||||
...row,
|
||||
image_assets: parseJsonArray(row.image_assets),
|
||||
total_interactions:
|
||||
row.latest_likes == null
|
||||
? null
|
||||
: Number(row.latest_likes) +
|
||||
Number(row.latest_comments ?? 0) +
|
||||
Number(row.latest_collects ?? 0),
|
||||
})),
|
||||
claims: claims.results,
|
||||
collection_runs: runs.results,
|
||||
};
|
||||
}
|
||||
|
||||
export async function recoveryList(
|
||||
input: Pagination & {
|
||||
taskId?: string;
|
||||
stage?: "all" | "published" | "unfilled" | "waiting_day7" | "day7_due";
|
||||
},
|
||||
) {
|
||||
await ensureSchema();
|
||||
const { limit, offset } = page(input);
|
||||
const db = getRawDb();
|
||||
const conditions: string[] = [];
|
||||
const bindings: unknown[] = [];
|
||||
if (input.taskId?.trim()) {
|
||||
conditions.push("d.task_id = ?");
|
||||
bindings.push(input.taskId.trim());
|
||||
}
|
||||
const stage = input.stage ?? "all";
|
||||
if (stage === "published") conditions.push("COALESCE(d.publish_url, '') != ''");
|
||||
if (stage === "unfilled") conditions.push("COALESCE(d.publish_url, '') = ''");
|
||||
if (stage === "waiting_day7") {
|
||||
conditions.push("COALESCE(d.publish_url, '') != '' AND (d.exposure IS NULL OR d.views IS NULL)");
|
||||
}
|
||||
if (stage === "day7_due") {
|
||||
conditions.push(
|
||||
"COALESCE(d.publish_url, '') != '' AND (d.exposure IS NULL OR d.views IS NULL) AND datetime(d.publish_time, '+7 days') <= CURRENT_TIMESTAMP",
|
||||
);
|
||||
}
|
||||
const where = conditions.length ? `WHERE ${conditions.join(" AND ")}` : "";
|
||||
const base = `FROM distributions d
|
||||
JOIN tasks t ON t.id = d.task_id
|
||||
JOIN contents c ON c.id = d.content_id
|
||||
LEFT JOIN accounts a ON a.id = d.account_id
|
||||
LEFT JOIN partners p ON p.id = d.partner_id
|
||||
LEFT JOIN claims cl ON cl.id = d.claim_id ${where}`;
|
||||
const [rows, count] = await Promise.all([
|
||||
db
|
||||
.prepare(
|
||||
`SELECT d.*, t.name AS task_name, c.source_row, c.title,
|
||||
a.nickname AS account_nickname, a.profile_url, p.name AS cooperation_source,
|
||||
cl.claimant_name ${base}
|
||||
ORDER BY COALESCE(d.publish_time, d.claimed_at) DESC LIMIT ? OFFSET ?`,
|
||||
)
|
||||
.bind(...bindings, limit, offset)
|
||||
.all<Record<string, unknown>>(),
|
||||
db
|
||||
.prepare(`SELECT COUNT(*) AS total ${base}`)
|
||||
.bind(...bindings)
|
||||
.first<{ total: number }>(),
|
||||
]);
|
||||
return {
|
||||
total: count?.total ?? 0,
|
||||
limit,
|
||||
offset,
|
||||
items: rows.results.map((row) => ({
|
||||
...row,
|
||||
total_interactions:
|
||||
row.latest_likes == null
|
||||
? null
|
||||
: Number(row.latest_likes) + Number(row.latest_comments ?? 0) + Number(row.latest_collects ?? 0),
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
export async function recoveryExport(taskId: string, origin: string) {
|
||||
await ensureSchema();
|
||||
await taskExists(taskId);
|
||||
const issued = await issueMcpExportToken("recovery", { taskId });
|
||||
return {
|
||||
task_id: taskId,
|
||||
download_url: `${origin}/api/recovery-export?token=${encodeURIComponent(issued.token)}`,
|
||||
expires_at: issued.expiresAt,
|
||||
};
|
||||
}
|
||||
|
||||
export async function setCollectionPlan(
|
||||
taskId: string,
|
||||
startDate: string,
|
||||
days: number[],
|
||||
bindings: McpOperationBindings,
|
||||
) {
|
||||
await ensureSchema();
|
||||
await taskExists(taskId);
|
||||
const normalizedDays = [...new Set(days)]
|
||||
.filter((day) => Number.isInteger(day) && day >= 1 && day <= 7)
|
||||
.sort((a, b) => a - b);
|
||||
if (!/^\d{4}-\d{2}-\d{2}$/.test(startDate) || Number.isNaN(Date.parse(`${startDate}T00:00:00+08:00`))) {
|
||||
throw new Error("开始采集日期无效");
|
||||
}
|
||||
if (!normalizedDays.length) throw new Error("请至少选择一个采集日");
|
||||
const db = getRawDb();
|
||||
await db.batch([
|
||||
db.prepare(
|
||||
`UPDATE tasks SET collection_start_date = ?, collection_days = ?,
|
||||
collection_schedule_updated_at = CURRENT_TIMESTAMP WHERE id = ?`,
|
||||
).bind(startDate, JSON.stringify(normalizedDays), taskId),
|
||||
db.prepare(
|
||||
`UPDATE distributions SET
|
||||
collection_status = CASE WHEN latest_likes IS NULL THEN 'scheduled' ELSE collection_status END,
|
||||
collection_status_description = CASE WHEN latest_likes IS NULL THEN ? ELSE collection_status_description END,
|
||||
updated_at = CURRENT_TIMESTAMP
|
||||
WHERE task_id = ? AND COALESCE(publish_url, '') != ''`,
|
||||
).bind(`已安排${normalizedDays.length}个采集日，每日10:00执行`, taskId),
|
||||
]);
|
||||
const created = await createCollectionRunTasks(db, taskId, startDate, normalizedDays);
|
||||
const catchup = await runDueScheduledCollections(
|
||||
db,
|
||||
Date.now(),
|
||||
resolveCollectionMcpConfig(bindings),
|
||||
"catchup",
|
||||
taskId,
|
||||
);
|
||||
return { task_id: taskId, start_date: startDate, days: normalizedDays, created, catchup };
|
||||
}
|
||||
|
||||
export async function runDueCollections(taskId: string | undefined, bindings: McpOperationBindings) {
|
||||
await ensureSchema();
|
||||
if (taskId) await taskExists(taskId);
|
||||
return runDueScheduledCollections(
|
||||
getRawDb(),
|
||||
Date.now(),
|
||||
resolveCollectionMcpConfig(bindings),
|
||||
"catchup",
|
||||
taskId,
|
||||
);
|
||||
}
|
||||
|
||||
export async function collectNow(
|
||||
distributionId: string,
|
||||
scheduleDay: number | undefined,
|
||||
bindings: McpOperationBindings,
|
||||
) {
|
||||
await ensureSchema();
|
||||
const day = scheduleDay ?? null;
|
||||
const metrics = await collectDistributionMetrics(
|
||||
getRawDb(),
|
||||
distributionId,
|
||||
shanghaiDateFromTimestamp(Date.now()),
|
||||
day,
|
||||
"manual",
|
||||
resolveCollectionMcpConfig(bindings),
|
||||
);
|
||||
return { distribution_id: distributionId, schedule_day: day, ...metrics };
|
||||
}
|
||||
|
||||
export async function retryFailed(taskId: string | undefined, bindings: McpOperationBindings) {
|
||||
await ensureSchema();
|
||||
const db = getRawDb();
|
||||
const ids = taskId
|
||||
? [taskId]
|
||||
: (
|
||||
await db
|
||||
.prepare(
|
||||
`SELECT DISTINCT task_id FROM distributions
|
||||
WHERE collection_status = 'failed' ORDER BY updated_at DESC LIMIT 20`,
|
||||
)
|
||||
.all<{ task_id: string }>()
|
||||
).results.map((row) => row.task_id);
|
||||
if (taskId) await taskExists(taskId);
|
||||
const results = [];
|
||||
for (const id of ids) {
|
||||
results.push({ task_id: id, ...(await retryFailedCollections(db, id, resolveCollectionMcpConfig(bindings))) });
|
||||
}
|
||||
return {
|
||||
task_count: results.length,
|
||||
attempted: results.reduce((sum, item) => sum + item.attempted, 0),
|
||||
succeeded: results.reduce((sum, item) => sum + item.succeeded, 0),
|
||||
failed: results.reduce((sum, item) => sum + item.failed, 0),
|
||||
tasks: results,
|
||||
};
|
||||
}
|
||||
|
||||
function resourceWhere(input: ResourceFilters) {
|
||||
const conditions: string[] = [];
|
||||
const bindings: unknown[] = [];
|
||||
if (input.query?.trim()) {
|
||||
conditions.push("(a.nickname LIKE ? ESCAPE '\\' OR a.public_account_id LIKE ? ESCAPE '\\')");
|
||||
const pattern = like(input.query.trim());
|
||||
bindings.push(pattern, pattern);
|
||||
}
|
||||
if (input.ipLocation?.trim()) {
|
||||
conditions.push("a.ip_location LIKE ? ESCAPE '\\'");
|
||||
bindings.push(like(input.ipLocation.trim()));
|
||||
}
|
||||
if (input.cooperationSource?.trim()) {
|
||||
conditions.push(
|
||||
`EXISTS (SELECT 1 FROM distributions dx JOIN partners px ON px.id = dx.partner_id
|
||||
WHERE dx.account_id = a.id AND px.name LIKE ? ESCAPE '\\')`,
|
||||
);
|
||||
bindings.push(like(input.cooperationSource.trim()));
|
||||
}
|
||||
if (input.platform?.trim() && input.platform !== "all") {
|
||||
conditions.push("a.platform = ?");
|
||||
bindings.push(input.platform.trim());
|
||||
}
|
||||
return { where: conditions.length ? `WHERE ${conditions.join(" AND ")}` : "", bindings };
|
||||
}
|
||||
|
||||
export async function resourceSearch(input: ResourceFilters) {
|
||||
await ensureSchema();
|
||||
const { limit, offset } = page(input);
|
||||
const { where, bindings } = resourceWhere(input);
|
||||
const db = getRawDb();
|
||||
const select = `SELECT a.*,
|
||||
(SELECT COUNT(*) FROM distributions d WHERE d.account_id = a.id) AS cooperation_count,
|
||||
(SELECT GROUP_CONCAT(DISTINCT p.name) FROM distributions d JOIN partners p ON p.id = d.partner_id WHERE d.account_id = a.id) AS cooperation_sources`;
|
||||
const [rows, count] = await Promise.all([
|
||||
db.prepare(`${select} FROM accounts a ${where} ORDER BY a.last_seen_at DESC LIMIT ? OFFSET ?`)
|
||||
.bind(...bindings, limit, offset).all<Record<string, unknown>>(),
|
||||
db.prepare(`SELECT COUNT(*) AS total FROM accounts a ${where}`).bind(...bindings).first<{ total: number }>(),
|
||||
]);
|
||||
return {
|
||||
total: count?.total ?? 0,
|
||||
limit,
|
||||
offset,
|
||||
accounts: rows.results.map((row) => ({
|
||||
...row,
|
||||
cooperation_sources: String(row.cooperation_sources ?? "").split(",").filter(Boolean),
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
export async function resourceGet(accountId: string) {
|
||||
await ensureSchema();
|
||||
const db = getRawDb();
|
||||
const account = await db.prepare("SELECT * FROM accounts WHERE id = ?").bind(accountId).first<Record<string, unknown>>();
|
||||
if (!account) throw new Error("账号不存在");
|
||||
const history = await db.prepare(
|
||||
`SELECT d.id AS distribution_id, d.task_id, t.name AS task_name, c.title,
|
||||
d.publish_url, d.publish_time, d.latest_likes, d.latest_comments, d.latest_collects,
|
||||
d.exposure, d.views, p.name AS cooperation_source, cl.claimant_name
|
||||
FROM distributions d JOIN tasks t ON t.id = d.task_id
|
||||
JOIN contents c ON c.id = d.content_id JOIN partners p ON p.id = d.partner_id
|
||||
LEFT JOIN claims cl ON cl.id = d.claim_id
|
||||
WHERE d.account_id = ? ORDER BY COALESCE(d.publish_time, d.claimed_at) DESC`,
|
||||
).bind(accountId).all<Record<string, unknown>>();
|
||||
return { account, cooperation_history: history.results };
|
||||
}
|
||||
|
||||
export async function backfillResourceProfile(
|
||||
input: { distributionId?: string; publishUrl?: string },
|
||||
bindings: McpOperationBindings,
|
||||
) {
|
||||
await ensureSchema();
|
||||
const db = getRawDb();
|
||||
const publishUrl = input.publishUrl ? extractXhsPublishUrl(input.publishUrl) : "";
|
||||
const row = input.distributionId
|
||||
? await db.prepare(
|
||||
`SELECT d.id, d.publish_url, COALESCE(a.nickname, '') AS nickname
|
||||
FROM distributions d LEFT JOIN accounts a ON a.id = d.account_id WHERE d.id = ?`,
|
||||
).bind(input.distributionId).first<{ id: string; publish_url: string | null; nickname: string }>()
|
||||
: publishUrl
|
||||
? await db.prepare(
|
||||
`SELECT d.id, d.publish_url, COALESCE(a.nickname, '') AS nickname
|
||||
FROM distributions d LEFT JOIN accounts a ON a.id = d.account_id
|
||||
WHERE d.publish_url = ? ORDER BY d.updated_at DESC LIMIT 1`,
|
||||
).bind(publishUrl).first<{ id: string; publish_url: string | null; nickname: string }>()
|
||||
: null;
|
||||
if (!row?.publish_url) throw new Error("没有找到可补全的发布记录");
|
||||
const result = await enrichDistributionAccount(
|
||||
db,
|
||||
row.id,
|
||||
row.publish_url,
|
||||
row.nickname,
|
||||
resolveCollectionMcpConfig(bindings),
|
||||
);
|
||||
const account = result.updated && result.accountId
|
||||
? await db.prepare("SELECT * FROM accounts WHERE id = ?").bind(result.accountId).first<Record<string, unknown>>()
|
||||
: null;
|
||||
return { distribution_id: row.id, ...result, account };
|
||||
}
|
||||
|
||||
export async function resourceExport(input: ResourceFilters, origin: string) {
|
||||
const result = await resourceSearch({ ...input, limit: 200, offset: 0 });
|
||||
const allIds: string[] = result.accounts.map((account) => String(account.id));
|
||||
if (result.total > result.accounts.length) {
|
||||
const { where, bindings } = resourceWhere(input);
|
||||
const rows = await getRawDb().prepare(`SELECT a.id FROM accounts a ${where} ORDER BY a.last_seen_at DESC LIMIT 5000`)
|
||||
.bind(...bindings).all<{ id: string }>();
|
||||
allIds.splice(0, allIds.length, ...rows.results.map((row) => row.id));
|
||||
}
|
||||
if (!allIds.length) throw new Error("当前筛选结果为空");
|
||||
const issued = await issueMcpExportToken("resources", { accountIds: allIds });
|
||||
return {
|
||||
account_count: allIds.length,
|
||||
download_url: `${origin}/api/resources-export?token=${encodeURIComponent(issued.token)}`,
|
||||
expires_at: issued.expiresAt,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user