#!/usr/bin/env python3 """Проактивный лимитер отправки SMS для SuperSam (правило sms.ru). Главное ограничение sms.ru — **не более 8 SMS за 10 минут по аккаунту суммарно, на любые номера** (антибомбинг; при превышении отправка отклоняется кодами 231/232). Отдельно страхуемся правилом «не более N SMS на один номер за окно». Сейчас кампании узнают о лимите только постфактум — по коду ошибки 231/232, после чего группа помечается send_failed, а часть сообщений уже потеряна. Этот модуль проверяет лимит ДО отправки и придерживает сообщение: оно не теряется, а уходит в следующий запуск кампании. Отправку всех кампаний сериализует общая блокировка (acquire_send_lock): аккаунтный счётчик один на всех, а таймеры кампаний стартуют в одну секунду. Правило и окно настраиваются в sms_campaign_settings (админка/БД): sms_limit_per_number_10min — сколько SMS на один номер за окно (по умолчанию 8; 0 = проверка выключена) sms_limit_account_10min — сколько SMS всего за окно, по аккаунту (боевое значение 8 = правило sms.ru; 0 = проверка выключена) sms_rate_window_seconds — окно в секундах (по умолчанию 600) Что считается отправкой: строки sms_campaign_log, реально принятые sms.ru (sms_id IS NOT NULL) в пределах окна. Получатель берётся как COALESCE(NULLIF(sent_to,''), customer_phone) — то, кому SMS ушла на самом деле; в тестовом режиме все сообщения идут на test_phone и считаются за него, как их видит sms.ru. Номера сравниваются по последним 10 цифрам (8/7/+7 и т.п.). Поведение при сбое БД — fail-open: недоступный счётчик не должен останавливать доставку, в лог пишется ERROR. Повторные Telegram-алерты подавляются (alert_due: не чаще одного раза в час, отметка в sms_campaign_settings). """ import logging log = logging.getLogger("sms_rate_limit") DEFAULT_PER_NUMBER = 8 # правило sms.ru: 8 сообщений на номер DEFAULT_WINDOW_SECONDS = 600 # ... за 10 минут DEFAULT_ACCOUNT = 0 # 0 = аккаунтный лимит не проверяем ALERT_THROTTLE_SECONDS = 3600 # не чаще одного алерта в час COUNT_SQL = """ SELECT COUNT(*) FROM sms_campaign_log WHERE created_at > NOW() - (%(win)s || ' seconds')::interval AND sms_id IS NOT NULL {phone_filter} """ PHONE_FILTER = ("AND right(regexp_replace(" "COALESCE(NULLIF(sent_to, ''), customer_phone), '\\D', '', 'g'), 10) " "= right(regexp_replace(%(phone)s, '\\D', '', 'g'), 10)") def _int(value, default): """Аккуратно приводит настройку к int (None, '' и мусор → default).""" try: if value is None or value == "": return int(default) return int(value) except (TypeError, ValueError): return int(default) def rules(settings): """Правила лимита из настроек кампании: (на номер, всего, окно).""" s = settings or {} return ( _int(s.get("sms_limit_per_number_10min"), DEFAULT_PER_NUMBER), _int(s.get("sms_limit_account_10min"), DEFAULT_ACCOUNT), _int(s.get("sms_rate_window_seconds"), DEFAULT_WINDOW_SECONDS), ) def _count(conn, window_seconds, phone=None): sql = COUNT_SQL.format(phone_filter=PHONE_FILTER if phone else "") params = {"win": window_seconds} if phone: params["phone"] = phone with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchone()[0] def check_send_allowed(conn, phone, settings): """Можно ли отправить ещё одну SMS на этот номер. Возвращает (ok, reason, info): ok — True (можно) / False (придержать) reason — короткое объяснение для лога и Telegram info — {'per_number': n, 'account': n, 'window': w} для алерта """ per_number, account, window = rules(settings) info = {"per_number": per_number, "account": account, "window": window} if per_number <= 0 and account <= 0: return True, "лимит выключен в настройках", info if not phone: return True, "пустой номер", info try: if per_number > 0: n = _count(conn, window, phone) info["number_count"] = n if n >= per_number: return False, (f"номер {phone}: уже {n} SMS за {window}с " f"(лимит {per_number})"), info if account > 0: total = _count(conn, window) info["account_count"] = total if total >= account: return False, (f"аккаунт: уже {total} SMS за {window}с " f"(лимит {account})"), info except Exception as exc: # noqa: BLE001 — fail-open, доставку не стопим log.error("sms rate limit check failed (fail-open): %s", exc) info["db_error"] = str(exc) return True, f"проверка лимита не удалась ({exc}) — отправляем", info return True, "лимит не выбран", info def alert_due(conn, settings, throttle_seconds=ALERT_THROTTLE_SECONDS): """Можно ли отправить Telegram-алерт про придержанные SMS (не чаще раза в час). Отметка пишется в sms_campaign_settings.last_limit_alert_at, таблица кампаний — единственная «своя» запись под рукой. Ошибки БД → False, чтобы не заваливать чат алертами. """ campaign_type = (settings or {}).get("campaign_type") if not campaign_type: return True # настроек нет — не рискуем потерять уведомление try: with conn.cursor() as cur: cur.execute(""" UPDATE sms_campaign_settings SET last_limit_alert_at = NOW() WHERE campaign_type = %s AND (last_limit_alert_at IS NULL OR last_limit_alert_at < NOW() - (%s || ' seconds')::interval) RETURNING id """, (campaign_type, throttle_seconds)) ok = cur.fetchone() is not None conn.commit() return ok except Exception as exc: # noqa: BLE001 conn.rollback() log.error("alert_due failed: %s", exc) return False # ─── Блокировка отправки (сериализация кампаний) ───────────────────────────── SEND_LOCK_KEY = 0x53535331 # 'SSS1' — общий ключ отправки всех SMS-кампаний LOCK_TIMEOUT_SECONDS = 120 # ждём чужой прогон не дольше двух минут def acquire_send_lock(conn, timeout_seconds=LOCK_TIMEOUT_SECONDS): """Захватить общую блокировку отправки (pg_advisory_lock). Аккаунтный лимит sms.ru общий для всех кампаний, а их таймеры стартуют в одну и ту же секунду: без сериализации каждый процесс видит «свободно 8» и вместе они отправят больше лимита. Блокировка — на соединении, снимается сама при закрытии коннекта. True — блокировка взята, можно отправлять. False — чужой прогон не завершился за timeout_seconds: прогон пропускаем (лучше задержка, чем блокировка аккаунта у sms.ru), повторим по таймеру. """ try: with conn.cursor() as cur: cur.execute("SET lock_timeout = %s", (f"{int(timeout_seconds)}s",)) cur.execute("SELECT pg_advisory_lock(%s)", (SEND_LOCK_KEY,)) return True except Exception as e: # noqa: BLE001 — недоступная блокировка не должна ломать прогон try: conn.rollback() except Exception: # noqa: BLE001 pass log.error("Не удалось взять блокировку отправки (%s) — прогон пропущен", e) return False def release_send_lock(conn): """Снять блокировку отправки. Ошибки не критичны: снимет закрытие коннекта.""" try: with conn.cursor() as cur: cur.execute("SELECT pg_advisory_unlock(%s)", (SEND_LOCK_KEY,)) return True except Exception as e: # noqa: BLE001 log.warning("Не удалось снять блокировку отправки: %s", e) return False