update

.env.example

@@ -0,0 +1,5 @@
+TG_BOT_HOST="insert your host here"
+ENCRYPTION_KEY="insert your encryption key here"
+TG_BOT_TOKEN="insert your telegram bot token here"
+ADMIN_USER="insert addnin user to access /update_webhook"
+ADMIN_PASSWORD="insert addnin password to access /update_webhook"

.github/workflows/python-app.yml

@@ -0,0 +1,35 @@
+# This workflow will install Python dependencies, run tests and lint with a single version of Python
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Test
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python 3.10
+      uses: actions/setup-python@v3
+      with:
+        python-version: "3.10"
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        if [ -f requirements-dev.txt ]; then pip install -r requirements-dev.txt; fi
+    - name: Lint with flake8
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics

README.md

@@ -1,2 +1,21 @@
-# notification_bot
-Stateless telegram notification bot
+[![status workflow test](https://github.com/cirebon-dev/notification_bot/actions/workflows/python-app.yml/badge.svg)](https://github.com/cirebon-dev/notification_bot/actions) 
+
+With [@sendh_bot](https://t.me/sendh_bot) you can easy send notification/message to telegram from anywhere (Terminal, CI/CD pipeline, IoT device etc)
+
+![Screenshot](screenshot.jpg)
+
+This service is stateless, so your privacy is highly protected, and you can deploy it in your own instance.
+
+## Run Locally
+
+1. type command `poetry self add poetry-dotenv-plugin`.
+
+2. type command `poetry update`.
+
+3. edit `.env.example` and renamed to `.env`.
+
+4. type command `poetry run python app.py`
+
+5. open browser `https://your-host/update_webhook`
+
+now you can start chatting with your bot :)

app.py

@@ -0,0 +1,80 @@
+# -*-coding:utf8;-*-
+from cryptography.fernet import Fernet
+from bottle import route, request, post, auth_basic
+import json
+import requests
+import bottle
+import telegram
+import os
+"""
+Telegram notification bot with highly privacy/data protection.
+author: guangrei
+"""
+
+HOST = os.environ.get("TG_BOT_HOST")
+ENCRYPTION_KEY = os.environ.get("ENCRYPTION_KEY")
+admin_user = os.environ.get("ADMIN_USER")
+admin_password = os.environ.get("ADMIN_PASSWORD")
+template = """
+now you can start sending messages, for example:
+`curl -X POST https://end-points -d '{"text": "Hello, world 🐣"}'`
+"""
+help_template = """
+to get support please open issues <a href="https://github.com/cirebon-dev/notification_bot">here</a>
+and follow our channel @anak_tkj
+"""
+key = ENCRYPTION_KEY.encode()
+fernet = Fernet(key)
+
+
+def is_authenticated_user(user, password):
+    return user == admin_user and password == admin_password
+
+
+@route('/')
+def root_handler():
+    return "its works!"
+
+
+@route("/update_webhook")
+@auth_basic(is_authenticated_user)
+def update_handler():
+    url = f'https://{HOST}/{telegram.token.replace(":","_")}'
+    return telegram.set_webhook(url)
+
+
+@post('/'+telegram.token.replace(":", "_"))
+def telegram_hook():
+    data = request.json
+    msg = telegram.get_message(data)
+    msg_id = telegram.get_message_id(data)
+    chat_id = telegram.get_chat_id(data)
+    if msg == "/start":
+        token = fernet.encrypt(str(chat_id).encode())
+        msg = f"{HOST}/h/{token.decode()}"
+        msg = template.replace("end-points", msg)
+        telegram.send_message(
+            msg, chat_id, parse_mode="Markdown", disable_web_page_preview=True, reply_to_message_id=msg_id)
+    elif msg == "/help":
+        telegram.send_message(help_template, chat_id, parse_mode="Html",
+                              reply_to_message_id=msg_id)
+
+    return "OK"
+
+
+@post('/h/<token>')
+def notify_handler(token):
+    try:
+        data = request.body.read().decode('utf-8')
+        data = json.loads(data)
+        chat_id = int(fernet.decrypt(token.encode()).decode())
+        return telegram.send_message(data["text"], chat_id)
+    except BaseException as e:
+        return {"ok": False, "ServerError": str(e)}
+
+
+bottle.debug(True)
+app = application = bottle.default_app()
+
+if __name__ == "__main__":
+    bottle.run(host='0.0.0.0', port=80, debug=True)

pyproject.toml

@@ -0,0 +1,21 @@
+[tool.poetry]
+name = "notification  bot"
+version = "0.1.0"
+description = ""
+authors = ["Guangrei"]
+license = "mit"
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "3.10"
+requests = "^2.31.0"
+bottle = "^0.12.25"
+cryptography = "^41.0.7"
+
+[tool.poetry.group.dev.dependencies]
+autopep8 = "^2.0.4"
+flake8 = "^6.1.0"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

requirements.txt

@@ -0,0 +1,3 @@
+requests
+bottle
+cryptography

telegram.py

@@ -0,0 +1,37 @@
+# -*-coding:utf8;-*-
+import os
+import requests
+
+
+token = os.environ.get("TG_BOT_TOKEN")
+BOT_URL = f'https://api.telegram.org/bot{token}/'
+
+
+def get_chat_id(data):
+    chat_id = data['message']['chat']['id']
+    return chat_id
+
+def get_message(data):
+    message_text = data['message']['text']
+    return message_text
+
+def get_message_id(data):
+    message_id = data['message']['message_id']
+    return message_id
+
+def send_message(text, chat_id, parse_mode=False, disable_web_page_preview=False, disable_notification=False, reply_to_message_id=False, reply_markup=False):
+    message_url = BOT_URL + 'sendMessage'
+    json_data = {"chat_id": chat_id, "text": text, "disable_web_page_preview":
+                 disable_web_page_preview, "disable_notification": disable_notification}
+    if parse_mode:
+        json_data['parse_mode'] = parse_mode
+    if reply_to_message_id:
+        json_data["reply_to_message_id"] = reply_to_message_id
+    if reply_markup:
+        json_data["reply_markup"] = reply_markup
+    return requests.post(message_url, json=json_data).json()
+
+
+def set_webhook(uri):
+    api = BOT_URL + 'setWebHook?url='+uri
+    return requests.get(api).json()

vercel.json

@@ -0,0 +1,10 @@
+{
+  "version": 2,
+  "public": false,
+  "builds": [{ "src": "app.py", "use": "@vercel/python" }],
+  "routes": [
+    { "src": "/", "dest": "app.py" },
+    { "src": "/(.*)", "dest": "app.py" }
+
+  ]
+}