From 7d4f8da59d60156c2d45dc4a44891eea48082315 Mon Sep 17 00:00:00 2001
From: Evgeny Zinoviev <me@ch1p.io>
Date: Sun, 25 Apr 2021 23:02:49 +0300
Subject: initial

---
 ch1p/__init__.py  |  2 ++
 ch1p/functions.py | 34 ++++++++++++++++++++++++++++++++++
 ch1p/state.py     | 28 ++++++++++++++++++++++++++++
 3 files changed, 64 insertions(+)
 create mode 100644 ch1p/__init__.py
 create mode 100644 ch1p/functions.py
 create mode 100644 ch1p/state.py

(limited to 'ch1p')

diff --git a/ch1p/__init__.py b/ch1p/__init__.py
new file mode 100644
index 0000000..878f8fa
--- /dev/null
+++ b/ch1p/__init__.py
@@ -0,0 +1,2 @@
+from .functions import telegram_notify
+from .state import State
\ No newline at end of file
diff --git a/ch1p/functions.py b/ch1p/functions.py
new file mode 100644
index 0000000..c4f9c99
--- /dev/null
+++ b/ch1p/functions.py
@@ -0,0 +1,34 @@
+import os, requests
+
+from typing import List, Tuple, AnyStr
+
+
+def _get_vars(params: List[Tuple], kw: dict) -> List[AnyStr]:
+    result = []
+
+    for kw_name, env_name in params:
+        env_name = f'MY_{env_name}'
+        if kw_name in kw:
+            result.append(kw[kw_name])
+        elif env_name in os.environ:
+            result.append(os.environ[env_name])
+        else:
+            raise RuntimeError("missing parameter %s or variable %s" % (kw_name, env_name))
+
+    return result
+
+
+def telegram_notify(text: str, parse_mode: str = 'html', **kwargs):
+    token, chat_id = _get_vars([
+        ('chat_id', 'TELEGRAM_NOTIFY_CHAT_ID'),
+        ('token', 'TELEGRAM_NOTIFY_TOKEN')
+    ], kwargs)
+
+    r = requests.post('https://api.telegram.org/bot%s/sendMessage' % token, data={
+        'chat_id': chat_id,
+        'text': text,
+        'parse_mode': parse_mode
+    })
+
+    if r.status_code != 200:
+        raise RuntimeError("telegram returned %d" % r.status_code)
diff --git a/ch1p/state.py b/ch1p/state.py
new file mode 100644
index 0000000..f677f59
--- /dev/null
+++ b/ch1p/state.py
@@ -0,0 +1,28 @@
+import os, json
+
+from functions import _get_vars
+
+
+class State:
+    def __init__(self, default=None, **kwargs):
+        file, = _get_vars([
+            ('file', 'STATE_FILE')
+        ], kwargs)
+
+        if default is None:
+            default = {}
+
+        self.file = file
+        self.default = default
+
+    def read(self) -> dict:
+        if not os.path.exists(self.file):
+            self.write(self.default)
+            return self.default
+
+        with open(self.file, 'r') as f:
+            return json.loads(f.read())
+
+    def write(self, state: dict):
+        with open(self.file, 'w') as f:
+            f.write(json.dumps(state))
\ No newline at end of file
-- 
cgit v1.2.3