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

import {
  resolveXhsPublicAccountDetails,
  resolveXhsAccountProfileFromMcp,
  resolveXhsProfileDetailsFromMcp,
  type CollectionMcpConfig,
} from "./mcp-collection-client";
import { hashText } from "./mvp-db";

type DistributionAccountRow = {
  id: string;
  account_id: string | null;
  publish_url: string | null;
};

type BackfillRow = DistributionAccountRow & {
  nickname: string | null;
  platform: string | null;
  platform_uid: string | null;
  public_account_id: string | null;
  profile_url: string | null;
  followers: number | 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;
  }
}

export async function enrichDistributionAccount(
  db: D1Database,
  distributionId: string,
  publishUrl: string,
  fallbackNickname: string,
  mcpConfig: CollectionMcpConfig,
) {
  const profile = await resolveXhsAccountProfileFromMcp(
    publishUrl,
    fallbackNickname,
    mcpConfig,
  );
  const current = await db
    .prepare(
      `SELECT id, account_id, publish_url
       FROM distributions
       WHERE id = ?`,
    )
    .bind(distributionId)
    .first<DistributionAccountRow>();
  if (!current || current.publish_url !== publishUrl) {
    return { updated: false, reason: "stale" as const };
  }

  const canonicalAccountId = `account-${hashText(
    `小红书:${profile.platformUid}`,
  )}`;
  if (current.account_id === canonicalAccountId) {
    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,
          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,
        canonicalAccountId,
      )
      .run();
    return {
      updated: true,
      accountId: canonicalAccountId,
      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, 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,
           post_count = accounts.post_count + 1,
           last_seen_at = CURRENT_TIMESTAMP`,
      )
      .bind(
        canonicalAccountId,
        profile.platformUid,
        profile.redId,
        profile.nickname || fallbackNickname,
        profile.profileUrl,
        profile.ipLocation,
        profile.followers ?? 0,
      ),
    db
      .prepare(
        `UPDATE distributions SET
          account_id = ?,
          updated_at = CURRENT_TIMESTAMP
         WHERE id = ? AND publish_url = ?`,
      )
      .bind(canonicalAccountId, distributionId, publishUrl),
  ]);

  if (provisionalAccountId) {
    await db
      .prepare(
        `DELETE FROM accounts
         WHERE id = ?
           AND id != ?
           AND NOT EXISTS (
             SELECT 1 FROM distributions WHERE account_id = ?
           )`,
      )
      .bind(
        provisionalAccountId,
        canonicalAccountId,
        provisionalAccountId,
      )
      .run();
  }
  return {
    updated: true,
    accountId: canonicalAccountId,
    profileUrl: profile.profileUrl,
  };
}

export async function backfillAccountProfiles(
  db: D1Database,
  mcpConfig: CollectionMcpConfig,
  limit = 10,
) {
  const rows = await db
    .prepare(
      `SELECT
        d.id,
        d.account_id,
        d.publish_url,
        a.nickname,
        a.platform,
        a.platform_uid,
        a.public_account_id,
        a.profile_url,
        a.followers
       FROM distributions d
       LEFT JOIN accounts a ON a.id = d.account_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;

  for (const row of rows.results) {
    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)
    ) {
      attempted += 1;
      attemptedThisRow = true;
      const details = await resolveXhsProfileDetailsFromMcp(
        row.profile_url ?? "",
        mcpConfig,
      ).catch(() =>
        resolveXhsPublicAccountDetails(row.profile_url ?? ""),
      );
      if (
        row.account_id &&
        (details.redId || details.followers !== null)
      ) {
        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,
                 last_seen_at = CURRENT_TIMESTAMP
             WHERE id = ?`,
          )
          .bind(
            details.redId,
            details.redId,
            details.followers,
            details.followers,
            details.followers,
            details.ipLocation ?? "",
            details.ipLocation ?? "",
            details.ipLocation ?? "",
            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.account_id ||
      (row.platform === "小红书" &&
        !isVerifiedXhsProfileUrl(row.profile_url)) ||
      (row.platform === "小红书" && !row.public_account_id) ||
      (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 };
}