Sweetbread/lair.moe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Activity

Refactoring

Browse Source
This commit is contained in:
Sweetbread
2026-02-07 18:39:58 +03:00
parent 0f5d1b5221
commit db11fabe1a
5 changed files with 120 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blueprints/risdeveau/modules/api/lb.py
+46 -24
View File
@@ -1,14 +1,31 @@
from flask import Flask, jsonify
import atexit
import re
from dataclasses import dataclass
from hashlib import md5
from json import dumps
from time import time
from urllib.parse import parse_qs, urlparse
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
import requests
from time import time
import atexit
import re
from urllib.parse import urlparse, parse_qs
from flask import Flask, jsonify
listens = {}
listening = {}
@dataclass
class Cache:
data = {}
last_updated = time()
status = None
data = {
"caches": {
"now": Cache(),
"listens": Cache()
},
"last_updated": time(),
"etag": ""
}
def yt_cover(youtube_url):
parsed_url = urlparse(youtube_url)
@@ -23,15 +40,15 @@ def yt_cover(youtube_url):
if not video_id:
return
return f"http://img.youtube.com/vi/{video_id}/sddefault.jpg"
return f"https://img.youtube.com/vi/{video_id}/sddefault.jpg"
def parse_listens(data: dict) -> dict:
new_data = {
"count": data["count"],
def parse_listens(json: dict) -> dict:
new_json = {
"count": json["count"],
"listens": []
}
for track in data["listens"]:
for track in json["listens"]:
track = track["track_metadata"]
new_track = {
@@ -53,33 +70,38 @@ def parse_listens(data: dict) -> dict:
if "cover_url" not in new_track.keys() and "id" in new_track.keys():
new_track["cover_url"] = "/asset/mb/" + new_track["id"]
new_data["listens"].append(new_track)
new_json["listens"].append(new_track)
return new_data
return new_json
def api_request(url: str, cache):
def api_request(url: str, cache: Cache):
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
cache.update({
'data': parse_listens(response.json().get("payload")),
'last_updated': time(),
'status': 'success'
})
json = parse_listens(response.json().get("payload"))
cache.status = 'success'
if cache.data != json:
cache.data = json
cache.last_updated = time()
data['last_updated'] = time()
data['etag'] = md5(''.join(
( dumps(data['caches'][x].data) for x in data['caches'] )
).encode()).hexdigest()
else:
cache['status'] = f'error: {response.status_code}'
cache.status = f'error: {response.status_code}'
except Exception as e:
cache['status'] = f'error: {str(e)}'
cache.status = f'error: {str(e)}'
scheduler = BackgroundScheduler()
scheduler.add_job(
func=lambda: api_request("https://api.listenbrainz.org/1/user/risdeveau/listens?count=5", listens),
func=lambda: api_request("https://api.listenbrainz.org/1/user/risdeveau/listens?count=5", data['caches']['listens']),
trigger=IntervalTrigger(minutes=1),
id='risdeveau.listenbrainz.listens',
replace_existing=True
)
scheduler.add_job(
func=lambda: api_request("https://api.listenbrainz.org/1/user/risdeveau/playing-now", listening),
func=lambda: api_request("https://api.listenbrainz.org/1/user/risdeveau/playing-now", data['caches']['now']),
trigger=IntervalTrigger(seconds=15),
id='risdeveau.listenbrainz.playing-now',
replace_existing=True
blueprints/risdeveau/modules/api/steam.py
+41 -18
View File
@@ -1,18 +1,35 @@
import atexit
import re
from dataclasses import dataclass
from hashlib import md5
from json import dumps
from os import environ
from flask import Flask, jsonify
from time import time
from urllib.parse import parse_qs, urlparse
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
import requests
from time import time
import atexit
import re
from urllib.parse import urlparse, parse_qs
from flask import Flask
TOKEN = environ.get("STEAM_TOKEN")
MY_ID = 76561198826355942
recent = {}
owned = {}
@dataclass
class Cache:
data = {}
last_updated = time()
status = None
data = {
"caches": {
"recent": Cache(),
"owned": Cache()
},
"last_updated": time(),
"etag": ""
}
def modify_game_list(json: dict) -> dict:
if 'games' in json.keys():
@@ -37,34 +54,40 @@ def api_request(cache, *args, **kwargs):
try:
response = steam_request(*args, **kwargs)
if response.status_code == 200:
cache.update({
'data': modify_game_list(response.json().get("response")),
'last_updated': time(),
'status': 'success'
})
json = modify_game_list(response.json().get("response"))
cache.status = 'success'
if cache.data != json:
cache.data = json
cache.last_updated = time()
data['last_updated'] = time()
data['etag'] = md5(''.join(
( dumps(data['caches'][x].data) for x in data['caches'] )
).encode()).hexdigest()
else:
cache['status'] = f'error: {response.status_code}'
cache.status = f'error: {response.status_code}'
print("x")
except Exception as e:
cache['status'] = f'error: {str(e)}'
cache.status = f'error: {str(e)}'
if TOKEN:
scheduler = BackgroundScheduler()
scheduler.add_job(
func=lambda: api_request(recent, "IPlayerService", "GetRecentlyPlayedGames", steamid=76561198826355942),
func=lambda: api_request(data['caches']['recent'], "IPlayerService", "GetRecentlyPlayedGames", steamid=76561198826355942),
trigger=IntervalTrigger(minutes=15),
id='risdeveau.steam.recent',
replace_existing=True
)
scheduler.add_job(
func=lambda: api_request(owned, "IPlayerService", "GetOwnedGames", steamid=76561198826355942, include_appinfo=1, include_played_free_games=1),
func=lambda: api_request(data['caches']['owned'], "IPlayerService", "GetOwnedGames", steamid=76561198826355942, include_appinfo=1, include_played_free_games=1),
trigger=IntervalTrigger(minutes=60),
id='risdeveau.steam.owned',
replace_existing=True
)
scheduler.start()
api_request(recent, "IPlayerService", "GetRecentlyPlayedGames", steamid=76561198826355942)
api_request(owned, "IPlayerService", "GetOwnedGames", steamid=76561198826355942, include_appinfo=1, include_played_free_games=1)
api_request(data['caches']['recent'], "IPlayerService", "GetRecentlyPlayedGames", steamid=76561198826355942)
api_request(data['caches']['owned'], "IPlayerService", "GetOwnedGames", steamid=76561198826355942, include_appinfo=1, include_played_free_games=1)
atexit.register(lambda: scheduler.shutdown())
else: