...
 
Commits (7)
......@@ -14,7 +14,7 @@ from datetime import (
timezone,
)
from pathlib import Path
from typing import Dict, List, Optional, Tuple, Union
from typing import Dict, List, Optional, Tuple, Union, Any
import os
import subprocess
......@@ -44,7 +44,7 @@ def _get_output_of_command(command: str) -> Optional[List[str]]:
return None
def _is_in_path(command: str, return_abs_path=False) -> Union[bool, str]:
def _is_in_path(command: str, return_abs_path: bool = False) -> Union[bool, str]:
"""
Check if *command* is in the $PATH or not.
......@@ -111,10 +111,12 @@ def get_os_info() -> str:
stdout=subprocess.PIPE,
close_fds=True)
process.wait()
output = process.stdout.readline().decode('utf-8').strip()
# some distros put n/a in places, so remove those
output = output.replace('n/a', '').replace('N/A', '')
return output
if process.stdout is not None:
out = process.stdout.readline().decode('utf-8').strip()
# some distros put n/a in places, so remove those
out = out.replace('n/a', '').replace('N/A', '')
return out
return ''
# lsb_release executable not available, so parse files
for distro_name in DISTRO_INFO:
......@@ -287,7 +289,7 @@ def shell_split(st: str) -> List[str]:
return ret
def find_argument(pos: int, text: str, quoted=True) -> int:
def find_argument(pos: int, text: str, quoted: bool = True) -> int:
"""
Split an input into a list of arguments, return the number of the
argument selected by pos.
......@@ -342,7 +344,7 @@ def _find_argument_unquoted(pos: int, text: str) -> int:
return argnum + 1
def parse_str_to_secs(duration='') -> int:
def parse_str_to_secs(duration: str = '') -> int:
"""
Parse a string of with a number of d, h, m, s.
......@@ -370,7 +372,7 @@ def parse_str_to_secs(duration='') -> int:
return result
def parse_secs_to_str(duration=0) -> str:
def parse_secs_to_str(duration: int = 0) -> str:
"""
Do the reverse operation of :py:func:`parse_str_to_secs`.
......@@ -457,7 +459,7 @@ def format_gaming_string(infos: Dict[str, str]) -> str:
return name
def safeJID(*args, **kwargs) -> JID:
def safeJID(*args: Any, **kwargs: Any) -> JID:
"""
Construct a :py:class:`slixmpp.JID` object from a string.
......
......@@ -15,7 +15,17 @@ import shutil
import time
import uuid
from collections import defaultdict
from typing import Callable, Dict, List, Optional, Set, Tuple, Type
from typing import (
cast,
Callable,
Dict,
List,
Optional,
Set,
Tuple,
Type,
TypeVar,
)
from xml.etree import ElementTree as ET
from functools import partial
......@@ -61,10 +71,15 @@ from poezio.core.structs import (
POSSIBLE_SHOW,
)
from poezio.ui.types import Message, InfoMessage
from poezio.ui.types import (
Message,
InfoMessage,
PersistentInfoMessage,
)
log = logging.getLogger(__name__)
T = TypeVar('T', bound=tabs.Tab)
class Core:
"""
......@@ -99,8 +114,10 @@ class Core:
# that are displayed in almost all tabs, in an
# information window.
self.information_buffer = TextBuffer()
self.information_win_size = config.get(
'info_win_height', section='var')
self.information_win_size = cast(
int,
config.get('info_win_height', section='var'),
)
self.information_win = windows.TextWin(300)
self.information_buffer.add_window(self.information_win)
self.left_tab_win = None
......@@ -813,7 +830,7 @@ class Core:
####################### XMPP-related actions ##################################
def get_status(self) -> str:
def get_status(self) -> Status:
"""
Get the last status that was previously set
"""
......@@ -1016,7 +1033,7 @@ class Core:
### Tab getters ###
def get_tabs(self, cls: Type[tabs.Tab] = None) -> List[tabs.Tab]:
def get_tabs(self, cls: Type[T] = None) -> List[T]:
"Get all the tabs of a type"
if cls is None:
return self.tabs.get_tabs()
......@@ -1324,7 +1341,7 @@ class Core:
if tab.name.startswith(room_name):
tab.activate(reason=reason)
def on_user_changed_status_in_private(self, jid: JID, status: str) -> None:
def on_user_changed_status_in_private(self, jid: JID, status: Status) -> None:
tab = self.tabs.by_name_and_class(jid, tabs.ChatTab)
if tab is not None: # display the message in private
tab.update_status(status)
......@@ -1359,7 +1376,7 @@ class Core:
"""
tab = self.tabs.by_name_and_class(jid, tabs.ConversationTab)
if tab is not None:
tab.add_message(InfoMessage(msg), typ=2)
tab.add_message(PersistentInfoMessage(msg))
if self.tabs.current_tab is tab:
self.refresh_window()
......@@ -1652,7 +1669,7 @@ class Core:
return
else:
scr = self.stdscr
tabs.Tab.resize(scr)
tabs.Tab.initial_resize(scr)
self.resize_global_info_bar()
self.resize_global_information_win()
for tab in self.tabs:
......@@ -2105,7 +2122,7 @@ class Core:
self.bookmarks.get_remote(self.xmpp, self.information,
_join_remote_only)
def room_error(self, error, room_name):
def room_error(self, error: IqError, room_name: str) -> None:
"""
Display the error in the tab
"""
......@@ -2120,12 +2137,11 @@ class Core:
nickname='Error',
nick_color=get_theme().COLOR_ERROR_MSG,
),
typ=2,
)
code = error['error']['code']
if code == '401':
msg = 'To provide a password in order to join the room, type "/join / password" (replace "password" by the real password)'
tab.add_message(InfoMessage(msg), typ=2)
tab.add_message(PersistentInfoMessage(msg))
if code == '409':
if config.get('alternative_nickname') != '':
if not tab.joined:
......@@ -2134,11 +2150,10 @@ class Core:
else:
if not tab.joined:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'You can join the room with another nick, '
'by typing "/join /other_nick"'
),
typ=2,
)
)
self.refresh_window()
......
......@@ -39,7 +39,13 @@ from poezio.logger import logger
from poezio.roster import roster
from poezio.text_buffer import CorrectionError, AckError
from poezio.theming import dump_tuple, get_theme
from poezio.ui.types import XMLLog, Message as PMessage, BaseMessage, InfoMessage
from poezio.ui.types import (
XMLLog,
Message as PMessage,
BaseMessage,
InfoMessage,
PersistentInfoMessage,
)
from poezio.core.commands import dumb_callback
......@@ -327,7 +333,7 @@ class HandlerCore:
error = '\x19%s}%s\x19o' % (dump_tuple(get_theme().COLOR_CHAR_NACK),
error_msg)
if not tab.nack_message('\n' + error, message['id'], message['to']):
tab.add_message(InfoMessage(error), typ=0)
tab.add_message(InfoMessage(error))
self.core.refresh_window()
def on_normal_message(self, message):
......@@ -430,8 +436,7 @@ class HandlerCore:
history=delayed,
identifier=message['id'],
jid=jid,
),
typ=1,
)
)
if not own and 'private' in config.get('beep_on').split():
......@@ -811,7 +816,7 @@ class HandlerCore:
identifier=message['id'],
)
typ = 2
tab.add_message(ui_msg, typ)
tab.add_message(ui_msg)
if highlight:
self.core.events.trigger('highlight', message, tab)
......@@ -906,8 +911,7 @@ class HandlerCore:
user=user,
identifier=message['id'],
jid=message['from'],
),
typ=1,
)
)
if sent:
tab.set_last_sent_message(message, correct=replaced)
......@@ -1400,53 +1404,53 @@ class HandlerCore:
if show_unavailable or hide_unavailable or non_priv or logging_off\
or non_anon or semi_anon or full_anon:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'Info: A configuration change not privacy-related occurred.'
),
typ=2)
)
modif = True
if show_unavailable:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'Info: The unavailable members are now shown.'
),
typ=2)
)
elif hide_unavailable:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'Info: The unavailable members are now hidden.',
),
typ=2)
)
if non_anon:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'\x191}Warning:\x19%(info_col)s} The room is now not anonymous. (public JID)' % info_col
),
typ=2)
)
elif semi_anon:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'Info: The room is now semi-anonymous. (moderators-only JID)',
),
typ=2)
)
elif full_anon:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'Info: The room is now fully anonymous.',
),
typ=2)
)
if logging_on:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'\x191}Warning: \x19%(info_col)s}This room is publicly logged' % info_col
),
typ=2)
)
elif logging_off:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
'Info: This room is not logged anymore.',
),
typ=2)
)
if modif:
self.core.refresh_window()
......@@ -1489,18 +1493,18 @@ class HandlerCore:
if nick_from:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
"%(user)s set the subject to: \x19%(text_col)s}%(subject)s" % fmt,
time=time,
),
typ=2)
)
else:
tab.add_message(
InfoMessage(
PersistentInfoMessage(
"The subject is: \x19%(text_col)s}%(subject)s" % fmt,
time=time,
),
typ=2)
)
tab.topic = subject
tab.topic_from = nick_from
if self.core.tabs.by_name_and_class(
......
"""
Module defining structures useful to the core class and related methods
"""
from dataclasses import dataclass
from typing import Any, Callable, List, Dict
__all__ = [
'ERROR_AND_STATUS_CODES', 'DEPRECATED_ERRORS', 'POSSIBLE_SHOW', 'Status',
......@@ -51,23 +53,11 @@ POSSIBLE_SHOW = {
}
@dataclass
class Status:
__slots__ = ('show', 'message')
def __init__(self, show, message):
self.show = show
self.message = message
class Command:
__slots__ = ('func', 'desc', 'comp', 'short_desc', 'usage')
def __init__(self, func, desc, comp, short_desc, usage):
self.func = func
self.desc = desc
self.comp = comp
self.short_desc = short_desc
self.usage = usage
show: str
message: str
class Completion:
......@@ -75,8 +65,13 @@ class Completion:
A completion result essentially currying the input completion call.
"""
__slots__ = ['func', 'args', 'kwargs', 'comp_list']
def __init__(self, func, comp_list, *args, **kwargs):
def __init__(
self,
func: Callable[..., Any],
comp_list: List[str],
*args: Any,
**kwargs: Any
) -> None:
self.func = func
self.comp_list = comp_list
self.args = args
......@@ -84,3 +79,12 @@ class Completion:
def run(self):
return self.func(self.comp_list, *self.args, **self.kwargs)
@dataclass
class Command:
__slots__ = ('func', 'desc', 'comp', 'short_desc', 'usage')
func: Callable[..., Any]
desc: str
comp: Callable[['windows.Input'], Completion]
short_desc: str
usage: str
......@@ -347,16 +347,16 @@ class Tabs:
if new_pos < len(self._tabs):
old_tab = self._tabs[old_pos]
self._tabs[new_pos], self._tabs[
old_pos] = old_tab, tabs.GapTab(self)
old_pos] = old_tab, tabs.GapTab(None)
else:
self._tabs.append(self._tabs[old_pos])
self._tabs[old_pos] = tabs.GapTab(self)
self._tabs[old_pos] = tabs.GapTab(None)
else:
if new_pos > old_pos:
self._tabs.insert(new_pos, tab)
self._tabs[old_pos] = tabs.GapTab(self)
self._tabs[old_pos] = tabs.GapTab(None)
elif new_pos < old_pos:
self._tabs[old_pos] = tabs.GapTab(self)
self._tabs[old_pos] = tabs.GapTab(None)
self._tabs.insert(new_pos, tab)
else:
return False
......
"""
Module containing various decorators
"""
from typing import Any, Callable, List, Optional
from typing import (
cast,
Any,
Callable,
List,
Optional,
TypeVar,
TYPE_CHECKING,
)
from poezio import common
if TYPE_CHECKING:
from poezio.tabs import RosterInfoTab
T = TypeVar('T', bound=Callable[..., Any])
class RefreshWrapper:
def __init__(self):
def __init__(self) -> None:
self.core = None
def conditional(self, func: Callable) -> Callable:
def conditional(self, func: T) -> T:
"""
Decorator to refresh the UI if the wrapped function
returns True
"""
def wrap(*args, **kwargs):
def wrap(*args: Any, **kwargs: Any) -> Any:
ret = func(*args, **kwargs)
if self.core and ret:
self.core.refresh_window()
return ret
return wrap
return cast(T, wrap)
def always(self, func: Callable) -> Callable:
def always(self, func: T) -> T:
"""
Decorator that refreshs the UI no matter what after the function
"""
def wrap(*args, **kwargs):
def wrap(*args: Any, **kwargs: Any) -> Any:
ret = func(*args, **kwargs)
if self.core:
self.core.refresh_window()
return ret
return wrap
return cast(T, wrap)
def update(self, func: Callable) -> Callable:
def update(self, func: T) -> T:
"""
Decorator that only updates the screen
"""
def wrap(*args, **kwargs):
def wrap(*args: Any, **kwargs: Any) -> Any:
ret = func(*args, **kwargs)
if self.core:
self.core.doupdate()
return ret
return wrap
return cast(T, wrap)
refresh_wrapper = RefreshWrapper()
......@@ -61,32 +75,32 @@ class CommandArgParser:
"""
@staticmethod
def raw(func: Callable) -> Callable:
def raw(func: T) -> T:
"""Just call the function with a single string, which is the original string
untouched
"""
def wrap(self, args, *a, **kw):
def wrap(self: Any, args: Any, *a: Any, **kw: Any) -> Any:
return func(self, args, *a, **kw)
return wrap
return cast(T, wrap)
@staticmethod
def ignored(func: Callable) -> Callable:
def ignored(func: T) -> T:
"""
Call the function without any argument
"""
def wrap(self, args=None, *a, **kw):
def wrap(self: Any, args: Any = None, *a: Any, **kw: Any) -> Any:
return func(self, *a, **kw)
return wrap
return cast(T, wrap)
@staticmethod
def quoted(mandatory: int,
optional=0,
optional: int = 0,
defaults: Optional[List[Any]] = None,
ignore_trailing_arguments=False):
ignore_trailing_arguments: bool = False) -> Callable[[T], T]:
"""The function receives a list with a number of arguments that is between
the numbers `mandatory` and `optional`.
......@@ -131,8 +145,8 @@ class CommandArgParser:
"""
default_args_outer = defaults or []
def first(func: Callable):
def second(self, args: str, *a, **kw):
def first(func: T) -> T:
def second(self: Any, args: str, *a: Any, **kw: Any) -> Any:
default_args = default_args_outer
if args and args.strip():
split_args = common.shell_split(args)
......@@ -156,8 +170,7 @@ class CommandArgParser:
res[-1] += " " + " ".join(split_args)
return func(self, res, *a, **kw)
return second
return cast(T, second)
return first
......@@ -166,11 +179,11 @@ command_args_parser = CommandArgParser()
def deny_anonymous(func: Callable) -> Callable:
"""Decorator to disable commands when using an anonymous account."""
def wrap(self: 'RosterInfoTab', *args, **kwargs):
def wrap(self: 'RosterInfoTab', *args: Any, **kwargs: Any) -> Any:
if self.core.xmpp.anon:
return self.core.information(
'This command is not available for anonymous accounts.',
'Info'
)
return func(self, *args, **kwargs)
return wrap
return cast(T, wrap)
......@@ -5,7 +5,8 @@ upstream.
TODO: Check that they are fixed and remove those hacks
"""
from slixmpp.stanza import Message
from typing import Callable, Any
from slixmpp import Message, Iq, ClientXMPP
from slixmpp.xmlstream import ET
import logging
......@@ -25,7 +26,7 @@ def has_identity(xmpp, jid, identity, on_true=None, on_false=None):
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=_cb)
def get_room_form(xmpp, room, callback):
def get_room_form(xmpp: ClientXMPP, room: str, callback: Callable[[Iq], Any]):
def _cb(result):
if result["type"] == "error":
return callback(None)
......
......@@ -11,13 +11,14 @@ conversations and roster changes
import mmap
import re
from typing import List, Dict, Optional, IO, Any
from typing import List, Dict, Optional, IO, Any, Union
from datetime import datetime
from poezio import common
from poezio.config import config
from poezio.xhtml import clean_text
from poezio.theming import dump_tuple, get_theme
from poezio.ui.types import Message, BaseMessage, LoggableTrait
import logging
......@@ -134,10 +135,7 @@ class Logger:
def log_message(self,
jid: str,
nick: str,
msg: str,
date: Optional[datetime] = None,
typ: int = 1) -> bool:
msg: Union[BaseMessage, Message]) -> bool:
"""
log the message in the appropriate jid's file
type:
......@@ -147,7 +145,16 @@ class Logger:
"""
if not config.get_by_tabname('use_log', jid):
return True
logged_msg = build_log_message(nick, msg, date=date, typ=typ)
if not isinstance(msg, LoggableTrait):
return True
date = msg.time
txt = msg.txt
typ = 2
nick = ''
if isinstance(msg, Message):
nick = msg.nickname
typ = 1
logged_msg = build_log_message(nick, txt, date=date, typ=typ)
if not logged_msg:
return True
jid = str(jid).replace('/', '\\')
......
......@@ -11,18 +11,39 @@ slix plugin
"""
from xml.etree import ElementTree as ET
from typing import (
Callable,
Optional,
TYPE_CHECKING,
)
from poezio.common import safeJID
from slixmpp import JID
from slixmpp.exceptions import IqError, IqTimeout
from slixmpp import (
JID,
ClientXMPP,
Iq,
)
import logging
log = logging.getLogger(__name__)
if TYPE_CHECKING:
from poezio.core import Core
from poezio.tabs import Tab
from slixmpp.plugins.xep_0004 import Form
NS_MUC_ADMIN = 'http://jabber.org/protocol/muc#admin'
NS_MUC_OWNER = 'http://jabber.org/protocol/muc#owner'
def destroy_room(xmpp, room, reason='', altroom=''):
def destroy_room(
xmpp: ClientXMPP,
room: str,
reason: str = '',
altroom: str = ''
) -> bool:
"""
destroy a room
"""
......@@ -42,7 +63,7 @@ def destroy_room(xmpp, room, reason='', altroom=''):
query.append(destroy)
iq.append(query)
def callback(iq):
def callback(iq: Iq) -> None:
if not iq or iq['type'] == 'error':
xmpp.core.information('Unable to destroy room %s' % room, 'Info')
else:
......@@ -52,23 +73,13 @@ def destroy_room(xmpp, room, reason='', altroom=''):
return True
def send_private_message(xmpp, jid, line):
"""
Send a private message
"""
jid = safeJID(jid)
xmpp.send_message(mto=jid, mbody=line, mtype='chat')
def send_groupchat_message(xmpp, jid, line):
"""
Send a message to the groupchat
"""
jid = safeJID(jid)
xmpp.send_message(mto=jid, mbody=line, mtype='groupchat')
def change_show(xmpp, jid: JID, own_nick: str, show, status):
def change_show(
xmpp: ClientXMPP,
jid: JID,
own_nick: str,
show: str,
status: Optional[str]
) -> None:
"""
Change our 'Show'
"""
......@@ -81,7 +92,7 @@ def change_show(xmpp, jid: JID, own_nick: str, show, status):
pres.send()
def change_subject(xmpp, jid, subject):
def change_subject(xmpp: ClientXMPP, jid: JID, subject: str) -> None:
"""
Change the room subject
"""
......@@ -92,7 +103,13 @@ def change_subject(xmpp, jid, subject):
msg.send()
def change_nick(core, jid, nick, status=None, show=None):
def change_nick(
core: 'Core',
jid: JID,
nick: str,
status: Optional[str] = None,
show: Optional[str] = None
) -> None:
"""
Change our own nick in a room
"""
......@@ -103,14 +120,16 @@ def change_nick(core, jid, nick, status=None, show=None):
presence.send()
def join_groupchat(core,
jid,
nick,
passwd='',
status=None,
show=None,
seconds=None,
tab=None):
def join_groupchat(
core: 'Core',
jid: JID,
nick: str,
passwd: str = '',
status: Optional[str] = None,
show: Optional[str] = None,
seconds: Optional[int] = None,
tab: Optional['Tab'] = None
) -> None:
xmpp = core.xmpp
stanza = xmpp.make_presence(
pto='%s/%s' % (jid, nick), pstatus=status, pshow=show)
......@@ -119,8 +138,10 @@ def join_groupchat(core,
passelement = ET.Element('password')
passelement.text = passwd
x.append(passelement)
def on_disco(iq):
if 'urn:xmpp:mam:2' in iq['disco_info'].get_features() or (tab and tab._text_buffer.last_message):
def on_disco(iq: Iq) -> None:
if ('urn:xmpp:mam:2' in iq['disco_info'].get_features()
or (tab and tab._text_buffer.last_message)):
history = ET.Element('{http://jabber.org/protocol/muc}history')
history.attrib['seconds'] = str(0)
x.append(history)
......@@ -136,13 +157,15 @@ def join_groupchat(core,
xmpp.plugin['xep_0045'].rooms[jid] = {}
xmpp.plugin['xep_0045'].our_nicks[jid] = to.resource
try:
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco)
except (IqError, IqTimeout):
return core.information('Failed to retrieve messages', 'Error')
xmpp.plugin['xep_0030'].get_info(jid=jid, callback=on_disco)
def leave_groupchat(xmpp, jid, own_nick, msg):
def leave_groupchat(
xmpp: ClientXMPP,
jid: JID,
own_nick: str,
msg: str
) -> None:
"""
Leave the groupchat
"""
......@@ -156,7 +179,14 @@ def leave_groupchat(xmpp, jid, own_nick, msg):
exc_info=True)
def set_user_role(xmpp, jid, nick, reason, role, callback=None):
def set_user_role(
xmpp: ClientXMPP,
jid: JID,
nick: str,
reason: str,
role: str,
callback: Callable[[Iq], None]
) -> None:
"""
(try to) Set the role of a MUC user
(role = 'none': eject user)
......@@ -172,21 +202,18 @@ def set_user_role(xmpp, jid, nick, reason, role, callback=None):
query.append(item)
iq.append(query)
iq['to'] = jid
if callback:
return iq.send(callback=callback)
try:
return iq.send()
except (IqError, IqTimeout) as e:
return e.iq
def set_user_affiliation(xmpp,
muc_jid,
affiliation,
nick=None,
jid=None,
reason=None,
callback=None):
iq.send(callback=callback)
def set_user_affiliation(
xmpp: ClientXMPP,
muc_jid: JID,
affiliation: str,
callback: Callable[[Iq], None],
nick: Optional[str] = None,
jid: Optional[JID] = None,
reason: Optional[str] = None
) -> None:
"""
(try to) Set the affiliation of a MUC user
"""
......@@ -212,18 +239,10 @@ def set_user_affiliation(xmpp,
query.append(item)
iq = xmpp.make_iq_set(query)
iq['to'] = muc_jid
if callback:
return iq.send(callback=callback)
try:
return xmpp.plugin['xep_0045'].set_affiliation(
str(muc_jid),
str(jid) if jid else None, nick, affiliation)
except:
log.debug('Error setting the affiliation: %s', exc_info=True)
return False
iq.send(callback=callback)
def cancel_config(xmpp, room):
def cancel_config(xmpp: ClientXMPP, room: str) -> None:
query = ET.Element('{http://jabber.org/protocol/muc#owner}query')
x = ET.Element('{jabber:x:data}x', type='cancel')
query.append(x)
......@@ -232,7 +251,7 @@ def cancel_config(xmpp, room):
iq.send()
def configure_room(xmpp, room, form):
def configure_room(xmpp: ClientXMPP, room: str, form: 'Form') -> None:
if form is None:
return
iq = xmpp.make_iq_set()
......
......@@ -104,7 +104,7 @@ def main():
logger.create_logger()
from poezio import roster
roster.create_roster()
roster.roster.reset()
from poezio.core.core import Core
......
from typing import List, Tuple, Any, TextIO, Union, Optional, Iterable, TypeVar
import sys
def split(s: str, comments: bool = ..., posix: bool = ...) -> List[str]: ...
if sys.version_info >= (3, 8):
def join(split_command: Iterable[str]) -> str: ...
def quote(s: str) -> str: ...
_SLT = TypeVar('_SLT', bound=shlex)
class shlex(Iterable[str]):
commenters: str
wordchars: str
whitespace: str
escape: str
quotes: str
escapedquotes: str
whitespace_split: bool
infile: str
instream: TextIO
source: str
debug: int
lineno: int
token: str
eof: str
if sys.version_info >= (3, 6):
punctuation_chars: str
if sys.version_info >= (3, 6):
def __init__(self, instream: Union[str, TextIO] = ..., infile: Optional[str] = ...,
posix: bool = ..., punctuation_chars: Union[bool, str] = ...) -> None: ...
else:
def __init__(self, instream: Union[str, TextIO] = ..., infile: Optional[str] = ...,
posix: bool = ...) -> None: ...
def get_token(self) -> Tuple[int, int, str]: ...
def push_token(self, tok: str) -> None: ...
def read_token(self) -> str: ...
def sourcehook(self, filename: str) -> Tuple[str, TextIO]: ...
# TODO argument types
def push_source(self, newstream: Any, newfile: Any = ...) -> None: ...
def pop_source(self) -> None: ...
def error_leader(self, infile: str = ...,
lineno: int = ...) -> None: ...
def __iter__(self: _SLT) -> _SLT: ...
def __next__(self) -> str: ...
......@@ -10,6 +10,8 @@ Defines the Roster and RosterGroup classes
import logging
log = logging.getLogger(__name__)
from typing import List
from poezio.config import config
from poezio.contact import Contact
from poezio.roster_sorting import SORTING_METHODS, GROUP_SORTING_METHODS
......@@ -18,6 +20,7 @@ from os import path as p
from datetime import datetime
from poezio.common import safeJID
from slixmpp.exceptions import IqError, IqTimeout
from slixmpp import JID
class Roster:
......@@ -29,6 +32,22 @@ class Roster:
DEFAULT_FILTER = (lambda x, y: None, None)
def __init__(self):
self.__node = None
# A tuple(function, *args) function to filter contacts
# on search, for example
self.contact_filter = self.DEFAULT_FILTER
self.groups = {}
self.contacts = {}
self.length = 0
self.connected = 0
self.folded_groups = []
# Used for caching roster infos
self.last_built = datetime.now()
self.last_modified = datetime.now()
def reset(self):
"""
node: the RosterSingle from slixmpp
"""
......@@ -143,7 +162,7 @@ class Roster:
"""Subscribe to a jid"""
self.__node.subscribe(jid)
def jids(self):
def jids(self) -> List[JID]:
"""List of the contact JIDS"""
l = []
for key in self.__node.keys():
......@@ -335,11 +354,6 @@ class RosterGroup:
return len([1 for contact in self.contacts if len(contact)])
def create_roster():
"Create the global roster object"
global roster
roster = Roster()
# Shared roster object
roster = None
roster = Roster()
......@@ -28,6 +28,7 @@ from typing import (
List,
Optional,
Union,
Tuple,
TYPE_CHECKING,
)
......@@ -46,12 +47,20 @@ from poezio.text_buffer import TextBuffer
from poezio.theming import get_theme, dump_tuple
from poezio.ui.funcs import truncate_nick
from poezio.ui.consts import LONG_FORMAT_LENGTH
from poezio.ui.types import BaseMessage, InfoMessage, Message
from poezio.ui.types import (
BaseMessage,
InfoMessage,
Message,
PersistentInfoMessage,
LoggableTrait,
)
from slixmpp import JID, InvalidJID, Message as SMessage
if TYPE_CHECKING:
from _curses import _CursesWindow # pylint: disable=E0611
from poezio.size_manager import SizeManager
from poezio.core.core import Core
log = logging.getLogger(__name__)
......@@ -117,7 +126,7 @@ class Tab:
height = 1
width = 1
def __init__(self, core):
def __init__(self, core: 'Core'):
self.core = core
self.nb = 0
if not hasattr(self, 'name'):
......@@ -133,7 +142,7 @@ class Tab:
self.commands = {} # and their own commands
@property
def size(self) -> int:
def size(self) -> 'SizeManager':
return self.core.size
@staticmethod
......@@ -196,7 +205,7 @@ class Tab:
self._state = 'normal'
@staticmethod
def resize(scr: '_CursesWindow'):
def initial_resize(scr: '_CursesWindow'):
Tab.height, Tab.width = scr.getmaxyx()
windows.base_wins.TAB_WIN = scr
......@@ -327,7 +336,7 @@ class Tab:
else:
return False
def refresh_tab_win(self):
def refresh_tab_win(self) -> None:
if config.get('enable_vertical_tab_list'):
left_tab_win = self.core.left_tab_win
if left_tab_win and not self.size.core_degrade_x:
......@@ -371,12 +380,12 @@ class Tab:
"""
pass
def update_commands(self):
def update_commands(self) -> None:
for c in self.plugin_commands:
if c not in self.commands:
self.commands[c] = self.plugin_commands[c]
def update_keys(self):
def update_keys(self) -> None:
for k in self.plugin_keys:
if k not in self.key_func:
self.key_func[k] = self.plugin_keys[k]
......@@ -435,7 +444,7 @@ class Tab:
"""
pass
def on_close(self):
def on_close(self) -> None:
"""
Called when the tab is to be closed
"""
......@@ -443,7 +452,7 @@ class Tab:
self.input.on_delete()
self.closed = True
def matching_names(self) -> List[str]:
def matching_names(self) -> List[Tuple[int, str]]:
"""
Returns a list of strings that are used to name a tab with the /win
command. For example you could switch to a tab that returns
......@@ -532,7 +541,7 @@ class ChatTab(Tab):
desc='Fix the last message with whatever you want.',
shortdesc='Correct the last message.',
completion=self.completion_correct)
self.chat_state = None
self.chat_state = None # type: Optional[str]
self.update_commands()
self.update_keys()
......@@ -542,6 +551,11 @@ class ChatTab(Tab):
return self._name
return self._jid.full
@property
def log_name(self) -> str:
"""Name used for the log filename"""
return self.jid.bare
@name.setter
def name(self, value: Union[JID, str]) -> None:
if isinstance(value, JID):
......@@ -571,18 +585,17 @@ class ChatTab(Tab):
def general_jid(self) -> JID:
raise NotImplementedError
def log_message(self, message: BaseMessage, typ=1):
def log_message(self, message: BaseMessage):
"""
Log the messages in the archives.
"""
name = self.jid.bare
if not isinstance(message, Message):
if not isinstance(message, LoggableTrait):
return
if not logger.log_message(name, message.nickname, message.txt, date=message.time, typ=typ):
if not logger.log_message(self.log_name, message):
self.core.information('Unable to write in the log file', 'Error')
def add_message(self, message: BaseMessage, typ=1):
self.log_message(message, typ=typ)
def add_message(self, message: BaseMessage):
self.log_message(message)
self._text_buffer.add_message(message)
def modify_message(self,
......@@ -595,7 +608,7 @@ class ChatTab(Tab):
message = self._text_buffer.modify_message(
txt, old_id, new_id, user=user, jid=jid)
if message:
self.log_message(message, typ=1)
self.log_message(message)
self.text_win.modify_message(message.identifier, message)
self.core.refresh_window()
return True
......@@ -667,11 +680,11 @@ class ChatTab(Tab):
self._text_buffer.messages = []
self.text_win.rebuild_everything(self._text_buffer)
def check_send_chat_state(self):
def check_send_chat_state(self) -> bool:
"If we should send a chat state"
return True
def send_chat_state(self, state, always_send=False):
def send_chat_state(self, state: str, always_send: bool = False) -> None:
"""
Send an empty chatstate message
"""
......@@ -691,9 +704,8 @@ class ChatTab(Tab):
x = ET.Element('{%s}x' % NS_MUC_USER)
msg.append(x)
msg.send()
return True
def send_composing_chat_state(self, empty_after):
def send_composing_chat_state(self, empty_after: bool) -> None:
"""
Send the "active" or "composing" chatstate, depending
on the the current status of the input
......@@ -729,7 +741,7 @@ class ChatTab(Tab):
self.core.add_timed_event(new_event)
self.timed_event_not_paused = new_event
def cancel_paused_delay(self):
def cancel_paused_delay(self) -> None:
"""
Remove that event from the list and set it to None.
Called for example when the input is emptied, or when the message
......@@ -741,7 +753,7 @@ class ChatTab(Tab):
self.core.remove_timed_event(self.timed_event_not_paused)
self.timed_event_not_paused = None
def set_last_sent_message(self, msg, correct=False):
def set_last_sent_message(self, msg: SMessage, correct: bool = False) -> None:
"""Ensure last_sent_message is set with the correct attributes"""
if correct:
# XXX: Is the copy needed. Is the object passed here reused
......@@ -751,7 +763,7 @@ class ChatTab(Tab):
self.last_sent_message = msg
@command_args_parser.raw
def command_correct(self, line):
def command_correct(self, line: str) -> None:
"""
/correct <fixed message>
"""
......@@ -777,7 +789,7 @@ class ChatTab(Tab):
return self.core.status.show in ('xa', 'away') or\
(hasattr(self, 'directed_presence') and not self.directed_presence)
def move_separator(self):
def move_separator(self) -> None:
self.text_win.remove_line_separator()
self.text_win.add_line_separator(self._text_buffer)
self.text_win.refresh()
......@@ -786,7 +798,7 @@ class ChatTab(Tab):
def get_conversation_messages(self):
return self._text_buffer.messages
def check_scrolled(self):
def check_scrolled(self) -> None:
if self.text_win.pos != 0:
self.state = 'scrolled'
......@@ -990,8 +1002,7 @@ class OneToOneTab(ChatTab):
if status.show in SHOW_NAME:
msg += 'show: %s, ' % SHOW_NAME[status.show]
self.add_message(
InfoMessage(txt=msg[:-2]),
typ=2,
PersistentInfoMessage(txt=msg[:-2])
)
def ack_message(self, msg_id: str, msg_jid: JID):
......
......@@ -178,7 +178,7 @@ class ConversationTab(OneToOneTab):
(' and their last status was %s' % status)
if status else '',
)
self.add_message(InfoMessage(msg), typ=0)
self.add_message(InfoMessage(msg))
self.core.refresh_window()
self.core.xmpp.plugin['xep_0012'].get_last_activity(
......@@ -206,15 +206,12 @@ class ConversationTab(OneToOneTab):
'status': status,
}
),
typ=0,
)
return True
else:
self.add_message(
InfoMessage("No information available"),
typ=0,
)
return True
self.add_message(
InfoMessage("No information available"),
)
return True
@command_args_parser.quoted(0, 1)
def command_version(self, args):
......
This diff is collapsed.
......@@ -27,7 +27,12 @@ from poezio.decorators import refresh_wrapper
from poezio.logger import logger
from poezio.theming import get_theme, dump_tuple
from poezio.decorators import command_args_parser
from poezio.ui.types import BaseMessage, Message, InfoMessage
from poezio.ui.types import (
BaseMessage,
InfoMessage,
Message,
PersistentInfoMessage,
)
log = logging.getLogger(__name__)
......@@ -69,6 +74,11 @@ class PrivateTab(OneToOneTab):
self.update_commands()
self.update_keys()
@property
def log_name(self) -> str:
"""Overriden from ChatTab because this is a case where we want the full JID"""
return self.jid.full
def remote_user_color(self):
user = self.parent_muc.get_user_by_name(self.jid.resource)
if user:
......@@ -105,16 +115,6 @@ class PrivateTab(OneToOneTab):
def remove_information_element(plugin_name):
del PrivateTab.additional_information[plugin_name]
def log_message(self, msg: BaseMessage, typ=1):
"""
Log the messages in the archives.
"""
if not isinstance(msg, Message):
return
if not logger.log_message(
self.jid.full, msg.nickname, msg.txt, date=msg.time, typ=typ):
self.core.information('Unable to write in the log file', 'Error')
def on_close(self):
super().on_close()
self.parent_muc.privates.remove(self)
......@@ -145,7 +145,7 @@ class PrivateTab(OneToOneTab):
@refresh_wrapper.always
@command_args_parser.raw
def command_say(self, line, attention=False, correct=False):
def command_say(self, line: str, attention: bool = False, correct: bool = False) -> None:
if not self.on:
return
our_jid = JID(self.jid.bare)
......@@ -310,7 +310,7 @@ class PrivateTab(OneToOneTab):
display a message.
"""
self.add_message(
InfoMessage(
PersistentInfoMessage(
'\x19%(nick_col)s}%(old)s\x19%(info_col)s} is now '
'known as \x19%(nick_col)s}%(new)s' % {
'old': old_nick,
......@@ -319,7 +319,7 @@ class PrivateTab(OneToOneTab):
'info_col': dump_tuple(get_theme().COLOR_INFORMATION_TEXT)
},
),
typ=2)
)
new_jid = self.jid.bare + '/' + user.nick
self.name = new_jid
return self.core.tabs.current_tab is self
......@@ -339,7 +339,7 @@ class PrivateTab(OneToOneTab):
if not status_message:
self.add_message(
InfoMessage(
PersistentInfoMessage(
'\x19%(quit_col)s}%(spec)s \x19%(nick_col)s}'
'%(nick)s\x19%(info_col)s} has left the room' % {
'nick': user.nick,
......@@ -349,10 +349,10 @@ class PrivateTab(OneToOneTab):
'info_col': dump_tuple(theme.COLOR_INFORMATION_TEXT)
},
),
typ=2)
)
else:
self.add_message(
InfoMessage(
PersistentInfoMessage(
'\x19%(quit_col)s}%(spec)s \x19%(nick_col)s}'
'%(nick)s\x19%(info_col)s} has left the room'
' (%(status)s)' % {
......@@ -364,7 +364,7 @@ class PrivateTab(OneToOneTab):
'info_col': dump_tuple(theme.COLOR_INFORMATION_TEXT)
},
),
typ=2)
)
return self.core.tabs.current_tab is self
@refresh_wrapper.conditional
......@@ -383,7 +383,7 @@ class PrivateTab(OneToOneTab):
if user:
color = dump_tuple(user.color)
self.add_message(
InfoMessage(
PersistentInfoMessage(
'\x19%(join_col)s}%(spec)s \x19%(color)s}%(nick)s\x19'
'%(info_col)s} joined the room' % {
'nick': nick,
......@@ -393,18 +393,18 @@ class PrivateTab(OneToOneTab):
'info_col': dump_tuple(theme.COLOR_INFORMATION_TEXT)
},
),
typ=2)
)
return self.core.tabs.current_tab is self
def activate(self, reason=None):
self.on = True
if reason:
self.add_message(InfoMessage(reason), typ=2)
self.add_message(PersistentInfoMessage(reason))
def deactivate(self, reason=None):
self.on = False
if reason:
self.add_message(InfoMessage(reason), typ=2)
self.add_message(PersistentInfoMessage(reason))
def matching_names(self):
return [(3, self.jid.resource), (4, self.name)]
......@@ -420,6 +420,5 @@ class PrivateTab(OneToOneTab):
nickname='Error',
nick_color=theme.COLOR_ERROR_MSG,
),
typ=2,
)
self.core.refresh_window()
......@@ -377,7 +377,7 @@ class RosterInfoTab(Tab):
'info_col': dump_tuple(get_theme().COLOR_INFORMATION_TEXT),
'jid': message['from'],
}
tab.add_message(InfoMessage(message), typ=0)
tab.add_message(InfoMessage(message))
@command_args_parser.ignored
def command_list_blocks(self):
......
......@@ -32,6 +32,7 @@ from poezio.ui.types import (
if TYPE_CHECKING:
from poezio.windows.text_win import TextWin
from poezio.user import User
class CorrectionError(Exception):
......@@ -249,7 +250,7 @@ class TextBuffer:
new_id: str,
highlight: bool = False,
time: Optional[datetime] = None,
user: Optional[str] = None,
user: Optional['User'] = None,
jid: Optional[str] = None) -> Message:
"""
Correct a message in a text buffer.
......
......@@ -12,7 +12,6 @@ from poezio.ui.consts import (
)
class BaseMessage:
__slots__ = ('txt', 'time', 'identifier')
......@@ -38,11 +37,20 @@ class InfoMessage(BaseMessage):
super().__init__(txt=txt, identifier=identifier, time=time)
class MucOwnLeaveMessage(InfoMessage):
class LoggableTrait:
"""Trait for classes of messages that should go through the logger"""
pass
class PersistentInfoMessage(InfoMessage, LoggableTrait):
pass
class MucOwnLeaveMessage(InfoMessage, LoggableTrait):
"""Status message displayed on our room leave/kick/ban"""
class MucOwnJoinMessage(InfoMessage):
class MucOwnJoinMessage(InfoMessage, LoggableTrait):
"""Status message displayed on our room join"""
......@@ -97,7 +105,7 @@ class StatusMessage(BaseMessage):
self.txt = self.format_string.format(**real_args)
class Message(BaseMessage):
class Message(BaseMessage, LoggableTrait):
__slots__ = ('txt', 'nick_color', 'time', 'nickname', 'user', 'delayed', 'history',
'identifier', 'top', 'highlight', 'me', 'old_message', 'revisions',
'jid', 'ack')
......
......@@ -55,7 +55,7 @@ class User:
else:
self.color = choice(get_theme().LIST_COLOR_NICKNAMES)
def set_deterministic_color(self):
def set_deterministic_color(self) -> None:
theme = get_theme()
if theme.ccg_palette:
# use XEP-0392 CCG
......
......@@ -3,6 +3,8 @@ Module defining all the "info wins", ie the bar which is on top of the
info buffer in normal tabs
"""
from typing import Optional, Dict, TYPE_CHECKING, Any
import logging
log = logging.getLogger(__name__)
......@@ -13,6 +15,11 @@ from poezio.windows.base_wins import Win
from poezio.ui.funcs import truncate_nick
from poezio.theming import get_theme, to_curses_attr
if TYPE_CHECKING:
from poezio.user import User
from poezio.tabs import MucTab
from poezio.windows import TextWin
class InfoWin(Win):
"""
......@@ -260,10 +267,16 @@ class MucInfoWin(InfoWin):
__slots__ = ()
def __init__(self):
def __init__(self) -> None:
InfoWin.__init__(self)
def refresh(self, room, window=None, user=None, information=None):
def refresh(
self,
room: 'MucTab',
window: Optional['TextWin'] = None,