setup

create broadcast function

program/function/broadcast.py

@@ -0,0 +1,94 @@
+import os
+import time
+import random
+import string
+import aiofiles
+import asyncio
+import datetime
+import traceback
+
+from pyrogram.errors import (
+    FloodWait,
+    InputUserDeactivated,
+    PeerIdInvalid,
+    UserIsBlocked,
+)
+from config import STREAM_COPY
+
+
+stream_id = {}
+
+
+async def send_msg(user_id, message):
+    try:
+        if STREAM_COPY is False:
+            await message.forward(chat_id=user_id)
+        elif STREAM_COPY is True:
+            await message.copy(chat_id=user_id)
+        return 200, None
+    except FloodWait as e:
+        await asyncio.sleep(e.x)
+        return send_msg(user_id, message)
+    except InputUserDeactivated:
+        return 400, f"{user_id} : deactivated\n"
+    except UserIsBlocked:
+        return 400, f"{user_id} : blocked the bot\n"
+    except PeerIdInvalid:
+        return 400, f"{user_id} : user id invalid\n"
+    except Exception:
+        return 500, f"{user_id} : {traceback.format_exc()}\n"
+
+
+async def broadcast(m, db):
+    all_users = await db.get_user()
+    broadcast_msg = m.reply_to_message
+    while True:
+        broadcast_id = "".join([random.choice(string.ascii_letters) for i in range(3)])
+        if not stream_id.get(broadcast_id):
+            break
+    out = await m.reply_text(
+        text=f"Broadcast Started! You will be notified with log file when all the users are notified."
+    )
+    start_time = time.time()
+    total_users = await db.total_count()
+    done = 0
+    failed = 0
+    success = 0
+    stream_id[broadcast_id] = dict(
+        total=total_users, current=done, failed=failed, success=success
+    )
+    async with aiofiles.open("broadcast.txt", "w") as broadcast_log_file:
+        async for user in all_users:
+            sts, msg = await send_msg(user_id=int(user["id"]), message=broadcast_msg)
+            if msg is not None:
+                await broadcast_log_file.write(msg)
+            if sts == 200:
+                success += 1
+            else:
+                failed += 1
+            if sts == 400:
+                await db.delete_user(user["id"])
+            done += 1
+            if stream_id.get(broadcast_id) is None:
+                break
+            else:
+                stream_id[broadcast_id].update(
+                    dict(current=done, failed=failed, success=success)
+                )
+    if stream_id.get(broadcast_id):
+        stream_id.pop(broadcast_id)
+    completed_in = datetime.timedelta(seconds=int(time.time() - start_time))
+    await asyncio.sleep(3)
+    await out.delete()
+    if failed == 0:
+        await m.reply_text(
+            text=f"broadcast completed in `{completed_in}`\n\nTotal users {total_users}.\nTotal done {done}, {success} success and {failed} failed.",
+            quote=True,
+        )
+    else:
+        await m.reply_document(
+            document="broadcast.txt",
+            caption=f"broadcast completed in `{completed_in}`\n\nTotal users {total_users}.\nTotal done {done}, {success} success and {failed} failed.",
+            quote=True,
+        )
+    os.remove("broadcast.txt")