import { resolveXhsPublicAccountDetails, resolveAccountProfileFromMcp, resolveProfileDetailsFromMcp, type CollectionMcpConfig, } from "./mcp-collection-client"; import { hashText } from "./mvp-db"; import type { DatabaseClient } from "./database"; type DistributionAccountRow = { id: string; account_id: string | null; publish_url: string | null; platform: string; claimant_contact: string | null; }; type BackfillRow = DistributionAccountRow & { resolved_account_id: string | null; nickname: string | null; platform_uid: string | null; public_account_id: string | null; profile_url: string | null; followers: number | null; gender: string | null; bio: string | null; tags: string | null; }; function isVerifiedXhsProfileUrl(value: string | null) { if (!value) return false; try { const url = new URL(value); return ( url.protocol === "https:" && (url.hostname === "xiaohongshu.com" || url.hostname.endsWith(".xiaohongshu.com")) && url.pathname.startsWith("/user/profile/") ); } catch { return false; } } function isVerifiedDouyinProfileUrl(value: string | null) { if (!value) return false; try { const url = new URL(value); const secUid = decodeURIComponent(url.pathname.match(/^\/user\/([^/?#]+)/)?.[1] ?? ""); return ( url.protocol === "https:" && (url.hostname === "douyin.com" || url.hostname.endsWith(".douyin.com")) && /^[A-Za-z0-9_-]{20,220}$/.test(secUid) && !/^\d+$/.test(secUid) ); } catch { return false; } } export async function enrichDistributionAccount( db: DatabaseClient, distributionId: string, publishUrl: string, fallbackNickname: string, mcpConfig: CollectionMcpConfig, ) { const current = await db .prepare( `SELECT d.id, d.account_id, d.publish_url, t.platform, cl.claimant_name AS claimant_contact FROM distributions d JOIN tasks t ON t.id = d.task_id LEFT JOIN claims cl ON cl.id = d.claim_id WHERE d.id = ?`, ) .bind(distributionId) .first(); if (!current || current.publish_url !== publishUrl) { return { updated: false, reason: "stale" as const }; } const platform = current.platform === "抖音" ? "抖音" : "小红书"; const profile = await resolveAccountProfileFromMcp( publishUrl, fallbackNickname, platform, mcpConfig, ); const canonicalAccountId = `account-${hashText( `${platform}:${profile.platformUid}`, )}`; const existingAccount = await db .prepare( `SELECT id FROM accounts WHERE platform = ? AND platform_uid = ? LIMIT 1`, ) .bind(platform, profile.platformUid) .first<{ id: string }>(); const targetAccountId = existingAccount?.id || canonicalAccountId; const currentContact = (current.claimant_contact || "").trim(); if (current.account_id === targetAccountId) { await db .prepare( `UPDATE accounts SET nickname = ?, profile_url = ?, public_account_id = CASE WHEN ? != '' THEN ? ELSE public_account_id END, ip_location = CASE WHEN ? != '' AND ? != '待识别' THEN ? ELSE ip_location END, followers = CASE WHEN ? IS NOT NULL AND (? > 0 OR followers = 0) THEN ? ELSE followers END, gender = CASE WHEN ? != '' THEN ? ELSE gender END, bio = CASE WHEN ? != '' THEN ? ELSE bio END, current_contact = CASE WHEN ? != '' THEN ? ELSE current_contact END, post_count = ( SELECT COUNT(*) FROM distributions WHERE account_id = ? ), last_seen_at = CURRENT_TIMESTAMP WHERE id = ?`, ) .bind( profile.nickname || fallbackNickname, profile.profileUrl, profile.redId, profile.redId, profile.ipLocation, profile.ipLocation, profile.ipLocation, profile.followers, profile.followers, profile.followers, profile.gender, profile.gender, profile.bio, profile.bio, currentContact, currentContact, targetAccountId, targetAccountId, ) .run(); return { updated: true, accountId: targetAccountId, profileUrl: profile.profileUrl, }; } const provisionalAccountId = current.account_id; await db.batch([ db .prepare( `INSERT INTO accounts (id, platform, platform_uid, public_account_id, nickname, profile_url, ip_location, followers, gender, bio, current_contact, post_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1) ON CONFLICT(platform, platform_uid) DO UPDATE SET public_account_id = CASE WHEN excluded.public_account_id != '' THEN excluded.public_account_id ELSE accounts.public_account_id END, nickname = excluded.nickname, profile_url = excluded.profile_url, ip_location = CASE WHEN excluded.ip_location != '' AND excluded.ip_location != '待识别' THEN excluded.ip_location ELSE accounts.ip_location END, followers = CASE WHEN excluded.followers > 0 OR accounts.followers = 0 THEN excluded.followers ELSE accounts.followers END, gender = CASE WHEN excluded.gender != '' THEN excluded.gender ELSE accounts.gender END, bio = CASE WHEN excluded.bio != '' THEN excluded.bio ELSE accounts.bio END, current_contact = CASE WHEN excluded.current_contact != '' THEN excluded.current_contact ELSE accounts.current_contact END, last_seen_at = CURRENT_TIMESTAMP`, ) .bind( targetAccountId, platform, profile.platformUid, profile.redId, profile.nickname || fallbackNickname, profile.profileUrl, profile.ipLocation, profile.followers ?? 0, profile.gender, profile.bio, currentContact, ), db .prepare( `UPDATE distributions SET account_id = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ? AND publish_url = ?`, ) .bind(targetAccountId, distributionId, publishUrl), db .prepare( `UPDATE accounts SET post_count = ( SELECT COUNT(*) FROM distributions WHERE account_id = ? ) WHERE id = ?`, ) .bind(targetAccountId, targetAccountId), ]); if (provisionalAccountId && provisionalAccountId !== targetAccountId) { await db .prepare( `UPDATE accounts SET post_count = ( SELECT COUNT(*) FROM distributions WHERE account_id = ? ) WHERE id = ?`, ) .bind(provisionalAccountId, provisionalAccountId) .run(); await db .prepare( `DELETE FROM accounts WHERE id = ? AND id != ? AND NOT EXISTS ( SELECT 1 FROM distributions WHERE account_id = ? )`, ) .bind( provisionalAccountId, targetAccountId, provisionalAccountId, ) .run(); } return { updated: true, accountId: targetAccountId, profileUrl: profile.profileUrl, }; } export async function backfillAccountProfiles( db: DatabaseClient, mcpConfig: CollectionMcpConfig, limit = 10, ) { const rows = await db .prepare( `SELECT d.id, d.account_id, d.publish_url, a.id AS resolved_account_id, a.nickname, COALESCE(a.platform, t.platform) AS platform, a.platform_uid, a.public_account_id, a.profile_url, a.followers, a.gender, a.bio, a.tags, cl.claimant_name AS claimant_contact FROM distributions d JOIN tasks t ON t.id = d.task_id LEFT JOIN accounts a ON a.id = d.account_id LEFT JOIN claims cl ON cl.id = d.claim_id WHERE d.publish_url IS NOT NULL AND d.publish_url != '' ORDER BY d.updated_at DESC LIMIT 100`, ) .all(); let attempted = 0; let updated = 0; let failed = 0; const backfilledAccounts = new Set(); for (const row of rows.results) { if ( row.resolved_account_id && !backfilledAccounts.has(row.resolved_account_id) ) { backfilledAccounts.add(row.resolved_account_id); const claimantContact = row.claimant_contact?.trim() || ""; await db .prepare( `UPDATE accounts SET current_contact = CASE WHEN ? != '' THEN ? ELSE current_contact END, post_count = ( SELECT COUNT(*) FROM distributions WHERE account_id = ? ) WHERE id = ?`, ) .bind( claimantContact, claimantContact, row.resolved_account_id, row.resolved_account_id, ) .run(); } if (attempted >= Math.max(1, Math.min(25, limit))) break; const noteId = (() => { try { const url = new URL(row.publish_url ?? ""); return ( url.pathname.match( /\/(?:discovery\/item|explore)\/([A-Za-z0-9_-]{8,80})/, )?.[1] ?? "" ); } catch { return ""; } })(); const isDemoAccount = row.platform_uid?.startsWith("xhs-"); let attemptedThisRow = false; let publicProfileUpdated = false; if ( row.platform === "小红书" && !isDemoAccount && isVerifiedXhsProfileUrl(row.profile_url) && (!row.public_account_id || Number(row.followers ?? 0) === 0 || !row.gender || !row.bio) ) { attempted += 1; attemptedThisRow = true; const details = await resolveProfileDetailsFromMcp( row.profile_url ?? "", "小红书", mcpConfig, ).catch(async () => ({ ...(await resolveXhsPublicAccountDetails(row.profile_url ?? "")), gender: "" as const, bio: "", recentNoteTitles: [] as string[], providerTags: [] as string[], })); if ( row.account_id && (details.redId || details.followers !== null || details.gender || details.bio || details.recentNoteTitles.length > 0) ) { await db .prepare( `UPDATE accounts SET public_account_id = CASE WHEN ? != '' THEN ? ELSE public_account_id END, followers = CASE WHEN ? IS NOT NULL AND (? > 0 OR followers = 0) THEN ? ELSE followers END, ip_location = CASE WHEN ? != '' AND ? != '待识别' THEN ? ELSE ip_location END, gender = CASE WHEN ? != '' THEN ? ELSE gender END, bio = CASE WHEN ? != '' THEN ? ELSE bio END, last_seen_at = CURRENT_TIMESTAMP WHERE id = ?`, ) .bind( details.redId, details.redId, details.followers, details.followers, details.followers, details.ipLocation ?? "", details.ipLocation ?? "", details.ipLocation ?? "", details.gender, details.gender, details.bio, details.bio, row.account_id, ) .run(); publicProfileUpdated = true; } if ( (details.redId || row.public_account_id) && Number(details.followers ?? row.followers ?? 0) > 0 ) { updated += 1; continue; } if (attempted >= Math.max(1, Math.min(25, limit))) { if (publicProfileUpdated) updated += 1; continue; } } const needsEnrichment = !isDemoAccount && (!row.resolved_account_id || (row.platform === "小红书" && !isVerifiedXhsProfileUrl(row.profile_url)) || (row.platform === "抖音" && !isVerifiedDouyinProfileUrl(row.profile_url)) || (row.platform === "小红书" && !row.public_account_id) || (row.platform === "抖音" && !row.public_account_id) || (row.platform === "小红书" && Number(row.followers ?? 0) === 0) || (row.platform === "抖音" && Number(row.followers ?? 0) === 0) || row.platform_uid?.startsWith("pending-") || Boolean(noteId && row.platform_uid === noteId)); if (!needsEnrichment || !row.publish_url) { if (publicProfileUpdated) updated += 1; continue; } if (!attemptedThisRow) attempted += 1; try { const result = await enrichDistributionAccount( db, row.id, row.publish_url, row.nickname || "待识别账号", mcpConfig, ); if (result.updated || publicProfileUpdated) updated += 1; } catch { if (publicProfileUpdated) updated += 1; else failed += 1; } } return { attempted, updated, failed }; }