deploy: auto commit server changes 2026-06-11 20:44:36

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 = [

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())

server/app/common/command/KbEnqueueRecent.php

@@ -0,0 +1,49 @@
+<?php
+
+namespace app\common\command;
+
+use app\common\service\ai\KbRedisQueueService;
+use think\console\Command;
+use think\console\Input;
+use think\console\Output;
+use think\console\input\Option;
+
+class KbEnqueueRecent extends Command
+{
+    protected function configure()
+    {
+        $this->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;
+    }
+}

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;
     }

server/app/common/service/ai/KbRedisQueueService.php

@@ -0,0 +1,235 @@
+<?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;
+    }
+}

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_created_at' => self::sourceTimestamp($data['create_time'] ?? 0),
-            'source_updated_at' => (int) ($data['update_time'] ?? $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_created_at' => self::sourceTimestamp($data['create_time'] ?? 0),
-            'source_updated_at' => (int) ($data['update_time'] ?? $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_created_at' => self::sourceTimestamp($data['create_time'] ?? 0),
-            'source_updated_at' => (int) ($data['update_time'] ?? $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_created_at' => self::sourceTimestamp($data['create_time'] ?? 0, $data['match_time'] ?? 0),
-            'source_updated_at' => (int) ($data['update_time'] ?? $data['match_time'] ?? $data['create_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_created_at' => self::sourceTimestamp($data['create_time'] ?? 0),
-            'source_updated_at' => (int) ($data['update_time'] ?? $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([

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('draw_result', 'lottery_draw_result', $limit, 80);
-        self::enqueueMissingLotteryRows('ai_history', 'lottery_ai_analysis', $limit);
+        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([

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',
     ],
 ];

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

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)<br\s*/?>|</p>|</div>|</li>", "\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())

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()