fix regex query (#15)

Co-authored-by: Jan Brugger <[email protected]>

src/queries.py

@@ -53,4 +53,7 @@ def __init__(self, query: str):
         self._regex_query = re.compile(query.removeprefix("r/").strip(), flags=re.IGNORECASE)
 
     def matches(self, text: str) -> bool:
-        return bool(re.search(self._regex_query, text))
+        try:
+            return bool(re.search(self._regex_query, text))
+        except re.error:
+            return False

src/telegram/patterns.py

@@ -1,4 +1,4 @@
 ALLOWED_CHARACTERS = r"\ &,!\w\[\]+-"
-QUERY_PATTERN = rf"^(r\/)?[{ALLOWED_CHARACTERS}]+$"
-QUERY_PATTERN_LIMITED_CHARS = rf"^(r\/)?[{ALLOWED_CHARACTERS}]{{1,60}}$"
+QUERY_PATTERN = rf"^[{ALLOWED_CHARACTERS}]+$"
+QUERY_PATTERN_LIMITED_CHARS = rf"^[{ALLOWED_CHARACTERS}]{{1,60}}$"
 PRICE_PATTERN = r"^\d+([,\.]\d{1,2})?$"

src/telegram/routers/error_router.py

@@ -5,11 +5,12 @@
 from aiogram import Bot, Router
 from aiogram.exceptions import TelegramAPIError
 from aiogram.filters import ExceptionTypeFilter
-from aiogram.types import ErrorEvent, Message
+from aiogram.types import CallbackQuery, ErrorEvent, Message
 
 from src import config
 from src.exceptions import NotificationNotFoundError
 from src.telegram.messages import Messages
+from src.telegram.routers import overwrite_or_answer
 
 error_router = Router()
 
@@ -18,27 +19,16 @@
 
 @error_router.error(ExceptionTypeFilter(NotificationNotFoundError))
 async def notification_not_found(error: ErrorEvent) -> bool:
-    if (callback_query := error.update.callback_query) and callback_query.message:
-        message = callback_query.message
-    elif error.update.message:
-        message = error.update.message
-    else:
-        return False
+    telegram_object = error.update.callback_query or error.update.message
 
-    error_text = Messages.notification_not_found()
+    if not telegram_object:
+        return False
 
-    if isinstance(message, Message):
+    if isinstance(telegram_object, CallbackQuery) and isinstance(telegram_object.message, Message):
         with suppress(TelegramAPIError):
-            await message.delete_reply_markup()
-
-        try:
-            await message.edit_text(text=error_text)
-        except TelegramAPIError:
-            pass
-        else:
-            return True
+            await telegram_object.message.delete_reply_markup()
 
-    await message.answer(text=error_text)
+    await overwrite_or_answer(telegram_object, Messages.notification_not_found())
 
     return True
 

src/telegram/routers/notification_router.py

@@ -26,7 +26,7 @@
 from src.telegram.patterns import PRICE_PATTERN, QUERY_PATTERN, QUERY_PATTERN_LIMITED_CHARS
 from src.telegram.routers import get_id, overwrite_or_answer, store_id
 from src.telegram.states import States
-from src.utils import prettify_query
+from src.utils import is_valid_regex_query, prettify_query
 
 if TYPE_CHECKING:
     from aiogram.fsm.context import FSMContext
@@ -47,6 +47,7 @@ async def new_notification(callback_query: CallbackQuery, state: FSMContext) ->
 
 
 @notification_router.message(States.ADD_NOTIFICATION, F.text.regexp(QUERY_PATTERN))
+@notification_router.message(States.ADD_NOTIFICATION, F.text.func(is_valid_regex_query))
 @notification_router.callback_query(AddNotificationCB.filter())
 async def add_notification(
     telegram_object: Message | CallbackQuery,
@@ -97,8 +98,10 @@ async def update_query(
 
 
 @notification_router.message(States.UPDATE_QUERY, F.text.regexp(QUERY_PATTERN))
+@notification_router.message(States.UPDATE_QUERY, F.text.func(is_valid_regex_query))
 async def process_update_query(message: Message, state: FSMContext) -> None:
     if not message.text:
+        logger.error("Empty message text in process update should not be possible")
         return
 
     notification = NotificationClient().update_query(
@@ -224,14 +227,14 @@ def __message_to_price(price_str: str | None) -> int:
 
 
 @notification_router.message(F.text.regexp(QUERY_PATTERN_LIMITED_CHARS))
+@notification_router.message(F.text.func(is_valid_regex_query))
 async def add_notification_inconclusive(message: Message) -> None:
-    text = message.text
-
-    if not text:
+    if not message.text:
+        logger.error("Empty message text in inconclusive update should not be possible")
         return
 
     await overwrite_or_answer(
         message,
-        Messages.add_notification_inconclusive(text),
-        reply_markup=Keyboards.add_notification_inconclusive(text),
+        Messages.add_notification_inconclusive(message.text),
+        reply_markup=Keyboards.add_notification_inconclusive(message.text),
     )

src/utils.py

@@ -5,8 +5,20 @@
 from src.models import PriceModel
 
 
-def prettify_query(query: str) -> str:
+def is_valid_regex_query(query: str) -> bool:
     if query.startswith("r/"):
+        try:
+            re.compile(query.lstrip("r/"))
+        except re.error:
+            pass
+        else:
+            return True
+
+    return False
+
+
+def prettify_query(query: str) -> str:
+    if is_valid_regex_query(query):
         return query
 
     query = " ".join(query.split()).lower()  # remove unnecessary whitespaces and lower