feat: automatically record UMO names

astrbot/core/core_lifecycle.py

@@ -462,7 +462,7 @@ async def load_pipeline_scheduler(self) -> dict[str, PipelineScheduler]:
         mapping = {}
         for conf_id, ab_config in self.astrbot_config_mgr.confs.items():
             scheduler = PipelineScheduler(
-                PipelineContext(ab_config, self.plugin_manager, conf_id),
+                PipelineContext(ab_config, self.plugin_manager, conf_id, self.db),
             )
             await scheduler.initialize()
             mapping[conf_id] = scheduler
@@ -479,7 +479,7 @@ async def reload_pipeline_scheduler(self, conf_id: str) -> None:
         if not ab_config:
             raise ValueError(f"配置文件 {conf_id} 不存在")
         scheduler = PipelineScheduler(
-            PipelineContext(ab_config, self.plugin_manager, conf_id),
+            PipelineContext(ab_config, self.plugin_manager, conf_id, self.db),
         )
         await scheduler.initialize()
         self.pipeline_scheduler_mapping[conf_id] = scheduler

astrbot/core/db/__init__.py

@@ -850,6 +850,22 @@ async def upsert_umo_alias(
         """Create or update the display alias metadata for a UMO."""
         ...
 
+    @abc.abstractmethod
+    async def upsert_umo_auto_name(
+        self,
+        umo: str,
+        creator_sender_id: str,
+        auto_name: str,
+    ) -> None:
+        """Create or update only the automatically discovered UMO name.
+
+        Args:
+            umo: Unified message origin to name.
+            creator_sender_id: Sender that first caused the UMO to be recorded.
+            auto_name: Name discovered from the inbound platform message.
+        """
+        ...
+
     @abc.abstractmethod
     async def get_umo_alias(self, umo: str) -> UmoAlias | None:
         """Get alias metadata for one UMO."""

astrbot/core/db/po.py

@@ -346,7 +346,7 @@ class UmoAlias(TimestampMixin, SQLModel, table=True):
         sa_column_kwargs={"autoincrement": True},
         default=None,
     )
-    umo: str = Field(nullable=False, max_length=512, unique=True, index=True)
+    umo: str = Field(nullable=False, max_length=512)
     creator_sender_id: str = Field(nullable=False, max_length=255)
     auto_name: str | None = Field(default=None, max_length=255)
     user_alias: str | None = Field(default=None, max_length=255)

astrbot/core/db/sqlite.py

@@ -10,6 +10,7 @@
 from deprecated import deprecated
 from sqlalchemy import CursorResult, Row, case, not_
 from sqlalchemy.dialects.sqlite import dialect as sqlite_dialect
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import defer
 from sqlmodel import col, delete, desc, func, or_, select, text, update
@@ -72,6 +73,9 @@ async def initialize(self) -> None:
             await self._ensure_platform_message_history_checkpoint_column(conn)
             await self._ensure_chatui_project_workspace_columns(conn)
             await self._ensure_conversation_indexes(conn)
+            # The table-level unique constraint already provides an index for UMO
+            # lookups. Older schemas also created this redundant explicit index.
+            await conn.execute(text("DROP INDEX IF EXISTS ix_umo_aliases_umo"))
             await conn.commit()
 
     async def _ensure_conversation_indexes(self, conn) -> None:
@@ -2152,30 +2156,80 @@ async def upsert_umo_alias(
         auto_name: str | None,
         user_alias: str | None,
     ) -> UmoAlias:
-        """Create or update alias metadata for a UMO."""
+        """Create or replace user-controlled alias metadata for a UMO.
+
+        Args:
+            umo: Unified message origin to name.
+            creator_sender_id: Sender responsible for the manual alias update.
+            auto_name: Latest name discovered from platform metadata.
+            user_alias: User-controlled display alias.
+
+        Returns:
+            Persisted UMO alias record.
+        """
+        now = datetime.now(timezone.utc)
+        statement = sqlite_insert(UmoAlias).values(
+            umo=umo,
+            creator_sender_id=creator_sender_id,
+            auto_name=auto_name,
+            user_alias=user_alias,
+            created_at=now,
+            updated_at=now,
+        )
+        statement = statement.on_conflict_do_update(
+            index_elements=[UmoAlias.umo],
+            set_={
+                "creator_sender_id": statement.excluded.creator_sender_id,
+                "auto_name": statement.excluded.auto_name,
+                "user_alias": statement.excluded.user_alias,
+                "updated_at": now,
+            },
+        )
         async with self.get_db() as session:
             session: AsyncSession
             async with session.begin():
+                await session.execute(statement)
                 result = await session.execute(
                     select(UmoAlias).where(col(UmoAlias.umo) == umo)
                 )
-                alias = result.scalar_one_or_none()
-                if alias:
-                    alias.creator_sender_id = creator_sender_id
-                    alias.auto_name = auto_name
-                    alias.user_alias = user_alias
-                    alias.updated_at = datetime.now(timezone.utc)
-                else:
-                    alias = UmoAlias(
-                        umo=umo,
-                        creator_sender_id=creator_sender_id,
-                        auto_name=auto_name,
-                        user_alias=user_alias,
-                    )
-                    session.add(alias)
-                await session.flush()
-                await session.refresh(alias)
-                return alias
+                return result.scalar_one()
+
+    async def upsert_umo_auto_name(
+        self,
+        umo: str,
+        creator_sender_id: str,
+        auto_name: str,
+    ) -> None:
+        """Persist an automatic UMO name without changing its manual alias.
+
+        Args:
+            umo: Unified message origin to name.
+            creator_sender_id: Sender that first caused the UMO to be recorded.
+            auto_name: Name discovered from the inbound platform message.
+        """
+        now = datetime.now(timezone.utc)
+        statement = sqlite_insert(UmoAlias).values(
+            umo=umo,
+            creator_sender_id=creator_sender_id,
+            auto_name=auto_name,
+            user_alias=None,
+            created_at=now,
+            updated_at=now,
+        )
+        statement = statement.on_conflict_do_update(
+            index_elements=[UmoAlias.umo],
+            set_={
+                "auto_name": statement.excluded.auto_name,
+                "updated_at": now,
+            },
+            where=col(UmoAlias.auto_name).is_distinct_from(
+                statement.excluded.auto_name
+            ),
+        )
+        async with self.get_db() as session:
+            session: AsyncSession
+            async with session.begin():
+                await session.execute(statement)
 
     async def get_umo_alias(self, umo: str) -> UmoAlias | None:
         """Get alias metadata for one UMO."""

astrbot/core/pipeline/context.py

@@ -8,6 +8,7 @@
 from .context_utils import call_event_hook, call_handler
 
 if TYPE_CHECKING:
+    from astrbot.core.db import BaseDatabase
     from astrbot.core.star import PluginManager
 
 
@@ -18,5 +19,6 @@ class PipelineContext:
     astrbot_config: AstrBotConfig  # AstrBot 配置对象
     plugin_manager: PluginManager  # 插件管理器对象
     astrbot_config_id: str
+    db_helper: BaseDatabase | None = None
     call_handler = call_handler
     call_event_hook = call_event_hook

astrbot/core/pipeline/waking_check/stage.py

@@ -13,6 +13,7 @@
 
 from ..context import PipelineContext
 from ..stage import Stage, register_stage
+from .umo_auto_name import UmoAutoNameRecorder
 
 UNIQUE_SESSION_ID_BUILDERS: dict[str, Callable[[AstrMessageEvent], str | None]] = {
     "aiocqhttp": lambda e: f"{e.get_sender_id()}_{e.get_group_id()}",
@@ -73,6 +74,10 @@ async def initialize(self, ctx: PipelineContext) -> None:
         )
         platform_settings = self.ctx.astrbot_config.get("platform_settings", {})
         self.unique_session = platform_settings.get("unique_session", False)
+        self._umo_auto_name_recorder = UmoAutoNameRecorder(
+            ctx.db_helper,
+            ctx.astrbot_config_id,
+        )
 
     async def process(
         self,
@@ -218,6 +223,8 @@ async def process(
                         f"{star_map[handler.handler_module_path].name}.",
                     )
                     event.stop_event()
+                    if event.is_wake:
+                        self._umo_auto_name_recorder.schedule(event)
                     return
 
                 is_wake = True
@@ -244,5 +251,7 @@ async def process(
         event.set_extra("activated_handlers", activated_handlers)
         event.set_extra("handlers_parsed_params", handlers_parsed_params)
 
-        if not is_wake:
+        if is_wake:
+            self._umo_auto_name_recorder.schedule(event)
+        else:
             event.stop_event()

astrbot/core/pipeline/waking_check/umo_auto_name.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import asyncio
+from collections import OrderedDict
+from typing import TYPE_CHECKING
+
+from astrbot import logger
+from astrbot.core.umo_alias import get_event_auto_name
+
+if TYPE_CHECKING:
+    from astrbot.core.db import BaseDatabase
+    from astrbot.core.platform.astr_message_event import AstrMessageEvent
+
+MAX_UMO_AUTO_NAME_CACHE_SIZE = 10_000
+
+
+class UmoAutoNameRecorder:
+    """Persist changed UMO names without blocking the waking stage."""
+
+    def __init__(
+        self,
+        db_helper: BaseDatabase | None,
+        config_id: str,
+    ) -> None:
+        """Initialize the bounded cache and background writer state.
+
+        Args:
+            db_helper: Database used to persist automatic names.
+            config_id: Pipeline configuration identifier used in the task name.
+        """
+        self.db_helper = db_helper
+        self.config_id = config_id
+        self._cache: OrderedDict[str, str] = OrderedDict()
+        self._pending: OrderedDict[str, tuple[str, str]] = OrderedDict()
+        self._writer_task: asyncio.Task[None] | None = None
+
+    def schedule(self, event: AstrMessageEvent) -> None:
+        """Queue a changed automatic name from an awakened event.
+
+        Args:
+            event: Awakened event containing the UMO and display metadata.
+        """
+        if self.db_helper is None:
+            return
+
+        umo = event.unified_msg_origin
+        auto_name = get_event_auto_name(event, fallback_to_id=False)
+        if not auto_name:
+            return
+        if self._cache.get(umo) == auto_name:
+            self._cache.move_to_end(umo)
+            return
+
+        self._cache[umo] = auto_name
+        self._cache.move_to_end(umo)
+        if len(self._cache) > MAX_UMO_AUTO_NAME_CACHE_SIZE:
+            self._cache.popitem(last=False)
+
+        self._pending[umo] = (str(event.get_sender_id() or ""), auto_name)
+        self._pending.move_to_end(umo)
+        if len(self._pending) > MAX_UMO_AUTO_NAME_CACHE_SIZE:
+            dropped_umo, (_, dropped_name) = self._pending.popitem(last=False)
+            if self._cache.get(dropped_umo) == dropped_name:
+                self._cache.pop(dropped_umo, None)
+
+        if self._writer_task is None or self._writer_task.done():
+            task = asyncio.create_task(
+                self._flush(),
+                name=f"umo_auto_name_writer:{self.config_id}",
+            )
+            self._writer_task = task
+            task.add_done_callback(self._on_writer_done)
+
+    async def _flush(self) -> None:
+        """Persist queued names sequentially, coalescing changes per UMO."""
+        if self.db_helper is None:
+            return
+
+        try:
+            while self._pending:
+                umo, (creator_sender_id, auto_name) = self._pending.popitem(last=False)
+                try:
+                    await self.db_helper.upsert_umo_auto_name(
+                        umo=umo,
+                        creator_sender_id=creator_sender_id,
+                        auto_name=auto_name,
+                    )
+                except Exception as exc:
+                    logger.warning(
+                        "Failed to persist automatic UMO name for %s: %s",
+                        umo,
+                        exc,
+                    )
+                    if umo not in self._pending and self._cache.get(umo) == auto_name:
+                        self._cache.pop(umo, None)
+        finally:
+            self._writer_task = None
+
+    @staticmethod
+    def _on_writer_done(task: asyncio.Task[None]) -> None:
+        """Expose unexpected writer failures.
+
+        Args:
+            task: Completed automatic-name writer task.
+        """
+        if task.cancelled():
+            return
+        exc = task.exception()
+        if exc is not None:
+            logger.error("UMO automatic-name writer failed.", exc_info=exc)

astrbot/core/platform/sources/discord/discord_platform_adapter.py

@@ -83,14 +83,11 @@ async def send_by_session(
 
         if channel:
             message_obj.type = self._get_message_type(channel)
-            message_obj.group_id = self._get_channel_id(channel)
-            group_name = self._get_group_name(channel)
-            if (
-                message_obj.type == MessageType.GROUP_MESSAGE
-                and message_obj.group
-                and group_name
-            ):
-                message_obj.group.group_name = group_name
+            if message_obj.type == MessageType.GROUP_MESSAGE:
+                message_obj.group_id = self._get_channel_id(channel)
+                group_name = self._get_group_name(channel)
+                if message_obj.group and group_name:
+                    message_obj.group.group_name = group_name
         else:
             logger.warning(
                 f"[Discord] Can't get channel info for {channel_id_str}, will guess message type.",
@@ -253,10 +250,11 @@ def _convert_message_to_abm(self, data: dict) -> AstrBotMessage:
 
         abm = AstrBotMessage()
         abm.type = self._get_message_type(message.channel)
-        abm.group_id = self._get_channel_id(message.channel)
-        group_name = self._get_group_name(message.channel)
-        if abm.type == MessageType.GROUP_MESSAGE and abm.group and group_name:
-            abm.group.group_name = group_name
+        if abm.type == MessageType.GROUP_MESSAGE:
+            abm.group_id = self._get_channel_id(message.channel)
+            group_name = self._get_group_name(message.channel)
+            if abm.group and group_name:
+                abm.group.group_name = group_name
         abm.message_str = content
         abm.sender = MessageMember(
             user_id=str(message.author.id),
@@ -542,18 +540,20 @@ async def dynamic_callback(
             abm = AstrBotMessage()
             if channel is not None:
                 abm.type = self._get_message_type(channel, ctx.guild_id)
-                abm.group_id = self._get_channel_id(channel)
-                group_name = self._get_group_name(channel)
-                if abm.type == MessageType.GROUP_MESSAGE and abm.group and group_name:
-                    abm.group.group_name = group_name
+                if abm.type == MessageType.GROUP_MESSAGE:
+                    abm.group_id = self._get_channel_id(channel)
+                    group_name = self._get_group_name(channel)
+                    if abm.group and group_name:
+                        abm.group.group_name = group_name
             else:
                 # 防守式兜底：channel 取不到时，仍能根据 guild_id/channel_id 推断会话信息
                 abm.type = (
                     MessageType.GROUP_MESSAGE
                     if ctx.guild_id is not None
                     else MessageType.FRIEND_MESSAGE
                 )
-                abm.group_id = str(ctx.channel_id)
+                if abm.type == MessageType.GROUP_MESSAGE:
+                    abm.group_id = str(ctx.channel_id)
 
             abm.message_str = message_str_for_filter
             abm.sender = MessageMember(