|
1
|
|
|
""" |
|
2
|
|
|
byceps.util.irc |
|
3
|
|
|
~~~~~~~~~~~~~~~ |
|
4
|
|
|
|
|
5
|
|
|
Send IRC messages to a bot via HTTP. |
|
6
|
|
|
|
|
7
|
|
|
:Copyright: 2006-2020 Jochen Kupperschmidt |
|
8
|
|
|
:License: Modified BSD, see LICENSE for details. |
|
9
|
|
|
""" |
|
10
|
|
|
|
|
11
|
1 |
|
from time import sleep |
|
12
|
1 |
|
from typing import Optional |
|
13
|
|
|
|
|
14
|
1 |
|
from flask import current_app |
|
15
|
1 |
|
import requests |
|
16
|
|
|
|
|
17
|
1 |
|
from ..services.global_setting import service as global_settings_service |
|
18
|
|
|
|
|
19
|
|
|
|
|
20
|
1 |
|
def send_message(channel: str, text: str) -> None: |
|
21
|
|
|
"""Write the text to the channel by sending it to the bot via HTTP.""" |
|
22
|
1 |
|
if not _is_enabled(): |
|
23
|
1 |
|
current_app.logger.warning('Announcements on IRC are disabled.') |
|
24
|
1 |
|
return |
|
25
|
|
|
|
|
26
|
1 |
|
url = _get_webhook_url() |
|
27
|
1 |
|
if not url: |
|
28
|
|
|
current_app.logger.warning( |
|
29
|
|
|
'No webhook URL configured for announcements on IRC.' |
|
30
|
|
|
) |
|
31
|
|
|
return |
|
32
|
|
|
|
|
33
|
1 |
|
text_prefix = _get_text_prefix() |
|
34
|
1 |
|
if text_prefix: |
|
35
|
|
|
text = text_prefix + text |
|
36
|
|
|
|
|
37
|
1 |
|
data = {'channel': channel, 'text': text} |
|
38
|
|
|
|
|
39
|
|
|
# Delay a bit as an attempt to avoid getting kicked from server |
|
40
|
|
|
# because of flooding. |
|
41
|
1 |
|
delay = _get_delay() |
|
42
|
1 |
|
sleep(delay) |
|
43
|
|
|
|
|
44
|
1 |
|
requests.post(url, json=data) # Ignore response code for now. |
|
45
|
|
|
|
|
46
|
|
|
|
|
47
|
1 |
|
def _is_enabled() -> bool: |
|
48
|
|
|
"""Return `true' if announcements on IRC are enabled. |
|
49
|
|
|
|
|
50
|
|
|
Disabled by default. |
|
51
|
|
|
""" |
|
52
|
1 |
|
value = global_settings_service.find_setting_value('announce_irc_enabled') |
|
53
|
1 |
|
return value == 'true' |
|
54
|
|
|
|
|
55
|
|
|
|
|
56
|
1 |
|
def _get_webhook_url() -> Optional[str]: |
|
57
|
|
|
"""Return the configured webhook URL.""" |
|
58
|
1 |
|
return global_settings_service.find_setting_value( |
|
59
|
|
|
'announce_irc_webhook_url' |
|
60
|
|
|
) |
|
61
|
|
|
|
|
62
|
|
|
|
|
63
|
1 |
|
def _get_text_prefix() -> Optional[str]: |
|
64
|
|
|
"""Return the configured text prefix.""" |
|
65
|
1 |
|
return global_settings_service.find_setting_value( |
|
66
|
|
|
'announce_irc_text_prefix' |
|
67
|
|
|
) |
|
68
|
|
|
|
|
69
|
|
|
|
|
70
|
1 |
|
def _get_delay(default=2) -> int: |
|
71
|
|
|
"""Return the configured delay, in seconds.""" |
|
72
|
1 |
|
value = global_settings_service.find_setting_value('announce_irc_delay') |
|
73
|
1 |
|
if value is None: |
|
74
|
|
|
return default |
|
75
|
|
|
|
|
76
|
1 |
|
try: |
|
77
|
1 |
|
value = int(value) |
|
78
|
|
|
except (TypeError, ValueError) as e: |
|
79
|
|
|
current_app.logger.warning( |
|
80
|
|
|
f'Invalid delay value configured for announcements on IRC: {e}' |
|
81
|
|
|
) |
|
82
|
|
|
return default |
|
83
|
|
|
|
|
84
|
|
|
return value |
|
85
|
|
|
|
byceps /
byceps
