1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
"""
Slixmpp: The Slick XMPP Library
Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
import logging
from asyncio import Future
from typing import (
List,
Optional,
Set,
Union,
)
from slixmpp.stanza import Iq
from slixmpp.plugins import BasePlugin
from slixmpp.xmlstream.handler import Callback
from slixmpp.xmlstream.matcher import StanzaPath
from slixmpp.xmlstream import register_stanza_plugin, JID
from slixmpp.plugins.xep_0191 import stanza, Block, Unblock, BlockList
log = logging.getLogger(__name__)
BlockedJIDs = Union[
JID,
Set[JID],
List[JID]
]
class XEP_0191(BasePlugin):
name = 'xep_0191'
description = 'XEP-0191: Blocking Command'
dependencies = {'xep_0030'}
stanza = stanza
def plugin_init(self):
register_stanza_plugin(Iq, BlockList)
register_stanza_plugin(Iq, Block)
register_stanza_plugin(Iq, Unblock)
self.xmpp.register_handler(
Callback('Blocked Contact',
StanzaPath('iq@type=set/block'),
self._handle_blocked))
self.xmpp.register_handler(
Callback('Unblocked Contact',
StanzaPath('iq@type=set/unblock'),
self._handle_unblocked))
def plugin_end(self):
self.xmpp.remove_handler('Blocked Contact')
self.xmpp.remove_handler('Unblocked Contact')
def get_blocked(self, ifrom: Optional[JID] = None, **iqkwargs) -> Future:
"""Get the list of blocked JIDs."""
iq = self.xmpp.make_iq_get(ifrom=ifrom)
iq.enable('blocklist')
return iq.send(**iqkwargs)
def block(self, jids: BlockedJIDs,
ifrom: Optional[JID] = None, **iqkwargs) -> Future:
"""Block a JID or a list of JIDs.
:param jids: JID(s) to block.
"""
iq = self.xmpp.make_iq_set(ifrom=ifrom)
if not isinstance(jids, (set, list)):
jids = [jids]
iq['block']['items'] = jids
return iq.send(**iqkwargs)
def unblock(self, jids: BlockedJIDs, ifrom: Optional[JID] = None, **iqkwargs) -> Future:
"""Unblock a JID or a list of JIDs.
:param jids: JID(s) to unblock.
"""
if jids is None:
raise ValueError("jids cannot be empty.")
iq = self.xmpp.make_iq_set(ifrom=ifrom)
if not isinstance(jids, (set, list)):
jids = [jids]
iq['unblock']['items'] = jids
return iq.send(**iqkwargs)
def _handle_blocked(self, iq):
self.xmpp.event('blocked', iq)
def _handle_unblocked(self, iq):
self.xmpp.event('unblocked', iq)
|
