""" NetSapiens CDR puller with an incremental SQLite accumulator. Modes: seed_from_csv(path) one-time load of an existing export (CSV or .csv.gz) pull_range(start, end) pull a date range from the API, upsert by id incremental() pull everything newer than the DB's max start-datetime The SQLite table `cdrs` has one TEXT column per API field (99) plus the id PK. Upsert-by-id makes every pull idempotent: re-pulling an overlapping window never duplicates rows. """ import csv import gzip import os import sqlite3 import sys import time import config # All 99 fields the API returns, in order (validated against the live v2 API). FIELDS = [ "id", "domain", "reseller", "call-account-code", "call-answer-datetime", "call-audio-codec", "call-audio-relay-side-a-local-port", "call-audio-relay-side-a-packet-count", "call-audio-relay-side-a-remote-ip", "call-audio-relay-side-b-packet-count", "call-audio-relay-side-b-remote-ip", "call-batch-answer-datetime", "call-batch-on-hold-duration-seconds", "call-batch-sequence-marker", "call-batch-start-datetime", "call-batch-total-duration-seconds", "call-direction", "call-disconnect-datetime", "call-disconnect-reason-text", "call-disposition", "call-disposition-direction", "call-disposition-notes", "call-disposition-reason", "call-disposition-submitted-datetime", "call-fax-codec", "call-fax-relay-side-a-local-port", "call-fax-relay-side-a-packet-count", "call-fax-relay-side-a-remote-ip", "call-fax-relay-side-b-packet-count", "call-fax-relay-side-b-remote-ip", "call-intelligence-job-id", "call-intelligence-percent-negative", "call-intelligence-percent-neutral", "call-intelligence-percent-positive", "call-intelligence-topics-top", "call-leg-ordinal-index", "call-on-hold-duration-seconds", "call-orig-call-id", "call-orig-caller-id", "call-orig-department", "call-orig-domain", "call-orig-from-host", "call-orig-from-name", "call-orig-from-uri", "call-orig-from-user", "call-orig-ip-address", "call-orig-match-uri", "call-orig-pre-routing-uri", "call-orig-request-host", "call-orig-request-uri", "call-orig-request-user", "call-orig-reseller", "call-orig-site", "call-orig-to-host", "call-orig-to-uri", "call-orig-to-user", "call-orig-user", "call-parent-call-id", "call-parent-cdr-id", "call-record-creation-datetime", "call-ringing-datetime", "call-routing-class", "call-routing-match-uri", "call-server-mac-address", "call-start-datetime", "call-tag", "call-talking-duration-seconds", "call-term-call-id", "call-term-caller-id", "call-term-department", "call-term-domain", "call-term-ip-address", "call-term-match-uri", "call-term-pre-routing-uri", "call-term-reseller", "call-term-site", "call-term-to-uri", "call-term-user", "call-through-action", "call-through-call-id", "call-through-caller-id", "call-through-department", "call-through-domain", "call-through-reseller", "call-through-site", "call-through-uri", "call-through-user", "call-total-duration-seconds", "call-video-codec", "call-video-relay-side-a-local-port", "call-video-relay-side-a-packet-count", "call-video-relay-side-a-remote-ip", "call-video-relay-side-b-packet-count", "call-video-relay-side-b-remote-ip", "core-server", "ending_sentiment", "hide-from-results", "is-trace-expected", "prefilled-trace-api", ] # Quote column names because many contain hyphens. "id" is the PK, declared separately. _COLS_SQL = ", ".join(f'"{c}" TEXT' for c in FIELDS if c != "id") _COL_NAMES = ", ".join(f'"{c}"' for c in FIELDS) _PLACEHOLDERS = ", ".join("?" for _ in FIELDS) def connect(db_path=None): conn = sqlite3.connect(db_path or config.DB_PATH) conn.execute(f'CREATE TABLE IF NOT EXISTS cdrs ("id" TEXT PRIMARY KEY, {_COLS_SQL})') conn.execute('CREATE INDEX IF NOT EXISTS idx_start ON cdrs("call-start-datetime")') conn.commit() return conn def _row_values(rec): return [("" if rec.get(f) is None else str(rec.get(f))) for f in FIELDS] def _upsert(conn, records): sql = (f'INSERT INTO cdrs ({_COL_NAMES}) VALUES ({_PLACEHOLDERS}) ' f'ON CONFLICT("id") DO UPDATE SET ' + ", ".join(f'"{c}"=excluded."{c}"' for c in FIELDS if c != "id")) conn.executemany(sql, [_row_values(r) for r in records]) conn.commit() def row_count(conn): return conn.execute("SELECT COUNT(*) FROM cdrs").fetchone()[0] def max_start_datetime(conn): if row_count(conn) == 0: return None return conn.execute( 'SELECT MAX("call-start-datetime") FROM cdrs WHERE "call-start-datetime" != ""' ).fetchone()[0] def prune_older_than(months=None, db_path=None): """Delete CDRs older than `months` (rolling retention). Returns rows removed.""" from datetime import date months = months if months is not None else config.RETENTION_MONTHS today = date.today() # first day of the month, `months` months back total = today.year * 12 + (today.month - 1) - months cutoff = f"{total // 12:04d}-{total % 12 + 1:02d}-01" conn = connect(db_path) cur = conn.execute('DELETE FROM cdrs WHERE "call-start-datetime" != "" ' 'AND "call-start-datetime" < ?', (cutoff,)) removed = cur.rowcount conn.commit() print(f"prune_older_than({months}mo): removed {removed:,} rows older than {cutoff} " f"(DB now {row_count(conn):,})") conn.close() return removed # --------------------------------------------------------------------------- # Seed from an existing CSV / gzip export # --------------------------------------------------------------------------- def seed_from_csv(csv_path, db_path=None): conn = connect(db_path) opener = gzip.open if csv_path.endswith(".gz") else open n = 0 with opener(csv_path, "rt", encoding="utf-8", newline="") as f: reader = csv.DictReader(f) batch = [] for rec in reader: batch.append(rec) if len(batch) >= 2000: _upsert(conn, batch); n += len(batch); batch = [] if batch: _upsert(conn, batch); n += len(batch) print(f"seed_from_csv: upserted {n:,} rows from {os.path.basename(csv_path)} " f"(DB now {row_count(conn):,})") conn.close() return n # --------------------------------------------------------------------------- # API pulls # --------------------------------------------------------------------------- def _headers(): return {"Authorization": f"Bearer {config.bearer_token()}", "Accept": "application/json"} def _fetch_page(requests, start, dt_start, dt_end): url = f"{config.API_BASE}/domains/{config.DOMAIN}/cdrs" params = {"datetime-start": dt_start, "datetime-end": dt_end, "start": start, "limit": config.PAGE_SIZE} for attempt in range(1, config.MAX_RETRIES + 1): try: r = requests.get(url, headers=_headers(), params=params, timeout=config.REQUEST_TIMEOUT) r.raise_for_status() data = r.json() if isinstance(data, dict) and "items" in data: return data["items"] return data if isinstance(data, list) else [] except Exception as e: # noqa: BLE001 - network/json errors all retried print(f" [!] offset {start} attempt {attempt}/{config.MAX_RETRIES}: {e}") if attempt < config.MAX_RETRIES: time.sleep(2 ** attempt) print(f" [X] giving up on offset {start}") return [] def pull_range(dt_start, dt_end, db_path=None): """Pull [dt_start, dt_end] from the API and upsert. ISO8601 with offset.""" import requests # local import so seed/analyze don't require the package conn = connect(db_path) before = row_count(conn) start, page, total_new = 0, 0, 0 print(f"pull_range: {dt_start} -> {dt_end}") while True: records = _fetch_page(requests, start, dt_start, dt_end) if not records: break _upsert(conn, records) total_new += len(records) page += 1 print(f" page {page}: +{len(records)} (cumulative {total_new})") if len(records) < config.PAGE_SIZE: break start += config.PAGE_SIZE time.sleep(config.DELAY_BETWEEN_PAGES) after = row_count(conn) print(f"pull_range done: fetched {total_new:,}; DB {before:,} -> {after:,} " f"({after - before:,} new)") conn.close() return total_new def incremental(db_path=None, until=None): """Pull everything newer than the DB's max start-datetime up to `until` (now).""" from datetime import datetime, timezone, timedelta conn = connect(db_path) last = max_start_datetime(conn) conn.close() if last is None: raise RuntimeError("DB is empty - run seed_from_csv or pull_range first.") # Re-pull from one minute before the last seen record to be safe (upsert dedupes). eastern = timezone(timedelta(hours=-5)) start_dt = last # already ISO8601 with offset from the API end_dt = until or datetime.now(eastern).strftime("%Y-%m-%dT%H:%M:%S%z") end_dt = end_dt[:-2] + ":" + end_dt[-2:] if end_dt and end_dt[-3] != ":" else end_dt print(f"incremental: last seen {start_dt}") return pull_range(start_dt, end_dt, db_path) def _get_all(requests, path, limit=100): """Paginate a v2 collection endpoint under the domain.""" url = f"{config.API_BASE}/domains/{config.DOMAIN}/{path}" out, start = [], 0 while True: r = requests.get(url, headers=_headers(), params={"limit": limit, "start": start}, timeout=config.REQUEST_TIMEOUT) r.raise_for_status() data = r.json() items = data if isinstance(data, list) else data.get("items", []) if not items: break out.extend(items) if len(items) < limit: break start += limit return out def pull_departments(out_path=None): """Pull ext -> {department, name, type} for every user, call queue, and auto attendant in the domain; cache to JSON. This is what lets a brand-new user or queue classify into a Reporting Division through its department, without waiting for an xlsx re-export. """ import json import requests out_path = out_path or config.DEPARTMENTS_JSON users = _get_all(requests, "users") queues = _get_all(requests, "callqueues") aas = _get_all(requests, "autoattendants") def ext_of(o, *keys): for k in keys: v = o.get(k) if v is not None and str(v).strip(): return str(v).strip() return "" combined = {} for u in users: ext = ext_of(u, "user") if not ext: continue combined[ext] = { "department": (u.get("department") or "").strip(), "name": f"{(u.get('name-first-name') or '').strip()} " f"{(u.get('name-last-name') or '').strip()}".strip(), "type": "user", } for q in queues: ext = ext_of(q, "callqueue") if not ext or not ext.isdigit(): # skip Reports* pseudo-queues continue # NetSapiens exposes a call queue's department via `subscriber_group` # (the plain `department` field stays blank for ACD queues). sub = (q.get("subscriber_group") or "").strip() combined[ext] = { "department": (q.get("department") or "").strip() or sub, "name": (q.get("description") or "").strip(), "type": "callqueue", } for a in aas: ext = ext_of(a, "user", "aa", "attendant") if not ext: continue combined[ext] = { "department": (a.get("department") or "").strip(), "name": (a.get("attendant-name") or a.get("description") or "").strip(), "type": "autoattendant", } os.makedirs(os.path.dirname(out_path), exist_ok=True) with open(out_path, "w") as f: json.dump(combined, f, indent=2) n_dept = sum(1 for v in combined.values() if v["department"]) print(f"pull_departments: {len(users)} users, {len(queues)} callqueues, " f"{len(aas)} auto attendants -> {len(combined)} extensions " f"({n_dept} with a department) -> {out_path}") return combined if __name__ == "__main__": # CLI: python3 pull.py seed | range | inc | depts if len(sys.argv) < 2: print(__doc__); sys.exit(0) cmd = sys.argv[1] if cmd == "seed": seed_from_csv(sys.argv[2]) elif cmd == "range": pull_range(sys.argv[2], sys.argv[3]) elif cmd == "inc": incremental() elif cmd == "depts": pull_departments() else: print("unknown command:", cmd)