读取 XutheringWavesUID 的 WavesSubscribe 表（只读） 与 XutheringWavesUID 共用同一张表，避免直接依赖
| 146 | |
| 147 | |
| 148 | class WavesSubscribeReader(BaseModel, table=True): |
| 149 | """读取 XutheringWavesUID 的 WavesSubscribe 表（只读） |
| 150 | |
| 151 | 与 XutheringWavesUID 共用同一张表，避免直接依赖 |
| 152 | """ |
| 153 | |
| 154 | __tablename__ = "WavesSubscribe" |
| 155 | __table_args__: Dict[str, Any] = {"extend_existing": True} |
| 156 | |
| 157 | group_id: str = Field(default="", title="群组ID", unique=True) |
| 158 | bot_self_id: str = Field(default="", title="BotSelfID") |
| 159 | updated_at: Optional[int] = Field(default=None, title="最后更新时间") |
| 160 | |
| 161 | @classmethod |
| 162 | @with_session |
| 163 | async def get_group_bot( |
| 164 | cls: Type[T_WavesSubscribeReader], |
| 165 | session: AsyncSession, |
| 166 | group_id: str, |
| 167 | ) -> Optional[str]: |
| 168 | """获取群组当前的bot_self_id""" |
| 169 | sql = select(cls).where(cls.group_id == group_id) |
| 170 | result = await session.execute(sql) |
| 171 | record = result.scalars().first() |
| 172 | return record.bot_self_id if record else None |
nothing calls this directly
no outgoing calls
no test coverage detected