fix: ограничить время запросов и исключить параллельные запуски

Добавлены таймауты запросов к сайту и Telegram, общий предел выполнения 9 минут и файловая блокировка запусков. База сообщений сохраняется атомарно. Регрессионные тесты запускаются перед публикацией Docker-образа.

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

@@ -19,6 +19,17 @@ 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:
@@ -38,4 +49,4 @@ jobs:
           context: .
           push: true
           tags: ${{ steps.meta.outputs.tags }}
-          labels: ${{ steps.meta.outputs.labels }}
+          labels: ${{ steps.meta.outputs.labels }}

README.md

@@ -25,6 +25,16 @@ docker compose up -d --build
 
 ## Интеграции
 
+Бот проверяет сообщения каждые 10 минут. Параллельные запуски защищены
+файловой блокировкой: если предыдущий запуск ещё работает, очередная проверка
+пропускается, не помечая сообщения обработанными. Следующий успешный запуск
+снова сравнит сообщения за текущий день с базой.
+
+Один запуск ограничен 9 минутами. Запрос к сайту водоканала имеет таймаут
+10 секунд на подключение и 60 секунд ожидания данных, запросы к Telegram —
+60 секунд. База `data/db.json` заменяется атомарно, чтобы прерывание записи
+не повредило предыдущие данные.
+
 Общая логика парсинга, работы с `data/db.json` и поиска новых сообщений вынесена в [`parser.py`](./parser.py).
 
 Для новой интеграции достаточно создать свой адаптер отправки и передать его в `publish_new_posts`:

parser.py

@@ -8,6 +8,7 @@ from bs4 import BeautifulSoup
 
 
 DEFAULT_DB_PATH = "data/db.json"
+SOURCE_TIMEOUT = (10, 60)
 
 
 @dataclass(frozen=True)
@@ -46,13 +47,15 @@ def save_database(posts, path=DEFAULT_DB_PATH):
     if directory and not os.path.exists(directory):
         os.makedirs(directory)
 
-    with open(path, "w", encoding="utf-8") as f:
+    temporary_path = os.fspath(path) + ".tmp"
+    with open(temporary_path, "w", encoding="utf-8") as f:
         json.dump([post.key for post in posts], f, ensure_ascii=False)
-        print("Database updated")
+    os.replace(temporary_path, path)
+    print("Database updated")
 
 
 def fetch_posts(session, url, today=None):
-    req = session.get(url)
+    req = session.get(url, timeout=SOURCE_TIMEOUT)
 
     if req.status_code != 200:
         raise RuntimeError("Request error: " + str(req.status_code))

start_telegram.py

@@ -1,9 +1,13 @@
+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', '')
@@ -18,6 +22,7 @@ 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))
@@ -47,5 +52,25 @@ def main():
         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__":
-    main()
+    run_once()

tests/test_run_safety.py

@@ -0,0 +1,147 @@
+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()