131 lines
4.2 KiB
Python
131 lines
4.2 KiB
Python
import re
|
|
import sys
|
|
from datetime import datetime, timedelta
|
|
import pytz
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
from telegram import Update, User
|
|
from telegram.ext import Application, CommandHandler, ContextTypes
|
|
|
|
reminders = {}
|
|
|
|
# TODO: fix max delay 24h
|
|
class CustomContext(ContextTypes.DEFAULT_TYPE):
|
|
def __init__(self, application, chat_id=None, user_id=None):
|
|
super().__init__(application=application)
|
|
self.chat_id = chat_id
|
|
self.user_id = user_id
|
|
if 'scheduler' not in application.bot_data:
|
|
application.bot_data['scheduler'] = AsyncIOScheduler(timezone="Europe/Warsaw")
|
|
application.bot_data['scheduler'].start()
|
|
self.scheduler = application.bot_data['scheduler']
|
|
|
|
|
|
async def schedule_reminder(context: CustomContext, chat_id: int, user: User, delay: timedelta):
|
|
max_delay = timedelta(hours=24)
|
|
if delay > max_delay:
|
|
await context.bot.send_message(
|
|
chat_id=chat_id,
|
|
text="maksymalnie 24 degeneracie ty",
|
|
parse_mode="HTML"
|
|
)
|
|
return
|
|
|
|
now = datetime.now(pytz.timezone("Europe/Warsaw"))
|
|
reminder_time = now + delay
|
|
job_id = f"pranie_{chat_id}_{reminder_time.timestamp()}"
|
|
|
|
async def send_reminder():
|
|
await context.bot.send_message(
|
|
chat_id=chat_id,
|
|
text=f"🔔 PRANIE!!! PRANIE KURWA ROZWIEŚ TEJ <a href='tg://user?id={user.id}'>@{user.username}</a>",
|
|
parse_mode="HTML"
|
|
)
|
|
|
|
context.scheduler.add_job(
|
|
send_reminder,
|
|
trigger="date",
|
|
run_date=reminder_time,
|
|
id=job_id,
|
|
replace_existing=True,
|
|
)
|
|
|
|
if chat_id not in reminders:
|
|
reminders[chat_id] = []
|
|
reminders[chat_id].append((job_id, reminder_time.strftime("%H:%M"), user.id))
|
|
|
|
|
|
async def pranie(update: Update, context: CustomContext):
|
|
chat_id = update.effective_chat.id
|
|
user = update.effective_user
|
|
username = user.first_name
|
|
|
|
text = " ".join(context.args)
|
|
|
|
if not text:
|
|
await update.message.reply_text("usage: /pranie HH:MM, /pranie Xmin/Xh")
|
|
return
|
|
|
|
if re.match(r"^\d{1,2}:\d{2}$", text):
|
|
hour, minute = map(int, text.split(":"))
|
|
now = datetime.now(pytz.timezone("Europe/Warsaw"))
|
|
reminder_time = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
|
|
if reminder_time < now:
|
|
reminder_time += timedelta(days=1)
|
|
delay = reminder_time - now
|
|
elif re.match(r"^\d+\s*(m|min|minutes?|h|hours?)$", text):
|
|
num, unit = re.findall(r"(\d+)\s*(\w+)", text)[0]
|
|
if "h" in unit:
|
|
delay = timedelta(hours=int(num))
|
|
else:
|
|
delay = timedelta(minutes=int(num))
|
|
else:
|
|
await update.message.reply_text("zły format, /pranie HH:MM albo /pranie [X min/hours]")
|
|
return
|
|
|
|
await schedule_reminder(context, chat_id, user, delay)
|
|
await update.message.reply_text(f"przypominaja nastawiona, {username}")
|
|
|
|
async def praniehelp(update: Update, context: CustomContext):
|
|
help_text = (
|
|
"🧺 *PRANIEBOT - komendy:*\n"
|
|
"/pranie HH:MM - przypominajka na godzinę\n"
|
|
" _example: /pranie 14:30_\n\n"
|
|
"/pranie Xmin/Xh - przypominajka po czasie\n"
|
|
" _example: /pranie 30min_\n"
|
|
" _example: /pranie 2h_\n\n"
|
|
"/praniehelp - niestety nie rozwiesi prania"
|
|
)
|
|
await update.message.reply_text(help_text, parse_mode="Markdown")
|
|
|
|
|
|
async def start(update: Update):
|
|
await update.message.reply_text(
|
|
"/pranie HH:MM albo /pranie Xmin/Xh do ustawienia przypominajki\n"
|
|
"/praniehelp niestety nie rozwiesi prania"
|
|
)
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
if len(args) != 1:
|
|
print(f"Usage: {sys.argv[0]} <path to token file>", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
token_file, = args
|
|
token = open(token_file).read().strip()
|
|
|
|
context_types = ContextTypes(context=CustomContext)
|
|
application = Application.builder().token(token).context_types(context_types).build()
|
|
|
|
application.add_handler(CommandHandler("start", start))
|
|
application.add_handler(CommandHandler("pranie", pranie))
|
|
application.add_handler(CommandHandler("praniehelp", praniehelp))
|
|
|
|
print("zapierdala")
|
|
|
|
application.run_polling()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|