Compare commits
8 Commits
fdbca572e2
...
cb0a90f72a
| Author | SHA1 | Date | |
|---|---|---|---|
| cb0a90f72a | |||
| a010a6f5d0 | |||
| c78640a6b5 | |||
| 9ca4528e25 | |||
| 80a00833f2 | |||
| 01f7c71555 | |||
| b94ec999a4 | |||
| f84d9444b4 |
@ -1,5 +1,5 @@
|
|||||||
"""
|
"""
|
||||||
Module implementing various chat classes and functions for managing a chat history.
|
Module implementing the 'Chat' class and functions for managing a chat history.
|
||||||
"""
|
"""
|
||||||
import pathlib
|
import pathlib
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
@ -7,7 +7,6 @@ from typing import TypeVar, Type, Optional
|
|||||||
from .message import Message, MessageFilter, MessageError
|
from .message import Message, MessageFilter, MessageError
|
||||||
|
|
||||||
ChatInst = TypeVar('ChatInst', bound='Chat')
|
ChatInst = TypeVar('ChatInst', bound='Chat')
|
||||||
ChatDirInst = TypeVar('ChatDirInst', bound='ChatDir')
|
|
||||||
|
|
||||||
|
|
||||||
class ChatError(Exception):
|
class ChatError(Exception):
|
||||||
@ -17,45 +16,22 @@ class ChatError(Exception):
|
|||||||
@dataclass
|
@dataclass
|
||||||
class Chat:
|
class Chat:
|
||||||
"""
|
"""
|
||||||
A class containing a complete chat history.
|
Class containing a complete chat history.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
messages: list[Message]
|
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:
|
|
||||||
"""
|
|
||||||
A Chat class that is bound to a given directory. Supports reading
|
|
||||||
and writing messages from / to that directory.
|
|
||||||
"""
|
|
||||||
|
|
||||||
messages: list[Message]
|
|
||||||
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
|
@classmethod
|
||||||
def from_dir(cls: Type[ChatDirInst],
|
def from_dir(cls: Type[ChatInst],
|
||||||
path: pathlib.Path,
|
path: pathlib.Path,
|
||||||
glob: Optional[str] = None,
|
glob: Optional[str] = None,
|
||||||
mfilter: Optional[MessageFilter] = None) -> ChatDirInst:
|
mfilter: Optional[MessageFilter] = None) -> ChatInst:
|
||||||
"""
|
"""
|
||||||
Create a ChatDir instance from the given directory. If 'glob' is specified,
|
Create a Chat instance from the message in the given directory.
|
||||||
files will be filtered using 'path.glob()', otherwise it uses 'path.iterdir()'.
|
If glob is specified, files will be filtered using path.glob(),
|
||||||
Messages are created using 'Message.from_file()' and the optional MessageFilter.
|
otherwise it uses path.iterdir(). Messages are created using
|
||||||
|
Message.from_file() and the optional MessageFilter.
|
||||||
"""
|
"""
|
||||||
messages: list[Message] = []
|
messages: list[Message] = []
|
||||||
message_files: set[str] = set()
|
|
||||||
file_iter = path.glob(glob) if glob else path.iterdir()
|
file_iter = path.glob(glob) if glob else path.iterdir()
|
||||||
for file_path in file_iter:
|
for file_path in file_iter:
|
||||||
if file_path.is_file():
|
if file_path.is_file():
|
||||||
@ -63,30 +39,6 @@ class ChatDir:
|
|||||||
message = Message.from_file(file_path, mfilter)
|
message = Message.from_file(file_path, mfilter)
|
||||||
if message:
|
if message:
|
||||||
messages.append(message)
|
messages.append(message)
|
||||||
message_files.add(file_path.name)
|
|
||||||
except MessageError as e:
|
except MessageError as e:
|
||||||
print(f"Error processing message in '{file_path}': {str(e)}")
|
print(f"Error processing message in '{file_path}': {str(e)}")
|
||||||
return cls(messages, path, mfilter, message_files)
|
return cls(messages)
|
||||||
|
|
||||||
@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()
|
|
||||||
|
|||||||
@ -326,7 +326,7 @@ class Message():
|
|||||||
data[cls.file_yaml_key] = file_path
|
data[cls.file_yaml_key] = file_path
|
||||||
return cls.from_dict(data)
|
return cls.from_dict(data)
|
||||||
|
|
||||||
def to_file(self, file_path: Optional[pathlib.Path]=None) -> None: # noqa: 11
|
def to_file(self, file_path: Optional[pathlib.Path]) -> None: # noqa: 11
|
||||||
"""
|
"""
|
||||||
Write a Message to the given file. Type is determined based on the suffix.
|
Write a Message to the given file. Type is determined based on the suffix.
|
||||||
Currently supported suffixes: ['.txt', '.yaml']
|
Currently supported suffixes: ['.txt', '.yaml']
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user