answering machine detection is routing humans to voicemail

"""Report how Twilio's answering machine detection is classifying your calls.

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_amd_classification_audit")

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

# DetectMessageEnd waits for the greeting to finish before deciding, so its
# verdicts arrive as this family rather than as machine_start.
MACHINE_END = ("machine_end_beep", "machine_end_silence", "machine_end_other")

GRADED = ("human", "machine", "machine-short", "unknown", "fax")


def seconds(value):
    """A call duration as an int. It arrives as a string and can be absent."""
    try:
        return int(str(value or "0").strip() or 0)
    except ValueError:
        return 0


def bucket(call, short_seconds=8):
    """Put one call in an answering-machine bucket. Pure, so the rules can be
    tested without a network.

    A machine_start call of a few seconds is the misroute this note is about: a
    person answered, detection called them a machine, and they hung up on the
    voicemail drop. The machine_end_* family is deliberately not split the same
    way, because there Twilio waited for the greeting to end and a short call
    means something else entirely.
    """
    if str(call.get("status") or "").strip().lower() != "completed":
        return "not-completed"

    answered = str(call.get("answered_by") or "").strip().lower()
    if not answered:
        return "no-amd"
    if answered in ("human", "fax", "unknown"):
        return answered
    if answered == "machine_start":
        return "machine-short" if seconds(call.get("duration")) <= short_seconds else "machine"
    if answered in MACHINE_END:
        return "machine"
    return "other"


def verdict(tally, min_calls=50, unknown_pct=3.0, machine_pct=40.0, short_pct=25.0):
    """Turn a tally of buckets into a verdict. Pure.

    The thresholds are arguments rather than constants because they are
    defaults, not truths: a debt collector's real voicemail rate is nothing like
    a delivery notification's. Returns (state, detail).
    """
    graded = sum(tally.get(k, 0) for k in GRADED)
    if graded == 0:
        return ("no-amd",
                "no call in this window carries answered_by, so machine "
                "detection was never requested and there is nothing to tune.")
    if graded < min_calls:
        return ("thin-sample",
                "only %d graded call(s), under the %d needed to read a "
                "distribution. Widen the window rather than trusting this."
                % (graded, min_calls))

    machines = tally.get("machine", 0) + tally.get("machine-short", 0)
    unknown_share = 100.0 * tally.get("unknown", 0) / graded
    machine_share = 100.0 * machines / graded
    short_share = (100.0 * tally.get("machine-short", 0) / machines) if machines else 0.0

    if unknown_share > unknown_pct:
        return ("detection-timing-out",
                "%.1f%% of %d graded call(s) came back unknown, over the %.1f%% "
                "threshold. unknown is a timeout, not a category: detection ran "
                "out of time and your flow branched on a value it has no case "
                "for." % (unknown_share, graded, unknown_pct))

    if machine_share > machine_pct and short_share > short_pct:
        return ("over-classifying",
                "%.1f%% of %d graded call(s) were called machines and %.1f%% of "
                "those lasted seconds. That short tail is people hanging up on a "
                "voicemail drop aimed at them."
                % (machine_share, graded, short_share))

    if machine_share > machine_pct:
        return ("machine-heavy",
                "%.1f%% of %d graded call(s) were machines, over the %.1f%% "
                "threshold, but only %.1f%% of them were short. This looks like "
                "a list that really does reach voicemail, not a detector fault."
                % (machine_share, graded, machine_pct, short_share))

    return ("healthy",
            "%d graded call(s): human %.1f%%, machine %.1f%%, unknown %.1f%%"
            % (graded, 100.0 * tally.get("human", 0) / graded,
               machine_share, unknown_share))


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 listing. next_page_uri here is a path, not a URL."""
    url = "%s/Accounts/%s/Calls.json" % (BASE, account)
    params = {"StartTime>=": since, "PageSize": 1000}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("calls", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=7, help="window to tally")
    ap.add_argument("--max-calls", type=int, default=20000,
                    help="stop after this many calls")
    ap.add_argument("--short-seconds", type=int, default=8,
                    help="a machine_start call this short is a suspected misroute")
    ap.add_argument("--min-calls", type=int, default=50,
                    help="fewer graded calls than this is not a distribution")
    ap.add_argument("--unknown-pct", type=float, default=3.0,
                    help="unknown share above this is a detection timeout")
    ap.add_argument("--machine-pct", type=float, default=40.0,
                    help="machine share above this is worth explaining")
    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)

    since = (dt.date.today() - dt.timedelta(days=args.days)).isoformat()
    calls = list_calls(session, account, since, args.max_calls)
    if not calls:
        log.info("no calls in the last %d day(s)", args.days)
        return 0

    tally = {}
    for c in calls:
        b = bucket(c, args.short_seconds)
        tally[b] = tally.get(b, 0) + 1

    for name in sorted(tally):
        log.info("%-14s %d", name, tally[name])

    state, detail = verdict(tally, args.min_calls, args.unknown_pct,
                            args.machine_pct)
    if state in ("healthy", "no-amd", "thin-sample"):
        log.info("%s  %s", state, detail)
        return 0

    log.warning("%s  %s", state, detail)
    log.warning("  repair: on the outbound create request set "
                "MachineDetection=DetectMessageEnd, or raise "
                "MachineDetectionTimeout and MachineDetectionSpeechThreshold")
    log.warning("  repair: or set AsyncAmd=true with AsyncAmdStatusCallback so "
                "the call connects first and is reclassified after")
    return 1


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

/**
 * Report how Twilio's answering machine detection is classifying your calls.
 *
 * 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`;

// DetectMessageEnd waits for the greeting to finish, so its verdicts arrive as
// this family rather than as machine_start.
const MACHINE_END = ['machine_end_beep', 'machine_end_silence', 'machine_end_other'];

const GRADED = ['human', 'machine', 'machine-short', 'unknown', 'fax'];

/** A call duration as a number. It arrives as a string and can be absent. */
export function seconds(value) {
  const n = Number.parseInt(String(value ?? '0').trim(), 10);
  return Number.isFinite(n) ? n : 0;
}

/**
 * Put one call in an answering-machine bucket. Pure.
 *
 * A machine_start call of a few seconds is the misroute: a person answered,
 * detection called them a machine, and they hung up on the voicemail drop. The
 * machine_end_* family is not split the same way, because there Twilio waited
 * for the greeting to end and a short call means something else.
 */
export function bucket(call, shortSeconds = 8) {
  if (String(call.status ?? '').trim().toLowerCase() !== 'completed') return 'not-completed';

  const answered = String(call.answered_by ?? '').trim().toLowerCase();
  if (!answered) return 'no-amd';
  if (['human', 'fax', 'unknown'].includes(answered)) return answered;
  if (answered === 'machine_start') {
    return seconds(call.duration) <= shortSeconds ? 'machine-short' : 'machine';
  }
  if (MACHINE_END.includes(answered)) return 'machine';
  return 'other';
}

/**
 * Turn a tally of buckets into a verdict. Pure. The thresholds are arguments
 * rather than constants because they are defaults, not truths. Returns
 * [state, detail].
 */
export function verdict(tally, minCalls = 50, unknownPct = 3.0, machinePct = 40.0,
                        shortPct = 25.0) {
  const graded = GRADED.reduce((n, k) => n + (tally[k] ?? 0), 0);
  if (graded === 0) {
    return ['no-amd',
      'no call in this window carries answered_by, so machine detection was ' +
      'never requested and there is nothing to tune.'];
  }
  if (graded < minCalls) {
    return ['thin-sample',
      `only ${graded} graded call(s), under the ${minCalls} needed to read a ` +
      'distribution. Widen the window rather than trusting this.'];
  }

  const machines = (tally.machine ?? 0) + (tally['machine-short'] ?? 0);
  const unknownShare = (100 * (tally.unknown ?? 0)) / graded;
  const machineShare = (100 * machines) / graded;
  const shortShare = machines ? (100 * (tally['machine-short'] ?? 0)) / machines : 0;

  if (unknownShare > unknownPct) {
    return ['detection-timing-out',
      `${unknownShare.toFixed(1)}% of ${graded} graded call(s) came back ` +
      `unknown, over the ${unknownPct.toFixed(1)}% threshold. unknown is a ` +
      'timeout, not a category: detection ran out of time and your flow ' +
      'branched on a value it has no case for.'];
  }

  if (machineShare > machinePct && shortShare > shortPct) {
    return ['over-classifying',
      `${machineShare.toFixed(1)}% of ${graded} graded call(s) were called ` +
      `machines and ${shortShare.toFixed(1)}% of those lasted seconds. That ` +
      'short tail is people hanging up on a voicemail drop aimed at them.'];
  }

  if (machineShare > machinePct) {
    return ['machine-heavy',
      `${machineShare.toFixed(1)}% of ${graded} graded call(s) were machines, ` +
      `over the ${machinePct.toFixed(1)}% threshold, but only ` +
      `${shortShare.toFixed(1)}% of them were short. This looks like a list ` +
      'that really does reach voicemail, not a detector fault.'];
  }

  return ['healthy',
    `${graded} graded call(s): human ` +
    `${((100 * (tally.human ?? 0)) / graded).toFixed(1)}%, machine ` +
    `${machineShare.toFixed(1)}%, unknown ${unknownShare.toFixed(1)}%`];
}

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 = 20000) {
  let url = `${BASE}/Accounts/${account}/Calls.json`;
  let params = { 'StartTime>=': since, PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.calls ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  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 arg = (name, fallback) => {
    const i = process.argv.indexOf(name);
    return i === -1 ? fallback : Number(process.argv[i + 1]);
  };
  const days = arg('--days', 7);
  const shortSeconds = arg('--short-seconds', 8);

  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
  const calls = await listCalls(auth, account, since);
  if (calls.length === 0) {
    console.log(`no calls in the last ${days} day(s)`);
    return;
  }

  const tally = {};
  for (const c of calls) {
    const b = bucket(c, shortSeconds);
    tally[b] = (tally[b] ?? 0) + 1;
  }
  for (const name of Object.keys(tally).sort()) {
    console.log(`${name.padEnd(14)} ${tally[name]}`);
  }

  const [state, detail] = verdict(tally, arg('--min-calls', 50),
                                  arg('--unknown-pct', 3.0), arg('--machine-pct', 40.0));
  if (['healthy', 'no-amd', 'thin-sample'].includes(state)) {
    console.log(`${state}  ${detail}`);
    return;
  }
  console.warn(`${state}  ${detail}`);
  console.warn('  repair: on the outbound create request set ' +
               'MachineDetection=DetectMessageEnd, or raise ' +
               'MachineDetectionTimeout and MachineDetectionSpeechThreshold');
  console.warn('  repair: or set AsyncAmd=true with AsyncAmdStatusCallback so ' +
               'the call connects first and is reclassified after');
  process.exitCode = 1;
}

// Only run when invoked directly, so importing this module from the test file
// does not fire main(), fail on the missing credentials and set an exit code
// that fails the suite 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_amd_classification_audit import bucket, verdict


def test_short_machine_start_is_the_misroute_bucket():
    assert bucket({"status": "completed", "answered_by": "machine_start",
                   "duration": "4"}) == "machine-short"


def test_machine_end_beep_is_not_split_by_duration():
    # DetectMessageEnd waited for the greeting, so a short call means something
    # else and must not land in the misroute bucket.
    assert bucket({"status": "completed", "answered_by": "machine_end_beep",
                   "duration": "4"}) == "machine"


def test_calls_without_detection_stay_out_of_the_denominator():
    assert bucket({"status": "completed", "duration": "90"}) == "no-amd"
    assert bucket({"status": "no-answer", "answered_by": "unknown"}) == "not-completed"


def test_unknown_share_over_the_threshold_reads_as_a_timeout():
    state, detail = verdict({"human": 400, "machine": 80, "unknown": 30})
    assert state == "detection-timing-out"
    assert "timeout, not a category" in detail


def test_machine_heavy_with_a_short_tail_is_over_classifying():
    state, detail = verdict({"human": 100, "machine": 60, "machine-short": 40})
    assert state == "over-classifying"
    assert "hanging up" in detail


def test_machine_heavy_without_a_short_tail_is_a_list_not_a_detector():
    state, _ = verdict({"human": 100, "machine": 98, "machine-short": 2})
    assert state == "machine-heavy"


def test_thin_sample_is_reported_rather_than_scored():
    assert verdict({"human": 10, "machine": 4})[0] == "thin-sample"


def test_no_graded_calls_means_detection_was_never_asked_for():
    assert verdict({"no-amd": 900, "not-completed": 100})[0] == "no-amd"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { bucket, verdict } from './twilio-amd-classification-audit.mjs';

test('short machine_start is the misroute bucket', () => {
  assert.equal(
    bucket({ status: 'completed', answered_by: 'machine_start', duration: '4' }),
    'machine-short');
});

test('machine_end_beep is not split by duration', () => {
  assert.equal(
    bucket({ status: 'completed', answered_by: 'machine_end_beep', duration: '4' }),
    'machine');
});

test('calls without detection stay out of the denominator', () => {
  assert.equal(bucket({ status: 'completed', duration: '90' }), 'no-amd');
  assert.equal(bucket({ status: 'no-answer', answered_by: 'unknown' }), 'not-completed');
});

test('unknown share over the threshold reads as a timeout', () => {
  const [state, detail] = verdict({ human: 400, machine: 80, unknown: 30 });
  assert.equal(state, 'detection-timing-out');
  assert.match(detail, /timeout, not a category/);
});

test('machine heavy with a short tail is over classifying', () => {
  const [state, detail] = verdict({ human: 100, machine: 60, 'machine-short': 40 });
  assert.equal(state, 'over-classifying');
  assert.match(detail, /hanging up/);
});

test('machine heavy without a short tail is a list not a detector', () => {
  assert.equal(verdict({ human: 100, machine: 98, 'machine-short': 2 })[0],
               'machine-heavy');
});

test('thin sample is reported rather than scored', () => {
  assert.equal(verdict({ human: 10, machine: 4 })[0], 'thin-sample');
});

test('no graded calls means detection was never asked for', () => {
  assert.equal(verdict({ 'no-amd': 900, 'not-completed': 100 })[0], 'no-amd');
});