feat: 批量更新账号资料支持勾选字段（含AI打标）
- 更新全部账号ID 升级为 批量更新账号资料：弹窗勾选 账号ID/昵称/粉丝数/IP属地/性别/简介/标签 - 标签勾选时调用 LLM 重新打标（覆盖已有标签），AI_TAGGING_TOOL 未配置自动跳过 - 只写入选中字段，爬取为空的字段跳过不清空；演示账号（xhs- 前缀）不参与 - .env.app.example 为 SEED_DEMO_DATA 补充注释
This commit is contained in:
ABAPPLO
253
lib/account-field-refresh-service.ts
Normal file
253
lib/account-field-refresh-service.ts
Normal file
@@ -0,0 +1,253 @@
|
||||
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;
|
||||
}
|
||||
@@ -1,134 +0,0 @@
|
||||
import { runInBackground } from "./background";
|
||||
import {
|
||||
resolveProfileDetailsFromMcp,
|
||||
type CollectionMcpConfig,
|
||||
} from "./mcp-collection-client";
|
||||
import type { DatabaseClient } from "./database";
|
||||
|
||||
export type PublicAccountIdRefreshStatus = {
|
||||
status: "running" | "done";
|
||||
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 ProfileScraper = (
|
||||
profileUrl: string,
|
||||
platform: "小红书" | "抖音",
|
||||
) => Promise<{ redId: string | null }>;
|
||||
|
||||
const defaultScraper =
|
||||
(config: CollectionMcpConfig): ProfileScraper =>
|
||||
(profileUrl, platform) =>
|
||||
resolveProfileDetailsFromMcp(profileUrl, platform, config);
|
||||
|
||||
let currentJob: PublicAccountIdRefreshStatus | null = null;
|
||||
|
||||
export function getPublicAccountIdRefreshStatus(): PublicAccountIdRefreshStatus | null {
|
||||
return currentJob ? { ...currentJob } : null;
|
||||
}
|
||||
|
||||
export function startPublicAccountIdRefresh(
|
||||
db: DatabaseClient,
|
||||
config: CollectionMcpConfig,
|
||||
): PublicAccountIdRefreshStatus {
|
||||
if (currentJob?.status === "running") {
|
||||
return { ...currentJob };
|
||||
}
|
||||
const job: PublicAccountIdRefreshStatus = {
|
||||
status: "running",
|
||||
total: 0,
|
||||
processed: 0,
|
||||
updated: 0,
|
||||
failed: 0,
|
||||
skipped: 0,
|
||||
startedAt: new Date().toISOString(),
|
||||
finishedAt: null,
|
||||
};
|
||||
currentJob = job;
|
||||
runInBackground(
|
||||
executePublicAccountIdRefresh(job, db, defaultScraper(config)),
|
||||
"public account id refresh",
|
||||
);
|
||||
return { ...job };
|
||||
}
|
||||
|
||||
async function executePublicAccountIdRefresh(
|
||||
job: PublicAccountIdRefreshStatus,
|
||||
db: DatabaseClient,
|
||||
scraper: ProfileScraper,
|
||||
) {
|
||||
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;
|
||||
|
||||
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 redId = details.redId?.trim();
|
||||
if (redId) {
|
||||
await db
|
||||
.prepare(`UPDATE accounts SET public_account_id = ? WHERE id = ?`)
|
||||
.bind(redId, row.id)
|
||||
.run();
|
||||
job.updated += 1;
|
||||
} else {
|
||||
job.skipped += 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 runPublicAccountIdRefresh(
|
||||
db: DatabaseClient,
|
||||
scraper: ProfileScraper,
|
||||
): Promise<PublicAccountIdRefreshStatus> {
|
||||
const job: PublicAccountIdRefreshStatus = {
|
||||
status: "running",
|
||||
total: 0,
|
||||
processed: 0,
|
||||
updated: 0,
|
||||
failed: 0,
|
||||
skipped: 0,
|
||||
startedAt: new Date().toISOString(),
|
||||
finishedAt: null,
|
||||
};
|
||||
await executePublicAccountIdRefresh(job, db, scraper);
|
||||
return job;
|
||||
}
|
||||
Reference in New Issue
Block a user