minimal design

Makefile

@@ -1,21 +1,2 @@
-env := .venv
-deps := requirements.txt
-
 run:
-	@clear;./$(env)/bin/python start.py
-
-build:
-	@./$(env)/bin/pip install build;./$(env)/bin/python -m build
-
-clean:
-	@rm -rf dist build *.egg-info **/__pycache__/
-
-stable: clean build
-	git push;git checkout releases;git merge main;git push;twine upload dist/*;git checkout main;
-
-$(env): $(deps)
-	python -m venv $@
-
-install: $(env)
-	@./$(env)/bin/pip install -r $(deps)
-
+	.venv\Scripts\python start.py

Makefile.unix

@@ -0,0 +1,21 @@
+env := .venv
+deps := requirements.txt
+
+run:
+	@clear;./$(env)/bin/python start.py
+
+build:
+	@./$(env)/bin/pip install build;./$(env)/bin/python -m build
+
+clean:
+	@rm -rf dist build *.egg-info **/__pycache__/
+
+stable: clean build
+	git push;git checkout releases;git merge main;git push;twine upload dist/*;git checkout main;
+
+$(env): $(deps)
+	python -m venv $@
+
+install: $(env)
+	@./$(env)/bin/pip install -r $(deps)
+

autogram/app.py

@@ -1,74 +1,18 @@
-import os
-import time
 import queue
 from autogram.base import Bot
-from requests.exceptions import ConnectionError
-
 
+# --
 class Autogram(Bot):
   def __init__(self, config) -> None:
     """Initialize parent object"""
-    self.update_handler = None
-    super().__init__(config)
-    return
-
-  def addHandler(self, function):
-    self.update_handler = function
-    return function
-
-  def prepare(self):
-    """Confirm auth through getMe(), then check update methods"""
-    res = self.getMe()
-    if not res.ok:
-      self.do_err(msg=str(res.json()))
-    self.webhook_addr = self.config.get('AUTOGRAM_ENDPOINT') or os.getenv('AUTOGRAM_ENDPOINT')  # noqa: E501
-    if self.webhook_addr:
-      res = self.setWebhook(self.webhook_addr)
-      if not res.ok:
-        self.do_err(msg='/setWebhook failed!')
-    else:
-      res = self.deleteWebhook()
-      if not res.ok:
-        self.do_err('/deleteWebhook failed!')
-      else:
-        self.short_poll()
-    return
+    return super().__init__(config)
 
   def start(self):
     """Launch the bot"""
-    try:
-      self.prepare()
-      while not self.terminate.is_set():
-        try:
-          if not self.update_handler:
-            time.sleep(5)
-            continue
-          self.update_handler(self.updates.get())
-        except queue.Empty:
-          continue
-    except ConnectionError:
-      self.terminate.set()
-      self.logger.critical('Connection Error!')
-    finally:
-      self.shutdown()
+    if (bot := self.getMe()).ok:
+      print(self.poll().json())
+
 
   def shutdown(self):
     """Gracefully terminate the bot"""
-    if self.terminate.is_set():
-      try:
-        res = self.getWebhookInfo()
-        if not res.ok:
-          return
-        if not res.json()['result']['url']:
-          return
-      except Exception:
-        return
-    # delete webhook and exit
-    try:
-      res = self.deleteWebhook()
-      if not res.ok:
-        raise RuntimeError()
-    except Exception:
-      self.logger.critical('/deleteWebhook failed!')
-    finally:
-      self.terminate.set()
+    self.terminate.set()

autogram/base.py

@@ -1,38 +1,29 @@
 import os
 import re
-import time
 import json
-import loguru
-import threading
 import requests
 from typing import Any
-from queue import Queue
+from loguru import logger
+from . import chat_actions
 from requests.models import Response
-from autogram.webserver import WebServer
-from bottle import request, response, post, run, get
 from autogram.config import save_config
-from . import chat_actions
-
 
+# --
 class Bot():
   endpoint = 'https://api.telegram.org/'
 
   def __init__(self, config :dict) -> None:
     """Initialize parent database object"""
     super().__init__()
-    self.updates = Queue()
-    self.logger = loguru.logger
-    self.terminate = threading.Event()
     self.requests = requests.session()
     self.config = config or self.do_err(msg='Please pass a config <object :Dict>!')
     if not self.config.get("telegram-token"):
       self.config.update({
           "telegram-token" : os.getenv('AUTOGRAM_TG_TOKEN') or self.do_err(msg='Missing bot token!') # noqa: E501
         })
 
-  def do_err(self, err_type =RuntimeError, msg ='Error!'):
+  def do_err(self, msg :str, err_type =RuntimeError):
     """Clean terminate the program on errors."""
-    self.terminate.set()
     raise err_type(msg)
 
   def settings(self, key :str, val: Any|None=None):
@@ -56,87 +47,24 @@ def media_quality(self):
   def setWebhook(self, hook_addr : str):
     if not re.search('^(https?):\\/\\/[^\\s/$.?#].[^\\s]*$', hook_addr):
       raise RuntimeError('Invalid webhook url. format <https://...>')
-    # ensure hook_addr stays reachable
-    @get('/')
-    def ping():
-      return json.dumps({'ok': True})
-    # keep-alive service
-    def keep_alive():
-      self.logger.info('Keep-alive started.')
-      while not self.terminate.is_set():
-        try:
-          res = self.requests.get(hook_addr)
-          if not res.ok:
-            self.logger.debug('Ngrok tunnel disconnected!')
-        except Exception:
-          self.logger.debug('Connection error.')
-        time.sleep(3)
-    # start keep-alive
-    alive_guard = threading.Thread(target=keep_alive)
-    alive_guard.name = 'Autogram:Keep-alive'
-    alive_guard.daemon = True
-    alive_guard.start()
-    # receive updates
-    @post('/')
-    def hookHandler():
-      response.content_type = 'application/json'
-      self.updates.put(request.json)
-      return json.dumps({'ok': True})
-    #
-    def runServer(server: Any):
-      return run(server=server, quiet=True)
-    #
-    server = WebServer(host="0.0.0.0", port=self.settings('lport'))
-    serv_thread = threading.Thread(target=runServer, args=(server,))
-    serv_thread.name = 'Autogram:Bottle'
-    serv_thread.daemon = True
-    serv_thread.start()
-    # inform telegram
+    #--
     url = f'{self.endpoint}bot{self.settings("telegram-token")}/setWebhook'
     params = {
       'url': hook_addr
     }
     return self.requests.get(url, params=params)
 
-  def short_poll(self):
-    """Start fetching updates in seperate thread"""
-    def getter():
-      failed = False
-      offset = 0
-      while not self.terminate.is_set():
-        try:
-          data = {
-            'timeout': 3,
-            'params': {
-              'offset': offset,
-              'limit': 10,
-              'timeout': 1
-            }
-          }
-          res = self.getUpdates(**data)
-        except Exception:
-          time.sleep(2)
-          continue
-        #
-        if not res.ok:
-          if not failed:
-            time.sleep(2)
-            failed = True
-          else:
-            self.terminate.set()
-        else:
-          updates = res.json()['result']
-          for update in updates:
-            offset = update['update_id'] + 1
-            self.updates.put(update)
-          # rate-limit
-          poll_interval = 2
-          time.sleep(poll_interval)
-      return
-    poll = threading.Thread(target=getter)
-    poll.name = 'Autogram:short_polling'
-    poll.daemon = True
-    poll.start()
+  def poll(self, offset=0, limit=10, timeout=7):
+    """Poll updates"""
+    data = {
+      'timeout': timeout,
+      'params': {
+        'offset': offset,
+        'limit': limit,
+        'timeout': timeout // 2
+      }
+    }
+    return self.getUpdates(**data)
 
   def getMe(self) -> Response:
     """Fetch `bot` information"""

autogram/webserver.py → autogram/bottle-rest.py

@@ -27,3 +27,9 @@ class server_cls(server_cls):
 
   def shutdown(self):
     self.srv.shutdown()
+
+# server = WebServer(host="0.0.0.0", port=self.settings('lport'))
+# serv_thread = threading.Thread(target=runServer, args=(server,))
+# serv_thread.name = 'Autogram:Bottle'
+# serv_thread.daemon = True
+# serv_thread.start()

autogram/config.py

@@ -2,7 +2,7 @@
 import sys
 import json
 from loguru import logger
-from typing import Callable, Dict
+from typing import Callable
 
 default_config = {
   'lport': 4004,
@@ -26,7 +26,7 @@ def load_config(config_file : str, config_path : str):
     config |= json.load(conf)
   return config
 
-def save_config(config :Dict):
+def save_config(config :dict):
   """config-file must be in the dictionary"""
   try:
     conffile = config.pop('config-file')
@@ -50,4 +50,8 @@ def wrapper(func: Callable):
   return wrapper
 #
 
-__all__ = [ "Start", "save_config", "load_config"]
+__all__ = [
+  "Start",
+  "save_config",
+  "load_config"
+]

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "autogram"
-version = "3.4.5"
+version = "3.5.0"
 description = "An easily extensible telegram API wrapper"
 readme = "README.md"
 authors = [{ name = "droi9", email = "[email protected]" }]
@@ -18,10 +18,8 @@ classifiers = [
 ]
 keywords = ["telegram", "API", "wrapper"]
 dependencies = [
-	"SQLAlchemy==2.0.19",
-	"loguru==0.7.0",
-	"bottle==0.12.25",
-	"requests==2.31.0",
+	"loguru==0.7.2",
+	"requests==2.32.3",
 ]
 requires-python = ">=3.6"
 

requirements.txt

@@ -1,4 +1,2 @@
-arrow>=1.3.0
 loguru>=0.7.2
-requests>=2.32.2
-python-dotenv>=1.0.1
+requests>=2.32.3

start.py

@@ -1,49 +1,8 @@
-"""
-@author: drui9
-@config values:
-    - update-endpoint: optional
-    - ngrok-token: optional
-    - bot-token: required
-    - cava-auth: optional
-"""
-import os
-import requests
-from unittest import mock
 from autogram import Autogram
-from dotenv import load_dotenv
-from autogram.config import load_config
-
-def get_update_endpoint(validator_fn, inject_key=None):
-    """Get updates endpoint, silently default to telegram servers"""
-    if key := os.getenv('ngrok-api-key', inject_key):
-        header = {'Authorization': f'Bearer {key}', 'Ngrok-Version': '2'}
-        try:
-            def getter():
-                if os.getenv('TESTING') == '1':
-                    return 'http://localhost:8000'
-                rep = requests.get('https://api.ngrok.com/tunnels', headers=header, timeout=6)
-                if rep.ok and (out := validator_fn(rep.json())):
-                    return out
-            return getter
-        except Exception:
-            raise
-    return Autogram.api_endpoint
-
-# modify to select one ngrok tunnel from list of tunnels
-def select_tunnel(tunnels):
-    for tunnel in tunnels['tunnels']:
-        if tunnel['forwards_to'] == 'http://api:8000':
-            return tunnel['public_url']
-
-# launcher
-if __name__ == '__main__':
-    load_dotenv()
-    config = Autogram.cfg_template()
-    with load_config(config):
-        if ngrok_token := config.get('ngrok-token'):
-            if getter := get_update_endpoint(select_tunnel, ngrok_token):
-                config['update-endpoint'] = getter()
-                bot = Autogram(config)
-                bot.getter = getter # fn to get updated endpoint
-                bot.loop()
+from autogram.config import load_config, Start
 
+#--
+@Start(config_file='web-auto.json')
+def main(config):
+    bot = Autogram(config)
+    bot.start()