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
|
from __future__ import annotations
from typing import TYPE_CHECKING
from nbxmpp.modules.base import BaseModule
from nbxmpp.namespaces import Namespace
from nbxmpp.protocol import NodeProcessed
from nbxmpp.protocol import Presence
from nbxmpp.structs import Hat
from nbxmpp.structs import HatData
from nbxmpp.structs import PresenceProperties
from nbxmpp.structs import StanzaHandler
if TYPE_CHECKING:
from nbxmpp.client import Client
class Hats(BaseModule):
def __init__(self, client: Client) -> None:
BaseModule.__init__(self, client)
self._client = client
self.handlers = [
StanzaHandler(
name="presence",
callback=self._process_hats,
ns=Namespace.HATS,
priority=15,
),
StanzaHandler(
name="presence",
callback=self._process_hats,
ns=Namespace.HATS_LEGACY,
priority=15,
),
]
def _process_hats(
self,
_client: Client,
stanza: Presence,
properties: PresenceProperties,
) -> None:
hats = stanza.getTag("hats", namespace=Namespace.HATS)
if hats is None:
hats = stanza.getTag("hats", namespace=Namespace.HATS_LEGACY)
if hats is None:
return
hat_data = HatData()
for hat in hats.getTags("hat"):
uri = hat.getAttr("uri")
title = hat.getAttr("title")
if not uri or not title:
self._log.warning("Invalid hat received")
self._log.warning(stanza)
raise NodeProcessed
hue_str = hat.getAttr("hue")
if hue_str is None:
hue: float | None = None
else:
try:
hue = float(hue_str)
except ValueError:
self._log.warning("Invalid hue value: %s", hue_str)
hue = None
lang = hat.getAttr("xml:lang")
hat_data.add_hat(Hat(uri, title, hue), lang)
properties.hats = hat_data
|
