-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
173 lines (150 loc) · 4.83 KB
/
app.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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
"""
# Project: github-music-status
# FilePath: /app.py
# File: app.py
# Created Date: Saturday, June 27th 2020, 2:12:09 pm
# Author: Craig Bojko (craig@pixelventures.co.uk)
# -----
# Last Modified: Sun Jul 12 2020
# Modified By: Craig Bojko
# -----
# Copyright (c) 2020 Pixel Ventures Ltd.
# ------------------------------------
# <<licensetext>>
"""
import os
import sys
import http.client
import json
import time
import datetime
# Add ./vendors to import locations
# Allows lambda to have vendors packaged to local dir
vendors = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(vendors, "./vendor"))
from termcolor import colored
from custom_logging import create_logger
LAST_FM_USERNAME = os.environ["LAST_FM_USERNAME"]
LAST_FM_API_TOKEN = os.environ["LAST_FM_TOKEN"]
LAST_FM_ORIGIN = "ws.audioscrobbler.com"
LAST_FM_PATH = "/2.0/?method=user.getrecenttracks&user={}&limit=3&api_key={}&format=json"\
.format(LAST_FM_USERNAME, LAST_FM_API_TOKEN)
GITHUB_TOKEN = os.environ["GITHUB_TOKEN"]
GITHUB_GRAPH_ORIGIN = "api.github.com"
GITHUB_GRAPH_PATH = "/graphql"
STATUS_TIMEOUT = 300
LOGGER = create_logger(level="INFO", colors=True)
def fetch_last_fm_status():
"""
Query last_FM and return a list of recent tracks
"""
data = {}
try:
connection = http.client.HTTPSConnection(LAST_FM_ORIGIN, 443)
connection.request("GET", LAST_FM_PATH)
response = connection.getresponse()
LOGGER(
"Last_FM: Status: {} Reason: {}".format(response.status, response.reason),
level="debug",
)
if response.status == 200:
data = response.read().decode()
connection.close()
return data
except http.client.HTTPException as error:
LOGGER("Error fetching Last_FM data", error, level="error")
return False
def push_github_status(status, expires=False):
"""
POST a status message to github
"""
expires_at_str = ""
if expires is not False:
expires_at_str = ('expiresAt: "{}"'.format(expires))
mutator = """mutation UpdateUserStatus {
changeUserStatus(
input: {
emoji: ":notes:"
message: \"""" + status + """\"
""" + expires_at_str + """
}
) {
clientMutationId
}
}"""
LOGGER("QUERY: ", mutator, level="debug")
try:
connection = http.client.HTTPSConnection(GITHUB_GRAPH_ORIGIN, 443)
headers = {
"Authorization": "Bearer %s" % GITHUB_TOKEN,
"User-Agent": "python bot",
"Content-Type": "application/json",
}
connection.request(
"POST",
GITHUB_GRAPH_PATH,
body=json.dumps({"query": mutator}),
headers=headers,
)
response = connection.getresponse()
data = response.read().decode()
connection.close()
LOGGER(
"Github: Status: {} Reason: {}".format(response.status, response.reason),
data,
level="debug",
)
return True
except http.client.HTTPException as error:
LOGGER("Error pushing Github graph data", error, level="error")
return error
def main(colors=True):
"""
Main function - run the app
"""
json_data = fetch_last_fm_status()
if json_data is False:
return False
data = json.loads(json_data)
tracks = data.get("recenttracks", {}).get("track")
now_playing = False
now_playing_track = None
for track in tracks:
try:
attr = track.get("@attr", {})
if attr and attr.get("nowplaying") == "true":
now_playing = True
now_playing_track = track
else: continue
except KeyError:
continue
now_playing_str = (
colored(now_playing, "green" if now_playing is True else "red")
if colors is True
else now_playing
)
if now_playing:
now_playing_track_data = "Listening to {}: {}".format(
now_playing_track.get("artist", {}).get("#text"),
now_playing_track.get("name", ""),
)
LOGGER("Now Playing: {}".format(now_playing_str), now_playing_track_data)
now = time.time()
in_five_minutes = now + STATUS_TIMEOUT
expiry_str = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(in_five_minutes))
push_github_status(now_playing_track_data, expires=expiry_str)
else:
LOGGER("Now Playing: {}".format(now_playing_str))
return now_playing_track
def lambda_handler(event, context):
"""
Handler function for AWS Lambda - initial entry point
"""
global LOGGER
LOGGER = create_logger(level="DEBUG", colors=False)
message = main(colors=False)
return {"now_playing": message}
if __name__ == "__main__":
main(colors=True)