Add /donors endpoint totalling up per-donor value

Latent-Logic · May 5, 2024 · 314a2cd · 314a2cd
1 parent 3f57dba
commit 314a2cd
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 1 deletion.
diff --git a/test_tracker.py b/test_tracker.py
@@ -643,6 +643,30 @@ async def get_events(timezone: Optional[str] = None):
     return f"<html><head><style>{style}</style></head><body>{build_table}</body></html>"
 
 
+@app.get("/donors", response_class=HTMLResponse)
+async def get_donors():
+    donor_db = {}
+    with Path(SETTINGS.db.events).open("r", encoding="utf-8") as f:
+        reader = csv.DictReader(f, delimiter=",")
+        assert reader.fieldnames == CSV_COLUMNS
+        for row in reader:
+            user_db = donor_db.setdefault(
+                row["user"].lower(), {"name": row["user"], "total": 0, **{k: 0 for k in CSV_TYPES}}
+            )
+            amount = float(row["amount"]) if row["type"] == TIPS else int(row["amount"])
+            user_db[row["type"]] += amount
+            user_db["total"] += amount * SETTINGS.get_value(row["type"])
+
+    build_table = "<table>\n"
+    build_table += "<tr>" + "".join(f"<th>{s}</th>" for s in ("name", "total", *CSV_TYPES)) + "</tr>\n"
+    for row in sorted(donor_db.values(), key=lambda x: x["total"], reverse=True):
+        build_table += "<tr>" + "".join(f"<td>{row[s]}</td>" for s in ("name", "total", *CSV_TYPES)) + "</tr>\n"
+    build_table += "</table>\n"
+    style = """table {border: 2px solid rgb(140 140 140);}
+    th,td {border: 1px solid rgb(160 160 160);}"""
+    return f"<html><head><style>{style}</style></head><body>{build_table}</body></html>"
+
+
 websocket_html = """
 <!DOCTYPE html>
 <html>

diff --git a/twitch_dono_clock/config.py b/twitch_dono_clock/config.py
@@ -1,6 +1,6 @@
 import re
 from datetime import datetime
-from typing import Dict, List, Tuple
+from typing import Dict, List, Tuple, Union
 
 import toml
 from pydantic import BaseModel, Field, model_validator
@@ -122,6 +122,18 @@ def compile_regex(self):
     def compiled_re(self) -> List[Tuple[str, re.Pattern, str]]:
         return self._compiled_re
 
+    def get_value(self, type_name: str) -> Union[float, int]:
+        if type_name == "bits":
+            return self.bits.money
+        if type_name == "tips":
+            return self.tips.money
+        if type_name == "subs_t1":
+            return self.subs.tier.t1.money
+        if type_name == "subs_t2":
+            return self.subs.tier.t2.money
+        if type_name == "subs_t3":
+            return self.subs.tier.t3.money
+
 
 try:
     SETTINGS = Settings.model_validate(toml.load("settings.toml"), strict=True)