text_buffer.py 6.61 KB
Newer Older
1 2
"""
Define the TextBuffer class
3 4 5 6 7

A text buffer contains a list of intermediate representations of messages
(not xml stanzas, but neither the Lines used in windows.py.

Each text buffer can be linked to multiple windows, that will be rendered
Link Mauve's avatar
Link Mauve committed
8
independently by their TextWins.
9 10
"""

11 12 13
import logging
log = logging.getLogger(__name__)

mathieui's avatar
mathieui committed
14 15 16 17 18 19 20 21
from typing import (
    Dict,
    List,
    Optional,
    TYPE_CHECKING,
    Tuple,
    Union,
)
22
from datetime import datetime
mathieui's avatar
mathieui committed
23
from poezio.config import config
24
from poezio.ui.types import Message, BaseMessage
25

mathieui's avatar
mathieui committed
26 27
if TYPE_CHECKING:
    from poezio.windows.text_win import TextWin
mathieui's avatar
mathieui committed
28 29


mathieui's avatar
mathieui committed
30 31
class CorrectionError(Exception):
    pass
mathieui's avatar
mathieui committed
32

mathieui's avatar
mathieui committed
33

34 35 36
class AckError(Exception):
    pass

mathieui's avatar
mathieui committed
37

38
class TextBuffer:
39 40
    """
    This class just keep trace of messages, in a list with various
41
    information and attributes.
42
    """
mathieui's avatar
mathieui committed
43

44
    def __init__(self, messages_nb_limit: Optional[int] = None) -> None:
45 46

        if messages_nb_limit is None:
47
            messages_nb_limit = config.get('max_messages_in_memory')
48
        self._messages_nb_limit = messages_nb_limit  # type: int
49
        # Message objects
50
        self.messages = []  # type: List[BaseMessage]
51
        # COMPAT: Correction id -> Original message id.
Maxime Buquet's avatar
Maxime Buquet committed
52
        self.correction_ids = {}  # type: Dict[str, str]
53
        # we keep track of one or more windows
54
        # so we can pass the new messages to them, as they are added, so
55
        # they (the windows) can build the lines from the new message
mathieui's avatar
mathieui committed
56
        self._windows = []  # type: List[TextWin]
57

58
    def add_window(self, win) -> None:
59
        self._windows.append(win)
60

61
    @property
mathieui's avatar
mathieui committed
62
    def last_message(self) -> Optional[BaseMessage]:
63 64
        return self.messages[-1] if self.messages else None

65
    def add_message(self, msg: BaseMessage):
66 67 68
        """
        Create a message and add it to the text buffer
        """
69
        self.messages.append(msg)
70

71
        while len(self.messages) > self._messages_nb_limit:
72
            self.messages.pop(0)
73

74
        ret_val = 0
75
        show_timestamps = config.get('show_timestamps')
76
        nick_size = config.get('max_nick_length')
mathieui's avatar
mathieui committed
77 78
        for window in self._windows:  # make the associated windows
            # build the lines from the new message
Madhur Garg's avatar
Madhur Garg committed
79 80 81 82 83 84
            nb = window.build_new_message(
                msg,
                timestamp=show_timestamps,
                nick_size=nick_size)
            if ret_val == 0:
                ret_val = nb
85
            top = isinstance(msg, Message) and msg.top
86
            if window.pos != 0 and top is False:
Madhur Garg's avatar
Madhur Garg committed
87
                window.scroll_up(nb)
88

89
        return min(ret_val, 1)
90

91
    def _find_message(self, orig_id: str) -> Tuple[str, int]:
92 93 94
        """
        Find a message in the text buffer from its message id
        """
Maxime Buquet's avatar
Maxime Buquet committed
95 96 97
        # When looking for a message, ensure the id doesn't appear in a
        # message we've removed from our message list. If so return the index
        # of the corresponding id for the original message instead.
98
        orig_id = self.correction_ids.get(orig_id, orig_id)
Maxime Buquet's avatar
Maxime Buquet committed
99

mathieui's avatar
mathieui committed
100
        for i in range(len(self.messages) - 1, -1, -1):
101
            msg = self.messages[i]
102 103 104
            if msg.identifier == orig_id:
                return (orig_id, i)
        return (orig_id, -1)
105

106
    def ack_message(self, old_id: str, jid: str) -> Union[None, bool, Message]:
107
        """Mark a message as acked"""
108
        return self._edit_ack(1, old_id, jid)
109

Link Mauve's avatar
Link Mauve committed
110 111
    def nack_message(self, error: str, old_id: str,
                     jid: str) -> Union[None, bool, Message]:
112
        """Mark a message as errored"""
113
        return self._edit_ack(-1, old_id, jid, append=error)
114

115
    def _edit_ack(self, value: int, old_id: str, jid: str,
Link Mauve's avatar
Link Mauve committed
116
                  append: str = '') -> Union[None, bool, Message]:
117
        """
118 119
        Edit the ack status of a message, and optionally
        append some text.
120
        """
Maxime Buquet's avatar
Maxime Buquet committed
121
        _, i = self._find_message(old_id)
122
        if i == -1:
123
            return None
124
        msg = self.messages[i]
mathieui's avatar
mathieui committed
125 126
        if not isinstance(msg, Message):
            return None
127 128
        if msg.ack == 1:  # Message was already acked
            return False
129
        if msg.jid != jid:
mathieui's avatar
mathieui committed
130 131
            raise AckError('Wrong JID for message id %s (was %s, expected %s)'
                           % (old_id, msg.jid, jid))
132

133
        msg.ack = value
134
        if append:
135 136
            msg.txt += append
        return msg
137

mathieui's avatar
mathieui committed
138
    def modify_message(self,
139
                       txt: str,
140
                       orig_id: str,
141 142 143 144
                       new_id: str,
                       highlight: bool = False,
                       time: Optional[datetime] = None,
                       user: Optional[str] = None,
145
                       jid: Optional[str] = None) -> Message:
146 147
        """
        Correct a message in a text buffer.
Maxime Buquet's avatar
Maxime Buquet committed
148 149 150 151 152

        Version 1.1.0 of Last Message Correction (0308) added clarifications
        that break the way poezio handles corrections. Instead of linking
        corrections to the previous correction/message as we were doing, we
        are now required to link all corrections to the original messages.
153 154
        """

155
        orig_id, i = self._find_message(orig_id)
156 157

        if i == -1:
mathieui's avatar
mathieui committed
158 159
            log.debug(
                'Message %s not found in text_buffer, abort replacement.',
160
                orig_id)
161 162 163
            raise CorrectionError("nothing to replace")

        msg = self.messages[i]
mathieui's avatar
mathieui committed
164 165
        if not isinstance(msg, Message):
            raise CorrectionError('Wrong message type')
166 167
        if msg.user and msg.user is not user:
            raise CorrectionError("Different users")
168
        elif msg.history:
169 170 171 172 173
            raise CorrectionError("Delayed message")
        elif not msg.user and (msg.jid is None or jid is None):
            raise CorrectionError('Could not check the '
                                  'identity of the sender')
        elif not msg.user and msg.jid != jid:
mathieui's avatar
mathieui committed
174 175
            raise CorrectionError(
                'Messages %s and %s have not been '
176
                'sent by the same fullJID' % (orig_id, new_id))
177 178 179

        if not time:
            time = msg.time
Maxime Buquet's avatar
Maxime Buquet committed
180

181
        self.correction_ids[new_id] = orig_id
mathieui's avatar
mathieui committed
182
        message = Message(
183 184 185 186 187 188 189
            txt=txt,
            time=time,
            nickname=msg.nickname,
            nick_color=msg.nick_color,
            history=False,
            user=msg.user,
            identifier=orig_id,
mathieui's avatar
mathieui committed
190 191 192 193
            highlight=highlight,
            old_message=msg,
            revisions=msg.revisions + 1,
            jid=jid)
194
        self.messages[i] = message
195
        log.debug('Replacing message %s with %s.', orig_id, new_id)
196
        return message
197

198
    def del_window(self, win) -> None:
199
        self._windows.remove(win)
200 201

    def __del__(self):
202 203
        size = len(self.messages)
        log.debug('** Deleting %s messages from textbuffer', size)