Добавил поддержку MeshCore через HA

.github/workflows/github-push-docker.yml

@@ -11,6 +11,13 @@ env:
 jobs:
   build-and-push-image:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        image:
+          - suffix: ''
+            start_file: start_telegram.py
+          - suffix: '-meshcore-ha'
+            start_file: start_meshcore_ha.py
     permissions:
       contents: read
       packages: write
@@ -19,17 +26,6 @@ jobs:
       - name: Checkout repository
         uses: actions/checkout@v3
 
-      - name: Set up Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: '3.11'
-
-      - name: Install dependencies and run tests
-        run: |
-          python -m pip install -r requirements.txt
-          python -m py_compile parser.py start_telegram.py
-          python -m unittest discover -s tests -v
-
       - name: Log in to the Container registry
         uses: docker/login-action@f054a8b539a109f9f41c372932f1ae047eff08c9
         with:
@@ -41,12 +37,14 @@ jobs:
         id: meta
         uses: docker/metadata-action@98669ae865ea3cffbcbaa878cf57c20bbf1c6c38
         with:
-          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+          images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}${{ matrix.image.suffix }}
 
       - name: Build and push Docker image
         uses: docker/build-push-action@ad44023a93711e3deb337508980b4b5e9bcdc5dc
         with:
           context: .
+          build-args: |
+            START_FILE=${{ matrix.image.start_file }}
           push: true
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}

Dockerfile

@@ -1,8 +1,9 @@
-FROM python:3.11.2-alpine3.17
+FROM python:3.12-alpine
 
 LABEL Maintainer="serega404"
 
 WORKDIR /app
+ARG START_FILE=start_telegram.py
 
 COPY requirements.txt requirements.txt
 RUN pip3 install -r requirements.txt
@@ -11,7 +12,7 @@ RUN pip3 install -r requirements.txt
 COPY crontab /tmp/crontab
 RUN cat /tmp/crontab > /etc/crontabs/root
 
-COPY start_telegram.py start_telegram.py
+COPY ${START_FILE} start.py
 COPY parser.py parser.py
 
 # run crond as main process of container

README.md

@@ -2,7 +2,7 @@
 
 [![MIT License](https://img.shields.io/github/license/serega404/VodokanalBot)](https://github.com/serega404/VodokanalBot)
 
-## Запуск в Docker
+## Запуск TG интеграции в Docker
 
 ``` Docker
 docker volume create vodokanal_bot_data
@@ -15,7 +15,7 @@ docker run -d --name VodokanalBot \
     ghcr.io/serega404/vodokanalbot:main
 ```
 
-## Запуск в Docker Compose
+### Запуск в Docker Compose
 
 Укажи `TELEGRAM_TOKEN` и `TELEGRAM_CHANNEL` в [`docker-compose.yml`](./docker-compose.yml), затем запусти:
 
@@ -23,18 +23,35 @@ docker run -d --name VodokanalBot \
 docker compose up -d --build
 ```
 
+## Запуск с MeshCore + Home Assistant
+
+Вариант для MeshCore отправляет новые сообщения в webhook Home Assistant, а Home Assistant пересылает их в канал MeshCore.
+
+1. Добавь в Home Assistant автоматизацию [`Webhook to MeshCore channel`](https://github.com/serega404/HA-MeshCore#webhook-to-meshcore-channel).
+2. Замени `CHANGE_ME_RANDOM_WEBHOOK_ID` в автоматизации на случайную строку.
+
+``` Docker
+docker volume create vodokanal_bot_data
+docker run -d --name VodokanalBot \
+    --restart=always \
+    -v vodokanal_bot_data:/app/data \
+    -e TZ='Europe/Moscow' \
+    -e HOME_ASSISTANT_WEBHOOK_ID='CHANGE_ME_RANDOM_WEBHOOK_ID' \
+    -e HOME_ASSISTANT_URL='http://homeassistant.local:8123' \
+    -e HOME_ASSISTANT_WEBHOOK_CHANNEL='0' \
+    ghcr.io/serega404/vodokanalbot-meshcore-ha:main
+```
+
+### Запуск в Docker Compose
+
+Укажи `HOME_ASSISTANT_WEBHOOK_ID` и `HOME_ASSISTANT_WEBHOOK_CHANNEL` в [`docker-compose.ha.yml`](./docker-compose.ha.yml), затем запусти:
+
+``` Docker
+docker compose up -d --build -f docker-compose.ha.yml
+```
+
 ## Интеграции
 
-Бот проверяет сообщения каждые 10 минут. Параллельные запуски защищены
-файловой блокировкой: если предыдущий запуск ещё работает, очередная проверка
-пропускается, не помечая сообщения обработанными. Следующий успешный запуск
-снова сравнит сообщения за текущий день с базой.
-
-Один запуск ограничен 9 минутами. Запрос к сайту водоканала имеет таймаут
-10 секунд на подключение и 60 секунд ожидания данных, запросы к Telegram —
-60 секунд. База `data/db.json` заменяется атомарно, чтобы прерывание записи
-не повредило предыдущие данные.
-
 Общая логика парсинга, работы с `data/db.json` и поиска новых сообщений вынесена в [`parser.py`](./parser.py).
 
 Для новой интеграции достаточно создать свой адаптер отправки и передать его в `publish_new_posts`:

crontab

@@ -1,2 +1,2 @@
-@reboot cd /app && python3 /app/start_telegram.py 
+@reboot cd /app && python3 /app/start.py
-*/10 * * * *  cd /app && python3 /app/start_telegram.py 
+*/10 * * * * cd /app && python3 /app/start.py

docker-compose.ha.yml

@@ -0,0 +1,22 @@
+services:
+  vodokanalbot-ha:
+    image: ghcr.io/serega404/vodokanalbot-meshcore-ha:main
+    build:
+      context: .
+      args:
+        START_FILE: start_meshcore_ha.py
+    container_name: VodokanalBotMeshCoreHA
+    restart: always
+    environment:
+      TZ: Europe/Moscow
+      HOME_ASSISTANT_URL: https://your-home-assistant-url
+      HOME_ASSISTANT_WEBHOOK_ID: CHANGE_ME_RANDOM_WEBHOOK_ID
+      HOME_ASSISTANT_WEBHOOK_CHANNEL: "0"
+      # VODOKANAL_URL: http://www.tgnvoda.ru/avarii.php
+      # PROXY_URL: socks5h://user:password@proxy-host:1080
+      # PROXY_URL: http://user:password@proxy-host:3128
+    volumes:
+      - vodokanal_bot_data:/app/data
+
+volumes:
+  vodokanal_bot_data:

docker-compose.yml

@@ -1,12 +1,17 @@
 services:
   vodokanalbot:
     image: ghcr.io/serega404/vodokanalbot:main
+    build:
+      context: .
+      args:
+        START_FILE: start_telegram.py
     container_name: VodokanalBot
     restart: always
     environment:
       TZ: Europe/Moscow
       TELEGRAM_TOKEN: TOKEN
       TELEGRAM_CHANNEL: CHAT_ID
+      # VODOKANAL_URL: http://www.tgnvoda.ru/avarii.php
       # PROXY_URL: socks5h://user:password@proxy-host:1080
       # PROXY_URL: http://user:password@proxy-host:3128
     volumes:

parser.py

@@ -8,8 +8,7 @@ from bs4 import BeautifulSoup
 
 
 DEFAULT_DB_PATH = "data/db.json"
-SOURCE_TIMEOUT = (10, 60)
+DEFAULT_VODOKANAL_URL = "http://www.tgnvoda.ru/avarii.php"
-
 
 @dataclass(frozen=True)
 class Post:
@@ -32,7 +31,6 @@ def create_session(proxy_url=""):
 
     return session
 
-
 def load_database(path=DEFAULT_DB_PATH):
     if not os.path.isfile(path):
         print("Database not loaded")
@@ -47,15 +45,13 @@ def save_database(posts, path=DEFAULT_DB_PATH):
     if directory and not os.path.exists(directory):
         os.makedirs(directory)
 
-    temporary_path = os.fspath(path) + ".tmp"
+    with open(path, "w", encoding="utf-8") as f:
-    with open(temporary_path, "w", encoding="utf-8") as f:
         json.dump([post.key for post in posts], f, ensure_ascii=False)
-    os.replace(temporary_path, path)
         print("Database updated")
 
 
 def fetch_posts(session, url, today=None):
-    req = session.get(url, timeout=SOURCE_TIMEOUT)
+    req = session.get(url)
 
     if req.status_code != 200:
         raise RuntimeError("Request error: " + str(req.status_code))
@@ -92,9 +88,9 @@ def get_new_posts(posts, database):
     return [post for post in posts if post.key not in database_keys]
 
 
-def publish_new_posts(send_message, session, url, db_path=DEFAULT_DB_PATH):
+def publish_new_posts(send_message, session, db_path=DEFAULT_DB_PATH):
     database = load_database(db_path)
-    posts = fetch_posts(session, url)
+    posts = fetch_posts(session, DEFAULT_VODOKANAL_URL)
 
     if not posts:
         print("No posts")

start_meshcore_ha.py

@@ -0,0 +1,56 @@
+import os
+
+from parser import create_session, publish_new_posts
+
+
+PROXY_URL = os.environ.get('PROXY_URL', '')
+HOME_ASSISTANT_URL = os.environ.get('HOME_ASSISTANT_URL', '')
+HOME_ASSISTANT_WEBHOOK_ID = os.environ.get('HOME_ASSISTANT_WEBHOOK_ID', '')
+HOME_ASSISTANT_WEBHOOK_CHANNEL = os.environ.get('HOME_ASSISTANT_WEBHOOK_CHANNEL', '0')
+
+
+def create_webhook_url():
+    return (
+        HOME_ASSISTANT_URL.rstrip('/')
+        + "/api/webhook/"
+        + HOME_ASSISTANT_WEBHOOK_ID
+    )
+
+
+def send_webhook_message(session, message):
+    req = session.get(
+        create_webhook_url(),
+        params={
+            'channel': HOME_ASSISTANT_WEBHOOK_CHANNEL,
+            'message': message,
+        },
+    )
+    if not 200 <= req.status_code < 300:
+        print("Home Assistant webhook request error: " + str(req.status_code))
+        exit()
+    else:
+        print("Home Assistant webhook message sent")
+
+
+def main():
+    if HOME_ASSISTANT_URL == '':
+        print("Home Assistant URL is not set")
+        exit()
+
+    if HOME_ASSISTANT_WEBHOOK_ID == '':
+        print("Home Assistant webhook id is not set")
+        exit()
+
+    session = create_session(PROXY_URL)
+    try:
+        publish_new_posts(
+            send_message=lambda message: send_webhook_message(session, message),
+            session=session,
+        )
+    except RuntimeError as error:
+        print(error)
+        exit()
+
+
+if __name__ == "__main__":
+    main()

start_telegram.py

@@ -1,13 +1,8 @@
-import fcntl
 import os
-import signal
 
 from parser import create_session, publish_new_posts
 
 
-URL = os.environ.get('VODOKANAL_URL', 'http://www.tgnvoda.ru/avarii.php')
-RUN_TIMEOUT_SECONDS = 9 * 60
-TELEGRAM_TIMEOUT = 60
 SEND_SILENT = os.environ.get('SEND_SILENT', False)
 TELEGRAM_TOKEN = os.environ.get('TELEGRAM_TOKEN', '')
 TELEGRAM_CHANNEL = os.environ.get('TELEGRAM_CHANNEL', '')
@@ -22,7 +17,6 @@ def send_telegram_message(session, message):
             'disable_notification': str(SEND_SILENT),
             'text': message,
         },
-        timeout=TELEGRAM_TIMEOUT,
     )
     if (req.status_code != 200):
         print("Telegram request error: " + str(req.status_code))
@@ -45,32 +39,11 @@ def main():
         publish_new_posts(
             send_message=lambda message: send_telegram_message(session, message),
             session=session,
-            url=URL,
         )
     except RuntimeError as error:
         print(error)
         exit()
 
 
-def _stop_overdue_run(signum, frame):
-    raise SystemExit(f"Bot run exceeded time limit ({RUN_TIMEOUT_SECONDS} seconds)")
-
-
-def run_once(lock_path="/tmp/vodokanalbot.lock"):
-    with open(lock_path, "a") as lock:
-        try:
-            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        except BlockingIOError:
-            print("Previous run is still active; skipping")
-            return
-        previous_handler = signal.signal(signal.SIGALRM, _stop_overdue_run)
-        try:
-            signal.alarm(RUN_TIMEOUT_SECONDS)
-            main()
-        finally:
-            signal.alarm(0)
-            signal.signal(signal.SIGALRM, previous_handler)
-
-
 if __name__ == "__main__":
-    run_once()
+    main()

tests/test_run_safety.py

@@ -1,147 +0,0 @@
-import fcntl
-import signal
-import tempfile
-import threading
-import time
-import unittest
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
-from pathlib import Path
-from unittest.mock import patch
-
-import requests
-
-import parser
-import start_telegram
-
-
-class RunSafetyTests(unittest.TestCase):
-    def test_next_run_delivers_new_posts_after_skipped_check(self):
-        with tempfile.TemporaryDirectory() as directory:
-            path = Path(directory) / "bot.lock"
-            db_path = str(Path(directory) / "db.json")
-            old = parser.Post("01.01", "old")
-            new = parser.Post("01.01", "new")
-            parser.save_database([old], db_path)
-            sent = []
-            with patch.object(parser, "fetch_posts", return_value=[old, new]), patch.object(
-                start_telegram, "main", side_effect=lambda: parser.publish_new_posts(
-                    sent.append, None, "http://source", db_path=db_path
-                )
-            ):
-                with path.open("a") as lock:
-                    fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
-                    start_telegram.run_once(path)
-                    self.assertEqual([], sent)
-                    self.assertEqual([old.key], parser.load_database(db_path))
-                start_telegram.run_once(path)
-                self.assertEqual([new.text], sent)
-                self.assertEqual([old.key, new.key], parser.load_database(db_path))
-
-    def test_unresponsive_telegram_times_out(self):
-        class SlowHandler(BaseHTTPRequestHandler):
-            def do_GET(self):
-                time.sleep(0.2)
-                self.send_response(200)
-                self.end_headers()
-                try:
-                    self.wfile.write(b'{"result": {"message_id": 42}}')
-                except BrokenPipeError:
-                    pass
-
-            def log_message(self, *args):
-                pass
-
-        with ThreadingHTTPServer(("127.0.0.1", 0), SlowHandler) as server:
-            class LocalTelegramSession(requests.Session):
-                def get(self, url, **kwargs):
-                    return super().get(f"http://127.0.0.1:{server.server_port}", **kwargs)
-
-            worker = threading.Thread(target=server.serve_forever, daemon=True)
-            worker.start()
-            try:
-                with LocalTelegramSession() as session, patch.object(
-                    start_telegram, "TELEGRAM_TIMEOUT", (0.05, 0.05), create=True
-                ):
-                    session.trust_env = False
-                    with self.assertRaises(requests.exceptions.Timeout):
-                        start_telegram.send_telegram_message(session, "test")
-            finally:
-                server.shutdown()
-                worker.join(timeout=2)
-
-    def test_interrupted_database_save_preserves_previous_data(self):
-        def interrupt_save(value, file, **kwargs):
-            file.write("[")
-            raise SystemExit("time limit")
-
-        with tempfile.TemporaryDirectory() as directory:
-            path = str(Path(directory) / "db.json")
-            parser.save_database([parser.Post("01.01", "previous")], path)
-            with patch.object(parser.json, "dump", side_effect=interrupt_save):
-                with self.assertRaisesRegex(SystemExit, "time limit"):
-                    parser.save_database([parser.Post("01.01", "new")], path)
-            self.assertEqual(["01.01$previous"], parser.load_database(path))
-
-    def test_unresponsive_source_times_out(self):
-        class SlowHandler(BaseHTTPRequestHandler):
-            def do_GET(self):
-                time.sleep(0.2)
-                self.send_response(200)
-                self.end_headers()
-
-            def log_message(self, *args):
-                pass
-
-        with ThreadingHTTPServer(("127.0.0.1", 0), SlowHandler) as server:
-            worker = threading.Thread(target=server.serve_forever, daemon=True)
-            worker.start()
-            try:
-                with requests.Session() as session, patch.object(
-                    parser, "SOURCE_TIMEOUT", (0.05, 0.05), create=True
-                ):
-                    session.trust_env = False
-                    with self.assertRaises(requests.exceptions.Timeout):
-                        parser.fetch_posts(session, f"http://127.0.0.1:{server.server_port}")
-            finally:
-                server.shutdown()
-                worker.join(timeout=2)
-
-    def test_overlapping_run_is_skipped(self):
-        with tempfile.TemporaryDirectory() as directory:
-            path = Path(directory) / "bot.lock"
-            with path.open("a") as lock, patch.object(start_telegram, "main") as main:
-                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
-                start_telegram.run_once(path)
-                main.assert_not_called()
-
-    def test_completed_run_releases_lock(self):
-        with tempfile.TemporaryDirectory() as directory, patch.object(start_telegram, "main") as main:
-            path = Path(directory) / "bot.lock"
-            start_telegram.run_once(path)
-            start_telegram.run_once(path)
-            self.assertEqual(2, main.call_count)
-
-    def test_failed_run_releases_lock(self):
-        with tempfile.TemporaryDirectory() as directory, patch.object(
-            start_telegram, "main", side_effect=[RuntimeError("source failed"), None]
-        ) as main:
-            path = Path(directory) / "bot.lock"
-            with self.assertRaisesRegex(RuntimeError, "source failed"):
-                start_telegram.run_once(path)
-            start_telegram.run_once(path)
-            self.assertEqual(2, main.call_count)
-
-    def test_overdue_run_stops_and_allows_next_run(self):
-        previous_handler = signal.getsignal(signal.SIGALRM)
-        with tempfile.TemporaryDirectory() as directory, patch.object(
-            start_telegram, "RUN_TIMEOUT_SECONDS", 1, create=True
-        ):
-            path = Path(directory) / "bot.lock"
-            with patch.object(start_telegram, "main", side_effect=lambda: time.sleep(2)):
-                with self.assertRaisesRegex(SystemExit, "time limit"):
-                    start_telegram.run_once(path)
-            self.assertEqual(previous_handler, signal.getsignal(signal.SIGALRM))
-            self.assertEqual(0, signal.alarm(0))
-            with patch.object(start_telegram, "main") as main:
-                start_telegram.run_once(path)
-                main.assert_called_once_with()