(
self,
chat_id: Union[int, str],
limit: int,
offset: int = 0,
filters: dict = None,
max_size: int = None,
)
| 201 | # Ask Doubt on telegram @KingVJ01 |
| 202 | |
| 203 | async def iter_messages( |
| 204 | self, |
| 205 | chat_id: Union[int, str], |
| 206 | limit: int, |
| 207 | offset: int = 0, |
| 208 | filters: dict = None, |
| 209 | max_size: int = None, |
| 210 | ) -> Optional[AsyncGenerator["types.Message", None]]: |
| 211 | current = offset |
| 212 | dup_files = [] |
| 213 | while True: |
| 214 | new_diff = min(200, limit - current) |
| 215 | if new_diff <= 0: |
| 216 | return |
| 217 | |
| 218 | messages = await self.get_messages(chat_id, list(range(current, current + new_diff + 1))) |
| 219 | for message in messages: |
| 220 | if any(getattr(message, media_type, False) for media_type in filters): |
| 221 | yield "FILTERED" |
| 222 | else: |
| 223 | yield message |
| 224 | |
| 225 | current += 1 |
| 226 | |
| 227 | # Don't Remove Credit Tg - @VJ_Botz |
| 228 | # Subscribe YouTube Channel For Amazing Bot https://youtube.com/@Tech_VJ |
no outgoing calls
no test coverage detected