Starred Gists
Starred Gists订阅

user_metrics.py

#!/bin/env python3
import requests
import os
from typing import Literal
from tqdm import tqdm
from dotenv import load_dotenv
load_dotenv()
USER_ID = int(os.environ["DANBOORU_USER_ID"])
def api_request(kind: Literal["get", "put"], endpoint: str, *, params: dict = {}, headers: dict = {}, json: dict = {}, data: dict = {}) -> dict | list:
url = f"https://danbooru.donmai.us/{endpoint}.json"
r = requests.request(kind, url,
auth=(os.environ["DANBOORU_LOGIN"], os.environ["DANBOORU_API_KEY"]),
params=params if len(params) > 0 else None,
headers={ "User-Agent": f"user #{USER_ID}", **headers },
json=json if len(json) > 0 else None,
data=data if len(data) > 0 else None
)
return r.json()
def api_get(endpoint: str, *, params: dict = {}, headers: dict = {}, json: dict = {}, data: dict = {}) -> dict | list:
return api_request("get", endpoint, params=params, headers=headers, json=json, data=data)
print(f"Fetching metrics for user #{USER_ID}")
params: dict[str, str | int] = {
"limit": 1000,
"only": "id,status,media_asset_count,media_assets[id,file_ext,file_size,image_width,image_height,post[uploader_id]]",
"search[uploader_id]": USER_ID
}
def format_bytes(n: int) -> str:
if n < 1024:
return f"{n} bytes"
elif n < 1024*1024:
return f"{n / 1024:.3f} KiB"
elif n < 1024*1024*1024:
return f"{n / 1024 / 1024:.3f} MiB"
elif n < 1024*1024*1024*1024:
return f"{n / 1024 / 1024 / 1024:.3f} GiB"
else:
return f"{n / 1024 / 1024 / 1024 / 1024:.3f} TiB"
def format_pixels(n: int) -> str:
if n < 1e3:
return f"{n}"
elif n < 1e6:
return f"{n / 1e3:.3f} KP"
elif n < 1e9:
return f"{n / 1e6:.3f} MP"
elif n < 1e12:
return f"{n / 1e9:.3f} GP"
else:
return f"{n / 1e12:.3f} TP"
class Stats:
pixels: int
files: dict[str, int]
bytes: int
def __init__(self):
self.pixels = 0
self.files = {}
self.bytes = 0
def add_pixels(self, w: int, h: int):
self.pixels += w * h
def add_ext(self, ext: str):
if ext not in self.files:
self.files[ext] = 1
else:
self.files[ext] += 1
def add_bytes(self, n: int):
self.bytes += n
def update(self, w: int, h: int, ext: str, size: int):
self.add_pixels(w, h)
self.add_ext(ext)
self.add_bytes(size)
def update_asset(self, asset: dict):
self.update(asset["image_width"], asset["image_height"], asset["file_ext"], asset["file_size"])
def dump(self, title: str):
print(f"{title}:")
print(" ", self.pixels, f"pixels ({format_pixels(self.pixels)})")
print(" ", self.bytes, f"bytes ({format_bytes(self.bytes)})")
print(" File formats:")
for ext, n in sorted(self.files.items(), key=lambda kv: kv[1], reverse=True):
print(f" {ext}: {n}")
print()
total_uploads = 0
total_assets = 0
statuses = {
"pending": 0,
"processing": 0,
"completed": 0,
"error": 0
}
self_upload = Stats()
other_upload = Stats()
not_uploaded = Stats()
total = Stats()
all_assets = set()
with tqdm() as progress:
cur = 0
while True:
params["page"] = f"a{cur}"
res = api_get("uploads", params=params)
if len(res) == 0:
break
total_uploads += len(res)
for upload in res:
cur = max(cur, upload["id"])
total_assets += upload["media_asset_count"]
statuses[upload["status"]] += 1
try:
for asset in upload["media_assets"]:
if asset["id"] not in all_assets:
all_assets.add(asset["id"])
else:
continue
if "post" in asset:
if asset["post"]["uploader_id"] == USER_ID:
self_upload.update_asset(asset)
total.update_asset(asset)
else:
other_upload.update_asset(asset)
else:
not_uploaded.update_asset(asset)
total.update_asset(asset)
except KeyError:
print(f"Exploded on https://danbooru.donmai.us/uploads/{upload["id"]}.json")
raise
progress.update()
progress.display()
print()
print(total_assets, "assets,", len(all_assets), "unique assets")
print(total_uploads, "uploads")
for ext, n in sorted(statuses.items(), key=lambda kv: kv[1], reverse=True):
print(f" {ext}: {n}")
print()
self_upload.dump("Uploaded")
other_upload.dump("Uploaded by someone else")
not_uploaded.dump("Not uploaded")
total.dump("Total")
参与讨论