From e17a9832afcaa74086fe84ea02e73b0433caef30 Mon Sep 17 00:00:00 2001 From: hajimi Date: Thu, 11 Jun 2026 20:44:36 +0800 Subject: [PATCH] deploy: auto commit server changes 2026-06-11 20:44:36 --- .../validate/ai/KbSyncJobValidate.php | 2 +- server/app/common/command/KbConsume.php | 3 +- server/app/common/command/KbEnqueueRecent.php | 49 ++ server/app/common/command/KbRebuild.php | 10 +- .../common/service/ai/KbRedisQueueService.php | 235 ++++++ server/app/common/service/ai/KbService.php | 41 +- .../app/common/service/ai/KbSyncService.php | 68 +- server/config/console.php | 1 + .../dongqiudi-crawler/docker/entrypoint.sh | 4 + .../dongqiudi-crawler/scripts/kb_worker.py | 694 ++++++++++++++++++ .../dongqiudi-crawler/tests/test_kb_worker.py | 111 +++ 11 files changed, 1195 insertions(+), 23 deletions(-) create mode 100644 server/app/common/command/KbEnqueueRecent.php create mode 100644 server/app/common/service/ai/KbRedisQueueService.php create mode 100644 server/public/dongqiudi-crawler/scripts/kb_worker.py create mode 100644 server/public/dongqiudi-crawler/tests/test_kb_worker.py diff --git a/server/app/adminapi/validate/ai/KbSyncJobValidate.php b/server/app/adminapi/validate/ai/KbSyncJobValidate.php index e0b50ae..3a09eb1 100644 --- a/server/app/adminapi/validate/ai/KbSyncJobValidate.php +++ b/server/app/adminapi/validate/ai/KbSyncJobValidate.php @@ -7,7 +7,7 @@ use app\common\validate\BaseValidate; class KbSyncJobValidate extends BaseValidate { protected $rule = [ - 'domain' => 'require|in:article,post,lottery,all', + 'domain' => 'require|in:article,post,lottery,match,all', ]; protected $message = [ diff --git a/server/app/common/command/KbConsume.php b/server/app/common/command/KbConsume.php index 2614ff7..a2f8a47 100644 --- a/server/app/common/command/KbConsume.php +++ b/server/app/common/command/KbConsume.php @@ -22,8 +22,9 @@ class KbConsume extends Command protected function execute(Input $input, Output $output) { $batch = max(1, (int) $input->getOption('batch')); - KbSyncService::markLotteryBackfillJobs($batch); KbSyncService::markMatchBackfillJobs($batch); + KbSyncService::markArticleBackfillJobs($batch); + KbSyncService::markLotteryBackfillJobs($batch); $jobs = AiKbSyncJob::where('status', AiKbSyncJob::STATUS_PENDING) ->where('scheduled_at', '<=', time()) diff --git a/server/app/common/command/KbEnqueueRecent.php b/server/app/common/command/KbEnqueueRecent.php new file mode 100644 index 0000000..65d8030 --- /dev/null +++ b/server/app/common/command/KbEnqueueRecent.php @@ -0,0 +1,49 @@ +setName('kb:enqueue-recent') + ->setDescription('扫描最近更新内容并写入AI知识库Redis队列') + ->addOption('limit', null, Option::VALUE_OPTIONAL, '每个来源最多扫描数量', 1000) + ->addOption('minutes', null, Option::VALUE_OPTIONAL, '扫描最近多少分钟更新的数据', 30); + } + + protected function execute(Input $input, Output $output) + { + $limit = max(1, (int) $input->getOption('limit')); + $minutes = max(1, (int) $input->getOption('minutes')); + $summary = KbRedisQueueService::enqueueRecent($limit, $minutes); + + foreach ($summary['targets'] as $label => $target) { + $output->writeln(sprintf( + '[kb:enqueue-recent] %s scanned=%d queued=%d deduped=%d', + $label, + (int) $target['scanned'], + (int) $target['queued'], + (int) $target['deduped'] + )); + } + + $output->writeln(sprintf( + '[kb:enqueue-recent] stream=%s scanned=%d queued=%d deduped=%d limit=%d minutes=%d', + $summary['stream'], + (int) $summary['scanned'], + (int) $summary['queued'], + (int) $summary['deduped'], + (int) $summary['limit'], + (int) $summary['minutes'] + )); + + return 0; + } +} diff --git a/server/app/common/command/KbRebuild.php b/server/app/common/command/KbRebuild.php index 94f9991..ac72a36 100644 --- a/server/app/common/command/KbRebuild.php +++ b/server/app/common/command/KbRebuild.php @@ -43,7 +43,15 @@ class KbRebuild extends Command } } - KbSyncService::markLotteryBackfillJobs(); + if (in_array('article', $targets, true)) { + KbSyncService::markArticleBackfillJobs(); + } + if (in_array('match', $targets, true)) { + KbSyncService::markMatchBackfillJobs(); + } + if (in_array('lottery', $targets, true)) { + KbSyncService::markLotteryBackfillJobs(); + } $output->writeln(sprintf('[kb:rebuild] success=%d failed=%d', $summary['success'], $summary['failed'])); return 0; } diff --git a/server/app/common/service/ai/KbRedisQueueService.php b/server/app/common/service/ai/KbRedisQueueService.php new file mode 100644 index 0000000..e4a8bad --- /dev/null +++ b/server/app/common/service/ai/KbRedisQueueService.php @@ -0,0 +1,235 @@ + 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; + } +} diff --git a/server/app/common/service/ai/KbService.php b/server/app/common/service/ai/KbService.php index e44583f..f78bd89 100644 --- a/server/app/common/service/ai/KbService.php +++ b/server/app/common/service/ai/KbService.php @@ -395,8 +395,8 @@ class KbService 'published_at' => (string) ($data['published_at'] ?? ''), 'is_video' => (int) ($data['is_video'] ?? 0), ], - 'source_created_at' => (int) ($data['create_time'] ?? 0), - 'source_updated_at' => (int) ($data['update_time'] ?? $data['create_time'] ?? 0), + 'source_created_at' => self::sourceTimestamp($data['create_time'] ?? 0), + 'source_updated_at' => self::sourceTimestamp($data['update_time'] ?? 0, $data['create_time'] ?? 0), ]; } @@ -435,8 +435,8 @@ class KbService 'user_id' => (int) ($data['user_id'] ?? 0), 'tags' => array_values($tags), ], - 'source_created_at' => (int) ($data['create_time'] ?? 0), - 'source_updated_at' => (int) ($data['update_time'] ?? $data['create_time'] ?? 0), + 'source_created_at' => self::sourceTimestamp($data['create_time'] ?? 0), + 'source_updated_at' => self::sourceTimestamp($data['update_time'] ?? 0, $data['create_time'] ?? 0), ]; } @@ -482,8 +482,8 @@ class KbService 'game_name' => (string) ($game['name'] ?? ''), 'draw_time' => (string) ($data['draw_time'] ?? ''), ], - 'source_created_at' => (int) ($data['create_time'] ?? 0), - 'source_updated_at' => (int) ($data['update_time'] ?? $data['create_time'] ?? 0), + 'source_created_at' => self::sourceTimestamp($data['create_time'] ?? 0), + 'source_updated_at' => self::sourceTimestamp($data['update_time'] ?? 0, $data['create_time'] ?? 0), ]; } @@ -543,8 +543,8 @@ class KbService 'match_time' => $matchTime, 'status' => (int) ($data['status'] ?? 0), ], - 'source_created_at' => (int) ($data['create_time'] ?? $data['match_time'] ?? 0), - 'source_updated_at' => (int) ($data['update_time'] ?? $data['match_time'] ?? $data['create_time'] ?? 0), + 'source_created_at' => self::sourceTimestamp($data['create_time'] ?? 0, $data['match_time'] ?? 0), + 'source_updated_at' => self::sourceTimestamp($data['update_time'] ?? 0, $data['match_time'] ?? ($data['create_time'] ?? 0)), ]; } @@ -576,11 +576,32 @@ class KbService 'issue' => (string) ($data['issue'] ?? ''), 'module' => (string) ($data['module'] ?? ''), ], - 'source_created_at' => (int) ($data['create_time'] ?? 0), - 'source_updated_at' => (int) ($data['update_time'] ?? $data['create_time'] ?? 0), + 'source_created_at' => self::sourceTimestamp($data['create_time'] ?? 0), + 'source_updated_at' => self::sourceTimestamp($data['update_time'] ?? 0, $data['create_time'] ?? 0), ]; } + private static function sourceTimestamp($value, $fallback = 0): int + { + if (is_int($value)) { + return $value; + } + + if (is_numeric($value)) { + return (int) $value; + } + + $value = trim((string) $value); + if ($value !== '') { + $timestamp = strtotime($value); + if ($timestamp !== false) { + return $timestamp; + } + } + + return is_array($fallback) ? 0 : self::sourceTimestamp($fallback); + } + private static function buildKeywordText(array $payload): string { return implode(' ', array_filter([ diff --git a/server/app/common/service/ai/KbSyncService.php b/server/app/common/service/ai/KbSyncService.php index 2094fcc..87f411c 100644 --- a/server/app/common/service/ai/KbSyncService.php +++ b/server/app/common/service/ai/KbSyncService.php @@ -55,41 +55,89 @@ class KbSyncService ]); } + public static function markArticleBackfillJobs(int $limit = 200): void + { + self::enqueueUnsyncedRows( + 'article', + 'article', + 'article', + $limit, + 85, + 'COALESCE(s.update_time, s.create_time, 0)' + ); + } + public static function markLotteryBackfillJobs(int $limit = 200): void { - self::enqueueMissingLotteryRows('draw_result', 'lottery_draw_result', $limit); - self::enqueueMissingLotteryRows('ai_history', 'lottery_ai_analysis', $limit); + self::enqueueMissingLotteryRows('draw_result', 'lottery_draw_result', $limit, 80); + self::enqueueMissingLotteryRows('ai_history', 'lottery_ai_analysis', $limit, 90); } public static function markMatchBackfillJobs(int $limit = 200): void { - self::enqueueMissingRows('match', 'event', 'match', $limit); + self::enqueueUnsyncedRows( + 'match', + 'event', + 'match', + $limit, + 60, + 'COALESCE(s.update_time, s.create_time, s.match_time, 0)' + ); } - private static function enqueueMissingLotteryRows(string $subtype, string $table, int $limit): void + private static function enqueueMissingLotteryRows(string $subtype, string $table, int $limit, int $priority): void { - self::enqueueMissingRows('lottery', $subtype, $table, $limit); + self::enqueueUnsyncedRows( + 'lottery', + $subtype, + $table, + $limit, + $priority, + 'COALESCE(s.update_time, s.create_time, 0)' + ); } - private static function enqueueMissingRows(string $domain, string $subtype, string $table, int $limit): void - { + private static function enqueueUnsyncedRows( + string $domain, + string $subtype, + string $table, + int $limit, + int $priority, + string $sourceUpdatedExpr + ): void { try { - $rows = Db::name($table)->alias('s') + $query = Db::name($table)->alias('s') ->leftJoin('ai_kb_document d', "d.domain = '{$domain}' AND d.subtype = '{$subtype}' AND d.source_id = s.id") - ->whereNull('d.id') ->where('s.id', '>', 0) + ->whereRaw("(d.id IS NULL OR d.is_active <> 1 OR d.source_updated_at < {$sourceUpdatedExpr})"); + + self::applySourceFilters($query, $domain); + + $rows = $query ->order('s.id', 'desc') ->limit($limit) ->column('s.id'); foreach ($rows as $sourceId) { - self::enqueue($domain, $subtype, (int) $sourceId, AiKbSyncJob::ACTION_UPSERT, 80); + self::enqueue($domain, $subtype, (int) $sourceId, AiKbSyncJob::ACTION_UPSERT, $priority); } } catch (\Throwable $e) { // Ignore in environments where KB tables are not ready yet. } } + private static function applySourceFilters($query, string $domain): void + { + if ($domain === 'article') { + $query->where('s.is_show', 1)->whereNull('s.delete_time'); + return; + } + + if ($domain === 'match') { + $query->where('s.is_show', 1); + } + } + public static function markDocumentInactive(string $domain, string $subtype, int $sourceId): void { AiKbDocument::where([ diff --git a/server/config/console.php b/server/config/console.php index 8750899..e88951c 100644 --- a/server/config/console.php +++ b/server/config/console.php @@ -16,5 +16,6 @@ return [ 'sms_verify' => 'app\common\command\SmsVerifyCommand', 'kb:rebuild' => 'app\common\command\KbRebuild', 'kb:consume' => 'app\common\command\KbConsume', + 'kb:enqueue-recent' => 'app\common\command\KbEnqueueRecent', ], ]; diff --git a/server/public/dongqiudi-crawler/docker/entrypoint.sh b/server/public/dongqiudi-crawler/docker/entrypoint.sh index a8e755a..3eefcc2 100644 --- a/server/public/dongqiudi-crawler/docker/entrypoint.sh +++ b/server/public/dongqiudi-crawler/docker/entrypoint.sh @@ -7,6 +7,10 @@ PYTHON_BIN="${PYTHON_BIN:-/usr/local/bin/python}" mkdir -p /app/logs /app/data/locks /etc/sport-era-crawler +if [ "$#" -gt 0 ]; then + exec "$@" +fi + python - <<'PY' import os import shlex diff --git a/server/public/dongqiudi-crawler/scripts/kb_worker.py b/server/public/dongqiudi-crawler/scripts/kb_worker.py new file mode 100644 index 0000000..1945801 --- /dev/null +++ b/server/public/dongqiudi-crawler/scripts/kb_worker.py @@ -0,0 +1,694 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import html +import json +import math +import os +import re +import socket +import sys +import time +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional + +import aiohttp +import aiomysql +import redis.asyncio as aioredis +from loguru import logger + + +APP_DIR = Path(__file__).resolve().parents[1] +if str(APP_DIR) not in sys.path: + sys.path.insert(0, str(APP_DIR)) + +from src.core.config import get_config, load_config # noqa: E402 + + +STREAM_KEY = "ai:kb:sync:stream" +DEAD_STREAM_KEY = "ai:kb:sync:dead" +GROUP_NAME = "ai_kb_workers" +COLLECTION_NAME = "global" +DEFAULT_CHUNK_SIZE = 600 +DEFAULT_CHUNK_OVERLAP = 80 +DEFAULT_EMBEDDING_BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode" +DEFAULT_EMBEDDING_MODEL = "text-embedding-v4" +DEFAULT_EMBEDDING_DIM = 1024 + + +def now_ts() -> int: + return int(time.time()) + + +def normalize_plain_text(text: Any) -> str: + value = html.unescape(str(text or "")) + value = re.sub(r"\s+", " ", value.strip(), flags=re.UNICODE) + return value.strip() + + +def normalize_html_text(text: Any) -> str: + value = str(text or "") + value = re.sub(r"(?i)|

||", "\n", value) + value = re.sub(r"<[^>]+>", "", value) + return normalize_plain_text(value) + + +def truncate_text(text: str, length: int) -> str: + text = text.strip() + if len(text) <= length: + return text + return text[:length] + "..." + + +def split_text(text: str, chunk_size: int = DEFAULT_CHUNK_SIZE, overlap: int = DEFAULT_CHUNK_OVERLAP) -> List[str]: + text = text.strip() + if not text: + return [] + if len(text) <= chunk_size: + return [text] + chunks: List[str] = [] + start = 0 + step = max(1, chunk_size - overlap) + while start < len(text): + chunks.append(text[start : start + chunk_size].strip()) + if start + chunk_size >= len(text): + break + start += step + return [chunk for chunk in chunks if chunk] + + +def vector_norm(vector: Iterable[float]) -> float: + total = 0.0 + for value in vector: + total += float(value) * float(value) + return math.sqrt(total) if total > 0 else 0.0 + + +def json_loads_maybe(value: Any, default: Any) -> Any: + if value is None: + return default + if isinstance(value, (dict, list)): + return value + try: + return json.loads(str(value)) + except Exception: + return default + + +def json_text(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), default=str) + + +class KbDatabase: + def __init__(self): + cfg = get_config().database + self.cfg = cfg + self.prefix = cfg.prefix + self.pool: Optional[aiomysql.Pool] = None + + async def connect(self) -> None: + if self.pool: + return + self.pool = await aiomysql.create_pool( + host=self.cfg.host, + port=self.cfg.port, + user=self.cfg.username, + password=self.cfg.password, + db=self.cfg.database, + charset=self.cfg.charset, + autocommit=False, + minsize=1, + maxsize=max(2, min(10, self.cfg.pool_size)), + ) + logger.info("KB worker MySQL pool ready: {}:{}/{}", self.cfg.host, self.cfg.port, self.cfg.database) + + async def close(self) -> None: + if self.pool: + self.pool.close() + await self.pool.wait_closed() + self.pool = None + + def table(self, name: str) -> str: + return f"`{self.prefix}{name}`" + + async def fetchone(self, sql: str, args: tuple = ()) -> Optional[Dict[str, Any]]: + await self.connect() + assert self.pool is not None + async with self.pool.acquire() as conn: + async with conn.cursor(aiomysql.DictCursor) as cur: + await cur.execute(sql, args) + return await cur.fetchone() + + async def fetchall(self, sql: str, args: tuple = ()) -> List[Dict[str, Any]]: + await self.connect() + assert self.pool is not None + async with self.pool.acquire() as conn: + async with conn.cursor(aiomysql.DictCursor) as cur: + await cur.execute(sql, args) + return list(await cur.fetchall()) + + async def build_payload(self, domain: str, subtype: str, source_id: int) -> Optional[Dict[str, Any]]: + key = f"{domain}:{subtype}" + if key == "article:article": + return await self.build_article_payload(source_id) + if key == "post:post": + return await self.build_post_payload(source_id) + if key == "lottery:draw_result": + return await self.build_lottery_draw_result_payload(source_id) + if key == "lottery:ai_history": + return await self.build_lottery_ai_history_payload(source_id) + if key == "match:event": + return await self.build_match_event_payload(source_id) + return None + + async def build_article_payload(self, source_id: int) -> Optional[Dict[str, Any]]: + row = await self.fetchone( + f"SELECT * FROM {self.table('article')} WHERE id=%s AND is_show=1 AND delete_time IS NULL LIMIT 1", + (source_id,), + ) + if not row: + return None + cate = await self.fetchone(f"SELECT name FROM {self.table('article_cate')} WHERE id=%s LIMIT 1", (row.get("cid") or 0,)) + cate_name = str((cate or {}).get("name") or "") + ext = json_loads_maybe(row.get("ext"), {}) + translated = str(ext.get("translated_content") or "").strip() if isinstance(ext, dict) else "" + content = normalize_html_text(translated or row.get("content") or "") + summary = normalize_plain_text(row.get("abstract") or row.get("desc") or "") + keywords = list(dict.fromkeys(filter(None, [ + str(row.get("title") or ""), + cate_name, + str(row.get("author") or ""), + str(row.get("category") or ""), + ]))) + return { + "title": str(row.get("title") or ""), + "summary": summary, + "canonical_text": "\n\n".join(filter(None, [str(row.get("title") or ""), summary, content])).strip(), + "keywords": ",".join(keywords), + "metadata_json": { + "cid": int(row.get("cid") or 0), + "cate_name": cate_name, + "author": str(row.get("author") or ""), + "category": str(row.get("category") or ""), + "article_id": int(row.get("article_id") or 0), + "published_at": str(row.get("published_at") or ""), + "is_video": int(row.get("is_video") or 0), + }, + "source_created_at": int(row.get("create_time") or 0), + "source_updated_at": int(row.get("update_time") or row.get("create_time") or 0), + } + + async def build_post_payload(self, source_id: int) -> Optional[Dict[str, Any]]: + row = await self.fetchone( + f"SELECT * FROM {self.table('community_post')} WHERE id=%s AND status=1 AND delete_time IS NULL LIMIT 1", + (source_id,), + ) + if not row: + return None + tags = await self.fetchall( + f""" + SELECT t.name + FROM {self.table('community_post_tag')} pt + JOIN {self.table('community_tag')} t ON t.id = pt.tag_id + WHERE pt.post_id=%s + """, + (source_id,), + ) + tag_names = [str(item.get("name") or "") for item in tags if item.get("name")] + ext = json_loads_maybe(row.get("ext"), {}) + segments = [ + str(row.get("content") or ""), + str(ext.get("translated_content") or "") if isinstance(ext, dict) else "", + str(ext.get("lottery_analysis_content") or "") if isinstance(ext, dict) else "", + " ".join(tag_names), + ] + canonical = normalize_plain_text("\n\n".join(filter(None, segments))) + return { + "title": truncate_text(canonical, 40) or f"社区帖子#{source_id}", + "summary": truncate_text(canonical, 200), + "canonical_text": canonical, + "keywords": ",".join(dict.fromkeys(tag_names)), + "metadata_json": { + "post_type": int(row.get("post_type") or 0), + "match_id": int(row.get("match_id") or 0), + "is_paid": int(row.get("is_paid") or 0), + "user_id": int(row.get("user_id") or 0), + "tags": tag_names, + }, + "source_created_at": int(row.get("create_time") or 0), + "source_updated_at": int(row.get("update_time") or row.get("create_time") or 0), + } + + async def build_lottery_draw_result_payload(self, source_id: int) -> Optional[Dict[str, Any]]: + row = await self.fetchone(f"SELECT * FROM {self.table('lottery_draw_result')} WHERE id=%s LIMIT 1", (source_id,)) + if not row: + return None + game = await self.fetchone( + f"SELECT name, template FROM {self.table('lottery_game')} WHERE code=%s LIMIT 1", + (row.get("code") or "",), + ) + game = game or {} + numbers = [item for item in str(row.get("draw_code") or "").split(",") if item] + title = f"{game.get('name') or row.get('code')} 第{row.get('issue')}期开奖" + summary = f"{title} 于 {row.get('draw_time') or ''} 开奖，开奖号码为 {'、'.join(numbers)}。" + trend = json_loads_maybe(row.get("trend"), {}) + canonical = "\n".join(filter(None, [ + summary, + f"下一期期号：{row.get('next_issue')}" if row.get("next_issue") else "", + f"下期开奖时间：{row.get('next_time')}" if row.get("next_time") else "", + "走势数据：" + normalize_plain_text(json_text(trend)) if trend not in ({}, None) else "", + ])).strip() + return { + "title": title, + "summary": summary, + "canonical_text": canonical, + "keywords": ",".join(filter(None, [ + str(row.get("code") or ""), + str(game.get("name") or ""), + str(game.get("template") or ""), + str(row.get("issue") or ""), + ])), + "metadata_json": { + "code": str(row.get("code") or ""), + "issue": str(row.get("issue") or ""), + "template": str(game.get("template") or ""), + "game_name": str(game.get("name") or ""), + "draw_time": str(row.get("draw_time") or ""), + }, + "source_created_at": int(row.get("create_time") or 0), + "source_updated_at": int(row.get("update_time") or row.get("create_time") or 0), + } + + async def build_lottery_ai_history_payload(self, source_id: int) -> Optional[Dict[str, Any]]: + row = await self.fetchone(f"SELECT * FROM {self.table('lottery_ai_analysis')} WHERE id=%s LIMIT 1", (source_id,)) + if not row: + return None + result = row.get("result") if row.get("result") is not None else "" + result_text = result if isinstance(result, str) else json_text(result) + canonical = normalize_plain_text(result_text) + title = f"{row.get('code') or '彩票'} 第{row.get('issue') or '-'}期 {row.get('module') or 'module'} 历史AI分析" + return { + "title": title, + "summary": truncate_text(canonical, 220), + "canonical_text": canonical, + "keywords": ",".join(filter(None, [str(row.get("code") or ""), str(row.get("issue") or ""), str(row.get("module") or "")])), + "metadata_json": { + "code": str(row.get("code") or ""), + "issue": str(row.get("issue") or ""), + "module": str(row.get("module") or ""), + }, + "source_created_at": int(row.get("create_time") or 0), + "source_updated_at": int(row.get("update_time") or row.get("create_time") or 0), + } + + async def build_match_event_payload(self, source_id: int) -> Optional[Dict[str, Any]]: + row = await self.fetchone(f"SELECT * FROM {self.table('match')} WHERE id=%s AND is_show=1 LIMIT 1", (source_id,)) + if not row: + return None + home = str(row.get("home_team") or "") + away = str(row.get("away_team") or "") + league = str(row.get("league_name") or "") + match_time_value = int(row.get("match_time") or 0) + match_time_text = time.strftime("%Y-%m-%d %H:%M", time.localtime(match_time_value)) if match_time_value > 0 else "" + history = await self.fetchall( + f""" + SELECT * FROM {self.table('match_history')} + WHERE (home_team=%s AND away_team=%s) OR (home_team=%s AND away_team=%s) + ORDER BY match_time DESC LIMIT 5 + """, + (home, away, away, home), + ) + home_recent = await self.fetchall( + f"SELECT * FROM {self.table('match_recent')} WHERE team_name=%s ORDER BY match_time DESC LIMIT 5", + (home,), + ) + away_recent = await self.fetchall( + f"SELECT * FROM {self.table('match_recent')} WHERE team_name=%s ORDER BY match_time DESC LIMIT 5", + (away,), + ) + title = f"{home} vs {away}".strip() + summary = ( + f"{league} {title}，{home} 对阵 {away}，比赛时间 {match_time_text}，" + f"当前比分 {row.get('home_score') or 0}-{row.get('away_score') or 0}。" + ) + canonical = "\n".join([ + summary, + f"赔率：主胜{row.get('home_odds') or '-'}，平局{row.get('draw_odds') or '-'}，客胜{row.get('away_odds') or '-'}。", + "历史交锋：" + normalize_plain_text(json_text(history)), + f"{home}近期战绩：" + normalize_plain_text(json_text(home_recent)), + f"{away}近期战绩：" + normalize_plain_text(json_text(away_recent)), + ]) + return { + "title": title or f"赛事#{source_id}", + "summary": normalize_plain_text(summary), + "canonical_text": canonical.strip(), + "keywords": ",".join(filter(None, [league, home, away])), + "metadata_json": { + "league": league, + "teams": [team for team in [home, away] if team], + "match_time": match_time_text, + "status": int(row.get("status") or 0), + }, + "source_created_at": int(row.get("create_time") or row.get("match_time") or 0), + "source_updated_at": int(row.get("update_time") or row.get("match_time") or row.get("create_time") or 0), + } + + async def upsert_document(self, domain: str, subtype: str, source_id: int, embedder: "EmbeddingClient") -> Dict[str, Any]: + payload = await self.build_payload(domain, subtype, source_id) + if not payload: + await self.mark_document_inactive(domain, subtype, source_id) + return {"success": False, "error": "源内容不存在或不可用"} + + content_hash = hashlib.md5(json_text([ + payload["title"], + payload["summary"], + payload["canonical_text"], + payload["keywords"], + payload["metadata_json"], + ]).encode("utf-8")).hexdigest() + + existing = await self.fetchone( + f""" + SELECT * FROM {self.table('ai_kb_document')} + WHERE domain=%s AND subtype=%s AND source_id=%s LIMIT 1 + """, + (domain, subtype, source_id), + ) + if existing and str(existing.get("content_hash") or "") == content_hash and int(existing.get("is_active") or 0) == 1: + await self.update_document_metadata(int(existing["id"]), payload) + chunk_count = await self.fetchone( + f"SELECT COUNT(*) AS total FROM {self.table('ai_kb_chunk')} WHERE document_id=%s", + (int(existing["id"]),), + ) + return {"success": True, "document_id": int(existing["id"]), "chunk_count": int((chunk_count or {}).get("total") or 0), "skipped": True} + + chunks = split_text(str(payload["canonical_text"] or "")) + if not chunks: + chunks = [payload["summary"] or payload["title"] or "empty"] + embedding_result = await embedder.embed(chunks) + embeddings = embedding_result.get("embeddings") or [] + if not embedding_result.get("success") or not embeddings: + return {"success": False, "error": embedding_result.get("error") or "embedding失败"} + + await self.connect() + assert self.pool is not None + async with self.pool.acquire() as conn: + async with conn.cursor(aiomysql.DictCursor) as cur: + await conn.begin() + try: + metadata_json = json_text(payload["metadata_json"]) + if existing: + document_id = int(existing["id"]) + await cur.execute( + f""" + UPDATE {self.table('ai_kb_document')} + SET title=%s, summary=%s, canonical_text=%s, keywords=%s, metadata_json=%s, + source_created_at=%s, source_updated_at=%s, content_hash=%s, is_active=1, update_time=%s + WHERE id=%s + """, + ( + payload["title"], + payload["summary"], + payload["canonical_text"], + payload["keywords"], + metadata_json, + int(payload["source_created_at"]), + int(payload["source_updated_at"]), + content_hash, + now_ts(), + document_id, + ), + ) + else: + await cur.execute( + f""" + INSERT INTO {self.table('ai_kb_document')} + (domain, subtype, source_id, title, summary, canonical_text, keywords, metadata_json, + source_created_at, source_updated_at, content_hash, is_active, create_time, update_time) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,1,%s,%s) + """, + ( + domain, + subtype, + source_id, + payload["title"], + payload["summary"], + payload["canonical_text"], + payload["keywords"], + metadata_json, + int(payload["source_created_at"]), + int(payload["source_updated_at"]), + content_hash, + now_ts(), + now_ts(), + ), + ) + document_id = int(cur.lastrowid) + + await cur.execute(f"DELETE FROM {self.table('ai_kb_chunk')} WHERE document_id=%s", (document_id,)) + dimension = int(embedding_result.get("dimension") or len(embeddings[0] or [])) + keyword_text = " ".join(filter(None, [ + str(payload.get("title") or ""), + str(payload.get("summary") or ""), + str(payload.get("keywords") or ""), + json_text(payload.get("metadata_json") or {}), + ])) + for index, chunk_text in enumerate(chunks): + embedding = [float(value) for value in embeddings[index]] + await cur.execute( + f""" + INSERT INTO {self.table('ai_kb_chunk')} + (document_id, collection_name, chunk_index, chunk_text, embedding_json, + embedding_model, embedding_dim, vector_norm, keyword_text, create_time, update_time) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) + """, + ( + document_id, + COLLECTION_NAME, + index, + chunk_text, + json_text(embedding), + embedding_result.get("model") or embedder.model, + dimension, + vector_norm(embedding), + keyword_text, + now_ts(), + now_ts(), + ), + ) + await conn.commit() + except Exception: + await conn.rollback() + raise + return {"success": True, "document_id": document_id, "chunk_count": len(chunks)} + + async def update_document_metadata(self, document_id: int, payload: Dict[str, Any]) -> None: + await self.connect() + assert self.pool is not None + async with self.pool.acquire() as conn: + async with conn.cursor() as cur: + await cur.execute( + f""" + UPDATE {self.table('ai_kb_document')} + SET title=%s, summary=%s, keywords=%s, metadata_json=%s, + source_created_at=%s, source_updated_at=%s, update_time=%s + WHERE id=%s + """, + ( + payload["title"], + payload["summary"], + payload["keywords"], + json_text(payload["metadata_json"]), + int(payload["source_created_at"]), + int(payload["source_updated_at"]), + now_ts(), + document_id, + ), + ) + await conn.commit() + + async def mark_document_inactive(self, domain: str, subtype: str, source_id: int) -> None: + await self.connect() + assert self.pool is not None + async with self.pool.acquire() as conn: + async with conn.cursor() as cur: + await cur.execute( + f""" + UPDATE {self.table('ai_kb_document')} + SET is_active=0, update_time=%s + WHERE domain=%s AND subtype=%s AND source_id=%s + """, + (now_ts(), domain, subtype, source_id), + ) + await conn.commit() + + +class EmbeddingClient: + def __init__(self): + self.api_key = os.environ.get("EMBEDDING_API_KEY") or os.environ.get("DASHSCOPE_API_KEY") or "" + self.base_url = (os.environ.get("EMBEDDING_BASE_URL") or DEFAULT_EMBEDDING_BASE_URL).rstrip("/") + self.model = os.environ.get("EMBEDDING_MODEL") or DEFAULT_EMBEDDING_MODEL + self.dimension = int(os.environ.get("EMBEDDING_DIM") or DEFAULT_EMBEDDING_DIM) + + async def embed(self, chunks: List[str]) -> Dict[str, Any]: + if not self.api_key or not self.base_url or not self.model: + return {"success": False, "error": "Embedding API配置未完成", "embeddings": []} + payload: Dict[str, Any] = {"model": self.model, "input": chunks} + if self.dimension > 0: + payload["dimensions"] = self.dimension + start = time.monotonic() + try: + async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60)) as session: + async with session.post( + self.base_url + "/v1/embeddings", + json=payload, + headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}, + ssl=False, + ) as resp: + body = await resp.text() + try: + data = json.loads(body) + except Exception: + data = {} + if resp.status != 200 or not isinstance(data.get("data"), list): + message = ((data.get("error") or {}).get("message") if isinstance(data, dict) else "") or f"HTTP {resp.status}" + return {"success": False, "error": message, "embeddings": [], "cost_ms": int((time.monotonic() - start) * 1000)} + embeddings = [] + for row in data.get("data") or []: + embedding = row.get("embedding") if isinstance(row, dict) else None + if isinstance(embedding, list) and embedding: + embeddings.append([float(value) for value in embedding]) + return { + "success": bool(embeddings), + "error": "" if embeddings else "Embedding结果为空", + "embeddings": embeddings, + "model": self.model, + "dimension": len(embeddings[0]) if embeddings else 0, + "tokens": int(((data.get("usage") or {}).get("total_tokens") or 0) if isinstance(data, dict) else 0), + "cost_ms": int((time.monotonic() - start) * 1000), + } + except Exception as exc: + return {"success": False, "error": f"Embedding请求失败: {exc}", "embeddings": []} + + +class KbWorker: + def __init__(self, batch: int, block_ms: int, max_retries: int, worker_id: str): + self.batch = max(1, batch) + self.block_ms = max(100, block_ms) + self.max_retries = max(0, max_retries) + self.worker_id = worker_id + redis_cfg = get_config().redis + self.redis = aioredis.Redis( + host=redis_cfg.host, + port=redis_cfg.port, + password=redis_cfg.password or None, + db=redis_cfg.db, + decode_responses=True, + ) + self.db = KbDatabase() + self.embedder = EmbeddingClient() + + async def close(self) -> None: + await self.redis.close() + await self.db.close() + + async def ensure_group(self) -> None: + try: + await self.redis.xgroup_create(STREAM_KEY, GROUP_NAME, id="0", mkstream=True) + logger.info("created redis stream group {} on {}", GROUP_NAME, STREAM_KEY) + except Exception as exc: + if "BUSYGROUP" not in str(exc): + raise + + async def run_forever(self) -> None: + await self.ensure_group() + logger.info("KB worker started id={} batch={} block_ms={}", self.worker_id, self.batch, self.block_ms) + while True: + await self.consume_once() + + async def consume_once(self) -> int: + rows = await self.redis.xreadgroup( + GROUP_NAME, + self.worker_id, + streams={STREAM_KEY: ">"}, + count=self.batch, + block=self.block_ms, + ) + processed = 0 + for _, messages in rows or []: + for message_id, fields in messages: + await self.handle_message(message_id, fields) + processed += 1 + return processed + + async def handle_message(self, message_id: str, fields: Dict[str, Any]) -> None: + domain = str(fields.get("domain") or "") + subtype = str(fields.get("subtype") or "") + source_id = int(fields.get("source_id") or 0) + attempts = int(fields.get("attempts") or 0) + started = time.monotonic() + try: + if not domain or not subtype or source_id <= 0: + raise ValueError("消息缺少 domain/subtype/source_id") + result = await self.db.upsert_document(domain, subtype, source_id, self.embedder) + if not result.get("success"): + raise RuntimeError(str(result.get("error") or "upsert failed")) + await self.redis.xack(STREAM_KEY, GROUP_NAME, message_id) + logger.info( + "kb upsert ok message={} source={}/{}/{} result={} elapsed={:.2f}s", + message_id, + domain, + subtype, + source_id, + result, + time.monotonic() - started, + ) + except Exception as exc: + await self.handle_failure(message_id, fields, attempts, str(exc)) + + async def handle_failure(self, message_id: str, fields: Dict[str, Any], attempts: int, error: str) -> None: + next_attempt = attempts + 1 + payload = dict(fields) + payload["attempts"] = str(next_attempt) + payload["last_error"] = error[:500] + payload["failed_at"] = str(now_ts()) + if next_attempt > self.max_retries: + await self.redis.xadd(DEAD_STREAM_KEY, payload) + await self.redis.xack(STREAM_KEY, GROUP_NAME, message_id) + logger.error("kb message={} moved to dead stream after {} attempts: {}", message_id, next_attempt, error) + return + await self.redis.xadd(STREAM_KEY, payload) + await self.redis.xack(STREAM_KEY, GROUP_NAME, message_id) + logger.warning("kb message={} retry {}/{}: {}", message_id, next_attempt, self.max_retries, error) + + +def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Sport Era AI KB Redis Stream worker") + parser.add_argument("--batch", type=int, default=50) + parser.add_argument("--block-ms", type=int, default=5000) + parser.add_argument("--max-retries", type=int, default=3) + parser.add_argument("--worker-id", default="") + return parser.parse_args(argv) + + +async def async_main(argv: Optional[List[str]] = None) -> int: + args = parse_args(argv) + load_config() + worker_id = args.worker_id or f"{socket.gethostname()}-{os.getpid()}" + worker = KbWorker(args.batch, args.block_ms, args.max_retries, worker_id) + try: + await worker.run_forever() + finally: + await worker.close() + return 0 + + +def main(argv: Optional[List[str]] = None) -> int: + return asyncio.run(async_main(argv)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/server/public/dongqiudi-crawler/tests/test_kb_worker.py b/server/public/dongqiudi-crawler/tests/test_kb_worker.py new file mode 100644 index 0000000..08c453d --- /dev/null +++ b/server/public/dongqiudi-crawler/tests/test_kb_worker.py @@ -0,0 +1,111 @@ +import unittest +from pathlib import Path +import sys +import types + + +APP_DIR = Path(__file__).resolve().parents[1] +if str(APP_DIR) not in sys.path: + sys.path.insert(0, str(APP_DIR)) + +fake_aiomysql = types.ModuleType("aiomysql") +fake_aiomysql.DictCursor = object +fake_aiomysql.create_pool = None +sys.modules.setdefault("aiomysql", fake_aiomysql) + +fake_aiohttp = types.ModuleType("aiohttp") +fake_aiohttp.ClientSession = object +fake_aiohttp.ClientTimeout = object +sys.modules.setdefault("aiohttp", fake_aiohttp) + +fake_redis = types.ModuleType("redis") +fake_redis_asyncio = types.ModuleType("redis.asyncio") +fake_redis_asyncio.Redis = object +fake_redis.asyncio = fake_redis_asyncio +sys.modules.setdefault("redis", fake_redis) +sys.modules.setdefault("redis.asyncio", fake_redis_asyncio) + +fake_loguru = types.ModuleType("loguru") +fake_loguru.logger = types.SimpleNamespace(info=lambda *args, **kwargs: None, warning=lambda *args, **kwargs: None, error=lambda *args, **kwargs: None) +sys.modules.setdefault("loguru", fake_loguru) + +fake_src = types.ModuleType("src") +fake_core = types.ModuleType("src.core") +fake_config = types.ModuleType("src.core.config") +fake_config.get_config = lambda: None +fake_config.load_config = lambda: None +fake_core.config = fake_config +fake_src.core = fake_core +sys.modules.setdefault("src", fake_src) +sys.modules.setdefault("src.core", fake_core) +sys.modules.setdefault("src.core.config", fake_config) + +from scripts.kb_worker import DEAD_STREAM_KEY, GROUP_NAME, STREAM_KEY, KbWorker # noqa: E402 + + +class FakeRedis: + def __init__(self): + self.acked = [] + self.added = [] + + async def xack(self, stream, group, message_id): + self.acked.append((stream, group, message_id)) + + async def xadd(self, stream, payload): + self.added.append((stream, payload)) + return "2-0" + + +class FakeDb: + def __init__(self, result=None, exc=None): + self.result = result if result is not None else {"success": True, "document_id": 1, "chunk_count": 1} + self.exc = exc + self.calls = [] + + async def upsert_document(self, domain, subtype, source_id, embedder): + self.calls.append((domain, subtype, source_id)) + if self.exc: + raise self.exc + return self.result + + +class FakeEmbedder: + pass + + +class KbWorkerQueueTest(unittest.IsolatedAsyncioTestCase): + def make_worker(self, db): + worker = object.__new__(KbWorker) + worker.redis = FakeRedis() + worker.db = db + worker.embedder = FakeEmbedder() + worker.max_retries = 1 + return worker + + async def test_successful_message_is_acked(self): + worker = self.make_worker(FakeDb()) + + await worker.handle_message("1-0", {"domain": "article", "subtype": "article", "source_id": "123"}) + + self.assertEqual(worker.db.calls, [("article", "article", 123)]) + self.assertEqual(worker.redis.acked, [(STREAM_KEY, GROUP_NAME, "1-0")]) + self.assertEqual(worker.redis.added, []) + + async def test_message_goes_to_dead_stream_after_retry_limit(self): + worker = self.make_worker(FakeDb(exc=RuntimeError("boom"))) + + await worker.handle_message( + "1-0", + {"domain": "article", "subtype": "article", "source_id": "123", "attempts": "1"}, + ) + + self.assertEqual(worker.redis.acked, [(STREAM_KEY, GROUP_NAME, "1-0")]) + self.assertEqual(len(worker.redis.added), 1) + stream, payload = worker.redis.added[0] + self.assertEqual(stream, DEAD_STREAM_KEY) + self.assertEqual(payload["attempts"], "2") + self.assertIn("boom", payload["last_error"]) + + +if __name__ == "__main__": + unittest.main()