libervia-backend/libervia/cli/cmd_contacts.py

#!/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.")
        )