179 lines
6.3 KiB
Python
179 lines
6.3 KiB
Python
import simplematrixbotlib as botlib
|
|
import requests
|
|
import datetime
|
|
import asyncio
|
|
import pytz
|
|
from typing import Dict
|
|
import os
|
|
import logging
|
|
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()
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
|
|
|
|
@bot.listener.on_startup
|
|
async def initial_schedule(room_id):
|
|
logging.info("starting praying times bot")
|
|
users = session.query(User).filter(User.location != "", User.reminder_time_in_minutes is not None).all()
|
|
for user in users:
|
|
logging.info("scheduling initial reminder for {}".format(user.username))
|
|
# reset last reminder schedule time
|
|
user.current_reminder_date = None
|
|
session.add(user)
|
|
session.commit()
|
|
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() and room.room_id:
|
|
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_minutes = minutes
|
|
user.room_id = room.room_id
|
|
session.add(user)
|
|
session.commit()
|
|
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:
|
|
logging.info("scheduling reminder for {}".format(username))
|
|
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"))
|
|
logging.info("scheduling reminder for {} with message: {}".format(username, message))
|
|
asyncio.ensure_future(remind(username, message, seconds - user.reminder_time_in_minutes * 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()
|