import { runInBackground } from "./background"; import { generateAccountTagsFromMcp, resolveProfileDetailsFromMcp, type AccountTaggingEnv, type CollectionMcpConfig, } from "./mcp-collection-client"; import { getRuntimeEnv } from "./runtime-env"; import type { DatabaseClient } from "./database"; export const ACCOUNT_REFRESH_FIELDS = [ "publicAccountId", "nickname", "followers", "ipLocation", "gender", "bio", "tags", ] as const; export type AccountRefreshField = (typeof ACCOUNT_REFRESH_FIELDS)[number]; export const ACCOUNT_REFRESH_FIELD_LABELS: Record = { publicAccountId: "账号ID(小红书号/抖音号)", nickname: "昵称", followers: "粉丝数", ipLocation: "IP属地", gender: "性别", bio: "简介", tags: "标签(AI 重新打标,覆盖已有标签)", }; const FIELD_COLUMNS: Record = { publicAccountId: "public_account_id", nickname: "nickname", followers: "followers", ipLocation: "ip_location", gender: "gender", bio: "bio", tags: "tags", }; export function sanitizeRefreshFields(value: unknown): AccountRefreshField[] { const requested = Array.isArray(value) ? value : []; const fields = ACCOUNT_REFRESH_FIELDS.filter((field) => requested.includes(field), ); return fields.length > 0 ? fields : ["publicAccountId"]; } export type AccountFieldRefreshStatus = { status: "running" | "done"; fields: AccountRefreshField[]; 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 ScrapedProfile = { nickname: string | null; redId: string | null; followers: number | null; ipLocation: string | null; gender: string; bio: string; recentNoteTitles: string[]; providerTags: string[]; }; export type ProfileScraper = ( profileUrl: string, platform: "小红书" | "抖音", ) => Promise; export type TagGenerator = (input: { platform: string; nickname: string; bio: string; gender: string; ipLocation: string; followers: number | null; providerTags: string[]; recentNoteTitles: string[]; }) => Promise; function scrapedFieldValues( details: ScrapedProfile, ): Partial> { const values: Partial> = {}; const redId = details.redId?.trim(); if (redId) values.publicAccountId = redId; const nickname = details.nickname?.trim(); if (nickname) values.nickname = nickname; if (details.followers !== null && Number.isFinite(details.followers)) { values.followers = details.followers; } const ipLocation = details.ipLocation?.trim(); if (ipLocation) values.ipLocation = ipLocation; if (details.gender === "男" || details.gender === "女") { values.gender = details.gender; } const bio = details.bio?.trim(); if (bio) values.bio = bio; return values; } let currentJob: AccountFieldRefreshStatus | null = null; export function getAccountFieldRefreshStatus(): AccountFieldRefreshStatus | null { return currentJob ? { ...currentJob } : null; } export function startAccountFieldRefresh( db: DatabaseClient, config: CollectionMcpConfig, requestedFields: unknown, ): AccountFieldRefreshStatus { if (currentJob?.status === "running") { return { ...currentJob }; } const fields = sanitizeRefreshFields(requestedFields); const job: AccountFieldRefreshStatus = { status: "running", fields, total: 0, processed: 0, updated: 0, failed: 0, skipped: 0, startedAt: new Date().toISOString(), finishedAt: null, }; currentJob = job; const scraper: ProfileScraper = (profileUrl, platform) => resolveProfileDetailsFromMcp(profileUrl, platform, config); const tagger: TagGenerator = (input) => generateAccountTagsFromMcp( input, config, getRuntimeEnv() as AccountTaggingEnv, ); runInBackground( executeAccountFieldRefresh(job, db, scraper, tagger), "account field refresh", ); return { ...job }; } async function executeAccountFieldRefresh( job: AccountFieldRefreshStatus, db: DatabaseClient, scraper: ProfileScraper, tagger: TagGenerator, ) { 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; const fields = job.fields; const wantTags = fields.includes("tags"); 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 values = scrapedFieldValues(details); if (wantTags) { const tags = await tagger({ platform: row.platform, nickname: details.nickname?.trim() || "", bio: details.bio, gender: details.gender, ipLocation: details.ipLocation ?? "", followers: details.followers, providerTags: details.providerTags, recentNoteTitles: details.recentNoteTitles, }).catch(() => []); if (tags.length > 0) values.tags = tags.join(","); } const updates = fields.filter( (field) => values[field] !== undefined, ); if (updates.length === 0) { job.skipped += 1; } else { const assignments = updates .map((field) => `${FIELD_COLUMNS[field]} = ?`) .join(", "); await db .prepare(`UPDATE accounts SET ${assignments} WHERE id = ?`) .bind( ...updates.map((field) => values[field]), row.id, ) .run(); job.updated += 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 runAccountFieldRefresh( db: DatabaseClient, scraper: ProfileScraper, tagger: TagGenerator, fields: AccountRefreshField[], ): Promise { const job: AccountFieldRefreshStatus = { status: "running", fields: sanitizeRefreshFields(fields), total: 0, processed: 0, updated: 0, failed: 0, skipped: 0, startedAt: new Date().toISOString(), finishedAt: null, }; await executeAccountFieldRefresh(job, db, scraper, tagger); return job; }