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
|
# Copyright (C) 2018 Philipp Hörist <philipp AT hoerist.com>
#
# This file is part of nbxmpp.
#
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING
from nbxmpp.modules.base import BaseModule
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import isMucPM
from nbxmpp.protocol import Message
from nbxmpp.structs import MessageProperties
from nbxmpp.structs import ReceiptData
from nbxmpp.structs import StanzaHandler
from nbxmpp.util import generate_id
if TYPE_CHECKING:
from nbxmpp.client import Client
class Receipts(BaseModule):
def __init__(self, client: Client) -> None:
BaseModule.__init__(self, client)
self._client = client
self.handlers = [
StanzaHandler(
name="message",
callback=self._process_message_receipt,
ns=Namespace.RECEIPTS,
priority=15,
),
]
def _process_message_receipt(
self, _client: Client, stanza: Message, properties: MessageProperties
) -> None:
request = stanza.getTag("request", namespace=Namespace.RECEIPTS)
if request is not None:
properties.receipt = ReceiptData(request.getName())
return
received = stanza.getTag("received", namespace=Namespace.RECEIPTS)
if received is not None:
id_ = received.getAttr("id")
if id_ is None:
self._log.warning("Receipt without id attr")
self._log.warning(stanza)
return
properties.receipt = ReceiptData(received.getName(), id_)
def build_receipt(stanza: Message | Any) -> Message:
if not isinstance(stanza, Message):
raise ValueError("Stanza type must be protocol.Message")
if stanza.getType() == "error":
raise ValueError("Receipt can not be generated for type error messages")
if stanza.getID() is None:
raise ValueError("Receipt can not be generated for messages without id")
if stanza.getTag("received", namespace=Namespace.RECEIPTS) is not None:
raise ValueError("Receipt can not be generated for receipts")
is_muc_pm = isMucPM(stanza)
jid = stanza.getFrom()
typ = stanza.getType()
if typ == "groupchat" or not is_muc_pm:
jid = jid.new_as_bare()
message = Message(to=jid, typ=typ)
if is_muc_pm:
message.setTag("x", namespace=Namespace.MUC_USER)
message_id = generate_id()
message.setID(message_id)
message.setReceiptReceived(stanza.getID())
message.setHint("store")
message.setOriginID(message_id)
return message
|
