koc-loop/lib/resource-write.ts

import { ensureSchema, getRawDb } from "./mvp-db";
import {
  generateAccountTagsFromMcp,
  resolveCollectionMcpConfig,
  resolveProfileDetailsFromMcp,
  resolveXhsPublicAccountDetails,
  type AccountTaggingEnv,
  type CollectionMcpBindings,
} from "./mcp-collection-client";
import { getRuntimeEnv } from "./runtime-env";
import {
  mergeCooperationSources,
  normalizeProfileUrl,
  resourceImportMissingFields,
  resourcePlatformUid,
  type ResourceImportRow,
} from "./resource-import";

export type AccountRow = {
  id: string;
  platform: string;
  platform_uid: string;
  public_account_id: string;
  nickname: string;
  profile_url: string;
  ip_location: string;
  followers: number;
  gender: string;
  bio: string;
  tags: string;
  cooperation_source: string;
};

export type AnalyzedRow = ResourceImportRow & {
  action: "create" | "update" | "error";
  accountId: string;
  platformUid: string;
  cooperationSource: string;
};

export const RESOURCE_IMPORT_SYNC_ENRICH_ROWS = 100;
const RESOURCE_IMPORT_DB_BATCH_SIZE = 100;

export function identityKey(platform: string, value: string) {
  return `${platform.trim().toLocaleLowerCase("zh-CN")}|${value
    .trim()
    .toLocaleLowerCase("zh-CN")}`;
}

export async function loadAccounts() {
  await ensureSchema();
  return getRawDb()
    .prepare(
      `SELECT id, platform, platform_uid, public_account_id, nickname,
              profile_url, ip_location, followers, gender, bio, tags,
              cooperation_source
       FROM accounts`,
    )
    .all<AccountRow>();
}

async function mapConcurrent<T, R>(
  items: T[],
  limit: number,
  worker: (item: T) => Promise<R>,
) {
  const results = new Array<R>(items.length);
  let cursor = 0;
  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, async () => {
      while (cursor < items.length) {
        const index = cursor;
        cursor += 1;
        results[index] = await worker(items[index]);
      }
    }),
  );
  return results;
}

export function mergeExistingFields(
  rows: ResourceImportRow[],
  accounts: AccountRow[],
) {
  const existingByProfile = new Map<string, AccountRow>();
  for (const account of accounts) {
    const profileUrl = normalizeProfileUrl(account.profile_url || "");
    if (profileUrl) {
      existingByProfile.set(identityKey(account.platform, profileUrl), account);
    }
  }
  return rows.map((row) => {
    if (
      row.errors.length > 0 ||
      !row.profileUrl ||
      !["小红书", "抖音"].includes(row.platform)
    ) {
      return row;
    }
    const existing = existingByProfile.get(identityKey(row.platform, row.profileUrl));
    const existingNickname =
      existing?.nickname && existing.nickname !== "待识别账号"
        ? existing.nickname
        : "";
    const existingIpLocation =
      existing?.ip_location && existing.ip_location !== "待识别"
        ? existing.ip_location
        : "";
    const existingGender: ResourceImportRow["gender"] =
      existing?.gender === "男" || existing?.gender === "女"
        ? existing.gender
        : "";
    return {
      ...row,
      nickname: row.nickname || existingNickname,
      publicAccountId: row.publicAccountId || existing?.public_account_id || "",
      ipLocation: row.ipLocation || existingIpLocation,
      followers: row.followersResolved
        ? row.followers
        : Number(existing?.followers || 0),
      followersResolved:
        row.followersResolved || Number(existing?.followers || 0) > 0,
      gender: row.gender || existingGender,
      bio: row.bio || existing?.bio || "",
      tags: row.tags.length > 0
        ? row.tags
        : (existing?.tags || "")
            .split(/[,，、;；|]/)
            .map((item) => item.trim())
            .filter(Boolean)
            .slice(0, 5),
    };
  });
}

export async function enrichRows(
  rows: ResourceImportRow[],
  accounts: AccountRow[],
) {
  const env = getRuntimeEnv() as unknown as CollectionMcpBindings;
  const mcpConfig = resolveCollectionMcpConfig(env);
  const baselineRows = mergeExistingFields(rows, accounts);
  return mapConcurrent(baselineRows, 4, async (baseline) => {
    const row = baseline;
    if (
      row.errors.length > 0 ||
      !row.profileUrl ||
      !["小红书", "抖音"].includes(row.platform)
    ) {
      return row;
    }
    if (resourceImportMissingFields(baseline).length === 0) {
      return baseline;
    }

    let details: {
      nickname: string | null;
      redId: string | null;
      followers: number | null;
      ipLocation: string | null;
      gender: "" | "男" | "女";
      bio: string;
      recentNoteTitles: string[];
      providerTags: string[];
    } = await resolveProfileDetailsFromMcp(
      row.profileUrl,
      row.platform === "抖音" ? "抖音" : "小红书",
      mcpConfig,
    ).catch(() => ({
      nickname: null,
      redId: null,
      followers: null,
      ipLocation: null,
      gender: "" as const,
      bio: "",
      recentNoteTitles: [],
      providerTags: [],
    }));
    const mcpResult = {
      nickname: baseline.nickname || details.nickname?.trim() || "",
      publicAccountId: baseline.publicAccountId || details.redId?.trim() || "",
      ipLocation: baseline.ipLocation || details.ipLocation?.trim() || "",
      followersResolved: baseline.followersResolved || details.followers !== null,
      gender: baseline.gender || details.gender,
      bio: baseline.bio || details.bio,
      tags: baseline.tags,
    };
    if (
      row.platform === "小红书" &&
      resourceImportMissingFields(mcpResult).length > 0
    ) {
      const publicDetails = await resolveXhsPublicAccountDetails(row.profileUrl).catch(
        () => ({ nickname: null, redId: null, followers: null, ipLocation: null }),
      );
      details = {
        ...details,
        nickname: details.nickname || publicDetails.nickname,
        redId: details.redId || publicDetails.redId,
        followers: details.followers ?? publicDetails.followers,
        ipLocation: details.ipLocation || publicDetails.ipLocation,
      };
    }
    return {
      ...baseline,
      nickname: baseline.nickname || details.nickname?.trim() || "待识别账号",
      publicAccountId: baseline.publicAccountId || details.redId?.trim() || "",
      ipLocation: baseline.ipLocation || details.ipLocation?.trim() || "待识别",
      followers: baseline.followersResolved
        ? baseline.followers
        : (details.followers ?? 0),
      followersResolved:
        baseline.followersResolved || details.followers !== null,
      gender: baseline.gender || details.gender,
      bio: baseline.bio || details.bio,
      tags: baseline.tags,
    };
  });
}

type UntaggedAccountRow = {
  id: string;
  platform: string;
  platform_uid: string;
  profile_url: string;
  nickname: string;
};

export async function tagUntaggedAccounts(
  rows: Array<Pick<AnalyzedRow, "platform" | "platformUid">>,
) {
  const env = getRuntimeEnv() as unknown as CollectionMcpBindings &
    AccountTaggingEnv;
  const toolName = String(env.AI_TAGGING_TOOL ?? "").trim();
  const refs = new Set(
    rows
      .filter((row) => row.platformUid)
      .map((row) => `${row.platform}:${row.platformUid}`),
  );
  if (!toolName || refs.size === 0) return;

  const db = getRawDb();
  const uids = [...new Set([...refs].map((key) => key.slice(key.indexOf(":") + 1)))];
  const candidates: UntaggedAccountRow[] = [];
  for (let index = 0; index < uids.length; index += 100) {
    const chunk = uids.slice(index, index + 100);
    const placeholders = chunk.map(() => "?").join(", ");
    const result = await db
      .prepare(
        `SELECT id, platform, platform_uid, profile_url, nickname
         FROM accounts
         WHERE (tags = '' OR tags IS NULL)
           AND profile_url IS NOT NULL AND profile_url != ''
           AND platform_uid IN (${placeholders})`,
      )
      .bind(...chunk)
      .all<UntaggedAccountRow>();
    candidates.push(...result.results);
  }
  const targets = candidates.filter((account) =>
    refs.has(`${account.platform}:${account.platform_uid}`),
  );
  if (targets.length === 0) return;

  const mcpConfig = resolveCollectionMcpConfig(env);
  await mapConcurrent(targets, 4, async (account) => {
    const details = await resolveProfileDetailsFromMcp(
      account.profile_url,
      account.platform === "抖音" ? "抖音" : "小红书",
      mcpConfig,
    ).catch(() => null);
    if (
      !details ||
      (!details.bio &&
        details.providerTags.length === 0 &&
        details.recentNoteTitles.length === 0)
    ) {
      return;
    }
    const tags = await generateAccountTagsFromMcp(
      {
        platform: account.platform,
        nickname: details.nickname?.trim() || account.nickname || "",
        bio: details.bio,
        gender: details.gender,
        ipLocation: details.ipLocation ?? "",
        followers: details.followers,
        providerTags: details.providerTags,
        recentNoteTitles: details.recentNoteTitles,
      },
      mcpConfig,
      { AI_TAGGING_TOOL: toolName },
    ).catch(() => []);
    if (tags.length === 0) return;
    await db
      .prepare(
        `UPDATE accounts SET tags = ? WHERE id = ? AND (tags = '' OR tags IS NULL)`,
      )
      .bind(tags.join(","), account.id)
      .run();
  });
}

export function analyzeRows(
  rows: ResourceImportRow[],
  accountRows: AccountRow[],
) {
  const profileMap = new Map<string, AccountRow>();
  const publicIdMap = new Map<string, AccountRow>();
  const platformUidMap = new Map<string, AccountRow>();
  for (const account of accountRows) {
    const profileUrl = normalizeProfileUrl(account.profile_url || "");
    if (profileUrl) profileMap.set(identityKey(account.platform, profileUrl), account);
    if (account.public_account_id) {
      publicIdMap.set(identityKey(account.platform, account.public_account_id), account);
    }
    platformUidMap.set(identityKey(account.platform, account.platform_uid), account);
  }

  return rows.map<AnalyzedRow>((row) => {
    const platformUid = resourcePlatformUid(row);
    const profileMatch = row.profileUrl
      ? profileMap.get(identityKey(row.platform, row.profileUrl))
      : undefined;
    const publicIdMatch = row.publicAccountId
      ? publicIdMap.get(identityKey(row.platform, row.publicAccountId))
      : undefined;
    const uidMatch = platformUidMap.get(identityKey(row.platform, platformUid));
    const matches = [profileMatch, publicIdMatch, uidMatch].filter(
      (account): account is AccountRow => Boolean(account),
    );
    const matchedIds = [...new Set(matches.map((account) => account.id))];
    const errors = [...row.errors];
    if (matchedIds.length > 1) {
      errors.push("账号主页和账号ID匹配到不同的现有账号，请先核对");
    }
    const existing = matchedIds.length === 1 ? matches[0] : undefined;
    const accountId = existing?.id ?? `account-${crypto.randomUUID().slice(0, 12)}`;
    const analyzed: AnalyzedRow = {
      ...row,
      errors,
      action: errors.length > 0 ? "error" : existing ? "update" : "create",
      accountId,
      platformUid: existing?.platform_uid ?? platformUid,
      cooperationSource: mergeCooperationSources(
        existing?.cooperation_source ?? "",
        row.cooperationSource,
      ),
    };
    if (analyzed.action !== "error") {
      const virtual: AccountRow = {
        id: accountId,
        platform: row.platform,
        platform_uid: analyzed.platformUid,
        public_account_id: row.publicAccountId || existing?.public_account_id || "",
        nickname: row.nickname,
        profile_url: row.profileUrl || existing?.profile_url || "",
        ip_location: row.ipLocation || existing?.ip_location || "待识别",
        followers: row.followers || existing?.followers || 0,
        gender: row.gender || existing?.gender || "",
        bio: row.bio || existing?.bio || "",
        tags: (row.tags.length > 0
          ? row.tags
          : (existing?.tags || "").split(/[,，、;；|]/).filter(Boolean)
        ).slice(0, 5).join(","),
        cooperation_source: analyzed.cooperationSource,
      };
      if (row.profileUrl) profileMap.set(identityKey(row.platform, row.profileUrl), virtual);
      if (row.publicAccountId) {
        publicIdMap.set(identityKey(row.platform, row.publicAccountId), virtual);
      }
      platformUidMap.set(identityKey(row.platform, analyzed.platformUid), virtual);
    }
    return analyzed;
  });
}

function statementForAnalyzedRow(
  db: ReturnType<typeof getRawDb>,
  row: AnalyzedRow,
) {
  return row.action === "update"
    ? db
        .prepare(
          `UPDATE accounts SET
             nickname = ?,
             public_account_id = CASE WHEN ? != '' THEN ? ELSE public_account_id END,
             profile_url = CASE WHEN ? != '' THEN ? ELSE profile_url END,
             ip_location = CASE
               WHEN ? != '' AND ? != '待识别' THEN ? ELSE ip_location END,
             followers = CASE WHEN ? = 1 THEN ? ELSE followers END,
             gender = CASE WHEN ? != '' THEN ? ELSE gender END,
             bio = CASE WHEN ? != '' THEN ? ELSE bio END,
             tags = CASE WHEN ? != '' THEN ? ELSE tags END,
             cooperation_source = ?,
             last_seen_at = CURRENT_TIMESTAMP
           WHERE id = ?`,
        )
        .bind(
          row.nickname || "待识别账号",
          row.publicAccountId,
          row.publicAccountId,
          row.profileUrl,
          row.profileUrl,
          row.ipLocation,
          row.ipLocation,
          row.ipLocation,
          row.followersResolved ? 1 : 0,
          row.followers,
          row.gender,
          row.gender,
          row.bio,
          row.bio,
          row.tags.join(","),
          row.tags.join(","),
          row.cooperationSource,
          row.accountId,
        )
    : db
        .prepare(
          `INSERT INTO accounts
            (id, platform, platform_uid, public_account_id, nickname,
             profile_url, ip_location, followers, post_count, avg_views,
             gender, bio, tags, cooperation_source)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?, ?, ?, ?)`,
        )
        .bind(
          row.accountId,
          row.platform,
          row.platformUid,
          row.publicAccountId,
          row.nickname || "待识别账号",
          row.profileUrl,
          row.ipLocation || "待识别",
          row.followers,
          row.gender,
          row.bio,
          row.tags.join(","),
          row.cooperationSource,
        );
}

export async function writeAnalyzedRows(rows: AnalyzedRow[]) {
  const db = getRawDb();
  const statements = rows
    .filter((row) => row.action !== "error")
    .map((row) => statementForAnalyzedRow(db, row));
  for (let index = 0; index < statements.length; index += RESOURCE_IMPORT_DB_BATCH_SIZE) {
    await db.batch(statements.slice(index, index + RESOURCE_IMPORT_DB_BATCH_SIZE));
  }
}