T-Mobile caps daily segments per brand, not per campaign

"""Measure today's segment burn against T-Mobile's daily cap on your brand.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. Nothing here can raise the cap; the script
exists so the ceiling is a number somebody knows before the afternoon batch
runs into it.
"""
import argparse
import datetime
import logging
import os
import sys

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("twilio_tmobile_daily_cap_report")

API = "https://api.twilio.com/2010-04-01"
MSG = "https://messaging.twilio.com/v1"

DAILY_CAP_ERROR = 30023

# Published tier defaults. Everything between these two is assigned by T-Mobile
# from the trust tier and is not exposed as a field, so it has to be supplied.
SOLE_PROP_DAILY_SEGMENTS = 1000
RUSSELL_3000_DAILY_SEGMENTS = 200000


def brand_ceiling(brand):
    """Derive the daily segment ceiling from a brand registration. Pure.

    Returns (ceiling, source). ceiling is None when the brand fields do not
    determine it, which is the common case: the tier comes from T-Mobile and
    there is no field to read it from.
    """
    if not brand:
        return (None, "no brand registration to read")

    brand_type = str(brand.get("brand_type") or "").upper()
    if brand_type == "SOLE_PROPRIETOR":
        return (SOLE_PROP_DAILY_SEGMENTS, "sole proprietor brands are capped at "
                "1,000 segments a day")

    if brand.get("russell_3000"):
        return (RUSSELL_3000_DAILY_SEGMENTS,
                "russell_3000 is true, which defaults to 200,000 segments a day")

    score = brand.get("brand_score")
    return (None,
            "brand_type is %s with brand_score %s: the tier is assigned by "
            "T-Mobile and is not exposed as a field, so pass --ceiling with the "
            "value from your tier"
            % (brand_type or "unset", "unset" if score is None else score))


def summarise(messages):
    """Total segments and capped-message count for a day of messages. Pure.

    num_segments arrives as a string. There is no ErrorCode filter on the
    Messages list, so 30023 is counted here rather than asked for.
    """
    segments = 0
    capped = 0
    for m in messages or []:
        try:
            segments += int(m.get("num_segments") or 0)
        except (TypeError, ValueError):
            pass
        try:
            code = int(m.get("error_code") or 0)
        except (TypeError, ValueError):
            code = 0
        if code == DAILY_CAP_ERROR:
            capped += 1
    return (segments, capped)


def verdict(ceiling, segments, capped, warn_ratio=0.8):
    """Classify one brand's position against the daily cap. Pure.

    An observed 30023 outranks the arithmetic, because the segment total is an
    upper bound: the Messages list does not say which carrier a destination
    belongs to, so the T-Mobile share of it cannot be isolated.
    Returns (state, detail).
    """
    if capped:
        return ("cap-hit",
                "%d message(s) today came back %d. The daily allowance ran out; "
                "it resets at midnight US Pacific." % (capped, DAILY_CAP_ERROR))

    if segments is None:
        return ("burn-unknown",
                "today's messages could not be read, so the burn is unknown.")

    if ceiling is None:
        return ("ceiling-unknown",
                "%d segment(s) sent today and no ceiling could be derived from "
                "the brand. Supply the tier value to turn this into a warning."
                % segments)

    if segments >= ceiling:
        return ("over-estimate",
                "%d segment(s) today against a ceiling of %d. That total is "
                "every carrier, so it is an upper bound on the T-Mobile share, "
                "but it is past the line and nothing has failed yet only "
                "because not all of it went to T-Mobile."
                % (segments, ceiling))

    if segments >= ceiling * warn_ratio:
        return ("near-cap",
                "%d segment(s) today, %.0f%% of the %d ceiling. Spread the rest "
                "of the day's volume." % (segments, 100.0 * segments / ceiling,
                                          ceiling))

    return ("under-cap",
            "%d segment(s) today against a ceiling of %d." % (segments, ceiling))


def get(session, url, **params):
    r = session.get(url, params=params, timeout=30)
    if r.status_code in (401, 403):
        raise SystemExit("%d from Twilio: check TWILIO_ACCOUNT_SID and that the "
                         "API key belongs to that account with read access"
                         % r.status_code)
    r.raise_for_status()
    return r.json()


def list_v1(session, url, key, limit=1000):
    """Page a messaging.twilio.com list. meta.next_page_url is absolute."""
    out = []
    while url and len(out) < limit:
        page = get(session, url, PageSize=50)
        out.extend(page.get(key, []))
        url = (page.get("meta") or {}).get("next_page_url")
    return out[:limit]


def list_messages(session, account, since, limit=20000):
    """Page the Messages list. next_page_uri is a path, not an absolute URL."""
    out = []
    page = get(session, "%s/Accounts/%s/Messages.json" % (API, account),
               PageSize=1000, **{"DateSent>": since})
    while page:
        out.extend(page.get("messages", []))
        nxt = page.get("next_page_uri")
        if not nxt or len(out) >= limit:
            break
        page = get(session, "https://api.twilio.com" + nxt)
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--ceiling", type=int, default=None,
                    help="daily segment ceiling for your T-Mobile tier, when "
                         "the brand fields do not determine it")
    ap.add_argument("--warn-ratio", type=float, default=0.8)
    ap.add_argument("--max-services", type=int, default=200)
    args = ap.parse_args()

    account = os.environ.get("TWILIO_ACCOUNT_SID")
    key = os.environ.get("TWILIO_API_KEY")
    secret = os.environ.get("TWILIO_API_SECRET")
    if not (account and key and secret):
        log.error("set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET "
                  "(an API Key with read access, not the auth token)")
        return 2

    session = requests.Session()
    session.auth = (key, secret)

    # The counter resets at midnight US Pacific, which is not your servers' day.
    pacific = datetime.timezone(datetime.timedelta(hours=-7))
    today = datetime.datetime.now(pacific).date().isoformat()

    messages = list_messages(session, account, today)
    segments, capped = summarise(messages)

    services = list_v1(session, MSG + "/Services", "services", args.max_services)
    brands = {}
    for svc in services:
        campaigns = list_v1(session, "%s/Services/%s/Compliance/Usa2p"
                            % (MSG, svc["sid"]), "compliance")
        for campaign in campaigns:
            brand_sid = campaign.get("brand_registration_sid")
            if not brand_sid or brand_sid in brands:
                continue
            brands[brand_sid] = get(session, "%s/a2p/BrandRegistrations/%s"
                                    % (MSG, brand_sid))
            limits = campaign.get("rate_limits")
            if limits:
                log.info("rate_limits on %s: %s", svc["sid"], limits)

    if not brands:
        log.info("no A2P brands reachable from the Messaging Services on this account")
        return 0

    bad = 0
    for brand_sid, brand in brands.items():
        derived, source = brand_ceiling(brand)
        ceiling = args.ceiling if args.ceiling is not None else derived
        state, detail = verdict(ceiling, segments, capped, args.warn_ratio)
        line = "%-16s %s  %s" % (state, brand_sid, detail)
        if state == "under-cap":
            log.info(line)
            continue
        bad += 1
        log.warning(line)
        log.warning("  ceiling: %s", source if args.ceiling is None
                    else "supplied on the command line")
        if state in ("cap-hit", "over-estimate", "near-cap"):
            log.warning("  repair: the cap cannot be raised by API. Move the "
                        "brand up a tier (Sole Proprietor to Standard, then "
                        "secondary vetting to lift brand_score), or request a "
                        "T-Mobile Special Business Review through Twilio Support")
            log.warning("  repair: until then, spread the day's volume and "
                        "shorten bodies, since the cap counts segments and a "
                        "160 character overflow doubles the cost of every send")

    log.info("%d brand(s), %d segment(s) today, %d capped",
             len(brands), segments, capped)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Measure today's segment burn against T-Mobile's daily cap on your brand.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. Nothing here can raise the cap.
 */
const API = 'https://api.twilio.com/2010-04-01';
const MSG = 'https://messaging.twilio.com/v1';

export const DAILY_CAP_ERROR = 30023;

// Published tier defaults. Everything between these two is assigned by T-Mobile
// from the trust tier and is not exposed as a field, so it has to be supplied.
const SOLE_PROP_DAILY_SEGMENTS = 1000;
const RUSSELL_3000_DAILY_SEGMENTS = 200000;

/**
 * Derive the daily segment ceiling from a brand registration. Pure.
 * Returns [ceiling, source]; ceiling is null when the fields do not determine it.
 */
export function brandCeiling(brand) {
  if (!brand) return [null, 'no brand registration to read'];

  const brandType = String(brand.brand_type ?? '').toUpperCase();
  if (brandType === 'SOLE_PROPRIETOR') {
    return [SOLE_PROP_DAILY_SEGMENTS,
      'sole proprietor brands are capped at 1,000 segments a day'];
  }

  if (brand.russell_3000) {
    return [RUSSELL_3000_DAILY_SEGMENTS,
      'russell_3000 is true, which defaults to 200,000 segments a day'];
  }

  const score = brand.brand_score;
  return [null,
    `brand_type is ${brandType || 'unset'} with brand_score ` +
    `${score === null || score === undefined ? 'unset' : score}: the tier is ` +
    'assigned by T-Mobile and is not exposed as a field, so pass --ceiling with ' +
    'the value from your tier'];
}

/**
 * Total segments and capped-message count for a day of messages. Pure.
 * num_segments arrives as a string, and there is no ErrorCode filter on the list.
 */
export function summarise(messages) {
  let segments = 0;
  let capped = 0;
  for (const m of messages ?? []) {
    const n = Number.parseInt(m.num_segments ?? 0, 10);
    if (Number.isFinite(n)) segments += n;
    const code = Number.parseInt(m.error_code ?? 0, 10);
    if (code === DAILY_CAP_ERROR) capped += 1;
  }
  return [segments, capped];
}

/**
 * Classify one brand's position against the daily cap. Pure.
 * An observed 30023 outranks the arithmetic, because the segment total is an
 * upper bound: the Messages list does not say which carrier a destination is on.
 * Returns [state, detail].
 */
export function verdict(ceiling, segments, capped, warnRatio = 0.8) {
  if (capped) {
    return ['cap-hit',
      `${capped} message(s) today came back ${DAILY_CAP_ERROR}. The daily ` +
      'allowance ran out; it resets at midnight US Pacific.'];
  }

  if (segments === null || segments === undefined) {
    return ['burn-unknown', "today's messages could not be read, so the burn is unknown."];
  }

  if (ceiling === null || ceiling === undefined) {
    return ['ceiling-unknown',
      `${segments} segment(s) sent today and no ceiling could be derived from ` +
      'the brand. Supply the tier value to turn this into a warning.'];
  }

  if (segments >= ceiling) {
    return ['over-estimate',
      `${segments} segment(s) today against a ceiling of ${ceiling}. That total ` +
      'is every carrier, so it is an upper bound on the T-Mobile share, but it ' +
      'is past the line and nothing has failed yet only because not all of it ' +
      'went to T-Mobile.'];
  }

  if (segments >= ceiling * warnRatio) {
    return ['near-cap',
      `${segments} segment(s) today, ${(100 * segments / ceiling).toFixed(0)}% of ` +
      `the ${ceiling} ceiling. Spread the rest of the day's volume.`];
  }

  return ['under-cap', `${segments} segment(s) today against a ceiling of ${ceiling}.`];
}

function authHeader(key, secret) {
  return `Basic ${Buffer.from(`${key}:${secret}`).toString('base64')}`;
}

async function get(auth, url, params = {}) {
  const u = new URL(url);
  for (const [k, v] of Object.entries(params)) u.searchParams.set(k, v);
  const res = await fetch(u, { headers: { Authorization: auth } });
  if (res.status === 401 || res.status === 403) {
    throw new Error(`${res.status} from Twilio: check TWILIO_ACCOUNT_SID and ` +
                    'that the API key belongs to that account with read access');
  }
  if (!res.ok) throw new Error(`${res.status} from ${u.pathname}`);
  return res.json();
}

export async function listV1(auth, url, key, limit = 1000) {
  const out = [];
  let next = url;
  while (next && out.length < limit) {
    const page = await get(auth, next, { PageSize: 50 });
    out.push(...(page[key] ?? []));
    next = page.meta?.next_page_url ?? null;
  }
  return out.slice(0, limit);
}

async function listMessages(auth, account, since, limit = 20000) {
  const out = [];
  let page = await get(auth, `${API}/Accounts/${account}/Messages.json`,
                       { PageSize: 1000, 'DateSent>': since });
  while (page) {
    out.push(...(page.messages ?? []));
    const nxt = page.next_page_uri;
    if (!nxt || out.length >= limit) break;
    page = await get(auth, `https://api.twilio.com${nxt}`);
  }
  return out.slice(0, limit);
}

async function main() {
  const account = process.env.TWILIO_ACCOUNT_SID;
  const key = process.env.TWILIO_API_KEY;
  const secret = process.env.TWILIO_API_SECRET;
  if (!account || !key || !secret) {
    console.error('set TWILIO_ACCOUNT_SID, TWILIO_API_KEY and TWILIO_API_SECRET ' +
                  '(an API Key with read access, not the auth token)');
    process.exitCode = 2;
    return;
  }
  const auth = authHeader(key, secret);
  const flag = process.argv.indexOf('--ceiling');
  const supplied = flag >= 0 ? Number(process.argv[flag + 1]) : null;

  // The counter resets at midnight US Pacific, which is not your servers' day.
  const pacific = new Date(Date.now() - 7 * 3600000);
  const today = pacific.toISOString().slice(0, 10);

  const messages = await listMessages(auth, account, today);
  const [segments, capped] = summarise(messages);

  const services = await listV1(auth, `${MSG}/Services`, 'services');
  const brands = {};
  for (const svc of services) {
    const campaigns = await listV1(auth, `${MSG}/Services/${svc.sid}/Compliance/Usa2p`,
                                   'compliance');
    for (const campaign of campaigns) {
      const brandSid = campaign.brand_registration_sid;
      if (!brandSid || brandSid in brands) continue;
      brands[brandSid] = await get(auth, `${MSG}/a2p/BrandRegistrations/${brandSid}`);
      if (campaign.rate_limits) {
        console.log(`rate_limits on ${svc.sid}:`, campaign.rate_limits);
      }
    }
  }

  const sids = Object.keys(brands);
  if (sids.length === 0) {
    console.log('no A2P brands reachable from the Messaging Services on this account');
    return;
  }

  let bad = 0;
  for (const brandSid of sids) {
    const [derived, source] = brandCeiling(brands[brandSid]);
    const ceiling = supplied === null ? derived : supplied;
    const [state, detail] = verdict(ceiling, segments, capped);
    const line = `${state.padEnd(16)} ${brandSid}  ${detail}`;
    if (state === 'under-cap') { console.log(line); continue; }
    bad += 1;
    console.warn(line);
    console.warn(`  ceiling: ${supplied === null ? source : 'supplied on the command line'}`);
    if (state === 'cap-hit' || state === 'over-estimate' || state === 'near-cap') {
      console.warn('  repair: the cap cannot be raised by API. Move the brand up ' +
                   'a tier (Sole Proprietor to Standard, then secondary vetting ' +
                   'to lift brand_score), or request a T-Mobile Special Business ' +
                   'Review through Twilio Support');
      console.warn("  repair: until then, spread the day's volume and shorten " +
                   'bodies, since the cap counts segments and a 160 character ' +
                   'overflow doubles the cost of every send');
    }
  }

  console.log(`${sids.length} brand(s), ${segments} segment(s) today, ${capped} capped`);
  process.exitCode = bad ? 1 : 0;
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing credentials and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from twilio_tmobile_daily_cap_report import brand_ceiling, summarise, verdict

SOLE_PROP = {"brand_type": "SOLE_PROPRIETOR", "brand_score": None}
RUSSELL = {"brand_type": "STANDARD", "russell_3000": True}
STANDARD = {"brand_type": "STANDARD", "brand_score": 62, "russell_3000": False}


def test_segments_are_summed_not_messages_counted():
    # Four messages, ten segments. Counting rows would report 40% of a 25 cap.
    messages = [{"num_segments": "4"}, {"num_segments": "1"},
                {"num_segments": "3"}, {"num_segments": "2"}]
    assert summarise(messages) == (10, 0)


def test_unreadable_segment_counts_do_not_abort_the_sum():
    assert summarise([{"num_segments": "2"}, {"num_segments": None},
                      {"num_segments": "x"}]) == (2, 0)


def test_30023_is_counted_client_side():
    messages = [{"num_segments": "1", "error_code": 30023},
                {"num_segments": "1", "error_code": "30023"},
                {"num_segments": "1", "error_code": 30007}]
    assert summarise(messages) == (3, 2)


def test_an_observed_cap_hit_outranks_the_estimate():
    # The segment total is an upper bound across all carriers. A 30023 is not.
    state, detail = verdict(200000, 12, capped=3)
    assert state == "cap-hit"
    assert "midnight US Pacific" in detail


def test_sole_proprietor_ceiling_is_derived_from_the_brand():
    ceiling, source = brand_ceiling(SOLE_PROP)
    assert ceiling == 1000
    assert "1,000 segments" in source


def test_russell_3000_defaults_to_two_hundred_thousand():
    assert brand_ceiling(RUSSELL)[0] == 200000


def test_an_ordinary_standard_brand_has_no_readable_ceiling():
    ceiling, source = brand_ceiling(STANDARD)
    assert ceiling is None
    assert "--ceiling" in source


def test_the_warning_band_sits_below_the_line():
    assert verdict(1000, 850, 0)[0] == "near-cap"
    assert verdict(1000, 400, 0)[0] == "under-cap"
    assert verdict(1000, 1200, 0)[0] == "over-estimate"
    assert verdict(None, 400, 0)[0] == "ceiling-unknown"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { brandCeiling, summarise, verdict } from './twilio-tmobile-daily-cap-report.mjs';

const SOLE_PROP = { brand_type: 'SOLE_PROPRIETOR', brand_score: null };
const RUSSELL = { brand_type: 'STANDARD', russell_3000: true };
const STANDARD = { brand_type: 'STANDARD', brand_score: 62, russell_3000: false };

test('segments are summed, not messages counted', () => {
  const messages = [{ num_segments: '4' }, { num_segments: '1' },
                    { num_segments: '3' }, { num_segments: '2' }];
  assert.deepEqual(summarise(messages), [10, 0]);
});

test('unreadable segment counts do not abort the sum', () => {
  assert.deepEqual(summarise([{ num_segments: '2' }, { num_segments: null },
                              { num_segments: 'x' }]), [2, 0]);
});

test('30023 is counted client side', () => {
  const messages = [{ num_segments: '1', error_code: 30023 },
                    { num_segments: '1', error_code: '30023' },
                    { num_segments: '1', error_code: 30007 }];
  assert.deepEqual(summarise(messages), [3, 2]);
});

test('an observed cap hit outranks the estimate', () => {
  const [state, detail] = verdict(200000, 12, 3);
  assert.equal(state, 'cap-hit');
  assert.match(detail, /midnight US Pacific/);
});

test('sole proprietor ceiling is derived from the brand', () => {
  const [ceiling, source] = brandCeiling(SOLE_PROP);
  assert.equal(ceiling, 1000);
  assert.match(source, /1,000 segments/);
});

test('russell 3000 defaults to two hundred thousand', () => {
  assert.equal(brandCeiling(RUSSELL)[0], 200000);
});

test('an ordinary standard brand has no readable ceiling', () => {
  const [ceiling, source] = brandCeiling(STANDARD);
  assert.equal(ceiling, null);
  assert.match(source, /--ceiling/);
});

test('the warning band sits below the line', () => {
  assert.equal(verdict(1000, 850, 0)[0], 'near-cap');
  assert.equal(verdict(1000, 400, 0)[0], 'under-cap');
  assert.equal(verdict(1000, 1200, 0)[0], 'over-estimate');
  assert.equal(verdict(null, 400, 0)[0], 'ceiling-unknown');
});