ALLOWED_TAGS = [

     'ci',

     'feat',

     'fix',

     'perf',

 Every commit message must be prefixed with one of the following tags:

 - ci        (ci related changes)

 - feat      (a new feature was added)

 - fix       (something was fixed)

 - perf      (performance related changes)

 CONSONANTS = 'bcdfghjklmnpqrstvwxyz'

 RFC5646_LANGUAGE_TAGS = {

         self._con.row_factory = self._namedtuple_factory

         self._get_jid_ids_from_db()

         assert cursor.description is not None

         fields = [col[0] for col in cursor.description]

         named_row = Row(*row)

         if 'additional_data' in fields:

             _dict = json.loads(named_row.additional_data or '{}')

     @staticmethod

     def _get_timeout() -> int:

         returns the timeout in epoch

         timeout = app.settings.get('restore_timeout')

         now = int(time.time())

     @timeit

     def _get_jid_ids_from_db(self) -> None:

         Load all jid/jid_id tuples into a dict for faster access

         rows = self._con.execute(

             'SELECT jid_id, jid, type FROM jids').fetchall()

         for row in rows:

                    kind: Optional[KindConstant] = None,

                    type_: Optional[JIDConstant] = None

                    ) -> int:

         Get the jid id from a jid.

         In case the jid id is not found create a new one.

         :param type_:   The JIDConstant

         return the jid id

         if kind in (KindConstant.GC_MSG, KindConstant.GCSTATUS):

             type_ = JIDConstant.ROOM_TYPE

     @staticmethod

     def convert_show_values_to_db_api_values(show: Optional[str]

                                              ) -> Optional[ShowConstant]:

         Convert from string style to constant ints for db

         if show == 'online':

             return ShowConstant.ONLINE

                                       timestamp: float,

                                       n_lines: int

                                       ) -> list[ConversationRow]:

         Load n_lines lines of conversation with jid before or after timestamp

         :param account:         The account

         :param timestamp:       timestamp

         returns a list of namedtuples

         jids = [jid]

         account_id = self.get_account_id(account)

         kinds = map(str, [KindConstant.ERROR,

                                    account: str,

                                    jid: JID

                                    ) -> LastConversationRow:

         Load the last line of a conversation with jid for account.

         Loads messages, but no status messages or error messages.

         :param jid:             The jid for which we request the conversation

         returns a list of namedtuples

         jids = [jid]

         account_id = self.get_account_id(account)

                                 timestamp: float

                                 ) -> tuple[list[ConversationRow],

                                            list[ConversationRow]]:

         Load all lines of conversation with jid around a specific timestamp

         :param account:         The account

         :param timestamp:       Timestamp around which to fetch messages

         returns a list of namedtuples

         jids = [jid]

         account_id = self.get_account_id(account)

         kinds = map(str, [KindConstant.ERROR])

                                  jid: str,

                                  before: float,

                                  after: float) -> list[ConversationRow]:

         Load all lines of conversation with jid between two timestamps

         :param account:         The account

         :param after:           earliest timestamp

         returns a list of namedtuples

         jids = [jid]

         account_id = self.get_account_id(account)

         kinds = map(str, [KindConstant.ERROR])

                    before: Optional[datetime.datetime] = None,

                    after: Optional[datetime.datetime] = None

                    ) -> Iterator[SearchLogRow]:

         Search the conversation log for messages containing the `query` string.

         The search can either span the complete log for the given

         :param after: A datetime.datetime instance or None

         returns a list of namedtuples

         jids = [jid]

         kinds = map(str, [KindConstant.STATUS,

                         before: Optional[datetime.datetime] = None,

                         after: Optional[datetime.datetime] = None

                         ) -> Iterator[SearchLogRow]:

         Search all conversation logs for messages containing the `query`

         string.

         :param after: A datetime.datetime instance or None

         returns a list of namedtuples

         account_ids = self.get_active_account_ids()

         kinds = map(str, [KindConstant.STATUS,

                           KindConstant.GCSTATUS])

                               year: int,

                               month: int

                               ) -> list[HistoryDayRow]:

         Get days in month where messages for 'jid' exist

         jids = [jid]

         account_id = self.get_account_id(account)

         kinds = map(str, [KindConstant.STATUS,

         delta = datetime.timedelta(

             days=days, hours=23, minutes=59, seconds=59, microseconds=999999)

             SELECT DISTINCT

             CAST(strftime('%d', time, 'unixepoch', 'localtime') AS INTEGER)

             AS day FROM logs NATURAL JOIN jids WHERE jid IN ({jids})

             AND time BETWEEN ? AND ?

             AND kind NOT IN ({kinds})

             ORDER BY time

                        account_id=account_id,

                        kinds=', '.join(kinds))

                                    account: str,

                                    jid: str

                                    ) -> Optional[float]:

         Get the timestamp of the last message we received for the jid

         jids = [jid]

         account_id = self.get_account_id(account)

         kinds = map(str, [KindConstant.STATUS,

                                     account: str,

                                     jid: str

                                     ) -> Optional[float]:

         Get the timestamp of the first message we received for the jid

         jids = [jid]

         account_id = self.get_account_id(account)

         kinds = map(str, [KindConstant.STATUS,

                          jid: str,

                          date: datetime.datetime

                          ) -> Optional[float]:

         Get a single timestamp of a message for 'jid'

         in time range of one day

         jids = [jid]

         account_id = self.get_account_id(account)

         delta = datetime.timedelta(

                                         jid: str,

                                         date: datetime.datetime

                                         ) -> Optional[MessageMetaRow]:

         Load meta data for the first message of a specific date

         jids = [jid]

         account_id = self.get_account_id(account)

                                 timestamp: float,

                                 message_id: str

                                 ) -> bool:

         Check if a message is already in the `logs` table

         :param account:     The account

         :param timestamp:   The timestamp in UTC epoch

         :param message_id:  The message-id

         # Add 60 seconds around the timestamp

         start_time = timestamp - 60

                        origin_id: Optional[str] = None,

                        groupchat: bool = False

                        ) -> bool:

         Checks if a stanza-id is already in the `logs` table

         :param account:     The account

         :param groupchat:   stanza-id is from a groupchat

         return True if the stanza-id was found

         ids: list[str] = []

         if stanza_id is not None:

             ids.append(stanza_id)

                    kind: Optional[KindConstant] = None,

                    type_: JIDConstant = JIDConstant.NORMAL_TYPE

                    ) -> int:

         Insert a new jid into the `jids` table.

         This is an alias of get_jid_id() for better readablility.

         :param kind:    A KindConstant

         :param type_:   A JIDConstant

         return self.get_jid_id(jid, kind, type_)

     @timeit

                          kind: KindConstant,

                          **kwargs: Any

                          ) -> int:

         Insert a new message into the `logs` table

         :param jid:     The jid as string

         :param kwargs:  Every additional named argument must correspond to

                         a field in the `logs` table

         jid_id = self.get_jid_id(jid, kind=kind)

         account_id = self.get_account_id(account)

                           message_id: str,

                           error: str

                           ) -> None:

         Update the corresponding message with the error

         :param account_jid: The jid of the account

         :param error:       The error stanza as string

         account_id = self.get_jid_id(account_jid)

         try:

                    message_id: str,

                    state: str

                    ) -> None:

         Update the marker state of the corresponding message

         :param account_jid: The jid of the account

         :param state:       The state, 'received' or 'displayed'

         if state not in ('received', 'displayed'):

             raise ValueError('Invalid marker state')

     @timeit

     def get_archive_infos(self, jid: str) -> Optional[LastArchiveMessageRow]:

         Get the archive infos

         :param jid:     The jid that belongs to the avatar

         jid_id = self.get_jid_id(jid, type_=JIDConstant.ROOM_TYPE)

         sql = '''SELECT * FROM last_archive_message WHERE jid_id = ?'''

         return self._con.execute(sql, (jid_id,)).fetchone()

     @timeit

     def set_archive_infos(self, jid: str, **kwargs: Any) -> None:

         Set archive infos

         :param jid:                     The jid that belongs to the avatar

         :param sync_threshold:          The max days that we request from a

                                         MUC archive

         jid_id = self.get_jid_id(jid)

         exists = self.get_archive_infos(jid)

         if not exists:

     @timeit

     def reset_archive_infos(self, jid: str) -> None:

         Set archive infos

         :param jid:                     The jid of the archive

         jid_id = self.get_jid_id(jid)

         sql = '''UPDATE last_archive_message

                  SET last_mam_id = NULL, oldest_mam_timestamp = NULL,

                 yield result

     def remove_history(self, account: str, jid: JID) -> None:

         Remove history for a specific chat.

         If it's a group chat, remove last MAM ID as well.

         account_id = self.get_account_id(account)

         try:

             jid_id = self.get_jid_id(jid)

         log.info('Removed history for: %s', jid)

     def remove_all_history(self) -> None:

         Remove all messages for all accounts

         statements = [

             'DELETE FROM logs',

             'DELETE FROM jids',

         log.info('Removed all chat history')

     def cleanup_chat_history(self) -> None:

         Remove messages from account where messages are older than max_age

         for account in app.settings.get_accounts():

             max_age = app.settings.get_account_setting(

                 account, 'chat_history_max_age')

 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 # POSSIBILITY OF SUCH DAMAGE.

 Provides IPython console widget

 @author: Eitan Isaacson

 All rights reserved. This program and the accompanying materials are made

 available under the terms of the BSD which accompanies this distribution, and

 is available at U{http://www.opensource.org/licenses/bsd-license.php}

 import re

 import sys

 class IterableIPShell:

     Create an IPython instance. Does not start a blocking event loop,

     instead allow single iterations. This allows embedding in GTK

     without blockage

     @type history_level: integer

     @ivar complete_sep: Separation delimiters for completion function.

     @type complete_sep: _sre.SRE_Pattern

     def __init__(self, argv=None, user_ns=None, user_global_ns=None, cin=None,

                  cout=None, cerr=None, input_func=None):

         @param argv: Command line options for IPython

         @type argv: list

         @param user_ns: User namespace.

         @type cerr: IO stream

         @param input_func: Replacement for builtin raw_input()

         @type input_func: function

         if argv is None:

             argv = []

         from traitlets.config.loader import Config

         cfg = Config()

         # InteractiveShell's __init__ gets a reference of stdout and stderr

         # so we save the standard here to revert it after init

                 self.IP.user_ns.update({k: v})

     def execute(self):

         Execute the current line provided by the shell object

         self.history_level = 0

         orig_stdout = sys.stdout

         sys.stdout = IPython.utils.io.stdout

         '''

         # TODO: Update to IPython 5.x and later

         return prompt

     def historyBack(self):

         Provide one history command back

         @return: The command string.

         @rtype: string

         self.history_level -= 1

         return self._getHistory()

     def historyForward(self):

         Provide one history command forward

         @return: The command string.

         @rtype: string

         self.history_level += 1

         return self._getHistory()

     def _getHistory(self):

         Get the command string of the current history level

         @return: Historic command string.

         @rtype: string

         try:

             rv = self.IP.user_ns['In'][self.history_level].strip('\n')

         except IndexError:

         return rv

     def updateNamespace(self, ns_dict):

         Add the current dictionary to the shell namespace

         @param ns_dict: A dictionary of symbol-values.

         @type ns_dict: dictionary

         self.IP.user_ns.update(ns_dict)

     def complete(self, line):

         Returns an auto completed line and/or possibilities for completion

         @param line: Given line so far.

         @return: Line completed as for as possible,

         and possible further completions.

         @rtype: tuple

         split_line = self.complete_sep.split(line)

         if split_line[-1]:

             possibilities = self.IP.complete(split_line[-1])

 class ConsoleView(Gtk.TextView):

     Specialized text view for console-like workflow

     @cvar ANSI_COLORS: Mapping of terminal colors to X11 names.

     @type mark: Gtk.TextMark

     @ivar line_start: Start of command line mark.

     @type line_start: Gtk.TextMark

     ANSI_COLORS = {'0;30': 'Black', '0;31': 'Red',

                    '0;32': 'Green', '0;33': 'Brown',

                    '1;36': 'LightCyan', '1;37': 'White'}

     def __init__(self):

         Initialize console view

         GObject.GObject.__init__(self)

         self.override_font(Pango.FontDescription('Mono'))

         self.set_cursor_visible(True)

             GLib.idle_add(self._write5, text, editable)

     def _write5(self, text, editable=False):

         Write given text to buffer

         @param text: Text to append.

         @type text: list of (token: string)

         @param editable: If true, added text is editable.

         @type editable: boolean

         start_mark = self.text_buffer.create_mark(

             None, self.text_buffer.get_end_iter(), True)

         self.scroll_mark_onscreen(self.mark)

     def _write(self, text, editable=False):

         Write given text to buffer

         @param text: Text to append.

         @type text: string

         @param editable: If true, added text is editable.

         @type editable: boolean

         if isinstance(text, list):

             self._write5(text, editable)

             return

         GLib.idle_add(self._showPrompt, prompt)

     def _showPrompt(self, prompt):

         Print prompt at start of line

         @param prompt: Prompt to print.

         @type prompt: string

         self._write(prompt)

         self.text_buffer.move_mark(self.line_start,

                                    self.text_buffer.get_end_iter())

         GLib.idle_add(self._changeLine, text)

     def _changeLine(self, text):

         Replace currently entered command line with given text

         @param text: Text to use as replacement.

         @type text: string

         iter_ = self.text_buffer.get_iter_at_mark(self.line_start)

         iter_.forward_to_line_end()

         self.text_buffer.delete(

         self._write(text, True)

     def getCurrentLine(self):

         Get text in current command line

         @return: Text of current command line.

         @rtype: string

         rv = self.text_buffer.get_slice(

             self.text_buffer.get_iter_at_mark(self.line_start),

             self.text_buffer.get_end_iter(), False)

         GLib.idle_add(self._showReturned, text)

     def _showReturned(self, text):

         Show returned text from last command and print new prompt

         @param text: Text to show.

         @type text: string

         iter_ = self.text_buffer.get_iter_at_mark(self.line_start)

         iter_.forward_to_line_end()

         self.text_buffer.apply_tag_by_name(

             self.text_buffer.insert_at_cursor(indentation)

     def onKeyPress(self, _widget, event):

         Key press callback used for correcting behavior for console-like

         interfaces. For example 'home' should go to prompt, not to beginning of

         line

         @return: Return True if event should not trickle.

         @rtype: boolean

         insert_mark = self.text_buffer.get_insert()

         insert_iter = self.text_buffer.get_iter_at_mark(insert_mark)