view libervia/cli/cmd_pubsub.py @ 4180:b86912d3fd33

plugin IP: fix use of legacy URL + coroutine use: An https:/salut-a-toi.org URL was used to retrieve external IP, but it's not valid anymore, resulting in an exception. This feature is currently disabled. Also moved several methods from legacy inline callbacks to coroutines.
author Goffi <goffi@goffi.org>
date Sat, 09 Dec 2023 14:30:54 +0100
parents 47401850dec6
children 3f7ca590a5da
line wrap: on
line source

#!/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 argparse
import os.path
import re
import sys
import subprocess
import asyncio
import json
from . import base
from libervia.backend.core.i18n import _
from libervia.backend.core import exceptions
from libervia.cli.constants import Const as C
from libervia.cli import common
from libervia.cli import arg_tools
from libervia.cli import xml_tools
from functools import partial
from libervia.backend.tools.common import data_format
from libervia.backend.tools.common import uri
from libervia.backend.tools.common.ansi import ANSI as A
from libervia.backend.tools.common import date_utils
from libervia.frontends.tools import jid, strings
from libervia.frontends.bridge.bridge_frontend import BridgeException

__commands__ = ["Pubsub"]

PUBSUB_TMP_DIR = "pubsub"
PUBSUB_SCHEMA_TMP_DIR = PUBSUB_TMP_DIR + "_schema"
ALLOWED_SUBSCRIPTIONS_OWNER = ("subscribed", "pending", "none")

# TODO: need to split this class in several modules, plugin should handle subcommands


class NodeInfo(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "info",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("retrieve node configuration"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-k",
            "--key",
            action="append",
            dest="keys",
            help=_("data key to filter"),
        )

    def remove_prefix(self, key):
        return key[7:] if key.startswith("pubsub#") else key

    def filter_key(self, key):
        return any((key == k or key == "pubsub#" + k) for k in self.args.keys)

    async def start(self):
        try:
            config_dict = await self.host.bridge.ps_node_configuration_get(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except BridgeException as e:
            if e.condition == "item-not-found":
                self.disp(
                    f"The node {self.args.node} doesn't exist on {self.args.service}",
                    error=True,
                )
                self.host.quit(C.EXIT_NOT_FOUND)
            else:
                self.disp(f"can't get node configuration: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            key_filter = (lambda k: True) if not self.args.keys else self.filter_key
            config_dict = {
                self.remove_prefix(k): v for k, v in config_dict.items() if key_filter(k)
            }
            await self.output(config_dict)
            self.host.quit()


class NodeCreate(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "create",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("create a node"),
        )

    @staticmethod
    def add_node_config_options(parser):
        parser.add_argument(
            "-f",
            "--field",
            action="append",
            nargs=2,
            dest="fields",
            default=[],
            metavar=("KEY", "VALUE"),
            help=_("configuration field to set"),
        )
        parser.add_argument(
            "-F",
            "--full-prefix",
            action="store_true",
            help=_('don\'t prepend "pubsub#" prefix to field names'),
        )

    def add_parser_options(self):
        self.add_node_config_options(self.parser)

    @staticmethod
    def get_config_options(args):
        if not args.full_prefix:
            return {"pubsub#" + k: v for k, v in args.fields}
        else:
            return dict(args.fields)

    async def start(self):
        options = self.get_config_options(self.args)
        try:
            node_id = await self.host.bridge.ps_node_create(
                self.args.service,
                self.args.node,
                options,
                self.profile,
            )
        except Exception as e:
            self.disp(msg=_("can't create node: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            if self.host.verbosity:
                announce = _("node created successfully: ")
            else:
                announce = ""
            self.disp(announce + node_id)
            self.host.quit()


class NodePurge(base.CommandBase):
    def __init__(self, host):
        super(NodePurge, self).__init__(
            host,
            "purge",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("purge a node (i.e. remove all items from it)"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-f",
            "--force",
            action="store_true",
            help=_("purge node without confirmation"),
        )

    async def start(self):
        if not self.args.force:
            if not self.args.service:
                message = _(
                    "Are you sure to purge PEP node [{node}]? This will "
                    "delete ALL items from it!"
                ).format(node=self.args.node)
            else:
                message = _(
                    "Are you sure to delete node [{node}] on service "
                    "[{service}]? This will delete ALL items from it!"
                ).format(node=self.args.node, service=self.args.service)
            await self.host.confirm_or_quit(message, _("node purge cancelled"))

        try:
            await self.host.bridge.ps_node_purge(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except Exception as e:
            self.disp(msg=_("can't purge node: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("node [{node}] purged successfully").format(node=self.args.node))
            self.host.quit()


class NodeDelete(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "delete",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("delete a node"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-f",
            "--force",
            action="store_true",
            help=_("delete node without confirmation"),
        )

    async def start(self):
        if not self.args.force:
            if not self.args.service:
                message = _("Are you sure to delete PEP node [{node}] ?").format(
                    node=self.args.node
                )
            else:
                message = _(
                    "Are you sure to delete node [{node}] on " "service [{service}]?"
                ).format(node=self.args.node, service=self.args.service)
            await self.host.confirm_or_quit(message, _("node deletion cancelled"))

        try:
            await self.host.bridge.ps_node_delete(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't delete node: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("node [{node}] deleted successfully").format(node=self.args.node))
            self.host.quit()


class NodeSet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "set",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("set node configuration"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-f",
            "--field",
            action="append",
            nargs=2,
            dest="fields",
            required=True,
            metavar=("KEY", "VALUE"),
            help=_("configuration field to set (required)"),
        )
        self.parser.add_argument(
            "-F",
            "--full-prefix",
            action="store_true",
            help=_('don\'t prepend "pubsub#" prefix to field names'),
        )

    def get_key_name(self, k):
        if self.args.full_prefix or k.startswith("pubsub#"):
            return k
        else:
            return "pubsub#" + k

    async def start(self):
        try:
            await self.host.bridge.ps_node_configuration_set(
                self.args.service,
                self.args.node,
                {self.get_key_name(k): v for k, v in self.args.fields},
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't set node configuration: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("node configuration successful"), 1)
            self.host.quit()


class NodeImport(base.CommandBase):
    def __init__(self, host):
        super(NodeImport, self).__init__(
            host,
            "import",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("import raw XML to a node"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "--admin",
            action="store_true",
            help=_("do a pubsub admin request, needed to change publisher"),
        )
        self.parser.add_argument(
            "import_file",
            type=argparse.FileType(),
            help=_(
                "path to the XML file with data to import. The file must contain "
                "whole XML of each item to import."
            ),
        )

    async def start(self):
        try:
            element, etree = xml_tools.etree_parse(
                self, self.args.import_file, reraise=True
            )
        except Exception as e:
            from lxml.etree import XMLSyntaxError

            if isinstance(e, XMLSyntaxError) and e.code == 5:
                # we have extra content, this probaby means that item are not wrapped
                # so we wrap them here and try again
                self.args.import_file.seek(0)
                xml_buf = "<import>" + self.args.import_file.read() + "</import>"
                element, etree = xml_tools.etree_parse(self, xml_buf)

                # we reverse element as we expect to have most recently published element first
                # TODO: make this more explicit and add an option
        element[:] = reversed(element)

        if not all([i.tag == "{http://jabber.org/protocol/pubsub}item" for i in element]):
            self.disp(
                _("You are not using list of pubsub items, we can't import this file"),
                error=True,
            )
            self.host.quit(C.EXIT_DATA_ERROR)
            return

        items = [etree.tostring(i, encoding="unicode") for i in element]
        if self.args.admin:
            method = self.host.bridge.ps_admin_items_send
        else:
            self.disp(
                _(
                    "Items are imported without using admin mode, publisher can't "
                    "be changed"
                )
            )
            method = self.host.bridge.ps_items_send

        try:
            items_ids = await method(
                self.args.service,
                self.args.node,
                items,
                "",
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't send items: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            if items_ids:
                self.disp(
                    _("items published with id(s) {items_ids}").format(
                        items_ids=", ".join(items_ids)
                    )
                )
            else:
                self.disp(_("items published"))
            self.host.quit()


class NodeAffiliationsGet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "get",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("retrieve node affiliations (for node owner)"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            affiliations = await self.host.bridge.ps_node_affiliations_get(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't get node affiliations: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(affiliations)
            self.host.quit()


class NodeAffiliationsSet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "set",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("set affiliations (for node owner)"),
        )

    def add_parser_options(self):
        # XXX: we use optional argument syntax for a required one because list of list of 2 elements
        #      (used to construct dicts) don't work with positional arguments
        self.parser.add_argument(
            "-a",
            "--affiliation",
            dest="affiliations",
            metavar=("JID", "AFFILIATION"),
            required=True,
            action="append",
            nargs=2,
            help=_("entity/affiliation couple(s)"),
        )

    async def start(self):
        affiliations = dict(self.args.affiliations)
        try:
            await self.host.bridge.ps_node_affiliations_set(
                self.args.service,
                self.args.node,
                affiliations,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't set node affiliations: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("affiliations have been set"), 1)
            self.host.quit()


class NodeAffiliations(base.CommandBase):
    subcommands = (NodeAffiliationsGet, NodeAffiliationsSet)

    def __init__(self, host):
        super(NodeAffiliations, self).__init__(
            host,
            "affiliations",
            use_profile=False,
            help=_("set or retrieve node affiliations"),
        )


class NodeSubscriptionsGet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "get",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("retrieve node subscriptions (for node owner)"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "--public",
            action="store_true",
            help=_("get public subscriptions"),
        )

    async def start(self):
        if self.args.public:
            method = self.host.bridge.ps_public_node_subscriptions_get
        else:
            method = self.host.bridge.ps_node_subscriptions_get
        try:
            subscriptions = await method(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't get node subscriptions: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(subscriptions)
            self.host.quit()


class StoreSubscriptionAction(argparse.Action):
    """Action which handle subscription parameter for owner

    list is given by pairs: jid and subscription state
    if subscription state is not specified, it default to "subscribed"
    """

    def __call__(self, parser, namespace, values, option_string):
        dest_dict = getattr(namespace, self.dest)
        while values:
            jid_s = values.pop(0)
            try:
                subscription = values.pop(0)
            except IndexError:
                subscription = "subscribed"
            if subscription not in ALLOWED_SUBSCRIPTIONS_OWNER:
                parser.error(
                    _("subscription must be one of {}").format(
                        ", ".join(ALLOWED_SUBSCRIPTIONS_OWNER)
                    )
                )
            dest_dict[jid_s] = subscription


class NodeSubscriptionsSet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "set",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("set/modify subscriptions (for node owner)"),
        )

    def add_parser_options(self):
        # XXX: we use optional argument syntax for a required one because list of list of 2 elements
        #      (uses to construct dicts) don't work with positional arguments
        self.parser.add_argument(
            "-S",
            "--subscription",
            dest="subscriptions",
            default={},
            nargs="+",
            metavar=("JID [SUSBSCRIPTION]"),
            required=True,
            action=StoreSubscriptionAction,
            help=_("entity/subscription couple(s)"),
        )

    async def start(self):
        try:
            self.host.bridge.ps_node_subscriptions_set(
                self.args.service,
                self.args.node,
                self.args.subscriptions,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't set node subscriptions: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("subscriptions have been set"), 1)
            self.host.quit()


class NodeSubscriptions(base.CommandBase):
    subcommands = (NodeSubscriptionsGet, NodeSubscriptionsSet)

    def __init__(self, host):
        super(NodeSubscriptions, self).__init__(
            host,
            "subscriptions",
            use_profile=False,
            help=_("get or modify node subscriptions"),
        )


class NodeSchemaSet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "set",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("set/replace a schema"),
        )

    def add_parser_options(self):
        self.parser.add_argument("schema", help=_("schema to set (must be XML)"))

    async def start(self):
        try:
            await self.host.bridge.ps_schema_set(
                self.args.service,
                self.args.node,
                self.args.schema,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't set schema: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("schema has been set"), 1)
            self.host.quit()


class NodeSchemaEdit(base.CommandBase, common.BaseEdit):
    use_items = False

    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "edit",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_draft=True,
            use_verbose=True,
            help=_("edit a schema"),
        )
        common.BaseEdit.__init__(self, self.host, PUBSUB_SCHEMA_TMP_DIR)

    def add_parser_options(self):
        pass

    async def publish(self, schema):
        try:
            await self.host.bridge.ps_schema_set(
                self.args.service,
                self.args.node,
                schema,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't set schema: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("schema has been set"), 1)
            self.host.quit()

    async def ps_schema_get_cb(self, schema):
        try:
            from lxml import etree
        except ImportError:
            self.disp(
                "lxml module must be installed to use edit, please install it "
                'with "pip install lxml"',
                error=True,
            )
            self.host.quit(1)
        content_file_obj, content_file_path = self.get_tmp_file()
        schema = schema.strip()
        if schema:
            parser = etree.XMLParser(remove_blank_text=True)
            schema_elt = etree.fromstring(schema, parser)
            content_file_obj.write(
                etree.tostring(schema_elt, encoding="utf-8", pretty_print=True)
            )
            content_file_obj.seek(0)
        await self.run_editor(
            "pubsub_schema_editor_args", content_file_path, content_file_obj
        )

    async def start(self):
        try:
            schema = await self.host.bridge.ps_schema_get(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except BridgeException as e:
            if e.condition == "item-not-found" or e.classname == "NotFound":
                schema = ""
            else:
                self.disp(f"can't edit schema: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)

        await self.ps_schema_get_cb(schema)


class NodeSchemaGet(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "get",
            use_output=C.OUTPUT_XML,
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("get schema"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            schema = await self.host.bridge.ps_schema_get(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except BridgeException as e:
            if e.condition == "item-not-found" or e.classname == "NotFound":
                schema = None
            else:
                self.disp(f"can't get schema: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)

        if schema:
            await self.output(schema)
            self.host.quit()
        else:
            self.disp(_("no schema found"), 1)
            self.host.quit(C.EXIT_NOT_FOUND)


class NodeSchema(base.CommandBase):
    subcommands = (NodeSchemaSet, NodeSchemaEdit, NodeSchemaGet)

    def __init__(self, host):
        super(NodeSchema, self).__init__(
            host, "schema", use_profile=False, help=_("data schema manipulation")
        )


class Node(base.CommandBase):
    subcommands = (
        NodeInfo,
        NodeCreate,
        NodePurge,
        NodeDelete,
        NodeSet,
        NodeImport,
        NodeAffiliations,
        NodeSubscriptions,
        NodeSchema,
    )

    def __init__(self, host):
        super(Node, self).__init__(
            host, "node", use_profile=False, help=_("node handling")
        )


class CacheGet(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "get",
            use_output=C.OUTPUT_LIST_XML,
            use_pubsub=True,
            pubsub_flags={C.NODE, C.MULTI_ITEMS, C.CACHE},
            help=_("get pubsub item(s) from cache"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-S",
            "--sub-id",
            default="",
            help=_("subscription id"),
        )

    async def start(self):
        try:
            ps_result = data_format.deserialise(
                await self.host.bridge.ps_cache_get(
                    self.args.service,
                    self.args.node,
                    self.args.max,
                    self.args.items,
                    self.args.sub_id,
                    self.get_pubsub_extra(),
                    self.profile,
                )
            )
        except BridgeException as e:
            if e.classname == "NotFound":
                self.disp(
                    f"The node {self.args.node} from {self.args.service} is not in cache "
                    f"for {self.profile}",
                    error=True,
                )
                self.host.quit(C.EXIT_NOT_FOUND)
            else:
                self.disp(f"can't get pubsub items from cache: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            await self.output(ps_result["items"])
            self.host.quit(C.EXIT_OK)


class CacheSync(base.CommandBase):

    def __init__(self, host):
        super().__init__(
            host,
            "sync",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("(re)synchronise a pubsub node"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            await self.host.bridge.ps_cache_sync(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except BridgeException as e:
            if e.condition == "item-not-found" or e.classname == "NotFound":
                self.disp(
                    f"The node {self.args.node} doesn't exist on {self.args.service}",
                    error=True,
                )
                self.host.quit(C.EXIT_NOT_FOUND)
            else:
                self.disp(f"can't synchronise pubsub node: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            self.host.quit(C.EXIT_OK)


class CachePurge(base.CommandBase):

    def __init__(self, host):
        super().__init__(
            host,
            "purge",
            use_profile=False,
            help=_("purge (delete) items from cache"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-s", "--service", action="append", metavar="JID", dest="services",
            help="purge items only for these services. If not specified, items from ALL "
            "services will be purged. May be used several times."
        )
        self.parser.add_argument(
            "-n", "--node", action="append", dest="nodes",
            help="purge items only for these nodes. If not specified, items from ALL "
            "nodes will be purged. May be used several times."
        )
        self.parser.add_argument(
            "-p", "--profile", action="append", dest="profiles",
            help="purge items only for these profiles. If not specified, items from ALL "
            "profiles will be purged. May be used several times."
        )
        self.parser.add_argument(
            "-b", "--updated-before", type=base.date_decoder, metavar="TIME_PATTERN",
            help="purge items which have been last updated before given time."
        )
        self.parser.add_argument(
            "-C", "--created-before", type=base.date_decoder, metavar="TIME_PATTERN",
            help="purge items which have been last created before given time."
        )
        self.parser.add_argument(
            "-t", "--type", action="append", dest="types",
            help="purge items flagged with TYPE. May be used several times."
        )
        self.parser.add_argument(
            "-S", "--subtype", action="append", dest="subtypes",
            help="purge items flagged with SUBTYPE. May be used several times."
        )
        self.parser.add_argument(
            "-f", "--force", action="store_true",
            help=_("purge items without confirmation")
        )

    async def start(self):
        if not self.args.force:
            await self.host.confirm_or_quit(
                _(
                    "Are you sure to purge items from cache? You'll have to bypass cache "
                    "or resynchronise nodes to access deleted items again."
                ),
                _("Items purgins has been cancelled.")
            )
        purge_data = {}
        for key in (
                "services", "nodes", "profiles", "updated_before", "created_before",
                "types", "subtypes"
        ):
            value = getattr(self.args, key)
            if value is not None:
                purge_data[key] = value
        try:
            await self.host.bridge.ps_cache_purge(
                data_format.serialise(
                    purge_data
                )
            )
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            self.host.quit(C.EXIT_OK)


class CacheReset(base.CommandBase):

    def __init__(self, host):
        super().__init__(
            host,
            "reset",
            use_profile=False,
            help=_("remove everything from cache"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-f", "--force", action="store_true",
            help=_("reset cache without confirmation")
        )

    async def start(self):
        if not self.args.force:
            await self.host.confirm_or_quit(
                _(
                    "Are you sure to reset cache? All nodes and items will be removed "
                    "from it, then it will be progressively refilled as if it were new. "
                    "This may be resources intensive."
                ),
                _("Pubsub cache reset has been cancelled.")
            )
        try:
            await self.host.bridge.ps_cache_reset()
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            self.host.quit(C.EXIT_OK)


class CacheSearch(base.CommandBase):
    def __init__(self, host):
        extra_outputs = {
            "default": self.default_output,
            "xml": self.xml_output,
            "xml-raw": self.xml_raw_output,
        }
        super().__init__(
            host,
            "search",
            use_profile=False,
            use_output=C.OUTPUT_LIST_DICT,
            extra_outputs=extra_outputs,
            help=_("search for pubsub items in cache"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-f", "--fts", help=_("Full-Text Search query"), metavar="FTS_QUERY"
        )
        self.parser.add_argument(
            "-p", "--profile", action="append", dest="profiles", metavar="PROFILE",
            help="search items only from these profiles. May be used several times."
        )
        self.parser.add_argument(
            "-s", "--service", action="append", dest="services", metavar="SERVICE",
            help="items must be from specified service. May be used several times."
        )
        self.parser.add_argument(
            "-n", "--node", action="append", dest="nodes", metavar="NODE",
            help="items must be in the specified node. May be used several times."
        )
        self.parser.add_argument(
            "-t", "--type", action="append", dest="types", metavar="TYPE",
            help="items must be of specified type. May be used several times."
        )
        self.parser.add_argument(
            "-S", "--subtype", action="append", dest="subtypes", metavar="SUBTYPE",
            help="items must be of specified subtype. May be used several times."
        )
        self.parser.add_argument(
            "-P", "--payload", action="store_true", help=_("include item XML payload")
        )
        self.parser.add_argument(
            "-o", "--order-by", action="append", nargs="+",
            metavar=("ORDER", "[FIELD] [DIRECTION]"),
            help=_("how items must be ordered. May be used several times.")
        )
        self.parser.add_argument(
            "-l", "--limit", type=int, help=_("maximum number of items to return")
        )
        self.parser.add_argument(
            "-i", "--index", type=int, help=_("return results starting from this index")
        )
        self.parser.add_argument(
            "-F",
            "--field",
            action="append",
            nargs=3,
            dest="fields",
            default=[],
            metavar=("PATH", "OPERATOR", "VALUE"),
            help=_("parsed data field filter. May be used several times."),
        )
        self.parser.add_argument(
            "-k",
            "--key",
            action="append",
            dest="keys",
            metavar="KEY",
            help=_(
                "data key(s) to display. May be used several times. DEFAULT: show all "
                "keys"
            ),
        )

    async def start(self):
        query = {}
        for arg in ("fts", "profiles", "services", "nodes", "types", "subtypes"):
            value = getattr(self.args, arg)
            if value:
                if arg in ("types", "subtypes"):
                    # empty string is used to find items without type and/or subtype
                    value = [v or None for v in value]
                query[arg] = value
        for arg in ("limit", "index"):
            value = getattr(self.args, arg)
            if value is not None:
                query[arg] = value
        if self.args.order_by is not None:
            for order_data in self.args.order_by:
                order, *args = order_data
                if order == "field":
                    if not args:
                        self.parser.error(_("field data must be specified in --order-by"))
                    elif len(args) == 1:
                        path = args[0]
                        direction = "asc"
                    elif len(args) == 2:
                        path, direction = args
                    else:
                        self.parser.error(_(
                            "You can't specify more that 2 arguments for a field in "
                            "--order-by"
                        ))
                    try:
                        path = json.loads(path)
                    except json.JSONDecodeError:
                        pass
                    order_query = {
                        "path": path,
                    }
                else:
                    order_query = {
                        "order": order
                    }
                    if not args:
                        direction = "asc"
                    elif len(args) == 1:
                        direction = args[0]
                    else:
                        self.parser.error(_(
                            "there are too many arguments in --order-by option"
                        ))
                if direction.lower() not in ("asc", "desc"):
                    self.parser.error(_("invalid --order-by direction: {direction!r}"))
                order_query["direction"] = direction
                query.setdefault("order-by", []).append(order_query)

        if self.args.fields:
            parsed = []
            for field in self.args.fields:
                path, operator, value = field
                try:
                    path = json.loads(path)
                except json.JSONDecodeError:
                    # this is not a JSON encoded value, we keep it as a string
                    pass

                if not isinstance(path, list):
                    path = [path]

                # handling of TP(<time pattern>)
                if operator in (">", "gt", "<", "le", "between"):
                    def datetime_sub(match):
                        return str(date_utils.date_parse_ext(
                            match.group(1), default_tz=date_utils.TZ_LOCAL
                        ))
                    value = re.sub(r"\bTP\(([^)]+)\)", datetime_sub, value)

                try:
                    value = json.loads(value)
                except json.JSONDecodeError:
                    # not JSON, as above we keep it as string
                    pass

                if operator in ("overlap", "ioverlap", "disjoint", "idisjoint"):
                    if not isinstance(value, list):
                        value = [value]

                parsed.append({
                    "path": path,
                    "op": operator,
                    "value": value
                })

            query["parsed"] = parsed

        if self.args.payload or "xml" in self.args.output:
            query["with_payload"] = True
            if self.args.keys:
                self.args.keys.append("item_payload")
        try:
            found_items = data_format.deserialise(
                await self.host.bridge.ps_cache_search(
                    data_format.serialise(query)
                ),
                type_check=list,
            )
        except BridgeException as e:
            self.disp(f"can't search for pubsub items in cache: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            if self.args.keys:
                found_items = [
                    {k: v for k,v in item.items() if k in self.args.keys}
                    for item in found_items
                ]
            await self.output(found_items)
            self.host.quit(C.EXIT_OK)

    def default_output(self, found_items):
        for item in found_items:
            for field in ("created", "published", "updated"):
                try:
                    timestamp = item[field]
                except KeyError:
                    pass
                else:
                    try:
                        item[field] = common.format_time(timestamp)
                    except ValueError:
                        pass
        self.host._outputs[C.OUTPUT_LIST_DICT]["simple"]["callback"](found_items)

    def xml_output(self, found_items):
        """Output prettified item payload"""
        cb = self.host._outputs[C.OUTPUT_XML][C.OUTPUT_NAME_XML]["callback"]
        for item in found_items:
            cb(item["item_payload"])

    def xml_raw_output(self, found_items):
        """Output item payload without prettifying"""
        cb = self.host._outputs[C.OUTPUT_XML][C.OUTPUT_NAME_XML_RAW]["callback"]
        for item in found_items:
            cb(item["item_payload"])


class Cache(base.CommandBase):
    subcommands = (
        CacheGet,
        CacheSync,
        CachePurge,
        CacheReset,
        CacheSearch,
    )

    def __init__(self, host):
        super(Cache, self).__init__(
            host, "cache", use_profile=False, help=_("pubsub cache handling")
        )


class Set(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "set",
            use_pubsub=True,
            use_quiet=True,
            pubsub_flags={C.NODE},
            help=_("publish a new item or update an existing one"),
        )

    def add_parser_options(self):
        NodeCreate.add_node_config_options(self.parser)
        self.parser.add_argument(
            "-e",
            "--encrypt",
            action="store_true",
            help=_("end-to-end encrypt the blog item")
        )
        self.parser.add_argument(
            "--encrypt-for",
            metavar="JID",
            action="append",
            help=_("encrypt a single item for")
        )
        self.parser.add_argument(
            "-X",
            "--sign",
            action="store_true",
            help=_("cryptographically sign the blog post")
        )
        self.parser.add_argument(
            "item",
            nargs="?",
            default="",
            help=_("id, URL of the item to update, keyword, or nothing for new item"),
        )

    async def start(self):
        element, etree = xml_tools.etree_parse(self, sys.stdin)
        element = xml_tools.get_payload(self, element)
        payload = etree.tostring(element, encoding="unicode")
        extra = {}
        if self.args.encrypt:
            extra["encrypted"] = True
        if self.args.encrypt_for:
            extra["encrypted_for"] = {"targets": self.args.encrypt_for}
        if self.args.sign:
            extra["signed"] = True
        publish_options = NodeCreate.get_config_options(self.args)
        if publish_options:
            extra["publish_options"] = publish_options

        try:
            published_id = await self.host.bridge.ps_item_send(
                self.args.service,
                self.args.node,
                payload,
                self.args.item,
                data_format.serialise(extra),
                self.profile,
            )
        except Exception as e:
            self.disp(_("can't send item: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            if published_id:
                if self.args.quiet:
                    self.disp(published_id, end="")
                else:
                    self.disp(f"Item published at {published_id}")
            else:
                self.disp("Item published")
            self.host.quit(C.EXIT_OK)


class Get(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "get",
            use_output=C.OUTPUT_LIST_XML,
            use_pubsub=True,
            pubsub_flags={C.NODE, C.MULTI_ITEMS, C.CACHE},
            help=_("get pubsub item(s)"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-S",
            "--sub-id",
            default="",
            help=_("subscription id"),
        )
        self.parser.add_argument(
            "--no-decrypt",
            action="store_true",
            help=_("don't do automatic decryption of e2ee items"),
        )
        #  TODO: a key(s) argument to select keys to display

    async def start(self):
        extra = {}
        if self.args.no_decrypt:
            extra["decrypt"] = False
        try:
            ps_result = data_format.deserialise(
                await self.host.bridge.ps_items_get(
                    self.args.service,
                    self.args.node,
                    self.args.max,
                    self.args.items,
                    self.args.sub_id,
                    self.get_pubsub_extra(extra),
                    self.profile,
                )
            )
        except BridgeException as e:
            if e.condition == "item-not-found" or e.classname == "NotFound":
                self.disp(
                    f"The node {self.args.node} doesn't exist on {self.args.service}",
                    error=True,
                )
                self.host.quit(C.EXIT_NOT_FOUND)
            else:
                self.disp(f"can't get pubsub items: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        except Exception as e:
            self.disp(f"Internal error: {e}", error=True)
            self.host.quit(C.EXIT_INTERNAL_ERROR)
        else:
            await self.output(ps_result["items"])
            self.host.quit(C.EXIT_OK)


class Delete(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "delete",
            use_pubsub=True,
            pubsub_flags={C.NODE, C.ITEM, C.SINGLE_ITEM},
            help=_("delete an item"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-f", "--force", action="store_true", help=_("delete without confirmation")
        )
        self.parser.add_argument(
            "--no-notification", dest="notify", action="store_false",
            help=_("do not send notification (not recommended)")
        )

    async def start(self):
        if not self.args.item:
            self.parser.error(_("You need to specify an item to delete"))
        if not self.args.force:
            message = _("Are you sure to delete item {item_id} ?").format(
                item_id=self.args.item
            )
            await self.host.confirm_or_quit(message, _("item deletion cancelled"))
        try:
            await self.host.bridge.ps_item_retract(
                self.args.service,
                self.args.node,
                self.args.item,
                self.args.notify,
                self.profile,
            )
        except Exception as e:
            self.disp(_("can't delete item: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("item {item} has been deleted").format(item=self.args.item))
            self.host.quit(C.EXIT_OK)


class Edit(base.CommandBase, common.BaseEdit):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "edit",
            use_verbose=True,
            use_pubsub=True,
            pubsub_flags={C.NODE, C.SINGLE_ITEM},
            use_draft=True,
            help=_("edit an existing or new pubsub item"),
        )
        common.BaseEdit.__init__(self, self.host, PUBSUB_TMP_DIR)

    def add_parser_options(self):
        self.parser.add_argument(
            "-e",
            "--encrypt",
            action="store_true",
            help=_("end-to-end encrypt the blog item")
        )
        self.parser.add_argument(
            "--encrypt-for",
            metavar="JID",
            action="append",
            help=_("encrypt a single item for")
        )
        self.parser.add_argument(
            "-X",
            "--sign",
            action="store_true",
            help=_("cryptographically sign the blog post")
        )

    async def publish(self, content):
        extra = {}
        if self.args.encrypt:
            extra["encrypted"] = True
        if self.args.encrypt_for:
            extra["encrypted_for"] = {"targets": self.args.encrypt_for}
        if self.args.sign:
            extra["signed"] = True
        published_id = await self.host.bridge.ps_item_send(
            self.pubsub_service,
            self.pubsub_node,
            content,
            self.pubsub_item or "",
            data_format.serialise(extra),
            self.profile,
        )
        if published_id:
            self.disp("Item published at {pub_id}".format(pub_id=published_id))
        else:
            self.disp("Item published")

    async def get_item_data(self, service, node, item):
        try:
            from lxml import etree
        except ImportError:
            self.disp(
                "lxml module must be installed to use edit, please install it "
                'with "pip install lxml"',
                error=True,
            )
            self.host.quit(1)
        items = [item] if item else []
        ps_result = data_format.deserialise(
            await self.host.bridge.ps_items_get(
                service, node, 1, items, "", data_format.serialise({}), self.profile
            )
        )
        item_raw = ps_result["items"][0]
        parser = etree.XMLParser(remove_blank_text=True, recover=True)
        item_elt = etree.fromstring(item_raw, parser)
        item_id = item_elt.get("id")
        try:
            payload = item_elt[0]
        except IndexError:
            self.disp(_("Item has not payload"), 1)
            return "", item_id
        return etree.tostring(payload, encoding="unicode", pretty_print=True), item_id

    async def start(self):
        (
            self.pubsub_service,
            self.pubsub_node,
            self.pubsub_item,
            content_file_path,
            content_file_obj,
        ) = await self.get_item_path()
        await self.run_editor("pubsub_editor_args", content_file_path, content_file_obj)
        self.host.quit()


class Rename(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "rename",
            use_pubsub=True,
            pubsub_flags={C.NODE, C.SINGLE_ITEM},
            help=_("rename a pubsub item"),
        )

    def add_parser_options(self):
        self.parser.add_argument("new_id", help=_("new item id to use"))

    async def start(self):
        try:
            await self.host.bridge.ps_item_rename(
                self.args.service,
                self.args.node,
                self.args.item,
                self.args.new_id,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't rename item: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp("Item renamed")
            self.host.quit(C.EXIT_OK)


class Subscribe(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "subscribe",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("subscribe to a node"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "--public",
            action="store_true",
            help=_("make the registration visible for everybody"),
        )

    async def start(self):
        options = {}
        if self.args.public:
            namespaces = await self.host.bridge.namespaces_get()
            try:
                ns_pps = namespaces["pps"]
            except KeyError:
                self.disp(
                    "Pubsub Public Subscription plugin is not loaded, can't use --public "
                    "option, subscription stopped", error=True
                )
                self.host.quit(C.EXIT_MISSING_FEATURE)
            else:
                options[f"{{{ns_pps}}}public"] = True
        try:
            sub_id = await self.host.bridge.ps_subscribe(
                self.args.service,
                self.args.node,
                data_format.serialise(options),
                self.profile,
            )
        except Exception as e:
            self.disp(_("can't subscribe to node: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("subscription done"), 1)
            if sub_id:
                self.disp(_("subscription id: {sub_id}").format(sub_id=sub_id))
            self.host.quit()


class Unsubscribe(base.CommandBase):
    # FIXME: check why we get a a NodeNotFound on subscribe just after unsubscribe

    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "unsubscribe",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            use_verbose=True,
            help=_("unsubscribe from a node"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            await self.host.bridge.ps_unsubscribe(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except Exception as e:
            self.disp(_("can't unsubscribe from node: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(_("subscription removed"), 1)
            self.host.quit()


class Subscriptions(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "subscriptions",
            use_output=C.OUTPUT_LIST_DICT,
            use_pubsub=True,
            help=_("retrieve all subscriptions on a service"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "--public",
            action="store_true",
            help=_("get public subscriptions"),
        )

    async def start(self):
        if self.args.public:
            method = self.host.bridge.ps_public_subscriptions_get
        else:
            method = self.host.bridge.ps_subscriptions_get
        try:
            subscriptions = data_format.deserialise(
                await method(
                    self.args.service,
                    self.args.node,
                    self.profile,
                ),
                type_check=list
            )
        except Exception as e:
            self.disp(_("can't retrieve subscriptions: {e}").format(e=e), error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(subscriptions)
            self.host.quit()


class Affiliations(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "affiliations",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            help=_("retrieve all affiliations on a service"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            affiliations = await self.host.bridge.ps_affiliations_get(
                self.args.service,
                self.args.node,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't get node affiliations: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(affiliations)
            self.host.quit()


class Reference(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "reference",
            use_pubsub=True,
            pubsub_flags={C.NODE, C.SINGLE_ITEM},
            help=_("send a reference/mention to pubsub item"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-t",
            "--type",
            default="mention",
            choices=("data", "mention"),
            help=_("type of reference to send (DEFAULT: mention)"),
        )
        self.parser.add_argument(
            "recipient",
            help=_("recipient of the reference")
        )

    async def start(self):
        service = self.args.service or await self.host.get_profile_jid()
        if self.args.item:
            anchor = uri.build_xmpp_uri(
                "pubsub", path=service, node=self.args.node, item=self.args.item
            )
        else:
            anchor = uri.build_xmpp_uri("pubsub", path=service, node=self.args.node)

        try:
            await self.host.bridge.reference_send(
                self.args.recipient,
                anchor,
                self.args.type,
                "",
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't send reference: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.host.quit()


class Search(base.CommandBase):
    """This command do a search without using MAM

    This commands checks every items it finds by itself,
    so it may be heavy in resources both for server and client
    """

    RE_FLAGS = re.MULTILINE | re.UNICODE
    EXEC_ACTIONS = ("exec", "external")

    def __init__(self, host):
        # FIXME: C.NO_MAX is not needed here, and this can be globally removed from consts
        #        the only interest is to change the help string, but this can be explained
        #        extensively in man pages (max is for each node found)
        base.CommandBase.__init__(
            self,
            host,
            "search",
            use_output=C.OUTPUT_XML,
            use_pubsub=True,
            pubsub_flags={C.MULTI_ITEMS, C.NO_MAX},
            use_verbose=True,
            help=_("search items corresponding to filters"),
        )

    @property
    def etree(self):
        """load lxml.etree only if needed"""
        if self._etree is None:
            from lxml import etree

            self._etree = etree
        return self._etree

    def filter_opt(self, value, type_):
        return (type_, value)

    def filter_flag(self, value, type_):
        value = C.bool(value)
        return (type_, value)

    def add_parser_options(self):
        self.parser.add_argument(
            "-D",
            "--max-depth",
            type=int,
            default=0,
            help=_(
                "maximum depth of recursion (will search linked nodes if > 0, "
                "DEFAULT: 0)"
            ),
        )
        self.parser.add_argument(
            "-M",
            "--node-max",
            type=int,
            default=30,
            help=_(
                "maximum number of items to get per node ({} to get all items, "
                "DEFAULT: 30)".format(C.NO_LIMIT)
            ),
        )
        self.parser.add_argument(
            "-N",
            "--namespace",
            action="append",
            nargs=2,
            default=[],
            metavar="NAME NAMESPACE",
            help=_("namespace to use for xpath"),
        )

        # filters
        filter_text = partial(self.filter_opt, type_="text")
        filter_re = partial(self.filter_opt, type_="regex")
        filter_xpath = partial(self.filter_opt, type_="xpath")
        filter_python = partial(self.filter_opt, type_="python")
        filters = self.parser.add_argument_group(
            _("filters"),
            _("only items corresponding to following filters will be kept"),
        )
        filters.add_argument(
            "-t",
            "--text",
            action="append",
            dest="filters",
            type=filter_text,
            metavar="TEXT",
            help=_("full text filter, item must contain this string (XML included)"),
        )
        filters.add_argument(
            "-r",
            "--regex",
            action="append",
            dest="filters",
            type=filter_re,
            metavar="EXPRESSION",
            help=_("like --text but using a regular expression"),
        )
        filters.add_argument(
            "-x",
            "--xpath",
            action="append",
            dest="filters",
            type=filter_xpath,
            metavar="XPATH",
            help=_("filter items which has elements matching this xpath"),
        )
        filters.add_argument(
            "-P",
            "--python",
            action="append",
            dest="filters",
            type=filter_python,
            metavar="PYTHON_CODE",
            help=_(
                "Python expression which much return a bool (True to keep item, "
                'False to reject it). "item" is raw text item, "item_xml" is '
                "lxml's etree.Element"
            ),
        )

        # filters flags
        flag_case = partial(self.filter_flag, type_="ignore-case")
        flag_invert = partial(self.filter_flag, type_="invert")
        flag_dotall = partial(self.filter_flag, type_="dotall")
        flag_matching = partial(self.filter_flag, type_="only-matching")
        flags = self.parser.add_argument_group(
            _("filters flags"),
            _("filters modifiers (change behaviour of following filters)"),
        )
        flags.add_argument(
            "-C",
            "--ignore-case",
            action="append",
            dest="filters",
            type=flag_case,
            const=("ignore-case", True),
            nargs="?",
            metavar="BOOLEAN",
            help=_("(don't) ignore case in following filters (DEFAULT: case sensitive)"),
        )
        flags.add_argument(
            "-I",
            "--invert",
            action="append",
            dest="filters",
            type=flag_invert,
            const=("invert", True),
            nargs="?",
            metavar="BOOLEAN",
            help=_("(don't) invert effect of following filters (DEFAULT: don't invert)"),
        )
        flags.add_argument(
            "-A",
            "--dot-all",
            action="append",
            dest="filters",
            type=flag_dotall,
            const=("dotall", True),
            nargs="?",
            metavar="BOOLEAN",
            help=_("(don't) use DOTALL option for regex (DEFAULT: don't use)"),
        )
        flags.add_argument(
            "-k",
            "--only-matching",
            action="append",
            dest="filters",
            type=flag_matching,
            const=("only-matching", True),
            nargs="?",
            metavar="BOOLEAN",
            help=_("keep only the matching part of the item"),
        )

        # action
        self.parser.add_argument(
            "action",
            default="print",
            nargs="?",
            choices=("print", "exec", "external"),
            help=_("action to do on found items (DEFAULT: print)"),
        )
        self.parser.add_argument("command", nargs=argparse.REMAINDER)

    async def get_items(self, depth, service, node, items):
        self.to_get += 1
        try:
            ps_result = data_format.deserialise(
                await self.host.bridge.ps_items_get(
                    service,
                    node,
                    self.args.node_max,
                    items,
                    "",
                    self.get_pubsub_extra(),
                    self.profile,
                )
            )
        except Exception as e:
            self.disp(
                f"can't get pubsub items at {service} (node: {node}): {e}",
                error=True,
            )
            self.to_get -= 1
        else:
            await self.search(ps_result, depth)

    def _check_pubsub_url(self, match, found_nodes):
        """check that the matched URL is an xmpp: one

        @param found_nodes(list[unicode]): found_nodes
            this list will be filled while xmpp: URIs are discovered
        """
        url = match.group(0)
        if url.startswith("xmpp"):
            try:
                url_data = uri.parse_xmpp_uri(url)
            except ValueError:
                return
            if url_data["type"] == "pubsub":
                found_node = {"service": url_data["path"], "node": url_data["node"]}
                if "item" in url_data:
                    found_node["item"] = url_data["item"]
                found_nodes.append(found_node)

    async def get_sub_nodes(self, item, depth):
        """look for pubsub URIs in item, and get_items on the linked nodes"""
        found_nodes = []
        checkURI = partial(self._check_pubsub_url, found_nodes=found_nodes)
        strings.RE_URL.sub(checkURI, item)
        for data in found_nodes:
            await self.get_items(
                depth + 1,
                data["service"],
                data["node"],
                [data["item"]] if "item" in data else [],
            )

    def parseXml(self, item):
        try:
            return self.etree.fromstring(item)
        except self.etree.XMLSyntaxError:
            self.disp(
                _(
                    "item doesn't looks like XML, you have probably used --only-matching "
                    "somewhere before and we have no more XML"
                ),
                error=True,
            )
            self.host.quit(C.EXIT_BAD_ARG)

    def filter(self, item):
        """apply filters given on command line

        if only-matching is used, item may be modified
        @return (tuple[bool, unicode]): a tuple with:
            - keep: True if item passed the filters
            - item: it is returned in case of modifications
        """
        ignore_case = False
        invert = False
        dotall = False
        only_matching = False
        item_xml = None
        for type_, value in self.args.filters:
            keep = True

            ## filters

            if type_ == "text":
                if ignore_case:
                    if value.lower() not in item.lower():
                        keep = False
                else:
                    if value not in item:
                        keep = False
                if keep and only_matching:
                    # doesn't really make sens to keep a fixed string
                    # so we raise an error
                    self.host.disp(
                        _("--only-matching used with fixed --text string, are you sure?"),
                        error=True,
                    )
                    self.host.quit(C.EXIT_BAD_ARG)
            elif type_ == "regex":
                flags = self.RE_FLAGS
                if ignore_case:
                    flags |= re.IGNORECASE
                if dotall:
                    flags |= re.DOTALL
                match = re.search(value, item, flags)
                keep = match != None
                if keep and only_matching:
                    item = match.group()
                    item_xml = None
            elif type_ == "xpath":
                if item_xml is None:
                    item_xml = self.parseXml(item)
                try:
                    elts = item_xml.xpath(value, namespaces=self.args.namespace)
                except self.etree.XPathEvalError as e:
                    self.disp(_("can't use xpath: {reason}").format(reason=e), error=True)
                    self.host.quit(C.EXIT_BAD_ARG)
                keep = bool(elts)
                if keep and only_matching:
                    item_xml = elts[0]
                    try:
                        item = self.etree.tostring(item_xml, encoding="unicode")
                    except TypeError:
                        # we have a string only, not an element
                        item = str(item_xml)
                        item_xml = None
            elif type_ == "python":
                if item_xml is None:
                    item_xml = self.parseXml(item)
                cmd_ns = {"etree": self.etree, "item": item, "item_xml": item_xml}
                try:
                    keep = eval(value, cmd_ns)
                except SyntaxError as e:
                    self.disp(str(e), error=True)
                    self.host.quit(C.EXIT_BAD_ARG)

                    ## flags

            elif type_ == "ignore-case":
                ignore_case = value
            elif type_ == "invert":
                invert = value
                #  we need to continue, else loop would end here
                continue
            elif type_ == "dotall":
                dotall = value
            elif type_ == "only-matching":
                only_matching = value
            else:
                raise exceptions.InternalError(
                    _("unknown filter type {type}").format(type=type_)
                )

            if invert:
                keep = not keep
            if not keep:
                return False, item

        return True, item

    async def do_item_action(self, item, metadata):
        """called when item has been kepts and the action need to be done

        @param item(unicode): accepted item
        """
        action = self.args.action
        if action == "print" or self.host.verbosity > 0:
            try:
                await self.output(item)
            except self.etree.XMLSyntaxError:
                # item is not valid XML, but a string
                # can happen when --only-matching is used
                self.disp(item)
        if action in self.EXEC_ACTIONS:
            item_elt = self.parseXml(item)
            if action == "exec":
                use = {
                    "service": metadata["service"],
                    "node": metadata["node"],
                    "item": item_elt.get("id"),
                    "profile": self.profile,
                }
                # we need to send a copy of self.args.command
                # else it would be modified
                parser_args, use_args = arg_tools.get_use_args(
                    self.host, self.args.command, use, verbose=self.host.verbosity > 1
                )
                cmd_args = sys.argv[0:1] + parser_args + use_args
            else:
                cmd_args = self.args.command

            self.disp(
                "COMMAND: {command}".format(
                    command=" ".join([arg_tools.escape(a) for a in cmd_args])
                ),
                2,
            )
            if action == "exec":
                p = await asyncio.create_subprocess_exec(*cmd_args)
                ret = await p.wait()
            else:
                p = await asyncio.create_subprocess_exec(*cmd_args, stdin=subprocess.PIPE)
                await p.communicate(item.encode(sys.getfilesystemencoding()))
                ret = p.returncode
            if ret != 0:
                self.disp(
                    A.color(
                        C.A_FAILURE,
                        _("executed command failed with exit code {ret}").format(ret=ret),
                    )
                )

    async def search(self, ps_result, depth):
        """callback of get_items

        this method filters items, get sub nodes if needed,
        do the requested action, and exit the command when everything is done
        @param items_data(tuple): result of get_items
        @param depth(int): current depth level
            0 for first node, 1 for first children, and so on
        """
        for item in ps_result["items"]:
            if depth < self.args.max_depth:
                await self.get_sub_nodes(item, depth)
            keep, item = self.filter(item)
            if not keep:
                continue
            await self.do_item_action(item, ps_result)

            #  we check if we got all get_items results
        self.to_get -= 1
        if self.to_get == 0:
            # yes, we can quit
            self.host.quit()
        assert self.to_get > 0

    async def start(self):
        if self.args.command:
            if self.args.action not in self.EXEC_ACTIONS:
                self.parser.error(
                    _("Command can only be used with {actions} actions").format(
                        actions=", ".join(self.EXEC_ACTIONS)
                    )
                )
        else:
            if self.args.action in self.EXEC_ACTIONS:
                self.parser.error(_("you need to specify a command to execute"))
        if not self.args.node:
            # TODO: handle get service affiliations when node is not set
            self.parser.error(_("empty node is not handled yet"))
            # to_get is increased on each get and decreased on each answer
            # when it reach 0 again, the command is finished
        self.to_get = 0
        self._etree = None
        if self.args.filters is None:
            self.args.filters = []
        self.args.namespace = dict(
            self.args.namespace + [("pubsub", "http://jabber.org/protocol/pubsub")]
        )
        await self.get_items(0, self.args.service, self.args.node, self.args.items)


class Transform(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "transform",
            use_pubsub=True,
            pubsub_flags={C.NODE, C.MULTI_ITEMS},
            help=_("modify items of a node using an external command/script"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "--apply",
            action="store_true",
            help=_("apply transformation (DEFAULT: do a dry run)"),
        )
        self.parser.add_argument(
            "--admin",
            action="store_true",
            help=_("do a pubsub admin request, needed to change publisher"),
        )
        self.parser.add_argument(
            "-I",
            "--ignore-errors",
            action="store_true",
            help=_(
                "if command return a non zero exit code, ignore the item and continue"
            ),
        )
        self.parser.add_argument(
            "-A",
            "--all",
            action="store_true",
            help=_("get all items by looping over all pages using RSM"),
        )
        self.parser.add_argument(
            "command_path",
            help=_(
                "path to the command to use. Will be called repetitivly with an "
                "item as input. Output (full item XML) will be used as new one. "
                'Return "DELETE" string to delete the item, and "SKIP" to ignore it'
            ),
        )

    async def ps_items_send_cb(self, item_ids, metadata):
        if item_ids:
            self.disp(
                _("items published with ids {item_ids}").format(
                    item_ids=", ".join(item_ids)
                )
            )
        else:
            self.disp(_("items published"))
        if self.args.all:
            return await self.handle_next_page(metadata)
        else:
            self.host.quit()

    async def handle_next_page(self, metadata):
        """Retrieve new page through RSM or quit if we're in the last page

        use to handle --all option
        @param metadata(dict): metadata as returned by ps_items_get
        """
        try:
            last = metadata["rsm"]["last"]
            index = int(metadata["rsm"]["index"])
            count = int(metadata["rsm"]["count"])
        except KeyError:
            self.disp(
                _("Can't retrieve all items, RSM metadata not available"), error=True
            )
            self.host.quit(C.EXIT_MISSING_FEATURE)
        except ValueError as e:
            self.disp(
                _("Can't retrieve all items, bad RSM metadata: {msg}").format(msg=e),
                error=True,
            )
            self.host.quit(C.EXIT_ERROR)

        if index + self.args.rsm_max >= count:
            self.disp(_("All items transformed"))
            self.host.quit(0)

        self.disp(
            _("Retrieving next page ({page_idx}/{page_total})").format(
                page_idx=int(index / self.args.rsm_max) + 1,
                page_total=int(count / self.args.rsm_max),
            )
        )

        extra = self.get_pubsub_extra()
        extra["rsm_after"] = last
        try:
            ps_result = await data_format.deserialise(
                self.host.bridge.ps_items_get(
                    self.args.service,
                    self.args.node,
                    self.args.rsm_max,
                    self.args.items,
                    "",
                    data_format.serialise(extra),
                    self.profile,
                )
            )
        except Exception as e:
            self.disp(f"can't retrieve items: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.ps_items_get_cb(ps_result)

    async def ps_items_get_cb(self, ps_result):
        encoding = "utf-8"
        new_items = []

        for item in ps_result["items"]:
            if self.check_duplicates:
                # this is used when we are not ordering by creation
                # to avoid infinite loop
                item_elt, __ = xml_tools.etree_parse(self, item)
                item_id = item_elt.get("id")
                if item_id in self.items_ids:
                    self.disp(
                        _(
                            "Duplicate found on item {item_id}, we have probably handled "
                            "all items."
                        ).format(item_id=item_id)
                    )
                    self.host.quit()
                self.items_ids.append(item_id)

                # we launch the command to filter the item
            try:
                p = await asyncio.create_subprocess_exec(
                    self.args.command_path, stdin=subprocess.PIPE, stdout=subprocess.PIPE
                )
            except OSError as e:
                exit_code = C.EXIT_CMD_NOT_FOUND if e.errno == 2 else C.EXIT_ERROR
                self.disp(f"Can't execute the command: {e}", error=True)
                self.host.quit(exit_code)
            encoding = "utf-8"
            cmd_std_out, cmd_std_err = await p.communicate(item.encode(encoding))
            ret = p.returncode
            if ret != 0:
                self.disp(
                    f"The command returned a non zero status while parsing the "
                    f"following item:\n\n{item}",
                    error=True,
                )
                if self.args.ignore_errors:
                    continue
                else:
                    self.host.quit(C.EXIT_CMD_ERROR)
            if cmd_std_err is not None:
                cmd_std_err = cmd_std_err.decode(encoding, errors="ignore")
                self.disp(cmd_std_err, error=True)
            cmd_std_out = cmd_std_out.decode(encoding).strip()
            if cmd_std_out == "DELETE":
                item_elt, __ = xml_tools.etree_parse(self, item)
                item_id = item_elt.get("id")
                self.disp(_("Deleting item {item_id}").format(item_id=item_id))
                if self.args.apply:
                    try:
                        await self.host.bridge.ps_item_retract(
                            self.args.service,
                            self.args.node,
                            item_id,
                            False,
                            self.profile,
                        )
                    except Exception as e:
                        self.disp(f"can't delete item {item_id}: {e}", error=True)
                        self.host.quit(C.EXIT_BRIDGE_ERRBACK)
                continue
            elif cmd_std_out == "SKIP":
                item_elt, __ = xml_tools.etree_parse(self, item)
                item_id = item_elt.get("id")
                self.disp(_("Skipping item {item_id}").format(item_id=item_id))
                continue
            element, etree = xml_tools.etree_parse(self, cmd_std_out)

            # at this point command has been run and we have a etree.Element object
            if element.tag not in ("item", "{http://jabber.org/protocol/pubsub}item"):
                self.disp(
                    "your script must return a whole item, this is not:\n{xml}".format(
                        xml=etree.tostring(element, encoding="unicode")
                    ),
                    error=True,
                )
                self.host.quit(C.EXIT_DATA_ERROR)

            if not self.args.apply:
                # we have a dry run, we just display filtered items
                serialised = etree.tostring(
                    element, encoding="unicode", pretty_print=True
                )
                self.disp(serialised)
            else:
                new_items.append(etree.tostring(element, encoding="unicode"))

        if not self.args.apply:
            # on dry run we have nothing to wait for, we can quit
            if self.args.all:
                return await self.handle_next_page(ps_result)
            self.host.quit()
        else:
            if self.args.admin:
                bridge_method = self.host.bridge.ps_admin_items_send
            else:
                bridge_method = self.host.bridge.ps_items_send

            try:
                ps_items_send_result = await bridge_method(
                    self.args.service,
                    self.args.node,
                    new_items,
                    "",
                    self.profile,
                )
            except Exception as e:
                self.disp(f"can't send item: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)
            else:
                await self.ps_items_send_cb(ps_items_send_result, metadata=ps_result)

    async def start(self):
        if self.args.all and self.args.order_by != C.ORDER_BY_CREATION:
            self.check_duplicates = True
            self.items_ids = []
            self.disp(
                A.color(
                    A.FG_RED,
                    A.BOLD,
                    '/!\\ "--all" should be used with "--order-by creation" /!\\\n',
                    A.RESET,
                    "We'll update items, so order may change during transformation,\n"
                    "we'll try to mitigate that by stopping on first duplicate,\n"
                    "but this method is not safe, and some items may be missed.\n---\n",
                )
            )
        else:
            self.check_duplicates = False

        try:
            ps_result = data_format.deserialise(
                await self.host.bridge.ps_items_get(
                    self.args.service,
                    self.args.node,
                    self.args.max,
                    self.args.items,
                    "",
                    self.get_pubsub_extra(),
                    self.profile,
                )
            )
        except Exception as e:
            self.disp(f"can't retrieve items: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.ps_items_get_cb(ps_result)


class Uri(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "uri",
            use_profile=False,
            use_pubsub=True,
            pubsub_flags={C.NODE, C.SINGLE_ITEM},
            help=_("build URI"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-p",
            "--profile",
            default=C.PROF_KEY_DEFAULT,
            help=_("profile (used when no server is specified)"),
        )

    def display_uri(self, jid_):
        uri_args = {}
        if not self.args.service:
            self.args.service = jid.JID(jid_).bare

        for key in ("node", "service", "item"):
            value = getattr(self.args, key)
            if key == "service":
                key = "path"
            if value:
                uri_args[key] = value
        self.disp(uri.build_xmpp_uri("pubsub", **uri_args))
        self.host.quit()

    async def start(self):
        if not self.args.service:
            try:
                jid_ = await self.host.bridge.param_get_a_async(
                    "JabberID", "Connection", profile_key=self.args.profile
                )
            except Exception as e:
                self.disp(f"can't retrieve jid: {e}", error=True)
                self.host.quit(C.EXIT_BRIDGE_ERRBACK)
            else:
                self.display_uri(jid_)
        else:
            self.display_uri(None)


class AttachmentGet(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "get",
            use_output=C.OUTPUT_LIST_DICT,
            use_pubsub=True,
            pubsub_flags={C.SERVICE, C.NODE, C.SINGLE_ITEM},
            help=_("get data attached to an item"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-j",
            "--jid",
            action="append",
            dest="jids",
            help=_(
                "get attached data published only by those JIDs (DEFAULT: get all "
                "attached data)"
            )
        )

    async def start(self):
        try:
            attached_data, __ = await self.host.bridge.ps_attachments_get(
                self.args.service,
                self.args.node,
                self.args.item,
                self.args.jids or [],
                "",
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't get attached data: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            attached_data = data_format.deserialise(attached_data, type_check=list)
            await self.output(attached_data)
            self.host.quit(C.EXIT_OK)


class AttachmentSet(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "set",
            use_pubsub=True,
            pubsub_flags={C.SERVICE, C.NODE, C.SINGLE_ITEM},
            help=_("attach data to an item"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "--replace",
            action="store_true",
            help=_(
                "replace previous versions of attachments (DEFAULT: update previous "
                "version)"
            )
        )
        self.parser.add_argument(
            "-N",
            "--noticed",
            metavar="BOOLEAN",
            nargs="?",
            default="keep",
            help=_("mark item as (un)noticed (DEFAULT: keep current value))")
        )
        self.parser.add_argument(
            "-r",
            "--reactions",
            # FIXME: to be replaced by "extend" when we stop supporting python 3.7
            action="append",
            help=_("emojis to add to react to an item")
        )
        self.parser.add_argument(
            "-R",
            "--reactions-remove",
            # FIXME: to be replaced by "extend" when we stop supporting python 3.7
            action="append",
            help=_("emojis to remove from reactions to an item")
        )

    async def start(self):
        attachments_data = {
            "service": self.args.service,
            "node": self.args.node,
            "id": self.args.item,
            "extra": {}
        }
        operation = "replace" if self.args.replace else "update"
        if self.args.noticed != "keep":
            if self.args.noticed is None:
                self.args.noticed = C.BOOL_TRUE
            attachments_data["extra"]["noticed"] = C.bool(self.args.noticed)

        if self.args.reactions or self.args.reactions_remove:
            reactions = attachments_data["extra"]["reactions"] = {
                "operation": operation
            }
            if self.args.replace:
                reactions["reactions"] = self.args.reactions
            else:
                reactions["add"] = self.args.reactions
                reactions["remove"] = self.args.reactions_remove


        if not attachments_data["extra"]:
            self.parser.error(_("At leat one attachment must be specified."))

        try:
            await self.host.bridge.ps_attachments_set(
                data_format.serialise(attachments_data),
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't attach data to item: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp("data attached")
            self.host.quit(C.EXIT_OK)


class Attachments(base.CommandBase):
    subcommands = (AttachmentGet, AttachmentSet)

    def __init__(self, host):
        super().__init__(
            host,
            "attachments",
            use_profile=False,
            help=_("set or retrieve items attachments"),
        )


class SignatureSign(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "sign",
            use_pubsub=True,
            pubsub_flags={C.NODE, C.SINGLE_ITEM},
            help=_("sign an item"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        attachments_data = {
            "service": self.args.service,
            "node": self.args.node,
            "id": self.args.item,
            "extra": {
                # we set None to use profile's bare JID
                "signature": {"signer": None}
            }
        }
        try:
            await self.host.bridge.ps_attachments_set(
                data_format.serialise(attachments_data),
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't sign the item: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(f"item {self.args.item!r} has been signed")
            self.host.quit(C.EXIT_OK)


class SignatureCheck(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "check",
            use_output=C.OUTPUT_DICT,
            use_pubsub=True,
            pubsub_flags={C.SERVICE, C.NODE, C.SINGLE_ITEM},
            help=_("check the validity of pubsub signature"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "signature",
            metavar="JSON",
            help=_("signature data")
        )

    async def start(self):
        try:
            ret_s = await self.host.bridge.ps_signature_check(
                self.args.service,
                self.args.node,
                self.args.item,
                self.args.signature,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't check signature: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(data_format.deserialise((ret_s)))
            self.host.quit()


class Signature(base.CommandBase):
    subcommands = (
        SignatureSign,
        SignatureCheck,
    )

    def __init__(self, host):
        super().__init__(
            host, "signature", use_profile=False, help=_("items signatures")
        )


class SecretShare(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "share",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("share a secret to let other entity encrypt or decrypt items"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-k", "--key", metavar="ID", dest="secret_ids", action="append", default=[],
            help=_(
                "only share secrets with those IDs (default: share all secrets of the "
                "node)"
            )
        )
        self.parser.add_argument(
            "recipient", metavar="JID", help=_("entity who must get the shared secret")
        )

    async def start(self):
        try:
            await self.host.bridge.ps_secret_share(
                self.args.recipient,
                self.args.service,
                self.args.node,
                self.args.secret_ids,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't share secret: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp("secrets have been shared")
            self.host.quit(C.EXIT_OK)


class SecretRevoke(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "revoke",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("revoke an encrypted node secret"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "secret_id", help=_("ID of the secrets to revoke")
        )
        self.parser.add_argument(
            "-r", "--recipient", dest="recipients", metavar="JID", action="append",
            default=[], help=_(
                "entity who must get the revocation notification (default: send to all "
                "entities known to have the shared secret)"
            )
        )

    async def start(self):
        try:
            await self.host.bridge.ps_secret_revoke(
                self.args.service,
                self.args.node,
                self.args.secret_id,
                self.args.recipients,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't revoke secret: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp("secret {self.args.secret_id} has been revoked.")
            self.host.quit(C.EXIT_OK)


class SecretRotate(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "rotate",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("revoke existing secrets, create a new one and send notifications"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-r", "--recipient", dest="recipients", metavar="JID", action="append",
            default=[], help=_(
                "entity who must get the revocation and shared secret notifications "
                "(default: send to all entities known to have the shared secret)"
            )
        )

    async def start(self):
        try:
            await self.host.bridge.ps_secret_rotate(
                self.args.service,
                self.args.node,
                self.args.recipients,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't rotate secret: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp("secret has been rotated")
            self.host.quit(C.EXIT_OK)


class SecretList(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "list",
            use_pubsub=True,
            use_verbose=True,
            pubsub_flags={C.NODE},
            help=_("list known secrets for a pubsub node"),
            use_output=C.OUTPUT_LIST_DICT
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            secrets = data_format.deserialise(await self.host.bridge.ps_secrets_list(
                self.args.service,
                self.args.node,
                self.profile,
            ), type_check=list)
        except Exception as e:
            self.disp(f"can't list node secrets: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            if not self.verbosity:
                # we don't print key if verbosity is not a least one, to avoid showing it
                # on the screen accidentally
                for secret in secrets:
                    del secret["key"]
            await self.output(secrets)
            self.host.quit(C.EXIT_OK)


class Secret(base.CommandBase):
    subcommands = (SecretShare, SecretRevoke, SecretRotate, SecretList)

    def __init__(self, host):
        super().__init__(
            host,
            "secret",
            use_profile=False,
            help=_("handle encrypted nodes secrets"),
        )


class HookCreate(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "create",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("create a Pubsub hook"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-t",
            "--type",
            default="python",
            choices=("python", "python_file", "python_code"),
            help=_("hook type"),
        )
        self.parser.add_argument(
            "-P",
            "--persistent",
            action="store_true",
            help=_("make hook persistent across restarts"),
        )
        self.parser.add_argument(
            "hook_arg",
            help=_("argument of the hook (depend of the type)"),
        )

    @staticmethod
    def check_args(self):
        if self.args.type == "python_file":
            self.args.hook_arg = os.path.abspath(self.args.hook_arg)
            if not os.path.isfile(self.args.hook_arg):
                self.parser.error(
                    _("{path} is not a file").format(path=self.args.hook_arg)
                )

    async def start(self):
        self.check_args(self)
        try:
            await self.host.bridge.ps_hook_add(
                self.args.service,
                self.args.node,
                self.args.type,
                self.args.hook_arg,
                self.args.persistent,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't create hook: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.host.quit()


class HookDelete(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "delete",
            use_pubsub=True,
            pubsub_flags={C.NODE},
            help=_("delete a Pubsub hook"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "-t",
            "--type",
            default="",
            choices=("", "python", "python_file", "python_code"),
            help=_("hook type to remove, empty to remove all (DEFAULT: remove all)"),
        )
        self.parser.add_argument(
            "-a",
            "--arg",
            dest="hook_arg",
            default="",
            help=_(
                "argument of the hook to remove, empty to remove all (DEFAULT: remove all)"
            ),
        )

    async def start(self):
        HookCreate.check_args(self)
        try:
            nb_deleted = await self.host.bridge.ps_hook_remove(
                self.args.service,
                self.args.node,
                self.args.type,
                self.args.hook_arg,
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't delete hook: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.disp(
                _("{nb_deleted} hook(s) have been deleted").format(nb_deleted=nb_deleted)
            )
            self.host.quit()


class HookList(base.CommandBase):
    def __init__(self, host):
        base.CommandBase.__init__(
            self,
            host,
            "list",
            use_output=C.OUTPUT_LIST_DICT,
            help=_("list hooks of a profile"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            data = await self.host.bridge.ps_hook_list(
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't list hooks: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            if not data:
                self.disp(_("No hook found."))
            await self.output(data)
            self.host.quit()


class Hook(base.CommandBase):
    subcommands = (HookCreate, HookDelete, HookList)

    def __init__(self, host):
        super(Hook, self).__init__(
            host,
            "hook",
            use_profile=False,
            use_verbose=True,
            help=_("trigger action on Pubsub notifications"),
        )


class Pubsub(base.CommandBase):
    subcommands = (
        Set,
        Get,
        Delete,
        Edit,
        Rename,
        Subscribe,
        Unsubscribe,
        Subscriptions,
        Affiliations,
        Reference,
        Search,
        Transform,
        Attachments,
        Signature,
        Secret,
        Hook,
        Uri,
        Node,
        Cache,
    )

    def __init__(self, host):
        super(Pubsub, self).__init__(
            host, "pubsub", use_profile=False, help=_("PubSub nodes/items management")
        )