a recording row says absent and there is no media behind it

"""Report Twilio recordings whose media was never produced.

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

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

# Where the status callback lives for each mechanism that can start a recording.
# The finding is only half useful without this sentence: knowing a recording is
# missing does not tell you which of five places the callback belongs in.
SOURCES = {
    "DialVerb": "recordingStatusCallback is an attribute on the Dial verb",
    "RecordVerb": "recordingStatusCallback is an attribute on the Record verb",
    "Conference": "recordingStatusCallback is an attribute on the Conference noun",
    "OutboundAPI": "RecordingStatusCallback is a parameter on the call create request",
    "StartCallRecordingAPI":
        "RecordingStatusCallback is a parameter on the recording create request",
    "StartConferenceRecordingAPI":
        "RecordingStatusCallback is a parameter on the recording create request",
    "Trunking": "recording is configured on the trunk itself, so there is no "
                "per-call attribute to add here",
}


def source_meaning(source):
    """Where the recording status callback is configured for this source. Pure."""
    key = str(source or "").strip()
    return SOURCES.get(
        key, "the source is not one this script recognises, so check how the "
             "recording was started before deciding where the callback goes")


def seconds(value):
    """A recording 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 verdict(recording):
    """Classify one Recording. Pure, so the rules can be tested without a
    network.

    Returns (state, detail). The two states worth acting on are absent, where no
    media was ever produced, and empty, where media exists and holds no audio.
    Both fail the person who asks for the call, and only the first has an
    error_code to explain itself.
    """
    status = str(recording.get("status") or "").strip().lower()
    source = str(recording.get("source") or "").strip()
    code = str(recording.get("error_code") or "").strip()

    if status == "absent":
        why = ("error_code %s" % code) if code else \
            "no error_code, which is unusual on an absent row"
        return ("absent",
                "%s asked for this recording and no media was produced (%s). "
                "The call itself completed normally, so nothing else about it "
                "looks wrong. %s."
                % (source or "An unnamed source", why, source_meaning(source)))

    if status in ("processing", "in-progress"):
        return ("in-flight",
                "status is %s: the media is still being written, so this is a "
                "verdict about a moment rather than a fault." % status)

    if status == "deleted":
        return ("deleted",
                "the media has been deleted. The row survives deletion, so a "
                "check that only looks for the recording's existence will keep "
                "reporting this call as recorded.")

    if status == "completed":
        if seconds(recording.get("duration")) <= 0:
            return ("empty",
                    "completed with a duration of zero: the media exists, it "
                    "will play, and there is no audio in it. It passes every "
                    "check for presence and fails the only one that matters.")
        return ("stored", "completed with %ds of media."
                % seconds(recording.get("duration")))

    return ("other", "status is %s, which this script has no rule for."
            % (status or "empty"))


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


def list_alerts(session, since, limit, log_level):
    """Page the Monitor alerts at one log level. next_page_url is absolute."""
    url = MONITOR + "/Alerts"
    params = {"LogLevel": log_level, "StartDate": since, "PageSize": 1000}
    out = []
    while url and len(out) < limit:
        page = get(session, url, **params)
        out.extend(page.get("alerts", []))
        url = (page.get("meta") or {}).get("next_page_url")
        params = {}
    return out[:limit]


def alerts_by_call(session, since, limit):
    """Alert error codes keyed by the call sid they were raised against.

    Both log levels, because several voice failures are logged at warning and an
    error-only sweep will report that nothing else was wrong with these calls.
    """
    out = {}
    for level in ("error", "warning"):
        for a in list_alerts(session, since, limit, level):
            sid = str(a.get("resource_sid") or "")
            if sid.startswith("CA"):
                out.setdefault(sid, set()).add(str(a.get("error_code") or "?"))
    return out


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=30, help="window to audit")
    ap.add_argument("--max-recordings", type=int, default=20000,
                    help="stop after this many recordings")
    ap.add_argument("--with-alerts", action="store_true",
                    help="also sweep Alerts and match on the call sid")
    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()
    recordings = list_recordings(session, account, since, args.max_recordings)
    if not recordings:
        log.info("no recordings in the last %d day(s)", args.days)
        return 0

    alerts = {}
    if args.with_alerts:
        # Alerts are retained 30 days, so a longer recording window is only
        # partially covered by this cross-reference.
        alerts = alerts_by_call(session, since, 10000)

    absent = 0
    empty = 0
    by_source = {}
    for rec in recordings:
        state, detail = verdict(rec)
        if state not in ("absent", "empty"):
            continue
        if state == "absent":
            absent += 1
            src = str(rec.get("source") or "unknown")
            by_source[src] = by_source.get(src, 0) + 1
        else:
            empty += 1
        log.warning("%-9s %s  %s", state, rec.get("sid"), detail)
        call_sid = str(rec.get("call_sid") or "")
        if call_sid in alerts:
            log.warning("  same call raised alert(s): %s",
                        ", ".join(sorted(alerts[call_sid])))

    log.info("%d recording(s), %d absent, %d empty",
             len(recordings), absent, empty)
    if not (absent or empty):
        return 0
    if by_source:
        log.warning("absent by source: %s",
                    ", ".join("%s=%d" % kv for kv in sorted(by_source.items())))
    log.warning("  repair: attach a recording status callback where the "
                "recording is started, so the next failure alerts on the day "
                "instead of at the audit")
    log.warning("  repair: reconcile your own recording table against status, "
                "not against the presence of a recording sid")
    return 1


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

/**
 * Report Twilio recordings whose media was never produced.
 *
 * 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';

// Where the status callback lives for each mechanism that can start a
// recording. Knowing a recording is missing does not tell you which of five
// places the callback belongs in.
const SOURCES = {
  DialVerb: 'recordingStatusCallback is an attribute on the Dial verb',
  RecordVerb: 'recordingStatusCallback is an attribute on the Record verb',
  Conference: 'recordingStatusCallback is an attribute on the Conference noun',
  OutboundAPI: 'RecordingStatusCallback is a parameter on the call create request',
  StartCallRecordingAPI:
    'RecordingStatusCallback is a parameter on the recording create request',
  StartConferenceRecordingAPI:
    'RecordingStatusCallback is a parameter on the recording create request',
  Trunking: 'recording is configured on the trunk itself, so there is no ' +
            'per-call attribute to add here',
};

/** Where the recording status callback is configured for this source. Pure. */
export function sourceMeaning(source) {
  const key = String(source ?? '').trim();
  return SOURCES[key] ??
    'the source is not one this script recognises, so check how the recording ' +
    'was started before deciding where the callback goes';
}

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

/**
 * Classify one Recording. Pure. Returns [state, detail]. The two states worth
 * acting on are absent, where no media was ever produced, and empty, where
 * media exists and holds no audio.
 */
export function verdict(recording) {
  const status = String(recording.status ?? '').trim().toLowerCase();
  const source = String(recording.source ?? '').trim();
  const code = String(recording.error_code ?? '').trim();

  if (status === 'absent') {
    const why = code ? `error_code ${code}`
                     : 'no error_code, which is unusual on an absent row';
    return ['absent',
      `${source || 'An unnamed source'} asked for this recording and no media ` +
      `was produced (${why}). The call itself completed normally, so nothing ` +
      `else about it looks wrong. ${sourceMeaning(source)}.`];
  }

  if (status === 'processing' || status === 'in-progress') {
    return ['in-flight',
      `status is ${status}: the media is still being written, so this is a ` +
      'verdict about a moment rather than a fault.'];
  }

  if (status === 'deleted') {
    return ['deleted',
      'the media has been deleted. The row survives deletion, so a check that ' +
      'only looks for the recording\'s existence will keep reporting this call ' +
      'as recorded.'];
  }

  if (status === 'completed') {
    if (seconds(recording.duration) <= 0) {
      return ['empty',
        'completed with a duration of zero: the media exists, it will play, ' +
        'and there is no audio in it. It passes every check for presence and ' +
        'fails the only one that matters.'];
    }
    return ['stored', `completed with ${seconds(recording.duration)}s of media.`];
  }

  return ['other', `status is ${status || 'empty'}, which this script has no rule for.`];
}

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

async function listAlerts(auth, since, limit, logLevel) {
  let url = `${MONITOR}/Alerts`;
  let params = { LogLevel: logLevel, StartDate: since, PageSize: 1000 };
  const out = [];
  while (url && out.length < limit) {
    const page = await get(auth, url, params);
    out.push(...(page.alerts ?? []));
    url = page.meta?.next_page_url ?? null;
    params = {};
  }
  return out.slice(0, limit);
}

/** Alert error codes keyed by call sid, swept at both log levels. */
export async function alertsByCall(auth, since, limit = 10000) {
  const out = new Map();
  for (const level of ['error', 'warning']) {
    for (const a of await listAlerts(auth, since, limit, level)) {
      const sid = String(a.resource_sid ?? '');
      if (!sid.startsWith('CA')) continue;
      if (!out.has(sid)) out.set(sid, new Set());
      out.get(sid).add(String(a.error_code ?? '?'));
    }
  }
  return out;
}

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 i = process.argv.indexOf('--days');
  const days = i === -1 ? 30 : Number(process.argv[i + 1]);

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

  // Alerts are retained 30 days, so a longer recording window is only partially
  // covered by this cross-reference.
  const alerts = process.argv.includes('--with-alerts')
    ? await alertsByCall(auth, since)
    : new Map();

  let absent = 0;
  let empty = 0;
  const bySource = new Map();
  for (const rec of recordings) {
    const [state, detail] = verdict(rec);
    if (state !== 'absent' && state !== 'empty') continue;
    if (state === 'absent') {
      absent += 1;
      const src = String(rec.source ?? 'unknown');
      bySource.set(src, (bySource.get(src) ?? 0) + 1);
    } else {
      empty += 1;
    }
    console.warn(`${state.padEnd(9)} ${rec.sid}  ${detail}`);
    const callSid = String(rec.call_sid ?? '');
    if (alerts.has(callSid)) {
      console.warn(`  same call raised alert(s): ${[...alerts.get(callSid)].sort().join(', ')}`);
    }
  }

  console.log(`${recordings.length} recording(s), ${absent} absent, ${empty} empty`);
  if (!absent && !empty) return;
  if (bySource.size) {
    console.warn('absent by source: ' +
      [...bySource.entries()].sort().map(([k, v]) => `${k}=${v}`).join(', '));
  }
  console.warn('  repair: attach a recording status callback where the recording ' +
               'is started, so the next failure alerts on the day instead of at ' +
               'the audit');
  console.warn('  repair: reconcile your own recording table against status, not ' +
               'against the presence of a recording sid');
  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_absent_recordings_audit import source_meaning, verdict


def test_absent_row_names_the_error_code_and_the_source():
    state, detail = verdict({"status": "absent", "error_code": 12400,
                             "source": "DialVerb"})
    assert state == "absent"
    assert "error_code 12400" in detail
    assert "Dial verb" in detail


def test_absent_row_without_an_error_code_says_so():
    state, detail = verdict({"status": "absent", "source": "RecordVerb"})
    assert state == "absent"
    assert "unusual" in detail


def test_completed_with_zero_duration_is_its_own_finding():
    state, detail = verdict({"status": "completed", "duration": "0"})
    assert state == "empty"
    assert "no audio" in detail


def test_completed_with_media_is_stored():
    assert verdict({"status": "completed", "duration": "671"})[0] == "stored"


def test_in_progress_is_a_moment_not_a_fault():
    assert verdict({"status": "processing"})[0] == "in-flight"


def test_deleted_row_survives_the_media():
    state, detail = verdict({"status": "deleted"})
    assert state == "deleted"
    assert "recording sid" not in detail


def test_trunking_source_has_nowhere_to_put_a_per_call_callback():
    assert "trunk itself" in source_meaning("Trunking")


def test_unrecognised_source_does_not_invent_a_place_for_the_callback():
    assert "not one this script recognises" in source_meaning("SomethingNew")

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { sourceMeaning, verdict } from './twilio-absent-recordings-audit.mjs';

test('absent row names the error code and the source', () => {
  const [state, detail] = verdict({ status: 'absent', error_code: 12400,
                                    source: 'DialVerb' });
  assert.equal(state, 'absent');
  assert.match(detail, /error_code 12400/);
  assert.match(detail, /Dial verb/);
});

test('absent row without an error code says so', () => {
  const [state, detail] = verdict({ status: 'absent', source: 'RecordVerb' });
  assert.equal(state, 'absent');
  assert.match(detail, /unusual/);
});

test('completed with zero duration is its own finding', () => {
  const [state, detail] = verdict({ status: 'completed', duration: '0' });
  assert.equal(state, 'empty');
  assert.match(detail, /no audio/);
});

test('completed with media is stored', () => {
  assert.equal(verdict({ status: 'completed', duration: '671' })[0], 'stored');
});

test('in progress is a moment not a fault', () => {
  assert.equal(verdict({ status: 'processing' })[0], 'in-flight');
});

test('deleted row survives the media', () => {
  assert.equal(verdict({ status: 'deleted' })[0], 'deleted');
});

test('trunking source has nowhere to put a per call callback', () => {
  assert.match(sourceMeaning('Trunking'), /trunk itself/);
});

test('unrecognised source does not invent a place for the callback', () => {
  assert.match(sourceMeaning('SomethingNew'), /not one this script recognises/);
});