From c5c4a636686a8971a135fbc2b7140a4aab6e27d0 Mon Sep 17 00:00:00 2001 From: Sergey Karmanov Date: Wed, 28 Oct 2026 18:11:59 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20=D0=BE=D0=B3=D1=80=D0=B0=D0=BD=D0=B8?= =?UTF-8?q?=D1=87=D0=B8=D1=82=D1=8C=20=D0=B2=D1=80=D0=B5=D0=BC=D1=8F=20?= =?UTF-8?q?=D0=B7=D0=B0=D0=BF=D1=80=D0=BE=D1=81=D0=BE=D0=B2=20=D0=B8=20?= =?UTF-8?q?=D0=B8=D1=81=D0=BA=D0=BB=D1=8E=D1=87=D0=B8=D1=82=D1=8C=20=D0=BF?= =?UTF-8?q?=D0=B0=D1=80=D0=B0=D0=BB=D0=BB=D0=B5=D0=BB=D1=8C=D0=BD=D1=8B?= =?UTF-8?q?=D0=B5=20=D0=B7=D0=B0=D0=BF=D1=83=D1=81=D0=BA=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Добавлены таймауты запросов к сайту и Telegram, общий предел выполнения 9 минут и файловая блокировка запусков. База сообщений сохраняется атомарно. Регрессионные тесты запускаются перед публикацией Docker-образа. --- .github/workflows/github-push-docker.yml | 13 +- README.md | 10 ++ parser.py | 9 +- start_telegram.py | 27 ++++- tests/test_run_safety.py | 147 +++++++++++++++++++++++ 5 files changed, 201 insertions(+), 5 deletions(-) create mode 100644 tests/test_run_safety.py diff --git a/.github/workflows/github-push-docker.yml b/.github/workflows/github-push-docker.yml index 552f4dd..a916965 100644 --- a/.github/workflows/github-push-docker.yml +++ b/.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 }} \ No newline at end of file + labels: ${{ steps.meta.outputs.labels }} diff --git a/README.md b/README.md index 35789cc..75647db 100644 --- a/README.md +++ b/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`: diff --git a/parser.py b/parser.py index 86a25ae..974b84e 100644 --- a/parser.py +++ b/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)) diff --git a/start_telegram.py b/start_telegram.py index 1a3b6e0..78386c6 100644 --- a/start_telegram.py +++ b/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() diff --git a/tests/test_run_safety.py b/tests/test_run_safety.py new file mode 100644 index 0000000..c425c6b --- /dev/null +++ b/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()