"""Admin alert when DubLab runs out of Gemini quota. Sends one Telegram message via @mzbanx_bot to the BOT ERROR REPORT & REQUEST group, then stays quiet for COOLDOWN seconds so a queue full of failing jobs can't spam the group. A successful job re-arms it. Token/chat come from .env: DUBLAB_ALERT_BOT_TOKEN, DUBLAB_ALERT_CHAT_ID. Only stdlib is used. """ import json import os import time import urllib.request COOLDOWN = 3 * 3600 _last = 0.0 def _send(text): token = os.environ.get("DUBLAB_ALERT_BOT_TOKEN", "") chat = os.environ.get("DUBLAB_ALERT_CHAT_ID", "") if not token or not chat: return False body = json.dumps({"chat_id": chat, "text": text}).encode() req = urllib.request.Request( f"https://api.telegram.org/bot{token}/sendMessage", data=body, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=15) as r: return r.status == 200 except Exception: # noqa: BLE001 - an alert must never break a job return False def quota_alert(where, detail): """Alert once per cooldown window. Returns True if a message went out.""" global _last now = time.time() if now - _last < COOLDOWN: return False _last = now text = ("⚠️ DubLab — Gemini quota ပြည့်ပြီ\n" f"နေရာ: {where}\n" f"အသေးစိတ်: {detail[:200]}\n" "ဒီကာလအတွင်း dub job တွေ fail နေနိုင်တယ် — quota ပြန်ရရင် " "အလိုလို ပြန်ကောင်းမယ်။ နောက် ၃ နာရီအတွင်း ထပ် alert မပို့ဘူး") return _send(text) def quota_ok(): """A job succeeded — quota is back, re-arm the alert.""" global _last _last = 0.0