forked from wichmannpas/uberspace-downtime-notify
-
Notifications
You must be signed in to change notification settings - Fork 0
/
fetch.py
executable file
·93 lines (72 loc) · 2.42 KB
/
fetch.py
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
#!/usr/bin/env python3
"""
Simple twitter page fetcher.
A simple script which checks the twitter account of uberspace
(or any other twitter account as well) for messages containing
specific keywords.
"""
import hashlib
import os
from HTMLParser import HTMLParser
html = HTMLParser()
import re
import requests
import subprocess
RELEVANT_HOSTS = [
'alpheca',
]
#NOTIFY_COMMAND = 'cat; echo subject: {}'
NOTIFY_COMMAND = 'mail -s "{}" $USER'
URL = 'https://twitter.com/ubernauten'
# hashes of all messages stored in STORE_FILE to prevent duplicates
STORE_FILE = os.path.join(os.getcwd(), 'notified')
def main():
messages = get_messages()
def get_messages():
"""Get a list containing the plain-text messages."""
page = requests.get(URL)
pattern = re.compile(r'<div class="js-tweet-text-container">[^>]*>(.*?)</p>.*?</div>', re.DOTALL)
messages = pattern.findall(page.text)
# unescape
messages = (clean_message(message) for message in messages)
for message in messages:
host = check_relevance(message)
if host:
send_notification(host, message)
def send_notification(host, message):
"""Send a notification message."""
if check_for_done_notification(message):
# notification has already been sent
return
subject = 'New message regarding {}'.format(host)
subprocess.Popen(
NOTIFY_COMMAND.format(subject), shell=True,
stdin=subprocess.PIPE).communicate(message.encode())
def check_for_done_notification(message):
"""Check whether a notification message has been sent already and store
the hash."""
hash = hashlib.sha256(message.encode()).hexdigest()
if hash in get_stored_hashes():
return True
add_stored_hash(hash)
return False
def add_stored_hash(hash):
with open(STORE_FILE, 'a') as file:
file.write('{}\n'.format(hash))
def get_stored_hashes():
if not os.path.isfile(STORE_FILE):
return []
with open(STORE_FILE, 'r') as file:
return file.read().split('\n')
def check_relevance(message):
"""Check if a message is relevant and return host."""
for host in RELEVANT_HOSTS:
if host.lower() in message.lower():
return host
def clean_message(message):
"""Html-unescape and strip html from a message."""
message = html.unescape(message)
pattern = re.compile(r'<[^>]*>')
message = pattern.sub('', message)
return message
if __name__ == '__main__':