Class RoverSubscribe

RoverSign/utils/database/rover_subscribe.py:18–145

群组Bot记录表 自动记录每个群最后使用的bot_self_id 当检测到bot变化时，自动更新该群所有订阅的bot_id

class RoverSubscribe(BaseModel, table=True):
    """群组Bot记录表

    自动记录每个群最后使用的bot_self_id
    当检测到bot变化时，自动更新该群所有订阅的bot_id
    """

    __tablename__ = "RoverSubscribe"
    __table_args__: Dict[str, Any] = {"extend_existing": True}

    group_id: str = Field(default="", title="群组ID", unique=True)
    bot_self_id: str = Field(default="", title="BotSelfID")
    updated_at: Optional[int] = Field(default=None, title="最后更新时间")

    @classmethod
    @with_lock
    @with_session
    async def check_and_update_bot(
        cls: Type[T_RoverSubscribe],
        session: AsyncSession,
        group_id: str,
        bot_id: str,
        bot_self_id: str,
    ) -> bool:
        """检查并更新群组的bot_self_id

        如果bot_self_id发生变化，自动更新该群所有订阅的bot_self_id

        Args:
            group_id: 群组ID
            bot_id: 平台 bot_id（onebot / qqgroup / ...）
            bot_self_id: 新的bot_self_id

        Returns:
            bool: 是否发生了bot变更
        """
        import time
        from gsuid_core.logger import logger

        current_time = int(time.time())

        logger.debug(
            f"[库洛签到·订阅] check_and_update_bot 被调用: group_id={group_id}, bot_id={bot_id}, bot_self_id={bot_self_id}"
        )

        # 查询现有记录
        sql = select(cls).where(cls.group_id == group_id)
        result = await session.execute(sql)
        existing = result.scalars().first()

        if existing:
            logger.debug(
                f"[库洛签到·订阅] 找到现有记录: group_id={group_id}, existing.bot_self_id={existing.bot_self_id}, new_bot_self_id={bot_self_id}"
            )
            # 检查bot是否变化
            if existing.bot_self_id != bot_self_id:
                old_bot_self_id = existing.bot_self_id
                logger.info(