phone numbers with no traffic still bill every month

"""Report Twilio phone numbers carrying no traffic, priced per year.

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 send messages 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_idle_numbers_audit")

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

# One page of traffic settles the question. A number with this many messages in
# the window is in use, and the exact figure would not change the verdict.
PROBE = 50


def monthly_rate(records, number_count, override=None):
    """Dollars per number per month.

    IncomingPhoneNumbers carries no price, so the rate has to come from the
    monthly usage record for the phonenumbers category divided by the numbers on
    the account. That is an average and it under-reports toll-free and short
    codes, which cost more than a local number. Pass --monthly-cost when you
    know the real figure.

    Usage prices arrive as strings and the sign convention differs between usage
    records and the balance resource, so take the magnitude.
    """
    if override is not None:
        return max(0.0, float(override))
    rows = [r for r in records
            if str(r.get("category") or "") == "phonenumbers"]
    if not rows or not number_count:
        return 0.0
    latest = max(rows, key=lambda r: str(r.get("start_date") or ""))
    try:
        price = abs(float(latest.get("price") or 0))
    except (TypeError, ValueError):
        return 0.0
    return price / float(number_count)


def verdict(activity, rate, window_days=90, min_traffic=5, flag_above=24.0):
    """Classify one number by what it carried against what it costs.

    activity: counts keyed outbound_messages, inbound_messages, outbound_calls,
    inbound_calls. rate: dollars per month. Pure, so the thresholds and the
    arithmetic are visible and testable rather than buried in a request loop.

    Returns (state, detail, annual_cost).
    """
    out = (int(activity.get("outbound_messages") or 0)
           + int(activity.get("outbound_calls") or 0))
    inb = (int(activity.get("inbound_messages") or 0)
           + int(activity.get("inbound_calls") or 0))
    annual = max(0.0, float(rate)) * 12.0
    window_cost = max(0.0, float(rate)) * (float(window_days) / 30.44)

    if out == 0 and inb == 0:
        if annual >= flag_above:
            return ("idle-costly",
                    "no messages and no calls either way in %d days, and it is "
                    "one of the more expensive numbers on the account at $%.2f "
                    "a year. Release this one first."
                    % (window_days, annual),
                    annual)
        return ("idle",
                "no messages and no calls either way in %d days. $%.2f a year "
                "for a number nothing touches." % (window_days, annual),
                annual)

    if out == 0:
        return ("inbound-only",
                "%d inbound event(s) in %d days and nothing outbound. Often "
                "deliberate, so confirm before releasing: $%.2f a year."
                % (inb, window_days, annual),
                annual)

    total = out + inb
    if total < min_traffic:
        per = window_cost / total if total else window_cost
        return ("trickle",
                "%d event(s) in %d days at $%.2f of rent, which is $%.2f per "
                "message or call. Cheaper to fold this traffic onto a number "
                "you already keep." % (total, window_days, window_cost, per),
                annual)

    return ("active",
            "%d outbound and %d inbound event(s) in %d days"
            % (out, inb, window_days),
            annual)


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_numbers(session, account, limit):
    """Page IncomingPhoneNumbers. next_page_uri is a path, not an absolute URL."""
    url = "%s/Accounts/%s/IncomingPhoneNumbers.json" % (BASE, account)
    params = {"PageSize": 100}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("incoming_phone_numbers", []))
        nxt = page.get("next_page_uri")
        url, params = (HOST + nxt) if nxt else None, {}
    return out[:limit]


def activity_for(session, account, e164, since):
    """Four small reads: messages and calls, each direction."""
    msgs = "%s/Accounts/%s/Messages.json" % (BASE, account)
    calls = "%s/Accounts/%s/Calls.json" % (BASE, account)
    params = {"PageSize": PROBE}
    out = {}
    out["outbound_messages"] = len(get(session, msgs, **dict(
        params, **{"From": e164, "DateSent>": since})).get("messages", []))
    out["inbound_messages"] = len(get(session, msgs, **dict(
        params, **{"To": e164, "DateSent>": since})).get("messages", []))
    out["outbound_calls"] = len(get(session, calls, **dict(
        params, **{"From": e164, "StartTime>": since})).get("calls", []))
    out["inbound_calls"] = len(get(session, calls, **dict(
        params, **{"To": e164, "StartTime>": since})).get("calls", []))
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=90,
                    help="traffic window, in days")
    ap.add_argument("--max-numbers", type=int, default=200,
                    help="stop after this many numbers; each costs four reads")
    ap.add_argument("--monthly-cost", type=float, default=None,
                    help="dollars per number per month, overriding the average")
    ap.add_argument("--min-traffic", type=int, default=5,
                    help="fewer events than this in the window reads as a trickle")
    ap.add_argument("--flag-above", type=float, default=24.0,
                    help="annual dollars above which an idle number is urgent")
    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)

    numbers = list_numbers(session, account, args.max_numbers)
    if not numbers:
        log.info("no phone numbers on this account")
        return 0

    usage = get(session, "%s/Accounts/%s/Usage/Records/Monthly.json"
                % (BASE, account), Category="phonenumbers")
    rate = monthly_rate(usage.get("usage_records", []), len(numbers),
                        args.monthly_cost)
    log.info("%d number(s) at about $%.2f each per month", len(numbers), rate)

    since = (dt.date.today() - dt.timedelta(days=args.days)).isoformat()
    idle, wasted = 0, 0.0
    for n in numbers:
        e164 = n.get("phone_number", "?")
        state, detail, annual = verdict(
            activity_for(session, account, e164, since), rate,
            args.days, args.min_traffic, args.flag_above)
        label = n.get("friendly_name") or e164
        line = "%-13s %s (%s)  %s" % (state, e164, label, detail)
        if state == "active":
            log.info(line)
            continue
        log.warning(line)
        if state.startswith("idle"):
            idle += 1
            wasted += annual
            log.warning("  repair: release it with a delete on %s/Accounts/%s"
                        "/IncomingPhoneNumbers/%s.json. Release is free and "
                        "recoverable for a short window.", BASE, account,
                        n.get("sid"))

    log.info("%d number(s), %d idle, $%.2f/year in rent for numbers with no "
             "traffic", len(numbers), idle, wasted)
    return 1 if idle else 0


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

/**
 * Report Twilio phone numbers carrying no traffic, priced per year.
 *
 * 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`;

// One page of traffic settles the question. A number with this many messages in
// the window is in use, and the exact figure would not change the verdict.
const PROBE = 50;

/**
 * Dollars per number per month. IncomingPhoneNumbers carries no price, so the
 * rate comes from the monthly usage record for the phonenumbers category
 * divided by the numbers on the account. That is an average and it
 * under-reports toll-free and short codes. Prices arrive as strings and the
 * sign convention differs between resources, so take the magnitude.
 */
export function monthlyRate(records, numberCount, override = null) {
  if (override !== null && override !== undefined) return Math.max(0, Number(override));
  const rows = records.filter((r) => String(r.category ?? '') === 'phonenumbers');
  if (!rows.length || !numberCount) return 0;
  const latest = rows.reduce((a, b) =>
    String(b.start_date ?? '') > String(a.start_date ?? '') ? b : a);
  const price = Math.abs(Number(latest.price ?? 0));
  if (!Number.isFinite(price)) return 0;
  return price / Number(numberCount);
}

/**
 * Classify one number by what it carried against what it costs. Pure, so the
 * thresholds and the arithmetic are visible and testable.
 * Returns [state, detail, annualCost].
 */
export function verdict(activity, rate, windowDays = 90, minTraffic = 5, flagAbove = 24) {
  const out = Number(activity.outbound_messages ?? 0) + Number(activity.outbound_calls ?? 0);
  const inb = Number(activity.inbound_messages ?? 0) + Number(activity.inbound_calls ?? 0);
  const annual = Math.max(0, Number(rate)) * 12;
  const windowCost = Math.max(0, Number(rate)) * (Number(windowDays) / 30.44);

  if (out === 0 && inb === 0) {
    if (annual >= flagAbove) {
      return ['idle-costly',
        `no messages and no calls either way in ${windowDays} days, and it is ` +
        `one of the more expensive numbers on the account at $${annual.toFixed(2)} ` +
        'a year. Release this one first.', annual];
    }
    return ['idle',
      `no messages and no calls either way in ${windowDays} days. ` +
      `$${annual.toFixed(2)} a year for a number nothing touches.`, annual];
  }

  if (out === 0) {
    return ['inbound-only',
      `${inb} inbound event(s) in ${windowDays} days and nothing outbound. Often ` +
      `deliberate, so confirm before releasing: $${annual.toFixed(2)} a year.`, annual];
  }

  const total = out + inb;
  if (total < minTraffic) {
    const per = total ? windowCost / total : windowCost;
    return ['trickle',
      `${total} event(s) in ${windowDays} days at $${windowCost.toFixed(2)} of ` +
      `rent, which is $${per.toFixed(2)} per message or call. Cheaper to fold ` +
      'this traffic onto a number you already keep.', annual];
  }

  return ['active',
    `${out} outbound and ${inb} inbound event(s) in ${windowDays} days`, annual];
}

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 listNumbers(auth, account, limit = 200) {
  let url = `${BASE}/Accounts/${account}/IncomingPhoneNumbers.json`;
  let params = { PageSize: 100 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.incoming_phone_numbers ?? []));
    url = page.next_page_uri ? HOST + page.next_page_uri : null;
    params = {};
  }
  return out.slice(0, limit);
}

async function activityFor(auth, account, e164, since) {
  const msgs = `${BASE}/Accounts/${account}/Messages.json`;
  const calls = `${BASE}/Accounts/${account}/Calls.json`;
  const p = { PageSize: PROBE };
  const om = await get(auth, msgs, { ...p, From: e164, 'DateSent>': since });
  const im = await get(auth, msgs, { ...p, To: e164, 'DateSent>': since });
  const oc = await get(auth, calls, { ...p, From: e164, 'StartTime>': since });
  const ic = await get(auth, calls, { ...p, To: e164, 'StartTime>': since });
  return {
    outbound_messages: (om.messages ?? []).length,
    inbound_messages: (im.messages ?? []).length,
    outbound_calls: (oc.calls ?? []).length,
    inbound_calls: (ic.calls ?? []).length,
  };
}

function flag(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 = flag('--days', 90);
  const minTraffic = flag('--min-traffic', 5);
  const flagAbove = flag('--flag-above', 24);
  const override = process.argv.includes('--monthly-cost')
    ? flag('--monthly-cost', null) : null;

  const numbers = await listNumbers(auth, account);
  if (numbers.length === 0) {
    console.log('no phone numbers on this account');
    return;
  }

  const usage = await get(auth, `${BASE}/Accounts/${account}/Usage/Records/Monthly.json`,
                          { Category: 'phonenumbers' });
  const rate = monthlyRate(usage.usage_records ?? [], numbers.length, override);
  console.log(`${numbers.length} number(s) at about $${rate.toFixed(2)} each per month`);

  const since = new Date(Date.now() - days * 86400000).toISOString().slice(0, 10);
  let idle = 0;
  let wasted = 0;
  for (const n of numbers) {
    const e164 = n.phone_number ?? '?';
    const act = await activityFor(auth, account, e164, since);
    const [state, detail, annual] = verdict(act, rate, days, minTraffic, flagAbove);
    const label = n.friendly_name || e164;
    const line = `${state.padEnd(13)} ${e164} (${label})  ${detail}`;
    if (state === 'active') { console.log(line); continue; }
    console.warn(line);
    if (state.startsWith('idle')) {
      idle += 1;
      wasted += annual;
      console.warn(`  repair: release it with a delete on ${BASE}/Accounts/` +
                   `${account}/IncomingPhoneNumbers/${n.sid}.json. Release is ` +
                   'free and recoverable for a short window.');
    }
  }

  console.log(`${numbers.length} number(s), ${idle} idle, $${wasted.toFixed(2)}` +
              '/year in rent for numbers with no traffic');
  process.exitCode = idle ? 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_idle_numbers_audit import monthly_rate, verdict

NOTHING = {"outbound_messages": 0, "inbound_messages": 0,
           "outbound_calls": 0, "inbound_calls": 0}


def test_silent_number_is_idle_and_priced_for_the_year():
    state, detail, annual = verdict(NOTHING, 1.15)
    assert state == "idle"
    assert round(annual, 2) == 13.80
    assert "13.80" in detail


def test_expensive_idle_number_is_escalated():
    # A toll-free number rents for more, so it is the one to release first.
    state, _, annual = verdict(NOTHING, 2.15, flag_above=24.0)
    assert state == "idle-costly"
    assert annual > 24.0


def test_inbound_only_number_is_not_reported_as_idle():
    # Checking From= alone is how somebody releases a working support line.
    act = dict(NOTHING, inbound_calls=31)
    state, detail, _ = verdict(act, 1.15)
    assert state == "inbound-only"
    assert "confirm before releasing" in detail


def test_a_handful_of_messages_reports_cost_per_message():
    act = dict(NOTHING, outbound_messages=3)
    state, detail, _ = verdict(act, 1.15, window_days=90, min_traffic=5)
    assert state == "trickle"
    assert "per message or call" in detail


def test_busy_number_is_active():
    act = dict(NOTHING, outbound_messages=50, inbound_messages=12)
    state, _, _ = verdict(act, 1.15)
    assert state == "active"


def test_monthly_rate_uses_the_newest_month_and_divides_by_the_numbers():
    records = [
        {"category": "phonenumbers", "start_date": "2026-06-01", "price": "23.00"},
        {"category": "phonenumbers", "start_date": "2026-07-01", "price": "46.00"},
    ]
    assert monthly_rate(records, 40) == 1.15


def test_monthly_rate_takes_the_magnitude_of_a_signed_price():
    records = [{"category": "phonenumbers", "start_date": "2026-07-01",
                "price": "-46.00"}]
    assert monthly_rate(records, 40) == 1.15


def test_monthly_rate_override_wins_and_survives_an_empty_account():
    assert monthly_rate([], 0, override=2.0) == 2.0
    assert monthly_rate([], 0) == 0.0

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { monthlyRate, verdict } from './twilio-idle-numbers-audit.mjs';

const NOTHING = {
  outbound_messages: 0, inbound_messages: 0, outbound_calls: 0, inbound_calls: 0,
};

test('silent number is idle and priced for the year', () => {
  const [state, detail, annual] = verdict(NOTHING, 1.15);
  assert.equal(state, 'idle');
  assert.equal(Number(annual.toFixed(2)), 13.80);
  assert.match(detail, /13\.80/);
});

test('expensive idle number is escalated', () => {
  const [state, , annual] = verdict(NOTHING, 2.15, 90, 5, 24);
  assert.equal(state, 'idle-costly');
  assert.ok(annual > 24);
});

test('inbound only number is not reported as idle', () => {
  const [state, detail] = verdict({ ...NOTHING, inbound_calls: 31 }, 1.15);
  assert.equal(state, 'inbound-only');
  assert.match(detail, /confirm before releasing/);
});

test('a handful of messages reports cost per message', () => {
  const [state, detail] = verdict({ ...NOTHING, outbound_messages: 3 }, 1.15, 90, 5);
  assert.equal(state, 'trickle');
  assert.match(detail, /per message or call/);
});

test('busy number is active', () => {
  const [state] = verdict(
    { ...NOTHING, outbound_messages: 50, inbound_messages: 12 }, 1.15);
  assert.equal(state, 'active');
});

test('monthlyRate uses the newest month and divides by the numbers', () => {
  const records = [
    { category: 'phonenumbers', start_date: '2026-06-01', price: '23.00' },
    { category: 'phonenumbers', start_date: '2026-07-01', price: '46.00' },
  ];
  assert.equal(monthlyRate(records, 40), 1.15);
});

test('monthlyRate takes the magnitude of a signed price', () => {
  const records = [
    { category: 'phonenumbers', start_date: '2026-07-01', price: '-46.00' },
  ];
  assert.equal(monthlyRate(records, 40), 1.15);
});

test('monthlyRate override wins and survives an empty account', () => {
  assert.equal(monthlyRate([], 0, 2), 2);
  assert.equal(monthlyRate([], 0), 0);
});