olly/nosibakabot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

moved to user' object & fixes

Browse Source
This commit is contained in:
Artem Reznichenko
2024-11-07 22:25:32 +03:00
parent b5e6cfc7f8
commit dc4abee6bc
4 changed files with 63 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.vscode/settings.json vendored Normal file
View File

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

27
bot/bot.py
View File

@ -5,19 +5,17 @@ from secret import TOKEN
import textbook import textbook
import re import re
from urllib.parse import urlparse, urlunparse from urllib.parse import urlparse, urlunparse
from msgprocessor import (
TrackerRemovalMsgProcessor,
TrackerRemovalProcessorMessage
)
import random 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) bot = AsyncTeleBot(TOKEN)
def extract_links(text: str): 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) links = re.findall(url_pattern, text)
return links return links
@ -54,26 +52,25 @@ async def got_message(msg: Message):
return return
if msg.from_user is None: if msg.from_user is None:
return return
if msg.from_user.username is None:
return
tracker_removal_result = TrackerRemovalMsgProcessor( tracker_removal_result = TrackerRemovalMsgProcessor(
TrackerRemovalProcessorMessage( TrackerRemovalProcessorMessage(fromUser=msg.from_user, text=msg.text)
fromUsername=msg.from_user.username,
text=msg.text
)
).process() ).process()
if not tracker_removal_result.needsToReply: if not tracker_removal_result.needsToReply:
return return
try: try:
await bot.delete_message(msg.chat.id, msg.id, 5) await bot.delete_message(msg.chat.id, msg.id, timeout=5)
except Exception as e: except Exception as e:
print(e) # todo: логгер 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 return
await bot.send_message(msg.chat.id, tracker_removal_result.text) await bot.send_message(msg.chat.id, tracker_removal_result.text, parse_mode="html")
async def main(): async def main():

22
bot/msgprocessor.py
View File

@ -3,11 +3,12 @@ from dataclasses import dataclass
from exception import UrlRemoverNotImplementedException from exception import UrlRemoverNotImplementedException
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import re import re
from telebot.types import User
@dataclass(init=True, eq=True) @dataclass(init=True, eq=True)
class TrackerRemovalProcessorMessage: class TrackerRemovalProcessorMessage:
fromUsername: str fromUser: User
text: str text: str
@ -23,12 +24,17 @@ class TrackerRemovalMsgProcessor:
def process(self) -> TrackerRemovalResult: def process(self) -> TrackerRemovalResult:
if not self.__remove_trackers_from_msg_urls(): if not self.__remove_trackers_from_msg_urls():
return TrackerRemovalResult(needsToReply=False, text="") # дальнейшие трансформации смысла не имеют return TrackerRemovalResult(
needsToReply=False, text=""
) # дальнейшие трансформации смысла не имеют
self.__emplace_sender_into_msg_text() self.__emplace_sender_into_msg_text()
return TrackerRemovalResult(needsToReply=True, text=self.__msg.text) # сообщение было изменено, нужно ответ отослать return TrackerRemovalResult(
needsToReply=True, text=self.__msg.text
) # сообщение было изменено, нужно ответ отослать
def __remove_trackers_from_msg_urls(self) -> bool: def __remove_trackers_from_msg_urls(self) -> bool:
trackers_extracted = False trackers_extracted = False
# todo: вообще мы работаем с http и это юзкейс обскьюрный # todo: вообще мы работаем с http и это юзкейс обскьюрный
# но ссылка может быть и без указания схемы, телега может распарсить # но ссылка может быть и без указания схемы, телега может распарсить
# просто строку через точки и в конце какой то домен верхнего уровня как ссылку # просто строку через точки и в конце какой то домен верхнего уровня как ссылку
@ -37,14 +43,14 @@ class TrackerRemovalMsgProcessor:
return len([s for s in SCHEMES if url.startswith(s)]) != 0 return len([s for s in SCHEMES if url.startswith(s)]) != 0
SEPARATOR_CHARS = [" ", "\n"] SEPARATOR_CHARS = [" ", "\n"]
separator_regex = "("+"|".join(SEPARATOR_CHARS)+")" separator_regex = "(" + "|".join(SEPARATOR_CHARS) + ")"
lexems = re.split(separator_regex, self.__msg.text) lexems = re.split(separator_regex, self.__msg.text)
for i, l in enumerate(lexems): for i, l in enumerate(lexems):
if not is_url(l): if not is_url(l):
continue continue
removed_trackers_url = self.__remove_tracker(l) removed_trackers_url = self.__remove_tracker(l)
if l == removed_trackers_url: # изменений урла не было if l == removed_trackers_url: # изменений урла не было
continue continue
trackers_extracted = True trackers_extracted = True
@ -68,7 +74,7 @@ class TrackerRemovalMsgProcessor:
return url return url
def __emplace_sender_into_msg_text(self): def __emplace_sender_into_msg_text(self):
self.__msg.text = f'Message from @{self.__msg.fromUsername}:\n\n{self.__msg.text}' 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: class TrackerRemoverFactory:
@ -106,7 +112,5 @@ class TrackerRemoverFactory:
if QUERY_PARAMS_TRACKER in query_params: if QUERY_PARAMS_TRACKER in query_params:
del query_params[QUERY_PARAMS_TRACKER] del query_params[QUERY_PARAMS_TRACKER]
return urlunparse( return urlunparse(
parsed_url._replace( parsed_url._replace(query=urlencode(query_params, doseq=True))
query=urlencode(query_params, doseq=True)
)
) )

59
bot/msgprocessor_test.py
View File

@ -5,6 +5,13 @@ from msgprocessor import (
TrackerRemovalProcessorMessage, TrackerRemovalProcessorMessage,
TrackerRemovalResult, TrackerRemovalResult,
) )
from dataclasses import dataclass
@dataclass(init=True, eq=True)
class TestUser:
id: int
first_name: str
class TestRemoverFactory(unittest.TestCase): class TestRemoverFactory(unittest.TestCase):
@ -12,13 +19,10 @@ class TestRemoverFactory(unittest.TestCase):
def test_remove_strategy_constructor(self): def test_remove_strategy_constructor(self):
test_case_data = [ test_case_data = [
{ {"domain": "youtube.com", "remover": self.factory.remove_yt_trackers},
"domain": "youtube.com",
"remover": self.factory.remove_yt_trackers
},
{ {
"domain": "lowerlevel.youtube.com", "domain": "lowerlevel.youtube.com",
"remover": self.factory.remove_yt_trackers "remover": self.factory.remove_yt_trackers,
}, },
{ {
"domain": "youtu.be", "domain": "youtu.be",
@ -27,7 +31,7 @@ class TestRemoverFactory(unittest.TestCase):
{ {
"domain": "something.youtu.be", "domain": "something.youtu.be",
"remover": self.factory.remove_yt_trackers, "remover": self.factory.remove_yt_trackers,
} },
] ]
for test_case in test_case_data: for test_case in test_case_data:
self.assertIs( self.assertIs(
@ -39,23 +43,26 @@ class TestRemoverFactory(unittest.TestCase):
test_case_data = [ test_case_data = [
{ {
"url": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy", "url": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy",
"expected_url": "https://youtu.be/jNQXAC9IVRw" "expected_url": "https://youtu.be/jNQXAC9IVRw",
}, },
{ {
"url": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy&t=16", "url": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy&t=16",
"expected_url": "https://youtu.be/jNQXAC9IVRw?t=16" "expected_url": "https://youtu.be/jNQXAC9IVRw?t=16",
}, },
{ {
"url": "https://www.youtube.com/watch?v=jNQXAC9IVRw", "url": "https://www.youtube.com/watch?v=jNQXAC9IVRw",
"expected_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", "url": "http://www.youtube.com/watch?v=jNQXAC9IVRw&si=qLIZT1rvs99_jbgy&t=16",
"expected_url": "http://www.youtube.com/watch?v=jNQXAC9IVRw&t=16" "expected_url": "http://www.youtube.com/watch?v=jNQXAC9IVRw&t=16",
} },
] ]
for test_case in test_case_data: for test_case in test_case_data:
self.assertEqual(self.factory.remove_yt_trackers(test_case["url"]), test_case["expected_url"]) self.assertEqual(
self.factory.remove_yt_trackers(test_case["url"]),
test_case["expected_url"],
)
class TestRemovalMsgProcessor(unittest.TestCase): class TestRemovalMsgProcessor(unittest.TestCase):
@ -63,44 +70,44 @@ class TestRemovalMsgProcessor(unittest.TestCase):
test_case_data = [ test_case_data = [
{ {
"msg_text": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy", "msg_text": "https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy",
"sender_username": "Ghytro", "sender": TestUser(id=123, first_name="Ghytro"),
"bot_responded": True, "bot_responded": True,
"bot_response": "Message from @Ghytro:\n\nhttps://youtu.be/jNQXAC9IVRw" "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наш слон хд", "msg_text": "чекай https://youtu.be/jNQXAC9IVRw?si=qLIZT1rvs99_jbgy\nнаш слон хд",
"sender_username": "OllyHearn", "sender": TestUser(id=321, first_name="OllyHearn"),
"bot_responded": True, "bot_responded": True,
"bot_response": "Message from @OllyHearn:\n\nчекай https://youtu.be/jNQXAC9IVRw\nнаш слон хд" "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 дада", "msg_text": "а я такая нитакуся без si ссылки шлю сразу https://youtu.be/jNQXAC9IVRw и по нескольку штук\nhttp://www.youtube.com/watch?v=jNQXAC9IVRw&si=qLIZT1rvs99_jbgy&t=16 дада",
"sender_username": "OllyHearn", "sender": TestUser(id=321, first_name="OllyHearn"),
"bot_responded": True, "bot_responded": True,
"bot_response": "Message from @OllyHearn:\n\nа я такая нитакуся без si ссылки шлю сразу https://youtu.be/jNQXAC9IVRw и по нескольку штук\nhttp://www.youtube.com/watch?v=jNQXAC9IVRw&t=16 дада" "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", "msg_text": "asdasdasdasdasdasdasd asdasd asdasd asdad sasa dadsas",
"sender_username": "Ghytro", "sender": TestUser(id=123, first_name="Ghytro"),
"bot_responded": False, "bot_responded": False,
"bot_response": "" "bot_response": "",
} },
] ]
for test_case in test_case_data: for test_case in test_case_data:
result = TrackerRemovalMsgProcessor( result = TrackerRemovalMsgProcessor(
TrackerRemovalProcessorMessage( TrackerRemovalProcessorMessage(
fromUsername=test_case["sender_username"], fromUser=test_case["sender"], text=test_case["msg_text"]
text=test_case["msg_text"]
) )
).process() ).process()
self.assertEqual( self.assertEqual(
result, result,
TrackerRemovalResult( TrackerRemovalResult(
needsToReply=test_case["bot_responded"], needsToReply=test_case["bot_responded"],
text=test_case["bot_response"] text=test_case["bot_response"],
) ),
) )
if __name__ == "__main__": if __name__ == "__main__":
test_classes_to_run = [TestRemoverFactory, TestRemovalMsgProcessor] test_classes_to_run = [TestRemoverFactory, TestRemovalMsgProcessor]
@ -110,7 +117,7 @@ if __name__ == "__main__":
for test_class in test_classes_to_run: for test_class in test_classes_to_run:
suite = loader.loadTestsFromTestCase(test_class) suite = loader.loadTestsFromTestCase(test_class)
suites_list.append(suite) suites_list.append(suite)
big_suite = unittest.TestSuite(suites_list) big_suite = unittest.TestSuite(suites_list)
runner = unittest.TextTestRunner() runner = unittest.TextTestRunner()