stats.py (5121B)
1 """Public stats page: what is in the CDN (types, sizes, folders, growth). 2 3 Everything is computed from the file list storage.py already caches, then kept 4 for STATS_TTL seconds, so viewing the page never adds load on the bucket. 5 Only aggregate numbers are exposed - no IPs, no upload log. 6 """ 7 from __future__ import annotations 8 9 import asyncio 10 import statistics 11 import time 12 from collections import Counter, defaultdict 13 from datetime import datetime, timezone 14 15 from fastapi import APIRouter, Request 16 from fastapi.responses import JSONResponse 17 from .fast_html import FastTemplates 18 19 from . import storage 20 21 router = APIRouter() 22 templates = FastTemplates(directory="app/templates") 23 24 STATS_TTL = 600 25 _cache: dict = {"at": 0.0, "data": None} 26 _lock = asyncio.Lock() 27 28 CATEGORIES = [ 29 ("images", "Images", "png jpg jpeg gif webp avif svg ico bmp heic tiff"), 30 ("documents", "Documents", "pdf doc docx ppt pptx xls xlsx odt ods odp txt md rtf csv epub"), 31 ("code", "Code & data", "js mjs css html htm json xml yml yaml toml py ts tsx jsx c cpp h java go rs sh sql map wasm"), 32 ("audio", "Audio", "mp3 wav ogg flac m4a aac opus"), 33 ("video", "Video", "mp4 mkv webm mov avi m4v"), 34 ("archives", "Archives", "zip tar gz tgz 7z rar xz bz2"), 35 ("fonts", "Fonts", "woff woff2 ttf otf eot"), 36 ] 37 _EXT_TO_CAT = {e: (k, label) for k, label, exts in CATEGORIES for e in exts.split()} 38 SIZE_BUCKETS = [(10 * 1024, "< 10 KB"), (100 * 1024, "10-100 KB"), (1024 ** 2, "100 KB-1 MB"), 39 (10 * 1024 ** 2, "1-10 MB"), (100 * 1024 ** 2, "10-100 MB"), (float("inf"), "> 100 MB")] 40 TOP_LEVEL_FOLDERS = {"uploads", "third-party"} 41 42 43 def _ext(name: str) -> str: 44 return name.rsplit(".", 1)[-1].lower() if "." in name.strip(".") else "" 45 46 47 def _folder(path: str) -> str: 48 parts = path.split("/") 49 if len(parts) == 1: 50 return "(root)" 51 return parts[0] if parts[0] in TOP_LEVEL_FOLDERS else "/".join(parts[:2]) if len(parts) > 2 else parts[0] 52 53 54 def compute(items: list[dict]) -> dict: 55 files = [i for i in items if not i["is_dir"] and i["name"] != ".gitattributes"] 56 if not files: 57 return {"empty": True} 58 59 sizes = [f["size"] for f in files] 60 cats: dict[str, dict] = {} 61 exts: dict[str, list] = defaultdict(lambda: [0, 0]) 62 folders: dict[str, list] = defaultdict(lambda: [0, 0]) 63 buckets = Counter() 64 months: dict[str, list] = defaultdict(lambda: [0, 0]) 65 66 for f in files: 67 ext = _ext(f["name"]) 68 key, label = _EXT_TO_CAT.get(ext, ("other", "Other")) 69 c = cats.setdefault(key, {"key": key, "label": label, "files": 0, "bytes": 0}) 70 c["files"] += 1 71 c["bytes"] += f["size"] 72 exts[ext or "(none)"][0] += 1 73 exts[ext or "(none)"][1] += f["size"] 74 fo = folders[_folder(f["path"])] 75 fo[0] += 1 76 fo[1] += f["size"] 77 buckets[next(lbl for limit, lbl in SIZE_BUCKETS if f["size"] < limit)] += 1 78 if f.get("ts"): 79 m = months[datetime.fromtimestamp(f["ts"], timezone.utc).strftime("%Y-%m")] 80 m[0] += 1 81 m[1] += f["size"] 82 83 top_ext = sorted(exts.items(), key=lambda kv: kv[1][0], reverse=True) 84 top_folders = sorted(folders.items(), key=lambda kv: kv[1][1], reverse=True) 85 largest = sorted(files, key=lambda f: f["size"], reverse=True)[:8] 86 month_rows = [{"m": k, "files": v[0], "bytes": v[1]} for k, v in sorted(months.items())][-12:] 87 88 return { 89 "empty": False, 90 "generated": int(time.time()), 91 "totals": { 92 "files": len(files), "bytes": sum(sizes), "avg": int(sum(sizes) / len(sizes)), 93 "median": int(statistics.median(sizes)), "types": len(exts), 94 }, 95 "categories": sorted(cats.values(), key=lambda c: c["files"], reverse=True), 96 "extensions": [{"ext": e, "files": v[0], "bytes": v[1]} for e, v in top_ext[:10]], 97 "extensions_more": sum(v[0] for _, v in top_ext[10:]), 98 "sizes": [{"label": lbl, "files": buckets.get(lbl, 0)} for _, lbl in SIZE_BUCKETS], 99 "folders": [{"name": n, "files": v[0], "bytes": v[1]} for n, v in top_folders[:8]], 100 "largest": [{"name": f["name"], "path": f["path"], "size": f["size"]} for f in largest], 101 "months": month_rows, 102 } 103 104 105 async def get_stats() -> dict: 106 now = time.time() 107 if _cache["data"] is not None and now - _cache["at"] < STATS_TTL: 108 return _cache["data"] 109 async with _lock: 110 if _cache["data"] is not None and time.time() - _cache["at"] < STATS_TTL: 111 return _cache["data"] 112 items = await storage._get_full_tree() 113 if not items: # listing failed: keep serving the last good result if we have one 114 return _cache["data"] or {"empty": True} 115 data = await asyncio.to_thread(compute, items) 116 _cache.update(at=time.time(), data=data) 117 return data 118 119 120 @router.get("/stats") 121 async def stats_page(request: Request): 122 return templates.TemplateResponse(request, "stats.html", {"page": "stats"}) 123 124 125 @router.get("/api/stats") 126 async def stats_api(): 127 return JSONResponse(await get_stats(), headers={"Cache-Control": "public, max-age=300"})