import { mkdir, writeFile } from 'node:fs/promises'; import { Pool } from 'pg'; import type { JustOneCreator } from './justone.service.ts'; import { extractWorkbookTablesFromBuffer } from '../utils/file-parser.ts'; const PROVIDER_ARCHIVE_DIR = 'docs/provider-archives'; const RAW_CACHE_TTL_MS = 30 * 24 * 60 * 60 * 1000; const EMBEDDING_DIMENSION = 384; const pool = new Pool({ connectionString: process.env.DATABASE_URL || 'postgres://tihao_ai_app:tihao_ai_app@localhost:5432/tihao_ai', }); interface SearchCriteria { platforms: string[]; keywords: string[]; fanRange: { min: number; max: number }; budgetRange: { min: number; max: number }; region?: string | string[]; gender?: string; contentTags?: string[]; excludeTags?: string[]; } interface ParsedCreator extends JustOneCreator { sourceKind: 'uploaded' | 'company' | 'confirmed_api'; sourceFile?: string; embeddingText: string; sourceConfidence?: number; } export async function ingestUploadedCreatorLibrary(buffer: Buffer, fileName: string): Promise { const sheets = extractWorkbookTablesFromBuffer(buffer); const creators = sheets.flatMap((sheet) => parseCreatorRows(sheet.rows, fileName)); if (creators.length === 0) return 0; await upsertCleanCreators(creators); return creators.length; } export async function searchLocalCreators(criteria: SearchCriteria, limit = 200): Promise { const normalized = normalizeCriteria(criteria); const queryText = buildCriteriaEmbeddingText(normalized); const embedding = buildEmbedding(queryText); const platformFilters = normalized.platforms.map(mapPlatform).filter(Boolean); const params: unknown[] = [embedding, limit]; const where: string[] = []; if (platformFilters.length > 0) { params.push(platformFilters); where.push(`platform = ANY($${params.length}::text[])`); } if (normalized.fanRange?.min) { params.push(normalized.fanRange.min); where.push(`fans_count >= $${params.length}`); } if (normalized.fanRange?.max) { params.push(normalized.fanRange.max); where.push(`fans_count <= $${params.length}`); } const sql = ` SELECT *, cosine_similarity(embedding, $1::double precision[]) AS similarity FROM creator_profile_clean ${where.length ? `WHERE ${where.join(' AND ')}` : ''} ORDER BY cosine_similarity(embedding, $1::double precision[]) DESC, updated_at DESC LIMIT $2 `; const result = await pool.query(sql, params); const cleanCreators = result.rows .map((row) => ({ creator: rowToCreator(row), score: scoreRetrievedCreator(row, normalized) })) .filter((item) => item.score > 0) .sort((a, b) => b.score - a.score) .slice(0, limit) .map((item) => item.creator); const cachedProviderCreators = await searchCachedProviderCreators( normalized, Math.max(0, limit - cleanCreators.length), ); return mergeCreators([...cleanCreators, ...cachedProviderCreators]).slice(0, limit); } export async function archiveProviderResponse(params: { provider: string; endpoint: string; requestParams: Record; responseBody: unknown; }): Promise { const fetchedAt = new Date(); const expiresAt = new Date(fetchedAt.getTime() + RAW_CACHE_TTL_MS); const safeEndpoint = params.endpoint.replace(/[^a-z0-9]+/gi, '-').replace(/^-|-$/g, '').slice(0, 80); const archiveDir = `${PROVIDER_ARCHIVE_DIR}/${fetchedAt.toISOString().slice(0, 7)}`; await mkdir(archiveDir, { recursive: true }); const id = `${params.provider}_${fetchedAt.getTime()}_${Math.random().toString(36).slice(2, 8)}`; const archivePath = `${archiveDir}/${id}_${safeEndpoint}.json`; await writeFile(archivePath, JSON.stringify(params, null, 2), 'utf-8'); await pool.query( `INSERT INTO provider_raw_cache (provider, endpoint, request_params, response_body, fetched_at, expires_at, normalized_status, archive_path) VALUES ($1, $2, $3::jsonb, $4::jsonb, $5, $6, 'pending', $7)`, [ params.provider, params.endpoint, JSON.stringify(params.requestParams), JSON.stringify(params.responseBody), fetchedAt, expiresAt, archivePath, ], ); } export async function cacheProviderCreators(params: { provider: string; endpoint: string; requestParams: Record; creators: JustOneCreator[]; }): Promise { const creators = params.creators.filter((creator) => creator.userId && creator.nickname); if (creators.length === 0) return 0; const fetchedAt = new Date(); const expiresAt = new Date(fetchedAt.getTime() + RAW_CACHE_TTL_MS); const client = await pool.connect(); try { await client.query('BEGIN'); for (const creator of creators) { const embeddingText = buildCreatorEmbeddingText(creator); await client.query( `INSERT INTO provider_creator_raw_cache ( provider, endpoint, request_params, platform, platform_user_id, display_name, fans_count, liked_collect_count, content_type, content_tags, persona_tags, city, geo_location, profile_url, cooperation_method, image_price, video_price, min_price, cooperation_status, raw_creator, fetched_at, expires_at, normalized_status, embedding_text, embedding ) VALUES ( $1, $2, $3::jsonb, $4, $5, $6, $7, $8, $9, $10::jsonb, $11::jsonb, $12, $13, $14, $15, $16, $17, $18, $19, $20::jsonb, $21, $22, 'raw', $23, $24::double precision[] ) ON CONFLICT (provider, platform, platform_user_id) DO UPDATE SET endpoint = EXCLUDED.endpoint, request_params = EXCLUDED.request_params, display_name = EXCLUDED.display_name, fans_count = EXCLUDED.fans_count, liked_collect_count = EXCLUDED.liked_collect_count, content_type = EXCLUDED.content_type, content_tags = EXCLUDED.content_tags, persona_tags = EXCLUDED.persona_tags, city = EXCLUDED.city, geo_location = EXCLUDED.geo_location, profile_url = COALESCE(EXCLUDED.profile_url, provider_creator_raw_cache.profile_url), cooperation_method = EXCLUDED.cooperation_method, image_price = EXCLUDED.image_price, video_price = EXCLUDED.video_price, min_price = EXCLUDED.min_price, cooperation_status = EXCLUDED.cooperation_status, raw_creator = COALESCE(provider_creator_raw_cache.raw_creator, '{}'::jsonb) || EXCLUDED.raw_creator, fetched_at = EXCLUDED.fetched_at, expires_at = EXCLUDED.expires_at, normalized_status = 'raw', embedding_text = EXCLUDED.embedding_text, embedding = EXCLUDED.embedding`, [ params.provider, params.endpoint, JSON.stringify(params.requestParams), creator.platform, creator.userId, creator.nickname, creator.fansCount || 0, creator.likedCollectCount || 0, creator.contentType || creator.contentTags.join('、'), JSON.stringify(creator.contentTags || []), JSON.stringify(creator.personalTags || []), creator.city || creator.location || null, creator.geoLocation || creator.location || null, getCreatorProfileUrl(creator), creator.cooperationMethod || null, creator.imagePrice || 0, creator.videoPrice || 0, creator.minPrice || 0, creator.cooperationStatus || null, JSON.stringify(creator), fetchedAt, expiresAt, embeddingText, buildEmbedding(embeddingText), ], ); } await client.query('COMMIT'); return creators.length; } catch (error) { await client.query('ROLLBACK'); throw error; } finally { client.release(); } } export async function recordRetrievalEvent(params: { taskId?: string; queryText: string; criteria: unknown; localHitCount: number; providerHitCount: number; finalCount: number; }): Promise { await pool.query( `INSERT INTO creator_retrieval_event (task_id, query_text, criteria, local_hit_count, provider_hit_count, final_count) VALUES ($1, $2, $3::jsonb, $4, $5, $6)`, [ params.taskId || null, params.queryText, JSON.stringify(params.criteria || {}), params.localHitCount, params.providerHitCount, params.finalCount, ], ); } export async function updateProviderCreatorProfileUrl(params: { provider: string; platform: string; platformUserId: string; profileUrl: string; secUid?: string; uniqueId?: string; }): Promise { await pool.query( `UPDATE provider_creator_raw_cache SET profile_url = $4, raw_creator = coalesce(raw_creator, '{}'::jsonb) || $5::jsonb WHERE provider = $1 AND platform = $2 AND platform_user_id = $3`, [ params.provider, params.platform, params.platformUserId, params.profileUrl, JSON.stringify({ profileUrl: params.profileUrl, secUid: params.secUid || '', uniqueId: params.uniqueId || '', }), ], ); } async function upsertCleanCreators(creators: ParsedCreator[]): Promise { const client = await pool.connect(); try { await client.query('BEGIN'); for (const creator of creators) { const embeddingText = creator.embeddingText || buildCreatorEmbeddingText(creator); await client.query( `INSERT INTO creator_profile_clean ( platform, platform_user_id, display_name, gender, fans_count, liked_collect_count, content_type, content_tags, persona_tags, city, geo_location, profile_url, cooperation_method, image_price, video_price, min_price, cooperation_status, source_kind, source_provider, source_file, source_confidence, updated_at, embedding_text, embedding ) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8::jsonb, $9::jsonb, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, now(), $22, $23::double precision[] ) ON CONFLICT (platform, platform_user_id) DO UPDATE SET display_name = EXCLUDED.display_name, gender = EXCLUDED.gender, fans_count = EXCLUDED.fans_count, liked_collect_count = EXCLUDED.liked_collect_count, content_type = EXCLUDED.content_type, content_tags = EXCLUDED.content_tags, persona_tags = EXCLUDED.persona_tags, city = EXCLUDED.city, geo_location = EXCLUDED.geo_location, profile_url = EXCLUDED.profile_url, cooperation_method = EXCLUDED.cooperation_method, image_price = EXCLUDED.image_price, video_price = EXCLUDED.video_price, min_price = EXCLUDED.min_price, cooperation_status = EXCLUDED.cooperation_status, source_kind = EXCLUDED.source_kind, source_provider = EXCLUDED.source_provider, source_file = EXCLUDED.source_file, source_confidence = EXCLUDED.source_confidence, updated_at = now(), embedding_text = EXCLUDED.embedding_text, embedding = EXCLUDED.embedding`, [ creator.platform, creator.userId, creator.nickname, creator.gender || null, creator.fansCount || 0, creator.likedCollectCount || 0, creator.contentType || creator.contentTags.join('、'), JSON.stringify(creator.contentTags || []), JSON.stringify(creator.personalTags || []), creator.city || creator.location || null, creator.geoLocation || creator.location || null, creator.xiaohongshuUrl || null, creator.cooperationMethod || null, creator.imagePrice || 0, creator.videoPrice || 0, creator.minPrice || 0, creator.cooperationStatus || null, creator.sourceKind, creator.sourceProvider || 'local-upload', creator.sourceFile || null, creator.sourceConfidence || 90, embeddingText, buildEmbedding(embeddingText), ], ); } await client.query('COMMIT'); } catch (error) { await client.query('ROLLBACK'); throw error; } finally { client.release(); } } async function searchCachedProviderCreators(criteria: SearchCriteria, limit: number): Promise { if (limit <= 0) return []; const queryText = buildCriteriaEmbeddingText(criteria); const embedding = buildEmbedding(queryText); const platformFilters = criteria.platforms.map(mapPlatform).filter(Boolean); const params: unknown[] = [embedding, limit]; const where: string[] = ['expires_at > now()']; if (platformFilters.length > 0) { params.push(platformFilters); where.push(`platform = ANY($${params.length}::text[])`); } if (criteria.fanRange?.min) { params.push(criteria.fanRange.min); where.push(`fans_count >= $${params.length}`); } if (criteria.fanRange?.max) { params.push(criteria.fanRange.max); where.push(`fans_count <= $${params.length}`); } const result = await pool.query( `SELECT *, cosine_similarity(embedding, $1::double precision[]) AS similarity FROM provider_creator_raw_cache WHERE ${where.join(' AND ')} ORDER BY cosine_similarity(embedding, $1::double precision[]) DESC, fetched_at DESC LIMIT $2`, params, ); return result.rows .map((row) => ({ creator: rowToProviderRawCreator(row), score: scoreRetrievedCreator(row, criteria) })) .filter((item) => item.score > 0) .sort((a, b) => b.score - a.score) .map((item) => item.creator); } function parseCreatorRows(rows: string[][], sourceFile: string): ParsedCreator[] { const headerIndex = rows.findIndex((row) => row.some((cell) => normalizeHeader(cell) === '昵称')); if (headerIndex < 0) return []; const headers = rows[headerIndex].map(normalizeHeader); const getIndex = (...names: string[]) => headers.findIndex((header) => names.includes(header)); const indexMap = { nickname: getIndex('昵称', '账号名称'), gender: getIndex('性别'), fansWan: getIndex('粉丝数(万)', '粉丝数'), likedWan: getIndex('赞藏数(万)', '赞藏数'), contentType: getIndex('内容类型'), city: getIndex('城市'), geoLocation: getIndex('地理位置'), profileUrl: getIndex('小红书主页', '主页链接'), cooperationMethod: getIndex('合作方式'), imagePrice: headers.findIndex((header) => header.includes('图文') && header.includes('报价')), videoPrice: headers.findIndex((header) => header.includes('视频') && header.includes('报价')), }; if (indexMap.nickname < 0 || indexMap.profileUrl < 0) return []; return rows.slice(headerIndex + 1) .filter((row) => row[indexMap.nickname]) .map((row) => { const nickname = row[indexMap.nickname] || ''; const profileUrl = row[indexMap.profileUrl] || ''; const userId = extractProfileId(profileUrl) || hashText(`${sourceFile}:${nickname}:${profileUrl}`); const contentType = row[indexMap.contentType] || ''; const imagePrice = toNumber(row[indexMap.imagePrice]); const videoPrice = toNumber(row[indexMap.videoPrice]); const fansCount = toCountFromWan(row[indexMap.fansWan]); const likedCollectCount = toCountFromWan(row[indexMap.likedWan]); const city = row[indexMap.city] || ''; const geoLocation = row[indexMap.geoLocation] || ''; const contentTags = splitTags(contentType); const creator: ParsedCreator = { userId, platform: 'xiaohongshu', nickname, redId: userId, location: city || geoLocation, fansCount, imagePrice, videoPrice, minPrice: Math.min(...[imagePrice, videoPrice].filter((price) => price > 0)) || 0, cooperationStatus: row[indexMap.cooperationMethod] ? 'active' : '', personalTags: [], contentTags, sourceProvider: 'local-upload', sourceKind: 'uploaded', sourceFile, gender: row[indexMap.gender] || '', likedCollectCount, contentType, city, geoLocation, xiaohongshuUrl: profileUrl, cooperationMethod: row[indexMap.cooperationMethod] || '', embeddingText: '', }; creator.embeddingText = buildCreatorEmbeddingText(creator); return creator; }); } function rowToCreator(row: Record): JustOneCreator { return { userId: String(row.platform_user_id || ''), platform: String(row.platform || ''), nickname: String(row.display_name || ''), redId: String(row.platform_user_id || ''), location: String(row.city || row.geo_location || ''), fansCount: Number(row.fans_count || 0), imagePrice: Number(row.image_price || 0), videoPrice: Number(row.video_price || 0), minPrice: Number(row.min_price || 0), cooperationStatus: String(row.cooperation_status || ''), personalTags: Array.isArray(row.persona_tags) ? row.persona_tags as string[] : [], contentTags: Array.isArray(row.content_tags) ? row.content_tags as string[] : [], sourceProvider: String(row.source_provider || 'local-db'), gender: String(row.gender || ''), likedCollectCount: Number(row.liked_collect_count || 0), contentType: String(row.content_type || ''), city: String(row.city || ''), geoLocation: String(row.geo_location || ''), profileUrl: String(row.profile_url || ''), xiaohongshuUrl: String(row.platform || '') === 'xiaohongshu' ? String(row.profile_url || '') : '', cooperationMethod: String(row.cooperation_method || ''), }; } function rowToProviderRawCreator(row: Record): JustOneCreator { const rawCreator = parseRawCreator(row.raw_creator); return { userId: String(row.platform_user_id || ''), platform: String(row.platform || ''), nickname: String(row.display_name || ''), redId: String(row.platform_user_id || ''), location: String(row.city || row.geo_location || ''), fansCount: Number(row.fans_count || 0), imagePrice: Number(row.image_price || 0), videoPrice: Number(row.video_price || 0), minPrice: Number(row.min_price || 0), cooperationStatus: String(row.cooperation_status || ''), personalTags: Array.isArray(row.persona_tags) ? row.persona_tags as string[] : [], contentTags: Array.isArray(row.content_tags) ? row.content_tags as string[] : [], sourceProvider: `${String(row.provider || 'provider')}-raw-cache`, gender: '', likedCollectCount: Number(row.liked_collect_count || 0), contentType: String(row.content_type || ''), city: String(row.city || ''), geoLocation: String(row.geo_location || ''), profileUrl: String(row.profile_url || rawCreator.profileUrl || ''), coreUserId: String(rawCreator.coreUserId || ''), secUid: String(rawCreator.secUid || ''), uniqueId: String(rawCreator.uniqueId || ''), xiaohongshuUrl: String(row.platform || '') === 'xiaohongshu' ? String(row.profile_url || '') : '', cooperationMethod: String(row.cooperation_method || ''), commercialMetrics: parseRawObject(rawCreator.commercialMetrics), audienceMetrics: parseRawObject(rawCreator.audienceMetrics), }; } function parseRawCreator(value: unknown): Record { if (typeof value === 'object' && value !== null) return value as Record; if (typeof value !== 'string') return {}; try { const parsed = JSON.parse(value); return typeof parsed === 'object' && parsed !== null ? parsed as Record : {}; } catch { return {}; } } function parseRawObject(value: unknown): T | undefined { return typeof value === 'object' && value !== null ? value as T : undefined; } function mergeCreators(creators: JustOneCreator[]): JustOneCreator[] { const map = new Map(); for (const creator of creators) { const key = `${creator.platform}:${creator.userId}`; if (!map.has(key)) map.set(key, creator); } return Array.from(map.values()); } function scoreRetrievedCreator(row: Record, criteria: SearchCriteria): number { let score = Number(row.similarity || 0) * 70; const text = String(row.embedding_text || '').toLowerCase(); const keywords = [...(criteria.contentTags || []), ...(criteria.keywords || [])].map((item) => item.toLowerCase()); const regions = normalizeRegions(criteria.region); const meaningfulKeywords = keywords.filter((keyword) => isMeaningfulKeyword(keyword)); const hasMeaningfulKeywordMatch = meaningfulKeywords.length === 0 || meaningfulKeywords.some((keyword) => text.includes(keyword) || extractKeywordTerms(keyword).some((term) => text.includes(term))); if (!hasMeaningfulKeywordMatch) return 0; for (const keyword of keywords) { if (keyword && text.includes(keyword)) score += 12; } for (const region of regions) { if (region && text.includes(region.toLowerCase())) score += 10; } for (const exclude of criteria.excludeTags || []) { if (exclude && text.includes(exclude.toLowerCase())) score -= 4; } if (String(row.cooperation_method || '').includes('视频')) score += 8; if (Number(row.video_price || 0) <= (criteria.budgetRange?.max || Infinity)) score += 6; return score; } function buildCriteriaEmbeddingText(criteria: SearchCriteria): string { return [ ...(criteria.keywords || []), ...(criteria.contentTags || []), ...normalizeRegions(criteria.region), criteria.gender || '', `粉丝${criteria.fanRange?.min || 0}-${criteria.fanRange?.max || 0}`, `预算${criteria.budgetRange?.min || 0}-${criteria.budgetRange?.max || 0}`, ].filter(Boolean).join(' '); } function buildCreatorEmbeddingText(creator: JustOneCreator): string { return [ creator.nickname, creator.contentType, ...(creator.contentTags || []), ...(creator.personalTags || []), ...(creator.contentSamples || []).flatMap((sample) => [sample.title, sample.content]), creator.city, creator.geoLocation, creator.gender, creator.cooperationMethod, `粉丝${creator.fansCount}`, `图文报价${creator.imagePrice}`, `视频报价${creator.videoPrice}`, ].filter(Boolean).join(' '); } function isMeaningfulKeyword(keyword: string): boolean { const normalized = keyword.trim(); if (!normalized) return false; return !['生活方式', '垂直内容', '近期数据表现好', '真实自然', '非硬广', '图文', '视频'].includes(normalized); } function extractKeywordTerms(keyword: string): string[] { return keyword .replace(/类达人|达人|账号|粉丝号|垂类/g, '') .split(/[、,,\s/]+/) .map((item) => item.trim()) .filter((item) => item.length >= 2); } function getCreatorProfileUrl(creator: JustOneCreator): string { if (creator.profileUrl) return creator.profileUrl; if (creator.xiaohongshuUrl) return creator.xiaohongshuUrl; if (creator.platform === 'xiaohongshu') return `https://www.xiaohongshu.com/user/profile/${creator.userId}`; return ''; } function buildEmbedding(text: string): number[] { const vector = new Array(EMBEDDING_DIMENSION).fill(0); const normalized = text.toLowerCase().replace(/\s+/g, ' ').trim(); const grams = new Set(); for (const token of normalized.split(/[ ,,、/|]+/).filter(Boolean)) { grams.add(token); for (let size = 2; size <= 4; size++) { for (let i = 0; i <= token.length - size; i++) { grams.add(token.slice(i, i + size)); } } } for (const gram of grams) { const index = stableHash(gram) % EMBEDDING_DIMENSION; vector[index] += 1; } const norm = Math.sqrt(vector.reduce((sum, value) => sum + value * value, 0)); return norm > 0 ? vector.map((value) => Number((value / norm).toFixed(8))) : vector; } function normalizeCriteria(criteria: SearchCriteria): SearchCriteria { const maxFans = Number(criteria.fanRange?.max || 0); return { ...criteria, fanRange: maxFans > 0 && maxFans <= 1000 ? { min: Number(criteria.fanRange?.min || 0) * 10000, max: maxFans * 10000 } : criteria.fanRange, }; } function normalizeHeader(value: string): string { return String(value || '').replace(/\s+/g, '').replace(/ /g, '').trim(); } function normalizeRegions(region?: string | string[]): string[] { if (!region) return []; return (Array.isArray(region) ? region : String(region).split(/[、,,/]/)) .map((item) => item.trim()) .filter(Boolean); } function mapPlatform(platform: string): string { if (platform === '小红书') return 'xiaohongshu'; if (platform === '抖音') return 'douyin'; return platform; } function splitTags(value: string): string[] { return String(value || '').split(/[、,,/]/).map((item) => item.trim()).filter(Boolean); } function extractProfileId(url: string): string { return String(url || '').split('/').filter(Boolean).pop() || ''; } function toNumber(value: unknown): number { if (typeof value === 'number') return value; const text = String(value || '').replace(/[¥,,]/g, '').trim(); if (!text || text === '/') return 0; return Number(text) || 0; } function toCountFromWan(value: unknown): number { const num = toNumber(value); return num > 0 && num < 10000 ? Math.round(num * 10000) : Math.round(num); } function hashText(text: string): string { return stableHash(text).toString(36); } function stableHash(text: string): number { let hash = 2166136261; for (let i = 0; i < text.length; i++) { hash ^= text.charCodeAt(i); hash = Math.imul(hash, 16777619); } return Math.abs(hash >>> 0); }