Create rate_limit.py

core/rate_limit.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+import os, json, datetime
+from pathlib import Path
+from typing import Tuple
+from .config import settings
+
+def _today_key() -> str:
+    return datetime.date.today().isoformat()
+
+def _redis_client():
+    if not settings.redis_url:
+        return None
+    try:
+        from redis import Redis
+        return Redis.from_url(settings.redis_url, decode_responses=True)
+    except Exception:
+        return None
+
+def check_and_increment_global_ai_cap() -> Tuple[bool, dict]:
+    limit = settings.max_ai_summaries_per_day
+    day = _today_key()
+
+    r = _redis_client()
+    if r:
+        key = f"toxrai:ai_summaries:{day}"
+        val = r.incr(key)
+        if val == 1:
+            r.expire(key, 60 * 60 * 24 * 2)
+        return (val <= limit), {"store": "redis", "day": day, "count": int(val), "limit": limit}
+
+    # fallback: file (may reset on restart if no persistent volume)
+    path = Path("/data/toxrai_limits.json") if os.path.isdir("/data") else Path(".toxrai_limits.json")
+    data = {}
+    if path.exists():
+        try:
+            data = json.loads(path.read_text(encoding="utf-8") or "{}")
+        except Exception:
+            data = {}
+
+    data.setdefault("ai_summaries", {})
+    data["ai_summaries"].setdefault(day, 0)
+    data["ai_summaries"][day] += 1
+
+    try:
+        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
+    except Exception:
+        pass
+
+    count = int(data["ai_summaries"][day])
+    return (count <= limit), {"store": "file", "path": str(path), "day": day, "count": count, "limit": limit}