-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
80 lines (66 loc) · 2.47 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from apscheduler.job import Job
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from sqlalchemy import Engine, create_engine
from sqlalchemy.pool import QueuePool
from config_data.config import Config, load_config
from handlers import admin_handlers, user_handlers
from middlewares.middleware import DBMiddleware
from models.models import Base
from utils.utils import checking_apps
logger = logging.getLogger(__name__)
logging.basicConfig(
level=logging.DEBUG,
format='[{asctime}] #{levelname:8} {filename}:'
'{lineno} - {name} - {message}',
style='{'
)
def set_scheduled_jobs(scheduler: AsyncIOScheduler, engine: Engine, bot: Bot,
config: Config) -> Job:
"""Устанавливает задачи для планировщика."""
logger.debug('func. set_scheduled_jobs has worked.')
return scheduler.add_job(
checking_apps,
"interval",
minutes=int(config.interval_value.minutes),
args=(engine, bot)
)
async def main() -> None:
"""Точка входа в бота."""
try:
logger.debug('Starting bot...')
config: Config = load_config()
scheduler = AsyncIOScheduler()
bot = Bot(
token=config.tg_bot.token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML))
engine = create_engine(
'sqlite:///data/sqlite3.db',
poolclass=QueuePool,
pool_size=5,
max_overflow=10
)
Base.metadata.create_all(engine)
dp = Dispatcher()
dp.include_router(admin_handlers.router)
dp.include_router(user_handlers.router)
dp.update.outer_middleware(DBMiddleware())
job = set_scheduled_jobs(scheduler, engine, bot, config)
dp.workflow_data.update({
'db': engine, 'bot': bot, 'config': config, 'job': job,
'scheduler': scheduler})
await bot.delete_webhook(drop_pending_updates=True)
scheduler.start()
await dp.start_polling(bot)
logger.debug('Bot started')
except Exception as error:
logger.error(f'Ошибка в работе программы: {error}', exc_info=True)
if __name__ == '__main__':
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info('Bot has stopped by admin')