Skip to content

Instantly share code, notes, and snippets.

@Tishka17
Last active November 8, 2020 11:16
Show Gist options
  • Save Tishka17/7e49bc8dc46e8905a5a9d09ada768bee to your computer and use it in GitHub Desktop.
Save Tishka17/7e49bc8dc46e8905a5a9d09ada768bee to your computer and use it in GitHub Desktop.
Bot main and config
[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
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"]),
)
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