重複するメッセージと"訓練" については通知しない (original) (raw)

QZSS「みちびき」の災害・危機管理通報サービス(災危通報)のスクリプトについて。通知メッセージが重複する場合は通知しないように。また、"これは訓練です" という場合にも通知しないように、スクリプトを修正しました。GNSS受信基板 GR-M10-RP を使用しています。

www.kadenken.com

import time from datetime import datetime, timedelta from pushover import Client import os from dotenv import load_dotenv import re import json

load_dotenv()

PUSHOVER_USER_KEY = os.getenv('PUSHOVER_USER_KEY') PUSHOVER_API_TOKEN = os.getenv('PUSHOVER_API_TOKEN') client = Client(PUSHOVER_USER_KEY, api_token=PUSHOVER_API_TOKEN)

LOG_FILE_PATH = os.getenv('LOG_FILE_PATH')

STATE_FILE_PATH = os.getenv('STATE_FILE_PATH')

last_sent_message_content = None last_sent_time = None

def load_last_sent_state(): """前回送信したメッセージと時刻の状態をファイルから読み込む""" global last_sent_message_content, last_sent_time if os.path.exists(STATE_FILE_PATH): try: with open(STATE_FILE_PATH, 'r') as file: state = json.load(file) last_sent_message_content = state.get('last_sent_message_content') last_sent_time = datetime.fromisoformat(state.get('last_sent_time')) if state.get('last_sent_time') else None except (json.JSONDecodeError, FileNotFoundError) as e: print(f"Error loading state file: {e}") last_sent_message_content = None last_sent_time = None

def save_last_sent_state(): """前回送信したメッセージと時刻の状態をファイルに保存する""" global last_sent_message_content, last_sent_time state = { 'last_sent_message_content': last_sent_message_content, 'last_sent_time': last_sent_time.isoformat() if last_sent_time else None } with open(STATE_FILE_PATH, 'w') as file: json.dump(state, file)

def send_pushover_notification(full_message, message_content, announcement_time): """Pushover通知を送信し、状態を更新する""" global last_sent_message_content, last_sent_time

if message_content == last_sent_message_content and last_sent_time and (announcement_time - last_sent_time) <= timedelta(hours=6):
    return

client.send_message(full_message, title="Log Alert")
last_sent_message_content = message_content
last_sent_time = announcement_time
save_last_sent_state()

def extract_announcement_time(line): """行から発表時刻を抽出する""" match = re.search(r'発表時刻:\s*(\d{1,2})月(\d{1,2})日(\d{1,2})時(\d{1,2})分', line) if match: month, day, hour, minute = map(int, match.groups()) current_year = datetime.now().year return datetime(current_year, month, day, hour, minute) return None

def contains_training_message(lines, start_index): """発表時刻の行から30行前までに「これは訓練です」という文字列があるかを確認する""" for i in range(start_index - 1, max(0, start_index - 30) - 1, -1): if "発表時刻" in lines[i]: return False if "これは訓練です" in lines[i]: return True return False

def monitor_log_file(file_path): """ログファイルを監視し、条件に応じてPushover通知を送信する""" file_byte_pos = 0 while True: with open(file_path, 'r', encoding='utf-8') as file: file.seek(file_byte_pos) lines = file.readlines() file_byte_pos = file.tell()

        for i, line in enumerate(lines):
            if "埼玉" in line:
                
                start_index = i
                announcement_time = None
                for j in range(i, -1, -1):
                    if "発表時刻" in lines[j]:
                        start_index = j
                        announcement_time = extract_announcement_time(lines[j])
                        break
                if announcement_time:
                    
                    if contains_training_message(lines, start_index):
                        continue
                    
                    full_message = ''.join(lines[start_index:i+1])
                    
                    message_content = ''.join(lines[start_index+1:i+1])
                    send_pushover_notification(full_message, message_content, announcement_time)
    time.sleep(2)

if name == "main": load_last_sent_state() if os.path.exists(LOG_FILE_PATH): monitor_log_file(LOG_FILE_PATH) else: print(f"Log file {LOG_FILE_PATH} does not exist.")