sinc_with_api

migrations/env.py

@@ -1,12 +1,10 @@
 from logging.config import fileConfig
 
 from alembic import context
 from sqlalchemy import engine_from_config, pool
-
 from src.db import Base
 from src.settings import settings
 
-
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
 config = context.config
@@ -60,7 +58,7 @@
 
     """
     configuration = config.get_section(config.config_ini_section)
-    configuration['sqlalchemy.url'] = str(settings.DB_DSN)
+    configuration["sqlalchemy.url"] = str(settings.DB_DSN)
     connectable = engine_from_config(
         configuration,
         prefix="sqlalchemy.",

migrations/versions/75dab872e82b_init.py

@@ -1,17 +1,16 @@
 """Init
 
 Revision ID: 75dab872e82b
-Revises: 
+Revises:
 Create Date: 2023-02-25 19:05:46.473426
 
 """
 
 import sqlalchemy as sa
 from alembic import op
 
-
 # revision identifiers, used by Alembic.
-revision = '75dab872e82b'
+revision = "75dab872e82b"
 down_revision = None
 branch_labels = None
 depends_on = None
@@ -20,16 +19,16 @@
 def upgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
     op.create_table(
-        'vk_user',
-        sa.Column('vk_id', sa.BIGINT(), nullable=False),
-        sa.Column('surname', sa.String(), nullable=False),
-        sa.Column('number', sa.String(), nullable=False),
-        sa.PrimaryKeyConstraint('vk_id'),
+        "vk_user",
+        sa.Column("vk_id", sa.BIGINT(), nullable=False),
+        sa.Column("surname", sa.String(), nullable=False),
+        sa.Column("number", sa.String(), nullable=False),
+        sa.PrimaryKeyConstraint("vk_id"),
     )
     # ### end Alembic commands ###
 
 
 def downgrade() -> None:
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_table('vk_user')
+    op.drop_table("vk_user")
     # ### end Alembic commands ###

src/__main__.py

@@ -1,14 +1,21 @@
+import asyncio
 import logging
 
 from src.handlers import event_loop
-
+from src.settings import get_settings, sync_from_server
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 
-logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO, datefmt="%Y-%m-%d %H:%M:%S")
+logging.basicConfig(
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    level=logging.INFO,
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
+    asyncio.run(sync_from_server())
+    settings = get_settings()
     logging.info("=== BOT START ===")
     while True:
         event_loop()

src/answers.py

@@ -1,70 +1,87 @@
 # Marakulin Andrey @annndruha
 # 2023
 
 from dataclasses import dataclass
 
+from src.settings import get_settings
+
+settings = get_settings()
+
 
 @dataclass
 class Answers:
-    hey: str = 'Привет!'
-    not_auth = 'Не авторизовано'
-    inst = 'Инструкция'
-    conf = 'Конфиденциальность'
+    hey: str = "Привет!"
+    not_auth = "Не авторизовано"
+    inst = "Инструкция"
+    conf = "Конфиденциальность"
 
     conf_full = (
-        '❗️ Файлы, которые вы отправляете через бота, будут храниться в течение нескольких месяцев '
-        'на сервере в Москве, а также в этом чате ВКонтакте.'
-        '\nДоступ к файлам имеет узкий круг лиц, ответственных за работоспособность сервиса печати.'
-        '\nМы НЕ рекомендуем использовать данный сервис для печати конфиденциальных документов!'
+        "❗️ Файлы, которые вы отправляете через бота, будут храниться в течение нескольких месяцев "
+        "на сервере в Москве, а также в этом чате ВКонтакте."
+        "\nДоступ к файлам имеет узкий круг лиц, ответственных за работоспособность сервиса печати."
+        "\nМы НЕ рекомендуем использовать данный сервис для печати конфиденциальных документов!"
     )
 
-    ask_help = ['привет', 'помогите', 'помощь', 'help', 'как ', 'инструкция', 'начать', 'Начать', 'старт', 'start']
+    ask_help = [
+        "привет",
+        "помогите",
+        "помощь",
+        "help",
+        "как ",
+        "инструкция",
+        "начать",
+        "Начать",
+        "старт",
+        "start",
+    ]
     help = (
-        'Я вк-бот бесплатного принтера профкома студентов физического факультета МГУ!\n\n'
-        '❔ Отправьте PDF файл и получите PIN для печати. Поддерживаются только .PDF файлы размером не более 3МБ.'
-        '\nС этим PIN необходимо подойти к принтеру и ввести его в терминал печати. '
-        'Либо отсканировать QR-код на принтере с помощью ссылки. После этого начнётся печать.'
-        '\n\n💻 Бот разработан группой программистов профкома, как и приложение Твой ФФ!'
-        '\nВ приложении вы сможете найти больше настроек печати, расписание и много других возможностей.'
-        '\nТак же есть Telegram-бот для печати.'
+        "Я вк-бот бесплатного принтера профкома студентов физического факультета МГУ!\n\n"
+        "❔ Отправьте PDF файл и получите PIN для печати. Поддерживаются только .PDF файлы размером не более 3МБ."
+        "\nС этим PIN необходимо подойти к принтеру и ввести его в терминал печати. "
+        "Либо отсканировать QR-код на принтере с помощью ссылки. После этого начнётся печать."
+        "\n\n💻 Бот разработан группой программистов профкома, как и приложение Твой ФФ!"
+        "\nВ приложении вы сможете найти больше настроек печати, расписание и много других возможностей."
+        "\nТак же есть Telegram-бот для печати."
     )
 
-    err_bd = '❌ Ошибка базы данных. Попробуйте позже.'
-    err_print = '❌ Ошибка сервера печати. Попробуйте позже.'
-    err_payload = '🐩 Похоже бот обновился.\nВыполните команду /start'
-    err_vk = '🐩 Ошибка взаимодействия с ВКонтакте. Попробуйте позже.'
-    err_kb = '🐩 Ошибка клавиатуры, что-то пошло не так'
-    err_fatal = '🐩 Глубоко внутри меня что-то сломалось...'
+    err_bd = "❌ Ошибка базы данных. Попробуйте позже."
+    err_print = "❌ Ошибка сервера печати. Попробуйте позже."
+    err_payload = "🐩 Похоже бот обновился.\nВыполните команду /start"
+    err_vk = "🐩 Ошибка взаимодействия с ВКонтакте. Попробуйте позже."
+    err_kb = "🐩 Ошибка клавиатуры, что-то пошло не так"
+    err_fatal = "🐩 Глубоко внутри меня что-то сломалось..."
 
-    val_pass = '✅ Поздравляю! Проверка пройдена и данные сохранены для этого аккаунта ВК. Можете присылать pdf.'
-    val_name = 'Иванов\n1234567'
-    val_update_pass = '✅ Поздравляю! Проверка пройдена и данные обновлены.'
-    val_already = 'Вы уже успешно авторизованы. Можете присылать файл на печать.'
-    val_addition = '\n\nНо для начала нужно авторизоваться. Нажмите на кнопку ниже:'
+    val_pass = "✅ Поздравляю! Проверка пройдена и данные сохранены для этого аккаунта ВК. Можете присылать pdf."
+    val_name = "Иванов\n1234567"
+    val_update_pass = "✅ Поздравляю! Проверка пройдена и данные обновлены."
+    val_already = "Вы уже успешно авторизованы. Можете присылать файл на печать."
+    val_addition = "\n\nНо для начала нужно авторизоваться. Нажмите на кнопку ниже:"
     val_need = (
-        '⚠ Для использования принтера необходимо авторизоваться.'
-        '\nВведите фамилию и номер профсоюзного билета в формате:'
+        "⚠ Для использования принтера необходимо авторизоваться."
+        "\nВведите фамилию и номер профсоюзного билета в формате:"
     )
     val_fail = (
-        '❌ Проверка не пройдена. Удостоверьтесь что вы состоите в профсоюзе и правильно ввели данные.'
-        '\n\nВведите фамилию и номер профсоюзного билета в формате:'
+        "❌ Проверка не пройдена. Удостоверьтесь что вы состоите в профсоюзе и правильно ввели данные."
+        "\n\nВведите фамилию и номер профсоюзного билета в формате:"
     )
     val_fail_format = (
-        'Я вас не понимаю. Если вы хотите обновить данные авторизации введите фамилию и '
-        'номер профсоюзного билета в формате:'
+        "Я вас не понимаю. Если вы хотите обновить данные авторизации введите фамилию и "
+        "номер профсоюзного билета в формате:"
     )
     val_unknown_message = (
-        'Сообщение не распознано. Если вы хотите обновить данные авторизации введите фамилию и '
-        'номер профсоюзного билета в формате:'
+        "Сообщение не распознано. Если вы хотите обновить данные авторизации введите фамилию и "
+        "номер профсоюзного билета в формате:"
     )
 
-    warn_filesize = '⚠️ Файл слишком большой. Мы принимаем файлы размером ДО 3 МБ'
-    warn_many_files = '⚠️ Файлов слишком много. Прикрепите только один файл pdf.'
-    warn_unreadable_file = '⚠️ Я не смог прочитать файл. Проверьте его целостность и формат, я работаю только с pdf.'
-    warn_only_pdfs = '⚠️ Я умею печатать только документы в формате pdf.'
+    warn_filesize = f"⚠️ Файл слишком большой. Мы принимаем файлы размером ДО {settings.MAX_PDF_SIZE_MB} МБ"
+    warn_many_files = "⚠️ Файлов слишком много. Прикрепите только один файл pdf."
+    warn_unreadable_file = f"⚠️ Я не смог прочитать файл. Проверьте его целостность и формат, я работаю только с {settings.CONTENT_TYPES}."
+    warn_only_pdfs = (
+        f"⚠️ Я умею печатать только документы в формате {settings.CONTENT_TYPES}."
+    )
 
     send_to_print = '✅ Файл "{}" успешно загружен. Для печати подойдите к принтеру и введите PIN: \n\n{}'
-    qr_button_text = '📷 Печать по QR'
+    qr_button_text = "📷 Печать по QR"
 
 
 ans = Answers()

src/auth.py

@@ -1,24 +1,30 @@
 # Marakulin Andrey @annndruha
 # 2023
 
 import requests
-from sqlalchemy.orm import Session
-
 import src.vk as vk
+from sqlalchemy.orm import Session
 from src.db import Session, VkUser
-from src.settings import settings
+from src.settings import get_settings
+
+settings = get_settings()
 
 
 def check_union_member(user: vk.EventUser, surname, number) -> None | tuple:
-    r = requests.get(url=settings.PRINT_URL + '/is_union_member', params=dict(surname=surname, number=number, v=1))
+    r = requests.get(
+        url=settings.PRINT_URL + "/is_union_member",
+        params=dict(surname=surname, number=number, v=1),
+    )
     if r.json():
         return user.user_id, surname, number
     return None
 
 
 def check_user_in_db(user: vk.EventUser) -> None | tuple:
     with Session() as session:
-        data: VkUser | None = session.query(VkUser).filter(VkUser.vk_id == user.user_id).one_or_none()
+        data: VkUser | None = (
+            session.query(VkUser).filter(VkUser.vk_id == user.user_id).one_or_none()
+        )
     if data is not None:
         return user.user_id, data.surname, data.number
     return None
@@ -30,10 +36,13 @@
     :return: db_requisites tuple or None if user not authenticated
     """
     with Session() as session:
-        data: VkUser | None = session.query(VkUser).filter(VkUser.vk_id == user.user_id).one_or_none()
+        data: VkUser | None = (
+            session.query(VkUser).filter(VkUser.vk_id == user.user_id).one_or_none()
+        )
     if data is not None:
         r = requests.get(
-            url=settings.PRINT_URL + '/is_union_member', params=dict(surname=data.surname, number=data.number, v=1)
+            url=settings.PRINT_URL + "/is_union_member",
+            params=dict(surname=data.surname, number=data.number, v=1),
         )
         if r.json():
             return user.user_id, data.surname, data.number
@@ -48,7 +57,9 @@
 
 def update_user(user: vk.EventUser, surname, number) -> None:
     with Session() as session:
-        data: VkUser | None = session.query(VkUser).filter(VkUser.vk_id == user.user_id).one_or_none()
+        data: VkUser | None = (
+            session.query(VkUser).filter(VkUser.vk_id == user.user_id).one_or_none()
+        )
         data.surname = surname
         data.number = number
         session.commit()

src/db.py

@@ -5,8 +5,6 @@
 from sqlalchemy.ext.declarative import as_declarative, declared_attr
 from sqlalchemy.orm import Mapped, mapped_column, sessionmaker
 
-from src.settings import settings
-
 
 @as_declarative()
 class Base:
@@ -24,7 +22,7 @@ def __repr__(self) -> str:
         attrs = []
         for c in self.__table__.columns:
             attrs.append(f"{c.name}={getattr(self, c.name)}")
-        return "{}({})".format(self.__class__.__name__, ', '.join(attrs))
+        return "{}({})".format(self.__class__.__name__, ", ".join(attrs))
 
 
 class VkUser(Base):
@@ -33,5 +31,7 @@ class VkUser(Base):
     number: Mapped[int] = mapped_column(sqlalchemy.String, nullable=False)
 
 
-engine = create_engine(url=str(settings.DB_DSN), pool_pre_ping=True, isolation_level="AUTOCOMMIT")
+engine = create_engine(
+    url=str(settings.DB_DSN), pool_pre_ping=True, isolation_level="AUTOCOMMIT"
+)
 Session = sessionmaker(bind=engine)