защиты и лекции поменял местами

feat: add exams

.env.sample

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

.gitignore

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

Dockerfile

@@ -0,0 +1,15 @@
+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

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

main.py

@@ -1,13 +1,13 @@
-import datetime
-import logging
-import os
-import requests
-import asyncio
-from aiogram import Bot
 import datetime as dt
 import locale
+import logging
+import os
+import re
+import time
 import urllib.parse
 
+import requests
+
 # Modify the links and data below:
 DEADLINES_URL = "https://m3104.nawinds.dev/DEADLINES.json"
 ADD_DEADLINE_LINK = "https://m3104.nawinds.dev/deadlines-editing-instructions/"
@@ -15,37 +15,88 @@ 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"))
+MAIN_GROUP_ID = int(os.getenv("MAIN_GROUP_ID") or '0')
+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}&details={urllib.parse.quote(description)}&" \
-           f"color=6"
+           f"dates={formatted_time}/{formatted_time}"
     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
@@ -64,152 +115,133 @@ async 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"])
-    if dt_obj < dt.datetime.now(dt_obj.tzinfo):
-        return False
-    return True
+    return not dt_obj < dt.datetime.now(dt_obj.tzinfo)
 
-def deadline_type_filter_func(d: dict, dtype: str) -> bool:
-    if f"[{dtype.lower()}]" in d["name"].lower():
-        return True
-    return False
 
-def deadlines_filter_func(d: dict) -> bool:
-    return (not deadline_type_filter_func(d, "тест") and
-            not deadline_type_filter_func(d, "лекция"))
+def deadline_type_filter_func(d: dict, dtype: str = '') -> bool:
+    if not dtype:
+        return not re.match(r'^\[.*\]', d['name'])
 
-async def get_message_text() -> str:
+    return f"[{dtype.lower()}]" in d["name"].lower()
+
+
+def get_message_text() -> str:
     try:
         response = requests.get(DEADLINES_URL).json()
     except Exception as e:
-        print(f"{datetime.datetime.now()} Failed to fetch deadlines: {e}")
+        print(f"{dt.datetime.now()} Failed to fetch deadlines: {e}")
         return ""
     all_deadlines = response["deadlines"]
 
-    deadlines = list(filter(lambda d: deadlines_filter_func(d) and relevant_filter_func(d), all_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))
+    types = [
+        ('', ''), # deadlines
+        ('🧑‍💻 Тесты', 'тест'),
+        ('🛡 Защиты', 'защита'),
+        ('🎓 Лекции', 'лекция'),
+        ('🤓 Экзамены', 'экзамен'),
+        ('👞 Консультации', 'консультация'),
+    ]
 
-    text = f"🔥️️ <b>Дедлайны</b> (<i>Обновлено в {await get_current_time()} 🔄</i>):\n\n"
+    assignments = [(sorted(filter(lambda t: deadline_type_filter_func(t, x[1]) and relevant_filter_func(t), all_deadlines),
+                           key=lambda z: timestamp_func(z)), x[0], x[1]) for x in types]
 
-    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))
+    text = f"🔥️️ <b>Дедлайны</b> (<i>Обновлено в {get_current_time()} 🔄</i>):\n\n"
 
-    if len(deadlines) == 0:
+    if len(assignments[0]) == 0:
         text += "Дедлайнов нет)\n\n"
 
-    for i in range(len(deadlines)):
-        no = i + 1
-        if no < 11:
-            no = NUMBER_EMOJIS[no] + " "
-        else:
-            no += ". "
-        text += str(no) + "<b>"
+    def add_items(items: list, category_name: str = '', replace_name: str = ''):
+        if len(items) == 0:
+            return
 
-        if deadlines[i].get("url"):
-            text += f"<a href='{deadlines[i]['url']}'>{deadlines[i]['name']}</a>"
-        else:
-            text += deadlines[i]["name"]
+        nonlocal text
+        REPLACE_PATTERN = re.compile(rf'^\[{replace_name}\] ', flags=re.IGNORECASE)
 
-        text += "</b> — "
-        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"
+        if category_name:
+            text += f"\n<b>{category_name}</b>:\n\n"
 
-    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")
+        for i, item in enumerate(items):
             no = i + 1
-            if no < 11:
+            if no <= 10:
                 no = NUMBER_EMOJIS[no] + " "
             else:
-                no += ". "
-            text += str(no) + "<b>"
+                no = str(no) + ". "
 
-            if test_url:
-                text += f"<a href='{test_url}'>{test_name}</a>"
+            text += no + "<b>"
+
+            name = re.sub(REPLACE_PATTERN, '', item['name'])
+            url = item.get('url')
+
+            if url:
+                text += f"<a href='{url}'>{name}</a>"
             else:
-                text += test_name
+                text += name
 
             text += "</b> — "
-            text += await get_human_timedelta(tests[i]["time"])
-            text += f"\n(<a href='{await generate_link(test_name, tests[i]['time'])}'>"
-            text += await get_human_time(tests[i]["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] + " "
+            text += get_human_timedelta(item["time"])
+            if ADD_CALENDAR_LINK:
+                text += f"\n(<a href='{generate_link(name, item['time'])}'>"
+                text += get_human_time(item["time"]) + "</a>)\n\n"
             else:
-                no += ". "
-            text += str(no) + "<b>"
+                text += f'\n({get_human_time(item["time"])})\n\n'
 
-            if lecture_url:
-                text += f"<a href='{lecture_url}'>{lecture_name}</a>"
-            else:
-                text += lecture_name
+    for assignment_type in assignments:
+        add_items(*assignment_type)
 
-            text += "</b> — "
-            text += await get_human_timedelta(lectures[i]["time"])
-            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>"
+    )
 
-    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
 
-    msg = await bot.send_message(chat_id, text, parse_mode="HTML", disable_web_page_preview=True)
+def main() -> None:
+    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(datetime.datetime.now(), "Message sent. Msg id:", msg.message_id)
+    print(dt.datetime.now(), "Message sent. Msg id:", msg_id)
 
-    while dt.datetime.now() - started_updating < dt.timedelta(days=1):
-        await asyncio.sleep(60)
+    condition = (lambda: True) if EDIT_MESSAGE_ID else (lambda: dt.datetime.now() - started_updating < dt.timedelta(days=1))
+    while condition():
+        time.sleep(60)
         try:
-            new_text = await get_message_text()
+            new_text = get_message_text()
             if text != new_text and new_text != "":
-                await msg.edit_text(new_text, parse_mode="HTML", disable_web_page_preview=True)
+                edit_message(msg_id, new_text)
                 text = new_text
-                print(datetime.datetime.now(), "Message updated. Msg id:", msg.message_id)
+                print(dt.datetime.now(), "Message updated. Msg id:", msg_id)
             else:
-                print(datetime.datetime.now(), "Message update skipped. Msg id:", msg.message_id)
+                print(dt.datetime.now(), "Message update skipped. Msg id:", msg_id)
 
         except Exception as e:
-            logging.warning(datetime.datetime.now(),f"{datetime.datetime.now()} Error updating message: {e}")
+            logging.warning(dt.datetime.now(),f"{dt.datetime.now()} Error updating message: {e}")
             continue
-    await msg.delete()
 
-
-async def main():
-    await send_deadlines(MAIN_GROUP_ID)
-    await bot.session.close()
+    if not EDIT_MESSAGE_ID:
+        delete_message(msg_id)
 
 
 if __name__ == '__main__':
-    asyncio.run(main())
+    main()

requirements.txt

@@ -1,21 +1,5 @@
-aiofiles==24.1.0
-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
+certifi==2025.4.26
+charset-normalizer==3.4.2
 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
-typing_extensions==4.12.2
-urllib3==2.2.3
-yarl==1.15.5
+urllib3==2.4.0