koc-loop/lib/account-field-refresh-service.ts

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<AccountRefreshField, string> = {
  publicAccountId: "账号ID（小红书号/抖音号）",
  nickname: "昵称",
  followers: "粉丝数",
  ipLocation: "IP属地",
  gender: "性别",
  bio: "简介",
  tags: "标签（AI 重新打标，覆盖已有标签）",
};

const FIELD_COLUMNS: Record<AccountRefreshField, string> = {
  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<ScrapedProfile>;

export type TagGenerator = (input: {
  platform: string;
  nickname: string;
  bio: string;
  gender: string;
  ipLocation: string;
  followers: number | null;
  providerTags: string[];
  recentNoteTitles: string[];
}) => Promise<string[]>;

function scrapedFieldValues(
  details: ScrapedProfile,
): Partial<Record<AccountRefreshField, string | number>> {
  const values: Partial<Record<AccountRefreshField, string | number>> = {};
  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<RefreshTargetRow>();
  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<AccountFieldRefreshStatus> {
  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;
}