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