python 3.x. Get notifications of EVE Online and post it to Discord. notification_types.json > https://gist.github.com/Omochin/b8dff8b60dff63e753240ca8dd3b03e7#file-notification_types-json

eveonline_post_notifications2discord.py

# pylint: disable=C0103,C0301
import os
import json
import datetime
import requests
from preston.xmlapi import Preston
import sqlalchemy.ext.declarative
from sqlalchemy import Column, Integer, String, DateTime

EVEONLINE_API_KEY = os.environ['EVEONLINE_API_KEY']
EVEONLINE_API_CODE = os.environ['EVEONLINE_API_CODE']
DATABASE_URL = os.environ['DATABASE_URL']
DISCORD_ALERT_WEBHOOK_URL = os.environ['DISCORD_ALERT_WEBHOOK_URL']
DISCORD_LOG_WEBHOOK_URL = os.getenv('DISCORD_LOG_WEBHOOK_URL', None)
DISCORD_PING_WEBHOOK_URL = os.getenv('DISCORD_PING_WEBHOOK_URL', None)
ALERT_NOTIFICATION_TYPE_IDS = os.getenv('ALERT_NOTIFICATION_TYPE_IDS', '184').split(':')
ALERT_NOTIFICATION_TEXT = '''
=================================================
| %s 
=================================================
date: %s
%s
'''

Base = sqlalchemy.ext.declarative.declarative_base()
class Notification(Base):
    __tablename__ = 'notifications'
    id = Column(Integer, primary_key=True)
    type_id = Column(Integer)
    sender_id = Column(Integer)
    sender_name = Column(String)
    sent_date = Column(DateTime)

def create_all():
    engine = sqlalchemy.create_engine(DATABASE_URL)
    Base.metadata.create_all(bind=engine)

def create_session():
    engine = sqlalchemy.create_engine(DATABASE_URL)
    return sqlalchemy.orm.sessionmaker(bind=engine)()

def truncate_all():
    session = create_session()
    session.query(Notification).delete()
    session.commit()

def post_json(url, content):
    if url is None:
        return

    requests.post(
        url,
        data=json.dumps({'content': content}),
        headers={'content-type': 'application/json'},
    )

def main():
    post_json(DISCORD_PING_WEBHOOK_URL, 'ping: ' + datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
    preston = Preston(key=EVEONLINE_API_KEY, code=EVEONLINE_API_CODE)

    try:
        character_id = preston.account.Characters()['rowset']['row']['@characterID']
    except KeyError:
        print('This key({}) is invalid'.format(EVEONLINE_API_KEY))
        return

    rowset = preston.char.Notifications(characterID=character_id)['rowset']
    if 'row' not in rowset:
        return

    session = create_session()

    with open('notification_types.json', 'r') as file:
        notification_dict = json.load(file)

    notification_rows = []
    row = rowset['row']
    for notification in row if isinstance(row, list) else [row]:
        notification_id = notification['@notificationID']
        exists = sqlalchemy.sql.exists().where(Notification.id == notification_id)

        if not session.query(exists).scalar():
            notification_row = Notification(
                id=notification_id,
                type_id=notification['@typeID'],
                sender_id=notification['@senderID'],
                sender_name=notification['@senderName'],
                sent_date=datetime.datetime.strptime(notification['@sentDate'], '%Y-%m-%d %H:%M:%S'),
            )
            session.add(notification_row)
            notification_rows.append(notification_row)

    session.commit()

    for notification in notification_rows:
        type_id = str(notification.type_id)
        row = preston.char.NotificationTexts(characterID=character_id, IDs=notification.id)['rowset']['row']
        text = ALERT_NOTIFICATION_TEXT % (
            notification_dict[type_id],
            notification.sent_date.strftime('%Y-%m-%d %H:%M:%S'),
            row['#text'],
        )

        post_json(DISCORD_LOG_WEBHOOK_URL, text)
        if type_id in ALERT_NOTIFICATION_TYPE_IDS:
            post_json(DISCORD_ALERT_WEBHOOK_URL, '@everyone ' + text)

if __name__ == '__main__':
    main()