diff --git a/app/admin-app.tsx b/app/admin-app.tsx index bf75f26..607c48a 100644 --- a/app/admin-app.tsx +++ b/app/admin-app.tsx @@ -74,6 +74,17 @@ type Account = { last_seen_at: string; }; +type IdRefreshStatus = { + status: "running" | "done"; + total: number; + processed: number; + updated: number; + failed: number; + skipped: number; + startedAt: string; + finishedAt: string | null; +}; + type ResourceImportPreview = { summary: { total: number; create: number; update: number; error: number }; rows: Array<{ @@ -3200,6 +3211,10 @@ function ResourcesPage({ const [editAccountId, setEditAccountId] = useState(""); const [editWorking, setEditWorking] = useState(false); const [editError, setEditError] = useState(""); + const [idRefresh, setIdRefresh] = useState(null); + const [idRefreshStarting, setIdRefreshStarting] = useState(false); + const idRefreshNotifiedRef = useRef(true); + const [idRefreshError, setIdRefreshError] = useState(""); const resourceAccounts = useMemo(() => { const deliveriesByAccount = new Map(); distributions.forEach((distribution) => { @@ -3447,6 +3462,84 @@ function ResourcesPage({ setEditWorking(false); } }; + const runningIdRefresh = idRefresh?.status === "running"; + const fetchIdRefreshStatus = useCallback(async () => { + const response = await fetch("/api/action", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ action: "public_account_id_refresh_status" }), + }); + const result = await readApiResponse<{ refresh: IdRefreshStatus | null }>( + response, + "查询失败", + ); + if (!response.ok) throw new Error(result.error || "查询失败"); + return result.refresh; + }, []); + const applyIdRefreshStatus = useCallback( + (status: IdRefreshStatus) => { + if (status.status === "done" && !idRefreshNotifiedRef.current) { + idRefreshNotifiedRef.current = true; + const summary = `账号ID更新完成：成功 ${status.updated} 个${ + status.skipped ? `，未识别 ${status.skipped} 个` : "" + }${status.failed ? `，失败 ${status.failed} 个` : ""}`; + void onImported(summary); + } + setIdRefresh(status); + }, + [onImported], + ); + useEffect(() => { + void fetchIdRefreshStatus() + .then((status) => { + if (status?.status === "running") { + idRefreshNotifiedRef.current = false; + applyIdRefreshStatus(status); + } + }) + .catch(() => undefined); + }, [applyIdRefreshStatus, fetchIdRefreshStatus]); + useEffect(() => { + if (!runningIdRefresh) return; + const timer = window.setInterval(() => { + void fetchIdRefreshStatus() + .then((status) => { + if (status) applyIdRefreshStatus(status); + }) + .catch(() => undefined); + }, 3000); + return () => window.clearInterval(timer); + }, [applyIdRefreshStatus, fetchIdRefreshStatus, runningIdRefresh]); + const startIdRefresh = async () => { + if (runningIdRefresh || idRefreshStarting) return; + if ( + !window.confirm( + "将逐个爬取全部账号主页以更新小红书号/抖音号（其他资料不变），账号较多时需要几分钟。确定开始？", + ) + ) { + return; + } + try { + setIdRefreshStarting(true); + setIdRefreshError(""); + const response = await fetch("/api/action", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ action: "refresh_public_account_ids" }), + }); + const result = await readApiResponse<{ refresh: IdRefreshStatus }>( + response, + "启动失败", + ); + if (!response.ok) throw new Error(result.error || "启动失败"); + idRefreshNotifiedRef.current = false; + setIdRefresh(result.refresh); + } catch (reason) { + setIdRefreshError(reason instanceof Error ? reason.message : "启动失败"); + } finally { + setIdRefreshStarting(false); + } + }; return (
@@ -3542,6 +3635,18 @@ function ResourcesPage({
共 {filteredAccounts.length} 个结果 +
+ {idRefreshError &&
{idRefreshError}
}
{filteredAccounts.map(({ account, sources, partnerManagedOnly }) => { diff --git a/app/api/action/route.ts b/app/api/action/route.ts index be65ab2..1e59091 100644 --- a/app/api/action/route.ts +++ b/app/api/action/route.ts @@ -54,6 +54,10 @@ import { } from "../../../lib/wecom-group-push-service"; import { isManagerRequest } from "../../../lib/user-auth"; import { extractPublishUrl } from "../../../lib/publish-url"; +import { + getPublicAccountIdRefreshStatus, + startPublicAccountIdRefresh, +} from "../../../lib/account-id-refresh-service"; type ActionBody = { action?: string; @@ -683,6 +687,14 @@ export async function POST(request: Request) { const taskId = String(body.taskId ?? "").trim(); const pushes = await listGroupPushes(db, taskId || undefined); return Response.json({ pushes }); + } else if (body.action === "refresh_public_account_ids") { + const status = startPublicAccountIdRefresh( + db, + resolveCollectionMcpConfig(env as unknown as CollectionMcpBindings), + ); + return Response.json({ refresh: status }); + } else if (body.action === "public_account_id_refresh_status") { + return Response.json({ refresh: getPublicAccountIdRefreshStatus() }); } else if (body.action === "wecom_group_chats") { const groups = await listGroupChatRows(db); return Response.json({ groups }); diff --git a/lib/account-id-refresh-service.ts b/lib/account-id-refresh-service.ts new file mode 100644 index 0000000..c04c6e8 --- /dev/null +++ b/lib/account-id-refresh-service.ts @@ -0,0 +1,134 @@ +import { runInBackground } from "./background"; +import { + resolveProfileDetailsFromMcp, + type CollectionMcpConfig, +} from "./mcp-collection-client"; +import type { DatabaseClient } from "./database"; + +export type PublicAccountIdRefreshStatus = { + status: "running" | "done"; + total: number; + processed: number; + updated: number; + failed: number; + skipped: number; + startedAt: string; + finishedAt: string | null; +}; + +type RefreshTargetRow = { + id: string; + platform: string; + platform_uid: string; + profile_url: string; +}; + +export type ProfileScraper = ( + profileUrl: string, + platform: "小红书" | "抖音", +) => Promise<{ redId: string | null }>; + +const defaultScraper = + (config: CollectionMcpConfig): ProfileScraper => + (profileUrl, platform) => + resolveProfileDetailsFromMcp(profileUrl, platform, config); + +let currentJob: PublicAccountIdRefreshStatus | null = null; + +export function getPublicAccountIdRefreshStatus(): PublicAccountIdRefreshStatus | null { + return currentJob ? { ...currentJob } : null; +} + +export function startPublicAccountIdRefresh( + db: DatabaseClient, + config: CollectionMcpConfig, +): PublicAccountIdRefreshStatus { + if (currentJob?.status === "running") { + return { ...currentJob }; + } + const job: PublicAccountIdRefreshStatus = { + status: "running", + total: 0, + processed: 0, + updated: 0, + failed: 0, + skipped: 0, + startedAt: new Date().toISOString(), + finishedAt: null, + }; + currentJob = job; + runInBackground( + executePublicAccountIdRefresh(job, db, defaultScraper(config)), + "public account id refresh", + ); + return { ...job }; +} + +async function executePublicAccountIdRefresh( + job: PublicAccountIdRefreshStatus, + db: DatabaseClient, + scraper: ProfileScraper, +) { + const rows = await db + .prepare( + `SELECT id, platform, platform_uid, profile_url + FROM accounts + WHERE profile_url IS NOT NULL AND profile_url != '' + AND platform IN ('小红书', '抖音')`, + ) + .all(); + const targets = rows.results.filter( + (row) => !row.platform_uid.startsWith("xhs-"), + ); + job.total = targets.length; + + let cursor = 0; + const worker = async () => { + while (cursor < targets.length) { + const row = targets[cursor]; + cursor += 1; + try { + const details = await scraper( + row.profile_url, + row.platform === "抖音" ? "抖音" : "小红书", + ); + const redId = details.redId?.trim(); + if (redId) { + await db + .prepare(`UPDATE accounts SET public_account_id = ? WHERE id = ?`) + .bind(redId, row.id) + .run(); + job.updated += 1; + } else { + job.skipped += 1; + } + } catch { + job.failed += 1; + } + job.processed += 1; + } + }; + await Promise.all( + Array.from({ length: Math.min(4, targets.length) }, worker), + ); + job.status = "done"; + job.finishedAt = new Date().toISOString(); +} + +export async function runPublicAccountIdRefresh( + db: DatabaseClient, + scraper: ProfileScraper, +): Promise { + const job: PublicAccountIdRefreshStatus = { + status: "running", + total: 0, + processed: 0, + updated: 0, + failed: 0, + skipped: 0, + startedAt: new Date().toISOString(), + finishedAt: null, + }; + await executePublicAccountIdRefresh(job, db, scraper); + return job; +} diff --git a/tests/account-id-refresh.test.mjs b/tests/account-id-refresh.test.mjs new file mode 100644 index 0000000..851e8f5 --- /dev/null +++ b/tests/account-id-refresh.test.mjs @@ -0,0 +1,98 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { + runPublicAccountIdRefresh, +} from "../lib/account-id-refresh-service.ts"; + +function fakeDb(accounts) { + const updates = []; + return { + db: { + prepare: (sql) => { + if (sql.startsWith("SELECT")) { + return { + all: async () => ({ + results: accounts.map((account) => ({ + id: account.id, + platform: account.platform, + platform_uid: account.platform_uid, + profile_url: account.profile_url, + })), + success: true, + meta: { changes: 0, last_row_id: 0 }, + }), + }; + } + const match = sql.match(/WHERE id = \?$/); + assert.ok(match, `unexpected sql: ${sql}`); + assert.match(sql, /UPDATE accounts SET public_account_id = \?/); + assert.doesNotMatch(sql, /followers|tags|nickname|bio|gender|ip_location/); + return { + bind: (value, id) => ({ + run: async () => { + updates.push({ value, id }); + return { success: true, meta: { changes: 1, last_row_id: 0 } }; + }, + }), + }; + }, + }, + updates, + }; +} + +test("updates only public_account_id for scrapeable accounts", async () => { + const accounts = [ + { id: "a1", platform: "小红书", platform_uid: "uid-1", profile_url: "https://www.xiaohongshu.com/user/profile/a1" }, + { id: "a2", platform: "抖音", platform_uid: "uid-2", profile_url: "https://www.douyin.com/user/a2" }, + ]; + const { db, updates } = fakeDb(accounts); + const status = await runPublicAccountIdRefresh(db, async (url) => { + if (url.endsWith("a1")) return { redId: "xhs-id-1" }; + if (url.endsWith("a2")) return { redId: " dy-id-2 " }; + return { redId: null }; + }); + + assert.equal(status.status, "done"); + assert.equal(status.total, 2); + assert.equal(status.processed, 2); + assert.equal(status.updated, 2); + assert.equal(status.failed, 0); + assert.deepEqual(updates, [ + { value: "xhs-id-1", id: "a1" }, + { value: "dy-id-2", id: "a2" }, + ]); +}); + +test("counts failures and skips empty redId without writing", async () => { + const accounts = [ + { id: "a1", platform: "小红书", platform_uid: "uid-1", profile_url: "https://www.xiaohongshu.com/user/profile/a1" }, + { id: "a2", platform: "小红书", platform_uid: "uid-2", profile_url: "https://www.xiaohongshu.com/user/profile/a2" }, + { id: "a3", platform: "小红书", platform_uid: "uid-3", profile_url: "https://www.xiaohongshu.com/user/profile/a3" }, + ]; + const { db, updates } = fakeDb(accounts); + const status = await runPublicAccountIdRefresh(db, async (url) => { + if (url.endsWith("a1")) throw new Error("mcp down"); + if (url.endsWith("a2")) return { redId: " " }; + return { redId: "ok-3" }; + }); + + assert.equal(status.total, 3); + assert.equal(status.updated, 1); + assert.equal(status.failed, 1); + assert.equal(status.skipped, 1); + assert.deepEqual(updates, [{ value: "ok-3", id: "a3" }]); +}); + +test("ignores demo accounts and accounts without profile_url at query level", async () => { + const accounts = [ + { id: "demo", platform: "小红书", platform_uid: "xhs-8af3", profile_url: "https://www.xiaohongshu.com/user/profile/demo" }, + { id: "real", platform: "小红书", platform_uid: "uid-9", profile_url: "https://www.xiaohongshu.com/user/profile/real" }, + ]; + const { db, updates } = fakeDb(accounts); + const status = await runPublicAccountIdRefresh(db, async () => ({ redId: "id-9" })); + + assert.equal(status.total, 1); + assert.equal(status.updated, 1); + assert.deepEqual(updates, [{ value: "id-9", id: "real" }]); +});