diff --git a/dtek_notif/main.py b/dtek_notif/main.py index 3896473..2f82980 100644 --- a/dtek_notif/main.py +++ b/dtek_notif/main.py @@ -1,106 +1,100 @@ -import requests -import json import asyncio -from bs4 import BeautifulSoup -from datetime import datetime, timedelta -from aiogram import Bot, Dispatcher, types -from aiogram.filters import Command -from aiogram.types import Message, KeyboardButton -from aiogram.utils.keyboard import ReplyKeyboardBuilder +import contextlib import logging import os +from datetime import datetime, timedelta + +import requests +from aiogram import Bot, Dispatcher +from aiogram.filters import Command +from aiogram.types import KeyboardButton, Message +from aiogram.utils.keyboard import ReplyKeyboardBuilder +from bs4 import BeautifulSoup from dotenv import load_dotenv -from typing import Optional, List, Dict # Загрузка переменных окружения load_dotenv() # Настройки -TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "YOUR_TOKEN_HERE") -ALLOWED_CHAT_IDS = list(map(int, os.getenv("ALLOWED_CHAT_IDS", "").split(","))) if os.getenv("ALLOWED_CHAT_IDS") else [] -CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "120")) +TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN', 'YOUR_TOKEN_HERE') +ALLOWED_CHAT_IDS = list(map(int, os.getenv('ALLOWED_CHAT_IDS', '').split(','))) if os.getenv('ALLOWED_CHAT_IDS') else [] +CHECK_INTERVAL = int(os.getenv('CHECK_INTERVAL', '120')) # Параметры для запроса -VOE_CITY_ID = int(os.getenv("VOE_CITY_ID", "VOE_CITY_ID")) -VOE_STREET_ID = int(os.getenv("VOE_STREET_ID", "VOE_STREET_ID")) -VOE_HOUSE_ID = int(os.getenv("VOE_HOUSE_ID", "VOE_HOUSE_ID")) +VOE_CITY_ID = int(os.getenv('VOE_CITY_ID', 'VOE_CITY_ID')) +VOE_STREET_ID = int(os.getenv('VOE_STREET_ID', 'VOE_STREET_ID')) +VOE_HOUSE_ID = int(os.getenv('VOE_HOUSE_ID', 'VOE_HOUSE_ID')) # Настройка логирования -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' -) +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Глобальные переменные bot = Bot(token=TELEGRAM_TOKEN) dp = Dispatcher() -last_schedule: Optional[List[Dict]] = None -last_notification_time: Dict[str, datetime] = {} +last_schedule: list[dict] | None = None +last_notification_time: dict[str, datetime] = {} # ============================================================================ # УТИЛИТЫ # ============================================================================ + def format_time_duration(minutes: int) -> str: """Форматирует время из минут в часы и минуты""" hours = minutes // 60 mins = minutes % 60 - + if hours == 0: - return f"{mins}м" + return f'{mins}м' elif mins == 0: - return f"{hours}ч" - return f"{hours}ч {mins}м" + return f'{hours}ч' + return f'{hours}ч {mins}м' -def get_day_statistics(day_blocks: List[Dict]) -> Dict[str, int]: +def get_day_statistics(day_blocks: list[dict]) -> dict[str, int]: """Получает статистику по дню""" total_minutes = 0 confirmed_minutes = 0 possible_minutes = 0 - + for block in day_blocks: - for half in [block["first_half"], block["second_half"]]: - if half["status"] == "off": + for half in [block['first_half'], block['second_half']]: + if half['status'] == 'off': total_minutes += 30 - if half["confirmed"]: + if half['confirmed']: confirmed_minutes += 30 else: possible_minutes += 30 - - return { - "total": total_minutes, - "confirmed": confirmed_minutes, - "possible": possible_minutes - } + + return {'total': total_minutes, 'confirmed': confirmed_minutes, 'possible': possible_minutes} # ============================================================================ # ПАРСИНГ ДАННЫХ # ============================================================================ -def parse_html(html: str) -> List[Dict]: + +def parse_html(html: str) -> list[dict]: """Парсит HTML с графиком отключений (логика от 15.11.2024)""" - soup = BeautifulSoup(html, "html.parser") - cells = soup.select(".disconnection-detailed-table-cell.cell") - + soup = BeautifulSoup(html, 'html.parser') + cells = soup.select('.disconnection-detailed-table-cell.cell') + schedule = [] current_hour = 0 current_day = 0 - + for cell in cells: if 'legend' in cell.get('class', []) or 'head' in cell.get('class', []): continue - + cell_classes = cell.get('class', []) - - # ПРоверка статуса отключения на весь час + + # ПРоверка статуса отключения на весь час full_hour_off = 'has_disconnection' in cell_classes and 'full_hour' in cell_classes - no_disconnection_cell = 'no_disconnection' in cell_classes - - hour_block = cell.select_one(".hour_block") + + hour_block = cell.select_one('.hour_block') if not hour_block: continue @@ -110,109 +104,106 @@ def parse_html(html: str) -> List[Dict]: cell_confirmed = True elif 'confirm_0' in cell_classes: cell_confirmed = False - + # Проверка половин часа - left = hour_block.select_one(".half.left") - right = hour_block.select_one(".half.right") - - def parse_half(half, is_full_hour_off: bool) -> Dict: + left = hour_block.select_one('.half.left') + right = hour_block.select_one('.half.right') + + def parse_half(half, is_full_hour_off: bool, cell_confirmed: bool | None = None) -> dict: """Парсит половину часа""" if not half: - return {"status": "on", "queue": None, "confirmed": None} - + return {'status': 'on', 'queue': None, 'confirmed': None} + half_classes = half.get('class', []) - + # Если вся ячейка full_hour - используем статус ячейки if is_full_hour_off: - return {"status": "off", "queue": None, "confirmed": cell_confirmed} - + return {'status': 'off', 'queue': None, 'confirmed': cell_confirmed} + # Определяем статус половины if 'has_disconnection' in half_classes: - status = "off" + status = 'off' elif 'no_disconnection' in half_classes: - status = "on" + status = 'on' else: - status = "on" # По умолчанию считаем включенным - + status = 'on' # По умолчанию считаем включенным + # Если выключено - ищем подробности queue = None confirmed = None - - if status == "off": - disconnection_div = half.select_one(".disconnection") + + if status == 'off': + disconnection_div = half.select_one('.disconnection') if disconnection_div: # Ищем номер черги в title - if disconnection_div.has_attr("title"): - title = disconnection_div["title"] - if "Номер черги" in title or "Номер черги:" in title: - try: - queue = title.split(":")[-1].strip() - except: - pass - + if disconnection_div.has_attr('title'): + title = disconnection_div['title'] + if 'Номер черги' in title or 'Номер черги:' in title: + with contextlib.suppress(BaseException): + queue = title.split(':')[-1].strip() + # Определяем подтверждение disc_classes = disconnection_div.get('class', []) if 'disconnection_confirm_1' in disc_classes: confirmed = True elif 'disconnection_confirm_0' in disc_classes: confirmed = False - - return {"status": status, "queue": queue, "confirmed": confirmed} - - first_half_data = parse_half(left, full_hour_off) - second_half_data = parse_half(right, full_hour_off) - - schedule.append({ - "hour": current_hour, - "day": current_day, - "first_half": first_half_data, - "second_half": second_half_data, - }) - + + return {'status': status, 'queue': queue, 'confirmed': confirmed} + + first_half_data = parse_half(left, full_hour_off, cell_confirmed) + second_half_data = parse_half(right, full_hour_off, cell_confirmed) + + schedule.append( + { + 'hour': current_hour, + 'day': current_day, + 'first_half': first_half_data, + 'second_half': second_half_data, + } + ) + current_hour += 1 if current_hour >= 24: current_hour = 0 current_day += 1 - + return schedule def get_voe_html(city_id: int, street_id: int, house_id: int) -> str: """Получает HTML с сайта VOE""" - url = "https://www.voe.com.ua/disconnection/detailed?ajax_form=1&_wrapper_format=drupal_ajax" + url = 'https://www.voe.com.ua/disconnection/detailed?ajax_form=1&_wrapper_format=drupal_ajax' headers = { - "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", - "X-Requested-With": "XMLHttpRequest", - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', + 'X-Requested-With': 'XMLHttpRequest', + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', } data = { - "search_type": 0, - "city_id": city_id, - "street_id": street_id, - "house_id": house_id, - "form_build_id": "form-Irv5aHw1R2FT_Ik2apyHOZ47hTH5xPNH_LQnBrmpSTc", - "form_id": "disconnection_detailed_search_form", - "_triggering_element_name": "search", - "_triggering_element_value": "Показати", - "_drupal_ajax": 1, + 'search_type': 0, + 'city_id': city_id, + 'street_id': street_id, + 'house_id': house_id, + 'form_build_id': 'form-Irv5aHw1R2FT_Ik2apyHOZ47hTH5xPNH_LQnBrmpSTc', + 'form_id': 'disconnection_detailed_search_form', + '_triggering_element_name': 'search', + '_triggering_element_value': 'Показати', + '_drupal_ajax': 1, } - + try: response = requests.post(url, headers=headers, data=data, timeout=10) response.raise_for_status() resp_json = response.json() - - insert_html = next( - (item["data"] for item in resp_json if item.get("command") == "insert"), - None - ) - + + insert_html = next((item['data'] for item in resp_json if item.get('command') == 'insert'), None) + if not insert_html: - raise ValueError("HTML не найден в ответе") - + raise ValueError('HTML не найден в ответе') + return insert_html except requests.exceptions.RequestException as e: - logger.error(f"Ошибка запроса VOE: {e}") + logger.error(f'Ошибка запроса VOE: {e}') raise @@ -220,198 +211,189 @@ def get_voe_html(city_id: int, street_id: int, house_id: int) -> str: # ФОРМАТИРОВАНИЕ СООБЩЕНИЙ # ============================================================================ + def get_main_keyboard(): """Создает главную клавиатуру""" builder = ReplyKeyboardBuilder() - builder.row( - KeyboardButton(text="📊 Графік"), - KeyboardButton(text="🔄 Оновити") - ) - builder.row( - KeyboardButton(text="📅 Сьогодні"), - KeyboardButton(text="📅 Завтра") - ) - builder.row(KeyboardButton(text="ℹ️ Про бота")) + builder.row(KeyboardButton(text='📊 Графік'), KeyboardButton(text='🔄 Оновити')) + builder.row(KeyboardButton(text='📅 Сьогодні'), KeyboardButton(text='📅 Завтра')) + builder.row(KeyboardButton(text='ℹ️ Про бота')) return builder.as_markup(resize_keyboard=True) -def format_schedule_message(schedule: List[Dict], days_to_show: int = 2) -> str: +def format_schedule_message(schedule: list[dict], days_to_show: int = 2) -> str: """Форматирует полный график на несколько дней""" lines = [ - "⚡️ Графік відключень світла", - f"🕐 Оновлено: {datetime.now().strftime('%d.%m.%Y %H:%M:%S')}", - "─" * 30, - "" + '⚡️ Графік відключень світла', + f'🕐 Оновлено: {datetime.now().strftime("%d.%m.%Y %H:%M:%S")}', + '─' * 30, + '', ] - + start_date = datetime.now() - + for day in range(min(days_to_show, 2)): - day_blocks = [b for b in schedule if b["day"] == day] + day_blocks = [b for b in schedule if b['day'] == day] if not day_blocks: continue - + date_str = (start_date + timedelta(days=day)).strftime('%d.%m.%Y') - day_name = "🌅 Сьогодні" if day == 0 else "🌄 Завтра" - - lines.append(f"{day_name} ({date_str})") - + day_name = '🌅 Сьогодні' if day == 0 else '🌄 Завтра' + + lines.append(f'{day_name} ({date_str})') + # Статистика stats = get_day_statistics(day_blocks) - if stats["total"] > 0: - lines.append(f"⏱ Всього: {format_time_duration(stats['total'])}") - if stats["confirmed"] > 0: - lines.append(f"🔴 Підтверджено: {format_time_duration(stats['confirmed'])}") - if stats["possible"] > 0: - lines.append(f"🟠 Можливо: {format_time_duration(stats['possible'])}") + if stats['total'] > 0: + lines.append(f'⏱ Всього: {format_time_duration(stats["total"])}') + if stats['confirmed'] > 0: + lines.append(f'🔴 Підтверджено: {format_time_duration(stats["confirmed"])}') + if stats['possible'] > 0: + lines.append(f'🟠 Можливо: {format_time_duration(stats["possible"])}') else: - lines.append("🟢 Відключень немає!") - - lines.append("") - + lines.append('🟢 Відключень немає!') + + lines.append('') + # Детальный список отключений disconnections = [] current_status = None start_time = None current_confirmed = None current_queue = None - + for block in day_blocks: - hour = block["hour"] - - for half_idx, half in enumerate([block["first_half"], block["second_half"]]): - time_str = f"{hour:02d}:00" if half_idx == 0 else f"{hour:02d}:30" - - if half["status"] == "off": - if current_status != "off": + hour = block['hour'] + + for half_idx, half in enumerate([block['first_half'], block['second_half']]): + time_str = f'{hour:02d}:00' if half_idx == 0 else f'{hour:02d}:30' + + if half['status'] == 'off': + if current_status != 'off': start_time = time_str - current_confirmed = half["confirmed"] - current_queue = half["queue"] - current_status = "off" + current_confirmed = half['confirmed'] + current_queue = half['queue'] + current_status = 'off' else: - if current_status == "off": - icon = "🔴" if current_confirmed else "🟠" - queue_text = f" (Ч{current_queue})" if current_queue else "" - disconnections.append(f"{icon} {start_time} - {time_str}{queue_text}") - current_status = half["status"] - + if current_status == 'off': + icon = '🔴' if current_confirmed else '🟠' + queue_text = f' (Ч{current_queue})' if current_queue else '' + disconnections.append(f'{icon} {start_time} - {time_str}{queue_text}') + current_status = half['status'] + # Если день закончился на отключении - if current_status == "off": - icon = "🔴" if current_confirmed else "🟠" - queue_text = f" (Ч{current_queue})" if current_queue else "" - next_hour = (day_blocks[-1]["hour"] + 1) % 24 - end_time = f"{next_hour:02d}:00" - disconnections.append(f"{icon} {start_time} - {end_time}{queue_text}") - + if current_status == 'off': + icon = '🔴' if current_confirmed else '🟠' + queue_text = f' (Ч{current_queue})' if current_queue else '' + next_hour = (day_blocks[-1]['hour'] + 1) % 24 + end_time = f'{next_hour:02d}:00' + disconnections.append(f'{icon} {start_time} - {end_time}{queue_text}') + if disconnections: for idx, disc in enumerate(disconnections, 1): - lines.append(f"{idx}. {disc}") - - lines.append("") - - lines.append("🔴 = підтверджено • 🟠 = можливо • 🟢 = світло") - - return "\n".join(lines) + lines.append(f'{idx}. {disc}') + + lines.append('') + + lines.append('🔴 = підтверджено • 🟠 = можливо • 🟢 = світло') + + return '\n'.join(lines) -def format_single_day_schedule(schedule: List[Dict], day: int) -> str: +def format_single_day_schedule(schedule: list[dict], day: int) -> str: """Форматирует график на один день""" - day_blocks = [b for b in schedule if b["day"] == day] + day_blocks = [b for b in schedule if b['day'] == day] if not day_blocks: - return "❌ Немає даних для цього дня" - + return '❌ Немає даних для цього дня' + start_date = datetime.now() date_str = (start_date + timedelta(days=day)).strftime('%d.%m.%Y') - day_name = "🟠 Сьогодні" if day == 0 else "🔶 Завтра" - - lines = [ - f"{day_name} • {date_str}", - "" - ] - + day_name = '🟠 Сьогодні' if day == 0 else '🔶 Завтра' + + lines = [f'{day_name} • {date_str}', ''] + # Статистика - lines.append("📊 Статистика") + lines.append('📊 Статистика') stats = get_day_statistics(day_blocks) - - if stats["total"] == 0: - lines.append("└ 🟢 Відключень немає!") + + if stats['total'] == 0: + lines.append('└ 🟢 Відключень немає!') else: - total_time = format_time_duration(stats["total"]) - lines.append(f"├ ⏱ Всього: {total_time}") - - if stats["confirmed"] > 0: - confirmed_time = format_time_duration(stats["confirmed"]) - lines.append(f"├ 🔴 Підтверджено: {confirmed_time}") - - if stats["possible"] > 0: - possible_time = format_time_duration(stats["possible"]) - lines.append(f"└ 🟠 Можливо: {possible_time}") + total_time = format_time_duration(stats['total']) + lines.append(f'├ ⏱ Всього: {total_time}') + + if stats['confirmed'] > 0: + confirmed_time = format_time_duration(stats['confirmed']) + lines.append(f'├ 🔴 Підтверджено: {confirmed_time}') + + if stats['possible'] > 0: + possible_time = format_time_duration(stats['possible']) + lines.append(f'└ 🟠 Можливо: {possible_time}') else: - lines.append(f"└ 🟢 Решта часу світло") - - lines.append("") - + lines.append('└ 🟢 Решта часу світло') + + lines.append('') + # Детальный список отключений disconnections = [] current_status = None start_time = None current_confirmed = None current_queue = None - + for block in day_blocks: - hour = block["hour"] - - for half_idx, half in enumerate([block["first_half"], block["second_half"]]): - time_str = f"{hour:02d}:00" if half_idx == 0 else f"{hour:02d}:30" - - if half["status"] == "off": - if current_status != "off": + hour = block['hour'] + + for half_idx, half in enumerate([block['first_half'], block['second_half']]): + time_str = f'{hour:02d}:00' if half_idx == 0 else f'{hour:02d}:30' + + if half['status'] == 'off': + if current_status != 'off': start_time = time_str - current_confirmed = half["confirmed"] - current_queue = half["queue"] - current_status = "off" + current_confirmed = half['confirmed'] + current_queue = half['queue'] + current_status = 'off' else: - if current_status == "off": - icon = "🔴" if current_confirmed else "🟠" - queue_text = f" (Ч.{current_queue})" if current_queue else "" - disconnections.append(f"{icon} {start_time} - {time_str}{queue_text}") - current_status = half["status"] - + if current_status == 'off': + icon = '🔴' if current_confirmed else '🟠' + queue_text = f' (Ч.{current_queue})' if current_queue else '' + disconnections.append(f'{icon} {start_time} - {time_str}{queue_text}') + current_status = half['status'] + # Если день закончился на отключении - if current_status == "off": - icon = "🔴" if current_confirmed else "🟠" - queue_text = f" (Ч.{current_queue})" if current_queue else "" - next_hour = (day_blocks[-1]["hour"] + 1) % 24 - end_time = f"{next_hour:02d}:00" - disconnections.append(f"{icon} {start_time} - {end_time}{queue_text}") - + if current_status == 'off': + icon = '🔴' if current_confirmed else '🟠' + queue_text = f' (Ч.{current_queue})' if current_queue else '' + next_hour = (day_blocks[-1]['hour'] + 1) % 24 + end_time = f'{next_hour:02d}:00' + disconnections.append(f'{icon} {start_time} - {end_time}{queue_text}') + if disconnections: - lines.append("⚡️ Розклад відключень") + lines.append('⚡️ Розклад відключень') for idx, disc in enumerate(disconnections, 1): - lines.append(f"{idx}. {disc}") - - lines.append("") - lines.append("🔴 підтверджено • 🟠 можливо • 🟢 світло") - - return "\n".join(lines) + lines.append(f'{idx}. {disc}') + + lines.append('') + lines.append('🔴 підтверджено • 🟠 можливо • 🟢 світло') + + return '\n'.join(lines) -def schedules_differ(old_schedule: Optional[List[Dict]], new_schedule: Optional[List[Dict]]) -> bool: +def schedules_differ(old_schedule: list[dict] | None, new_schedule: list[dict] | None) -> bool: """Проверяет отличия между графиками""" if old_schedule is None or new_schedule is None: return True - + if len(old_schedule) != len(new_schedule): return True - - for old, new in zip(old_schedule, new_schedule): - if old["day"] >= 2: + + for old, new in zip(old_schedule, new_schedule, strict=False): + if old['day'] >= 2: break - - if (old["first_half"] != new["first_half"] or - old["second_half"] != new["second_half"]): + + if old['first_half'] != new['first_half'] or old['second_half'] != new['second_half']: return True - + return False @@ -419,94 +401,92 @@ def schedules_differ(old_schedule: Optional[List[Dict]], new_schedule: Optional[ # УВЕДОМЛЕНИЯ # ============================================================================ -async def send_to_all_users(message_text: str, parse_mode: str = "HTML"): + +async def send_to_all_users(message_text: str, parse_mode: str = 'HTML'): """Отправляет сообщение всем пользователям""" if not ALLOWED_CHAT_IDS: - logger.warning("Нет допущенных ID чатов для отправки уведомлений") + logger.warning('Нет допущенных ID чатов для отправки уведомлений') return - + for chat_id in ALLOWED_CHAT_IDS: try: await bot.send_message(chat_id, message_text, parse_mode=parse_mode) - logger.info(f"✅ Сообщение отправлено пользователю {chat_id}") + logger.info(f'✅ Сообщение отправлено пользователю {chat_id}') except Exception as e: - logger.error(f"❌ Ошибка отправки пользователю {chat_id}: {e}") + logger.error(f'❌ Ошибка отправки пользователю {chat_id}: {e}') await asyncio.sleep(0.5) async def check_schedule(): """Проверяет график и отправляет уведомления""" global last_schedule - + try: - logger.info("🔍 Проверка графика...") + logger.info('🔍 Проверка графика...') html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID) new_schedule = parse_html(html) - + if schedules_differ(last_schedule, new_schedule): - logger.info("✨ Обнаружены изменения!") + logger.info('✨ Обнаружены изменения!') message = format_schedule_message(new_schedule, days_to_show=2) - + if last_schedule is not None: - await send_to_all_users(f"🔄 Графік оновлено!\n\n{message}") - + await send_to_all_users(f'🔄 Графік оновлено!\n\n{message}') + last_schedule = new_schedule else: - logger.info("✓ Графік без змін") - + logger.info('✓ Графік без змін') + except Exception as e: - logger.error(f"❌ Ошибка при проверке графика: {e}") + logger.error(f'❌ Ошибка при проверке графика: {e}') async def check_upcoming_disconnections(): """Проверяет предстоящие события и отправляет предупреждения за 5 минут""" global last_notification_time - + if last_schedule is None: return - + now = datetime.now() - today_blocks = [b for b in last_schedule if b["day"] == 0] - + today_blocks = [b for b in last_schedule if b['day'] == 0] + # Создаем список всех переходов (off -> on или on -> off) transitions = [] prev_status = None - + for block in today_blocks: - hour = block["hour"] - - for half_idx, half in enumerate([block["first_half"], block["second_half"]]): + hour = block['hour'] + + for half_idx, half in enumerate([block['first_half'], block['second_half']]): minute = 0 if half_idx == 0 else 30 - time_str = f"{hour:02d}:{minute:02d}" - - current_status = half["status"] - + time_str = f'{hour:02d}:{minute:02d}' + + current_status = half['status'] + # Если статус изменился - это переход if prev_status is not None and prev_status != current_status: - transitions.append({ - "hour": hour, - "minute": minute, - "time_str": time_str, - "from_status": prev_status, - "to_status": current_status, - "confirmed": half.get("confirmed"), - "queue": half.get("queue") - }) - + transitions.append( + { + 'hour': hour, + 'minute': minute, + 'time_str': time_str, + 'from_status': prev_status, + 'to_status': current_status, + 'confirmed': half.get('confirmed'), + 'queue': half.get('queue'), + } + ) + prev_status = current_status - + # Проверяем переходы for transition in transitions: - event_time = now.replace( - hour=transition["hour"], - minute=transition["minute"], - second=0, - microsecond=0 - ) - + event_time = now.replace(hour=transition['hour'], minute=transition['minute'], second=0, microsecond=0) + time_until = (event_time - now).total_seconds() / 60 - notification_key = f"{transition['hour']}:{transition['minute']}_{transition['to_status']}" - + notification_key = f'{transition["hour"]}:{transition["minute"]}_{transition["to_status"]}' + # Если за 5 минут до события (±1 минута) и еще не отправляли if 4 <= time_until <= 6: # Проверяем, не отправляли ли уже уведомление сегодня @@ -514,48 +494,48 @@ async def check_upcoming_disconnections(): last_notif_time = last_notification_time[notification_key] if last_notif_time.date() == now.date(): continue # Уже отправляли сегодня - + # Переход на ОТКЛЮЧЕНИЕ (on -> off) - if transition["from_status"] == "on" and transition["to_status"] == "off": - icon = "🔴" if transition["confirmed"] else "🟠" - status = "підтверджено" if transition["confirmed"] else "можливе" - queue_info = f" (Черга {transition['queue']})" if transition['queue'] else "" - + if transition['from_status'] == 'on' and transition['to_status'] == 'off': + icon = '🔴' if transition['confirmed'] else '🟠' + status = 'підтверджено' if transition['confirmed'] else 'можливе' + queue_info = f' (Черга {transition["queue"]})' if transition['queue'] else '' + warning = ( - f"⚠️ УВАГА! ВІДКЛЮЧЕННЯ\n\n" - f"Через ~5 хвилин\n" - f"Час: {transition['time_str']}\n" - f"Статус: {icon} {status}{queue_info}" + f'⚠️ УВАГА! ВІДКЛЮЧЕННЯ\n\n' + f'Через ~5 хвилин\n' + f'Час: {transition["time_str"]}\n' + f'Статус: {icon} {status}{queue_info}' ) - + await send_to_all_users(warning) last_notification_time[notification_key] = now - logger.info(f"📢 Відправлено попередження про ВІДКЛЮЧЕННЯ в {transition['time_str']}") - + logger.info(f'📢 Відправлено попередження про ВІДКЛЮЧЕННЯ в {transition["time_str"]}') + # Переход на ВКЛЮЧЕНИЕ (off -> on) - elif transition["from_status"] == "off" and transition["to_status"] == "on": + elif transition['from_status'] == 'off' and transition['to_status'] == 'on': warning = ( - f"✅ УВАГА! ВКЛЮЧЕННЯ\n\n" - f"Через ~5 хвилин буде світло\n" - f"Час: {transition['time_str']}" + f'✅ УВАГА! ВКЛЮЧЕННЯ\n\n' + f'Через ~5 хвилин буде світло\n' + f'Час: {transition["time_str"]}' ) - + await send_to_all_users(warning) last_notification_time[notification_key] = now - logger.info(f"📢 Відправлено попередження про ВКЛЮЧЕННЯ в {transition['time_str']}") + logger.info(f'📢 Відправлено попередження про ВКЛЮЧЕННЯ в {transition["time_str"]}') async def monitoring_loop(): """Основной цикл мониторинга""" await check_schedule() - + while True: try: await asyncio.sleep(CHECK_INTERVAL) await check_schedule() await check_upcoming_disconnections() except Exception as e: - logger.error(f"Ошибка в цикле мониторинга: {e}") + logger.error(f'Ошибка в цикле мониторинга: {e}') await asyncio.sleep(5) @@ -563,138 +543,127 @@ async def monitoring_loop(): # ОБРАБОТЧИКИ КОМАНД # ============================================================================ -@dp.message(Command("start")) + +@dp.message(Command('start')) async def cmd_start(message: Message): """Обработчик /start""" if message.chat.id not in ALLOWED_CHAT_IDS: - await message.answer("❌ У вас немає доступу до цього бота.") + await message.answer('❌ У вас немає доступу до цього бота.') return - + await message.answer( - "👋 Ласкаво просимо!\n\n" - "🤖 Бот для моніторингу графіку відключень світла\n\n" - "✨ Можливості:\n" - "• 📊 Перегляд графіку на сьогодні і завтра\n" - "• 🔔 Автоматичні сповіщення за 5 хвилин до подій\n" - "• 🔄 Моніторинг змін графіку\n\n" - "Використовуйте кнопки нижче 👇", - parse_mode="HTML", - reply_markup=get_main_keyboard() + '👋 Ласкаво просимо!\n\n' + '🤖 Бот для моніторингу графіку відключень світла\n\n' + '✨ Можливості:\n' + '• 📊 Перегляд графіку на сьогодні і завтра\n' + '• 🔔 Автоматичні сповіщення за 5 хвилин до подій\n' + '• 🔄 Моніторинг змін графіку\n\n' + 'Використовуйте кнопки нижче 👇', + parse_mode='HTML', + reply_markup=get_main_keyboard(), ) -@dp.message(lambda msg: msg.text == "ℹ️ Про бота") +@dp.message(lambda msg: msg.text == 'ℹ️ Про бота') async def cmd_info(message: Message): """Показывает информацию о боте""" if message.chat.id not in ALLOWED_CHAT_IDS: return - + await message.answer( - "ℹ️ Про бота\n\n" - "🚀 Версія: 2.2 (Стабільна)\n\n" - "📝 Реліз-ноути:\n" - "├ 15.11.2024: Адаптація під оновлену логіку сайту VOE\n" - "├ Виправлено парсинг half.left та half.right\n" - "├ Покращено визначення підтвердження відключень\n" - "└ Оптимізовано обробку статусу для всієї години\n\n" - "⚡ Функціональність:\n" - "├ Моніторинг графіку 24/7\n" - "├ Сповіщення за 5 хвилин\n" - "├ Детальна статистика дня\n" - "└ Красива візуалізація\n\n" - "🔐 Безпека: Використовуються .env файли\n" - "💾 Джерело: voe.com.ua", - parse_mode="HTML", - reply_markup=get_main_keyboard() + 'ℹ️ Про бота\n\n' + '🚀 Версія: 2.2 (Стабільна)\n\n' + '📝 Реліз-ноути:\n' + '├ 15.11.2024: Адаптація під оновлену логіку сайту VOE\n' + '├ Виправлено парсинг half.left та half.right\n' + '├ Покращено визначення підтвердження відключень\n' + '└ Оптимізовано обробку статусу для всієї години\n\n' + '⚡ Функціональність:\n' + '├ Моніторинг графіку 24/7\n' + '├ Сповіщення за 5 хвилин\n' + '├ Детальна статистика дня\n' + '└ Красива візуалізація\n\n' + '🔐 Безпека: Використовуються .env файли\n' + '💾 Джерело: voe.com.ua', + parse_mode='HTML', + reply_markup=get_main_keyboard(), ) -@dp.message(Command("schedule")) +@dp.message(Command('schedule')) async def cmd_schedule(message: Message): """Показывает полный график""" if message.chat.id not in ALLOWED_CHAT_IDS: - await message.answer("❌ У вас немає доступу.") + await message.answer('❌ У вас немає доступу.') return - + try: - await message.answer("⏳ Завантаження графіку...") + await message.answer('⏳ Завантаження графіку...') html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID) schedule = parse_html(html) text = format_schedule_message(schedule, days_to_show=2) - await message.answer(text, parse_mode="HTML", reply_markup=get_main_keyboard()) + await message.answer(text, parse_mode='HTML', reply_markup=get_main_keyboard()) except Exception as e: - await message.answer( - f"❌ Помилка: {str(e)}", - parse_mode="HTML", - reply_markup=get_main_keyboard() - ) + await message.answer(f'❌ Помилка: {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard()) -@dp.message(Command("today")) +@dp.message(Command('today')) async def cmd_today(message: Message): """Показывает график на сегодня""" if message.chat.id not in ALLOWED_CHAT_IDS: - await message.answer("❌ У вас немає доступу.") + await message.answer('❌ У вас немає доступу.') return - + try: - await message.answer("⏳ Завантаження графіку сьогодні...") + await message.answer('⏳ Завантаження графіку сьогодні...') html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID) schedule = parse_html(html) text = format_single_day_schedule(schedule, 0) - await message.answer(text, parse_mode="HTML", reply_markup=get_main_keyboard()) + await message.answer(text, parse_mode='HTML', reply_markup=get_main_keyboard()) except Exception as e: - await message.answer( - f"❌ Помилка: {str(e)}", - parse_mode="HTML", - reply_markup=get_main_keyboard() - ) + await message.answer(f'❌ Помилка: {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard()) -@dp.message(Command("tomorrow")) +@dp.message(Command('tomorrow')) async def cmd_tomorrow(message: Message): """Показывает график на завтра""" if message.chat.id not in ALLOWED_CHAT_IDS: - await message.answer("❌ У вас немає доступу.") + await message.answer('❌ У вас немає доступу.') return - + try: - await message.answer("⏳ Завантаження графіку завтра...") + await message.answer('⏳ Завантаження графіку завтра...') html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID) schedule = parse_html(html) text = format_single_day_schedule(schedule, 1) - await message.answer(text, parse_mode="HTML", reply_markup=get_main_keyboard()) + await message.answer(text, parse_mode='HTML', reply_markup=get_main_keyboard()) # await message.answer("❌ Функція тимчасово недоступна. Чекаємо на оновлення сайту", parse_mode="HTML", reply_markup=get_main_keyboard()) except Exception as e: - await message.answer( - f"❌ Помилка: {str(e)}", - parse_mode="HTML", - reply_markup=get_main_keyboard() - ) + await message.answer(f'❌ Помилка: {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard()) -@dp.message(Command("check")) +@dp.message(Command('check')) async def cmd_check(message: Message): """Принудительная проверка графика""" if message.chat.id not in ALLOWED_CHAT_IDS: - await message.answer("❌ У вас немає доступу.") + await message.answer('❌ У вас немає доступу.') return - + try: - await message.answer("🔄 Перевіряю графік...", parse_mode="HTML") + await message.answer('🔄 Перевіряю графік...', parse_mode='HTML') html = get_voe_html(VOE_CITY_ID, VOE_STREET_ID, VOE_HOUSE_ID) new_schedule = parse_html(html) - - prefix = "✅ Знайдено зміни!\n\n" if schedules_differ(last_schedule, new_schedule) else "✓ Графік без змін\n\n" - result = prefix + format_schedule_message(new_schedule, days_to_show=2) - - await message.answer(result, parse_mode="HTML", reply_markup=get_main_keyboard()) - except Exception as e: - await message.answer( - f"❌ Помилка: {str(e)}", - parse_mode="HTML", - reply_markup=get_main_keyboard() + + prefix = ( + '✅ Знайдено зміни!\n\n' + if schedules_differ(last_schedule, new_schedule) + else '✓ Графік без змін\n\n' ) + result = prefix + format_schedule_message(new_schedule, days_to_show=2) + + await message.answer(result, parse_mode='HTML', reply_markup=get_main_keyboard()) + except Exception as e: + await message.answer(f'❌ Помилка: {str(e)}', parse_mode='HTML', reply_markup=get_main_keyboard()) @dp.message() @@ -702,37 +671,37 @@ async def handle_text(message: Message): """Обработчик текстовых сообщений и кнопок""" if message.chat.id not in ALLOWED_CHAT_IDS: return - + text = message.text - + # Кнопка "Графік" - if text == "📊 Графік": + if text == '📊 Графік': await cmd_schedule(message) - + # Кнопка "Сьогодні" - elif text == "📅 Сьогодні": + elif text == '📅 Сьогодні': await cmd_today(message) - + # Кнопка "Завтра" - elif text == "📅 Завтра": + elif text == '📅 Завтра': await cmd_tomorrow(message) - + # Кнопка "Оновити" - elif text == "🔄 Оновити": + elif text == '🔄 Оновити': await cmd_check(message) - + # Кнопка "Про бота" - elif text == "ℹ️ Про бота": + elif text == 'ℹ️ Про бота': await cmd_info(message) - + # Неизвестная команда else: await message.answer( - "❓ Команда не розпізнана\n\n" - "Використовуйте кнопки на клавіатурі або команди:\n" - "/start • /today • /tomorrow • /schedule • /check", - parse_mode="HTML", - reply_markup=get_main_keyboard() + '❓ Команда не розпізнана\n\n' + 'Використовуйте кнопки на клавіатурі або команди:\n' + '/start • /today • /tomorrow • /schedule • /check', + parse_mode='HTML', + reply_markup=get_main_keyboard(), ) @@ -740,39 +709,40 @@ async def handle_text(message: Message): # ГЛАВНАЯ ФУНКЦИЯ # ============================================================================ + async def main(): """Главная функция""" - logger.info("=" * 50) - logger.info("ЗАПУСК БОТА V2.2 (stable 2.2, 15.11.2025)") - logger.info("=" * 50) - - if not TELEGRAM_TOKEN or TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN", "YOUR_TOKEN_HERE") - logger.error("❌ TELEGRAM_TOKEN не конфігурований! Напишіть токен в .env файл") + logger.info('=' * 50) + logger.info('ЗАПУСК БОТА V2.2 (stable 2.2, 15.11.2025)') + logger.info('=' * 50) + + if not TELEGRAM_TOKEN or os.getenv('TELEGRAM_TOKEN', 'YOUR_TOKEN_HERE') == TELEGRAM_TOKEN: + logger.error('❌ TELEGRAM_TOKEN не конфігурований! Напишіть токен в .env файл') return - + if not ALLOWED_CHAT_IDS: - logger.error("❌ ALLOWED_CHAT_IDS не конфігуровані! Напишіть ID в .env файл") + logger.error('❌ ALLOWED_CHAT_IDS не конфігуровані! Напишіть ID в .env файл') return - - logger.info(f"📌 Allowed chat ids: {ALLOWED_CHAT_IDS}") - logger.info(f"⏱ Інтервал перевірки: {CHECK_INTERVAL} сек") - logger.info("=" * 50) - + + logger.info(f'📌 Allowed chat ids: {ALLOWED_CHAT_IDS}') + logger.info(f'⏱ Інтервал перевірки: {CHECK_INTERVAL} сек') + logger.info('=' * 50) + # Запускаем мониторинг monitoring_task = asyncio.create_task(monitoring_loop()) - + try: await dp.start_polling(bot) except KeyboardInterrupt: - logger.info("⏹ Бот зупинений користувачем") + logger.info('⏹ Бот зупинений користувачем') finally: monitoring_task.cancel() await bot.session.close() - logger.info("✓ Підключення закрито") + logger.info('✓ Підключення закрито') -if __name__ == "__main__": +if __name__ == '__main__': try: asyncio.run(main()) except KeyboardInterrupt: - logger.info("⏹ Завершено") \ No newline at end of file + logger.info('⏹ Завершено') diff --git a/edu_master/phpsessid-bot/Dockerfile b/edu_master/phpsessid-bot/Dockerfile index b66780c..18a7a51 100644 --- a/edu_master/phpsessid-bot/Dockerfile +++ b/edu_master/phpsessid-bot/Dockerfile @@ -3,10 +3,10 @@ FROM python:3.11-slim WORKDIR /app # Install system dependencies -RUN apt-get update && apt-get install -y redis-tools && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y --no-install-recommends redis-tools && rm -rf /var/lib/apt/lists/* # Install dependencies -RUN pip install requests redis +RUN pip install --no-cache-dir requests==2.32.3 redis==5.2.1 # Copy application code COPY . . diff --git a/edu_master/phpsessid-bot/bot.py b/edu_master/phpsessid-bot/bot.py index 4c3e3cc..963a143 100644 --- a/edu_master/phpsessid-bot/bot.py +++ b/edu_master/phpsessid-bot/bot.py @@ -7,12 +7,10 @@ import redis import requests # Configure logging -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s' -) +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) + # Load configuration (adapted to .env keys) def _env(key, default=None): v = os.getenv(key, default) @@ -20,40 +18,46 @@ def _env(key, default=None): return v[1:-1] return v + LOGIN = _env('KEEPER_LOGIN') PASSWORD = _env('KEEPER_PASSWORD') EDU_BASE = _env('EDU_URL_BASE', 'https://edu.edu.vn.ua') EDU_LOGIN_PATH = _env('EDU_URL_LOGIN', '/user/login') EDU_COURSES_PATH = _env('EDU_URL_COURSES', '/course/userlist') -URL_LOGIN = f"{EDU_BASE.rstrip('/')}/{EDU_LOGIN_PATH.lstrip('/')}" -URL_VERIFY = f"{EDU_BASE.rstrip('/')}/{EDU_COURSES_PATH.lstrip('/')}" +URL_LOGIN = f'{EDU_BASE.rstrip("/")}/{EDU_LOGIN_PATH.lstrip("/")}' +URL_VERIFY = f'{EDU_BASE.rstrip("/")}/{EDU_COURSES_PATH.lstrip("/")}' INTERVAL = int(_env('KEEPER_INTERVAL', 10)) -USER_AGENT = _env('USER_AGENT', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36') +USER_AGENT = _env( + 'USER_AGENT', + 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36', +) REDIS_HOST = _env('REDIS_HOST', 'redis') REDIS_PORT = int(_env('REDIS_PORT', 6379)) SUCCESS_FILE = '/tmp/last_success' # noqa: S108 + def touch_success_file(): """Updates the timestamp of the success file for healthchecks.""" try: with open(SUCCESS_FILE, 'w') as f: f.write(str(datetime.now().timestamp())) except Exception as e: - logger.error(f"Failed to touch success file: {e}") + logger.error(f'Failed to touch success file: {e}') + def main(): - logger.info("Starting Session Keeper Bot") + logger.info('Starting Session Keeper Bot') # Connect to Redis try: redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True) redis_client.ping() - logger.info(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}") + logger.info(f'Connected to Redis at {REDIS_HOST}:{REDIS_PORT}') except Exception as e: - logger.error(f"Failed to connect to Redis: {e}") + logger.error(f'Failed to connect to Redis: {e}') return session = requests.Session() @@ -73,19 +77,16 @@ def main(): 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"Linux"', 'Accept-Encoding': 'gzip, deflate, br', - 'Priority': 'u=0, i' + 'Priority': 'u=0, i', } session.headers.update(headers) while True: try: - logger.info("Attempting login...") + logger.info('Attempting login...') # Login payload - payload = { - 'login': LOGIN, - 'password': PASSWORD - } + payload = {'login': LOGIN, 'password': PASSWORD} # Perform Login # Note: The user request shows a POST to /user/login with form data @@ -94,17 +95,17 @@ def main(): login_response = session.post(URL_LOGIN, data=payload, allow_redirects=True) - logger.info(f"Login Response Status: {login_response.status_code}") - logger.info(f"Cookies after login: {session.cookies.get_dict()}") + logger.info(f'Login Response Status: {login_response.status_code}') + logger.info(f'Cookies after login: {session.cookies.get_dict()}') # Verify Session - logger.info("Verifying session...") + logger.info('Verifying session...') verify_response = session.get(URL_VERIFY, allow_redirects=False) - logger.info(f"Verify Response Status: {verify_response.status_code}") + logger.info(f'Verify Response Status: {verify_response.status_code}') if verify_response.status_code == 200: - logger.info("Session verification SUCCESS (200 OK).") + logger.info('Session verification SUCCESS (200 OK).') touch_success_file() # Save PHPSESSID to Redis @@ -112,19 +113,20 @@ def main(): if phpsessid: try: redis_client.set('EDU_PHPSESSID', phpsessid) - logger.info(f"Saved PHPSESSID to Redis: {phpsessid}") + logger.info(f'Saved PHPSESSID to Redis: {phpsessid}') except Exception as e: - logger.error(f"Failed to save PHPSESSID to Redis: {e}") + logger.error(f'Failed to save PHPSESSID to Redis: {e}') elif verify_response.status_code == 302: - logger.warning("Session verification FAILED (302 Redirect). Session might be invalid.") + logger.warning('Session verification FAILED (302 Redirect). Session might be invalid.') else: - logger.warning(f"Session verification returned unexpected status: {verify_response.status_code}") + logger.warning(f'Session verification returned unexpected status: {verify_response.status_code}') except Exception as e: - logger.error(f"An error occurred: {e}") + logger.error(f'An error occurred: {e}') - logger.info(f"Sleeping for {INTERVAL} minutes...") + logger.info(f'Sleeping for {INTERVAL} minutes...') time.sleep(INTERVAL * 60) -if __name__ == "__main__": + +if __name__ == '__main__': main() diff --git a/edu_master/webinar-checker/Dockerfile b/edu_master/webinar-checker/Dockerfile index ec581f7..833e02e 100644 --- a/edu_master/webinar-checker/Dockerfile +++ b/edu_master/webinar-checker/Dockerfile @@ -3,7 +3,7 @@ FROM python:3.11-slim WORKDIR /app # Install dependencies -RUN pip install --upgrade pip && pip install playwright==1.56.0 redis requests "python-telegram-bot[job-queue]" +RUN pip install --no-cache-dir --upgrade pip && pip install --no-cache-dir playwright==1.56.0 redis==5.2.1 requests==2.32.3 "python-telegram-bot[job-queue]==21.10" COPY checker.py . diff --git a/edu_master/webinar-checker/checker.py b/edu_master/webinar-checker/checker.py index e0cee47..e7cf468 100644 --- a/edu_master/webinar-checker/checker.py +++ b/edu_master/webinar-checker/checker.py @@ -13,10 +13,7 @@ from telegram.constants import ChatType from telegram.ext import Application, CallbackQueryHandler, CommandHandler, ContextTypes # Logger -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s' -) +logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Suppress HTTP request logs @@ -24,6 +21,7 @@ logging.getLogger('urllib3').setLevel(logging.WARNING) logging.getLogger('httpx').setLevel(logging.WARNING) logging.getLogger('telegram.ext._application').setLevel(logging.WARNING) + # Load environment variables def _env(key, default=None): v = os.getenv(key, default) @@ -31,169 +29,176 @@ def _env(key, default=None): return v[1:-1] return v + EDU_BASE = _env('EDU_URL_BASE', 'https://edu.edu.vn.ua') EDU_WEBINAR_PATH = _env('EDU_URL_WEBINAR', '/webinar/useractive') -WEBINAR_URL = f"{EDU_BASE.rstrip('/')}/{EDU_WEBINAR_PATH.lstrip('/')}" -DIARY_URL = f"{EDU_BASE.rstrip('/')}/user/diary" +WEBINAR_URL = f'{EDU_BASE.rstrip("/")}/{EDU_WEBINAR_PATH.lstrip("/")}' +DIARY_URL = f'{EDU_BASE.rstrip("/")}/user/diary' WEBINAR_CHECK_INTERVAL = int(_env('WEBINAR_CHECK_INTERVAL', 60)) REDIS_HOST = _env('REDIS_HOST', 'redis') REDIS_PORT = int(_env('REDIS_PORT', 6379)) PLAYWRIGHT_WS = _env('PLAYWRIGHT_WS', 'ws://playwright-service:3000/ws') -USER_AGENT = _env('USER_AGENT', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36') +USER_AGENT = _env( + 'USER_AGENT', + 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36', +) WEBINAR_TELEGRAM_TOKEN = _env('WEBINAR_TELEGRAM_TOKEN') ADMIN_ID = int(_env('WEBINAR_ADMIN_ID', '0')) # Redis Keys -KEY_WHITELIST = "bot:whitelist" -KEY_WHITELIST_ENABLED = "bot:whitelist_enabled" -KEY_SUBSCRIBERS = "bot:subscribers" -KEY_PHPSESSID = "EDU_PHPSESSID" -KEY_WEBINAR_HISTORY = "bot:webinar_history" # Stores last 3 webinars +KEY_WHITELIST = 'bot:whitelist' +KEY_WHITELIST_ENABLED = 'bot:whitelist_enabled' +KEY_SUBSCRIBERS = 'bot:subscribers' +KEY_PHPSESSID = 'EDU_PHPSESSID' +KEY_WEBINAR_HISTORY = 'bot:webinar_history' # Stores last 3 webinars # Initialize Redis try: redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True) redis_client.ping() - logger.info(f"Connected to Redis at {REDIS_HOST}:{REDIS_PORT}") + logger.info(f'Connected to Redis at {REDIS_HOST}:{REDIS_PORT}') except Exception as e: - logger.error(f"Failed to connect to Redis: {e}") + logger.error(f'Failed to connect to Redis: {e}') exit(1) # --- Translations --- TRANSLATIONS = { 'ru': { - 'welcome': "👋 Привет, {name}!\n\nЯ бот-уведомитель о вебинарах. Я буду сообщать вам, когда появится новый вебинар.\nВы подписаны на уведомления.", - 'welcome_admin': "\n\n👑 Режим администратора активен", - 'access_denied': "⛔ Доступ запрещен. Вас нет в белом списке.", - 'help_title': "🤖 Помощь по боту\n\n", - 'help_commands': "/start - Подписаться на уведомления\n/stop - Отписаться от уведомлений\n/help - Показать это сообщение\n/language - Сменить язык", - 'help_admin': "\nКоманды администратора:\n/adduser [user_id] - Добавить пользователя в белый список\n/removeuser [user_id] - Удалить пользователя из белого списка\nИли используйте панель ниже для управления настройками.", - 'admin_only': "⛔ Только для администратора!", - 'user_added': "✅ Пользователь {user_id} добавлен в белый список", - 'user_removed': "✅ Пользователь {user_id} удален из белого списка", - 'user_not_in_whitelist': "⚠️ Пользователь {user_id} не был в белом списке", - 'cannot_remove_admin': "❌ Невозможно удалить администратора из белого списка", - 'invalid_user_id': "❌ Неверный ID пользователя. Должно быть число.", - 'usage_adduser': "Использование: /adduser [user_id]", - 'usage_removeuser': "Использование: /removeuser [user_id]", - 'whitelist_enabled': "✅ Белый список включен", - 'whitelist_disabled': "✅ Белый список отключен", - 'whitelist_title': "📋 Белый список:\n", - 'subscribers_title': "👥 Подписчики:\n", - 'empty': "Пусто", - 'force_check_running': "🔄 Запускаю проверку...", - 'check_failed': "❌ Проверка не удалась. Смотрите логи.", - 'check_completed_none': "✅ Проверка завершена. Вебинаров не найдено.", - 'check_completed': "✅ Проверка завершена. Найдено {count} вебинар(ов)!", - 'toggle_whitelist_disable': "🔒 Отключить белый список", - 'toggle_whitelist_enable': "🔓 Включить белый список", - 'view_whitelist': "📋 Посмотреть белый список", - 'view_subscribers': "👥 Посмотреть подписчиков", - 'force_check': "🔄 Принудительная проверка", - 'webinar_found': "🎓 Новый вебинар!\n\n", - 'webinar_item': "📌 {name}\n🔗 https://edu.edu.vn.ua{url}", - 'select_language': "🌐 Выберите язык / Оберіть мову / Select language:", - 'language_changed': "✅ Язык изменен на русский", - 'flag_ru': "🇷🇺 Русский", - 'flag_uk': "🇺🇦 Українська", - 'flag_en': "🇬🇧 English", - 'history_cleared': "✅ История вебинаров очищена", - 'history_clear_failed': "❌ Ошибка при очистке истории", + 'welcome': '👋 Привет, {name}!\n\nЯ бот-уведомитель о вебинарах. Я буду сообщать вам, когда появится новый вебинар.\nВы подписаны на уведомления.', + 'welcome_admin': '\n\n👑 Режим администратора активен', + 'access_denied': '⛔ Доступ запрещен. Вас нет в белом списке.', + 'help_title': '🤖 Помощь по боту\n\n', + 'help_commands': '/start - Подписаться на уведомления\n/stop - Отписаться от уведомлений\n/help - Показать это сообщение\n/language - Сменить язык', + 'help_admin': '\nКоманды администратора:\n/adduser [user_id] - Добавить пользователя в белый список\n/removeuser [user_id] - Удалить пользователя из белого списка\nИли используйте панель ниже для управления настройками.', + 'admin_only': '⛔ Только для администратора!', + 'user_added': '✅ Пользователь {user_id} добавлен в белый список', + 'user_removed': '✅ Пользователь {user_id} удален из белого списка', + 'user_not_in_whitelist': '⚠️ Пользователь {user_id} не был в белом списке', + 'cannot_remove_admin': '❌ Невозможно удалить администратора из белого списка', + 'invalid_user_id': '❌ Неверный ID пользователя. Должно быть число.', + 'usage_adduser': 'Использование: /adduser [user_id]', + 'usage_removeuser': 'Использование: /removeuser [user_id]', + 'whitelist_enabled': '✅ Белый список включен', + 'whitelist_disabled': '✅ Белый список отключен', + 'whitelist_title': '📋 Белый список:\n', + 'subscribers_title': '👥 Подписчики:\n', + 'empty': 'Пусто', + 'force_check_running': '🔄 Запускаю проверку...', + 'check_failed': '❌ Проверка не удалась. Смотрите логи.', + 'check_completed_none': '✅ Проверка завершена. Вебинаров не найдено.', + 'check_completed': '✅ Проверка завершена. Найдено {count} вебинар(ов)!', + 'toggle_whitelist_disable': '🔒 Отключить белый список', + 'toggle_whitelist_enable': '🔓 Включить белый список', + 'view_whitelist': '📋 Посмотреть белый список', + 'view_subscribers': '👥 Посмотреть подписчиков', + 'force_check': '🔄 Принудительная проверка', + 'webinar_found': '🎓 Новый вебинар!\n\n', + 'webinar_item': '📌 {name}\n🔗 https://edu.edu.vn.ua{url}', + 'select_language': '🌐 Выберите язык / Оберіть мову / Select language:', + 'language_changed': '✅ Язык изменен на русский', + 'flag_ru': '🇷🇺 Русский', + 'flag_uk': '🇺🇦 Українська', + 'flag_en': '🇬🇧 English', + 'history_cleared': '✅ История вебинаров очищена', + 'history_clear_failed': '❌ Ошибка при очистке истории', }, 'uk': { 'welcome': "👋 Привіт, {name}!\n\nЯ бот-сповіщувач про вебінари. Я повідомлятиму вас, коли з'явиться новий вебінар.\nВи підписані на сповіщення.", - 'welcome_admin': "\n\n👑 Режим адміністратора активний", - 'access_denied': "⛔ Доступ заборонено. Вас немає в білому списку.", - 'help_title': "🤖 Довідка по боту\n\n", - 'help_commands': "/start - Підписатися на сповіщення\n/stop - Відписатися від сповіщень\n/help - Показати це повідомлення\n/language - Змінити мову", - 'help_admin': "\nКоманди адміністратора:\n/adduser [user_id] - Додати користувача до білого списку\n/removeuser [user_id] - Видалити користувача з білого списку\nАбо використовуйте панель нижче для керування налаштуваннями.", - 'admin_only': "⛔ Тільки для адміністратора!", - 'user_added': "✅ Користувач {user_id} доданий до білого списку", - 'user_removed': "✅ Користувач {user_id} видалений з білого списку", - 'user_not_in_whitelist': "⚠️ Користувач {user_id} не був у білому списку", - 'cannot_remove_admin': "❌ Неможливо видалити адміністратора з білого списку", - 'invalid_user_id': "❌ Невірний ID користувача. Має бути число.", - 'usage_adduser': "Використання: /adduser [user_id]", - 'usage_removeuser': "Використання: /removeuser [user_id]", - 'whitelist_enabled': "✅ Білий список увімкнено", - 'whitelist_disabled': "✅ Білий список вимкнено", - 'whitelist_title': "📋 Білий список:\n", - 'subscribers_title': "👥 Підписники:\n", - 'empty': "Порожньо", - 'force_check_running': "🔄 Запускаю перевірку...", - 'check_failed': "❌ Перевірка не вдалася. Дивіться логи.", - 'check_completed_none': "✅ Перевірка завершена. Вебінарів не знайдено.", - 'check_completed': "✅ Перевірка завершена. Знайдено {count} вебінар(ів)!", - 'toggle_whitelist_disable': "🔒 Вимкнути білий список", - 'toggle_whitelist_enable': "🔓 Увімкнути білий список", - 'view_whitelist': "📋 Переглянути білий список", - 'view_subscribers': "👥 Переглянути підписників", - 'force_check': "🔄 Примусова перевірка", - 'webinar_found': "🎓 Новий вебінар!\n\n", - 'webinar_item': "📌 {name}\n🔗 https://edu.edu.vn.ua{url}", - 'select_language': "🌐 Виберіть мову / Выберите язык / Select language:", - 'language_changed': "✅ Мову змінено на українську", - 'flag_ru': "🇷🇺 Русский", - 'flag_uk': "🇺🇦 Українська", - 'flag_en': "🇬🇧 English", - 'history_cleared': "✅ Історія вебінарів очищена", - 'history_clear_failed': "❌ Помилка при очищенні історії", + 'welcome_admin': '\n\n👑 Режим адміністратора активний', + 'access_denied': '⛔ Доступ заборонено. Вас немає в білому списку.', + 'help_title': '🤖 Довідка по боту\n\n', + 'help_commands': '/start - Підписатися на сповіщення\n/stop - Відписатися від сповіщень\n/help - Показати це повідомлення\n/language - Змінити мову', + 'help_admin': '\nКоманди адміністратора:\n/adduser [user_id] - Додати користувача до білого списку\n/removeuser [user_id] - Видалити користувача з білого списку\nАбо використовуйте панель нижче для керування налаштуваннями.', + 'admin_only': '⛔ Тільки для адміністратора!', + 'user_added': '✅ Користувач {user_id} доданий до білого списку', + 'user_removed': '✅ Користувач {user_id} видалений з білого списку', + 'user_not_in_whitelist': '⚠️ Користувач {user_id} не був у білому списку', + 'cannot_remove_admin': '❌ Неможливо видалити адміністратора з білого списку', + 'invalid_user_id': '❌ Невірний ID користувача. Має бути число.', + 'usage_adduser': 'Використання: /adduser [user_id]', + 'usage_removeuser': 'Використання: /removeuser [user_id]', + 'whitelist_enabled': '✅ Білий список увімкнено', + 'whitelist_disabled': '✅ Білий список вимкнено', + 'whitelist_title': '📋 Білий список:\n', + 'subscribers_title': '👥 Підписники:\n', + 'empty': 'Порожньо', + 'force_check_running': '🔄 Запускаю перевірку...', + 'check_failed': '❌ Перевірка не вдалася. Дивіться логи.', + 'check_completed_none': '✅ Перевірка завершена. Вебінарів не знайдено.', + 'check_completed': '✅ Перевірка завершена. Знайдено {count} вебінар(ів)!', + 'toggle_whitelist_disable': '🔒 Вимкнути білий список', + 'toggle_whitelist_enable': '🔓 Увімкнути білий список', + 'view_whitelist': '📋 Переглянути білий список', + 'view_subscribers': '👥 Переглянути підписників', + 'force_check': '🔄 Примусова перевірка', + 'webinar_found': '🎓 Новий вебінар!\n\n', + 'webinar_item': '📌 {name}\n🔗 https://edu.edu.vn.ua{url}', + 'select_language': '🌐 Виберіть мову / Выберите язык / Select language:', + 'language_changed': '✅ Мову змінено на українську', + 'flag_ru': '🇷🇺 Русский', + 'flag_uk': '🇺🇦 Українська', + 'flag_en': '🇬🇧 English', + 'history_cleared': '✅ Історія вебінарів очищена', + 'history_clear_failed': '❌ Помилка при очищенні історії', }, 'en': { - 'welcome': "👋 Hello, {name}!\n\nI am the Webinar Checker Bot. I will notify you when a new webinar appears.\nYou have been subscribed to notifications.", - 'welcome_admin': "\n\n👑 Admin Mode Active", - 'access_denied': "⛔ Access denied. You are not on the whitelist.", - 'help_title': "🤖 Bot Help\n\n", - 'help_commands': "/start - Subscribe to notifications\n/stop - Unsubscribe from notifications\n/help - Show this message\n/language - Change language", - 'help_admin': "\nAdmin Commands:\n/adduser [user_id] - Add user to whitelist\n/removeuser [user_id] - Remove user from whitelist\nOr use the panel below to manage settings.", - 'admin_only': "⛔ Admin only!", - 'user_added': "✅ User {user_id} added to whitelist", - 'user_removed': "✅ User {user_id} removed from whitelist", - 'user_not_in_whitelist': "⚠️ User {user_id} was not in whitelist", - 'cannot_remove_admin': "❌ Cannot remove admin from whitelist", - 'invalid_user_id': "❌ Invalid user ID. Must be a number.", - 'usage_adduser': "Usage: /adduser [user_id]", - 'usage_removeuser': "Usage: /removeuser [user_id]", - 'whitelist_enabled': "✅ Whitelist Enabled", - 'whitelist_disabled': "✅ Whitelist Disabled", - 'whitelist_title': "📋 Whitelist:\n", - 'subscribers_title': "👥 Subscribers:\n", - 'empty': "Empty", - 'force_check_running': "🔄 Running immediate check...", - 'check_failed': "❌ Check failed. See logs for details.", - 'check_completed_none': "✅ Check completed. No webinars found.", - 'check_completed': "✅ Check completed. Found {count} webinar(s)!", - 'toggle_whitelist_disable': "🔒 Disable Whitelist", - 'toggle_whitelist_enable': "🔓 Enable Whitelist", - 'view_whitelist': "📋 View Whitelist", - 'view_subscribers': "👥 View Subscribers", - 'force_check': "🔄 Force Check", - 'webinar_found': "🎓 New webinar found!\n\n", - 'webinar_item': "📌 {name}\n🔗 https://edu.edu.vn.ua{url}", - 'select_language': "🌐 Select language / Виберіть мову / Выберите язык:", - 'language_changed': "✅ Language changed to English", - 'flag_ru': "🇷🇺 Русский", - 'flag_uk': "🇺🇦 Українська", - 'flag_en': "🇬🇧 English", - 'history_cleared': "✅ Webinar history cleared", - 'history_clear_failed': "❌ Error clearing history", - } + 'welcome': '👋 Hello, {name}!\n\nI am the Webinar Checker Bot. I will notify you when a new webinar appears.\nYou have been subscribed to notifications.', + 'welcome_admin': '\n\n👑 Admin Mode Active', + 'access_denied': '⛔ Access denied. You are not on the whitelist.', + 'help_title': '🤖 Bot Help\n\n', + 'help_commands': '/start - Subscribe to notifications\n/stop - Unsubscribe from notifications\n/help - Show this message\n/language - Change language', + 'help_admin': '\nAdmin Commands:\n/adduser [user_id] - Add user to whitelist\n/removeuser [user_id] - Remove user from whitelist\nOr use the panel below to manage settings.', + 'admin_only': '⛔ Admin only!', + 'user_added': '✅ User {user_id} added to whitelist', + 'user_removed': '✅ User {user_id} removed from whitelist', + 'user_not_in_whitelist': '⚠️ User {user_id} was not in whitelist', + 'cannot_remove_admin': '❌ Cannot remove admin from whitelist', + 'invalid_user_id': '❌ Invalid user ID. Must be a number.', + 'usage_adduser': 'Usage: /adduser [user_id]', + 'usage_removeuser': 'Usage: /removeuser [user_id]', + 'whitelist_enabled': '✅ Whitelist Enabled', + 'whitelist_disabled': '✅ Whitelist Disabled', + 'whitelist_title': '📋 Whitelist:\n', + 'subscribers_title': '👥 Subscribers:\n', + 'empty': 'Empty', + 'force_check_running': '🔄 Running immediate check...', + 'check_failed': '❌ Check failed. See logs for details.', + 'check_completed_none': '✅ Check completed. No webinars found.', + 'check_completed': '✅ Check completed. Found {count} webinar(s)!', + 'toggle_whitelist_disable': '🔒 Disable Whitelist', + 'toggle_whitelist_enable': '🔓 Enable Whitelist', + 'view_whitelist': '📋 View Whitelist', + 'view_subscribers': '👥 View Subscribers', + 'force_check': '🔄 Force Check', + 'webinar_found': '🎓 New webinar found!\n\n', + 'webinar_item': '📌 {name}\n🔗 https://edu.edu.vn.ua{url}', + 'select_language': '🌐 Select language / Виберіть мову / Выберите язык:', + 'language_changed': '✅ Language changed to English', + 'flag_ru': '🇷🇺 Русский', + 'flag_uk': '🇺🇦 Українська', + 'flag_en': '🇬🇧 English', + 'history_cleared': '✅ Webinar history cleared', + 'history_clear_failed': '❌ Error clearing history', + }, } # --- Language Helper Functions --- + def get_user_language(user_id: int) -> str: """Get user's preferred language from Redis. Default: Ukrainian.""" - lang = redis_client.get(f"user:{user_id}:language") + lang = redis_client.get(f'user:{user_id}:language') return lang if lang in ['ru', 'uk', 'en'] else 'uk' + def set_user_language(user_id: int, lang: str): """Save user's language preference to Redis.""" if lang in ['ru', 'uk', 'en']: - redis_client.set(f"user:{user_id}:language", lang) - logger.info(f"User {user_id} language set to {lang}") + redis_client.set(f'user:{user_id}:language', lang) + logger.info(f'User {user_id} language set to {lang}') + def t(user_id: int, key: str, **kwargs) -> str: """Translate message for user with optional formatting.""" @@ -203,81 +208,102 @@ def t(user_id: int, key: str, **kwargs) -> str: return message.format(**kwargs) return message + def get_language_keyboard(): """Generate language selection keyboard.""" keyboard = [ [ - InlineKeyboardButton("🇷🇺 Русский", callback_data="lang_ru"), - InlineKeyboardButton("🇺🇦 Українська", callback_data="lang_uk"), + InlineKeyboardButton('🇷🇺 Русский', callback_data='lang_ru'), + InlineKeyboardButton('🇺🇦 Українська', callback_data='lang_uk'), ], [ - InlineKeyboardButton("🇬🇧 English", callback_data="lang_en"), - ] + InlineKeyboardButton('🇬🇧 English', callback_data='lang_en'), + ], ] return InlineKeyboardMarkup(keyboard) + # --- Helper Functions --- + def is_whitelisted(user_id: int) -> bool: """Check if user is allowed to use the bot.""" if user_id == ADMIN_ID: return True enabled = redis_client.get(KEY_WHITELIST_ENABLED) - if enabled == "0": # Whitelist disabled + if enabled == '0': # Whitelist disabled return True return redis_client.sismember(KEY_WHITELIST, str(user_id)) + async def is_group_admin(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: """Check if the user is an administrator in the group.""" user = update.effective_user chat = update.effective_chat - if chat.type in [ChatType.PRIVATE, "private"]: + if chat.type in [ChatType.PRIVATE, 'private']: return True try: member = await context.bot.get_chat_member(chat.id, user.id) return member.status in [ChatMember.OWNER, ChatMember.ADMINISTRATOR] except Exception as e: - logger.error(f"Failed to check admin status: {e}") + logger.error(f'Failed to check admin status: {e}') return False + def get_admin_keyboard(user_id: int): """Generate admin panel keyboard.""" - whitelist_enabled = redis_client.get(KEY_WHITELIST_ENABLED) != "0" + whitelist_enabled = redis_client.get(KEY_WHITELIST_ENABLED) != '0' toggle_text = t(user_id, 'toggle_whitelist_disable') if whitelist_enabled else t(user_id, 'toggle_whitelist_enable') keyboard = [ - [InlineKeyboardButton(toggle_text, callback_data="toggle_whitelist")], - [InlineKeyboardButton(t(user_id, 'view_whitelist'), callback_data="view_whitelist")], - [InlineKeyboardButton(t(user_id, 'view_subscribers'), callback_data="view_subscribers")], - [InlineKeyboardButton(t(user_id, 'force_check'), callback_data="force_check")] + [InlineKeyboardButton(toggle_text, callback_data='toggle_whitelist')], + [InlineKeyboardButton(t(user_id, 'view_whitelist'), callback_data='view_whitelist')], + [InlineKeyboardButton(t(user_id, 'view_subscribers'), callback_data='view_subscribers')], + [InlineKeyboardButton(t(user_id, 'force_check'), callback_data='force_check')], ] return InlineKeyboardMarkup(keyboard) + # --- Diary Functions --- -DIARY_MONTH_NAMES = ['', 'Січня', 'Лютого', 'Березня', 'Квітня', 'Травня', 'Червня', - 'Липня', 'Серпня', 'Вересня', 'Жовтня', 'Листопада', 'Грудня'] +DIARY_MONTH_NAMES = [ + '', + 'Січня', + 'Лютого', + 'Березня', + 'Квітня', + 'Травня', + 'Червня', + 'Липня', + 'Серпня', + 'Вересня', + 'Жовтня', + 'Листопада', + 'Грудня', +] DIARY_WEEKDAYS_SHORT = ['Пн', 'Вт', 'Ср', 'Чт', 'Пт', 'Сб', 'Нд'] + def get_diary_keyboard(): today = datetime.now() keyboard = [ [ - InlineKeyboardButton(f"📌 Сьогодні ({today.day}.{today.month:02d})", callback_data="diary_today"), - InlineKeyboardButton("📌 Завтра", callback_data="diary_tomorrow"), + InlineKeyboardButton(f'📌 Сьогодні ({today.day}.{today.month:02d})', callback_data='diary_today'), + InlineKeyboardButton('📌 Завтра', callback_data='diary_tomorrow'), ], [ - InlineKeyboardButton("📅 Цей тиждень", callback_data="diary_week"), - InlineKeyboardButton("📅 Весь місяць", callback_data="diary_month"), + InlineKeyboardButton('📅 Цей тиждень', callback_data='diary_week'), + InlineKeyboardButton('📅 Весь місяць', callback_data='diary_month'), ], ] return InlineKeyboardMarkup(keyboard) + def _parse_calendar_html(table_html: str) -> tuple: """Parse calendar HTML table into (month_text, {day_num: {weekday, events}}).""" days = {} @@ -328,18 +354,15 @@ def _parse_calendar_html(table_html: str) -> tuple: async def fetch_diary_data(phpsessid: str) -> dict | None: - logger.info("Fetching diary data via Playwright...") + logger.info('Fetching diary data via Playwright...') try: async with async_playwright() as p: browser = await p.chromium.connect(PLAYWRIGHT_WS) try: context_browser = await browser.new_context(user_agent=USER_AGENT) - await context_browser.add_cookies([{ - 'name': 'PHPSESSID', - 'value': phpsessid, - 'domain': 'edu.edu.vn.ua', - 'path': '/' - }]) + await context_browser.add_cookies( + [{'name': 'PHPSESSID', 'value': phpsessid, 'domain': 'edu.edu.vn.ua', 'path': '/'}] + ) page = await context_browser.new_page() try: @@ -354,21 +377,22 @@ async def fetch_diary_data(phpsessid: str) -> dict | None: } """) if not table_html: - logger.error("table.calendar not found in DOM") + logger.error('table.calendar not found in DOM') return None # Debug: save HTML for troubleshooting - with contextlib.suppress(Exception), \ - open('/tmp/diary_debug.html', 'w', encoding='utf-8') as f: # noqa: S108 + with contextlib.suppress(Exception), open('/tmp/diary_debug.html', 'w', encoding='utf-8') as f: # noqa: S108 f.write(table_html) month_text, days = _parse_calendar_html(table_html) - logger.info(f"Diary parsed: month={month_text!r}, days_with_events={sum(1 for d in days.values() if d['events'])}/{len(days)}") + logger.info( + f'Diary parsed: month={month_text!r}, days_with_events={sum(1 for d in days.values() if d["events"])}/{len(days)}' + ) return {'monthFullText': month_text, 'days': days} except Exception as e: - logger.error(f"Error parsing diary: {e}") + logger.error(f'Error parsing diary: {e}') return None finally: await page.close() @@ -376,64 +400,69 @@ async def fetch_diary_data(phpsessid: str) -> dict | None: finally: await browser.close() except Exception as e: - logger.error(f"Playwright error in diary fetch: {e}") + logger.error(f'Playwright error in diary fetch: {e}') return None + def _parse_diary_month(text: str) -> str: match = re.search(r'([А-Яа-яіїєґ\']+\s*:\s*\d{4})', text) if match: return match.group(1).replace(' : ', ' ').strip() return text.strip() + def format_diary_day(data: dict, day_num: int) -> str: days = data.get('days', {}) month_str = _parse_diary_month(data.get('monthFullText', '')) day_data = days.get(str(day_num)) - lines = [f"📅 {day_num} {month_str}", "─" * 18] + lines = [f'📅 {day_num} {month_str}', '─' * 18] if not day_data or not day_data.get('events'): - lines.append("Немає подій") + lines.append('Немає подій') else: for e in day_data['events']: - lines.append(f"📌 {e}") - lines.append(f"\n🔗 {DIARY_URL}") - return "\n".join(lines) + lines.append(f'📌 {e}') + lines.append(f'\n🔗 {DIARY_URL}') + return '\n'.join(lines) + def format_diary_week(data: dict, today: datetime) -> str: days = data.get('days', {}) _parse_diary_month(data.get('monthFullText', '')) monday = today - timedelta(days=today.weekday()) sunday = monday + timedelta(days=6) - lines = [f"📅 Тиждень {monday.day}.{monday.month} – {sunday.day}.{sunday.month}\n"] + lines = [f'📅 Тиждень {monday.day}.{monday.month} – {sunday.day}.{sunday.month}\n'] for i in range(7): d = monday + timedelta(days=i) day_data = days.get(str(d.day)) - lines.append(f"─ {DIARY_WEEKDAYS_SHORT[i]} {d.day}.{d.month} ─") + lines.append(f'─ {DIARY_WEEKDAYS_SHORT[i]} {d.day}.{d.month} ─') if not day_data or not day_data.get('events'): - lines.append("Немає подій\n") + lines.append('Немає подій\n') else: for e in day_data['events']: - lines.append(f"📌 {e}") - lines.append("") - lines.append(f"🔗 {DIARY_URL}") - return "\n".join(lines) + lines.append(f'📌 {e}') + lines.append('') + lines.append(f'🔗 {DIARY_URL}') + return '\n'.join(lines) + def format_diary_month(data: dict) -> str: days = data.get('days', {}) month_str = _parse_diary_month(data.get('monthFullText', '')) - lines = [f"📅 {month_str}\n"] + lines = [f'📅 {month_str}\n'] for day_num in sorted(days.keys(), key=int): day_data = days[day_num] events = day_data.get('events', []) weekday = day_data.get('weekday', '') - lines.append(f"─ {weekday} {day_num} ─") + lines.append(f'─ {weekday} {day_num} ─') if not events: - lines.append("Немає подій\n") + lines.append('Немає подій\n') else: for e in events: - lines.append(f"📌 {e}") - lines.append("") - lines.append(f"🔗 {DIARY_URL}") - return "\n".join(lines) + lines.append(f'📌 {e}') + lines.append('') + lines.append(f'🔗 {DIARY_URL}') + return '\n'.join(lines) + async def _get_diary_data(context: ContextTypes.DEFAULT_TYPE) -> dict | None: cached = context.user_data.get('diary_cache') @@ -448,13 +477,15 @@ async def _get_diary_data(context: ContextTypes.DEFAULT_TYPE) -> dict | None: context.user_data['diary_cache'] = {'data': data, 'timestamp': now_ts} return data + # --- Command Handlers --- + async def start(update: Update, _context: ContextTypes.DEFAULT_TYPE): """Handle /start command.""" user = update.effective_user chat = update.effective_chat - logger.info(f"User {user.id} ({user.username}) started the bot in chat {chat.id} ({chat.type}).") + logger.info(f'User {user.id} ({user.username}) started the bot in chat {chat.id} ({chat.type}).') # Check whitelist - MUST be the user executing the command if not is_whitelisted(user.id): @@ -466,24 +497,26 @@ async def start(update: Update, _context: ContextTypes.DEFAULT_TYPE): msg = t(chat.id, 'welcome', name=user.first_name) - if user.id == ADMIN_ID and chat.type == "private": + if user.id == ADMIN_ID and chat.type == 'private': msg += t(chat.id, 'welcome_admin') await update.message.reply_text(msg, parse_mode='HTML', reply_markup=get_admin_keyboard(user.id)) else: await update.message.reply_text(msg, parse_mode='HTML') + async def help_command(update: Update, _context: ContextTypes.DEFAULT_TYPE): """Handle /help command.""" user_id = update.effective_user.id chat_id = update.effective_chat.id msg = t(chat_id, 'help_title') + t(chat_id, 'help_commands') - if user_id == ADMIN_ID and update.effective_chat.type == "private": + if user_id == ADMIN_ID and update.effective_chat.type == 'private': msg += t(chat_id, 'help_admin') await update.message.reply_text(msg, parse_mode='HTML', reply_markup=get_admin_keyboard(user_id)) else: await update.message.reply_text(msg, parse_mode='HTML') + async def stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /stop command (unsubscribe).""" user = update.effective_user @@ -491,11 +524,19 @@ async def stop_command(update: Update, context: ContextTypes.DEFAULT_TYPE): # Permission check: Whitelisted user OR Group Admin if not (is_whitelisted(user.id) or await is_group_admin(update, context)): - await update.message.reply_text(t(chat.id, 'access_denied')) # Or specific "admin only" message + await update.message.reply_text(t(chat.id, 'access_denied')) # Or specific "admin only" message return redis_client.srem(KEY_SUBSCRIBERS, chat.id) - await update.message.reply_text(t(chat.id, 'whitelist_disabled').replace(" whitelist", " notifications").replace("Білий список", "Сповіщення").replace("Белый список", "Уведомления") if chat.id else "Unsubscribed") + await update.message.reply_text( + t(chat.id, 'whitelist_disabled') + .replace(' whitelist', ' notifications') + .replace('Білий список', 'Сповіщення') + .replace('Белый список', 'Уведомления') + if chat.id + else 'Unsubscribed' + ) + async def language_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle /language command.""" @@ -507,11 +548,10 @@ async def language_command(update: Update, context: ContextTypes.DEFAULT_TYPE): return await update.message.reply_text( - t(chat.id, 'select_language'), - parse_mode='HTML', - reply_markup=get_language_keyboard() + t(chat.id, 'select_language'), parse_mode='HTML', reply_markup=get_language_keyboard() ) + async def add_user(update: Update, context: ContextTypes.DEFAULT_TYPE): """Add user to whitelist (admin only).""" admin_id = update.effective_user.id @@ -527,10 +567,11 @@ async def add_user(update: Update, context: ContextTypes.DEFAULT_TYPE): user_id = int(context.args[0]) redis_client.sadd(KEY_WHITELIST, str(user_id)) await update.message.reply_text(t(admin_id, 'user_added', user_id=user_id)) - logger.info(f"Admin added user {user_id} to whitelist") + logger.info(f'Admin added user {user_id} to whitelist') except ValueError: await update.message.reply_text(t(admin_id, 'invalid_user_id')) + async def remove_user(update: Update, context: ContextTypes.DEFAULT_TYPE): """Remove user from whitelist (admin only).""" admin_id = update.effective_user.id @@ -551,12 +592,13 @@ async def remove_user(update: Update, context: ContextTypes.DEFAULT_TYPE): removed = redis_client.srem(KEY_WHITELIST, str(user_id)) if removed: await update.message.reply_text(t(admin_id, 'user_removed', user_id=user_id)) - logger.info(f"Admin removed user {user_id} from whitelist") + logger.info(f'Admin removed user {user_id} from whitelist') else: await update.message.reply_text(t(admin_id, 'user_not_in_whitelist', user_id=user_id)) except ValueError: await update.message.reply_text(t(admin_id, 'invalid_user_id')) + async def clear_history(update: Update, _context: ContextTypes.DEFAULT_TYPE): """Clear webinar history (admin only).""" admin_id = update.effective_user.id @@ -567,79 +609,75 @@ async def clear_history(update: Update, _context: ContextTypes.DEFAULT_TYPE): try: redis_client.delete(KEY_WEBINAR_HISTORY) await update.message.reply_text(t(admin_id, 'history_cleared')) - logger.info("Admin cleared webinar history") + logger.info('Admin cleared webinar history') except Exception as e: - logger.error(f"Failed to clear history: {e}") + logger.error(f'Failed to clear history: {e}') await update.message.reply_text(t(admin_id, 'history_clear_failed')) + async def diary_command(update: Update, _context: ContextTypes.DEFAULT_TYPE): user = update.effective_user if not is_whitelisted(user.id): await update.message.reply_text(t(user.id, 'access_denied')) return await update.message.reply_text( - "📅 Щоденник — виберіть період:", - parse_mode='HTML', - reply_markup=get_diary_keyboard() + '📅 Щоденник — виберіть період:', parse_mode='HTML', reply_markup=get_diary_keyboard() ) + async def diary_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query user_id = query.from_user.id await query.answer() if user_id != ADMIN_ID and not is_whitelisted(user_id): - await query.edit_message_text("⛔ Доступ заборонено.") + await query.edit_message_text('⛔ Доступ заборонено.') return data = query.data - if data == "diary_refresh": + if data == 'diary_refresh': context.user_data.pop('diary_cache', None) - await query.edit_message_text("🔄 Завантажую щоденник...") + await query.edit_message_text('🔄 Завантажую щоденник...') diary_data = await _get_diary_data(context) if not diary_data: - await query.edit_message_text("❌ Не вдалося завантажити щоденник. Немає сесії або помилка.") + await query.edit_message_text('❌ Не вдалося завантажити щоденник. Немає сесії або помилка.') return await query.edit_message_text( - "📅 Щоденник — виберіть період:", - parse_mode='HTML', - reply_markup=get_diary_keyboard() + '📅 Щоденник — виберіть період:', parse_mode='HTML', reply_markup=get_diary_keyboard() ) return - await query.edit_message_text("🔄 Завантажую щоденник...") + await query.edit_message_text('🔄 Завантажую щоденник...') diary_data = await _get_diary_data(context) if not diary_data: - await query.edit_message_text("❌ Не вдалося завантажити щоденник.") + await query.edit_message_text('❌ Не вдалося завантажити щоденник.') return today = datetime.now() - if data == "diary_today": + if data == 'diary_today': text = format_diary_day(diary_data, today.day) - elif data == "diary_tomorrow": + elif data == 'diary_tomorrow': tomorrow = today + timedelta(days=1) if tomorrow.day < today.day: - text = "❌ Дані за наступний місяць недоступні. Перейдіть на сайт." + text = '❌ Дані за наступний місяць недоступні. Перейдіть на сайт.' else: text = format_diary_day(diary_data, tomorrow.day) - elif data == "diary_week": + elif data == 'diary_week': text = format_diary_week(diary_data, today) - elif data == "diary_month": + elif data == 'diary_month': text = format_diary_month(diary_data) else: return if len(text) > 4096: - text = text[:4090] + "\n\n✂️ ...(обрізано)" + text = text[:4090] + '\n\n✂️ ...(обрізано)' + + await query.edit_message_text(text, parse_mode='HTML', reply_markup=get_diary_keyboard()) - await query.edit_message_text( - text, - parse_mode='HTML', - reply_markup=get_diary_keyboard() - ) # --- Admin Callbacks --- + async def admin_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): """Handle admin panel button clicks.""" query = update.callback_query @@ -652,25 +690,25 @@ async def admin_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): await query.answer() data = query.data - if data == "toggle_whitelist": + if data == 'toggle_whitelist': current = redis_client.get(KEY_WHITELIST_ENABLED) - new_state = "0" if current != "0" else "1" + new_state = '0' if current != '0' else '1' redis_client.set(KEY_WHITELIST_ENABLED, new_state) - state_text = t(user_id, 'whitelist_disabled') if new_state == "0" else t(user_id, 'whitelist_enabled') + state_text = t(user_id, 'whitelist_disabled') if new_state == '0' else t(user_id, 'whitelist_enabled') await query.edit_message_reply_markup(reply_markup=get_admin_keyboard(user_id)) await query.message.reply_text(state_text) - elif data == "view_whitelist": + elif data == 'view_whitelist': members = redis_client.smembers(KEY_WHITELIST) - msg = t(user_id, 'whitelist_title') + ("\n".join(members) if members else t(user_id, 'empty')) + msg = t(user_id, 'whitelist_title') + ('\n'.join(members) if members else t(user_id, 'empty')) await query.message.reply_text(msg, parse_mode='HTML') - elif data == "view_subscribers": + elif data == 'view_subscribers': subs = redis_client.smembers(KEY_SUBSCRIBERS) - msg = t(user_id, 'subscribers_title') + ("\n".join(subs) if subs else t(user_id, 'empty')) + msg = t(user_id, 'subscribers_title') + ('\n'.join(subs) if subs else t(user_id, 'empty')) await query.message.reply_text(msg, parse_mode='HTML') - elif data == "force_check": + elif data == 'force_check': await query.message.reply_text(t(user_id, 'force_check_running')) result = await check_webinars_job(context) @@ -681,27 +719,28 @@ async def admin_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): else: await query.message.reply_text(t(user_id, 'check_completed', count=result)) + async def language_callback(update: Update, _context: ContextTypes.DEFAULT_TYPE): """Handle language selection button clicks.""" query = update.callback_query user_id = query.from_user.id data = query.data - if data.startswith("lang_"): - lang = data.split("_")[1] + if data.startswith('lang_'): + lang = data.split('_')[1] set_user_language(user_id, lang) await query.answer() - await query.edit_message_text( - t(user_id, 'language_changed'), - parse_mode='HTML' - ) + await query.edit_message_text(t(user_id, 'language_changed'), parse_mode='HTML') + # --- Webinar Checking Job --- + def get_webinar_key(url: str) -> str: """Generate unique key for a webinar based on URL.""" return url + def get_stored_webinars() -> list: """Get list of stored webinar keys from Redis.""" data = redis_client.get(KEY_WEBINAR_HISTORY) @@ -709,18 +748,20 @@ def get_stored_webinars() -> list: try: return json.loads(data) except Exception as e: - logger.error(f"Failed to parse webinar history: {e}") + logger.error(f'Failed to parse webinar history: {e}') return [] + def store_webinars(webinar_keys: list): """Store up to 3 most recent webinar keys in Redis.""" # Keep only last 3 webinar_keys = webinar_keys[-3:] try: redis_client.set(KEY_WEBINAR_HISTORY, json.dumps(webinar_keys)) - logger.info(f"Stored {len(webinar_keys)} webinar(s) in history") + logger.info(f'Stored {len(webinar_keys)} webinar(s) in history') except Exception as e: - logger.error(f"Failed to store webinar history: {e}") + logger.error(f'Failed to store webinar history: {e}') + async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): """Background job to check for webinars using Async Playwright. @@ -728,22 +769,22 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): Returns: int: Number of webinars found, or None if check failed """ - logger.info("Running webinar check...") + logger.info('Running webinar check...') phpsessid = redis_client.get(KEY_PHPSESSID) if not phpsessid: - logger.warning("PHPSESSID missing. Skipping check.") + logger.warning('PHPSESSID missing. Skipping check.') # --- DEBUG LOGGING --- try: with open('phpsessid_missing.log', 'a') as f: - f.write(f"[{os.getcwd()}] PHPSESSID missing at {context.job.last_run: %Y-%m-%d %H:%M:%S}\n") + f.write(f'[{os.getcwd()}] PHPSESSID missing at {context.job.last_run: %Y-%m-%d %H:%M:%S}\n') except Exception as e: - logger.error(f"Failed to write PHPSESSID debug log: {e}") + logger.error(f'Failed to write PHPSESSID debug log: {e}') # --------------------- return None current_webinars = [] # List of dicts with name, url, and formatted text - content = "" + content = '' try: async with async_playwright() as p: @@ -755,12 +796,9 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): context_browser = await browser.new_context(user_agent=USER_AGENT) # Add PHPSESSID cookie - await context_browser.add_cookies([{ - 'name': 'PHPSESSID', - 'value': phpsessid, - 'domain': 'edu.edu.vn.ua', - 'path': '/' - }]) + await context_browser.add_cookies( + [{'name': 'PHPSESSID', 'value': phpsessid, 'domain': 'edu.edu.vn.ua', 'path': '/'}] + ) # Create new page page = await context_browser.new_page() @@ -777,8 +815,8 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): content = await page.content() # Check if "no webinar" message is present - if "Жодного онлайн уроку зараз" not in content: - logger.info("!!! WEBINAR FOUND !!!") + if 'Жодного онлайн уроку зараз' not in content: + logger.info('!!! WEBINAR FOUND !!!') # Extract webinar details from table rows rows = page.locator('#meetings table tbody tr') @@ -788,7 +826,7 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): row = rows.nth(i) text = await row.inner_text() - if "Жодного онлайн уроку зараз" not in text: + if 'Жодного онлайн уроку зараз' not in text: # Extract name (topic) from first column name_elem = row.locator('td').nth(0) name = await name_elem.inner_text() @@ -799,17 +837,13 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): url = await url_elem.get_attribute('href') if name and url: - current_webinars.append({ - 'name': name, - 'url': url, - 'text': text.strip() - }) - logger.info(f"Found webinar: {name} -> {url}") + current_webinars.append({'name': name, 'url': url, 'text': text.strip()}) + logger.info(f'Found webinar: {name} -> {url}') else: - logger.info("No webinars found (expected message present)") + logger.info('No webinars found (expected message present)') except Exception as e: - logger.error(f"Error checking page: {e}. Saving content for debug.") + logger.error(f'Error checking page: {e}. Saving content for debug.') # If page content is available, save it on error with contextlib.suppress(Exception): if page and not content: @@ -824,24 +858,24 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): await browser.close() except Exception as e: - logger.error(f"Playwright error: {e}") + logger.error(f'Playwright error: {e}') return None # --- DEBUG LOGGING (Saving last response content) --- - if not current_webinars and content: #if no webinars found, save the page content + if not current_webinars and content: # if no webinars found, save the page content try: with open('response.html', 'w', encoding='utf-8') as f: f.write(content) - logger.info("Saved page content to response.html for debug.") + logger.info('Saved page content to response.html for debug.') except Exception as e: - logger.error(f"Failed to write debug HTML: {e}") + logger.error(f'Failed to write debug HTML: {e}') # ----------------------------------------------------- # Check for NEW webinars and notify if current_webinars: # Get stored webinar history stored_keys = get_stored_webinars() - logger.info(f"Stored webinar keys: {stored_keys}") + logger.info(f'Stored webinar keys: {stored_keys}') # Find new webinars (not in history) new_webinars = [] @@ -853,7 +887,7 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): if key not in stored_keys: new_webinars.append(webinar) - logger.info(f"NEW webinar detected: {webinar['name']}") + logger.info(f'NEW webinar detected: {webinar["name"]}') # Update stored history with current webinars # Merge old and new, keeping only last 5 @@ -863,37 +897,40 @@ async def check_webinars_job(context: ContextTypes.DEFAULT_TYPE): # Notify subscribers ONLY about NEW webinars if new_webinars: subscribers = redis_client.smembers(KEY_SUBSCRIBERS) - logger.info(f"Sending notification about {len(new_webinars)} new webinar(s) to {len(subscribers)} subscriber(s)") + logger.info( + f'Sending notification about {len(new_webinars)} new webinar(s) to {len(subscribers)} subscriber(s)' + ) for sub_id in subscribers: try: # Build message in user's language # sub_id comes from redis set as string, convert to int for translation lookup - webinar_items = "\n\n".join([ - t(int(sub_id), 'webinar_item', name=w['name'], url=w['url']) - for w in new_webinars - ]) + webinar_items = '\n\n'.join( + [t(int(sub_id), 'webinar_item', name=w['name'], url=w['url']) for w in new_webinars] + ) message = t(int(sub_id), 'webinar_found') + webinar_items await context.bot.send_message(chat_id=sub_id, text=message, parse_mode='HTML') - logger.info(f"Notification sent to {sub_id}") + logger.info(f'Notification sent to {sub_id}') except Exception as e: - logger.error(f"Failed to send to {sub_id}: {e}") + logger.error(f'Failed to send to {sub_id}: {e}') else: - logger.info(f"Found {len(current_webinars)} webinar(s), but all are already known") + logger.info(f'Found {len(current_webinars)} webinar(s), but all are already known') return len(current_webinars) + # --- Main --- + def main(): if not WEBINAR_TELEGRAM_TOKEN: - logger.error("WEBINAR_TELEGRAM_TOKEN is missing!") + logger.error('WEBINAR_TELEGRAM_TOKEN is missing!') return # Set default whitelist state if not set if not redis_client.exists(KEY_WHITELIST_ENABLED): - redis_client.set(KEY_WHITELIST_ENABLED, "1") # Enabled by default + redis_client.set(KEY_WHITELIST_ENABLED, '1') # Enabled by default # Add admin to whitelist if ADMIN_ID: @@ -902,26 +939,27 @@ def main(): app = Application.builder().token(WEBINAR_TELEGRAM_TOKEN).build() # Handlers - app.add_handler(CommandHandler("start", start)) - app.add_handler(CommandHandler("stop", stop_command)) - app.add_handler(CommandHandler("help", help_command)) - app.add_handler(CommandHandler("language", language_command)) - app.add_handler(CommandHandler("adduser", add_user)) - app.add_handler(CommandHandler("removeuser", remove_user)) - app.add_handler(CommandHandler("clearhistory", clear_history)) - app.add_handler(CommandHandler("diary", diary_command)) + app.add_handler(CommandHandler('start', start)) + app.add_handler(CommandHandler('stop', stop_command)) + app.add_handler(CommandHandler('help', help_command)) + app.add_handler(CommandHandler('language', language_command)) + app.add_handler(CommandHandler('adduser', add_user)) + app.add_handler(CommandHandler('removeuser', remove_user)) + app.add_handler(CommandHandler('clearhistory', clear_history)) + app.add_handler(CommandHandler('diary', diary_command)) # Callback handlers - diary first, then language selection, then admin panel - app.add_handler(CallbackQueryHandler(diary_callback, pattern="^diary_")) - app.add_handler(CallbackQueryHandler(language_callback, pattern="^lang_")) + app.add_handler(CallbackQueryHandler(diary_callback, pattern='^diary_')) + app.add_handler(CallbackQueryHandler(language_callback, pattern='^lang_')) app.add_handler(CallbackQueryHandler(admin_callback)) # Job Queue job_queue = app.job_queue job_queue.run_repeating(check_webinars_job, interval=WEBINAR_CHECK_INTERVAL, first=10) - logger.info("Bot started polling...") + logger.info('Bot started polling...') app.run_polling() -if __name__ == "__main__": + +if __name__ == '__main__': main() diff --git a/userbot/Dockerfile b/userbot/Dockerfile index 6148d41..50455a7 100644 --- a/userbot/Dockerfile +++ b/userbot/Dockerfile @@ -1,11 +1,15 @@ FROM python:3.11-slim AS builder +SHELL ["/bin/bash", "-o", "pipefail", "-c"] + WORKDIR /src COPY requirements.txt . RUN pip install --no-cache-dir --user -r requirements.txt FROM python:3.11-slim +SHELL ["/bin/bash", "-o", "pipefail", "-c"] + RUN apt-get update && apt-get install -y --no-install-recommends \ git \ mediainfo \ diff --git a/userbot/images/icons.py b/userbot/images/icons.py index 0c9737d..ea0f247 100644 --- a/userbot/images/icons.py +++ b/userbot/images/icons.py @@ -3,7 +3,7 @@ import re from io import BytesIO import requests -from bs4 import BeautifulSoup as BS +from bs4 import BeautifulSoup from pyrogram import Client, filters from pyrogram.errors import RPCError from pyrogram.types import InputMediaPhoto, Message @@ -22,7 +22,7 @@ async def search_icon(_, message: Message): try: html_content = requests.get(url, timeout=10).text - soup = BS(html_content, 'html.parser') + soup = BeautifulSoup(html_content, 'html.parser') results = soup.find_all( 'img', src=re.compile(r'https://cdn-icons-png.flaticon.com/128/[0-9]+/[0-9]+.png'), diff --git a/userbot/images/pinterest.py b/userbot/images/pinterest.py index 355f4b2..7fc7c22 100644 --- a/userbot/images/pinterest.py +++ b/userbot/images/pinterest.py @@ -40,7 +40,7 @@ async def download_image(url): @Client.on_message(filters.command('pinterest', prefix) & filters.me) -async def pinterest_search(client: Client, message: Message): +async def pinterest_search(_client: Client, message: Message): if len(message.command) < 2: await message.edit('Usage: `pinterest [number] `', parse_mode=enums.ParseMode.MARKDOWN) return diff --git a/userbot/images/unsplash2.py b/userbot/images/unsplash2.py index 859a358..5b6c1cb 100644 --- a/userbot/images/unsplash2.py +++ b/userbot/images/unsplash2.py @@ -40,7 +40,7 @@ async def download_image(url): @Client.on_message(filters.command(['unsplash2', 'usp2'], prefix) & filters.me) -async def imgsearch(client: Client, message: Message): +async def imgsearch(_client: Client, message: Message): if len(message.command) < 2: await message.edit('Usage: `img [number] `', parse_mode=enums.ParseMode.MARKDOWN) return diff --git a/userbot/modules/amongus.py b/userbot/modules/amongus.py index 95851ed..594bce6 100644 --- a/userbot/modules/amongus.py +++ b/userbot/modules/amongus.py @@ -10,15 +10,16 @@ from PIL import Image, ImageDraw, ImageFont from pyrogram import Client, filters from pyrogram.types import Message from utils.misc import modules_help, prefix -from utils.scripts import edit_or_reply -from utils.scripts import reply_check as ReplyCheck +from utils.scripts import edit_or_reply, reply_check async def amongus_gen(text: str, clr: int) -> str: url = 'https://github.com/TgCatUB/CatUserbot-Resources/raw/master/Resources/Amongus/' font = ImageFont.truetype( BytesIO( - requests.get('https://github.com/TgCatUB/CatUserbot-Resources/raw/master/Resources/fonts/bold.ttf', timeout=10).content + requests.get( + 'https://github.com/TgCatUB/CatUserbot-Resources/raw/master/Resources/fonts/bold.ttf', timeout=10 + ).content ), 60, ) @@ -97,7 +98,7 @@ async def amongus_cmd(client: Client, message: Message): await client.send_sticker( message.chat.id, imposter_file, - reply_to_message_id=ReplyCheck(message), + reply_to_message_id=reply_check(message), ) @@ -119,7 +120,7 @@ async def imposter_cmd(client: Client, message: Message): await client.send_photo( message.chat.id, imposter_file, - reply_to_message_id=ReplyCheck(message), + reply_to_message_id=reply_check(message), ) diff --git a/userbot/modules/anime/anilist.py b/userbot/modules/anime/anilist.py index b0ea92a..5aef87a 100644 --- a/userbot/modules/anime/anilist.py +++ b/userbot/modules/anime/anilist.py @@ -43,22 +43,22 @@ async def anime_search(client: Client, message: Message): result = response.json() - averageScore = result['averageScore'] + average_score = result['average_score'] try: - coverImage_url = result['imageUrl'] - coverImage = requests.get(url=coverImage_url, timeout=10).content + cover_image_url = result['imageUrl'] + cover_image = requests.get(url=cover_image_url, timeout=10).content async with aiofiles.open('coverImage.jpg', mode='wb') as f: - await f.write(coverImage) + await f.write(cover_image) except Exception: - coverImage = None + cover_image = None title = result['title']['english'] trailer = result['trailer']['id'] description = result['description'] episodes = result['episodes'] genres = ', '.join(result['genres']) - isAdult = result['isAdult'] + is_adult = result['is_adult'] status = result['status'] studios = ', '.join(result['studios']) @@ -68,7 +68,7 @@ async def anime_search(client: Client, message: Message): [ InputMediaPhoto( 'coverImage.jpg', - caption=f"Title: {title}\nAverage Score: {averageScore}\nStatus: {status}\nGenres: {genres}\nEpisodes: {episodes}\nIs Adult: {isAdult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", + caption=f"Title: {title}\nAverage Score: {average_score}\nStatus: {status}\nGenres: {genres}\nEpisodes: {episodes}\nIs Adult: {is_adult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", ) ], ) @@ -81,7 +81,7 @@ async def anime_search(client: Client, message: Message): [ InputMediaPhoto( 'coverImage.jpg', - caption=f"Title: {title}\nAverage Score: {averageScore}\nStatus: {status}\nGenres: {genres}\nEpisodes: {episodes}\nIs Adult: {isAdult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", + caption=f"Title: {title}\nAverage Score: {average_score}\nStatus: {status}\nGenres: {genres}\nEpisodes: {episodes}\nIs Adult: {is_adult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", ) ], ) @@ -109,22 +109,22 @@ async def manga_search(client: Client, message: Message): result = response.json() - averageScore = result['averageScore'] + average_score = result['average_score'] try: - coverImage_url = result['imageUrl'] - coverImage = requests.get(url=coverImage_url, timeout=10).content + cover_image_url = result['imageUrl'] + cover_image = requests.get(url=cover_image_url, timeout=10).content async with aiofiles.open('coverImage.jpg', mode='wb') as f: - await f.write(coverImage) + await f.write(cover_image) except Exception: - coverImage = None + cover_image = None title = result['title']['english'] trailer = result['trailer']['id'] description = result['description'] chapters = result['chapters'] genres = ', '.join(result['genres']) - isAdult = result['isAdult'] + is_adult = result['is_adult'] status = result['status'] studios = ', '.join(result['studios']) @@ -134,7 +134,7 @@ async def manga_search(client: Client, message: Message): [ InputMediaPhoto( 'coverImage.jpg', - caption=f"Title: {title}\nAverage Score: {averageScore}\nStatus: {status}\nGenres: {genres}\nChapters: {chapters}\nIs Adult: {isAdult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", + caption=f"Title: {title}\nAverage Score: {average_score}\nStatus: {status}\nGenres: {genres}\nChapters: {chapters}\nIs Adult: {is_adult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", ) ], ) @@ -147,7 +147,7 @@ async def manga_search(client: Client, message: Message): [ InputMediaPhoto( 'coverImage.jpg', - caption=f"Title: {title}\nAverage Score: {averageScore}\nStatus: {status}\nGenres: {genres}\nChapters: {chapters}\nIs Adult: {isAdult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", + caption=f"Title: {title}\nAverage Score: {average_score}\nStatus: {status}\nGenres: {genres}\nChapters: {chapters}\nIs Adult: {is_adult}\nStudios: {studios}\nDescription: {description}\nTrailer: Click Here", ) ], ) @@ -176,13 +176,13 @@ async def character(client: Client, message: Message): result = response.json() try: - coverImage_url = result['image']['large'] - coverImage = requests.get(url=coverImage_url, timeout=10).content + cover_image_url = result['image']['large'] + cover_image = requests.get(url=cover_image_url, timeout=10).content async with aiofiles.open('coverImage.jpg', mode='wb') as f: - await f.write(coverImage) + await f.write(cover_image) except Exception: - coverImage = None + cover_image = None age = result['age'] description = result['description'] diff --git a/userbot/modules/anime/anime.py b/userbot/modules/anime/anime.py index 0b6554a..efa1180 100644 --- a/userbot/modules/anime/anime.py +++ b/userbot/modules/anime/anime.py @@ -49,7 +49,7 @@ async def random(): @Client.on_message(filters.command(['arnd', 'arandom'], prefix) & filters.me) -async def anime_handler(client: Client, message: Message): +async def anime_handler(_client: Client, message: Message): try: await message.edit('Searching art', parse_mode=enums.ParseMode.HTML) ra = await random() diff --git a/userbot/modules/blackbox.py b/userbot/modules/blackbox.py index 0a6bbf0..5ed3359 100644 --- a/userbot/modules/blackbox.py +++ b/userbot/modules/blackbox.py @@ -11,7 +11,7 @@ def id_generator() -> str: @Client.on_message(filters.command(['bbox', 'blackbox'], prefix) & filters.me) -async def blackbox(client, message): +async def blackbox(_client, message): m = message msg = await m.edit_text('🔍') diff --git a/userbot/modules/cohere.py b/userbot/modules/cohere.py index 4470739..fac15f3 100644 --- a/userbot/modules/cohere.py +++ b/userbot/modules/cohere.py @@ -5,7 +5,7 @@ from utils.scripts import import_library cohere = import_library('cohere') -import cohere # noqa: F811 +import cohere # noqa: F811, E402 co = cohere.Client(cohere_key) diff --git a/userbot/modules/demotivator.py b/userbot/modules/demotivator.py index c1b5d68..6c61672 100644 --- a/userbot/modules/demotivator.py +++ b/userbot/modules/demotivator.py @@ -15,7 +15,9 @@ from PIL import Image, ImageDraw, ImageFont # noqa: E402 @Client.on_message(filters.command(['dem'], prefix) & filters.me) async def demotivator(client: Client, message: Message): await message.edit('Process of demotivation...', parse_mode=enums.ParseMode.HTML) - font = requests.get('https://github.com/The-MoonTg-project/files/blob/main/Times%20New%20Roman.ttf?raw=true', timeout=10) + font = requests.get( + 'https://github.com/The-MoonTg-project/files/blob/main/Times%20New%20Roman.ttf?raw=true', timeout=10 + ) f = font.content template_dem = requests.get('https://raw.githubusercontent.com/files/main/demotivator.png', timeout=10) if message.reply_to_message: diff --git a/userbot/modules/destroy.py b/userbot/modules/destroy.py index 8c23595..95c2f39 100644 --- a/userbot/modules/destroy.py +++ b/userbot/modules/destroy.py @@ -11,7 +11,7 @@ from lottie.importers import importers # noqa: E402 @Client.on_message(filters.command('destroy', prefix) & filters.me) -async def destroy_sticker(client: Client, message: Message): +async def destroy_sticker(_client: Client, message: Message): """Destroy animated stickers by modifying their animation properties""" try: reply = message.reply_to_message diff --git a/userbot/modules/direct.py b/userbot/modules/direct.py index 498a444..6f4ce95 100644 --- a/userbot/modules/direct.py +++ b/userbot/modules/direct.py @@ -30,11 +30,11 @@ def subprocess_run(cmd): executable='bash', ) talk = subproc.communicate() - exitCode = subproc.returncode - if exitCode != 0: + exit_code = subproc.returncode + if exit_code != 0: reply += ( '```An error was detected while running the subprocess:\n' - f'exit code: {exitCode}\n' + f'exit code: {exit_code}\n' f'stdout: {talk[0]}\n' f'stderr: {talk[1]}```' ) diff --git a/userbot/modules/duckduckgo.py b/userbot/modules/duckduckgo.py index 1308dfc..31b59b2 100644 --- a/userbot/modules/duckduckgo.py +++ b/userbot/modules/duckduckgo.py @@ -4,7 +4,7 @@ from utils.misc import prefix @Client.on_message(filters.command('duck', prefix) & filters.me) -async def duckgo(client: Client, message: Message): +async def duckgo(_client: Client, message: Message): input_str = ' '.join(message.command[1:]) sample_url = 'https://duckduckgo.com/?q={}'.format(input_str.replace(' ', '+')) if sample_url: diff --git a/userbot/modules/example.py b/userbot/modules/example.py index ca3da04..78b925b 100644 --- a/userbot/modules/example.py +++ b/userbot/modules/example.py @@ -29,7 +29,7 @@ from utils.misc import modules_help, prefix @Client.on_message(filters.command('example_edit', prefix) & filters.me) -async def example_edit(client: Client, message: Message): +async def example_edit(_client: Client, message: Message): try: await message.edit('This is an example module') except Exception as e: diff --git a/userbot/modules/fliptext.py b/userbot/modules/fliptext.py index b85f36a..c625bca 100644 --- a/userbot/modules/fliptext.py +++ b/userbot/modules/fliptext.py @@ -85,7 +85,7 @@ REPLACEMENT_MAP = { @Client.on_message(filters.command('flip', prefix) & filters.me) -async def flip(client: Client, message: Message): +async def flip(_client: Client, message: Message): text = ' '.join(message.command[1:]) final_str = '' for char in text: diff --git a/userbot/modules/flux.py b/userbot/modules/flux.py index 4244a15..3b9ea80 100644 --- a/userbot/modules/flux.py +++ b/userbot/modules/flux.py @@ -9,9 +9,9 @@ from utils.scripts import format_exc, progress def schellwithflux(args): - API_URL = 'https://randydev-ryuzaki-api.hf.space/api/v1/akeno/fluxai' + api_url = 'https://randydev-ryuzaki-api.hf.space/api/v1/akeno/fluxai' payload = {'user_id': 1191668125, 'args': args} # Please don't edit here - response = requests.post(API_URL, json=payload, timeout=10) + response = requests.post(api_url, json=payload, timeout=10) if response.status_code != 200: print(f'Error status {response.status_code}') return None @@ -19,7 +19,7 @@ def schellwithflux(args): @Client.on_message(filters.command('fluxai', prefix) & filters.me) -async def imgfluxai_(client: Client, message: Message): +async def imgfluxai_(_client: Client, message: Message): question = message.text.split(' ', 1)[1] if len(message.command) > 1 else None if not question: return await message.reply_text('Please provide a question for Flux.') diff --git a/userbot/modules/hearts.py b/userbot/modules/hearts.py index 4b4f128..4bcf68c 100644 --- a/userbot/modules/hearts.py +++ b/userbot/modules/hearts.py @@ -37,20 +37,20 @@ async def _wrap_edit(message: Message, text: str): async def phase1(message: Message): """Big scroll""" - BIG_SCROLL = '🧡💛💚💙💜🖤🤎' + big_scroll = '🧡💛💚💙💜🖤🤎' await _wrap_edit(message, joined_heart) - for heart in BIG_SCROLL: + for heart in big_scroll: await _wrap_edit(message, joined_heart.replace(R, heart)) await asyncio.sleep(SLEEP) async def phase2(message: Message): """Per-heart randomiser""" - ALL = ['❤️'] + list('🧡💛💚💙💜🤎🖤') # don't include white heart + all_hearts = ['❤️'] + list('🧡💛💚💙💜🤎🖤') # don't include white heart format_heart = joined_heart.replace(R, '{}') for _ in range(5): - heart = format_heart.format(*random.choices(ALL, k=heartlet_len)) # noqa: S311 + heart = format_heart.format(*random.choices(all_hearts, k=heartlet_len)) # noqa: S311 await _wrap_edit(message, heart) await asyncio.sleep(SLEEP) @@ -75,7 +75,7 @@ async def phase4(message: Message): @Client.on_message(filters.command('hearts', prefix) & filters.me) -async def hearts(client: Client, message: Message): +async def hearts(_client: Client, message: Message): await phase1(message) await phase2(message) await phase3(message) diff --git a/userbot/modules/loader.py b/userbot/modules/loader.py index 7816d25..9c189f1 100644 --- a/userbot/modules/loader.py +++ b/userbot/modules/loader.py @@ -85,8 +85,8 @@ async def loadmod(_, message: Message): module_name = url.lower() try: f = requests.get( - 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt' - , timeout=10).text + 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt', timeout=10 + ).text except Exception: return await message.edit('Failed to fetch custom modules list') modules_dict = {line.split('/')[-1].split()[0]: line.strip() for line in f.splitlines()} @@ -97,8 +97,9 @@ async def loadmod(_, message: Message): return else: modules_hashes = requests.get( - 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/modules_hashes.txt' - , timeout=10).text + 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/modules_hashes.txt', + timeout=10, + ).text resp = requests.get(url, timeout=10) if not resp.ok: @@ -136,8 +137,8 @@ async def loadmod(_, message: Message): content = f.read() modules_hashes = requests.get( - 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/modules_hashes.txt' - , timeout=10).text + 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/modules_hashes.txt', timeout=10 + ).text if hashlib.sha256(content).hexdigest() not in modules_hashes: os.remove(file_name) @@ -214,7 +215,9 @@ async def load_all_mods(_, message: Message): os.mkdir(f'{BASE_PATH}/modules/custom_modules') try: - f = requests.get('https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt', timeout=10).text + f = requests.get( + 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt', timeout=10 + ).text except Exception: return await message.edit('Failed to fetch custom modules list') modules_list = f.splitlines() @@ -281,14 +284,17 @@ async def updateallmods(_, message: Message): if not module_name.endswith('.py'): continue try: - f = requests.get('https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt', timeout=10).text + f = requests.get( + 'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/full.txt', timeout=10 + ).text except Exception: return await message.edit('Failed to fetch custom modules list') modules_dict = {line.split('/')[-1].split()[0]: line.strip() for line in f.splitlines()} if module_name in modules_dict: resp = requests.get( - f'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/{modules_dict[module_name]}.py' - , timeout=10) + f'https://raw.githubusercontent.com/The-MoonTg-project/custom_modules/main/{modules_dict[module_name]}.py', + timeout=10, + ) if not resp.ok: modules_installed.remove(module_name) continue diff --git a/userbot/modules/misc/autofwd.py b/userbot/modules/misc/autofwd.py index 73b2285..0951653 100644 --- a/userbot/modules/misc/autofwd.py +++ b/userbot/modules/misc/autofwd.py @@ -140,20 +140,20 @@ async def autofwd_main(client: Client, message: Message): target_chats = db.get('custom.autofwd', 'chatto') if source_chats is not None and chat_id in source_chats and target_chats is not None: - for chat in target_chats: + for chat in target_chats: + try: + await message.copy(chat) + except Exception as e: try: - await message.copy(chat) - except Exception as e: - try: - await client.send_message( - 'me', - f'Auto Forwarding Failed for Chat with id: {chat_id} to {chat}\n\n{e}', - ) - except MessageTooLong: - await client.send_message( - 'me', - f'Auto Forwarding Failed for Chat with id: {chat_id} to {chat}, Please check logs!', - ) + await client.send_message( + 'me', + f'Auto Forwarding Failed for Chat with id: {chat_id} to {chat}\n\n{e}', + ) + except MessageTooLong: + await client.send_message( + 'me', + f'Auto Forwarding Failed for Chat with id: {chat_id} to {chat}, Please check logs!', + ) modules_help['autofwd'] = { diff --git a/userbot/modules/misc/backup.py b/userbot/modules/misc/backup.py index f6f33fc..1f69c23 100644 --- a/userbot/modules/misc/backup.py +++ b/userbot/modules/misc/backup.py @@ -63,7 +63,7 @@ async def backup(client: Client, message: Message): @Client.on_message(filters.command(['restore', 'res'], prefix) & filters.me) -async def restore(client: Client, message: Message): +async def restore(_client: Client, message: Message): """ Restore the database """ @@ -98,7 +98,7 @@ async def restore(client: Client, message: Message): @Client.on_message(filters.command(['backupmods', 'bms'], prefix) & filters.me) -async def backupmods(client: Client, message: Message): +async def backupmods(_client: Client, message: Message): """ Backup the modules """ @@ -112,8 +112,8 @@ async def backupmods(client: Client, message: Message): for mod in modules_help: if os.path.isfile(f'modules/custom_modules/{mod}.py'): - f = open(f'backups/{mod}.py', 'wb') - f.write(open(f'modules/custom_modules/{mod}.py', 'rb').read()) + with open(f'backups/{mod}.py', 'wb') as f, open(f'modules/custom_modules/{mod}.py', 'rb') as src: + f.write(src.read()) await message.edit( text='All modules backed up to: backups/ folder', parse_mode=enums.ParseMode.HTML, @@ -123,7 +123,7 @@ async def backupmods(client: Client, message: Message): @Client.on_message(filters.command(['backupmod', 'bm'], prefix) & filters.me) -async def backupmod(client: Client, message: Message): +async def backupmod(_client: Client, message: Message): """ Backup the module """ @@ -142,8 +142,8 @@ async def backupmod(client: Client, message: Message): await message.edit('Backing up module...', parse_mode=enums.ParseMode.HTML) if os.path.isfile(f'modules/custom_modules/{mod}.py'): - f = open(f'backups/{mod}.py', 'wb') - f.write(open(f'modules/custom_modules/{mod}.py', 'rb').read()) + with open(f'backups/{mod}.py', 'wb') as f, open(f'modules/custom_modules/{mod}.py', 'rb') as src: + f.write(src.read()) else: return await message.edit( f'Module {mod} not found.', @@ -159,7 +159,7 @@ async def backupmod(client: Client, message: Message): @Client.on_message(filters.command(['restoremod', 'resmod'], prefix) & filters.me) -async def restoremod(client: Client, message: Message): +async def restoremod(_client: Client, message: Message): """ Restore the module """ @@ -178,11 +178,11 @@ async def restoremod(client: Client, message: Message): await message.edit('Restoring module...', parse_mode=enums.ParseMode.HTML) if os.path.isfile(f'backups/{mod}.py'): - f = open(f'modules/custom_modules/{mod}.py', 'wb') - f.write(open(f'backups/{mod}.py', 'rb').read()) + with open(f'modules/custom_modules/{mod}.py', 'wb') as f, open(f'backups/{mod}.py', 'rb') as src: + f.write(src.read()) else: return await message.edit( - f'Module {mod} not found.', + text='Backup file {mod} not found', parse_mode=enums.ParseMode.HTML, ) await message.edit( @@ -195,7 +195,7 @@ async def restoremod(client: Client, message: Message): @Client.on_message(filters.command(['restoremods', 'resmods'], prefix) & filters.me) -async def restoremods(client: Client, message: Message): +async def restoremods(_client: Client, message: Message): """ Restore the modules """ @@ -209,8 +209,8 @@ async def restoremods(client: Client, message: Message): if mod.endswith('.py'): if os.path.isfile(f'modules/{mod}'): continue - f = open(f'modules/custom_modules/{mod}', 'wb') - f.write(open(f'backups/{mod}', 'rb').read()) + with open(f'modules/custom_modules/{mod}', 'wb') as f, open(f'backups/{mod}', 'rb') as src: + f.write(src.read()) await message.edit( text='All modules restored from: backups/ folder', parse_mode=enums.ParseMode.HTML, diff --git a/userbot/modules/misc/prayer.py b/userbot/modules/misc/prayer.py index bf2d33c..a115c51 100644 --- a/userbot/modules/misc/prayer.py +++ b/userbot/modules/misc/prayer.py @@ -41,7 +41,7 @@ def format_time_12hr(time_str: str) -> str: @Client.on_message(filters.command('prayer', prefix) & filters.me) -async def namaz_times(client: Client, message: Message): +async def namaz_times(_client: Client, message: Message): if message.reply_to_message: city_name = message.reply_to_message.text country_name = DEFAULT_COUNTRY # Default to Pakistan if no country is provided diff --git a/userbot/modules/misc/safone.py b/userbot/modules/misc/safone.py index d5520a6..ba2a44d 100644 --- a/userbot/modules/misc/safone.py +++ b/userbot/modules/misc/safone.py @@ -166,17 +166,17 @@ async def app(client: Client, message: Message): result = response.json() try: - coverImage_url = result['results'][0]['icon'] - coverImage = requests.get(url=coverImage_url, timeout=10).content - async with aiofiles.open('coverImage.jpg', mode='wb') as f: - await f.write(coverImage) + cover_image_url = result['results'][0]['icon'] + cover_image = requests.get(url=cover_image_url, timeout=10).content + async with aiofiles.open('cover_image.jpg', mode='wb') as f: + await f.write(cover_image) except Exception: - coverImage = None + cover_image = None description = result['results'][0]['description'] developer = result['results'][0]['developer'] - IsFree = result['results'][0]['free'] + is_free = result['results'][0]['free'] genre = result['results'][0]['genre'] package_name = result['results'][0]['id'] title = result['results'][0]['title'] @@ -189,8 +189,8 @@ async def app(client: Client, message: Message): chat_id, [ InputMediaPhoto( - 'coverImage.jpg', - caption=f'Title: {title}\nRating: {rating}\nIsFree: {IsFree}\nPrice: {price}\nPackage Name: {package_name}\nGenres: {genre}\nDeveloper: {developer}\nDescription: {description}\nLink: {link}', + 'cover_image.jpg', + caption=f'Title: {title}\nRating: {rating}\nis_free: {is_free}\nPrice: {price}\nPackage Name: {package_name}\nGenres: {genre}\nDeveloper: {developer}\nDescription: {description}\nLink: {link}', ) ], ) @@ -202,16 +202,16 @@ async def app(client: Client, message: Message): chat_id, [ InputMediaPhoto( - 'coverImage.jpg', - caption=f'Title: {title}\nRating: {rating}\nIsFree: {IsFree}\nPrice: {price}\nPackage Name: {package_name}\nGenres: {genre}\nDeveloper: {developer}\nDescription: {description}\nLink: {link}', + 'cover_image.jpg', + caption=f'Title: {title}\nRating: {rating}\nis_free: {is_free}\nPrice: {price}\nPackage Name: {package_name}\nGenres: {genre}\nDeveloper: {developer}\nDescription: {description}\nLink: {link}', ) ], ) except Exception as e: await message.edit_text(format_exc(e)) finally: - if os.path.exists('coverImage.jpg'): - os.remove('coverImage.jpg') + if os.path.exists('cover_image.jpg'): + os.remove('cover_image.jpg') @Client.on_message(filters.command('tsearch', prefix) & filters.me) @@ -232,7 +232,7 @@ async def tsearch(client: Client, message: Message): result = response.json() - coverImage_url = result['results'][0]['thumbnail'] + cover_image_url = result['results'][0]['thumbnail'] description = result['results'][0]['description'] genre = result['results'][0]['genre'] category = result['results'][0]['category'] @@ -264,17 +264,17 @@ async def tsearch(client: Client, message: Message): content=all_results_content, ) - if coverImage_url is not None: - coverImage = requests.get(url=coverImage_url, timeout=10).content - async with aiofiles.open('coverImage.jpg', mode='wb') as f: - await f.write(coverImage) + if cover_image_url is not None: + cover_image = requests.get(url=cover_image_url, timeout=10).content + async with aiofiles.open('cover_image.jpg', mode='wb') as f: + await f.write(cover_image) await message.delete() await client.send_media_group( chat_id, [ InputMediaPhoto( - 'coverImage.jpg', + 'cover_image.jpg', caption=f"Title: {title}\nCategory: {category}\nLanguage: {language}\nSize: {size}\nGenres: {genre}\nDescription: {description}\nMagnet Link: Click Here\nMore Results: Click Here", ) ], @@ -292,7 +292,7 @@ async def tsearch(client: Client, message: Message): chat_id, [ InputMediaPhoto( - 'coverImage.jpg', + 'cover_image.jpg', caption=f"Title: {title}\nCategory: {category}\nLanguage: {language}\nSize: {size}\nGenres: {genre}\nDescription: {description}\nMagnet Link: Click Here", ) ], @@ -308,8 +308,8 @@ async def tsearch(client: Client, message: Message): except Exception as e: await message.edit_text(format_exc(e)) finally: - if os.path.exists('coverImage.jpg'): - os.remove('coverImage.jpg') + if os.path.exists('cover_image.jpg'): + os.remove('cover_image.jpg') @Client.on_message(filters.command('stts', prefix) & filters.me) diff --git a/userbot/modules/python.py b/userbot/modules/python.py index 5e166f7..72d41c1 100644 --- a/userbot/modules/python.py +++ b/userbot/modules/python.py @@ -53,7 +53,7 @@ async def user_exec(_: Client, message: Message): # noinspection PyUnusedLocal @Client.on_message(filters.command(['ev', 'eval'], prefix) & filters.me) -async def user_eval(client: Client, message: Message): +async def user_eval(_client: Client, message: Message): if len(message.command) == 1: await message.edit("Code to eval isn't provided") return diff --git a/userbot/modules/removebg.py b/userbot/modules/removebg.py index db720e1..05e9521 100644 --- a/userbot/modules/removebg.py +++ b/userbot/modules/removebg.py @@ -119,22 +119,23 @@ async def rmbg(client: Client, message: Message): start = datetime.now() await pablo.edit('sending to ReMove.BG') input_file_name = cool - files = { - 'image_file': (input_file_name, open(input_file_name, 'rb')), - } - r = requests.post( - 'https://api.remove.bg/v1.0/removebg', - headers={'X-Api-Key': rmbg_key}, - files=files, - allow_redirects=True, - stream=True, - timeout=10, - ) + with open(input_file_name, 'rb') as f: + files = { + 'image_file': (input_file_name, f), + } + r = requests.post( + 'https://api.remove.bg/v1.0/removebg', + headers={'X-Api-Key': rmbg_key}, + files=files, + allow_redirects=True, + stream=True, + timeout=10, + ) if os.path.exists(cool): os.remove(cool) output_file_name = r - contentType = output_file_name.headers.get('content-type') - if 'image' in contentType: + content_type = output_file_name.headers.get('content-type') + if 'image' in content_type: with io.BytesIO(output_file_name.content) as remove_bg_image: remove_bg_image.name = 'BG_rem.png' await client.send_document(message.chat.id, remove_bg_image, reply_to_message_id=message.id) diff --git a/userbot/modules/socialstalk.py b/userbot/modules/socialstalk.py index f6769df..a154836 100644 --- a/userbot/modules/socialstalk.py +++ b/userbot/modules/socialstalk.py @@ -67,8 +67,9 @@ async def ipinfo(_, message: Message): await m.edit_text('🔎') try: url = requests.get( - f'http://ip-api.com/json/{searchip}?fields=status,message,continent,continentCode,country,countryCode,region,regionName,city,district,zip,lat,lon,timezone,offset,currency,isp,org,as,asname,reverse,mobile,proxy,hosting,query' - , timeout=10) + f'http://ip-api.com/json/{searchip}?fields=status,message,continent,continentCode,country,countryCode,region,regionName,city,district,zip,lat,lon,timezone,offset,currency,isp,org,as,asname,reverse,mobile,proxy,hosting,query', + timeout=10, + ) response = json.loads(url.text) text = f""" IP Address: {response['query']} diff --git a/userbot/modules/spin.py b/userbot/modules/spin.py index dac7778..24843d2 100644 --- a/userbot/modules/spin.py +++ b/userbot/modules/spin.py @@ -102,7 +102,8 @@ async def spin_handler(client: Client, message: Message): elif message.reply_to_message.text: result = await quote_cmd(client, message) filename = 'sticker.png' if result[1] else 'sticker.webp' - open('downloads/' + filename, 'wb').write(result[0].getbuffer()) + with open('downloads/' + filename, 'wb') as f: + f.write(result[0].getbuffer()) coro = False else: filename = 'photo.jpg' diff --git a/userbot/modules/thumbnail.py b/userbot/modules/thumbnail.py index 99764b4..0d8e3c6 100644 --- a/userbot/modules/thumbnail.py +++ b/userbot/modules/thumbnail.py @@ -24,14 +24,14 @@ from utils.misc import modules_help, prefix @Client.on_message(filters.command('setthumb', prefix) & filters.me) async def setthumb(_, message: Message): - THUMB_PATH = 'downloads/thumb' + thumb_path = 'downloads/thumb' if message.reply_to_message: - if not os.path.exists(THUMB_PATH): - os.makedirs(THUMB_PATH) + if not os.path.exists(thumb_path): + os.makedirs(thumb_path) new_thumb = await message.reply_to_message.download() with Image.open(new_thumb) as img: if img.format in ['PNG', 'JPG', 'JPEG']: - new_path = os.path.join(THUMB_PATH, 'thumb.jpg') + new_path = os.path.join(thumb_path, 'thumb.jpg') os.rename(new_thumb, new_path) await message.edit_text('Thumbnail set successfully!') else: diff --git a/userbot/modules/vt.py b/userbot/modules/vt.py index efc70be..30d4bff 100644 --- a/userbot/modules/vt.py +++ b/userbot/modules/vt.py @@ -48,8 +48,9 @@ async def scan_my_file(_, message: Message): url = 'https://www.virustotal.com/vtapi/v2/file/scan' params = {'apikey': vak} - files = {'file': (downloaded_file_name, open(downloaded_file_name, 'rb'))} - response = requests.post(url, files=files, params=params, timeout=10) + with open(downloaded_file_name, 'rb') as f: + files = {'file': (downloaded_file_name, f)} + response = requests.post(url, files=files, params=params, timeout=10) try: r_json = response.json() md5 = r_json['md5'] @@ -104,9 +105,10 @@ async def scan_my_large_file(_, message: Message): url = upl_data - files = {'file': (downloaded_file_name, open(downloaded_file_name, 'rb'))} - headers = {'accept': 'application/json', 'x-apikey': vak} - response = requests.post(url, files=files, headers=headers, timeout=10) + with open(downloaded_file_name, 'rb') as f: + files = {'file': (downloaded_file_name, f)} + headers = {'accept': 'application/json', 'x-apikey': vak} + response = requests.post(url, files=files, headers=headers, timeout=10) r_json = response.json() analysis_url = r_json['data']['links']['self'] diff --git a/userbot/utils/conv.py b/userbot/utils/conv.py index b56e909..cf8854b 100644 --- a/userbot/utils/conv.py +++ b/userbot/utils/conv.py @@ -23,7 +23,7 @@ from pyrogram.handlers import MessageHandler class _TrueFilter(filters.Filter): - async def __call__(self, client: Client, update: types.Message): + async def __call__(self, _client: Client, _update: types.Message): return True diff --git a/userbot/utils/db.py b/userbot/utils/db.py index 90a2dc2..672e817 100644 --- a/userbot/utils/db.py +++ b/userbot/utils/db.py @@ -149,7 +149,7 @@ class SqliteDatabase(Database): self._lock.release() def get(self, module: str, variable: str, default=None): - sql = f"SELECT * FROM '{module}' WHERE var=?" + sql = f"SELECT * FROM '{module}' WHERE var=?" # noqa: S608 cur = self._execute(module, sql, (variable,)) row = cur.fetchone() @@ -162,7 +162,7 @@ class SqliteDatabase(Database): INSERT INTO '{module}' VALUES ( ?, ?, ? ) ON CONFLICT (var) DO UPDATE SET val=?, type=? WHERE var=? - """ + """ # noqa: S608 if isinstance(value, bool): val = '1' if value else '0' @@ -183,7 +183,7 @@ class SqliteDatabase(Database): return True def remove(self, module: str, variable: str): - sql = f"DELETE FROM '{module}' WHERE var=?" + sql = f"DELETE FROM '{module}' WHERE var=?" # noqa: S608 self._execute(module, sql, (variable,)) self._conn.commit() @@ -192,7 +192,7 @@ class SqliteDatabase(Database): if not re.match(pattern, module): raise ValueError(f'Invalid module name format: {module}') - sql = f"SELECT * FROM '{module}'" + sql = f"SELECT * FROM '{module}'" # noqa: S608 cur = self._execute(module, sql) collection = {} diff --git a/userbot/utils/rentry.py b/userbot/utils/rentry.py index 32c6aae..c46ceb5 100644 --- a/userbot/utils/rentry.py +++ b/userbot/utils/rentry.py @@ -138,14 +138,14 @@ async def paste( print(f'URL: {url} - Edit Code: {edit_code} - Time: {ftime}') - rallUrls = db.get('core.rentry', 'urls', default={'allUrls': {}}) + rall_urls = db.get('core.rentry', 'urls', default={'allUrls': {}}) entry_id = str(uuid4()) - rallUrls['allUrls'][entry_id] = { + rall_urls['allUrls'][entry_id] = { 'url': short_url, 'edit_code': edit_code, 'time': ftime, } - db.set('core.rentry', 'urls', rallUrls) + db.set('core.rentry', 'urls', rall_urls) if return_edit: return (url, edit_code) @@ -156,19 +156,19 @@ async def rentry_cleanup_job(): """Periodically checks and deletes rentry pastes older than 24 hours""" while True: try: - rallUrls = db.get('core.rentry', 'urls', default={'allUrls': {}}) + rall_urls = db.get('core.rentry', 'urls', default={'allUrls': {}}) now = datetime.now() deleted_count = 0 error_count = 0 - for entry_id, entry in list(rallUrls['allUrls'].items()): + for entry_id, entry in list(rall_urls['allUrls'].items()): url = entry['url'] entry_time = datetime.strptime(entry['time'], '%d %I:%M:%S %p %Y') if now - entry_time > timedelta(days=1): try: delete(url, entry['edit_code']) - del rallUrls['allUrls'][entry_id] + del rall_urls['allUrls'][entry_id] deleted_count += 1 print(f'[#] Deleted expired rentry paste: {url}') except Exception as e: @@ -179,7 +179,7 @@ async def rentry_cleanup_job(): print(f'[*] Cleanup summary: {deleted_count} deleted, {error_count} failed') if deleted_count: - db.set('core.rentry', 'urls', rallUrls) + db.set('core.rentry', 'urls', rall_urls) except Exception as e: print(f'[!] Error in rentry cleanup job: {str(e)}')