tihao-ai/server/services/local-creator-db.service.ts

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<number> {
  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<JustOneCreator[]> {
  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<string, unknown>;
  responseBody: unknown;
}): Promise<void> {
  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<string, unknown>;
  creators: JustOneCreator[];
}): Promise<number> {
  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 = 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,
          raw_creator = 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<void> {
  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,
    ],
  );
}

async function upsertCleanCreators(creators: ParsedCreator[]): Promise<void> {
  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<JustOneCreator[]> {
  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<string, unknown>): 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 || ''),
    xiaohongshuUrl: String(row.profile_url || ''),
    cooperationMethod: String(row.cooperation_method || ''),
  };
}

function rowToProviderRawCreator(row: Record<string, unknown>): 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.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 || ''),
    xiaohongshuUrl: String(row.profile_url || ''),
    cooperationMethod: String(row.cooperation_method || ''),
  };
}

function mergeCreators(creators: JustOneCreator[]): JustOneCreator[] {
  const map = new Map<string, JustOneCreator>();
  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<string, unknown>, 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);

  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.city,
    creator.geoLocation,
    creator.gender,
    creator.cooperationMethod,
    `粉丝${creator.fansCount}`,
    `图文报价${creator.imagePrice}`,
    `视频报价${creator.videoPrice}`,
  ].filter(Boolean).join(' ');
}

function getCreatorProfileUrl(creator: JustOneCreator): string {
  if (creator.xiaohongshuUrl) return creator.xiaohongshuUrl;
  if (creator.platform === 'douyin') return `https://www.douyin.com/user/${creator.userId}`;
  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<string>();

  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(/&#10;/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);
}