From 640ffd161618aba81e82dcfedf821a2891e5155d Mon Sep 17 00:00:00 2001 From: Tamar Kalir Date: Tue, 14 Jan 2025 19:41:24 +0200 Subject: [PATCH 1/2] prevent trigger_generate_infographics_and_send_to_telegram from triggering dag if no widgets are found --- anyway/telegram_accident_notifications.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/anyway/telegram_accident_notifications.py b/anyway/telegram_accident_notifications.py index aac2d24ae..9686c6452 100644 --- a/anyway/telegram_accident_notifications.py +++ b/anyway/telegram_accident_notifications.py @@ -4,7 +4,7 @@ from anyway.models import TelegramForwardedMessages from anyway.utilities import trigger_airflow_dag from anyway.app_and_db import db -from anyway.infographics_utils import get_infographics_data_by_newsflash +from anyway.infographics_utils import get_infographics_data_by_newsflash, WIDGETS import telebot import boto3 import time @@ -131,7 +131,12 @@ def create_public_urls_for_infographics_images(folder_name): def trigger_generate_infographics_and_send_to_telegram(newsflash_id, pre_verification_chat=True): - dag_conf = {"news_flash_id": newsflash_id} - dag_conf["chat_id"] = TELEGRAM_CHANNEL_CHAT_ID if pre_verification_chat \ - else TELEGRAM_POST_VERIFICATION_CHANNEL_CHAT_ID - trigger_airflow_dag("generate-and-send-infographics-images", dag_conf) + infographics_data = get_infographics_data_by_newsflash(newsflash_id) + if WIDGETS not in infographics_data: + logging.warning(f"no infographics to send for newsflash {newsflash_id}") + else: + logging.info(f"widgets found in json for {newsflash_id}, triggering dag") + dag_conf = {"news_flash_id": newsflash_id} + dag_conf["chat_id"] = TELEGRAM_CHANNEL_CHAT_ID if pre_verification_chat \ + else TELEGRAM_POST_VERIFICATION_CHANNEL_CHAT_ID + trigger_airflow_dag("generate-and-send-infographics-images", dag_conf) From 763d64366e56179c3cca71baaa544aa9eceec3cc Mon Sep 17 00:00:00 2001 From: Tamar Kalir Date: Tue, 14 Jan 2025 20:47:00 +0200 Subject: [PATCH 2/2] added structured logs for infographic_image_generator.upload_directory_to_s3 and send_infographics_to_telegram --- anyway/infographic_image_generator.py | 32 ++++++++++++++--- anyway/telegram_accident_notifications.py | 42 ++++++++++++++++++----- 2 files changed, 62 insertions(+), 12 deletions(-) diff --git a/anyway/infographic_image_generator.py b/anyway/infographic_image_generator.py index 6878a8715..c2aad5831 100644 --- a/anyway/infographic_image_generator.py +++ b/anyway/infographic_image_generator.py @@ -133,11 +133,35 @@ def download_infographics_images(generated_images_names, newsflash_id, local_inf def upload_directory_to_s3(download_directory, newsflash_id): - s3uploader = S3Uploader() - path = f"{download_directory}/" - for filename in os.listdir(path): - s3uploader.upload_to_s3(f"{path}/{filename}", newsflash_id) + log_message = { + "event": "upload_to_s3", + "newsflash_id": newsflash_id + } + uploaded_files = [] + try: + s3uploader = S3Uploader() + path = f"{download_directory}/" + filenames = os.listdir(path) + for filename in filenames: + s3uploader.upload_to_s3(f"{path}/{filename}", newsflash_id) + uploaded_files.append(filename) + + log_message.update ({ + "status": "success", + "file_count": len(filenames), + "files": filenames + }) + logging.info(json.dumps(log_message)) + except Exception as e: + log_message.update ({ + "status": "failure", + "uploaded_files": uploaded_files, + "error": str(e), + "traceback": repr(e), + }) + logging.error(json.dumps(log_message)) + raise #raise exception again to fail dag class S3Uploader(S3DataClass): def __init__(self): diff --git a/anyway/telegram_accident_notifications.py b/anyway/telegram_accident_notifications.py index 9686c6452..ef48888a2 100644 --- a/anyway/telegram_accident_notifications.py +++ b/anyway/telegram_accident_notifications.py @@ -97,14 +97,40 @@ def send_infographics_to_telegram(root_message_id, newsflash_id, channel_of_init #to create a comment on the channel message, we need to send a reply to the #forwareded message in the discussion group. bot = telebot.TeleBot(secrets.get("BOT_TOKEN")) - - linked_group = telegram_linked_group_by_channel[channel_of_initial_message] - items_for_send = get_items_for_send(newsflash_id) - for url, text in items_for_send: - bot.send_photo(linked_group, url, reply_to_message_id=root_message_id, caption=text) - - send_after_infographics_message(bot, root_message_id, newsflash_id, linked_group) - logging.info("notification send done") + sent_items = [] + channel_type = "post verification" if channel_of_initial_message == TELEGRAM_POST_VERIFICATION_CHANNEL_CHAT_ID \ + else "pre verification" + + log_message = { + "event": "send infographics to telegram", + "newsflash_id": newsflash_id, + "root_message_id": root_message_id, + "channel_of_initial_message": channel_of_initial_message, + "channel_type": channel_type + } + try: + linked_group = telegram_linked_group_by_channel[channel_of_initial_message] + items_for_send = get_items_for_send(newsflash_id) + + for url, text in items_for_send: + bot.send_photo(linked_group, url, reply_to_message_id=root_message_id, caption=text) + sent_items.append((url, text)) + + send_after_infographics_message(bot, root_message_id, newsflash_id, linked_group) + log_message.update({ + "status": "success", + "sent_count": len(sent_items) + }) + + except Exception as e: + log_message.update({ + "sent_items": sent_items, + "sent_count": len(sent_items), + "status": "failure", + "error": str(e) + }) + logging.error(json.dumps(log_message)) + raise def extract_infographic_name_from_s3_object(s3_object_name):