92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""
|
|
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()
|