155 lines
4.1 KiB
Python
155 lines
4.1 KiB
Python
import requests
|
|
import time
|
|
import tomllib
|
|
from os import path
|
|
import json
|
|
|
|
if path.exists("config.toml"):
|
|
with open("config.toml", "rb") as f:
|
|
config = tomllib.load(f)
|
|
else:
|
|
raise RuntimeError("config.toml not found")
|
|
|
|
TTL = config["anilist"].get("TTL", 2 * 60 * 60) # 2 hour default if not specificed
|
|
|
|
def getMediaListCollection(
|
|
mediaType=config["anilist"]["mediaType"],
|
|
userId=config["anilist"]["userId"],
|
|
*,
|
|
cachePath="data.json"
|
|
):
|
|
"""
|
|
Returns a tuple of (entries, cached)
|
|
|
|
Entries is a dictionary of the values to be written to the cache file, such as:
|
|
- timestamp
|
|
- ttl
|
|
- params supplied
|
|
- response
|
|
|
|
cached is a boolean indicating whether the cache file was used or not
|
|
|
|
Takes the following arguments:
|
|
- mediaType: the media type to use for the request (if not supplied, defaults to value supplied from config.toml)
|
|
- userId: the user id to use for the request (if not supplied, defaults to value supplied from config.toml)
|
|
- cachePath: the path to the cache file (if not supplied, defaults to "data.json")
|
|
"""
|
|
|
|
if not mediaType or not userId:
|
|
raise RuntimeError("MediaType and UserId must be set")
|
|
|
|
if path.exists(cachePath):
|
|
with open(cachePath, "r") as f:
|
|
entries = json.load(f)
|
|
|
|
if (
|
|
"timestamp" in entries
|
|
and "ttl" in entries
|
|
and "response" in entries
|
|
and "params" in entries
|
|
):
|
|
timestamp = float(entries["timestamp"])
|
|
ttl = float(entries["ttl"])
|
|
|
|
cacheUserId = entries["params"].get("userId")
|
|
cacheMediaType = entries["params"].get("mediaType")
|
|
|
|
cacheExpired = time.time() - timestamp > ttl
|
|
parametersChanged = (
|
|
cacheUserId != userId
|
|
or cacheMediaType != mediaType
|
|
)
|
|
|
|
if not cacheExpired and not parametersChanged:
|
|
print("Cache is fresh, using cached data...")
|
|
return (entries["response"], True)
|
|
|
|
if cacheExpired:
|
|
print("Cache has gone stale, refreshing...")
|
|
elif parametersChanged:
|
|
print("Cache parameters changed, refreshing...")
|
|
else:
|
|
print("Cache is invalid, refreshing...")
|
|
|
|
# Fetch fresh data from AniList
|
|
query = """
|
|
query ($type: MediaType!, $userId: Int!, $chunk: Int, $perChunk: Int) {
|
|
MediaListCollection(
|
|
type: $type,
|
|
userId: $userId,
|
|
chunk: $chunk,
|
|
perChunk: $perChunk
|
|
) {
|
|
hasNextChunk
|
|
lists {
|
|
name
|
|
entries {
|
|
id
|
|
media {
|
|
id
|
|
title {
|
|
english
|
|
romaji
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
entries = {}
|
|
chunk = 1
|
|
|
|
while True:
|
|
variables = {
|
|
"type": mediaType,
|
|
"userId": int(userId),
|
|
"chunk": chunk,
|
|
"perChunk": 500,
|
|
}
|
|
|
|
resp = requests.post(
|
|
"https://graphql.anilist.co",
|
|
json={
|
|
"query": query,
|
|
"variables": variables,
|
|
},
|
|
timeout=30,
|
|
)
|
|
|
|
resp.raise_for_status()
|
|
|
|
data = resp.json()
|
|
|
|
if "errors" in data:
|
|
raise RuntimeError(data["errors"])
|
|
|
|
collection = data["data"]["MediaListCollection"]
|
|
|
|
for lst in collection["lists"]:
|
|
entries.setdefault(lst["name"], []).extend(lst["entries"])
|
|
|
|
if not collection["hasNextChunk"]:
|
|
break
|
|
|
|
chunk += 1
|
|
|
|
form = {
|
|
"timestamp": str(time.time()),
|
|
"ttl": str(TTL),
|
|
"params": {
|
|
"mediaType": mediaType,
|
|
"userId": userId,
|
|
},
|
|
"response": entries
|
|
}
|
|
|
|
return (form, False)
|
|
|
|
if __name__ == "__main__":
|
|
entries, cached = getMediaListCollection()
|
|
|
|
if not cached:
|
|
with open("data.json", "w") as f:
|
|
f.write(f"{json.dumps(entries, indent=2)}") |