cli (contacts): add commands to handle pubsub contacts:

Add commands to retrieve (`contacts get`) and set (`contacts set`) pubsub contacts. As
they are managed with pubsub node, deletion can be done using `pubsub delete`.

rel 475

libervia/cli/cmd_contacts.py

@@ -0,0 +1,163 @@
+#!/usr/bin/env python3
+
+# Libervia CLI
+# Copyright (C) 2009-2026 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/>.
+
+from rich.table import Table, Column
+from rich.text import Text
+from libervia.backend.core.i18n import _
+from libervia.backend.plugins.plugin_misc_pubsub_contacts import (
+    Contact,
+    ContactSetRequest,
+    Contacts,
+    ContactsRequest,
+    Group,
+    Groups,
+)
+from libervia.cli.constants import Const as C
+
+from . import base
+
+__commands__ = ["ContactsCmd"]
+
+
+class Get(base.CommandBase):
+
+    def __init__(self, host):
+        super().__init__(
+            host,
+            "get",
+            use_output=C.OUTPUT_MODEL,
+            extra_outputs={"default": self.default_output},
+            help=_("retrieve the contacts list"),
+        )
+
+    def add_parser_options(self):
+        self.parser.add_argument(
+            "--ignore-cache",
+            action="store_true",
+            help="don't use cache and re-fetch data from pubsub service.",
+        )
+
+    def default_output(self, contacts: Contacts):
+        for idx, contact in enumerate(contacts):
+            table = Table(
+                Column("key", style="header"),
+                "value",
+                box=None,
+                show_header=False,
+            )
+            table.add_row("🏷️name", Text(contact.name or "", style="primary"))
+            table.add_row("🆔id", Text(contact.id or "", style="italic"))
+            group_table = Table.grid("name", "id", "description", padding=(0, 1))
+            for group in contact.groups:
+                group_table.add_row(
+                    Text(group.name or "", style="bold"),
+                    Text(group.id or "", style="italic"),
+                    group.description,
+                    "🔒" if group.encrypted else ""
+                )
+            if idx:
+                self.console.print()
+            table.add_row("👥groups", group_table)
+            table.add_row("🔒encrypted", str(contact.encrypted).lower())
+
+            self.console.print(table)
+
+
+    async def start(self):
+        try:
+            contacts_s = await self.host.bridge.ps_contacts_get(
+                ContactsRequest(ignore_cache=self.args.ignore_cache).model_dump_json(),
+                self.host.profile,
+            )
+        except Exception as e:
+            self.disp(f"error while retrieving the contacts: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+        else:
+            contacts = Contacts.model_validate_json(contacts_s)
+            await self.output(contacts)
+            self.host.quit()
+
+
+class Set(base.CommandBase):
+
+    def __init__(self, host):
+        super().__init__(host, "set", help=_("add or update a contact"))
+
+    def add_parser_options(self):
+        self.parser.add_argument("-n", "--name", help=_("name to use for this contact"))
+        self.parser.add_argument(
+            "-g",
+            "--group-names",
+            action="append",
+            metavar="GROUP_NAME",
+            default=[],
+            help=_(
+                "named of the groups this contact belong to (can be used several times)."
+            ),
+        )
+        self.parser.add_argument(
+            "-G",
+            "--group-ids",
+            action="append",
+            metavar="GROUP_ID",
+            default=[],
+            help=_(
+                "IDs of the groups this contact belong to (can be used several times)."
+            ),
+        )
+        self.parser.add_argument(
+            "--no-encrypt",
+            action="store_true",
+            help="store the contact in clear text instead of encrypting it.",
+        )
+        self.parser.add_argument("jid", help=_("jid of the roster entity"))
+
+    async def start(self):
+        groups = Groups([])
+        for name in self.args.group_names:
+            groups.append(Group(name=name))
+        for id_ in self.args.group_ids:
+            groups.append(Group(id=id_))
+
+        try:
+            item_id = await self.host.bridge.ps_contact_set(
+                ContactSetRequest(
+                    contact=Contact(
+                        name=self.args.name,
+                        identities={"jid": self.args.jid},
+                        groups=groups,
+                        encrypted=not self.args.no_encrypt,
+                    )
+                ).model_dump_json(),
+                self.host.profile,
+            )
+        except Exception as e:
+            self.disp(f"error while updating the contact: {e}", error=True)
+            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
+        else:
+            self.disp(f"Contact published successfully with ID {item_id}.")
+        self.host.quit()
+
+
+class ContactsCmd(base.CommandBase):
+    subcommands = (Get, Set)
+
+    def __init__(self, host):
+        super(ContactsCmd, self).__init__(
+            host, "contacts", use_profile=True, help=_("Manage contacts of a profile.")
+        )

libervia/cli/constants.py

@@ -97,6 +97,7 @@ class Const(constants.Const):
             "failure": "bold red",
             "file": "white",
             "header": "bold yellow",
+            "primary": "bright_cyan",
             "instructions": "italic",
             "level1": "bold blue",
             "level2": "magenta",