Last active
November 8, 2020 11:16
-
-
Save Tishka17/7e49bc8dc46e8905a5a9d09ada768bee to your computer and use it in GitHub Desktop.
Bot main and config
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
[tg_bot] | |
token = | |
admin_id = | |
support_chat_id = | |
service_chat_id = | |
homework_chat_id = | |
use_redis = true | |
[tg_admin] | |
api_id = | |
api_hash = | |
[tinkoff] | |
terminal_key = | |
password = | |
callback_url = | |
[db] | |
user = postgres | |
password = postgres | |
database = postgres | |
host = 127.0.0.1 | |
[new_db] | |
user = postgres | |
password = postgres | |
database = postgres | |
host = 127.0.0.1 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import configparser | |
from dataclasses import dataclass | |
@dataclass | |
class DbConfig: | |
host: str | |
password: str | |
user: str | |
database: str | |
@dataclass | |
class TinkoffClient: | |
terminal_key: str | |
password: str | |
callback_url: str | |
@dataclass | |
class TgBot: | |
token: str | |
support_chat_id: int | |
admin_id: int | |
service_chat_id: int | |
use_redis: bool | |
@dataclass | |
class TgAdmin: | |
api_id: int | |
api_hash: str | |
@dataclass | |
class Config: | |
tg_bot: TgBot | |
tg_admin: TgAdmin | |
db: DbConfig | |
new_db: DbConfig | |
tinkoff: TinkoffClient | |
def cast_bool(value: str) -> bool: | |
if not value: | |
return False | |
return value.lower() in ("true", "t", "1", "yes") | |
def load_config(path: str): | |
config = configparser.ConfigParser() | |
config.read(path) | |
tg_bot = config["tg_bot"] | |
tg_admin = config["tg_admin"] | |
return Config( | |
tg_bot=TgBot( | |
token=tg_bot["token"], | |
support_chat_id=int(tg_bot["support_chat_id"]), | |
admin_id=int(tg_bot["admin_id"]), | |
service_chat_id=int(tg_bot["service_chat_id"]), | |
use_redis=cast_bool(tg_bot.get("use_redis")), | |
), | |
db=DbConfig(**config["db"]), | |
new_db=DbConfig(**config["new_db"]), | |
tinkoff=TinkoffClient(**config["tinkoff"]), | |
tg_admin=TgAdmin(int(tg_admin["api_id"]), tg_admin["api_hash"]), | |
) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import asyncio | |
import logging | |
import pyrogram | |
from aiogram import Bot, Dispatcher | |
from aiogram.contrib.fsm_storage.memory import MemoryStorage | |
from aiogram.contrib.fsm_storage.redis import RedisStorage | |
from aiopg.sa import create_engine | |
from coursebot.config import load_config | |
from coursebot.service.chat_cleaner import ChatCleaner | |
from coursebot.service.tinkoff_client import TinkoffClient | |
from coursebot.tg_handlers.admin import register_admin | |
from coursebot.tg_handlers.cleaner import register_cleaner | |
from coursebot.tg_handlers.homework import register_homework | |
from coursebot.tg_handlers.user import register_user | |
from coursebot.tg_handlers.utils.permissions import PermissionsFilter | |
from coursebot.tg_middlewares.db import DbMiddleware | |
from coursebot.tg_middlewares.last_message import LastMessageSaver | |
logger = logging.getLogger(__name__) | |
async def main(): | |
logging.basicConfig( | |
level=logging.INFO, | |
format="%(asctime)s - %(levelname)s - %(name)s - %(message)s", | |
) | |
logger.error("Starting bot") | |
config = load_config("bot.ini") | |
if config.tg_bot.use_redis: | |
storage = RedisStorage() | |
else: | |
storage = MemoryStorage() | |
engine = await create_engine( | |
user=config.db.user, | |
password=config.db.password, | |
database=config.db.database, | |
host=config.db.host, | |
echo=False, | |
) | |
tinkoff = TinkoffClient(config.tinkoff.terminal_key, config.tinkoff.password, config.tinkoff.callback_url) | |
client = pyrogram.Client("account", config.tg_admin.api_id, config.tg_admin.api_hash) | |
chat_cleaner = ChatCleaner(client, config.tg_bot.admin_id) | |
register_cleaner(client, chat_cleaner, engine) | |
asyncio.create_task(client.start()) | |
bot = Bot(token=config.tg_bot.token) | |
dp = Dispatcher(bot, storage=storage) | |
dp.middleware.setup(DbMiddleware(engine, tinkoff)) | |
dp.middleware.setup(LastMessageSaver()) | |
permissions = PermissionsFilter(config.tg_bot.admin_id) | |
register_support(dp, config.tg_bot.support_chat_id) | |
register_admin(dp, client, chat_cleaner, config.tg_bot.service_chat_id, permissions) | |
register_user(dp, veratag, config.tg_bot.service_chat_id) | |
# start | |
await dp.start_polling() | |
if __name__ == '__main__': | |
asyncio.run(main()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment