matrix-bot-praying-times/matrix_bot_praying_times/__main__.py

import simplematrixbotlib as botlib
import requests
import datetime
import asyncio
import pytz
from typing import Dict
import os
from dotenv import load_dotenv
from .models.user import User

from sqlalchemy import create_engine
from sqlalchemy.sql import select
from sqlalchemy.orm import sessionmaker

load_dotenv()

creds = botlib.Creds(
    os.getenv("BOT_HOMESERVER"),
    os.getenv("BOT_USERNAME"),
    os.getenv("BOT_PASSWORD"),
    session_stored_file="data/session.txt"
)

bot = botlib.Bot(creds)
PREFIX = '!'
utc = pytz.UTC

engine = create_engine(os.getenv("DB_URI"))
Session = sessionmaker(bind=engine)
session = Session()


@bot.listener.on_reaction_event
def initial_schedule() -> None:
    users = session.query(User).filter(User.location != "").all()
    for user in users:
        schedule_reminder(user.username)



@bot.listener.on_message_event
async def echo(room, message) -> None:
    match = botlib.MessageMatch(room, message, bot, PREFIX)

    if match.is_not_from_this_bot() and match.prefix() and match.command("echo"):

        await bot.api.send_text_message(
            room.room_id, " ".join(arg for arg in match.args())
        )


@bot.listener.on_message_event
async def times(room, message) -> None:
    match = botlib.MessageMatch(room, message, bot, PREFIX)
    username = str(message).split(': ')[0]
    response = ""

    user = session.query(User).filter(User.username == username, User.location != "").one_or_none()

    if match.is_not_from_this_bot() and match.prefix() and match.command("times"):
        if not user:
            response = "Please set your location first using *!set-location City, Country*"
        else:
            times = get_praying_times(datetime.datetime.today(), username)

            response = "Today's praying times for {}:\n\n".format(user.location)
            response += "\n".join("{}: {}".format(key, value) for key, value in times.items())

        await bot.api.send_markdown_message(room.room_id, response)


@bot.listener.on_message_event
async def usage(room, message) -> None:
    match = botlib.MessageMatch(room, message, bot, PREFIX)

    response = """usage:
- **!set-location** <City, Country>
- **!times**
- **!set-reminder** <X> where X is the number of minutes you want to receive the reminder before praying
    """

    if match.is_not_from_this_bot() and not match.prefix():
        await bot.api.send_markdown_message(room.room_id, response)


@bot.listener.on_message_event
async def set_location(room, message) -> None:
    match = botlib.MessageMatch(room, message, bot, PREFIX)

    response = ""

    if match.is_not_from_this_bot() and match.prefix() and match.command("set-location"):
        location = " ".join(arg for arg in match.args())
        username = str(message).split(': ')[0]

        user = session.query(User).filter_by(username=username).one_or_none()
        if not user:
            # create new user
            user = User(username=username, location=location)
            session.add(user)
        user.location = location
        session.commit()

        response = "Your location was set to: {}".format(location)
        await bot.api.send_markdown_message(room.room_id, response)


@bot.listener.on_message_event
async def set_reminder(room, message) -> None:
    match = botlib.MessageMatch(room, message, bot, PREFIX)

    response = ""

    if match.is_not_from_this_bot() and match.prefix() and match.command("set-reminder"):
        minutes = int(match.args()[0])
        username = str(message).split(': ')[0]
        user = session.query(User).filter(User.username == username, User.location != "").one_or_none()

        if not user:
            response = "You did not set your location yet."
            await bot.api.send_markdown_message(room.room_id, response)
        else:
            user.reminder_time_in_minues = minutes
            user.room_id = room.room_id
            response = "Your reminder was set to {} minutes before praying.".format(minutes)
            schedule_reminder(username)
            await bot.api.send_markdown_message(room.room_id, response)


def get_praying_times(date: datetime.date, username) -> Dict[str, str]:
    day = date.day
    month = date.month
    year = date.year
    user = session.query(User).filter_by(username=username).one_or_none()
    times_api_url = 'http://api.aladhan.com/v1/timingsByAddress/{}-{}-{}?address={}&method=7&iso8601=true'.format(day, month, year, user.location)
    times = requests.get(times_api_url)
    times = times.json()['data']['timings']
    return times


def schedule_reminder(username) -> None:
    now = datetime.datetime.now(datetime.UTC)
    user = session.query(User).filter_by(username=username).one_or_none()
    user.current_reminder_date = datetime.date.today()
    session.add(user)
    session.commit()
    times = get_praying_times(datetime.date.today(), username)
    for prayer, time in times.items():
        praying_time = datetime.datetime.fromisoformat(time)
        if praying_time > now:
            # schedule praying times but skip the ones which already have been passed today
            seconds = int((praying_time - now).total_seconds())
            message = "{} is at {}".format(prayer, praying_time.strftime("%H:%M"))
            asyncio.ensure_future(remind(username, message, seconds - user.reminder_time_in_minues * 60))
    # save date of last schedule for user


async def remind(username, message, seconds) -> None:
    await asyncio.sleep(seconds)
    user = session.query(User).filter_by(username=username).one_or_none()
    # check if a new reminder has to be scheduled
    if user.current_reminder_date < datetime.date.today():
        schedule_reminder(user.username)
    await bot.api.send_markdown_message(user.room_id, message)


bot.run()