only store one file per chat
This commit is contained in:
parent
87ccb2e627
commit
fb621074d0
|
@ -27,9 +27,8 @@ def setup(email: str, password: str, db: str, debug: bool) -> deltachat.Account:
|
|||
|
||||
def remind_chat(chat: deltachat.Chat):
|
||||
"""Remind a chat from the last sent file, with all lines which fit today's date."""
|
||||
messages = [msg.text for msg in chat.get_messages()]
|
||||
if chat_is_active(messages):
|
||||
file_path = get_file_path(messages)
|
||||
if chat_is_active(chat):
|
||||
file_path = get_file_path(chat)
|
||||
lines = get_lines_from_file(file_path)
|
||||
message_text = "\n".join(today_lines(lines))
|
||||
print(message_text)
|
||||
|
@ -38,8 +37,7 @@ def remind_chat(chat: deltachat.Chat):
|
|||
|
||||
def activate_chat(msg: deltachat.Message):
|
||||
"""Activate a Chat after the user sent /start"""
|
||||
messages = [message.text for message in msg.chat.get_messages()]
|
||||
file_path = get_file_path(messages)
|
||||
file_path = get_file_path(msg.chat)
|
||||
if not file_path:
|
||||
reply(
|
||||
msg.chat,
|
||||
|
@ -60,7 +58,7 @@ def store_file(msg: deltachat.Message):
|
|||
timer = msg.chat.get_ephemeral_timer()
|
||||
if timer:
|
||||
msg.chat.set_ephemeral_timer(0)
|
||||
new_filename = "./files/" + str(msg.chat.id) + "/" + msg.filename.split("/")[-1]
|
||||
new_filename = "./files/" + str(msg.chat.id)
|
||||
with open(msg.filename, "r") as read_file:
|
||||
os.makedirs(os.path.dirname(new_filename), exist_ok=True)
|
||||
with open(new_filename, "w+") as write_file:
|
||||
|
|
|
@ -1,5 +1,7 @@
|
|||
import datetime
|
||||
|
||||
import deltachat
|
||||
|
||||
|
||||
def check_new_day(current_day):
|
||||
if current_day != datetime.datetime.now().day:
|
||||
|
@ -11,12 +13,13 @@ def update_day():
|
|||
return datetime.datetime.now().day
|
||||
|
||||
|
||||
def chat_is_active(messages: [str]) -> bool:
|
||||
def chat_is_active(chat: deltachat.Chat) -> bool:
|
||||
"""Check if a user activated the bot.
|
||||
|
||||
:param messages: the text of all messages with the user
|
||||
:return: True if the user wants to receive reminders.
|
||||
"""
|
||||
messages = [msg.text for msg in chat.get_messages()]
|
||||
active = False
|
||||
for message_text in messages:
|
||||
if "/start" in message_text:
|
||||
|
@ -33,27 +36,13 @@ def chat_is_active(messages: [str]) -> bool:
|
|||
return active
|
||||
|
||||
|
||||
def get_file_path(messages: [str]) -> str:
|
||||
def get_file_path(chat: deltachat.Chat) -> str:
|
||||
"""Get the file path from the user's chat.
|
||||
|
||||
:param messages: the text of all messages with the user
|
||||
:param chat: the chat in which the file is requested
|
||||
:return: the file path with their reminders
|
||||
"""
|
||||
# :TODO SECURITY ISSUE - right now, this allows anyone to read any arbitrary file on the bot's machine.
|
||||
# at least the parts which are formatted as expected, tbh.
|
||||
|
||||
file_path = None
|
||||
for message_text in messages:
|
||||
if "Thanks, I stored the file at" in message_text:
|
||||
file_path = message_text.split("file at ", 1)[1]
|
||||
if message_text.startswith("/file"):
|
||||
try:
|
||||
file_path = message_text.split("file ", 1)[1]
|
||||
except IndexError as e:
|
||||
print("Can't split the message:", message_text)
|
||||
print(e)
|
||||
if not get_lines_from_file(file_path) == [f"File not found: {file_path}"]:
|
||||
return file_path
|
||||
return "./files/" + str(chat.id)
|
||||
|
||||
|
||||
def get_lines_from_file(file_path: str) -> [str]:
|
||||
|
|
|
@ -6,9 +6,8 @@ import requests
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_file_path(request, tmpdir):
|
||||
if request.param:
|
||||
path = str(tmpdir) + "/" + str(request.param)
|
||||
def tmp_file_path(tmpdir):
|
||||
path = str(tmpdir) + "/testfile"
|
||||
with open(path, "w+", encoding="utf-8") as f:
|
||||
f.write("test")
|
||||
return path
|
||||
|
|
|
@ -6,14 +6,13 @@ from remember_remember_bot.commands import store_file, reply
|
|||
from remember_remember_bot.util import get_file_path, get_lines_from_file
|
||||
|
||||
|
||||
def test_store_file(tmp_path, chat: deltachat.Chat):
|
||||
def test_store_file(tmp_file_path, chat: deltachat.Chat):
|
||||
chat.set_ephemeral_timer(1)
|
||||
reply(chat, "/file", attachment=str(tmp_path))
|
||||
reply(chat, "/file", attachment=str(tmp_file_path))
|
||||
last_message = chat.get_messages()[-1]
|
||||
store_file(last_message)
|
||||
time.sleep(1)
|
||||
messages = chat.get_messages()
|
||||
new_file_path = get_file_path([msg.text for msg in messages])
|
||||
new_file_path = get_file_path(chat)
|
||||
assert new_file_path
|
||||
assert get_lines_from_file(new_file_path) == ["test"]
|
||||
assert chat.get_ephemeral_timer() == 1
|
||||
|
|
|
@ -34,27 +34,10 @@ def test_update_day():
|
|||
(["/start", "/stop", "/start"], True),
|
||||
],
|
||||
)
|
||||
def test_chat_is_active(messages, active):
|
||||
assert chat_is_active(messages) == active
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"messages, tmp_file_path",
|
||||
[
|
||||
(["/file %file_path%"], "test1"),
|
||||
(["/file %file_path%", "/stop"], "test2"),
|
||||
(["I stored your reminders at the file %file_path%"], "test3"),
|
||||
(["/start", "/stop", "/start"], None),
|
||||
],
|
||||
indirect=["tmp_file_path"],
|
||||
)
|
||||
def test_get_file_path(messages: [], tmp_file_path: str):
|
||||
if tmp_file_path:
|
||||
for listitem in range(len(messages)):
|
||||
messages[listitem] = messages[listitem].replace(
|
||||
"%file_path%", tmp_file_path
|
||||
)
|
||||
assert get_file_path(messages) == tmp_file_path
|
||||
def test_chat_is_active(messages, active, chat):
|
||||
for msg in messages:
|
||||
chat.send_text(msg)
|
||||
assert chat_is_active(chat) == active
|
||||
|
||||
|
||||
def test_today_lines():
|
||||
|
|
Loading…
Reference in a new issue