plugin.py

"""群友语录 - 打通 Maibot 核心的独立语录表情包系统

将群聊投稿的图片语录接入 Maibot Planner：
- /投稿          回复含图片的消息投稿，VLM 自动生成描述+关键词
- send_group_quote  核心工具，Planner 根据上下文关键词检索并发送语录
- /群友语录      随机语录（手动 fallback）

语录库与 Maibot 主表情池完全隔离，不污染主表情系统。
"""

from __future__ import annotations

import asyncio
import base64
import binascii
import hashlib
import json
import os
import re
import sqlite3
import time
from typing import Any

from maibot_sdk import Command, Field, MaiBotPlugin, PluginConfigBase


# ---------------------------------------------------------------------------
# 路径常量
# ---------------------------------------------------------------------------

PLUGIN_DIR = os.path.dirname(os.path.abspath(__file__))
DATA_DIR = os.path.join(PLUGIN_DIR, "data")
DB_PATH = os.path.join(DATA_DIR, "quotes.db")
IMAGES_DIR = os.path.join(DATA_DIR, "images")


# ---------------------------------------------------------------------------
# VLM / LLM 提示词
# ---------------------------------------------------------------------------

VLM_DESC_PROMPT = """你是 QQ 群友语錄圖片的審核與結構化標註器。這張圖片正被投稿至 QQ 群語錄插件。

請依序審核：
1. 必須是可辨識的 QQ 對話截圖；其他軟體、網頁、純圖片或非 QQ 聊天不通過。
2. 必須具有可獨立理解、值得作為群友語錄保存的實質內容；普通聊天流水帳、無意義內容，以及只是不斷重複嵌套的聊天截圖不通過。
3. 不得含有直接色情、露骨性描述、色情擦邊、血腥或其他明顯敏感內容。
4. 不得含有歧視、仇恨、針對身分群體的侮辱、嚴重人身攻擊或煽動衝突的內容。
5. 發言人 QQ 暱稱與關鍵發言內容都必須清晰可辨；看不清、無法確認是 QQ 對話，或無法可靠辨認內容時不通過，不要猜測。

通過時請：
- 按畫面閱讀順序為發言編號，description 必須保留「發言人、發言內容、其回覆的內容（如有）」。
- speakers 填寫畫面中清晰可見的發言人 QQ 暱稱，必須照原文抄錄，不得省略、概括或虛構；看不清則不要填。
- keywords 必須優先包含所有 speakers，並補充有辨識度的原句短語、事件和主題，方便之後按群友名稱或聊天上下文檢索。
- 不要把頭像人物、投稿者或機器人臆測成發言人；只記錄圖片文字中明確可見的名稱。

不通過時只需在 reason 填簡短分類理由，例如「非QQ對話」「無實質內容」「低俗色情」「歧視衝突」「文字無法辨識」。
嚴格只返回一行合法 JSON，不要解釋，不要使用 Markdown；approved 必須是 JSON 布林值：
{"approved":true,"reason":"","description":"1. 發言人「暱稱」：發言內容；回覆內容：……","speakers":["暱稱"],"keywords":["暱稱","原句短語","主題"]}
"""

KW_EXTRACT_PROMPT = (
    "下面是最近的群聊记录。请从中提取 1~3 个最适合用来检索群友语录/梗图的关键词，"
    "只返回关键词，用逗号分隔，不要解释：\n\n{context}"
)

QUOTE_COMMAND_PATTERN = (
    r"^/(?:群友语录|uu语录|群u语录)"
    r"(?:(?:\s*(?:(?:编号|編號)\s*)?[#＃]\s*|\s+(?:(?:编号|編號)\s*)?)"
    r"(?P<quote_id>\d+))?\s*$"
)

MENTION_RANDOM_PATTERN = (
    r"^@.+?\s+(?:搬|发|發|来点[史屎石]|來點[史屎石]|岁月史书|歲月史書)\s*$"
)
MENTION_USER_PATTERN = (
    r"^@.+?\s+(?:来点|來點)\s*@.+?\s*的(?:语录|語錄|史)\s*$"
)


# ---------------------------------------------------------------------------
# 配置模型
# ---------------------------------------------------------------------------


class PluginSectionConfig(PluginConfigBase):
    """插件基础配置。"""

    __ui_label__ = "插件"
    __ui_icon__ = "package"
    __ui_order__ = 0

    enabled: bool = Field(default=True, description="是否启用群友语录插件")
    config_version: str = Field(default="2.0.0", description="配置版本")


class VlmConfig(PluginConfigBase):
    """视觉模型描述生成配置。"""

    __ui_label__ = "VLM 描述"
    __ui_icon__ = "eye"
    __ui_order__ = 1

    enabled: bool = Field(default=True, description="投稿时是否用视觉模型自动生成描述与关键词")
    model: str = Field(
        default="",
        description="视觉模型任务名，留空使用 host 默认 vlm 任务",
    )
    max_keywords: int = Field(default=6, description="每个语录最多保留的关键词数量")
    backfill_on_load: bool = Field(
        default=True, description="启动时是否后台补全存量无描述语录"
    )
    backfill_concurrency: int = Field(
        default=2, description="存量补全并发数"
    )


class RetrievalConfig(PluginConfigBase):
    """检索配置。"""

    __ui_label__ = "检索"
    __ui_icon__ = "search"
    __ui_order__ = 2

    context_message_limit: int = Field(
        default=12, description="无关键词时拉取的最近消息条数用于上下文提取"
    )
    random_fallback: bool = Field(
        default=True, description="关键词未命中时是否随机发一张兜底"
    )


class LimitsConfig(PluginConfigBase):
    """限制配置。"""

    __ui_label__ = "限制"
    __ui_icon__ = "shield"
    __ui_order__ = 3

    max_quotes_per_group: int = Field(
        default=500, description="每个群最大语录数量，超出后自动清理最旧记录"
    )


class GroupQuotesConfig(PluginConfigBase):
    """群友语录插件配置。"""

    plugin: PluginSectionConfig = Field(default_factory=PluginSectionConfig)
    vlm: VlmConfig = Field(default_factory=VlmConfig)
    retrieval: RetrievalConfig = Field(default_factory=RetrievalConfig)
    limits: LimitsConfig = Field(default_factory=LimitsConfig)




# ---------------------------------------------------------------------------
# 插件主类
# ---------------------------------------------------------------------------


class GroupQuotesPlugin(MaiBotPlugin):
    """群友语录插件：群内图片投稿 + Planner 关键词检索发送。"""

    config_model = GroupQuotesConfig

    def __init__(self) -> None:
        super().__init__()
        self._db: sqlite3.Connection | None = None
        self._backfill_task: asyncio.Task[None] | None = None

    # ---- 生命周期 ----

    async def on_load(self) -> None:
        """插件加载时初始化数据目录、数据库并启动存量补全。"""
        os.makedirs(DATA_DIR, exist_ok=True)
        os.makedirs(IMAGES_DIR, exist_ok=True)

        self._db = sqlite3.connect(DB_PATH)
        self._db.row_factory = sqlite3.Row
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute(
            """
            CREATE TABLE IF NOT EXISTS group_quotes (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                group_id TEXT NOT NULL,
                user_id TEXT NOT NULL,
                user_nickname TEXT NOT NULL DEFAULT '',
                image_hash TEXT NOT NULL,
                image_path TEXT NOT NULL,
                message_id TEXT NOT NULL DEFAULT '',
                created_at REAL NOT NULL
            )
            """
        )
        self._db.execute(
            "CREATE INDEX IF NOT EXISTS idx_group_quotes_group_id "
            "ON group_quotes(group_id)"
        )
        self._db.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_group_quotes_dedup "
            "ON group_quotes(group_id, image_hash)"
        )
        self._ensure_schema()
        self._db.commit()

        self.ctx.logger.info("群友语录插件已加载（/群友语录 优先按上下文关键词检索）")

        if self.config.vlm.enabled and self.config.vlm.backfill_on_load:
            self._backfill_task = asyncio.create_task(self._backfill_descriptions())

    async def on_unload(self) -> None:
        """插件卸载时关闭数据库连接。"""
        if self._backfill_task is not None and not self._backfill_task.done():
            self._backfill_task.cancel()
            try:
                await self._backfill_task
            except asyncio.CancelledError:
                pass
            self._backfill_task = None
        if self._db:
            self._db.close()
            self._db = None
        self.ctx.logger.info("群友语录插件已卸载")

    async def on_config_update(
        self, scope: str, config_data: dict[str, Any], version: str
    ) -> None:
        """配置热重载时执行。"""

    # ---- 数据库迁移 ----

    def _ensure_schema(self) -> None:
        """为旧库补充新字段（description / keywords / vlm_processed）。"""
        assert self._db is not None
        cursor = self._db.execute("PRAGMA table_info(group_quotes)")
        existing_cols = {row[1] for row in cursor.fetchall()}
        new_cols = {
            "description": "TEXT NOT NULL DEFAULT ''",
            "keywords": "TEXT NOT NULL DEFAULT ''",
            "vlm_processed": "INTEGER NOT NULL DEFAULT 0",
        }
        for col, decl in new_cols.items():
            if col not in existing_cols:
                self._db.execute(
                    f"ALTER TABLE group_quotes ADD COLUMN {col} {decl}"
                )
                self.ctx.logger.info(f"已迁移：新增字段 {col}")

    # ---- /投稿 ----

    @Command("submit_quote", description="投稿群友语录（回复图片使用）", pattern=r"/投稿")
    async def handle_submit_quote(
        self,
        stream_id: str = "",
        group_id: str = "",
        user_id: str = "",
        message: dict | None = None,
        **kwargs: Any,
    ) -> tuple[bool, str, bool]:
        """处理 /投稿：回复含图片的消息，将该图片投稿到本群语录库。"""
        del kwargs

        if not group_id:
            await self.ctx.send.text("群友语录仅支持群聊使用", stream_id)
            return False, "非群聊消息", True

        if not message or not isinstance(message, dict):
            await self.ctx.send.text("无法获取消息内容", stream_id)
            return False, "消息为空", True

        raw_message = message.get("raw_message", [])
        reply_component = _find_component(raw_message, "reply")
        if not reply_component:
            await self.ctx.send.text("请回复一张图片来投稿 /投稿", stream_id)
            return False, "无回复", True

        reply_data = reply_component.get("data", {})
        if isinstance(reply_data, dict):
            target_message_id = str(
                reply_data.get("target_message_id")
                or reply_data.get("message_id")
                or reply_data.get("id")
                or ""
            ).strip()
        else:
            target_message_id = str(reply_data).strip()

        if not target_message_id:
            await self.ctx.send.text("无法获取被回复的消息", stream_id)
            return False, "无目标消息 ID", True

        # 限定当前 stream，避免 QQ 消息 ID 在不同会话中碰撞；图片刚入站时
        # 可能尚未完成持久化，短暂重试可减少偶发的“引用图片不存在”。
        replied_message: dict[str, Any] | None = None
        image_component: dict[str, Any] | None = None
        for delay in (0.0, 0.4, 1.0):
            if delay:
                await asyncio.sleep(delay)
            try:
                result = await self.ctx.message.get_by_id(
                    target_message_id,
                    stream_id=stream_id,
                    include_binary_data=True,
                )
            except Exception as exc:
                self.ctx.logger.debug(
                    f"获取被回复消息异常 id={target_message_id}: {exc}"
                )
                continue
            if not isinstance(result, dict):
                continue
            replied_message = result
            replied_raw = replied_message.get("raw_message", [])
            # NapCat 会把部分 QQ 图片（例如自定义表情）归类成 emoji，投稿时
            # 两种媒体都应接受。
            image_component = (
                _find_component(replied_raw, "image")
                or _find_component(replied_raw, "emoji")
            )
            if image_component and str(
                image_component.get("binary_data_base64", "")
            ).strip():
                break

        if replied_message is None:
            self.ctx.logger.warning(
                f"获取被回复消息失败 id={target_message_id} stream={stream_id}"
            )
            await self.ctx.send.text("被回复的消息不存在或已过期", stream_id)
            return False, "获取消息失败", True
        if not image_component:
            await self.ctx.send.text("被回复的消息中没有可投稿的图片", stream_id)
            return False, "无图片", True

        image_base64 = str(image_component.get("binary_data_base64", "")).strip()
        if not image_base64:
            await self.ctx.send.text(
                "引用图片的数据尚未就绪或缓存已失效，请重新发送图片后立即投稿",
                stream_id,
            )
            return False, "无图片数据", True

        # 不信任平台提供的 hash：始终按实际图片字节计算内容哈希，避免同图因
        # 平台 ID/转发方式不同而绕过群内去重。
        try:
            image_bytes = base64.b64decode(image_base64, validate=True)
        except (ValueError, binascii.Error) as exc:
            self.ctx.logger.warning(f"投稿图片 Base64 无效: {exc}")
            await self.ctx.send.text("无法解析图片数据，请重新发送图片后再投稿", stream_id)
            return False, "图片数据无效", True
        image_hash = hashlib.sha256(image_bytes).hexdigest()

        user_nickname = _extract_user_nickname(message)
        assert self._db is not None

        # 先查重；最重要的是审核通过前绝不写入永久图目录或数据库。
        duplicate = self._db.execute(
            "SELECT 1 FROM group_quotes WHERE group_id=? AND image_hash=? LIMIT 1",
            (group_id, image_hash),
        ).fetchone()
        if duplicate:
            await self.ctx.send.text("这张图片已经投稿过了", stream_id)
            return False, "重复投稿", True

        # 同步 VLM 审核 + 描述（投稿时即时返回审核结果）。审核服务异常、
        # 超时或返回格式无效时一律拒绝本次投稿，避免未经审核的图片 fail-open。
        approved = True
        reason = ""
        description = ""
        keywords: list[str] = []
        if self.config.vlm.enabled:
            try:
                approved, reason, description, keywords = (
                    await self._vlm_review_and_describe(image_base64)
                )
            except Exception as exc:
                self.ctx.logger.warning(f"投稿审核异常 hash={image_hash}: {exc}")
                approved = False
                reason = "审核服务暂时不可用，请稍后重试"

        if not approved:
            short_reason = reason or "内容不适合收录"
            await self.ctx.send.text(f"投稿未通过审核：{short_reason}", stream_id)
            return False, f"审核未通过: {short_reason}", True

        # 只有审核通过后才落盘，再写数据库。
        group_image_dir = os.path.join(IMAGES_DIR, group_id)
        os.makedirs(group_image_dir, exist_ok=True)
        image_path = os.path.join(group_image_dir, f"{image_hash}.png")
        created_file = False
        if not os.path.exists(image_path):
            with open(image_path, "wb") as f:
                f.write(image_bytes)
            created_file = True

        # 审核通过：入库（带描述与关键词）
        kw_text = ",".join(keywords) if keywords else ""
        try:
            cursor = self._db.execute(
                """
                INSERT INTO group_quotes
                    (group_id, user_id, user_nickname, image_hash,
                     image_path, message_id, created_at,
                     description, keywords, vlm_processed)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    group_id, user_id, user_nickname, image_hash,
                    image_path, target_message_id, time.time(),
                    description, kw_text, 1 if self.config.vlm.enabled else 0,
                ),
            )
            self._db.commit()
        except sqlite3.IntegrityError:
            # 并发投稿可能在预查重后由另一请求先入库；此时文件由该记录共用，
            # 不应删除。
            await self.ctx.send.text("这张图片已经投稿过了", stream_id)
            return False, "重复投稿", True
        except Exception:
            # 数据库写入失败时不留下孤儿文件；若已有记录引用则保留。
            referenced = self._db.execute(
                "SELECT 1 FROM group_quotes WHERE image_path=? LIMIT 1", (image_path,)
            ).fetchone()
            if created_file and not referenced:
                try:
                    os.remove(image_path)
                except OSError as exc:
                    self.ctx.logger.warning(f"清理投稿孤儿图片失败 {image_path}: {exc}")
            raise

        quote_id = cursor.lastrowid
        self._enforce_group_limit(group_id)

        msg = f"投稿成功！编号 #{quote_id}"
        if description:
            msg += f"（{description}）"
        await self.ctx.send.text(msg, stream_id)
        return True, f"投稿成功 #{quote_id}", True

    # ---- /群友语录 ----

    @Command(
        "random_quote",
        description="获取群友语录；可用 /群友语录 123 指定编号",
        pattern=QUOTE_COMMAND_PATTERN,
    )
    async def handle_random_quote(
        self,
        stream_id: str = "",
        group_id: str = "",
        **kwargs: Any,
    ) -> tuple[bool, str, bool]:
        """处理 /群友语录：指定编号直取，否则按上下文检索后随机兜底。"""

        if not group_id:
            await self.ctx.send.text("群友语录仅支持群聊使用", stream_id)
            return False, "非群聊消息", True

        matched_groups = kwargs.get("matched_groups")
        if not isinstance(matched_groups, dict):
            matched_groups = {}
        raw_quote_id = str(matched_groups.get("quote_id") or "").strip()
        if not raw_quote_id:
            # 兼容未注入 matched_groups 的旧 Host/SDK：从原始命令文本回退解析。
            raw_text = str(kwargs.get("text") or kwargs.get("plain_text") or "").strip()
            message = kwargs.get("message")
            if not raw_text and isinstance(message, dict):
                raw_text = str(
                    message.get("plain_text")
                    or message.get("processed_plain_text")
                    or ""
                ).strip()
            match = re.fullmatch(QUOTE_COMMAND_PATTERN, raw_text)
            if match is not None:
                raw_quote_id = str(match.group("quote_id") or "").strip()

        # 指定编号时不调用上下文关键词模型，直接按本群 id 精确查询。
        if raw_quote_id:
            quote_id = int(raw_quote_id)
            row = self._quote_row_by_id(group_id, quote_id)
            if row is None:
                await self.ctx.send.text(
                    f"本群找不到编号 #{quote_id} 的语录", stream_id
                )
                return False, f"指定语录 #{quote_id} 不存在", True
            sent = await self._send_quote_row(row, stream_id)
            if not sent:
                await self.ctx.send.text(
                    f"编号 #{quote_id} 的语录图片已丢失或发送失败", stream_id
                )
            return sent, f"已发送指定语录 #{quote_id}", True

        # 未指定编号：优先从最近对话上下文提取关键词检索
        keyword = await self._extract_keywords_from_context(stream_id)
        row = None
        matched = False
        if keyword:
            row = self._search_quotes(group_id, keyword, limit=1)
            if row:
                matched = True
                self.ctx.logger.info(f"语录按上下文关键词命中: {keyword}")

        # 未命中 -> 随机兜底
        if row is None:
            row = self._random_quote_row(group_id)
        if not row:
            await self.ctx.send.text("本群还没有语录，快来 /投稿 吧！", stream_id)
            return False, "语录为空", True

        sent = await self._send_quote_row(row, stream_id)
        return (sent, f"已发送语录(关键词{keyword})" if matched else "已发送语录", True)

    @Command(
        "mention_random_quote",
        description="@机器人后用搬/发/来点史等口令随机发送群友语录",
        pattern=MENTION_RANDOM_PATTERN,
    )
    async def handle_mention_random_quote(
        self,
        stream_id: str = "",
        group_id: str = "",
        message: dict | None = None,
        **kwargs: Any,
    ) -> tuple[bool, str, bool]:
        """处理 @bot 随机口令；纯随机，不进行上下文检索。"""
        del kwargs
        if not group_id:
            await self.ctx.send.text("群友语录仅支持群聊使用", stream_id)
            return False, "非群聊消息", True
        if not _message_mentions_bot(message):
            return False, "随机语录口令未@机器人", False

        row = self._random_quote_row(group_id)
        if row is None:
            await self.ctx.send.text("本群还没有语录，快来 /投稿 吧！", stream_id)
            return False, "语录为空", True
        sent = await self._send_quote_row(row, stream_id)
        return sent, "已发送纯随机语录", True

    @Command(
        "mention_user_quote",
        description="@机器人 来点@用户 的语录/史，按被@用户昵称检索",
        pattern=MENTION_USER_PATTERN,
    )
    async def handle_mention_user_quote(
        self,
        stream_id: str = "",
        group_id: str = "",
        message: dict | None = None,
        **kwargs: Any,
    ) -> tuple[bool, str, bool]:
        """以被@群友的当前群昵称作为唯一关键词检索语录。"""
        del kwargs
        if not group_id:
            await self.ctx.send.text("群友语录仅支持群聊使用", stream_id)
            return False, "非群聊消息", True
        if not _message_mentions_bot(message):
            return False, "用户语录口令未@机器人", False

        target_name = _extract_target_mention_name(message)
        if not target_name:
            await self.ctx.send.text("请在口令中再 @ 一位群友", stream_id)
            return False, "未找到目标群友", True

        row = self._search_quotes(group_id, target_name, limit=1)
        if row is None:
            await self.ctx.send.text(
                f"本群找不到与「{target_name}」相关的语录", stream_id
            )
            return False, f"用户关键词未命中: {target_name}", True
        self.ctx.logger.info(f"语录按被@用户昵称命中: {target_name}")
        sent = await self._send_quote_row(row, stream_id)
        return sent, f"已发送 {target_name} 的语录", True


    # ---- 核心工具已移除 ----
    # 原 send_group_quote(@Tool core_tool) 会让 Planner 自然触发，冷不丁发图会吓人。
    # 改为 /群友语录 手动触发时优先按上下文关键词检索，未命中再随机（见 handle_random_quote）。

    # ---- VLM 描述生成 ----

    async def _generate_and_store_description(
        self, group_id: str, image_hash: str, image_base64: str
    ) -> None:
        """调用视觉模型为单条语录生成描述与关键词并写入数据库。"""
        assert self._db is not None
        try:
            _approved, _reason, description, keywords = (
                await self._vlm_review_and_describe(image_base64)
            )
        except Exception as exc:
            self.ctx.logger.warning(
                f"VLM 生成描述失败 hash={image_hash}: {exc}"
            )
            return

        if not description and not keywords:
            return

        kw_text = ",".join(keywords) if keywords else ""
        self._db.execute(
            "UPDATE group_quotes SET description=?, keywords=?, vlm_processed=1 "
            "WHERE group_id=? AND image_hash=?",
            (description, kw_text, group_id, image_hash),
        )
        self._db.commit()
        self.ctx.logger.info(
            f"语录描述已生成 hash={image_hash[:12]} desc={description!r} kw={keywords}"
        )

    async def _vlm_review_and_describe(
        self, image_base64: str
    ) -> tuple[bool, str, str, list[str]]:
        """调用视觉模型审核并生成描述，返回 (approved, reason, description, keywords)。

        VLM 失败、超时或结果无法解析时拒绝本次投稿（fail-closed）。
        """
        prompt = [
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": VLM_DESC_PROMPT},
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/png;base64,{image_base64}"
                        },
                    },
                ],
            }
        ]
        result = await self.ctx.llm.generate(
            prompt=prompt,
            model=self.config.vlm.model or "vlm",
        )
        if not isinstance(result, dict):
            return False, "审核服务返回格式异常，请稍后重试", "", []
        if not result.get("success", True):
            self.ctx.logger.debug(
                f"VLM 调用失败: {result.get('error') or result.get('response')}"
            )
            return False, "审核服务暂时不可用，请稍后重试", "", []
        text = str(
            result.get("response") or result.get("content") or ""
        ).strip()
        return _parse_vlm_result(text, self.config.vlm.max_keywords)

    async def _backfill_descriptions(self) -> None:
        """后台为存量无描述语录补全 VLM 描述。"""
        assert self._db is not None
        try:
            cursor = self._db.execute(
                "SELECT group_id, image_hash, image_path FROM group_quotes "
                "WHERE vlm_processed=0"
            )
            pending = cursor.fetchall()
        except Exception as exc:
            self.ctx.logger.warning(f"读取待补全语录失败: {exc}")
            return

        if not pending:
            return

        self.ctx.logger.info(f"开始后台补全 {len(pending)} 条存量语录描述")
        sem = asyncio.Semaphore(max(1, self.config.vlm.backfill_concurrency))

        async def _one(row: sqlite3.Row) -> None:
            async with sem:
                image_path = str(row["image_path"])
                if not os.path.isfile(image_path):
                    return
                try:
                    with open(image_path, "rb") as f:
                        image_base64 = base64.b64encode(f.read()).decode("utf-8")
                except Exception as exc:
                    self.ctx.logger.warning(f"读取图片失败 {image_path}: {exc}")
                    return
                await self._generate_and_store_description(
                    str(row["group_id"]),
                    str(row["image_hash"]),
                    image_base64,
                )

        await asyncio.gather(*[_one(r) for r in pending], return_exceptions=True)
        self.ctx.logger.info("存量语录描述补全完成")

    # ---- 上下文关键词提取 ----

    async def _extract_keywords_from_context(self, stream_id: str) -> str:
        """无关键词时，从最近对话中提取检索关键词。失败返回空。"""
        if not stream_id:
            return ""
        try:
            messages = await self.ctx.message.get_recent(
                stream_id,
                limit=max(1, self.config.retrieval.context_message_limit),
            )
        except Exception as exc:
            self.ctx.logger.debug(f"获取对话上下文异常: {exc}")
            return ""

        if not messages:
            return ""

        context_text = _build_context_text(messages)
        if not context_text.strip():
            return ""

        try:
            result = await self.ctx.llm.generate(
                prompt=KW_EXTRACT_PROMPT.format(context=context_text[:1200]),
                model="utils",
            )
        except Exception as exc:
            self.ctx.logger.debug(f"LLM 提取关键词异常: {exc}")
            return ""

        if not isinstance(result, dict):
            return ""
        text = str(
            result.get("response") or result.get("content") or ""
        ).strip()
        # 取第一个关键词
        first = re.split(r"[,，、\s]+", text)
        for kw in first:
            kw = kw.strip().strip("\"'`")
            if len(kw) >= 2:
                return kw
        return ""

    # ---- 检索 / 发送辅助 ----

    def _search_quotes(
        self, group_id: str, keyword: str, limit: int = 1
    ) -> sqlite3.Row | None:
        """按关键词模糊匹配 description / keywords，随机返回。"""
        assert self._db is not None
        like = f"%{keyword}%"
        cursor = self._db.execute(
            """
            SELECT id, image_path, image_hash, description, keywords,
                   user_id, user_nickname
            FROM group_quotes
            WHERE group_id = ? AND (
                keywords LIKE ? OR description LIKE ?
            )
            ORDER BY RANDOM() LIMIT ?
            """,
            (group_id, like, like, limit),
        )
        return cursor.fetchone()

    def _quote_row_by_id(
        self, group_id: str, quote_id: int
    ) -> sqlite3.Row | None:
        """按编号精确返回本群语录；group_id 条件防止跨群读取。"""
        assert self._db is not None
        cursor = self._db.execute(
            """
            SELECT id, image_path, image_hash, description, keywords,
                   user_id, user_nickname
            FROM group_quotes
            WHERE group_id = ? AND id = ?
            LIMIT 1
            """,
            (group_id, quote_id),
        )
        return cursor.fetchone()

    def _random_quote_row(self, group_id: str) -> sqlite3.Row | None:
        """随机返回一条本群语录。"""
        assert self._db is not None
        cursor = self._db.execute(
            """
            SELECT id, image_path, image_hash, description, keywords,
                   user_id, user_nickname
            FROM group_quotes
            WHERE group_id = ?
            ORDER BY RANDOM() LIMIT 1
            """,
            (group_id,),
        )
        return cursor.fetchone()

    async def _send_quote_row(
        self, row: sqlite3.Row, stream_id: str
    ) -> bool:
        """读取语录图片并发送，附带投稿人信息，自动清理丢失文件的孤立记录。"""
        assert self._db is not None
        image_path = str(row["image_path"])
        try:
            quote_id = int(row["id"])
        except (KeyError, TypeError, ValueError):
            quote_id = 0

        if not os.path.isfile(image_path):
            self._db.execute(
                "DELETE FROM group_quotes WHERE id=?",
                (quote_id,),
            )
            self._db.commit()
            self.ctx.logger.warning(f"语录图片丢失已清理: {image_path}")
            return False

        try:
            with open(image_path, "rb") as f:
                image_bytes = f.read()
        except Exception as exc:
            self.ctx.logger.warning(f"读取语录图片失败: {exc}")
            return False

        image_base64 = base64.b64encode(image_bytes).decode("utf-8")
        caption = _build_quote_caption(row)
        segments: list[dict[str, str]] = [
            {"type": "image", "content": image_base64}
        ]
        if caption:
            segments.append({"type": "text", "content": caption})

        # 使用 hybrid 将图片与编号/投稿人合并为同一条 QQ 消息。
        try:
            sent = await self.ctx.send.hybrid(
                segments,
                stream_id,
                processed_plain_text=caption,
            )
        except Exception as exc:
            self.ctx.logger.warning(f"发送图文语录失败: {exc}")
            return False
        if not sent:
            self.ctx.logger.warning("发送图文语录失败: send.hybrid 返回 False")
            return False
        return True

    def _enforce_group_limit(self, group_id: str) -> None:
        """超出上限时清理最旧记录及其图片文件。"""
        assert self._db is not None
        max_quotes = self.config.limits.max_quotes_per_group
        cursor = self._db.execute(
            "SELECT COUNT(*) FROM group_quotes WHERE group_id = ?",
            (group_id,),
        )
        count = cursor.fetchone()[0]
        if count <= max_quotes:
            return

        excess = count - max_quotes
        stale_cursor = self._db.execute(
            "SELECT image_path FROM group_quotes WHERE group_id = ? "
            "ORDER BY created_at ASC LIMIT ?",
            (group_id, excess),
        )
        for srow in stale_cursor:
            stale_path = srow[0]
            if stale_path and os.path.isfile(stale_path):
                try:
                    os.remove(stale_path)
                except OSError:
                    pass
        self._db.execute(
            "DELETE FROM group_quotes WHERE group_id = ? AND id IN ("
            " SELECT id FROM group_quotes WHERE group_id = ?"
            " ORDER BY created_at ASC LIMIT ?)",
            (group_id, group_id, excess),
        )
        self._db.commit()


# ---------------------------------------------------------------------------
# 模块级辅助函数
# ---------------------------------------------------------------------------


def _find_component(
    raw_message: list[Any], comp_type: str
) -> dict[str, Any] | None:
    """在 raw_message 列表中查找指定类型的组件。"""
    for comp in raw_message:
        if isinstance(comp, dict) and comp.get("type") == comp_type:
            return comp
    return None


def _extract_user_nickname(message: dict[str, Any]) -> str:
    """从消息字典中提取发送者昵称。"""
    message_info = message.get("message_info", {})
    if isinstance(message_info, dict):
        user_info = message_info.get("user_info", {})
        if isinstance(user_info, dict):
            return str(user_info.get("user_nickname", ""))
    return ""


def _message_mentions_bot(message: dict[str, Any] | None) -> bool:
    """使用 Host 已判定的 @bot/提及标记，避免普通 @群友 文本误触发。"""
    if not isinstance(message, dict):
        return False
    if bool(message.get("is_at") or message.get("is_mentioned")):
        return True
    message_info = message.get("message_info", {})
    if isinstance(message_info, dict):
        additional = message_info.get("additional_config", {})
        if isinstance(additional, dict):
            return bool(additional.get("at_bot") or additional.get("is_mentioned"))
    return False


def _extract_target_mention_name(message: dict[str, Any] | None) -> str:
    """取得 @bot 以外最后一个被 @ 用户的群名片/昵称，缺失时回退 QQ 号。"""
    if not isinstance(message, dict):
        return ""
    names: list[str] = []
    raw_message = message.get("raw_message", [])
    if not isinstance(raw_message, list):
        return ""
    for component in raw_message:
        if not isinstance(component, dict) or component.get("type") != "at":
            continue
        data = component.get("data", {})
        if isinstance(data, dict):
            name = str(
                data.get("target_user_cardname")
                or data.get("target_user_nickname")
                or data.get("target_user_id")
                or data.get("qq")
                or ""
            ).strip()
        else:
            name = str(data or "").strip()
        if name:
            names.append(name)
    # 正常 QQ 消息依次为 @bot、@目标；若 bot 由 mention_bot 表示，列表中
    # 可能只有目标，因此统一取最后一个，由 _message_mentions_bot 保证已 @bot。
    return names[-1] if names else ""


def _build_quote_caption(row: sqlite3.Row) -> str:
    """构造发送语录时的投稿人标注文字。

    格式：`编号#xxx，由"昵称(QQ号)"投稿!`
    缺少投稿人数据的老记录只写编号 `编号#xxx`。
    """
    try:
        quote_id = int(row["id"])
    except (KeyError, TypeError, ValueError):
        return ""

    nickname = str(row["user_nickname"] or "").strip()
    user_id = str(row["user_id"] or "").strip()

    if not user_id and not nickname:
        return f"编号#{quote_id}"

    if nickname and user_id:
        contributor = f'{nickname}({user_id})'
    elif user_id:
        contributor = user_id
    else:
        contributor = nickname

    if not contributor:
        return f"编号#{quote_id}"
    # 用户要求称昵称(QQ号)为其"昵称"，这里沿用"投稿人"语义
    return f'编号#{quote_id}，由"{contributor}"投稿!'


def _build_context_text(messages: list[Any]) -> str:
    """将消息列表拼成纯文本上下文。"""
    blocks: list[str] = []
    for msg in messages:
        if not isinstance(msg, dict):
            continue
        msg_info = msg.get("message_info", {})
        user_info = msg_info.get("user_info", {}) if isinstance(msg_info, dict) else {}
        nickname = (
            user_info.get("user_nickname")
            or user_info.get("user_name")
            or "群友"
        ) if isinstance(user_info, dict) else "群友"
        content = (
            msg.get("processed_plain_text")
            or msg.get("plain_text")
            or msg.get("content")
            or msg.get("text")
            or ""
        )
        if not str(content).strip():
            continue
        blocks.append(f"{nickname}: {content}")
    return "\n".join(blocks)


def _parse_vlm_result(text: str, max_keywords: int) -> tuple[bool, str, str, list[str]]:
    """从 VLM 返回文本中解析 approved/reason/description/keywords。"""
    if not text:
        return False, "审核服务未返回有效结果，请稍后重试", "", []

    # 尝试提取 JSON 块
    json_text = text
    match = re.search(r"\{.*\}", text, re.DOTALL)
    if match:
        json_text = match.group(0)

    try:
        data = json.loads(json_text)
        if isinstance(data, dict):
            raw_approved = data.get("approved")
            if isinstance(raw_approved, bool):
                approved = raw_approved
            elif isinstance(raw_approved, str) and raw_approved.strip().lower() in {
                "true", "false"
            }:
                approved = raw_approved.strip().lower() == "true"
            else:
                return False, "审核结果缺少有效的 approved 字段，请稍后重试", "", []
            reason = str(data.get("reason", "")).strip()
            description = str(data.get("description", "")).strip()
            raw_speakers = data.get("speakers", [])
            if isinstance(raw_speakers, str):
                speakers = [raw_speakers.strip()] if raw_speakers.strip() else []
            elif isinstance(raw_speakers, list):
                speakers = [str(v).strip() for v in raw_speakers if str(v).strip()]
            else:
                speakers = []

            raw_kw = data.get("keywords", [])
            if isinstance(raw_kw, str):
                raw_keywords = [
                    k.strip() for k in re.split(r"[,，、\s]+", raw_kw) if k.strip()
                ]
            elif isinstance(raw_kw, list):
                raw_keywords = [str(k).strip() for k in raw_kw if str(k).strip()]
            else:
                raw_keywords = []

            if approved and (not description or not speakers):
                return False, "发言人或发言内容无法可靠识别", "", []

            # speakers 置前并去重，确保 max_keywords 截断时仍优先保留用户名。
            keywords: list[str] = []
            seen: set[str] = set()
            for keyword in [*speakers, *raw_keywords]:
                normalized = keyword.casefold()
                if normalized in seen:
                    continue
                seen.add(normalized)
                keywords.append(keyword)
                if len(keywords) >= max_keywords:
                    break
            return approved, reason, description, keywords
    except (json.JSONDecodeError, ValueError):
        pass

    # 审核结果无法可靠解析时绝不放行，防止模型输出漂移绕过审核。
    return False, "审核结果格式异常，请稍后重试", "", []


# ---------------------------------------------------------------------------
# 工厂函数
# ---------------------------------------------------------------------------


def create_plugin() -> GroupQuotesPlugin:
    """创建群友语录插件实例。"""
    return GroupQuotesPlugin()