First commit

.gitignore

@@ -0,0 +1,5 @@
+.venv
+.vscode
+__pycache__
+*.session
+*.env

main.py

@@ -0,0 +1,77 @@
+#############################################
+#
+#   1. Check Permissions.
+#   2. IP address validation.
+#   3. Notify detect change.add()
+#   4. Add disable handler  
+#############################################
+
+from os import getenv
+import re
+from telethon import TelegramClient, events, custom
+from telethon.tl.custom.button import Button
+from telethon.tl.custom.message import Message
+from datetime import datetime, timedelta, time
+import asyncio
+from pinger import Pinger
+from dotenv import load_dotenv
+
+
+IP_ADDRESS_REGEX = '^((25[0-5]|(2[0-4]|1\d|[1-9]|)\d)(\.(?!$)|$)){4}$'
+
+
+def main():
+    load_dotenv()
+    pinger = Pinger()
+    
+    client = TelegramClient("pinger", getenv("API_ID"), getenv("API_HASH"))
+    
+    
+    async def send_telegram_message(result):
+        # print(result)
+        await client.send_message(383724428, str(result))
+    
+    async def check_pings():
+        await pinger.start(send_telegram_message)
+
+    @client.on(events.NewMessage(pattern="/start"))
+    async def start(event:Message):
+        if event.is_private:
+            await event.respond("שלום")
+
+    # @client.on(events.NewMessage(pattern="/help"))
+    # async def help(event:Message):
+    #     asyncio.create_task(help_message(event))
+
+    @client.on(events.NewMessage(pattern="/enable"))
+    async def enable(event:Message):
+        if event.is_private:
+            asyncio.create_task(check_pings())
+            await event.reply("Enable")
+
+    @client.on(events.NewMessage(pattern="/disable"))
+    async def disable(event:Message):
+        if event.is_private:
+            await pinger.stop()
+            await event.reply("Stopping...")
+
+    @client.on(events.NewMessage(pattern="/add_ip"))
+    async def add_ip(event:Message):
+        if event.is_private:
+            ip = event.raw_text.split(" ")[1]
+            if re.search(ip):
+                await pinger.add_new_ip(event.raw_text.split(" ")[1])
+
+    @client.on(events.NewMessage(pattern="/del_ip"))
+    async def del_ip(event:Message):
+        if event.is_private:
+            await pinger.del_ip(event.raw_text.split(" ")[1])
+
+    client.start(bot_token=f'{getenv("BOT_TOKEN")}')
+    print("Client was started!")
+    client.loop.create_task(check_pings())
+    client.run_until_disconnected()
+
+
+if __name__ == "__main__":
+    main()

pinger.py

@@ -0,0 +1,78 @@
+import asyncio
+
+class Pinger():
+    def __init__(self):
+        self.enabled = False
+        self.ips_list = []
+        self.ips_running_list = []
+
+    async def add_new_ip(self, ip):
+        self.ips_list.append(ip)
+        
+        
+    async def del_ip(self, ip):
+        self.ips_list.remove(ip)
+        
+    async def start(self, fn):
+        self.enabled = True
+        asyncio.create_task(self.ping_all(fn))
+        
+    async def stop(self):
+        self.enabled = False
+        while len(self.ips_running_list) > 0:
+            await asyncio.sleep(0)
+        
+    async def notify_ping(self, ip, fn):
+        success = False
+        cmd = f"ping -c 1 -W 5 -q {ip}"
+        proc = await asyncio.create_subprocess_shell(
+            cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE)
+
+        stdout, stderr = await proc.communicate()
+
+        # print(f'[{cmd!r} exited with {proc.returncode}]')
+        if stdout:
+            output = f'[stdout]\n{stdout.decode()}'
+            success = bool(int((output.split(",")[1].strip()[0])))
+            
+        # if stderr:
+        #     print(f'[stderr]\n{stderr.decode()}')
+        await fn(str({"ip" : ip, "success": success}))
+        return ip
+    
+
+    async def ping_all(self, fn):
+
+        def on_done(task_ip: asyncio.Task):
+            ip = task_ip.result()
+            self.ips_running_list.remove(ip)
+        
+        while self.enabled:
+            for ip in self.ips_list:
+                if ip not in self.ips_running_list:
+                    task = asyncio.create_task(self.notify_ping(ip, fn))
+                    self.ips_running_list.append(ip)
+                    task.add_done_callback(on_done)
+                else:     
+                    await asyncio.sleep(0)
+            await asyncio.sleep(0)
+            
+            
+async def output_print(result):
+    print(result)
+            
+            
+async def main():
+    p = Pinger()
+    await p.add_new_ip("192.168.14.121")
+    await p.add_new_ip("8.8.8.8")
+    await p.start(output_print)
+    await asyncio.sleep(20)
+    await p.stop()
+    print("Finished!")
+    
+
+if __name__ == "__main__":
+    asyncio.run(main())

text.py

@@ -0,0 +1,55 @@
+import asyncio
+import time
+
+async def say_after(delay, what):
+    print(1)
+    await asyncio.sleep(delay)
+    print(what)
+    
+async def say_after2(delay, what):
+    print(2)
+    await asyncio.sleep(delay)
+    print(what)
+
+async def say_after3(delay, what):
+    print(3)
+    await asyncio.sleep(delay)
+    print(what)
+    
+
+async def main():
+    l = []
+    
+    await say_after3(3, "delay1")
+    
+    task1 = asyncio.create_task(
+        say_after(1, 'hello'))
+
+    task2 = asyncio.create_task(
+        say_after2(2, 'world'))
+    
+    l.append(task1)
+    l.append(task2)
+    
+    task1.add_done_callback(l.remove)
+    task2.add_done_callback(l.remove)
+    
+    
+    await say_after3(3, "delay2")
+
+    print(f"started at {time.strftime('%X')}")
+
+    # Wait until both tasks are completed (should take
+    # around 2 seconds.)
+    # await task1   
+    # await task2
+    
+    while len(l) != 0:
+        print("wait...")
+        await asyncio.sleep(1)
+
+    print(f"finished at {time.strftime('%X')}")
+    
+    
+if __name__ == "__main__":
+    asyncio.run(main())