a carrier blocks your caller ID and 32017 is the only notice

"""Report Twilio numbers blocked with 32017 and the ones scoring like them.

Read only. GET requests and nothing else: give this an API Key with read access
rather than the account auth token. The repair is printed, never performed,
because this script holds a credential to an account that can place calls and
spend money.
"""
import argparse
import datetime as dt
import logging
import os
import sys

import requests

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

HOST = "https://api.twilio.com"
BASE = HOST + "/2010-04-01"
MONITOR = "https://monitor.twilio.com/v1"

CARRIER_BLOCKED = 32017

# A call that reached one of these was attempted and finished. Anything still in
# flight is excluded so a window that ends mid-campaign does not depress the rate.
TERMINAL = {"completed", "busy", "no-answer", "failed", "canceled"}


def seconds(value):
    """Duration as an integer. The API returns it as a string, and absent on
    calls that never connected."""
    try:
        return int(str(value or "0").strip())
    except ValueError:
        return 0


def tally(calls, blocked=None):
    """Fold call records into per-caller-ID counters. Pure, so it tests offline.

    blocked maps a number to how many 32017 alerts were raised against it.

    Answered seconds are summed only over completed calls. Averaging duration
    across calls that rang out gives every busy dialer a flattering number and
    is the mistake that makes this whole check useless.
    """
    out = {}
    for c in calls or []:
        frm = str(c.get("from") or "").strip()
        status = str(c.get("status") or "").strip().lower()
        if not frm or status not in TERMINAL:
            continue
        row = out.setdefault(frm, {"attempts": 0, "completed": 0,
                                   "answered_seconds": 0, "blocked": 0})
        row["attempts"] += 1
        if status == "completed":
            row["completed"] += 1
            row["answered_seconds"] += seconds(c.get("duration"))
    for number, count in (blocked or {}).items():
        row = out.setdefault(str(number).strip(),
                             {"attempts": 0, "completed": 0,
                              "answered_seconds": 0, "blocked": 0})
        row["blocked"] = count
    return out


def verdict(stats, min_attempts=20, min_answer_rate=0.30, min_mean_duration=30):
    """Judge one caller ID's reputation profile. Pure.

    The thresholds are defaults, not physics: analytics providers do not publish
    theirs. They are set where a legitimate outbound operation is comfortably
    clear and a short-call dialer is not.

    Returns (state, detail).
    """
    attempts = stats.get("attempts", 0)
    completed = stats.get("completed", 0)
    rate = (completed / float(attempts)) if attempts else 0.0
    mean = (stats.get("answered_seconds", 0) / float(completed)) if completed else 0.0
    shape = ("%d of %d answered (%.0f%%), mean answered call %.0fs"
             % (completed, attempts, rate * 100, mean))

    if stats.get("blocked"):
        return ("blocked",
                "%d call(s) refused with %d by a terminating carrier: %s. The "
                "block is carrier side, so there is nothing to change on the "
                "number itself." % (stats["blocked"], CARRIER_BLOCKED, shape))

    if attempts < min_attempts:
        return ("thin",
                "%d attempt(s) is too little traffic to read a reputation from. "
                "%s" % (attempts, shape))

    low_rate = rate < min_answer_rate
    short = mean < min_mean_duration
    if low_rate and short:
        return ("at-risk",
                "%s. Low answer rate and short answered calls together are the "
                "profile carrier analytics score as a nuisance dialer, and this "
                "number has not been blocked yet." % shape)
    if short:
        return ("short-calls",
                "%s. Mean answered duration under %ds is the single metric most "
                "likely to pull a score down." % (shape, min_mean_duration))
    if low_rate:
        return ("low-answer",
                "%s. An answer rate under %.0f%% suggests the number is already "
                "being labelled on some handsets, which lowers it further."
                % (shape, min_answer_rate * 100))

    return ("healthy", shape)


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_calls(session, account, since, limit):
    """Page the calls. next_page_uri here is a path, not an absolute URL, and
    this resource has no ErrorCode filter, so the bucketing is client-side."""
    url = "%s/Accounts/%s/Calls.json" % (BASE, account)
    params = {"StartTime>=": since, "PageSize": 1000}
    out = []
    while url and len(out) < limit:
        body = get(session, url, **params)
        out.extend(body.get("calls", []))
        nxt = body.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def sweep_alerts(session, since, limit, levels):
    """Both log levels, merged on sid.

    Several voice failures are logged at warning rather than error. Sweeping the
    error level alone is how a voice account reads clean while numbers are being
    refused.
    """
    seen = {}
    for level in levels:
        url = MONITOR + "/Alerts"
        params = {"LogLevel": level, "StartDate": since, "PageSize": 1000}
        got = 0
        while url and got < limit:
            page = get(session, url, **params)
            for a in page.get("alerts", []):
                seen.setdefault(a.get("sid"), a)
                got += 1
            url = (page.get("meta") or {}).get("next_page_url")
            params = {}
    return list(seen.values())


def blocked_numbers(session, account, alerts):
    """Resolve 32017 alerts to the caller ID each was raised against."""
    cache = {}
    counts = {}
    for a in alerts:
        sid = str(a.get("resource_sid") or "")
        if not sid.startswith("CA"):
            continue
        if sid not in cache:
            cache[sid] = get(session, "%s/Accounts/%s/Calls/%s.json"
                             % (BASE, account, sid))
        frm = str(cache[sid].get("from") or "").strip()
        if frm:
            counts[frm] = counts.get(frm, 0) + 1
    return counts


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=30,
                    help="window to measure; reputation moves over weeks")
    ap.add_argument("--max-calls", type=int, default=20000,
                    help="stop after this many calls")
    ap.add_argument("--min-attempts", type=int, default=20,
                    help="below this a number has too little traffic to judge")
    ap.add_argument("--errors-only", action="store_true",
                    help="skip the warning level, which will under-report")
    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)

    days = min(args.days, 30)
    since = (dt.date.today() - dt.timedelta(days=days)).isoformat()
    levels = ["error"] if args.errors_only else ["error", "warning"]

    alerts = sweep_alerts(session, since, 10000, levels)
    hits = [a for a in alerts
            if str(a.get("error_code") or "").strip() == str(CARRIER_BLOCKED)]
    blocked = blocked_numbers(session, account, hits)

    calls = list_calls(session, account, since, args.max_calls)
    rows = tally(calls, blocked)
    if not rows:
        log.info("no outbound calls with a caller ID in the last %d day(s)", days)
        return 0

    bad = 0
    at_risk = 0
    for number in sorted(rows):
        state, detail = verdict(rows[number], args.min_attempts)
        line = "%-12s %s  %s" % (state, number, detail)
        if state in ("healthy", "thin"):
            log.info(line)
            continue
        bad += 1
        if state != "blocked":
            at_risk += 1
        log.warning(line)

    if bad:
        log.warning("  repair: register the numbers at freecallerregistry.com "
                    "and, for T-Mobile, portal.firstorion.com")
        log.warning("  then change the traffic: fewer attempts per number, "
                    "call at hours people answer, raise mean duration. Rotating "
                    "to a fresh number without that earns the same score again")

    log.info("%d number(s), %d blocked, %d at risk",
             len(rows), len(blocked), at_risk)
    return 1 if bad else 0


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

/**
 * Report Twilio numbers blocked with 32017 and the ones scoring like them.
 *
 * Read only. GET requests and nothing else: give this an API Key with read
 * access rather than the account auth token. The repair is printed, never
 * performed.
 */
const HOST = 'https://api.twilio.com';
const BASE = `${HOST}/2010-04-01`;
const MONITOR = 'https://monitor.twilio.com/v1';

const CARRIER_BLOCKED = 32017;

// A call that reached one of these was attempted and finished. Anything still in
// flight is excluded so a window ending mid-campaign does not depress the rate.
const TERMINAL = new Set(['completed', 'busy', 'no-answer', 'failed', 'canceled']);

/** Duration as an integer. The API returns it as a string. */
export function seconds(value) {
  const n = Number.parseInt(String(value ?? '0').trim(), 10);
  return Number.isNaN(n) ? 0 : n;
}

/**
 * Fold call records into per-caller-ID counters. `blocked` maps a number to how
 * many 32017 alerts were raised against it. Answered seconds are summed only
 * over completed calls: averaging duration across calls that rang out gives
 * every busy dialer a flattering number. Pure.
 */
export function tally(calls, blocked = {}) {
  const out = {};
  const row = (n) => {
    if (!out[n]) out[n] = { attempts: 0, completed: 0, answered_seconds: 0, blocked: 0 };
    return out[n];
  };
  for (const c of calls ?? []) {
    const frm = String(c.from ?? '').trim();
    const status = String(c.status ?? '').trim().toLowerCase();
    if (!frm || !TERMINAL.has(status)) continue;
    const r = row(frm);
    r.attempts += 1;
    if (status === 'completed') {
      r.completed += 1;
      r.answered_seconds += seconds(c.duration);
    }
  }
  for (const [number, count] of Object.entries(blocked ?? {})) {
    row(String(number).trim()).blocked = count;
  }
  return out;
}

/**
 * Judge one caller ID's reputation profile. The thresholds are defaults, not
 * physics: analytics providers do not publish theirs. Pure. Returns
 * [state, detail].
 */
export function verdict(stats, minAttempts = 20, minAnswerRate = 0.30,
                        minMeanDuration = 30) {
  const attempts = stats.attempts ?? 0;
  const completed = stats.completed ?? 0;
  const rate = attempts ? completed / attempts : 0;
  const mean = completed ? (stats.answered_seconds ?? 0) / completed : 0;
  const shape = `${completed} of ${attempts} answered (${Math.round(rate * 100)}%), ` +
                `mean answered call ${Math.round(mean)}s`;

  if (stats.blocked) {
    return ['blocked',
      `${stats.blocked} call(s) refused with ${CARRIER_BLOCKED} by a terminating ` +
      `carrier: ${shape}. The block is carrier side, so there is nothing to ` +
      'change on the number itself.'];
  }

  if (attempts < minAttempts) {
    return ['thin',
      `${attempts} attempt(s) is too little traffic to read a reputation from. ${shape}`];
  }

  const lowRate = rate < minAnswerRate;
  const short = mean < minMeanDuration;
  if (lowRate && short) {
    return ['at-risk',
      `${shape}. Low answer rate and short answered calls together are the ` +
      'profile carrier analytics score as a nuisance dialer, and this number ' +
      'has not been blocked yet.'];
  }
  if (short) {
    return ['short-calls',
      `${shape}. Mean answered duration under ${minMeanDuration}s is the single ` +
      'metric most likely to pull a score down.'];
  }
  if (lowRate) {
    return ['low-answer',
      `${shape}. An answer rate under ${Math.round(minAnswerRate * 100)}% suggests ` +
      'the number is already being labelled on some handsets, which lowers it further.'];
  }

  return ['healthy', shape];
}

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 listCalls(auth, account, since, limit) {
  let url = `${BASE}/Accounts/${account}/Calls.json`;
  let params = { 'StartTime>=': since, PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const body = await get(auth, url, params);
    out.push(...(body.calls ?? []));
    url = body.next_page_uri ? HOST + body.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

/** Both log levels, merged on sid: several voice failures are warnings. */
export async function sweepAlerts(auth, since, limit, levels) {
  const seen = new Map();
  for (const level of levels) {
    let url = `${MONITOR}/Alerts`;
    let params = { LogLevel: level, StartDate: since, PageSize: 1000 };
    let got = 0;
    while (url && got < limit) {
      const page = await get(auth, url, params);
      for (const a of page.alerts ?? []) {
        if (!seen.has(a.sid)) seen.set(a.sid, a);
        got += 1;
      }
      url = page.meta?.next_page_url ?? null;
      params = {};
    }
  }
  return [...seen.values()];
}

async function blockedNumbers(auth, account, alerts) {
  const cache = new Map();
  const counts = {};
  for (const a of alerts) {
    const sid = String(a.resource_sid ?? '');
    if (!sid.startsWith('CA')) continue;
    if (!cache.has(sid)) {
      cache.set(sid, await get(auth, `${BASE}/Accounts/${account}/Calls/${sid}.json`));
    }
    const frm = String(cache.get(sid).from ?? '').trim();
    if (frm) counts[frm] = (counts[frm] ?? 0) + 1;
  }
  return counts;
}

function flagValue(name, fallback) {
  const i = process.argv.indexOf(name);
  return i === -1 ? fallback : Number(process.argv[i + 1]);
}

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 days = Math.min(flagValue('--days', 30), 30);
  const minAttempts = flagValue('--min-attempts', 20);
  const maxCalls = flagValue('--max-calls', 20000);
  const levels = process.argv.includes('--errors-only') ? ['error'] : ['error', 'warning'];
  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);

  const alerts = await sweepAlerts(auth, since, 10000, levels);
  const hits = alerts.filter(
    (a) => String(a.error_code ?? '').trim() === String(CARRIER_BLOCKED));
  const blocked = await blockedNumbers(auth, account, hits);

  const calls = await listCalls(auth, account, since, maxCalls);
  const rows = tally(calls, blocked);
  const numbers = Object.keys(rows).sort();
  if (numbers.length === 0) {
    console.log(`no outbound calls with a caller ID in the last ${days} day(s)`);
    return;
  }

  let bad = 0;
  let atRisk = 0;
  for (const number of numbers) {
    const [state, detail] = verdict(rows[number], minAttempts);
    const line = `${state.padEnd(12)} ${number}  ${detail}`;
    if (state === 'healthy' || state === 'thin') { console.log(line); continue; }
    bad += 1;
    if (state !== 'blocked') atRisk += 1;
    console.warn(line);
  }

  if (bad) {
    console.warn('  repair: register the numbers at freecallerregistry.com and, ' +
                 'for T-Mobile, portal.firstorion.com');
    console.warn('  then change the traffic: fewer attempts per number, call at ' +
                 'hours people answer, raise mean duration. Rotating to a fresh ' +
                 'number without that earns the same score again');
  }

  console.log(`${numbers.length} number(s), ${Object.keys(blocked).length} blocked, ` +
              `${atRisk} at risk`);
  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_caller_id_reputation_audit import seconds, tally, verdict


def call(frm, status, duration="0"):
    return {"from": frm, "status": status, "duration": duration}


def test_duration_parses_from_the_string_the_api_returns():
    assert seconds("45") == 45
    assert seconds(None) == 0
    assert seconds("") == 0
    assert seconds("n/a") == 0


def test_unanswered_calls_count_as_attempts_and_not_towards_the_mean():
    # The mistake that makes this whole check useless: a dialer whose calls ring
    # out looks fine if their zero durations are averaged in.
    calls = [call("+15005550006", "completed", "120"),
             call("+15005550006", "no-answer"),
             call("+15005550006", "busy")]
    row = tally(calls)["+15005550006"]
    assert row == {"attempts": 3, "completed": 1, "answered_seconds": 120,
                   "blocked": 0}


def test_calls_still_in_flight_are_excluded_from_the_denominator():
    calls = [call("+15005550006", "completed", "60"),
             call("+15005550006", "in-progress"),
             call("+15005550006", "queued")]
    assert tally(calls)["+15005550006"]["attempts"] == 1


def test_a_blocked_number_with_no_calls_in_the_window_still_appears():
    rows = tally([], {"+15005550006": 4})
    assert rows["+15005550006"]["blocked"] == 4
    assert verdict(rows["+15005550006"])[0] == "blocked"


def test_a_block_outranks_every_other_signal():
    stats = {"attempts": 500, "completed": 480, "answered_seconds": 96000,
             "blocked": 7}
    state, detail = verdict(stats)
    assert state == "blocked"
    assert "carrier side" in detail


def test_too_few_attempts_is_reported_as_thin_rather_than_scored():
    state, detail = verdict({"attempts": 4, "completed": 0, "answered_seconds": 0})
    assert state == "thin"
    assert "0 of 4" in detail


def test_low_answer_rate_and_short_calls_together_are_the_at_risk_profile():
    stats = {"attempts": 400, "completed": 40, "answered_seconds": 320}
    state, detail = verdict(stats)
    assert state == "at-risk"
    assert "10%" in detail
    assert "8s" in detail


def test_short_calls_alone_are_their_own_state():
    stats = {"attempts": 100, "completed": 90, "answered_seconds": 900}
    state, detail = verdict(stats)
    assert state == "short-calls"
    assert "under 30s" in detail


def test_a_low_answer_rate_on_long_calls_is_a_different_finding():
    stats = {"attempts": 200, "completed": 40, "answered_seconds": 8000}
    state, detail = verdict(stats)
    assert state == "low-answer"
    assert "labelled" in detail


def test_a_healthy_number_reports_the_two_numbers_that_matter():
    stats = {"attempts": 200, "completed": 150, "answered_seconds": 30000}
    state, detail = verdict(stats)
    assert state == "healthy"
    assert "150 of 200" in detail
    assert "200s" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { seconds, tally, verdict } from './twilio-caller-id-reputation-audit.mjs';

const call = (from, status, duration = '0') => ({ from, status, duration });

test('duration parses from the string the api returns', () => {
  assert.equal(seconds('45'), 45);
  assert.equal(seconds(null), 0);
  assert.equal(seconds(''), 0);
  assert.equal(seconds('n/a'), 0);
});

test('unanswered calls count as attempts and not towards the mean', () => {
  const calls = [call('+15005550006', 'completed', '120'),
                 call('+15005550006', 'no-answer'),
                 call('+15005550006', 'busy')];
  assert.deepEqual(tally(calls)['+15005550006'],
                   { attempts: 3, completed: 1, answered_seconds: 120, blocked: 0 });
});

test('calls still in flight are excluded from the denominator', () => {
  const calls = [call('+15005550006', 'completed', '60'),
                 call('+15005550006', 'in-progress'),
                 call('+15005550006', 'queued')];
  assert.equal(tally(calls)['+15005550006'].attempts, 1);
});

test('a blocked number with no calls in the window still appears', () => {
  const rows = tally([], { '+15005550006': 4 });
  assert.equal(rows['+15005550006'].blocked, 4);
  assert.equal(verdict(rows['+15005550006'])[0], 'blocked');
});

test('a block outranks every other signal', () => {
  const [state, detail] = verdict(
    { attempts: 500, completed: 480, answered_seconds: 96000, blocked: 7 });
  assert.equal(state, 'blocked');
  assert.match(detail, /carrier side/);
});

test('too few attempts is reported as thin rather than scored', () => {
  const [state, detail] = verdict({ attempts: 4, completed: 0, answered_seconds: 0 });
  assert.equal(state, 'thin');
  assert.match(detail, /0 of 4/);
});

test('low answer rate and short calls together are the at-risk profile', () => {
  const [state, detail] = verdict(
    { attempts: 400, completed: 40, answered_seconds: 320 });
  assert.equal(state, 'at-risk');
  assert.match(detail, /10%/);
  assert.match(detail, /8s/);
});

test('short calls alone are their own state', () => {
  const [state, detail] = verdict(
    { attempts: 100, completed: 90, answered_seconds: 900 });
  assert.equal(state, 'short-calls');
  assert.match(detail, /under 30s/);
});

test('a low answer rate on long calls is a different finding', () => {
  const [state, detail] = verdict(
    { attempts: 200, completed: 40, answered_seconds: 8000 });
  assert.equal(state, 'low-answer');
  assert.match(detail, /labelled/);
});

test('a healthy number reports the two numbers that matter', () => {
  const [state, detail] = verdict(
    { attempts: 200, completed: 150, answered_seconds: 30000 });
  assert.equal(state, 'healthy');
  assert.match(detail, /150 of 200/);
  assert.match(detail, /200s/);
});