refactor app

app/core/bot.py

@@ -0,0 +1,79 @@
+import asyncio
+
+from aiogram import Bot, types
+from aiogram.contrib.middlewares.logging import LoggingMiddleware
+from aiogram.dispatcher import Dispatcher
+from aiogram.dispatcher.webhook import SendMessage
+from aiogram.utils.callback_data import CallbackData
+
+from core.parse_web import parse_site, configure_firefox_driver
+from settings import API_TOKEN
+
+bot = Bot(token=API_TOKEN)
+dispatcher = Dispatcher(bot)
+dispatcher.middleware.setup(LoggingMiddleware())
+
+driver = configure_firefox_driver()
+
+stations_cb = CallbackData('station', 'direction')
+
+
+def get_keyboard() -> types.InlineKeyboardMarkup:
+    """
+    Generate keyboard with list of posts
+    """
+    markup = types.InlineKeyboardMarkup()
+
+    markup.row(
+        types.InlineKeyboardButton('Дом -> Офис', callback_data=stations_cb.new(direction='home->office')),
+        types.InlineKeyboardButton('Офис -> Дом', callback_data=stations_cb.new(direction='office->home')),
+    )
+    return markup
+
+
+@dispatcher.callback_query_handler(stations_cb.filter(direction='home->office'))
+async def home_office(query: types.CallbackQuery, callback_data: dict[str, str]) -> SendMessage:
+
+    text = parse_site(
+        driver=driver,
+        url='https://yandex.ru/maps/213/moscow/stops/stop__9640740/'
+            '?l=masstransit&ll=37.527754%2C55.823507&tab=overview&z=21',
+        message='Остановка Б. Академическая ул, д. 15'
+    )
+
+    return SendMessage(query.message.chat.id, text, reply_markup=get_keyboard())
+
+
+@dispatcher.callback_query_handler(stations_cb.filter(direction='office->home'))
+async def office_home(query: types.CallbackQuery, callback_data: dict[str, str]) -> SendMessage:
+
+    text = parse_site(
+        driver=driver,
+        url='https://yandex.ru/maps/213/moscow/stops/stop__9640288/?'
+            'l=masstransit&ll=37.505338%2C55.800160&tab=overview&z=211',
+        message='Остановка Улица Алабяна'
+    )
+    return SendMessage(query.message.chat.id, text, reply_markup=get_keyboard())
+
+
+@dispatcher.message_handler(commands=['chatid'])
+async def chat_id(message: types.Message) -> SendMessage:
+
+    return SendMessage(message.chat.id, message.chat.id)
+
+
+@dispatcher.message_handler()
+async def echo(message: types.Message) -> SendMessage:
+    return SendMessage(message.chat.id, 'Выбери остановку', reply_markup=get_keyboard())
+
+
+async def send_message(chat_ids: list[int]) -> None:
+    text = parse_site(
+        driver=driver,
+        url='https://yandex.ru/maps/213/moscow/stops/stop__9640740/'
+            '?l=masstransit&ll=37.527754%2C55.823507&tab=overview&z=21',
+        message='Остановка Б. Академическая ул, д. 15'
+    )
+    await asyncio.gather(
+        *[bot.send_message(chat_id=chat_id, text=text, parse_mode=types.ParseMode.HTML) for chat_id in chat_ids]
+    )

app/core/logger.py

@@ -0,0 +1,7 @@
+import sys
+from loguru import logger
+
+logger.remove()
+logger.add(sink=sys.stdout, colorize=True, level='DEBUG',
+           format="<cyan>{time:DD.MM.YYYY HH:mm:ss}</cyan> | <level>{level}</level> | "
+                  "<magenta>{message}</magenta>")

app/core/parse_web.py

@@ -0,0 +1,70 @@
+import os
+import tarfile
+import time
+from pathlib import Path
+
+import wget
+from selenium import webdriver
+from selenium.common.exceptions import NoSuchElementException
+from selenium.webdriver.firefox import options
+from selenium.webdriver.firefox.service import Service
+from selenium.webdriver.firefox.webdriver import WebDriver
+
+from core.logger import logger
+from settings import BASE_DIR, GECKO_DRIVER_VERSION
+
+
+def download_gecko_driver():
+    gecko_driver = f'https://github.com/mozilla/geckodriver/releases/download/v{GECKO_DRIVER_VERSION}/' \
+                   f'geckodriver-v{GECKO_DRIVER_VERSION}-linux64.tar.gz'
+
+    if not Path(f'{BASE_DIR}/geckodriver').exists():
+        logger.info(f'Downloading gecodriver v {GECKO_DRIVER_VERSION}...')
+        geckodriver_file = wget.download(url=gecko_driver, out=BASE_DIR)
+
+        with tarfile.open(geckodriver_file) as tar:
+            tar.extractall(BASE_DIR)
+        os.remove(f'{BASE_DIR}/geckodriver-v{GECKO_DRIVER_VERSION}-linux64.tar.gz')
+        logger.info(f'\ngeckodriver has been downloaded to folder {BASE_DIR}')
+
+
+def configure_firefox_driver(private_window: bool = False) -> WebDriver:
+    opt = options.Options()
+    opt.headless = True
+    opt.add_argument('-profile')
+    opt.add_argument(f'{Path.home()}/snap/firefox/common/.mozilla/firefox')
+    if private_window:
+        opt.set_preference("browser.privatebrowsing.autostart", True)
+    service = Service(executable_path=f'{BASE_DIR}/geckodriver')
+    firefox_driver = webdriver.Firefox(service=service, options=opt)
+
+    return firefox_driver
+
+
+def parse_site(driver: WebDriver, url: str, message: str) -> str:
+    driver.get(url)
+    time.sleep(4)
+    elements = driver.find_elements(by='class name', value='masstransit-vehicle-snippet-view')
+
+    bus_300, bus_t19 = None, None
+    bus_300_arrival, bus_t19_arrival = None, None
+
+    for element in elements:
+        try:
+            bus_300 = element.find_element(by='css selector', value='[aria-label="300"]')
+            bus_300_arrival = element.find_element(by='class name', value='masstransit-prognoses-view__title-text')
+        except NoSuchElementException:
+            pass
+        try:
+            bus_t19 = element.find_element(by='css selector', value='[aria-label="т19"]')
+            bus_t19_arrival = element.find_element(by='class name', value='masstransit-prognoses-view__title-text')
+        except NoSuchElementException:
+            pass
+    answer = f'{message}\n\n'
+    if not all([bus_300, bus_t19]):
+        return 'Автобусов 300 или Т19 не найдено. \n\nСмотри на карте :)'
+    if bus_300:
+        answer += f'Автобус {bus_300.text} - {bus_300_arrival.text}\n'
+    if bus_t19:
+        answer += f'Автобус {bus_t19.text} - {bus_t19_arrival.text}'
+    return answer

app/core/scheduler.py

@@ -0,0 +1,20 @@
+from core.bot import send_message
+
+cron_jobs = [
+        {'trigger': 'cron', 'day_of_week': 'mon-fri', 'hour': 8, 'minute': 59, 'second': 0},
+        {'trigger': 'cron', 'day_of_week': 'mon-fri', 'hour': 9, 'minute': 4, 'second': 0},
+        {'trigger': 'cron', 'day_of_week': 'mon-fri', 'hour': 9, 'minute': 9, 'second': 0},
+    ]
+
+user_chat_ids = {'chat_ids': [417070387,  # me
+                              431571617,  # Lenok
+                              ]}
+
+
+def asyncio_schedule() -> None:
+    from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+    scheduler = AsyncIOScheduler()
+    for cron in cron_jobs:
+        scheduler.add_job(send_message, kwargs=user_chat_ids, **cron)
+    scheduler.start()

app/main.py

@@ -0,0 +1,39 @@
+from aiogram.utils.executor import start_webhook
+
+from core.bot import bot, dispatcher
+from core.logger import logger
+from core.parse_web import download_gecko_driver
+from core.scheduler import asyncio_schedule
+from settings import WEBHOOK_URL, WEBHOOK_PATH, WEBAPP_HOST, WEBAPP_PORT
+
+
+async def on_startup(dispatcher) -> None:
+    await bot.set_webhook(WEBHOOK_URL)
+    asyncio_schedule()
+
+
+async def on_shutdown(dispatcher):
+    logger.warning('Shutting down..')
+
+    # Remove webhook (not acceptable in some cases)
+    await bot.delete_webhook()
+
+    # Close DB connection (if used)
+    await dispatcher.storage.close()
+    await dispatcher.storage.wait_closed()
+
+    logger.warning('Bye!')
+
+
+if __name__ == '__main__':
+    download_gecko_driver()
+
+    start_webhook(
+        dispatcher=dispatcher,
+        webhook_path=WEBHOOK_PATH,
+        on_startup=on_startup,
+        on_shutdown=on_shutdown,
+        skip_updates=True,
+        host=WEBAPP_HOST,
+        port=WEBAPP_PORT,
+    )

app/settings.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+from decouple import AutoConfig
+
+# Build paths inside the project like this: BASE_DIR.joinpath('some')
+# `pathlib` is better than writing: dirname(dirname(dirname(__file__)))
+BASE_DIR = Path(__file__).parent.parent
+
+# Loading `.env` files
+# See docs: https://gitlab.com/mkleehammer/autoconfig
+env_path = BASE_DIR.joinpath('config')
+
+config = AutoConfig(search_path=env_path)
+
+
+GECKO_DRIVER_VERSION = config('GECKO_DRIVER_VERSION')
+BASE_DIR = Path(__file__).parent.resolve().as_posix()
+
+API_TOKEN = config('API_TOKEN')
+
+# webhook settings
+WEBHOOK_HOST = config('WEBHOOK_HOST')
+WEBHOOK_PATH = config('WEBHOOK_PATH')
+WEBHOOK_URL = f"{WEBHOOK_HOST}{WEBHOOK_PATH}"
+
+# webserver settings
+WEBAPP_HOST = config('WEBAPP_HOST')  # or ip
+WEBAPP_PORT = config('WEBAPP_PORT', cast=int)