Merge branch 'feature/new-responses' into 'main'

extract links, delete incoming msg, add sender to reply

See merge request ollyhearn/nosibakabot!1

.gitignore

@@ -1 +1,164 @@
 **/secret.py
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/latest/usage/project/#working-with-version-control
+.pdm.toml
+.pdm-python
+.pdm-build/
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+  "python.testing.unittestArgs": ["-v", "-s", "./bot", "-p", "*_test.py"],
+  "python.testing.pytestEnabled": false,
+  "python.testing.unittestEnabled": true
+}

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)