clean up code

src/_dependencies/vk_api_client.py

@@ -15,7 +15,7 @@ def __init__(self, token: str):
 
     def send(
         self,
-        user_id: str,
+        user_id: int | str,
         random_id: int,
         message: str = '',
         lat: str = '',

src/vk_bot/_utils/bot_polling.py

@@ -12,7 +12,6 @@
 from pydantic import BaseModel
 from sqlalchemy.engine.base import Connection, Engine
 from vk_api.longpoll import Event, VkEventType, VkLongPoll
-from vk_api.vk_api import VkApiMethod
 
 from _dependencies.commons import get_app_config, sqlalchemy_get_pool
 from _dependencies.telegram_api_wrapper import make_invite_text_for_user
@@ -34,8 +33,7 @@ class EventObject(BaseModel):
 
 class UpdateEvent(BaseModel):
     type: str
-    object: EventObject
-    group_id: int
+    object: EventObject | None
 
 
 @dataclass
@@ -47,7 +45,7 @@ def connect(self) -> Iterator[Connection]:
         with self._pool.connect() as connection:
             yield connection
 
-    def set_user_vk_id(self, telegram_user_id: int, vk_id: str) -> None:
+    def set_user_vk_id(self, telegram_user_id: int, vk_id: int) -> None:
         """Write user's VKontakte id"""
 
         with self.connect() as connection:
@@ -64,42 +62,15 @@ def db() -> 'DBClient':
     return DBClient(_pool=sqlalchemy_get_pool())
 
 
-def process_incoming_message(vk: VkApiMethod, event: Event) -> None:
-    logging.info('processing message %s', event)
-    if event.from_me:
+def process_incoming_message(event: UpdateEvent) -> None:
+    logging.info('processing event %s', event)
+    if not event.object:
         return
 
-    msg = event.text.lower()
-
-    telegram_user_id, secret = get_invite_from_message(msg)
-    logging.info('got invite from user %s', telegram_user_id)
-    vk.messages.mark_as_read(peer_id=event.peer_id)
-    if not telegram_user_id:
+    if event.type != 'message_new':
         return
 
-    if make_invite_text_for_user(telegram_user_id).lower() != msg:
-        logging.warning(f'invite hash wrong for telegram user {telegram_user_id}')
-        return
-
-    random_id = random.randint(0, 10_000_000)
-    vk_user_id = event.user_id
-    try:
-        logging.info('connecting user %s', telegram_user_id)
-        db().set_user_vk_id(telegram_user_id, vk_user_id)
-        message = (
-            'Ваш акаунт связан с аккаунтом в Телеграм. \n'
-            'Вы будете получать здесь уведомления о поисках с теми же настройками, которые заданы в Телеграм'
-        )
-        vk.messages.send(user_id=vk_user_id, message=message, random_id=random_id)
-
-    except:
-        logging.exception("can't connect VK and Telegram users")
-        vk.messages.send(user_id=vk_user_id, message='не удалось привязать пользователя Телеграм', random_id=random_id)
-
-
-def process_incoming_message_2(event: UpdateEvent) -> None:
-    logging.info('processing message %s', event)
-    if event.object.message.from_id == event.group_id:
+    if not event.object.message:
         return
 
     msg = event.object.message.text.lower()
@@ -124,14 +95,12 @@ def process_incoming_message_2(event: UpdateEvent) -> None:
             'Вы будете получать здесь уведомления о поисках с теми же настройками, которые заданы в Телеграм'
         )
         get_default_vk_api_client().send(user_id=vk_user_id, message=message, random_id=random_id)
-        # vk.messages.send(user_id=vk_user_id, message=message, random_id=random_id)
 
     except:
         logging.exception("can't connect VK and Telegram users")
         get_default_vk_api_client().send(
             user_id=vk_user_id, message='не удалось привязать пользователя Телеграм', random_id=random_id
         )
-        # vk.messages.send(user_id=vk_user_id, message='не удалось привязать пользователя Телеграм', random_id=random_id)
 
 
 def get_invite_from_message(message: str) -> tuple[int, str] | tuple[None, None]:
@@ -151,8 +120,19 @@ def get_invite_from_message(message: str) -> tuple[int, str] | tuple[None, None]
 
 def run_polling() -> None:
     vk_session = vk_api.VkApi(token=get_app_config().vk_api_key)
-    vk = vk_session.get_api()
     longpoll = VkLongPoll(vk_session)
     for event in longpoll.listen():
-        if event.type == VkEventType.MESSAGE_NEW:
-            process_incoming_message(vk, event)
+        if event.type == VkEventType.MESSAGE_NEW and not event.from_me:
+            json_ = {
+                'type': 'message_new',
+                'object': {
+                    'message': {
+                        'from_id': event.from_user,
+                        'text': event.text,
+                        'peer_id': event.from_user,
+                    },
+                },
+            }
+            event_data = UpdateEvent.model_validate(json_)
+
+            process_incoming_message(event_data)

src/vk_bot/cli.py

@@ -1,9 +1,3 @@
-from _dependencies.misc import (
-    RequestWrapper,
-    ResponseWrapper,
-    request_response_converter,
-)
-
 from ._utils.bot_polling import run_polling
 from .main import main_raw
 

src/vk_bot/main.py

@@ -1,40 +1,31 @@
 import logging
 import random
 from contextlib import suppress
-from functools import lru_cache
 from typing import Any
 
-import vk_api
-from pydantic import BaseModel
-from sqlalchemy.engine.base import Connection, Engine
-from vk_api.longpoll import Event, VkEventType, VkLongPoll
-from vk_api.vk_api import VkApiMethod
-
-from _dependencies.commons import get_app_config, setup_logging, sqlalchemy_get_pool
+from _dependencies.commons import get_app_config, setup_logging
 from _dependencies.misc import (
     RequestWrapper,
     ResponseWrapper,
     request_response_converter,
 )
-from _dependencies.telegram_api_wrapper import make_invite_text_for_user
 
-from ._utils.bot_polling import UpdateEvent, process_incoming_message, process_incoming_message_2
+from ._utils.bot_polling import UpdateEvent, process_incoming_message, run_polling
 
 setup_logging(__package__)
 random.seed()
 
 
-@lru_cache
-def _get_vk_session() -> vk_api.VkApi:
-    vk_session = vk_api.VkApi(token=get_app_config().vk_api_key)
-    return vk_session
+def main_raw(request: dict) -> str:
+    with suppress(Exception):
+        if request['type'] == 'confirmation' and request['group_id'] == 237036024:
+            # confirmation, run once
+            return get_app_config().vk_confirmation_code
 
+    event = UpdateEvent.model_validate(request)
+    process_incoming_message(event)
 
-@lru_cache
-def _get_longpoll() -> VkLongPoll:
-    vk_session = _get_vk_session()
-    longpoll = VkLongPoll(vk_session)
-    return longpoll
+    return 'ok'
 
 
 @request_response_converter
@@ -43,25 +34,6 @@ def main(request: RequestWrapper, *args: Any, **kwargs: Any) -> ResponseWrapper:
     return ResponseWrapper(data=main_raw(request.json_))  # type:ignore[arg-type]
 
 
-def main_raw(request: dict) -> str:
-    logging.info(request)
-
-    with suppress(Exception):
-        if request['type'] == 'confirmation' and request['group_id'] == 237036024:
-            # confirmation, run once
-            return get_app_config().vk_confirmation_code
-
-    event = UpdateEvent.model_validate(request)
-    process_incoming_message_2(event)
-    # Cannot convert Flask request to Event. So we'll use endpoint just as trigger to run polling.
-    # logging.info('getting session')
-    # vk_session = _get_vk_session()
-    # vk = vk_session.get_api()
-
-    # logging.info('checking updates')
-    # for event in _get_longpoll().check():
-    #     logging.info('got event %s', event)
-    #     if event.type == VkEventType.MESSAGE_NEW:
-    #         process_incoming_message(vk, event)
-
-    return 'ok'
+# if __name__ == '__main__':
+#     run_polling()
+#     # run_flask()