178 lines
4.5 KiB
Python
178 lines
4.5 KiB
Python
import json
|
|
import aiohttp
|
|
import asyncio
|
|
import sys
|
|
import select
|
|
from datetime import datetime
|
|
|
|
# Read DEV-README.md to have example of requests and data returned
|
|
|
|
|
|
stop = False
|
|
tasks = []
|
|
|
|
|
|
def indent(string, spaces=4):
|
|
return (' ' * spaces) + string + '\n'
|
|
|
|
|
|
class ValheimNewsJSONDecoder(json.JSONDecoder):
|
|
|
|
def decode(dct):
|
|
value = dct
|
|
if "gid" in dct:
|
|
value = ValheimNewsItem(dct)
|
|
elif "newsitems" in dct:
|
|
value = ValheimNewsItems(dct)
|
|
elif "appnews" in dct:
|
|
value = ValheimNews(dct)
|
|
return value
|
|
|
|
|
|
class ValheimNewsBase:
|
|
KEYS = []
|
|
|
|
def __init__(self, dct):
|
|
for key, value in dct.items():
|
|
if key in self.KEYS:
|
|
setattr(self, key, value)
|
|
|
|
def to_string(self, spaces=0, in_list=False):
|
|
string = ""
|
|
content = list(vars(self).items())
|
|
if in_list:
|
|
(key, value), content = content[0], content[1:]
|
|
string += indent("- {}: {}".format(key, value), spaces - 2)
|
|
for key, value in content:
|
|
string += indent("{}: {}".format(key, value), spaces)
|
|
return string
|
|
|
|
def __repr__(self):
|
|
return self.to_string()
|
|
|
|
|
|
class ValheimNews(ValheimNewsBase):
|
|
KEYS = [
|
|
"appnews"
|
|
]
|
|
|
|
def to_string(self, spaces=0):
|
|
return "appnews: \n{}".format(self.appnews.to_string(spaces + 4))
|
|
|
|
|
|
class ValheimNewsItems(ValheimNewsBase):
|
|
|
|
KEYS = [
|
|
"appid",
|
|
"newsitems",
|
|
"count"
|
|
]
|
|
|
|
def to_string(self, spaces=0):
|
|
newsitems = ""
|
|
for newsitem in self.newsitems:
|
|
newsitems += newsitem.to_string(spaces + 4, True)
|
|
return \
|
|
indent("appid: {}".format(self.appid)) + \
|
|
indent("newsitems: \n{}".format(newsitems)) + \
|
|
indent("count: {}".format(self.count))
|
|
|
|
|
|
class ValheimNewsItem(ValheimNewsBase):
|
|
|
|
KEYS = [
|
|
"gid",
|
|
"title",
|
|
"url",
|
|
"is_external_url",
|
|
"author",
|
|
"contents",
|
|
"feedlabel",
|
|
"date",
|
|
"feedname",
|
|
"feed_type",
|
|
"appid"
|
|
]
|
|
|
|
|
|
async def getnews(count=2, maxlength=15):
|
|
url = "https://api.steampowered.com/ISteamNews/GetNewsForApp/v0002/?" \
|
|
"appid={}&count={}&maxlength={}&format={}" \
|
|
.format(892970, count, maxlength, "json")
|
|
|
|
print("Fetching {}".format(url))
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
print("Checking news... {}".format(response.status))
|
|
if (response.status != 200):
|
|
return
|
|
if (response.headers['content-type'].split(";")[0]
|
|
!= "application/json"):
|
|
return
|
|
result = json.loads(
|
|
await response.text(),
|
|
object_hook=ValheimNewsJSONDecoder.decode
|
|
)
|
|
return result
|
|
|
|
|
|
async def get_latest_dev_news():
|
|
news = await getnews(count=8, maxlength=140)
|
|
for newsitem in news.appnews.newsitems:
|
|
if newsitem.feed_type == 1:
|
|
return newsitem
|
|
|
|
|
|
async def get_latest_update():
|
|
url = "https://steamdb.info/depot/892971"
|
|
headers = {'User-Agent': 'curl/7.77.0'}
|
|
|
|
async with aiohttp.ClientSession(headers=headers) as session:
|
|
async with session.get(url) as response:
|
|
print("Checking latest update... {}".format(response.status))
|
|
if (response.status != 200):
|
|
return
|
|
response_strings = (await response.text()).split("\n")
|
|
last_update_found = False
|
|
for line in response_strings:
|
|
if last_update_found:
|
|
start = line.index("datetime") + 10
|
|
date = datetime.fromisoformat(line[start:start + 25])
|
|
return date
|
|
if "Last update" in line:
|
|
last_update_found = True
|
|
|
|
|
|
async def check_for_news(delay=5):
|
|
global stop
|
|
while not stop:
|
|
await get_latest_update()
|
|
await asyncio.sleep(5)
|
|
|
|
|
|
async def do_nothing():
|
|
global stop
|
|
while not stop:
|
|
i, o, e = select.select([sys.stdin], [], [], 0)
|
|
if i:
|
|
for task in tasks:
|
|
task.cancel()
|
|
await asyncio.sleep(0.1)
|
|
|
|
|
|
async def main():
|
|
global tasks
|
|
tasks = [
|
|
asyncio.create_task(do_nothing()),
|
|
asyncio.create_task(check_for_news()),
|
|
]
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.get_event_loop().run_until_complete(main())
|
|
except asyncio.CancelledError:
|
|
print("Goodbye!")
|