mirror of
https://github.com/matrix-org/pantalaimon.git
synced 2024-10-01 03:35:38 -04:00
panctl: Improve the completions.
This commit is contained in:
Damir Jelić
parent
10bb16b210
commit
6adedbaa0b
@ -23,19 +23,75 @@ use_asyncio_event_loop()
|
||||
@attr.s
|
||||
class PanCompleter(Completer):
|
||||
"""Completer for panctl commands."""
|
||||
|
||||
commands = attr.ib(type=List[str])
|
||||
ctl = attr.ib()
|
||||
devices = attr.ib()
|
||||
|
||||
def complete_commands(self, last_word):
|
||||
"""Complete the available commands."""
|
||||
compl_words = []
|
||||
compl_words = self.filter_words(self.commands, last_word)
|
||||
for compl_word in compl_words:
|
||||
yield Completion(compl_word, -len(last_word))
|
||||
|
||||
for command in self.commands:
|
||||
if last_word in command:
|
||||
compl_words.append(command)
|
||||
def complete_users(self, last_word, pan_user):
|
||||
devices = self.devices.list(
|
||||
pan_user,
|
||||
dbus_interface="org.pantalaimon.devices"
|
||||
)
|
||||
users = set(device["user_id"] for device in devices)
|
||||
compl_words = self.filter_words(users, last_word)
|
||||
|
||||
for compl_word in compl_words:
|
||||
yield Completion(compl_word, -len(last_word))
|
||||
|
||||
return ""
|
||||
|
||||
def complete_devices(self, last_word, pan_user, user_id):
|
||||
devices = self.devices.list_user_devices(
|
||||
pan_user,
|
||||
user_id,
|
||||
dbus_interface="org.pantalaimon.devices"
|
||||
)
|
||||
device_ids = [device["device_id"] for device in devices]
|
||||
compl_words = self.filter_words(device_ids, last_word)
|
||||
|
||||
for compl_word in compl_words:
|
||||
yield Completion(compl_word, -len(last_word))
|
||||
|
||||
return ""
|
||||
|
||||
def filter_words(self, words, last_word):
|
||||
compl_words = []
|
||||
|
||||
for word in words:
|
||||
if last_word in word:
|
||||
compl_words.append(word)
|
||||
|
||||
return compl_words
|
||||
|
||||
def complete_verification(self, command, last_word, words):
|
||||
def complete_pan_users():
|
||||
users = self.ctl.list_users(
|
||||
dbus_interface="org.pantalaimon.control"
|
||||
)
|
||||
compl_words = self.filter_words([i[0] for i in users], last_word)
|
||||
|
||||
for compl_word in compl_words:
|
||||
yield Completion(compl_word, -len(last_word))
|
||||
|
||||
if len(words) == 2:
|
||||
return complete_pan_users()
|
||||
elif len(words) == 3:
|
||||
pan_user = words[1]
|
||||
return self.complete_users(last_word, pan_user)
|
||||
elif len(words) == 4:
|
||||
pan_user = words[1]
|
||||
user_id = words[2]
|
||||
return self.complete_devices(last_word, pan_user, user_id)
|
||||
|
||||
return ""
|
||||
|
||||
def get_completions(self, document, complete_event):
|
||||
"""Build the completions."""
|
||||
text_before_cursor = document.text_before_cursor
|
||||
@ -47,6 +103,19 @@ class PanCompleter(Completer):
|
||||
if len(words) == 1:
|
||||
return self.complete_commands(last_word)
|
||||
|
||||
if len(words) > 1:
|
||||
command = words[0]
|
||||
|
||||
if command in [
|
||||
"start-verification",
|
||||
"accept-verification",
|
||||
"confirm-verification",
|
||||
"cancel-verification",
|
||||
"verify-device",
|
||||
"unverify-device",
|
||||
]:
|
||||
return self.complete_verification(command, last_word, words)
|
||||
|
||||
return ""
|
||||
|
||||
|
||||
@ -60,6 +129,9 @@ class PanCtl:
|
||||
"list-users",
|
||||
"export-keys",
|
||||
"import-keys",
|
||||
"verify-device",
|
||||
"unverify-device",
|
||||
"start-verification",
|
||||
"accept-verification",
|
||||
"confirm-verification"
|
||||
]
|
||||
@ -140,8 +212,8 @@ class PanCtl:
|
||||
dbus_interface="org.pantalaimon.control"
|
||||
)
|
||||
print("pantalaimon users:")
|
||||
for user, devic in users:
|
||||
print(" ", user, devic)
|
||||
for user, device in users:
|
||||
print(" ", user, device)
|
||||
|
||||
def import_keys(self, args):
|
||||
try:
|
||||
@ -187,7 +259,7 @@ class PanCtl:
|
||||
|
||||
async def loop(self):
|
||||
"""Event loop for panctl."""
|
||||
completer = PanCompleter(self.commands)
|
||||
completer = PanCompleter(self.commands, self.ctl, self.devices)
|
||||
promptsession = PromptSession("panctl> ", completer=completer)
|
||||
|
||||
while True:
|
||||
|
||||
@ -219,7 +219,7 @@ class PanStore:
|
||||
query = Accounts.select()
|
||||
|
||||
for account in query:
|
||||
device_store = []
|
||||
device_store = defaultdict(dict)
|
||||
|
||||
for d in account.device_keys:
|
||||
|
||||
@ -231,13 +231,13 @@ class PanStore:
|
||||
except IndexError:
|
||||
trust_state = TrustState.unset
|
||||
|
||||
device_store.append({
|
||||
device_store[d.user_id][d.device_id] = {
|
||||
"user_id": d.user_id,
|
||||
"device_id": d.device_id,
|
||||
"fingerprint_key": d.fp_key,
|
||||
"sender_key": d.sender_key,
|
||||
"trust_state": trust_state.name
|
||||
})
|
||||
}
|
||||
|
||||
store[account.user_id] = device_store
|
||||
|
||||
|
||||
@ -8,6 +8,7 @@ from dbus.mainloop.glib import DBusGMainLoop
|
||||
from gi.repository import GLib
|
||||
|
||||
from queue import Empty
|
||||
from nio.store import TrustState
|
||||
|
||||
from pantalaimon.store import PanStore
|
||||
from pantalaimon.thread_messages import (
|
||||
@ -32,9 +33,36 @@ class Devices(dbus.service.Object):
|
||||
self.queue = queue
|
||||
|
||||
@dbus.service.method("org.pantalaimon.devices",
|
||||
out_signature="a{sa{saa{ss}}}")
|
||||
def list(self):
|
||||
return self.device_list
|
||||
in_signature="s",
|
||||
out_signature="aa{ss}")
|
||||
def list(self, pan_user):
|
||||
device_store = self.device_list.get(pan_user, None)
|
||||
|
||||
if not device_store:
|
||||
return []
|
||||
|
||||
device_list = [
|
||||
device for device_list in device_store.values() for device in
|
||||
device_list.values()
|
||||
]
|
||||
|
||||
return device_list
|
||||
|
||||
@dbus.service.method("org.pantalaimon.devices",
|
||||
in_signature="ss",
|
||||
out_signature="aa{ss}")
|
||||
def list_user_devices(self, pan_user, user_id):
|
||||
device_store = self.device_list.get(pan_user, None)
|
||||
|
||||
if not device_store:
|
||||
return []
|
||||
|
||||
device_list = device_store.get(user_id, None)
|
||||
|
||||
if not device_list:
|
||||
return []
|
||||
|
||||
return device_list.values()
|
||||
|
||||
@dbus.service.method("org.pantalaimon.devices",
|
||||
in_signature="sss")
|
||||
@ -88,21 +116,18 @@ class Devices(dbus.service.Object):
|
||||
def update_devices(self, message):
|
||||
device_store = self.device_list[message.user_id]
|
||||
|
||||
# TODO the store type got changed to a list, fix adding/removing of
|
||||
# devices.
|
||||
|
||||
# for user_id, device_dict in message.devices.items():
|
||||
# for device in device_dict.values():
|
||||
# if device.deleted:
|
||||
# device_store[user_id].remove(device.id, None)
|
||||
# else:
|
||||
# device_store[user_id][device.id] = {
|
||||
# "user_id": device.user_id,
|
||||
# "device_id": device.id,
|
||||
# "fingerprint_key": device.ed25519,
|
||||
# "sender_key": device.curve25519,
|
||||
# "trust_state": TrustState.unset.name,
|
||||
# }
|
||||
for user_id, device_dict in message.devices.items():
|
||||
for device in device_dict.values():
|
||||
if device.deleted:
|
||||
device_store[user_id].remove(device.id, None)
|
||||
else:
|
||||
device_store[user_id][device.id] = {
|
||||
"user_id": device.user_id,
|
||||
"device_id": device.id,
|
||||
"fingerprint_key": device.ed25519,
|
||||
"sender_key": device.curve25519,
|
||||
"trust_state": TrustState.unset.name,
|
||||
}
|
||||
|
||||
|
||||
class Control(dbus.service.Object):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user