chore: remove aiogram as a dependency

This commit is contained in:
Arthur K. 2025-05-21 16:36:31 +03:00
parent 331008aa14
commit bbcfa2a601
Signed by: wzray
GPG key ID: B97F30FDC4636357
2 changed files with 69 additions and 43 deletions

90
main.py
View file

@ -1,15 +1,12 @@
import asyncio
import datetime as dt import datetime as dt
import locale import locale
import logging import logging
import os import os
import re import re
import time
import urllib.parse import urllib.parse
import requests import requests
from aiogram import Bot
from aiogram.client.default import DefaultBotProperties
from aiogram.types import LinkPreviewOptions
# Modify the links and data below: # Modify the links and data below:
DEADLINES_URL = "https://m3104.nawinds.dev/DEADLINES.json" DEADLINES_URL = "https://m3104.nawinds.dev/DEADLINES.json"
@ -18,6 +15,7 @@ BOT_NAME = "Дединсайдер M3104"
BOT_USERNAME = "m3104_deadliner_bot" BOT_USERNAME = "m3104_deadliner_bot"
# Environment variables that should be available: # Environment variables that should be available:
API_URL = 'https://api.telegram.org/bot'
TOKEN = os.getenv("TOKEN") TOKEN = os.getenv("TOKEN")
MAIN_GROUP_ID = int(os.getenv("MAIN_GROUP_ID") or '0') MAIN_GROUP_ID = int(os.getenv("MAIN_GROUP_ID") or '0')
EDIT_MESSAGE_ID = int(os.getenv("EDIT_MESSAGE_ID") or '0') EDIT_MESSAGE_ID = int(os.getenv("EDIT_MESSAGE_ID") or '0')
@ -28,21 +26,66 @@ assert MAIN_GROUP_ID, "Missing group ID!"
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
bot = Bot(TOKEN, default=DefaultBotProperties(parse_mode='HTML', link_preview=LinkPreviewOptions(is_disabled=True)))
NUMBER_EMOJIS = ['0.', '1', '2', '3', '4', '5', '6', '7', '8', '9', '🔟'] 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: def get_current_time() -> str:
current_time = dt.datetime.now() current_time = dt.datetime.now()
current_time_hour = current_time.hour if current_time.hour >= 10 else "0" + str(current_time.hour) 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) 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}" return f"{current_time_hour}:{current_time_minute}"
def get_dt_obj_from_string(time: str) -> dt.datetime: def get_dt_obj_from_string(time: str) -> dt.datetime:
time = time.replace('GMT+3', '+0300') time = time.replace('GMT+3', '+0300')
locale.setlocale(locale.LC_TIME, 'en_US.UTF-8') locale.setlocale(locale.LC_TIME, 'en_US.UTF-8')
return dt.datetime.strptime(time, "%d %b %Y %H:%M:%S %z") return dt.datetime.strptime(time, "%d %b %Y %H:%M:%S %z")
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) dt_obj = get_dt_obj_from_string(event_time)
formatted_time = dt_obj.strftime("%Y%m%d T%H%M%S%z") formatted_time = dt_obj.strftime("%Y%m%d T%H%M%S%z")
@ -53,6 +96,7 @@ def generate_link(event_name: str, event_time: str) -> str:
f"color=6" f"color=6"
return link return link
def get_human_timedelta(time: str) -> str: def get_human_timedelta(time: str) -> str:
dt_obj = get_dt_obj_from_string(time) dt_obj = get_dt_obj_from_string(time)
dt_now = dt.datetime.now(dt_obj.tzinfo) # Ensure timezones are consistent dt_now = dt.datetime.now(dt_obj.tzinfo) # Ensure timezones are consistent
@ -72,28 +116,33 @@ def get_human_timedelta(time: str) -> str:
else: else:
return f"{hours}ч {minutes}м" return f"{hours}ч {minutes}м"
def get_human_time(time: str) -> str: def get_human_time(time: str) -> str:
dt_obj = get_dt_obj_from_string(time) dt_obj = get_dt_obj_from_string(time)
locale.setlocale(locale.LC_TIME, 'ru_RU.UTF-8') locale.setlocale(locale.LC_TIME, 'ru_RU.UTF-8')
formatted_date = dt_obj.strftime("%a, %d %B в %H:%M") formatted_date = dt_obj.strftime("%a, %d %B в %H:%M")
return formatted_date return formatted_date
def timestamp_func(a: dict) -> float: def timestamp_func(a: dict) -> float:
time = a["time"].replace('GMT+3', '+0300') time = a["time"].replace('GMT+3', '+0300')
locale.setlocale(locale.LC_TIME, 'en_US.UTF-8') locale.setlocale(locale.LC_TIME, 'en_US.UTF-8')
a_timestamp = dt.datetime.strptime(time, "%d %b %Y %H:%M:%S %z").timestamp() 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 return a_timestamp # 29 Oct 2024 23:59:59 GMT+3
def relevant_filter_func(d: dict) -> bool: def relevant_filter_func(d: dict) -> bool:
dt_obj = get_dt_obj_from_string(d["time"]) dt_obj = get_dt_obj_from_string(d["time"])
return not dt_obj < dt.datetime.now(dt_obj.tzinfo) return not dt_obj < dt.datetime.now(dt_obj.tzinfo)
def deadline_type_filter_func(d: dict, dtype: str = '') -> bool: def deadline_type_filter_func(d: dict, dtype: str = '') -> bool:
if not dtype: if not dtype:
return not re.match(r'^\[.*\]', d['name']) return not re.match(r'^\[.*\]', d['name'])
return f"[{dtype.lower()}]" in d["name"].lower() return f"[{dtype.lower()}]" in d["name"].lower()
def get_message_text() -> str: def get_message_text() -> str:
try: try:
response = requests.get(DEADLINES_URL).json() response = requests.get(DEADLINES_URL).json()
@ -162,43 +211,36 @@ def get_message_text() -> str:
return text return text
async def send_deadlines(chat_id: int) -> None:
def main() -> None:
text = get_message_text() text = get_message_text()
if EDIT_MESSAGE_ID: if EDIT_MESSAGE_ID:
msg = await bot.edit_message_text(text, chat_id=chat_id, message_id=EDIT_MESSAGE_ID) msg_id = edit_message(EDIT_MESSAGE_ID, text)
assert not isinstance(msg, bool), 'EDIT_MESSAGE_ID is not a message!'
else: else:
msg = await bot.send_message(chat_id, text, parse_mode="HTML", disable_web_page_preview=True) msg_id = send_message(text)
started_updating = dt.datetime.now() started_updating = dt.datetime.now()
print(dt.datetime.now(), "Message sent. Msg id:", msg.message_id) print(dt.datetime.now(), "Message sent. Msg id:", msg_id)
condition = (lambda: True) if EDIT_MESSAGE_ID else (lambda: dt.datetime.now() - started_updating < dt.timedelta(days=1)) condition = (lambda: True) if EDIT_MESSAGE_ID else (lambda: dt.datetime.now() - started_updating < dt.timedelta(days=1))
while condition(): while condition():
await asyncio.sleep(60) time.sleep(60)
try: try:
new_text = get_message_text() new_text = get_message_text()
if text != new_text and new_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 text = new_text
print(dt.datetime.now(), "Message updated. Msg id:", msg.message_id) print(dt.datetime.now(), "Message updated. Msg id:", msg_id)
else: else:
print(dt.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: except Exception as e:
logging.warning(dt.datetime.now(),f"{dt.datetime.now()} Error updating message: {e}") logging.warning(dt.datetime.now(),f"{dt.datetime.now()} Error updating message: {e}")
continue continue
if not EDIT_MESSAGE_ID: if not EDIT_MESSAGE_ID:
await msg.delete() delete_message(msg_id)
async def main():
try:
await send_deadlines(MAIN_GROUP_ID)
finally:
await bot.session.close()
if __name__ == '__main__': if __name__ == '__main__':
asyncio.run(main()) main()

View file

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