koc-loop/lib/account-enrichment-service.ts

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<DistributionAccountRow>();
  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<BackfillRow>();
  let attempted = 0;
  let updated = 0;
  let failed = 0;
  const backfilledAccounts = new Set<string>();

  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 };
}