koc-loop/app/api/resources-import/route.ts

import { isManagerRequest, managerForbidden } from "../../../lib/user-auth";
import { ensureSchema, getRawDb } from "../../../lib/mvp-db";
import {
  resolveCollectionMcpConfig,
  resolveXhsProfileDetailsFromMcp,
  resolveXhsPublicAccountDetails,
  type CollectionMcpBindings,
} from "../../../lib/mcp-collection-client";
import { getRuntimeEnv } from "../../../lib/runtime-env";
import {
  mergeCooperationSources,
  normalizeProfileUrl,
  parseResourceImportFile,
  RESOURCE_IMPORT_MAX_BYTES,
  resourcePlatformUid,
  resourceImportMissingFields,
  type ResourceImportRow,
} from "../../../lib/resource-import";

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

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

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

async function loadAccounts() {
  return getRawDb()
    .prepare(
      `SELECT id, platform, platform_uid, public_account_id, nickname,
              profile_url, ip_location, followers, 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;
}

async function enrichRows(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);
    }
  }
  const mcpConfig = resolveCollectionMcpConfig(
    getRuntimeEnv() as unknown as CollectionMcpBindings,
  );
  return mapConcurrent(rows, 4, async (row) => {
    if (row.errors.length > 0 || !row.profileUrl || row.platform !== "小红书") {
      return row;
    }
    const existing = existingByProfile.get(identityKey(row.platform, row.profileUrl));
    const existingIpLocation =
      existing?.ip_location && existing.ip_location !== "待识别"
        ? existing.ip_location
        : "";
    const baseline: ResourceImportRow = {
      ...row,
      nickname: row.nickname || existing?.nickname || "",
      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,
    };
    if (resourceImportMissingFields(baseline).length === 0) {
      return baseline;
    }

    let details = await resolveXhsProfileDetailsFromMcp(row.profileUrl, mcpConfig).catch(
      () => ({ nickname: null, redId: null, followers: null, ipLocation: null }),
    );
    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,
    };
    if (resourceImportMissingFields(mcpResult).length > 0) {
      const publicDetails = await resolveXhsPublicAccountDetails(row.profileUrl).catch(
        () => ({ nickname: null, redId: null, followers: null, ipLocation: null }),
      );
      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,
    };
  });
}

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,
        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 summarize(rows: AnalyzedRow[]) {
  return {
    total: rows.length,
    create: rows.filter((row) => row.action === "create").length,
    update: rows.filter((row) => row.action === "update").length,
    error: rows.filter((row) => row.action === "error").length,
  };
}

export async function POST(request: Request) {
  if (!(await isManagerRequest(request))) return managerForbidden();
  try {
    await ensureSchema();
    const form = await request.formData();
    const file = form.get("file");
    const mode = String(form.get("mode") ?? "preview");
    if (!(file instanceof File)) {
      return Response.json({ error: "请选择需要导入的 Excel 文件" }, { status: 400 });
    }
    if (file.size <= 0 || file.size > RESOURCE_IMPORT_MAX_BYTES) {
      return Response.json({ error: "文件不能为空，且不能超过 5MB" }, { status: 400 });
    }
    const rows = parseResourceImportFile(file.name, new Uint8Array(await file.arrayBuffer()));
    const accounts = await loadAccounts();
    const enriched = await enrichRows(rows, accounts.results);
    const analyzed = analyzeRows(enriched, accounts.results);
    const summary = summarize(analyzed);
    if (mode !== "commit") {
      return Response.json({
        summary,
        rows: analyzed.slice(0, 100).map((row) => ({
          rowNumber: row.rowNumber,
          platform: row.platform,
          nickname: row.nickname,
          publicAccountId: row.publicAccountId,
          profileUrl: row.profileUrl,
          ipLocation: row.ipLocation,
          followers: row.followers,
          cooperationSource: row.cooperationSource,
          action: row.action,
          errors: row.errors,
        })),
        truncated: analyzed.length > 100,
      });
    }
    if (summary.error > 0) {
      return Response.json(
        { error: `有 ${summary.error} 行数据未通过校验，请修正后重新上传`, summary },
        { status: 400 },
      );
    }

    const db = getRawDb();
    const statements = analyzed.map((row) =>
      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,
                 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.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,
                 cooperation_source)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, 0, ?)`,
            )
            .bind(
              row.accountId,
              row.platform,
              row.platformUid,
              row.publicAccountId,
              row.nickname,
              row.profileUrl,
              row.ipLocation,
              row.followers,
              row.cooperationSource,
            ),
    );
    if (statements.length > 0) await db.batch(statements);
    return Response.json({
      summary,
      message: `已导入 ${summary.create} 个新账号，更新 ${summary.update} 个已有账号`,
    });
  } catch (error) {
    const message = error instanceof Error ? error.message : "导入失败";
    return Response.json({ error: message }, { status: 400 });
  }
}