245 lines
8.3 KiB
Python
245 lines
8.3 KiB
Python
import os
|
|
import sys
|
|
import json
|
|
import datetime
|
|
import configparser
|
|
import threading
|
|
import logging
|
|
import signal
|
|
import asyncio
|
|
from collections import namedtuple
|
|
|
|
import aioxmpp
|
|
from aioxmpp.structs import PresenceShow
|
|
|
|
import requests
|
|
|
|
from janine.utils import find_one, find_all
|
|
from janine.sources import sources_from_config, MiscDataSources
|
|
|
|
log = logging.getLogger('janine')
|
|
#log.basicConfig(level=logging.INFO)
|
|
|
|
Warning = namedtuple('Warning', ['id',
|
|
'sent',
|
|
'effective_from',
|
|
'expires',
|
|
'urgency',
|
|
'sender',
|
|
'headline',
|
|
'description',
|
|
'instruction'])
|
|
|
|
def to_warning(data):
|
|
'''
|
|
Returns a Warning given the raw data
|
|
'''
|
|
info = find_one(lambda x: 'headline' in x.keys(), data['info'])
|
|
return Warning(id=data['identifier'],
|
|
sent=data['sent'],
|
|
# Not all items have to have those
|
|
effective_from=info.get('effective', 'N/A'),
|
|
# Not all items have to have those
|
|
expires=info.get('expires', 'N/A'),
|
|
urgency=info['urgency'],
|
|
# Not all items have to have those
|
|
sender=info.get('senderName', 'N/A'),
|
|
headline=info['headline'],
|
|
description=info['description'],
|
|
instruction=info.get('instruction', 'N/A'))
|
|
|
|
def landkreis_filter(kreis, item):
|
|
'''
|
|
Returns True when the item is relevant to the Landkreis @kreis
|
|
'''
|
|
info = find_one(lambda e: 'area' in e.keys(), item['info'])
|
|
geocode = find_one(lambda e: 'geocode' in e.keys(), info['area'])
|
|
|
|
# Note: Some items may have multiple Landkreise
|
|
values = list(map(lambda e: e['valueName'], geocode['geocode']))
|
|
return kreis in values
|
|
|
|
def parse_data(text, filter_):
|
|
'''
|
|
Reads the remote response, parses it and returns a list of warnings.
|
|
'''
|
|
data = json.loads(text)
|
|
return [to_warning(raw) for raw in find_all(filter_, data)]
|
|
|
|
class WarningBot:
|
|
def __init__(self):
|
|
self._warnings0 = []
|
|
self._warnings1 = []
|
|
self._client = None
|
|
self._refresh_timeout = 630 # 15min
|
|
|
|
# Configuration stuff
|
|
self._data_dir = ''
|
|
|
|
self._load_config()
|
|
|
|
async def connect(self):
|
|
'''
|
|
Starts the "event loop" of the bot
|
|
'''
|
|
self._client = aioxmpp.PresenceManagedClient(
|
|
self._jid,
|
|
aioxmpp.make_security_layer(self._password))
|
|
|
|
# First we check if the entered Landkreis is valid
|
|
channels = {}
|
|
channel_file = os.path.join(self._data_dir, 'channels.json')
|
|
if not os.path.exists(channel_file):
|
|
log.debug('Requesting search channels')
|
|
req = requests.get(MiscDataSources.channels())
|
|
channels = json.loads(req.text)
|
|
|
|
try:
|
|
with open(channel_file, 'w') as f:
|
|
f.write(json.dumps(channels))
|
|
except Exception as err:
|
|
log.error('Failed to cache channel data:')
|
|
log.error(str(err))
|
|
else:
|
|
with open(channel_file, 'r') as f:
|
|
channels = json.loads(f.read())
|
|
|
|
for key in channels.keys():
|
|
if channels[key].get('NAME', '') == self._landkreis:
|
|
log.info('Landkreis valid')
|
|
break
|
|
else:
|
|
log.error('Invalid Landkreis')
|
|
return
|
|
|
|
async with self._client.connected() as stream:
|
|
logging.info('Client connected to server')
|
|
|
|
# In case you want a nice avatar
|
|
if self._avatar:
|
|
log.info('Setting avatar')
|
|
with open(self._avatar, 'rb') as f:
|
|
image_data = f.read()
|
|
|
|
avatar_set = aioxmpp.avatar.AvatarSet()
|
|
avatar_set.add_avatar_image('image/png', image_bytes=image_data)
|
|
await selg.avatar.publish_avatar_set(avatar_set)
|
|
|
|
# Set some presence information
|
|
self._client.set_presence(
|
|
aioxmpp.PresenceState(available=True,
|
|
show=PresenceShow.DND),
|
|
self._status)
|
|
|
|
# Start our fetch-send loop
|
|
# NOTE: Originally, I wanted to use a cronjob and
|
|
# signal.signal(...) for this but you can't
|
|
# use async in event handlers
|
|
loop = asyncio.get_event_loop()
|
|
periodic = loop.create_task(self._periodic_requests())
|
|
logging.info('Periodic ticker started')
|
|
await periodic
|
|
|
|
async def _periodic_requests(self):
|
|
'''
|
|
"Executes" every self._refresh_timeout seconds to fetch all
|
|
configured warnings and send them to the users, if there are
|
|
any new ones.
|
|
'''
|
|
while True:
|
|
logging.debug('Refreshing warning list')
|
|
await self._fetch_warnings()
|
|
await asyncio.sleep(self._refresh_timeout)
|
|
|
|
def _filter_func(self):
|
|
return lambda item: landkreis_filter(self._landkreis, item)
|
|
|
|
async def _fetch_warnings(self):
|
|
'''
|
|
Fetches all warnings and tries to find new ones
|
|
to send notifications.
|
|
'''
|
|
self._warnings1 = self._warnings0
|
|
self._warnings0 = []
|
|
for source in self._sources:
|
|
req = requests.get(source)
|
|
self._warnings0 = parse_data(req.text, self._filter_func())
|
|
|
|
# Find new warnings and send the new ones
|
|
ids = map(lambda x: x.id, self._warnings1)
|
|
for warning in self._warnings0:
|
|
if warning.id in ids:
|
|
continue
|
|
|
|
await self._send_notification(warning)
|
|
|
|
def __time_format(self, time_str):
|
|
'''
|
|
Reformat ISO style time data to a more
|
|
readable format.
|
|
'''
|
|
date = None
|
|
try:
|
|
date = datetime.datetime.fromisoformat(time_str)
|
|
except:
|
|
pass
|
|
|
|
if not date:
|
|
return time_str
|
|
|
|
return '{}.{}.{} {}:{}'.format(date.day,
|
|
date.month,
|
|
date.year,
|
|
date.hour,
|
|
date.minute)
|
|
|
|
async def _send_notification(self, warning):
|
|
'''
|
|
Send a warning to all the recipients
|
|
'''
|
|
# Reformat the message a bit
|
|
effective_time = self.__time_format(warning.effective_from)
|
|
expiry_time = self.__time_format(warning.expires)
|
|
body = '*{}*\n({} bis {})\n\n{}'.format(warning.headline,
|
|
effective_time,
|
|
expiry_time,
|
|
warning.description)
|
|
# Smells like script injection, but okay
|
|
body = body.replace('<br>', '\n')
|
|
body = body.replace('<br/>', '\n')
|
|
|
|
for recipient in self._recipients:
|
|
msg = aioxmpp.stanza.Message(
|
|
to=aioxmpp.JID.fromstr(recipient),
|
|
type_=aioxmpp.MessageType.CHAT)
|
|
msg.body[None] = body
|
|
await self._client.send(msg)
|
|
|
|
def _load_config(self):
|
|
# Load config
|
|
config_path = sys.argv[1] if len(sys.argv) == 2 else '/etc/janine/janine.conf'
|
|
config = configparser.ConfigParser()
|
|
config.read(config_path)
|
|
|
|
# Configure sources
|
|
self._sources = sources_from_config(config)
|
|
self._data_dir = config['General'].get('DataDir', '/etc/janine')
|
|
self._landkreis = config['General']['Landkreis']
|
|
self._recipients = config['General']['Recipients'].split(',')
|
|
self._refresh_timeout = int(config['General']['Timeout'])
|
|
|
|
# Bot Config
|
|
self._jid = aioxmpp.JID.fromstr(config['Bot']['JID'])
|
|
self._password = config['Bot']['Password']
|
|
self._avatar = config['Bot'].get('Avatar', None)
|
|
self._status = config['Bot'].get('Status', 'Warnt dich vor Katastrophen')
|
|
|
|
def main():
|
|
bot = WarningBot()
|
|
loop = asyncio.get_event_loop()
|
|
loop.run_until_complete(bot.connect())
|
|
loop.close()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|