136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""
|
|
Module implementing message related functions and classes.
|
|
"""
|
|
import pathlib
|
|
from typing import Type, TypeVar, Optional, Any, Final
|
|
from dataclasses import dataclass, asdict, field
|
|
from .tags import Tag
|
|
|
|
QuestionInst = TypeVar('QuestionInst', bound='Question')
|
|
AnswerInst = TypeVar('AnswerInst', bound='Answer')
|
|
MessageInst = TypeVar('MessageInst', bound='Message')
|
|
|
|
|
|
class MessageError(Exception):
|
|
pass
|
|
|
|
|
|
def source_code(text: str, include_delims: bool = False) -> list[str]:
|
|
"""
|
|
Extract all source code sections from the given text, i. e. all lines
|
|
surrounded by lines tarting with '```'. If 'include_delims' is True,
|
|
the surrounding lines are included, otherwise they are omitted. The
|
|
result list contains every source code section as a single string.
|
|
The order in the list represents the order of the sections in the text.
|
|
"""
|
|
code_sections: list[str] = []
|
|
code_lines: list[str] = []
|
|
in_code_block = False
|
|
|
|
for line in text.split('\n'):
|
|
if line.strip().startswith('```'):
|
|
if include_delims:
|
|
code_lines.append(line)
|
|
if in_code_block:
|
|
code_sections.append('\n'.join(code_lines) + '\n')
|
|
code_lines.clear()
|
|
in_code_block = not in_code_block
|
|
elif in_code_block:
|
|
code_lines.append(line)
|
|
|
|
return code_sections
|
|
|
|
|
|
class Question(str):
|
|
"""
|
|
A single question with a defined header.
|
|
"""
|
|
header: Final[str] = '=== QUESTION ==='
|
|
|
|
def __new__(cls: Type[QuestionInst], string: str) -> QuestionInst:
|
|
"""
|
|
Make sure the question string does not contain the header.
|
|
"""
|
|
if cls.header in string:
|
|
raise MessageError(f"Question '{string}' contains the header '{cls.header}'")
|
|
instance = super().__new__(cls, string)
|
|
return instance
|
|
|
|
def source_code(self, include_delims: bool = False) -> list[str]:
|
|
"""
|
|
Extract and return all source code sections.
|
|
"""
|
|
return source_code(self, include_delims)
|
|
|
|
|
|
class Answer(str):
|
|
"""
|
|
A single answer with a defined header.
|
|
"""
|
|
header: Final[str] = '=== ANSWER ==='
|
|
|
|
def __new__(cls: Type[AnswerInst], string: str) -> AnswerInst:
|
|
"""
|
|
Make sure the answer string does not contain the header.
|
|
"""
|
|
if cls.header in string:
|
|
raise MessageError(f"Answer '{string}' contains the header '{cls.header}'")
|
|
instance = super().__new__(cls, string)
|
|
return instance
|
|
|
|
def source_code(self, include_delims: bool = False) -> list[str]:
|
|
"""
|
|
Extract and return all source code sections.
|
|
"""
|
|
return source_code(self, include_delims)
|
|
|
|
|
|
@dataclass
|
|
class Message():
|
|
"""
|
|
Single message. Consists of a question and optionally an answer, a set of tags
|
|
and a file path.
|
|
"""
|
|
question: Question
|
|
answer: Optional[Answer]
|
|
tags: Optional[set[Tag]]
|
|
file_path: Optional[pathlib.Path]
|
|
file_suffixes: Final[list[str]] = field(default_factory=lambda: ['.txt', '.yaml'])
|
|
|
|
# @classmethod
|
|
# def from_file(cls: Type[MessageInst], file_path: pathlib.Path) -> MessageInst:
|
|
# """
|
|
# Create a Message from the given file. Expects the following file structure:
|
|
# * TagLine (from 'self.tags')
|
|
# * Question.Header
|
|
# * Question
|
|
# * Answer.Header
|
|
# """
|
|
# if file_path:
|
|
# self.file_path = file_path
|
|
# if not self.file_path:
|
|
# raise MessageError("Got no valid path to read message")
|
|
# if self.file_path.suffix not in self.file_suffixes:
|
|
# raise MessageError(f"File type '{self.file_path.suffix}' is not supported")
|
|
# pass
|
|
|
|
def to_file(self, file_path: Optional[pathlib.Path]) -> None:
|
|
"""
|
|
Write Message to the given file. Creates the following file structure:
|
|
* TagLine (from 'self.tags')
|
|
* Question.Header
|
|
* Question
|
|
* Answer.Header
|
|
* Answer
|
|
"""
|
|
if file_path:
|
|
self.file_path = file_path
|
|
if not self.file_path:
|
|
raise MessageError("Got no valid path to write message")
|
|
if self.file_path.suffix not in self.file_suffixes:
|
|
raise MessageError(f"File type '{self.file_path.suffix}' is not supported")
|
|
pass
|
|
|
|
def asdict(self) -> dict[str, Any]:
|
|
return asdict(self)
|