From 6b3855ef08f6d5d621202149d5f1c2bcc360a8dc Mon Sep 17 00:00:00 2001 From: juk0de Date: Thu, 24 Aug 2023 16:49:54 +0200 Subject: [PATCH] added new module 'chat.py' --- chatmastermind/chat.py | 91 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) create mode 100644 chatmastermind/chat.py diff --git a/chatmastermind/chat.py b/chatmastermind/chat.py new file mode 100644 index 0000000..72b369b --- /dev/null +++ b/chatmastermind/chat.py @@ -0,0 +1,91 @@ +""" +Module implementing various chat classes and functions for managing a chat history. +""" +import pathlib +from dataclasses import dataclass +from typing import TypeVar, Type, Optional +from .message import Message, MessageFilter, MessageError + +ChatInst = TypeVar('ChatInst', bound='Chat') +ChatDirInst = TypeVar('ChatDirInst', bound='ChatDir') + + +class ChatError(Exception): + pass + + +@dataclass +class Chat: + """ + A class containing a complete chat history. + """ + + messages: list[Message] + + def filter(self, mfilter: MessageFilter) -> None: + """ + Use 'Message.match(mfilter) to remove all messages that + don't fulfill the filter requirements. + """ + self.messages = [m for m in self.messages if m.match(mfilter)] + + +@dataclass +class ChatDir(Chat): + """ + A Chat class that is bound to a given directory. Supports reading + and writing messages from / to that directory. + """ + + directory: pathlib.Path + # a MessageFilter that all messages must match (if given) + mfilter: Optional[MessageFilter] = None + # set containing all file names of the current messages + message_files: set[str] = set() + + @classmethod + def from_dir(cls: Type[ChatDirInst], + path: pathlib.Path, + glob: Optional[str] = None, + mfilter: Optional[MessageFilter] = None) -> ChatDirInst: + """ + Create a ChatDir instance from the given directory. If 'glob' is specified, + files will be filtered using 'path.glob()', otherwise it uses 'path.iterdir()'. + Messages are created using 'Message.from_file()' and the optional MessageFilter. + """ + messages: list[Message] = [] + message_files: set[str] = set() + file_iter = path.glob(glob) if glob else path.iterdir() + for file_path in file_iter: + if file_path.is_file(): + try: + message = Message.from_file(file_path, mfilter) + if message: + messages.append(message) + message_files.add(file_path.name) + except MessageError as e: + print(f"Error processing message in '{file_path}': {str(e)}") + return cls(messages, path, mfilter, message_files) + + @classmethod + def from_messages(cls: Type[ChatDirInst], + path: pathlib.Path, + messages: list[Message], + mfilter: Optional[MessageFilter]) -> ChatDirInst: + """ + Create a ChatDir instance from the given message list. + Note that the next call to 'dump()' will write all files + in order to synchronize the messages. 'update()' is not + supported until after the first 'dump()'. + """ + return cls(messages, path, mfilter) + + # def dump(self) -> None: + # """ + # Writes all messages to the bound directory. If a message has no file_path, + # it will create a new one. + # """ + # for message in self.messages: + # # TODO: determine file name if message does not have one + # if message.file_path.name() not in self.message_files: + # message.to_file()