139 lines
3.4 KiB
Python
139 lines
3.4 KiB
Python
import requests
|
|
import time
|
|
import tomllib
|
|
from os import path
|
|
import json
|
|
|
|
TTL = 2 * 60 * 60 # 2 hours
|
|
|
|
if path.exists("config.toml"):
|
|
with open("config.toml", "rb") as f:
|
|
data = tomllib.load(f)
|
|
else:
|
|
raise RuntimeError("config.toml not found")
|
|
|
|
def getMediaListCollection(
|
|
mediaType=data["anilist"]["mediaType"],
|
|
userId=data["anilist"]["userId"],
|
|
*,
|
|
cachePath="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"]
|
|
|
|
if cacheExpired:
|
|
print("Cache has gone stale, refreshing...")
|
|
elif parametersChanged:
|
|
print("Cache parameters changed, refreshing...")
|
|
|
|
else:
|
|
print("Cache not found, 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"]:
|
|
for entry in lst["entries"]:
|
|
entries[entry["id"]] = entry
|
|
|
|
if not collection["hasNextChunk"]:
|
|
break
|
|
|
|
chunk += 1
|
|
|
|
form = {
|
|
"timestamp": str(time.time()),
|
|
"ttl": str(TTL),
|
|
"params": {
|
|
"mediaType": mediaType,
|
|
"userId": userId,
|
|
},
|
|
"response": list(entries.values()),
|
|
}
|
|
|
|
return form
|
|
|
|
if __name__ == "__main__":
|
|
entries = getMediaListCollection()
|
|
|
|
with open("data.json", "w") as f:
|
|
f.write(f"{json.dumps(entries, indent=2)}") |