| """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: |
| 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 |
|
|
