diff --git a/.env.sample b/.env.sample deleted file mode 100644 index c52b606..0000000 --- a/.env.sample +++ /dev/null @@ -1,4 +0,0 @@ -TOKEN= -MAIN_GROUP_ID= -EDIT_MESSAGE_ID= -ADD_CALENDAR_LINK= diff --git a/.gitignore b/.gitignore deleted file mode 100644 index 4c49bd7..0000000 --- a/.gitignore +++ /dev/null @@ -1 +0,0 @@ -.env diff --git a/Dockerfile b/Dockerfile deleted file mode 100644 index ed3af3e..0000000 --- a/Dockerfile +++ /dev/null @@ -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"] diff --git a/compose.yml b/compose.yml deleted file mode 100644 index c854143..0000000 --- a/compose.yml +++ /dev/null @@ -1,7 +0,0 @@ -services: - deadline_bot: - build: . - env_file: .env - volumes: - - /etc/localtime:/etc/localtime:ro - restart: unless-stopped diff --git a/main.py b/main.py index c82caff..54b798d 100644 --- a/main.py +++ b/main.py @@ -1,12 +1,12 @@ -import datetime as dt -import locale +import datetime 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') -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!" +MAIN_GROUP_ID = int(os.getenv("MAIN_GROUP_ID")) logging.basicConfig(level=logging.INFO) +bot = Bot(TOKEN) + NUMBER_EMOJIS = ['0.', '1️⃣', '2️⃣', '3️⃣', '4️⃣', '5️⃣', '6️⃣', '7️⃣', '8️⃣', '9️⃣', '🔟'] - -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: +async 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") - -def generate_link(event_name: str, event_time: str) -> str: +async 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 - -def get_human_timedelta(time: str) -> str: +async 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}м" - -def get_human_time(time: str) -> str: +async 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: - if not dtype: - return not re.match(r'^\[.*\]', d['name']) +def deadlines_filter_func(d: dict) -> bool: + return (not deadline_type_filter_func(d, "тест") and + not deadline_type_filter_func(d, "лекция")) - return f"[{dtype.lower()}]" in d["name"].lower() - - -def get_message_text() -> str: +async 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 - ('🧑‍💻 Тесты', 'тест'), - ('🛡 Защиты', 'защита'), - ('🎓 Лекции', 'лекция'), - ('🤓 Экзамены', 'экзамен'), - ('👞 Консультации', 'консультация'), - ] + 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)) - 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] + text = f"🔥️️ Дедлайны (Обновлено в {await get_current_time()} 🔄):\n\n" - text = f"🔥️️ Дедлайны (Обновлено в {get_current_time()} 🔄):\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 = ''): - if len(items) == 0: - return + for i in range(len(deadlines)): + no = i + 1 + if no < 11: + no = NUMBER_EMOJIS[no] + " " + else: + no += ". " + text += str(no) + "" - nonlocal text - REPLACE_PATTERN = re.compile(rf'^\[{replace_name}\] ', flags=re.IGNORECASE) + if deadlines[i].get("url"): + text += f"{deadlines[i]['name']}" + else: + text += deadlines[i]["name"] - if category_name: - text += f"\n{category_name}:\n\n" + text += " — " + text += await get_human_timedelta(deadlines[i]["time"]) + text += f"\n(" + text += await get_human_time(deadlines[i]["time"]) + ")\n\n" - for i, item in enumerate(items): + if len(tests) > 0: + text += f"\n🧑‍💻 Тесты:\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) + "" - text += no + "" - - name = re.sub(REPLACE_PATTERN, '', item['name']) - url = item.get('url') - - if url: - text += f"{name}" + if test_url: + text += f"{test_name}" else: - text += name + text += test_name text += " — " - text += get_human_timedelta(item["time"]) - if ADD_CALENDAR_LINK: - text += f"\n(" - text += get_human_time(item["time"]) + ")\n\n" + text += await get_human_timedelta(tests[i]["time"]) + text += f"\n(" + text += await get_human_time(tests[i]["time"]) + ")\n\n" + + if len(lectures) > 0: + text += f"\n👨‍🏫 Лекции:\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) + "" - for assignment_type in assignments: - add_items(*assignment_type) + if lecture_url: + text += f"{lecture_name}" + else: + text += lecture_name - text += ( - f"\n🆕 " - f"Добавить дедлайн" - ) + text += " — " + text += await get_human_timedelta(lectures[i]["time"]) + text += f"\n(" + text += await get_human_time(lectures[i]["time"]) + ")\n\n" + text += f"\n🆕 " \ + f"Добавить дедлайн" return text +async def send_deadlines(chat_id: int) -> None: + text = await get_message_text() + if text == "Дедлайнов нет)\n\n": + return -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) + msg = await bot.send_message(chat_id, text, parse_mode="HTML", disable_web_page_preview=True) 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 condition(): - time.sleep(60) + while dt.datetime.now() - started_updating < dt.timedelta(days=1): + await asyncio.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()) diff --git a/requirements.txt b/requirements.txt index 13b630a..1d84bca 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,21 @@ -certifi==2025.4.26 -charset-normalizer==3.4.2 +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 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