.env.sample

@@ -1,4 +0,0 @@
-TOKEN=
-MAIN_GROUP_ID=
-EDIT_MESSAGE_ID=
-ADD_CALENDAR_LINK=

.gitignore

@@ -1 +0,0 @@
-.env

Dockerfile

@@ -1,15 +0,0 @@
-FROM python:3.12-bookworm
-STOPSIGNAL SIGINT
-WORKDIR /app
-
-# make ru_RU.UTF-8 locale available
-RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install \
-    --no-install-recommends -y locales && rm -r /var/lib/apt/lists/* && \
-    printf "en_US.UTF-8 UTF-8\nru_RU.UTF-8 UTF-8" >/etc/locale.gen && \
-    dpkg-reconfigure --frontend=noninteractive locales
-
-COPY requirements.txt .
-RUN pip install -r requirements.txt
-
-COPY main.py .
-CMD ["python3", "-u", "main.py"]

compose.yml

@@ -1,7 +0,0 @@
-services:
-  deadline_bot:
-    build: .
-    env_file: .env
-    volumes:
-      - /etc/localtime:/etc/localtime:ro
-    restart: unless-stopped

main.py

@@ -1,12 +1,12 @@
-import datetime as dt
+import datetime
-import locale
 import logging
 import os
-import re
-import time
-import urllib.parse
-
 import requests
+import asyncio
+from aiogram import Bot
+import datetime as dt
+import locale
+import urllib.parse
 
 # Modify the links and data below:
 DEADLINES_URL = "https://m3104.nawinds.dev/DEADLINES.json"
@@ -15,88 +15,37 @@ BOT_NAME = "Дединсайдер M3104"
 BOT_USERNAME = "m3104_deadliner_bot"
 
 # Environment variables that should be available:
-API_URL = 'https://api.telegram.org/bot'
 TOKEN = os.getenv("TOKEN")
-MAIN_GROUP_ID = int(os.getenv("MAIN_GROUP_ID") or '0')
+MAIN_GROUP_ID = int(os.getenv("MAIN_GROUP_ID"))
-EDIT_MESSAGE_ID = int(os.getenv("EDIT_MESSAGE_ID") or '0')
-ADD_CALENDAR_LINK = os.getenv("ADD_CALENDAR_LINK") != 'false'
-
-assert TOKEN, "Missing token!"
-assert MAIN_GROUP_ID, "Missing group ID!"
 
 logging.basicConfig(level=logging.INFO)
 
+bot = Bot(TOKEN)
+
 NUMBER_EMOJIS = ['0.', '1️⃣', '2️⃣', '3️⃣', '4️⃣', '5️⃣', '6️⃣', '7️⃣', '8️⃣', '9️⃣', '🔟']
 
-
+async def get_current_time() -> str:
-class TelegramException(Exception):
-    def __init__(self, *, error_code: int, description: str, **_):
-        super().__init__(f'Error {error_code}: {description}')
-        self.error_code = error_code
-        self.description = description
-
-
-def telegram_request(method: str, args: dict):
-    data = requests.post(API_URL + f'{TOKEN}/{method}', json=args).json()
-    if not data['ok']:
-        raise TelegramException(**data)
-    return data
-
-
-def send_message(text: str) -> int:
-    return telegram_request('sendMessage', {
-        'chat_id': MAIN_GROUP_ID,
-        'parse_mode': 'HTML',
-        'text': text,
-        'link_preview_options': {
-            'is_disabled': True
-        }
-    })['result']['message_id']
-
-
-def edit_message(message_id: int, text: str) -> int:
-    return telegram_request('editMessageText', {
-        'chat_id': MAIN_GROUP_ID,
-        'parse_mode': 'HTML',
-        'message_id': message_id,
-        'text': text,
-        'link_preview_options': {
-            'is_disabled': True
-        }
-    })['result']['message_id']
-
-
-def delete_message(message_id: int) -> bool:
-    return telegram_request('deleteMessage', {
-        'chat_id': MAIN_GROUP_ID,
-        'message_id': message_id
-    })['result']
-
-
-def get_current_time() -> str:
     current_time = dt.datetime.now()
     current_time_hour = current_time.hour if current_time.hour >= 10 else "0" + str(current_time.hour)
     current_time_minute = current_time.minute if current_time.minute >= 10 else "0" + str(current_time.minute)
     return f"{current_time_hour}:{current_time_minute}"
 
-
 def get_dt_obj_from_string(time: str) -> dt.datetime:
     time = time.replace('GMT+3', '+0300')
     locale.setlocale(locale.LC_TIME, 'en_US.UTF-8')
     return dt.datetime.strptime(time, "%d %b %Y %H:%M:%S %z")
 
-
+async def generate_link(event_name: str, event_time: str) -> str:
-def generate_link(event_name: str, event_time: str) -> str:
     dt_obj = get_dt_obj_from_string(event_time)
     formatted_time = dt_obj.strftime("%Y%m%d T%H%M%S%z")
     description = f"Дедлайн добавлен ботом {BOT_NAME} (https://t.me/{BOT_USERNAME})"
     link = f"https://calendar.google.com/calendar/u/0/r/eventedit?" \
            f"text={urllib.parse.quote(event_name)}&" \
-           f"dates={formatted_time}/{formatted_time}"
+           f"dates={formatted_time}/{formatted_time}&details={urllib.parse.quote(description)}&" \
+           f"color=6"
     return link
 
-
+async def get_human_timedelta(time: str) -> str:
-def get_human_timedelta(time: str) -> str:
     dt_obj = get_dt_obj_from_string(time)
     dt_now = dt.datetime.now(dt_obj.tzinfo)  # Ensure timezones are consistent
     delta = dt_obj - dt_now
@@ -115,133 +64,152 @@ def get_human_timedelta(time: str) -> str:
     else:
         return f"{hours}ч {minutes}м"
 
-
+async def get_human_time(time: str) -> str:
-def get_human_time(time: str) -> str:
     dt_obj = get_dt_obj_from_string(time)
     locale.setlocale(locale.LC_TIME, 'ru_RU.UTF-8')
     formatted_date = dt_obj.strftime("%a, %d %B в %H:%M")
     return formatted_date
 
-
 def timestamp_func(a: dict) -> float:
     time = a["time"].replace('GMT+3', '+0300')
     locale.setlocale(locale.LC_TIME, 'en_US.UTF-8')
     a_timestamp = dt.datetime.strptime(time, "%d %b %Y %H:%M:%S %z").timestamp()
     return a_timestamp  # 29 Oct 2024 23:59:59 GMT+3
 
-
 def relevant_filter_func(d: dict) -> bool:
     dt_obj = get_dt_obj_from_string(d["time"])
-    return not dt_obj < dt.datetime.now(dt_obj.tzinfo)
+    if dt_obj < dt.datetime.now(dt_obj.tzinfo):
+        return False
+    return True
 
+def deadline_type_filter_func(d: dict, dtype: str) -> bool:
+    if f"[{dtype.lower()}]" in d["name"].lower():
+        return True
+    return False
 
-def deadline_type_filter_func(d: dict, dtype: str = '') -> bool:
+def deadlines_filter_func(d: dict) -> bool:
-    if not dtype:
+    return (not deadline_type_filter_func(d, "тест") and
-        return not re.match(r'^\[.*\]', d['name'])
+            not deadline_type_filter_func(d, "лекция"))
 
-    return f"[{dtype.lower()}]" in d["name"].lower()
+async def get_message_text() -> str:
-
-
-def get_message_text() -> str:
     try:
         response = requests.get(DEADLINES_URL).json()
     except Exception as e:
-        print(f"{dt.datetime.now()} Failed to fetch deadlines: {e}")
+        print(f"{datetime.datetime.now()} Failed to fetch deadlines: {e}")
         return ""
     all_deadlines = response["deadlines"]
 
-    types = [
+    deadlines = list(filter(lambda d: deadlines_filter_func(d) and relevant_filter_func(d), all_deadlines))
-        ('', ''), # deadlines
+    tests = list(filter(lambda t: deadline_type_filter_func(t, "тест") and relevant_filter_func(t), all_deadlines))
-        ('🧑‍💻 Тесты', 'тест'),
+    lectures = list(filter(lambda t: deadline_type_filter_func(t, "лекция") and relevant_filter_func(t), all_deadlines))
-        ('🛡 Защиты', 'защита'),
-        ('🎓 Лекции', 'лекция'),
-        ('🤓 Экзамены', 'экзамен'),
-        ('👞 Консультации', 'консультация'),
-    ]
 
-    assignments = [(sorted(filter(lambda t: deadline_type_filter_func(t, x[1]) and relevant_filter_func(t), all_deadlines),
+    text = f"🔥️️ <b>Дедлайны</b> (<i>Обновлено в {await get_current_time()} 🔄</i>):\n\n"
-                           key=lambda z: timestamp_func(z)), x[0], x[1]) for x in types]
 
-    text = f"🔥️️ <b>Дедлайны</b> (<i>Обновлено в {get_current_time()} 🔄</i>):\n\n"
+    deadlines = sorted(deadlines, key=lambda x: timestamp_func(x))
+    tests = sorted(tests, key=lambda x: timestamp_func(x))
+    lectures = sorted(lectures, key=lambda x: timestamp_func(x))
 
-    if len(assignments[0]) == 0:
+    if len(deadlines) == 0:
         text += "Дедлайнов нет)\n\n"
 
-    def add_items(items: list, category_name: str = '', replace_name: str = ''):
+    for i in range(len(deadlines)):
-        if len(items) == 0:
+        no = i + 1
-            return
+        if no < 11:
+            no = NUMBER_EMOJIS[no] + " "
+        else:
+            no += ". "
+        text += str(no) + "<b>"
 
-        nonlocal text
+        if deadlines[i].get("url"):
-        REPLACE_PATTERN = re.compile(rf'^\[{replace_name}\] ', flags=re.IGNORECASE)
+            text += f"<a href='{deadlines[i]['url']}'>{deadlines[i]['name']}</a>"
+        else:
+            text += deadlines[i]["name"]
 
-        if category_name:
+        text += "</b> — "
-            text += f"\n<b>{category_name}</b>:\n\n"
+        text += await get_human_timedelta(deadlines[i]["time"])
+        text += f"\n(<a href='{await generate_link(deadlines[i]['name'], deadlines[i]['time'])}'>"
+        text += await get_human_time(deadlines[i]["time"]) + "</a>)\n\n"
 
-        for i, item in enumerate(items):
+    if len(tests) > 0:
+        text += f"\n🧑‍💻 <b>Тесты</b>:\n\n"
+
+        for i in range(len(tests)):
+            test_name = tests[i]["name"].replace("[Тест] ", "").replace("[тест]", "")
+            test_url = tests[i].get("url")
             no = i + 1
-            if no <= 10:
+            if no < 11:
                 no = NUMBER_EMOJIS[no] + " "
             else:
-                no = str(no) + ". "
+                no += ". "
+            text += str(no) + "<b>"
 
-            text += no + "<b>"
+            if test_url:
-
+                text += f"<a href='{test_url}'>{test_name}</a>"
-            name = re.sub(REPLACE_PATTERN, '', item['name'])
-            url = item.get('url')
-
-            if url:
-                text += f"<a href='{url}'>{name}</a>"
             else:
-                text += name
+                text += test_name
 
             text += "</b> — "
-            text += get_human_timedelta(item["time"])
+            text += await get_human_timedelta(tests[i]["time"])
-            if ADD_CALENDAR_LINK:
+            text += f"\n(<a href='{await generate_link(test_name, tests[i]['time'])}'>"
-                text += f"\n(<a href='{generate_link(name, item['time'])}'>"
+            text += await get_human_time(tests[i]["time"]) + "</a>)\n\n"
-                text += get_human_time(item["time"]) + "</a>)\n\n"
+
+    if len(lectures) > 0:
+        text += f"\n👨‍🏫 <b>Лекции</b>:\n\n"
+
+        for i in range(len(lectures)):
+            lecture_name = lectures[i]["name"].replace("[Лекция] ", "").replace("[лекция]", "")
+            lecture_url = lectures[i].get("url")
+            no = i + 1
+            if no < 11:
+                no = NUMBER_EMOJIS[no] + " "
             else:
-                text += f'\n({get_human_time(item["time"])})\n\n'
+                no += ". "
+            text += str(no) + "<b>"
 
-    for assignment_type in assignments:
+            if lecture_url:
-        add_items(*assignment_type)
+                text += f"<a href='{lecture_url}'>{lecture_name}</a>"
+            else:
+                text += lecture_name
 
-    text += (
+            text += "</b> — "
-        f"\n🆕 <a href='{ADD_DEADLINE_LINK}'>"
+            text += await get_human_timedelta(lectures[i]["time"])
-        f"Добавить дедлайн</a>"
+            text += f"\n(<a href='{await generate_link(lecture_name, lectures[i]['time'])}'>"
-    )
+            text += await get_human_time(lectures[i]["time"]) + "</a>)\n\n"
 
+    text += f"\n🆕 <a href='{ADD_DEADLINE_LINK}'>" \
+            f"Добавить дедлайн</a>"
     return text
 
+async def send_deadlines(chat_id: int) -> None:
+    text = await get_message_text()
+    if text == "Дедлайнов нет)\n\n":
+        return
 
-def main() -> None:
+    msg = await bot.send_message(chat_id, text, parse_mode="HTML", disable_web_page_preview=True)
-    text = get_message_text()
-
-    if EDIT_MESSAGE_ID:
-        msg_id = edit_message(EDIT_MESSAGE_ID, text)
-    else:
-        msg_id = send_message(text)
     started_updating = dt.datetime.now()
-    print(dt.datetime.now(), "Message sent. Msg id:", msg_id)
+    print(datetime.datetime.now(), "Message sent. Msg id:", msg.message_id)
 
-    condition = (lambda: True) if EDIT_MESSAGE_ID else (lambda: dt.datetime.now() - started_updating < dt.timedelta(days=1))
+    while dt.datetime.now() - started_updating < dt.timedelta(days=1):
-    while condition():
+        await asyncio.sleep(60)
-        time.sleep(60)
         try:
-            new_text = get_message_text()
+            new_text = await get_message_text()
             if text != new_text and new_text != "":
-                edit_message(msg_id, new_text)
+                await msg.edit_text(new_text, parse_mode="HTML", disable_web_page_preview=True)
                 text = new_text
-                print(dt.datetime.now(), "Message updated. Msg id:", msg_id)
+                print(datetime.datetime.now(), "Message updated. Msg id:", msg.message_id)
             else:
-                print(dt.datetime.now(), "Message update skipped. Msg id:", msg_id)
+                print(datetime.datetime.now(), "Message update skipped. Msg id:", msg.message_id)
 
         except Exception as e:
-            logging.warning(dt.datetime.now(),f"{dt.datetime.now()} Error updating message: {e}")
+            logging.warning(datetime.datetime.now(),f"{datetime.datetime.now()} Error updating message: {e}")
             continue
+    await msg.delete()
 
-    if not EDIT_MESSAGE_ID:
+
-        delete_message(msg_id)
+async def main():
+    await send_deadlines(MAIN_GROUP_ID)
+    await bot.session.close()
 
 
 if __name__ == '__main__':
-    main()
+    asyncio.run(main())

requirements.txt

@@ -1,5 +1,21 @@
-certifi==2025.4.26
+aiofiles==24.1.0
-charset-normalizer==3.4.2
+aiogram==3.13.1
+aiohappyeyeballs==2.4.3
+aiohttp==3.10.10
+aiosignal==1.3.1
+annotated-types==0.7.0
+async-timeout==4.0.3
+attrs==24.2.0
+certifi==2024.8.30
+charset-normalizer==3.4.0
+frozenlist==1.4.1
 idna==3.10
+magic-filter==1.0.12
+multidict==6.1.0
+propcache==0.2.0
+pydantic==2.9.2
+pydantic_core==2.23.4
 requests==2.32.3
-urllib3==2.4.0
+typing_extensions==4.12.2
+urllib3==2.2.3
+yarl==1.15.5