extract links, delete incoming msg, add sender to reply

bot/bot.py

@@ -7,13 +7,15 @@ import re
 from urllib.parse import urlparse, urlunparse
 import random
 
-HAS_LINK_RE = r'(https?:\/\/[^\s]+|www\.[^\s]+)'
+from msgprocessor import TrackerRemovalMsgProcessor, TrackerRemovalProcessorMessage
+
+HAS_LINK_RE = r"(https?:\/\/[^\s]+|www\.[^\s]+)"
 
 bot = AsyncTeleBot(TOKEN)
 
 
 def extract_links(text: str):
-    url_pattern = r'(https?://[^\s]+|www\.[^\s]+)'
+    url_pattern = r"(https?://[^\s]+|www\.[^\s]+)"
     links = re.findall(url_pattern, text)
     return links
 
@@ -45,10 +47,30 @@ async def start(msg: Message):
 
 @bot.message_handler(func=lambda message: True)
 async def got_message(msg: Message):
-    if re.match(string=msg.text, pattern=HAS_LINK_RE):
-        fixed_reply = process_text(msg.text)
-        if fixed_reply:
-            await bot.reply_to(msg, fixed_reply)
+    # god i love nones as fuck
+    if msg.text is None:
+        return
+    if msg.from_user is None:
+        return
+
+    tracker_removal_result = TrackerRemovalMsgProcessor(
+        TrackerRemovalProcessorMessage(fromUser=msg.from_user, text=msg.text)
+    ).process()
+
+    if not tracker_removal_result.needsToReply:
+        return
+
+    try:
+        await bot.delete_message(msg.chat.id, msg.id, timeout=5)
+    except Exception as e:
+        await bot.reply_to(
+            message=msg.id,
+            text="Uoghhhh, i am not an admin here? I can't cleanup this tracking(",
+        )
+        print(e, flush=True)  # todo: логгер
+        return
+
+    await bot.send_message(msg.chat.id, tracker_removal_result.text, parse_mode="html")
 
 
 async def main():

bot/exception.py

@@ -0,0 +1,9 @@
+class UrlRemoverNotImplementedException(Exception):
+    def __init__(self, domain: str):
+        self.__base_message = "Url remover for domain not implemented"
+        self.domain = domain
+        super().__init__(self.__base_message)
+
+    
+    def __str__(self):
+        return f'{self.__base_message}: {self.domain}'

bot/msgprocessor.py

@@ -0,0 +1,116 @@
+from typing import Callable
+from dataclasses import dataclass
+from exception import UrlRemoverNotImplementedException
+from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
+import re
+from telebot.types import User
+
+
+@dataclass(init=True, eq=True)
+class TrackerRemovalProcessorMessage:
+    fromUser: User
+    text: str
+
+
+@dataclass(frozen=True, init=True, eq=True)
+class TrackerRemovalResult:
+    needsToReply: bool
+    text: str
+
+
+class TrackerRemovalMsgProcessor:
+    def __init__(self, msg: TrackerRemovalProcessorMessage):
+        self.__msg = msg
+
+    def process(self) -> TrackerRemovalResult:
+        if not self.__remove_trackers_from_msg_urls():
+            return TrackerRemovalResult(
+                needsToReply=False, text=""
+            )  # дальнейшие трансформации смысла не имеют
+        self.__emplace_sender_into_msg_text()
+        return TrackerRemovalResult(
+            needsToReply=True, text=self.__msg.text
+        )  # сообщение было изменено, нужно ответ отослать
+
+    def __remove_trackers_from_msg_urls(self) -> bool:
+        trackers_extracted = False
+
+        # todo: вообще мы работаем с http и это юзкейс обскьюрный
+        # но ссылка может быть и без указания схемы, телега может распарсить
+        # просто строку через точки и в конце какой то домен верхнего уровня как ссылку
+        def is_url(url: str) -> bool:
+            SCHEMES = ["http://", "https://"]
+            return len([s for s in SCHEMES if url.startswith(s)]) != 0
+
+        SEPARATOR_CHARS = [" ", "\n"]
+        separator_regex = "(" + "|".join(SEPARATOR_CHARS) + ")"
+        lexems = re.split(separator_regex, self.__msg.text)
+        for i, l in enumerate(lexems):
+            if not is_url(l):
+                continue
+
+            removed_trackers_url = self.__remove_tracker(l)
+            if l == removed_trackers_url:  # изменений урла не было
+                continue
+
+            trackers_extracted = True
+            lexems[i] = removed_trackers_url
+
+        self.__msg.text = "".join(lexems)
+        return trackers_extracted
+
+    @staticmethod
+    def __remove_tracker(url: str) -> str:
+        try:
+            parsed_url = urlparse(url)
+        except Exception:
+            return url
+        if parsed_url.hostname is None:
+            return url
+        hostname = str(parsed_url.hostname)
+        try:
+            return TrackerRemoverFactory.make_remover(hostname)(url)
+        except UrlRemoverNotImplementedException:
+            return url
+
+    def __emplace_sender_into_msg_text(self):
+        self.__msg.text = f'Message from <a href="tg://user?id={self.__msg.fromUser.id}">{self.__msg.fromUser.first_name}</a>:\n\n{self.__msg.text}'
+
+
+class TrackerRemoverFactory:
+    TrackerRemover = Callable[[str], str]
+
+    @staticmethod
+    def make_remover(domain: str) -> TrackerRemover:
+        @dataclass(frozen=True, init=True)
+        class RemoverIdentifyer:
+            domains: list[str]
+            remover: TrackerRemoverFactory.TrackerRemover
+
+        removers_by_domain = [
+            RemoverIdentifyer(
+                domains=["youtube.com", "youtu.be"],
+                remover=TrackerRemoverFactory.remove_yt_trackers,
+            )
+        ]
+        remover_one = [
+            r
+            for r in removers_by_domain
+            if len([d for d in r.domains if domain.endswith(d)]) != 0
+        ]
+        if len(remover_one) == 0:
+            raise UrlRemoverNotImplementedException(domain)
+        return remover_one[0].remover
+
+    @staticmethod
+    def remove_yt_trackers(url: str) -> str:
+        # todo: подумать как обобщить, мб билдер стратегии поиска трекера
+        # но эт сильно на потом
+        QUERY_PARAMS_TRACKER = "si"
+        parsed_url = urlparse(url)
+        query_params = parse_qs(parsed_url.query)
+        if QUERY_PARAMS_TRACKER in query_params:
+            del query_params[QUERY_PARAMS_TRACKER]
+        return urlunparse(
+            parsed_url._replace(query=urlencode(query_params, doseq=True))
+        )

bot/msgprocessor_test.py

@@ -0,0 +1,124 @@
+import unittest
+from msgprocessor import (
+    TrackerRemovalMsgProcessor,
+    TrackerRemoverFactory,
+    TrackerRemovalProcessorMessage,
+    TrackerRemovalResult,
+)
+from dataclasses import dataclass
+
+
+@dataclass(init=True, eq=True)
+class TestUser:
+    id: int
+    first_name: str
+
+
+class TestRemoverFactory(unittest.TestCase):
+    factory = TrackerRemoverFactory()
+
+    def test_remove_strategy_constructor(self):
+        test_case_data = [
+            {"domain": "youtube.com", "remover": self.factory.remove_yt_trackers},
+            {
+                "domain": "lowerlevel.youtube.com",
+                "remover": self.factory.remove_yt_trackers,
+            },
+            {
+                "domain": "youtu.be",
+                "remover": self.factory.remove_yt_trackers,
+            },
+            {
+                "domain": "something.youtu.be",
+                "remover": self.factory.remove_yt_trackers,
+            },
+        ]
+        for test_case in test_case_data:
+            self.assertIs(
+                self.factory.make_remover(test_case["domain"]),
+                self.factory.remove_yt_trackers,
+            )
+
+    def test_remove_yt_si(self):
+        test_case_data = [
+            {
+                "url": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy",
+                "expected_url": "https://youtu.be/jNQXAC9IVRw",
+            },
+            {
+                "url": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy&t=16",
+                "expected_url": "https://youtu.be/jNQXAC9IVRw?t=16",
+            },
+            {
+                "url": "https://www.youtube.com/watch?v=jNQXAC9IVRw",
+                "expected_url": "https://www.youtube.com/watch?v=jNQXAC9IVRw",
+            },
+            {
+                "url": "http://www.youtube.com/watch?v=jNQXAC9IVRw&si=qLIZT1rvs99_jbgy&t=16",
+                "expected_url": "http://www.youtube.com/watch?v=jNQXAC9IVRw&t=16",
+            },
+        ]
+        for test_case in test_case_data:
+            self.assertEqual(
+                self.factory.remove_yt_trackers(test_case["url"]),
+                test_case["expected_url"],
+            )
+
+
+class TestRemovalMsgProcessor(unittest.TestCase):
+    def test_remove_links(self):
+        test_case_data = [
+            {
+                "msg_text": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy",
+                "sender": TestUser(id=123, first_name="Ghytro"),
+                "bot_responded": True,
+                "bot_response": 'Message from <a href="tg://user?id=123">Ghytro</a>:\n\nhttps://youtu.be/jNQXAC9IVRw',
+            },
+            {
+                "msg_text": "чекай https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy\nнаш слон хд",
+                "sender": TestUser(id=321, first_name="OllyHearn"),
+                "bot_responded": True,
+                "bot_response": 'Message from <a href="tg://user?id=321">OllyHearn</a>:\n\nчекай https://youtu.be/jNQXAC9IVRw\nнаш слон хд',
+            },
+            {
+                "msg_text": "а я такая нитакуся без si ссылки шлю сразу https://youtu.be/jNQXAC9IVRw и по нескольку штук\nhttp://www.youtube.com/watch?v=jNQXAC9IVRw&si=qLIZT1rvs99_jbgy&t=16 дада",
+                "sender": TestUser(id=321, first_name="OllyHearn"),
+                "bot_responded": True,
+                "bot_response": 'Message from <a href="tg://user?id=321">OllyHearn</a>:\n\nа я такая нитакуся без si ссылки шлю сразу https://youtu.be/jNQXAC9IVRw и по нескольку штук\nhttp://www.youtube.com/watch?v=jNQXAC9IVRw&t=16 дада',
+            },
+            {
+                "msg_text": "asdasdasdasdasdasdasd asdasd asdasd asdad sasa dadsas",
+                "sender": TestUser(id=123, first_name="Ghytro"),
+                "bot_responded": False,
+                "bot_response": "",
+            },
+        ]
+        for test_case in test_case_data:
+            result = TrackerRemovalMsgProcessor(
+                TrackerRemovalProcessorMessage(
+                    fromUser=test_case["sender"], text=test_case["msg_text"]
+                )
+            ).process()
+            self.assertEqual(
+                result,
+                TrackerRemovalResult(
+                    needsToReply=test_case["bot_responded"],
+                    text=test_case["bot_response"],
+                ),
+            )
+
+
+if __name__ == "__main__":
+    test_classes_to_run = [TestRemoverFactory, TestRemovalMsgProcessor]
+
+    loader = unittest.TestLoader()
+
+    suites_list = []
+    for test_class in test_classes_to_run:
+        suite = loader.loadTestsFromTestCase(test_class)
+        suites_list.append(suite)
+
+    big_suite = unittest.TestSuite(suites_list)
+
+    runner = unittest.TextTestRunner()
+    results = runner.run(big_suite)