6a7b7bc36a
This is a huge configuration refactoring. The configuration system now uses Pydantic settings, with better validation, better support for environment variables and other goodies. The main config files now use TOML instead of INI, this allow for better handling of complex types, notably dict. Plugins, components and frontends can declare their config settings in a class which will be imported dynamically.
835 lines
32 KiB
Python
835 lines
32 KiB
Python
#!/usr/bin/env python3
|
||
|
||
|
||
# Libervia CLI
|
||
# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
|
||
|
||
# This program is free software: you can redistribute it and/or modify
|
||
# it under the terms of the GNU Affero General Public License as published by
|
||
# the Free Software Foundation, either version 3 of the License, or
|
||
# (at your option) any later version.
|
||
|
||
# This program is distributed in the hope that it will be useful,
|
||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||
# GNU Affero General Public License for more details.
|
||
|
||
# You should have received a copy of the GNU Affero General Public License
|
||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||
|
||
import json
|
||
import os
|
||
import os.path
|
||
import time
|
||
import tempfile
|
||
import asyncio
|
||
import shlex
|
||
import re
|
||
from pathlib import Path
|
||
from typing import TYPE_CHECKING
|
||
from libervia.cli.constants import Const as C
|
||
from libervia.backend.core.i18n import _
|
||
from libervia.backend.core import exceptions
|
||
from libervia.backend.tools.common import regex
|
||
from libervia.backend.tools.common.ansi import ANSI as A
|
||
from libervia.backend.tools.common import uri as xmpp_uri
|
||
from libervia.backend.tools import config
|
||
from configparser import NoSectionError, NoOptionError
|
||
from collections import namedtuple
|
||
|
||
if TYPE_CHECKING:
|
||
from .base import LiberviaCli
|
||
|
||
# default arguments used for some known editors (editing with metadata)
|
||
VIM_SPLIT_ARGS = "-c 'set nospr|vsplit|wincmd w|next|wincmd w'"
|
||
EMACS_SPLIT_ARGS = '--eval "(split-window-horizontally)"'
|
||
EDITOR_ARGS_MAGIC = {
|
||
"vim": VIM_SPLIT_ARGS + " {content_file} {metadata_file}",
|
||
"nvim": VIM_SPLIT_ARGS + " {content_file} {metadata_file}",
|
||
"gvim": VIM_SPLIT_ARGS + " --nofork {content_file} {metadata_file}",
|
||
"emacs": EMACS_SPLIT_ARGS + " {content_file} {metadata_file}",
|
||
"xemacs": EMACS_SPLIT_ARGS + " {content_file} {metadata_file}",
|
||
"nano": " -F {content_file} {metadata_file}",
|
||
}
|
||
|
||
SECURE_UNLINK_MAX = 10
|
||
SECURE_UNLINK_DIR = ".backup"
|
||
METADATA_SUFF = "_metadata.json"
|
||
|
||
|
||
def format_time(timestamp):
|
||
"""Return formatted date for timestamp
|
||
|
||
@param timestamp(str,int,float): unix timestamp
|
||
@return (unicode): formatted date
|
||
"""
|
||
fmt = "%d/%m/%Y %H:%M:%S %Z"
|
||
return time.strftime(fmt, time.localtime(float(timestamp)))
|
||
|
||
|
||
def ansi_ljust(s, width):
|
||
"""ljust method handling ANSI escape codes"""
|
||
cleaned = regex.ansi_remove(s)
|
||
return s + " " * (width - len(cleaned))
|
||
|
||
|
||
def ansi_center(s, width):
|
||
"""ljust method handling ANSI escape codes"""
|
||
cleaned = regex.ansi_remove(s)
|
||
diff = width - len(cleaned)
|
||
half = diff / 2
|
||
return half * " " + s + (half + diff % 2) * " "
|
||
|
||
|
||
def ansi_rjust(s, width):
|
||
"""ljust method handling ANSI escape codes"""
|
||
cleaned = regex.ansi_remove(s)
|
||
return " " * (width - len(cleaned)) + s
|
||
|
||
|
||
def get_tmp_dir(config: config.Settings, cat_dir, sub_dir=None):
|
||
"""Return directory used to store temporary files
|
||
|
||
@param sat_conf(ConfigParser.ConfigParser): instance opened on sat configuration
|
||
@param cat_dir(str): directory of the category (e.g. "blog")
|
||
@param sub_dir(str): sub directory where data need to be put
|
||
profile can be used here, or special directory name
|
||
sub_dir will be escaped to be usable in path (use regex.path_unescape to find
|
||
initial str)
|
||
@return (Path): path to the dir
|
||
"""
|
||
path_elts = [config.common.local_dir, cat_dir]
|
||
if sub_dir is not None:
|
||
path_elts.append(regex.path_escape(sub_dir))
|
||
return Path(*path_elts)
|
||
|
||
|
||
def parse_args(host, cmd_line, **format_kw):
|
||
"""Parse command arguments
|
||
|
||
@param cmd_line(unicode): command line as found in sat.conf
|
||
@param format_kw: keywords used for formating
|
||
@return (list(unicode)): list of arguments to pass to subprocess function
|
||
"""
|
||
try:
|
||
# we split the arguments and add the known fields
|
||
# we split arguments first to avoid escaping issues in file names
|
||
return [a.format(**format_kw) for a in shlex.split(cmd_line)]
|
||
except ValueError as e:
|
||
host.disp(
|
||
"Couldn't parse editor cmd [{cmd}]: {reason}".format(cmd=cmd_line, reason=e)
|
||
)
|
||
return []
|
||
|
||
|
||
class BaseEdit(object):
|
||
"""base class for editing commands
|
||
|
||
This class allows to edit file for PubSub or something else.
|
||
It works with temporary files in SàT local_dir, in a "cat_dir" subdir
|
||
"""
|
||
|
||
def __init__(self, host: "LiberviaCli", cat_dir, use_metadata=False):
|
||
"""
|
||
@param sat_conf(ConfigParser.ConfigParser): instance opened on sat configuration
|
||
@param cat_dir(unicode): directory to use for drafts
|
||
this will be a sub-directory of SàT's local_dir
|
||
@param use_metadata(bool): True is edition need a second file for metadata
|
||
most of signature change with use_metadata with an additional metadata
|
||
argument.
|
||
This is done to raise error if a command needs metadata but forget the flag,
|
||
and vice versa
|
||
"""
|
||
self.host = host
|
||
self.cat_dir = cat_dir
|
||
self.use_metadata = use_metadata
|
||
|
||
def secure_unlink(self, path):
|
||
"""Unlink given path after keeping it for a while
|
||
|
||
This method is used to prevent accidental deletion of a draft
|
||
If there are more file in SECURE_UNLINK_DIR than SECURE_UNLINK_MAX,
|
||
older file are deleted
|
||
@param path(Path, str): file to unlink
|
||
"""
|
||
path = Path(path).resolve()
|
||
if not path.is_file:
|
||
raise OSError("path must link to a regular file")
|
||
if path.parent != get_tmp_dir(self.host.settings, self.cat_dir):
|
||
self.disp(
|
||
f"File {path} is not in SàT temporary hierarchy, we do not remove " f"it",
|
||
2,
|
||
)
|
||
return
|
||
# we have 2 files per draft with use_metadata, so we double max
|
||
unlink_max = SECURE_UNLINK_MAX * 2 if self.use_metadata else SECURE_UNLINK_MAX
|
||
backup_dir = get_tmp_dir(self.host.settings, self.cat_dir, SECURE_UNLINK_DIR)
|
||
if not os.path.exists(backup_dir):
|
||
os.makedirs(backup_dir)
|
||
filename = os.path.basename(path)
|
||
backup_path = os.path.join(backup_dir, filename)
|
||
# we move file to backup dir
|
||
self.host.disp(
|
||
"Backuping file {src} to {dst}".format(src=path, dst=backup_path),
|
||
1,
|
||
)
|
||
os.rename(path, backup_path)
|
||
# and if we exceeded the limit, we remove older file
|
||
backup_files = [os.path.join(backup_dir, f) for f in os.listdir(backup_dir)]
|
||
if len(backup_files) > unlink_max:
|
||
backup_files.sort(key=lambda path: os.stat(path).st_mtime)
|
||
for path in backup_files[: len(backup_files) - unlink_max]:
|
||
self.host.disp("Purging backup file {}".format(path), 2)
|
||
os.unlink(path)
|
||
|
||
async def run_editor(
|
||
self,
|
||
editor_args_opt,
|
||
content_file_path,
|
||
content_file_obj,
|
||
meta_file_path=None,
|
||
meta_ori=None,
|
||
):
|
||
"""Run editor to edit content and metadata
|
||
|
||
@param editor_args_opt(unicode): option in [cli] section in configuration for
|
||
specific args
|
||
@param content_file_path(str): path to the content file
|
||
@param content_file_obj(file): opened file instance
|
||
@param meta_file_path(str, Path, None): metadata file path
|
||
if None metadata will not be used
|
||
@param meta_ori(dict, None): original cotent of metadata
|
||
can't be used if use_metadata is False
|
||
"""
|
||
if not self.use_metadata:
|
||
assert meta_file_path is None
|
||
assert meta_ori is None
|
||
|
||
# we calculate hashes to check for modifications
|
||
import hashlib
|
||
|
||
content_file_obj.seek(0)
|
||
tmp_ori_hash = hashlib.sha1(content_file_obj.read()).digest()
|
||
content_file_obj.close()
|
||
|
||
# we prepare arguments
|
||
editor = self.host.cli_settings.editor
|
||
editor_args = self.host.cli_settings.editor_args_opt
|
||
if editor_args is None:
|
||
# no, we check if we know the editor and have special arguments
|
||
if self.use_metadata:
|
||
editor_args = EDITOR_ARGS_MAGIC.get(os.path.basename(editor), "")
|
||
else:
|
||
editor_args = ""
|
||
parse_kwargs = {"content_file": content_file_path}
|
||
if self.use_metadata:
|
||
parse_kwargs["metadata_file"] = meta_file_path
|
||
args = parse_args(self.host, editor_args, **parse_kwargs)
|
||
if not args:
|
||
args = [content_file_path]
|
||
|
||
# actual editing
|
||
editor_process = await asyncio.create_subprocess_exec(
|
||
editor, *[str(a) for a in args]
|
||
)
|
||
editor_exit = await editor_process.wait()
|
||
|
||
# edition will now be checked, and data will be sent if it was a success
|
||
if editor_exit != 0:
|
||
self.disp(
|
||
f"Editor exited with an error code, so temporary file has not be "
|
||
f"deleted, and item is not published.\nYou can find temporary file "
|
||
f"at {content_file_path}",
|
||
error=True,
|
||
)
|
||
else:
|
||
# main content
|
||
try:
|
||
with content_file_path.open("rb") as f:
|
||
content = f.read()
|
||
except (OSError, IOError):
|
||
self.disp(
|
||
f"Can read file at {content_file_path}, have it been deleted?\n"
|
||
f"Cancelling edition",
|
||
error=True,
|
||
)
|
||
self.host.quit(C.EXIT_NOT_FOUND)
|
||
|
||
# metadata
|
||
if self.use_metadata:
|
||
try:
|
||
with meta_file_path.open("rb") as f:
|
||
metadata = json.load(f)
|
||
except (OSError, IOError):
|
||
self.disp(
|
||
f"Can read file at {meta_file_path}, have it been deleted?\n"
|
||
f"Cancelling edition",
|
||
error=True,
|
||
)
|
||
self.host.quit(C.EXIT_NOT_FOUND)
|
||
except ValueError:
|
||
self.disp(
|
||
f"Can't parse metadata, please check it is correct JSON format. "
|
||
f"Cancelling edition.\nYou can find tmp file at "
|
||
f"{content_file_path} and temporary meta file at "
|
||
f"{meta_file_path}.",
|
||
error=True,
|
||
)
|
||
self.host.quit(C.EXIT_DATA_ERROR)
|
||
|
||
if self.use_metadata and not metadata.get("publish", True):
|
||
self.disp(
|
||
f'Publication blocked by "publish" key in metadata, cancelling '
|
||
f"edition.\n\ntemporary file path:\t{content_file_path}\nmetadata "
|
||
f"file path:\t{meta_file_path}",
|
||
error=True,
|
||
)
|
||
self.host.quit()
|
||
|
||
if len(content) == 0:
|
||
self.disp("Content is empty, cancelling the edition")
|
||
if content_file_path.parent != get_tmp_dir(self.host.settings, self.cat_dir):
|
||
self.disp(
|
||
"File are not in SàT temporary hierarchy, we do not remove them",
|
||
2,
|
||
)
|
||
self.host.quit()
|
||
self.disp(f"Deletion of {content_file_path}", 2)
|
||
os.unlink(content_file_path)
|
||
if self.use_metadata:
|
||
self.disp(f"Deletion of {meta_file_path}".format(meta_file_path), 2)
|
||
os.unlink(meta_file_path)
|
||
self.host.quit()
|
||
|
||
# time to re-check the hash
|
||
elif tmp_ori_hash == hashlib.sha1(content).digest() and (
|
||
not self.use_metadata or meta_ori == metadata
|
||
):
|
||
self.disp("The content has not been modified, cancelling the edition")
|
||
self.host.quit()
|
||
|
||
else:
|
||
# we can now send the item
|
||
content = content.decode("utf-8-sig") # we use utf-8-sig to avoid BOM
|
||
try:
|
||
if self.use_metadata:
|
||
await self.publish(content, metadata)
|
||
else:
|
||
await self.publish(content)
|
||
except Exception as e:
|
||
if self.use_metadata:
|
||
self.disp(
|
||
f"Error while sending your item, the temporary files have "
|
||
f"been kept at {content_file_path} and {meta_file_path}: "
|
||
f"{e}",
|
||
error=True,
|
||
)
|
||
else:
|
||
self.disp(
|
||
f"Error while sending your item, the temporary file has been "
|
||
f"kept at {content_file_path}: {e}",
|
||
error=True,
|
||
)
|
||
self.host.quit(1)
|
||
|
||
self.secure_unlink(content_file_path)
|
||
if self.use_metadata:
|
||
self.secure_unlink(meta_file_path)
|
||
|
||
async def publish(self, content):
|
||
# if metadata is needed, publish will be called with it last argument
|
||
raise NotImplementedError
|
||
|
||
def get_tmp_file(self):
|
||
"""Create a temporary file
|
||
|
||
@return (tuple(file, Path)): opened (w+b) file object and file path
|
||
"""
|
||
suff = "." + self.get_tmp_suff()
|
||
cat_dir_str = self.cat_dir
|
||
tmp_dir = get_tmp_dir(self.host.settings, self.cat_dir, self.profile)
|
||
if not tmp_dir.exists():
|
||
try:
|
||
tmp_dir.mkdir(parents=True)
|
||
except OSError as e:
|
||
self.disp(
|
||
f"Can't create {tmp_dir} directory: {e}",
|
||
error=True,
|
||
)
|
||
self.host.quit(1)
|
||
try:
|
||
fd, path = tempfile.mkstemp(
|
||
suffix=suff,
|
||
prefix=time.strftime(cat_dir_str + "_%Y-%m-%d_%H:%M:%S_"),
|
||
dir=tmp_dir,
|
||
text=True,
|
||
)
|
||
return os.fdopen(fd, "w+b"), Path(path)
|
||
except OSError as e:
|
||
self.disp(f"Can't create temporary file: {e}", error=True)
|
||
self.host.quit(1)
|
||
|
||
def get_current_file(self, profile):
|
||
"""Get most recently edited file
|
||
|
||
@param profile(unicode): profile linked to the draft
|
||
@return(Path): full path of current file
|
||
"""
|
||
# we guess the item currently edited by choosing
|
||
# the most recent file corresponding to temp file pattern
|
||
# in tmp_dir, excluding metadata files
|
||
tmp_dir = get_tmp_dir(self.host.settings, self.cat_dir, profile)
|
||
available = [
|
||
p
|
||
for p in tmp_dir.glob(f"{self.cat_dir}_*")
|
||
if not p.match(f"*{METADATA_SUFF}")
|
||
]
|
||
if not available:
|
||
self.disp(
|
||
f"Could not find any content draft in {tmp_dir}",
|
||
error=True,
|
||
)
|
||
self.host.quit(1)
|
||
return max(available, key=lambda p: p.stat().st_mtime)
|
||
|
||
async def get_item_data(self, service, node, item):
|
||
"""return formatted content, metadata (or not if use_metadata is false), and item id"""
|
||
raise NotImplementedError
|
||
|
||
def get_tmp_suff(self):
|
||
"""return suffix used for content file"""
|
||
return "xml"
|
||
|
||
async def get_item_path(self):
|
||
"""Retrieve item path (i.e. service and node) from item argument
|
||
|
||
This method is obviously only useful for edition of PubSub based features
|
||
"""
|
||
service = self.args.service
|
||
node = self.args.node
|
||
item = self.args.item
|
||
last_item = self.args.last_item
|
||
|
||
if self.args.current:
|
||
# user wants to continue current draft
|
||
content_file_path = self.get_current_file(self.profile)
|
||
self.disp("Continuing edition of current draft", 2)
|
||
content_file_obj = content_file_path.open("r+b")
|
||
# we seek at the end of file in case of an item already exist
|
||
# this will write content of the existing item at the end of the draft.
|
||
# This way no data should be lost.
|
||
content_file_obj.seek(0, os.SEEK_END)
|
||
elif self.args.draft_path:
|
||
# there is an existing draft that we use
|
||
content_file_path = self.args.draft_path.expanduser()
|
||
content_file_obj = content_file_path.open("r+b")
|
||
# we seek at the end for the same reason as above
|
||
content_file_obj.seek(0, os.SEEK_END)
|
||
else:
|
||
# we need a temporary file
|
||
content_file_obj, content_file_path = self.get_tmp_file()
|
||
|
||
if item or last_item:
|
||
self.disp("Editing requested published item", 2)
|
||
try:
|
||
if self.use_metadata:
|
||
content, metadata, item = await self.get_item_data(
|
||
service, node, item
|
||
)
|
||
else:
|
||
content, item = await self.get_item_data(service, node, item)
|
||
except Exception as e:
|
||
# FIXME: ugly but we have not good may to check errors in bridge
|
||
if "item-not-found" in str(e):
|
||
# item doesn't exist, we create a new one with requested id
|
||
metadata = None
|
||
if last_item:
|
||
self.disp(_("no item found at all, we create a new one"), 2)
|
||
else:
|
||
self.disp(
|
||
_(
|
||
'item "{item}" not found, we create a new item with'
|
||
"this id"
|
||
).format(item=item),
|
||
2,
|
||
)
|
||
content_file_obj.seek(0)
|
||
else:
|
||
self.disp(f"Error while retrieving item: {e}")
|
||
self.host.quit(C.EXIT_ERROR)
|
||
else:
|
||
# item exists, we write content
|
||
if content_file_obj.tell() != 0:
|
||
# we already have a draft,
|
||
# we copy item content after it and add an indicator
|
||
content_file_obj.write("\n*****\n")
|
||
content_file_obj.write(content.encode("utf-8"))
|
||
content_file_obj.seek(0)
|
||
self.disp(_('item "{item}" found, we edit it').format(item=item), 2)
|
||
else:
|
||
self.disp("Editing a new item", 2)
|
||
if self.use_metadata:
|
||
metadata = None
|
||
|
||
if self.use_metadata:
|
||
return service, node, item, content_file_path, content_file_obj, metadata
|
||
else:
|
||
return service, node, item, content_file_path, content_file_obj
|
||
|
||
|
||
class Table(object):
|
||
def __init__(self, host, data, headers=None, filters=None, use_buffer=False):
|
||
"""
|
||
@param data(iterable[list]): table data
|
||
all lines must have the same number of columns
|
||
@param headers(iterable[unicode], None): names/titles of the columns
|
||
if not None, must have same number of columns as data
|
||
@param filters(iterable[(callable, unicode)], None): values filters
|
||
the callable will get 2 arguments:
|
||
- current column value
|
||
- RowData with all columns values
|
||
if may also only use 1 argument, which will then be current col value.
|
||
the callable must return a string
|
||
if it's unicode, it will be used with .format and must countain u'{}' which
|
||
will be replaced with the string.
|
||
if not None, must have same number of columns as data
|
||
@param use_buffer(bool): if True, bufferise output instead of printing it directly
|
||
"""
|
||
self.host = host
|
||
self._buffer = [] if use_buffer else None
|
||
# headers are columns names/titles, can be None
|
||
self.headers = headers
|
||
# sizes fof columns without headers,
|
||
# headers may be larger
|
||
self.sizes = []
|
||
# rows countains one list per row with columns values
|
||
self.rows = []
|
||
|
||
size = None
|
||
if headers:
|
||
# we use a namedtuple to make the value easily accessible from filters
|
||
headers_safe = [re.sub(r"[^a-zA-Z_]", "_", h) for h in headers]
|
||
row_cls = namedtuple("RowData", headers_safe)
|
||
else:
|
||
row_cls = tuple
|
||
|
||
for row_data in data:
|
||
new_row = []
|
||
row_data_list = list(row_data)
|
||
for idx, value in enumerate(row_data_list):
|
||
if filters is not None and filters[idx] is not None:
|
||
filter_ = filters[idx]
|
||
if isinstance(filter_, str):
|
||
col_value = filter_.format(value)
|
||
else:
|
||
try:
|
||
col_value = filter_(value, row_cls(*row_data_list))
|
||
except TypeError:
|
||
col_value = filter_(value)
|
||
# we count size without ANSI code as they will change length of the
|
||
# string when it's mostly style/color changes.
|
||
col_size = len(regex.ansi_remove(col_value))
|
||
else:
|
||
col_value = str(value)
|
||
col_size = len(col_value)
|
||
new_row.append(col_value)
|
||
if size is None:
|
||
self.sizes.append(col_size)
|
||
else:
|
||
self.sizes[idx] = max(self.sizes[idx], col_size)
|
||
if size is None:
|
||
size = len(new_row)
|
||
if headers is not None and len(headers) != size:
|
||
raise exceptions.DataError("headers size is not coherent with rows")
|
||
else:
|
||
if len(new_row) != size:
|
||
raise exceptions.DataError("rows size is not coherent")
|
||
self.rows.append(new_row)
|
||
|
||
if not data and headers is not None:
|
||
# the table is empty, we print headers at their lenght
|
||
self.sizes = [len(h) for h in headers]
|
||
|
||
@property
|
||
def string(self):
|
||
if self._buffer is None:
|
||
raise exceptions.InternalError("buffer must be used to get a string")
|
||
return "\n".join(self._buffer)
|
||
|
||
@staticmethod
|
||
def read_dict_values(data, keys, defaults=None):
|
||
if defaults is None:
|
||
defaults = {}
|
||
for key in keys:
|
||
try:
|
||
yield data[key]
|
||
except KeyError as e:
|
||
default = defaults.get(key)
|
||
if default is not None:
|
||
yield default
|
||
else:
|
||
raise e
|
||
|
||
@classmethod
|
||
def from_list_dict(
|
||
cls, host, data, keys=None, headers=None, filters=None, defaults=None
|
||
):
|
||
"""Create a table from a list of dictionaries
|
||
|
||
each dictionary is a row of the table, keys being columns names.
|
||
the whole data will be read and kept into memory, to be printed
|
||
@param data(list[dict[unicode, unicode]]): data to create the table from
|
||
@param keys(iterable[unicode], None): keys to get
|
||
if None, all keys will be used
|
||
@param headers(iterable[unicode], None): name of the columns
|
||
names must be in same order as keys
|
||
@param filters(dict[unicode, (callable,unicode)), None): filter to use on values
|
||
keys correspond to keys to filter, and value is the same as for Table.__init__
|
||
@param defaults(dict[unicode, unicode]): default value to use
|
||
if None, an exception will be raised if not value is found
|
||
"""
|
||
if keys is None and headers is not None:
|
||
# FIXME: keys are not needed with OrderedDict,
|
||
raise exceptions.DataError("You must specify keys order to used headers")
|
||
if keys is None:
|
||
keys = list(data[0].keys())
|
||
if headers is None:
|
||
headers = keys
|
||
if filters is None:
|
||
filters = {}
|
||
filters = [filters.get(k) for k in keys]
|
||
return cls(
|
||
host,
|
||
(cls.read_dict_values(d, keys, defaults) for d in data),
|
||
headers,
|
||
filters,
|
||
)
|
||
|
||
def _headers(self, head_sep, headers, sizes, alignment="left", style=None):
|
||
"""Render headers
|
||
|
||
@param head_sep(unicode): sequence to use as separator
|
||
@param alignment(unicode): how to align, can be left, center or right
|
||
@param style(unicode, iterable[unicode], None): ANSI escape sequences to apply
|
||
@param headers(list[unicode]): headers to show
|
||
@param sizes(list[int]): sizes of columns
|
||
"""
|
||
rendered_headers = []
|
||
if isinstance(style, str):
|
||
style = [style]
|
||
for idx, header in enumerate(headers):
|
||
size = sizes[idx]
|
||
if alignment == "left":
|
||
rendered = header[:size].ljust(size)
|
||
elif alignment == "center":
|
||
rendered = header[:size].center(size)
|
||
elif alignment == "right":
|
||
rendered = header[:size].rjust(size)
|
||
else:
|
||
raise exceptions.InternalError("bad alignment argument")
|
||
if style:
|
||
args = style + [rendered]
|
||
rendered = A.color(*args)
|
||
rendered_headers.append(rendered)
|
||
return head_sep.join(rendered_headers)
|
||
|
||
def _disp(self, data):
|
||
"""output data (can be either bufferised or printed)"""
|
||
if self._buffer is not None:
|
||
self._buffer.append(data)
|
||
else:
|
||
self.host.disp(data)
|
||
|
||
def display(
|
||
self,
|
||
head_alignment="left",
|
||
columns_alignment="left",
|
||
head_style=None,
|
||
show_header=True,
|
||
show_borders=True,
|
||
hide_cols=None,
|
||
col_sep=" │ ",
|
||
top_left="┌",
|
||
top="─",
|
||
top_sep="─┬─",
|
||
top_right="┐",
|
||
left="│",
|
||
right=None,
|
||
head_sep=None,
|
||
head_line="┄",
|
||
head_line_left="├",
|
||
head_line_sep="┄┼┄",
|
||
head_line_right="┤",
|
||
bottom_left="└",
|
||
bottom=None,
|
||
bottom_sep="─┴─",
|
||
bottom_right="┘",
|
||
):
|
||
"""Print the table
|
||
|
||
@param show_header(bool): True if header need no be shown
|
||
@param show_borders(bool): True if borders need no be shown
|
||
@param hide_cols(None, iterable(unicode)): columns which should not be displayed
|
||
@param head_alignment(unicode): how to align headers, can be left, center or right
|
||
@param columns_alignment(unicode): how to align columns, can be left, center or
|
||
right
|
||
@param col_sep(unicode): separator betweens columns
|
||
@param head_line(unicode): character to use to make line under head
|
||
@param disp(callable, None): method to use to display the table
|
||
None to use self.host.disp
|
||
"""
|
||
if not self.sizes:
|
||
# the table is empty
|
||
return
|
||
col_sep_size = len(regex.ansi_remove(col_sep))
|
||
|
||
# if we have columns to hide, we remove them from headers and size
|
||
if not hide_cols:
|
||
headers = self.headers
|
||
sizes = self.sizes
|
||
else:
|
||
headers = list(self.headers)
|
||
sizes = self.sizes[:]
|
||
ignore_idx = [headers.index(to_hide) for to_hide in hide_cols]
|
||
for to_hide in hide_cols:
|
||
hide_idx = headers.index(to_hide)
|
||
del headers[hide_idx]
|
||
del sizes[hide_idx]
|
||
|
||
if right is None:
|
||
right = left
|
||
if top_sep is None:
|
||
top_sep = col_sep_size * top
|
||
if head_sep is None:
|
||
head_sep = col_sep
|
||
if bottom is None:
|
||
bottom = top
|
||
if bottom_sep is None:
|
||
bottom_sep = col_sep_size * bottom
|
||
if not show_borders:
|
||
left = right = head_line_left = head_line_right = ""
|
||
# top border
|
||
if show_borders:
|
||
self._disp(
|
||
top_left + top_sep.join([top * size for size in sizes]) + top_right
|
||
)
|
||
|
||
# headers
|
||
if show_header and self.headers is not None:
|
||
self._disp(
|
||
left
|
||
+ self._headers(head_sep, headers, sizes, head_alignment, head_style)
|
||
+ right
|
||
)
|
||
# header line
|
||
self._disp(
|
||
head_line_left
|
||
+ head_line_sep.join([head_line * size for size in sizes])
|
||
+ head_line_right
|
||
)
|
||
|
||
# content
|
||
if columns_alignment == "left":
|
||
alignment = lambda idx, s: ansi_ljust(s, sizes[idx])
|
||
elif columns_alignment == "center":
|
||
alignment = lambda idx, s: ansi_center(s, sizes[idx])
|
||
elif columns_alignment == "right":
|
||
alignment = lambda idx, s: ansi_rjust(s, sizes[idx])
|
||
else:
|
||
raise exceptions.InternalError("bad columns alignment argument")
|
||
|
||
for row in self.rows:
|
||
if hide_cols:
|
||
row = [v for idx, v in enumerate(row) if idx not in ignore_idx]
|
||
self._disp(
|
||
left
|
||
+ col_sep.join([alignment(idx, c) for idx, c in enumerate(row)])
|
||
+ right
|
||
)
|
||
|
||
if show_borders:
|
||
# bottom border
|
||
self._disp(
|
||
bottom_left
|
||
+ bottom_sep.join([bottom * size for size in sizes])
|
||
+ bottom_right
|
||
)
|
||
# we return self so string can be used after display (table.display().string)
|
||
return self
|
||
|
||
def display_blank(self, **kwargs):
|
||
"""Display table without visible borders"""
|
||
kwargs_ = {"col_sep": " ", "head_line_sep": " ", "show_borders": False}
|
||
kwargs_.update(kwargs)
|
||
return self.display(**kwargs_)
|
||
|
||
|
||
async def fill_well_known_uri(command, path, key, meta_map=None):
|
||
"""Look for URIs in well-known location and fill appropriate args if suitable
|
||
|
||
@param command(CommandBase): command instance
|
||
args of this instance will be updated with found values
|
||
@param path(unicode): absolute path to use as a starting point to look for URIs
|
||
@param key(unicode): key to look for
|
||
@param meta_map(dict, None): if not None, map metadata to arg name
|
||
key is metadata used attribute name
|
||
value is name to actually use, or None to ignore
|
||
use empty dict to only retrieve URI
|
||
possible keys are currently:
|
||
- labels
|
||
"""
|
||
args = command.args
|
||
if args.service or args.node:
|
||
# we only look for URIs if a service and a node are not already specified
|
||
return
|
||
|
||
host = command.host
|
||
|
||
try:
|
||
uris_data = await host.bridge.uri_find(path, [key])
|
||
except Exception as e:
|
||
host.disp(f"can't find {key} URI: {e}", error=True)
|
||
host.quit(C.EXIT_BRIDGE_ERRBACK)
|
||
|
||
try:
|
||
uri_data = uris_data[key]
|
||
except KeyError:
|
||
host.disp(
|
||
_(
|
||
"No {key} URI specified for this project, please specify service and "
|
||
"node"
|
||
).format(key=key),
|
||
error=True,
|
||
)
|
||
host.quit(C.EXIT_NOT_FOUND)
|
||
|
||
uri = uri_data["uri"]
|
||
|
||
# set extra metadata if they are specified
|
||
for data_key in ["labels"]:
|
||
new_values_json = uri_data.get(data_key)
|
||
if uri_data is not None:
|
||
if meta_map is None:
|
||
dest = data_key
|
||
else:
|
||
dest = meta_map.get(data_key)
|
||
if dest is None:
|
||
continue
|
||
|
||
try:
|
||
values = getattr(args, data_key)
|
||
except AttributeError:
|
||
raise exceptions.InternalError(f"there is no {data_key!r} arguments")
|
||
else:
|
||
if values is None:
|
||
values = []
|
||
values.extend(json.loads(new_values_json))
|
||
setattr(args, dest, values)
|
||
|
||
parsed_uri = xmpp_uri.parse_xmpp_uri(uri)
|
||
try:
|
||
args.service = parsed_uri["path"]
|
||
args.node = parsed_uri["node"]
|
||
except KeyError:
|
||
host.disp(_("Invalid URI found: {uri}").format(uri=uri), error=True)
|
||
host.quit(C.EXIT_DATA_ERROR)
|