0aa9613693
`pubsub/attachments` subcommands has a `get` and a `set` subcommand to manage attachments of a pubsub item. `blog` has the same subcommands, the only difference is that if `--node` is not specified, the default blog namespace will be used. rel 370
2718 lines
88 KiB
Python
2718 lines
88 KiB
Python
#!/usr/bin/env python3
|
||
|
||
|
||
# jp: a SàT command line tool
|
||
# 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 sat.core.i18n import _
|
||
from sat.core import exceptions
|
||
from sat_frontends.jp.constants import Const as C
|
||
from sat_frontends.jp import common
|
||
from sat_frontends.jp import arg_tools
|
||
from sat_frontends.jp import xml_tools
|
||
from functools import partial
|
||
from sat.tools.common import data_format
|
||
from sat.tools.common import uri
|
||
from sat.tools.common.ansi import ANSI as A
|
||
from sat.tools.common import date_utils
|
||
from sat_frontends.tools import jid, strings
|
||
from sat_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 removePrefix(self, key):
|
||
return key[7:] if key.startswith("pubsub#") else key
|
||
|
||
def filterKey(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.psNodeConfigurationGet(
|
||
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.filterKey
|
||
config_dict = {
|
||
self.removePrefix(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.psNodeCreate(
|
||
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.confirmOrQuit(message, _("node purge cancelled"))
|
||
|
||
try:
|
||
await self.host.bridge.psNodePurge(
|
||
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.confirmOrQuit(message, _("node deletion cancelled"))
|
||
|
||
try:
|
||
await self.host.bridge.psNodeDelete(
|
||
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 getKeyName(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.psNodeConfigurationSet(
|
||
self.args.service,
|
||
self.args.node,
|
||
{self.getKeyName(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.etreeParse(
|
||
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.etreeParse(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.psAdminItemsSend
|
||
else:
|
||
self.disp(
|
||
_(
|
||
"Items are imported without using admin mode, publisher can't "
|
||
"be changed"
|
||
)
|
||
)
|
||
method = self.host.bridge.psItemsSend
|
||
|
||
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.psNodeAffiliationsGet(
|
||
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.psNodeAffiliationsSet(
|
||
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.psPublicNodeSubscriptionsGet
|
||
else:
|
||
method = self.host.bridge.psNodeSubscriptionsGet
|
||
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.psNodeSubscriptionsSet(
|
||
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.psSchemaSet(
|
||
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.psSchemaSet(
|
||
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 psSchemaGetCb(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.getTmpFile()
|
||
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.runEditor(
|
||
"pubsub_schema_editor_args", content_file_path, content_file_obj
|
||
)
|
||
|
||
async def start(self):
|
||
try:
|
||
schema = await self.host.bridge.psSchemaGet(
|
||
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.psSchemaGetCb(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.psSchemaGet(
|
||
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.psCacheGet(
|
||
self.args.service,
|
||
self.args.node,
|
||
self.args.max,
|
||
self.args.items,
|
||
self.args.sub_id,
|
||
self.getPubsubExtra(),
|
||
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.psCacheSync(
|
||
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.confirmOrQuit(
|
||
_(
|
||
"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.psCachePurge(
|
||
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.confirmOrQuit(
|
||
_(
|
||
"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.psCacheReset()
|
||
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.psCacheSearch(
|
||
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(
|
||
"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.etreeParse(self, sys.stdin)
|
||
element = xml_tools.getPayload(self, element)
|
||
payload = etree.tostring(element, encoding="unicode")
|
||
extra = {}
|
||
publish_options = NodeCreate.get_config_options(self.args)
|
||
if publish_options:
|
||
extra["publish_options"] = publish_options
|
||
|
||
try:
|
||
published_id = await self.host.bridge.psItemSend(
|
||
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"),
|
||
)
|
||
# TODO: a key(s) argument to select keys to display
|
||
|
||
async def start(self):
|
||
try:
|
||
ps_result = data_format.deserialise(
|
||
await self.host.bridge.psItemsGet(
|
||
self.args.service,
|
||
self.args.node,
|
||
self.args.max,
|
||
self.args.items,
|
||
self.args.sub_id,
|
||
self.getPubsubExtra(),
|
||
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.confirmOrQuit(message, _("item deletion cancelled"))
|
||
try:
|
||
await self.host.bridge.psItemRetract(
|
||
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):
|
||
pass
|
||
|
||
async def publish(self, content):
|
||
published_id = await self.host.bridge.psItemSend(
|
||
self.pubsub_service,
|
||
self.pubsub_node,
|
||
content,
|
||
self.pubsub_item or "",
|
||
"",
|
||
self.profile,
|
||
)
|
||
if published_id:
|
||
self.disp("Item published at {pub_id}".format(pub_id=published_id))
|
||
else:
|
||
self.disp("Item published")
|
||
|
||
async def getItemData(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.psItemsGet(
|
||
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.getItemPath()
|
||
await self.runEditor("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.psItemRename(
|
||
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.namespacesGet()
|
||
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.psSubscribe(
|
||
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.psUnsubscribe(
|
||
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.psPublicSubscriptionsGet
|
||
else:
|
||
method = self.host.bridge.psSubscriptionsGet
|
||
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.psAffiliationsGet(
|
||
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.buildXMPPUri(
|
||
"pubsub", path=service, node=self.args.node, item=self.args.item
|
||
)
|
||
else:
|
||
anchor = uri.buildXMPPUri("pubsub", path=service, node=self.args.node)
|
||
|
||
try:
|
||
await self.host.bridge.referenceSend(
|
||
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 getItems(self, depth, service, node, items):
|
||
self.to_get += 1
|
||
try:
|
||
ps_result = data_format.deserialise(
|
||
await self.host.bridge.psItemsGet(
|
||
service,
|
||
node,
|
||
self.args.node_max,
|
||
items,
|
||
"",
|
||
self.getPubsubExtra(),
|
||
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 _checkPubsubURL(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.parseXMPPUri(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 getSubNodes(self, item, depth):
|
||
"""look for pubsub URIs in item, and getItems on the linked nodes"""
|
||
found_nodes = []
|
||
checkURI = partial(self._checkPubsubURL, found_nodes=found_nodes)
|
||
strings.RE_URL.sub(checkURI, item)
|
||
for data in found_nodes:
|
||
await self.getItems(
|
||
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 doItemAction(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 getItems
|
||
|
||
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 getItems
|
||
@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.getSubNodes(item, depth)
|
||
keep, item = self.filter(item)
|
||
if not keep:
|
||
continue
|
||
await self.doItemAction(item, ps_result)
|
||
|
||
# we check if we got all getItems 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.getItems(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 psItemsSendCb(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.handleNextPage(metadata)
|
||
else:
|
||
self.host.quit()
|
||
|
||
async def handleNextPage(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 psItemsGet
|
||
"""
|
||
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.getPubsubExtra()
|
||
extra["rsm_after"] = last
|
||
try:
|
||
ps_result = await data_format.deserialise(
|
||
self.host.bridge.psItemsGet(
|
||
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.psItemsGetCb(ps_result)
|
||
|
||
async def psItemsGetCb(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.etreeParse(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.etreeParse(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.psItemRetract(
|
||
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.etreeParse(self, item)
|
||
item_id = item_elt.get("id")
|
||
self.disp(_("Skipping item {item_id}").format(item_id=item_id))
|
||
continue
|
||
element, etree = xml_tools.etreeParse(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.handleNextPage(ps_result)
|
||
self.host.quit()
|
||
else:
|
||
if self.args.admin:
|
||
bridge_method = self.host.bridge.psAdminItemsSend
|
||
else:
|
||
bridge_method = self.host.bridge.psItemsSend
|
||
|
||
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.psItemsSendCb(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.psItemsGet(
|
||
self.args.service,
|
||
self.args.node,
|
||
self.args.max,
|
||
self.args.items,
|
||
"",
|
||
self.getPubsubExtra(),
|
||
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.psItemsGetCb(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.buildXMPPUri("pubsub", **uri_args))
|
||
self.host.quit()
|
||
|
||
async def start(self):
|
||
if not self.args.service:
|
||
try:
|
||
jid_ = await self.host.bridge.asyncGetParamA(
|
||
"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.psAttachmentsGet(
|
||
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(
|
||
"-R",
|
||
"--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 replace by "extend" when we stop supporting python 3.7
|
||
action="append",
|
||
help=_("add emojis to react to an item")
|
||
)
|
||
|
||
async def start(self):
|
||
mb_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
|
||
mb_data["extra"]["noticed"] = C.bool(self.args.noticed)
|
||
|
||
if self.args.reactions is not None:
|
||
mb_data["extra"]["reaction"] = {
|
||
"operation": operation,
|
||
"reactions": [r for emojis in self.args.reactions for r in emojis]
|
||
}
|
||
|
||
if not mb_data["extra"]:
|
||
self.parser.error(_("At leat one attachment must be specified."))
|
||
|
||
try:
|
||
await self.host.bridge.psAttachmentsSet(
|
||
data_format.serialise(mb_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 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 checkArgs(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.checkArgs(self)
|
||
try:
|
||
await self.host.bridge.psHookAdd(
|
||
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.checkArgs(self)
|
||
try:
|
||
nb_deleted = await self.host.bridge.psHookRemove(
|
||
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.psHookList(
|
||
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,
|
||
Hook,
|
||
Uri,
|
||
Node,
|
||
Cache,
|
||
)
|
||
|
||
def __init__(self, host):
|
||
super(Pubsub, self).__init__(
|
||
host, "pubsub", use_profile=False, help=_("PubSub nodes/items management")
|
||
)
|