236 lines
8.0 KiB
PHP
236 lines
8.0 KiB
PHP
<?php
|
|
|
|
namespace app\common\service\ai;
|
|
|
|
use think\facade\Db;
|
|
|
|
class KbRedisQueueService
|
|
{
|
|
public const STREAM_KEY = 'ai:kb:sync:stream';
|
|
public const DEDUPE_PREFIX = 'ai:kb:sync:dedupe:';
|
|
|
|
private const DEFAULT_DEDUPE_TTL = 1800;
|
|
|
|
public static function enqueueRecent(int $limit = 1000, int $minutes = 30): array
|
|
{
|
|
$limit = max(1, $limit);
|
|
$minutes = max(1, $minutes);
|
|
$cutoff = time() - ($minutes * 60);
|
|
$redis = self::redis();
|
|
$summary = [
|
|
'stream' => self::STREAM_KEY,
|
|
'limit' => $limit,
|
|
'minutes' => $minutes,
|
|
'scanned' => 0,
|
|
'queued' => 0,
|
|
'deduped' => 0,
|
|
'targets' => [],
|
|
];
|
|
|
|
foreach (self::targets() as $target) {
|
|
$result = self::enqueueTarget($redis, $target, $limit, $cutoff, $minutes * 60);
|
|
$summary['targets'][$target['label']] = $result;
|
|
$summary['scanned'] += $result['scanned'];
|
|
$summary['queued'] += $result['queued'];
|
|
$summary['deduped'] += $result['deduped'];
|
|
}
|
|
|
|
return $summary;
|
|
}
|
|
|
|
private static function enqueueTarget(\Redis $redis, array $target, int $limit, int $cutoff, int $dedupeTtl): array
|
|
{
|
|
$result = [
|
|
'domain' => $target['domain'],
|
|
'subtype' => $target['subtype'],
|
|
'table' => $target['table'],
|
|
'scanned' => 0,
|
|
'queued' => 0,
|
|
'deduped' => 0,
|
|
];
|
|
|
|
$rows = self::recentUnsyncedRows($target, $limit, $cutoff);
|
|
$result['scanned'] = count($rows);
|
|
|
|
foreach ($rows as $row) {
|
|
$sourceId = (int) ($row['source_id'] ?? 0);
|
|
if ($sourceId <= 0) {
|
|
continue;
|
|
}
|
|
|
|
$dedupeKey = self::dedupeKey($target['domain'], $target['subtype'], $sourceId);
|
|
$setResult = $redis->rawCommand('SET', $dedupeKey, '1', 'EX', (string) max(60, $dedupeTtl), 'NX');
|
|
if ($setResult !== true && strtoupper((string) $setResult) !== 'OK') {
|
|
$result['deduped']++;
|
|
continue;
|
|
}
|
|
|
|
$messageId = self::xAdd($redis, [
|
|
'domain' => $target['domain'],
|
|
'subtype' => $target['subtype'],
|
|
'source_id' => (string) $sourceId,
|
|
'source_updated_at' => (string) (int) ($row['source_updated_at'] ?? 0),
|
|
'priority' => (string) (int) $target['priority'],
|
|
'trace_id' => self::traceId($target['label'], $sourceId),
|
|
]);
|
|
|
|
if ($messageId !== false && $messageId !== null && $messageId !== '') {
|
|
$result['queued']++;
|
|
}
|
|
}
|
|
|
|
return $result;
|
|
}
|
|
|
|
private static function recentUnsyncedRows(array $target, int $limit, int $cutoff): array
|
|
{
|
|
$sourceUpdatedExpr = $target['source_updated_expr'];
|
|
$query = Db::name($target['table'])->alias('s')
|
|
->leftJoin(
|
|
'ai_kb_document d',
|
|
"d.domain = '{$target['domain']}' AND d.subtype = '{$target['subtype']}' AND d.source_id = s.id"
|
|
)
|
|
->where('s.id', '>', 0)
|
|
->whereRaw("{$sourceUpdatedExpr} >= {$cutoff}")
|
|
->whereRaw("(d.id IS NULL OR d.is_active <> 1 OR d.source_updated_at < {$sourceUpdatedExpr})");
|
|
|
|
self::applySourceFilters($query, $target);
|
|
|
|
return $query
|
|
->fieldRaw("s.id AS source_id, {$sourceUpdatedExpr} AS source_updated_at")
|
|
->orderRaw("{$sourceUpdatedExpr} DESC, s.id DESC")
|
|
->limit($limit)
|
|
->select()
|
|
->toArray();
|
|
}
|
|
|
|
private static function applySourceFilters($query, array $target): void
|
|
{
|
|
if (($target['domain'] ?? '') === 'article') {
|
|
$query->where('s.is_show', 1)->whereNull('s.delete_time');
|
|
}
|
|
|
|
if (($target['domain'] ?? '') === 'match') {
|
|
$query->where('s.is_show', 1)->whereNull('s.delete_time');
|
|
}
|
|
|
|
if (($target['domain'] ?? '') === 'post') {
|
|
$query->where('s.status', 1)->whereNull('s.delete_time');
|
|
}
|
|
|
|
foreach (($target['where'] ?? []) as $field => $value) {
|
|
$query->where('s.' . $field, $value);
|
|
}
|
|
}
|
|
|
|
private static function targets(): array
|
|
{
|
|
return [
|
|
[
|
|
'label' => 'crypto_article',
|
|
'domain' => 'article',
|
|
'subtype' => 'article',
|
|
'table' => 'article',
|
|
'priority' => 82,
|
|
'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
|
|
'where' => ['cid' => 9],
|
|
],
|
|
[
|
|
'label' => 'article',
|
|
'domain' => 'article',
|
|
'subtype' => 'article',
|
|
'table' => 'article',
|
|
'priority' => 85,
|
|
'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
|
|
],
|
|
[
|
|
'label' => 'lottery_draw_result',
|
|
'domain' => 'lottery',
|
|
'subtype' => 'draw_result',
|
|
'table' => 'lottery_draw_result',
|
|
'priority' => 80,
|
|
'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
|
|
],
|
|
[
|
|
'label' => 'lottery_ai_history',
|
|
'domain' => 'lottery',
|
|
'subtype' => 'ai_history',
|
|
'table' => 'lottery_ai_analysis',
|
|
'priority' => 90,
|
|
'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
|
|
],
|
|
[
|
|
'label' => 'match',
|
|
'domain' => 'match',
|
|
'subtype' => 'event',
|
|
'table' => 'match',
|
|
'priority' => 60,
|
|
'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, s.match_time, 0)',
|
|
],
|
|
[
|
|
'label' => 'post',
|
|
'domain' => 'post',
|
|
'subtype' => 'post',
|
|
'table' => 'community_post',
|
|
'priority' => 75,
|
|
'source_updated_expr' => 'COALESCE(s.update_time, s.create_time, 0)',
|
|
],
|
|
];
|
|
}
|
|
|
|
private static function redis(): \Redis
|
|
{
|
|
if (!class_exists(\Redis::class)) {
|
|
throw new \RuntimeException('Redis 扩展不可用，无法写入 AI 知识库 Redis 队列');
|
|
}
|
|
|
|
$options = (array) config('cache.stores.redis');
|
|
$redis = new \Redis();
|
|
$host = (string) (env('AI_KB_REDIS_HOST', '') ?: ($options['host'] ?? '127.0.0.1'));
|
|
$port = (int) (env('AI_KB_REDIS_PORT', '') ?: ($options['port'] ?? 6379));
|
|
$timeout = (float) ($options['timeout'] ?? 0);
|
|
$redis->connect($host, $port, $timeout);
|
|
|
|
$password = (string) (env('AI_KB_REDIS_PASSWORD', '') ?: ($options['password'] ?? ''));
|
|
if ($password !== '') {
|
|
$redis->auth($password);
|
|
}
|
|
|
|
$select = (int) (env('AI_KB_REDIS_DB', '') ?: 4);
|
|
$redis->select($select);
|
|
|
|
return $redis;
|
|
}
|
|
|
|
private static function xAdd(\Redis $redis, array $fields)
|
|
{
|
|
if (method_exists($redis, 'xAdd')) {
|
|
return $redis->xAdd(self::STREAM_KEY, '*', $fields);
|
|
}
|
|
|
|
$args = ['XADD', self::STREAM_KEY, '*'];
|
|
foreach ($fields as $field => $value) {
|
|
$args[] = (string) $field;
|
|
$args[] = (string) $value;
|
|
}
|
|
|
|
return $redis->rawCommand(...$args);
|
|
}
|
|
|
|
private static function dedupeKey(string $domain, string $subtype, int $sourceId): string
|
|
{
|
|
return self::DEDUPE_PREFIX . $domain . ':' . $subtype . ':' . $sourceId;
|
|
}
|
|
|
|
private static function traceId(string $label, int $sourceId): string
|
|
{
|
|
try {
|
|
$suffix = bin2hex(random_bytes(4));
|
|
} catch (\Throwable $e) {
|
|
$suffix = (string) mt_rand(100000, 999999);
|
|
}
|
|
|
|
return $label . '-' . $sourceId . '-' . date('YmdHis') . '-' . $suffix;
|
|
}
|
|
}
|