view sat_frontends/jp/cmd_blocking.py @ 3922:0ff265725489

plugin XEP-0447: handle attachment and download: - plugin XEP-0447 can now be used in message attachments and to retrieve an attachment - plugin attach: `attachment` being processed is added to `extra` so the handler can inspect it - plugin attach: `size` is added to attachment - plugin download: a whole attachment dict is now used in `download` and `file_download`/`file_download_complete`. `download_uri` can be used as a shortcut when just a URI is used. In addition to URI scheme handler, whole attachment handlers can now be registered with `register_download_handler` - plugin XEP-0363: `file_http_upload` `XEP-0363_upload_size` triggers have been renamed to `XEP-0363_upload_pre_slot` and is now using a dict with arguments, allowing for the size but also the filename to be modified, which is necessary for encryption (filename may be hidden from URL this way). - plugin XEP-0446: fix wrong element name - plugin XEP-0447: source handler can now be registered (`url-data` is registered by default) - plugin XEP-0447: source parsing has been put in a separated `parse_sources_elt` method, as it may be useful to do it independently (notably with XEP-0448) - plugin XEP-0447: parse received message and complete attachments when suitable - plugin XEP-0447: can now be used with message attachments - plugin XEP-0447: can now be used with attachments download - renamed `options` arguments to `extra` for consistency - some style change (progressive move from legacy camelCase to PEP8 snake_case) - some typing rel 379
author Goffi <goffi@goffi.org>
date Thu, 06 Oct 2022 16:02:05 +0200
parents 12317ba98d99
children 524856bd7b19
line wrap: on
line source

#!/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 json
import os
from sat.core.i18n import _
from sat.tools.common import data_format
from sat_frontends.jp import common
from sat_frontends.jp.constants import Const as C
from . import base

__commands__ = ["Blocking"]


class List(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "list",
            use_output=C.OUTPUT_LIST,
            help=_("list blocked entities"),
        )

    def add_parser_options(self):
        pass

    async def start(self):
        try:
            blocked_jids = await self.host.bridge.blockingList(
                self.profile,
            )
        except Exception as e:
            self.disp(f"can't get blocked entities: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            await self.output(blocked_jids)
            self.host.quit(C.EXIT_OK)


class Block(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "block",
            help=_("block one or more entities"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "entities",
            nargs="+",
            metavar="JID",
            help=_("JIDs of entities to block"),
        )

    async def start(self):
        try:
            await self.host.bridge.blockingBlock(
                self.args.entities,
                self.profile
            )
        except Exception as e:
            self.disp(f"can't block entities: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.host.quit(C.EXIT_OK)


class Unblock(base.CommandBase):
    def __init__(self, host):
        super().__init__(
            host,
            "unblock",
            help=_("unblock one or more entities"),
        )

    def add_parser_options(self):
        self.parser.add_argument(
            "entities",
            nargs="+",
            metavar="JID",
            help=_("JIDs of entities to unblock"),
        )
        self.parser.add_argument(
            "-f",
            "--force",
            action="store_true",
            help=_('when "all" is used, unblock all entities without confirmation'),
        )

    async def start(self):
        if self.args.entities == ["all"]:
            if not self.args.force:
                await self.host.confirmOrQuit(
                    _("All entities will be unblocked, are you sure"),
                    _("unblock cancelled")
                )
            self.args.entities.clear()
        elif self.args.force:
            self.parser.error(_('--force is only allowed when "all" is used as target'))

        try:
            await self.host.bridge.blockingUnblock(
                self.args.entities,
                self.profile
            )
        except Exception as e:
            self.disp(f"can't unblock entities: {e}", error=True)
            self.host.quit(C.EXIT_BRIDGE_ERRBACK)
        else:
            self.host.quit(C.EXIT_OK)


class Blocking(base.CommandBase):
    subcommands = (List, Block, Unblock)

    def __init__(self, host):
        super().__init__(
            host, "blocking", use_profile=False, help=_("entities blocking")
        )