my_voice_assistant_v3_jamulix/app/quota.py

"""Pro-Nutzer-Tageskontingent (Kostenkontrolle).

Limit aus Settings (`daily_request_limit`), pro Nutzer ueber `prefs.daily_request_limit`
ueberschreibbar. 0 bedeutet unbegrenzt.
"""

from app.metrics import metrics


class QuotaExceededError(Exception):
    def __init__(self, limit: int, count: int):
        self.limit = limit
        self.count = count
        super().__init__(f"Daily request limit reached ({count}/{limit})")


def effective_limit(user, cfg=None) -> int:
    if cfg is None:
        from app.runtime_config import runtime_settings
        cfg = runtime_settings
    pref = user.prefs.get("daily_request_limit") if user and user.prefs else None
    if pref is not None:
        try:
            return int(pref)
        except (TypeError, ValueError):
            pass
    return cfg.daily_request_limit


def enforce_quota(user, store, cfg=None) -> None:
    """Wirft QuotaExceededError, wenn das Tageslimit erreicht ist."""
    limit = effective_limit(user, cfg)
    if limit and limit > 0:
        count = store.get_request_count(user.id)
        if count >= limit:
            metrics.inc("quota_exceeded_total")
            raise QuotaExceededError(limit, count)


def record_usage(user, store, units: int = 0) -> None:
    store.add_usage(user.id, units)
    metrics.inc("turns_total")


def web_search_limit(user, cfg=None) -> int:
    """Max. Web-Suchen pro Tag fuer diesen Nutzer (Pref > global; 0 = unbegrenzt)."""
    if cfg is None:
        from app.runtime_config import runtime_settings
        cfg = runtime_settings
    pref = user.prefs.get("web_search_daily_limit") if user and user.prefs else None
    if pref is not None:
        try:
            return int(pref)
        except (TypeError, ValueError):
            pass
    return cfg.web_search_daily_limit


def web_search_within_budget(user, store, cfg=None) -> bool:
    """Darf der Nutzer heute noch eine Web-Suche ausloesen? (Limit 0 = unbegrenzt)."""
    limit = web_search_limit(user, cfg)
    if not limit or limit <= 0:
        return True
    return store.web_search_count(user.id) < limit


def record_web_search(user, store, n: int = 1) -> None:
    """Zaehlt n tatsaechlich ausgefuehrte Web-Suchen fuer den Nutzer."""
    if n and n > 0 and user is not None:
        store.add_web_search_usage(user.id, n)