#!/usr/bin/env python3 """ IT Pro Partner — RingLogix Sync Engine Syncs customer VoIP data from RingLogix (NetSapiens/CPaaS) API to local DB. Designed to run as a cron job every 15-30 minutes. Requires: - psycopg2 or sqlite3 (depending on backend) - requests or urllib (stdlib-compatible version below) - RingLogix OAuth credentials (client_id, client_secret, reseller username/password) Usage: python3 sync_ringlogix.py # full sync all domains python3 sync_ringlogix.py --domain acme.ringlogix.com # single domain Environment variables (or .env file): RINGLOGIX_CLIENT_ID RINGLOGIX_CLIENT_SECRET RINGLOGIX_USERNAME # Reseller account username RINGLOGIX_PASSWORD # Reseller account password DATABASE_URL # postgresql://... or sqlite:///path """ from __future__ import annotations import base64 import hashlib import hmac import json import os import ssl import sys import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path from typing import Any # ─── Config ───────────────────────────────────────────────────────── BASE_URL = "https://api.ringlogix.com/pbx/v1/" TOKEN_URL = urllib.parse.urljoin(BASE_URL, "oauth2/token/") CLIENT_ID = os.environ.get("RINGLOGIX_CLIENT_ID", "") CLIENT_SECRET = os.environ.get("RINGLOGIX_CLIENT_SECRET", "") USERNAME = os.environ.get("RINGLOGIX_USERNAME", "") PASSWORD = os.environ.get("RINGLOGIX_PASSWORD", "") DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///portal.db") # ─── HTTP Helpers ─────────────────────────────────────────────────── _ctx = ssl.create_default_context() def _request( method: str, url: str, headers: dict[str, str] | None = None, body: str | bytes | None = None, ) -> tuple[int, bytes]: if isinstance(body, str): body = body.encode("utf-8") req = urllib.request.Request(url, data=body, headers=headers or {}, method=method) try: with urllib.request.urlopen(req, context=_ctx, timeout=30) as resp: return resp.status, resp.read() except urllib.error.HTTPError as e: return e.code, e.read() # ─── OAuth2 ───────────────────────────────────────────────────────── class RingLogixAuth: """Manages OAuth2 token lifecycle for RingLogix API.""" def __init__(self) -> None: self.access_token: str = "" self.refresh_token: str = "" self.expires_at: float = 0 # timestamp def _password_grant(self) -> dict[str, Any]: data = urllib.parse.urlencode({ "grant_type": "password", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "username": USERNAME, "password": PASSWORD, }) status, raw = _request("POST", TOKEN_URL, body=data) if status != 200: raise RuntimeError(f"OAuth password grant failed HTTP {status}: {raw[:300]}") return json.loads(raw) def _refresh_grant(self) -> dict[str, Any]: data = urllib.parse.urlencode({ "grant_type": "refresh_token", "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET, "refresh_token": self.refresh_token, }) status, raw = _request("POST", TOKEN_URL, body=data) if status != 200: # Fall back to password grant return self._password_grant() return json.loads(raw) def ensure_token(self) -> None: if self.access_token and time.time() < self.expires_at - 60: return # still valid if self.refresh_token: resp = self._refresh_grant() else: resp = self._password_grant() self.access_token = resp["access_token"] self.refresh_token = resp.get("refresh_token", "") self.expires_at = time.time() + resp.get("expires_in", 3600) def api_call( self, object_name: str, action: str, params: dict[str, str] | None = None, ) -> dict[str, Any]: """Make a POST to the RingLogix API with OAuth bearer token.""" self.ensure_token() payload = {"object": object_name, "action": action} if params: payload.update(params) headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/x-www-form-urlencoded; charset=utf-8", } body = urllib.parse.urlencode(payload) status, raw = _request("POST", BASE_URL, headers=headers, body=body) if status == 401: # Token expired mid-call, force refresh and retry once self.access_token = "" self.ensure_token() headers["Authorization"] = f"Bearer {self.access_token}" status, raw = _request("POST", BASE_URL, headers=headers, body=body) if status not in (200, 201): raise RuntimeError( f"API call {object_name}/{action} failed HTTP {status}: {raw[:500]}" ) result = json.loads(raw) if isinstance(result, list): return {"data": result, "total": len(result)} return result def api_get(self, object_name: str, action: str, params: dict[str, str] | None = None) -> list[dict[str, Any]]: """GET-style read returning a list of results.""" resp = self.api_call(object_name, action, params) # NetSapiens API returns data in various shapes if isinstance(resp, dict): data = resp.get("data", resp.get("response", [])) if isinstance(data, dict): return [data] if isinstance(data, list): return data return [resp] if isinstance(resp, dict) else [] # ─── DB Layer ─────────────────────────────────────────────────────── class Database: """Minimal SQLite DB for MVP. Swap for psycopg2/asyncpg in production.""" def __init__(self, dsn: str) -> None: import sqlite3 self.conn = sqlite3.connect(dsn.replace("sqlite:///", "")) self.conn.row_factory = sqlite3.Row self.conn.execute("PRAGMA journal_mode=WAL") self.conn.execute("PRAGMA foreign_keys=ON") self._ensure_schema() def _ensure_schema(self) -> None: self.conn.executescript(""" CREATE TABLE IF NOT EXISTS customers ( id TEXT PRIMARY KEY, company_name TEXT NOT NULL, contact_email TEXT NOT NULL, contact_phone TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS service_subscriptions ( id TEXT PRIMARY KEY, customer_id TEXT REFERENCES customers(id), service_type TEXT NOT NULL CHECK (service_type IN ('voip','web_hosting','fleet_tracking')), status TEXT NOT NULL DEFAULT 'active', plan_name TEXT, monthly_price REAL, next_billing_date TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS voip_subscriptions ( id TEXT PRIMARY KEY, subscription_id TEXT REFERENCES service_subscriptions(id), ringlogix_domain TEXT NOT NULL UNIQUE, max_extensions INTEGER DEFAULT 5, max_auto_attendants INTEGER DEFAULT 1, max_call_queues INTEGER DEFAULT 1, last_synced_at TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS voip_extensions ( id TEXT PRIMARY KEY, voip_subscription_id TEXT REFERENCES voip_subscriptions(id), extension TEXT NOT NULL, user_login TEXT, full_name TEXT, email TEXT, department TEXT, device_mac TEXT, device_model TEXT, device_status TEXT DEFAULT 'offline', features TEXT DEFAULT '{}', cf_always TEXT, cf_busy TEXT, cf_no_answer TEXT, vm_enabled INTEGER DEFAULT 1, vm_email_notification INTEGER DEFAULT 0, vm_transcription INTEGER DEFAULT 0, recording_enabled INTEGER DEFAULT 0, presence TEXT DEFAULT 'offline', updated_at TEXT DEFAULT (datetime('now')), UNIQUE(voip_subscription_id, extension) ); CREATE TABLE IF NOT EXISTS voip_dids ( id TEXT PRIMARY KEY, voip_subscription_id TEXT REFERENCES voip_subscriptions(id), phone_number TEXT NOT NULL UNIQUE, label TEXT, routing TEXT, routing_target TEXT, caller_id_name TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS voip_auto_attendants ( id TEXT PRIMARY KEY, voip_subscription_id TEXT REFERENCES voip_subscriptions(id), name TEXT NOT NULL, greeting_type TEXT, greeting_text TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS voip_call_queues ( id TEXT PRIMARY KEY, voip_subscription_id TEXT REFERENCES voip_subscriptions(id), name TEXT NOT NULL, routing_strategy TEXT, max_wait_time_seconds INTEGER DEFAULT 300, agents TEXT DEFAULT '[]', created_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS voip_feature_usage ( id TEXT PRIMARY KEY, voip_subscription_id TEXT REFERENCES voip_subscriptions(id), feature_key TEXT NOT NULL, enabled INTEGER DEFAULT 0, current_qty INTEGER DEFAULT 0, max_qty INTEGER, last_verified_at TEXT DEFAULT (datetime('now')), UNIQUE(voip_subscription_id, feature_key) ); CREATE TABLE IF NOT EXISTS voip_plan_features ( id TEXT PRIMARY KEY, plan_name TEXT NOT NULL, feature_key TEXT NOT NULL, feature_label TEXT NOT NULL, feature_category TEXT, included INTEGER DEFAULT 1, max_qty INTEGER, sort_order INTEGER DEFAULT 0, UNIQUE(plan_name, feature_key) ); """) self.conn.commit() def upsert(self, table: str, data: dict[str, Any], unique_cols: list[str]) -> None: """Upsert by deleting existing matching row(s) then inserting.""" if unique_cols: where = " AND ".join(f"{c}=?" for c in unique_cols) vals = tuple(data.get(c) for c in unique_cols) self.conn.execute(f"DELETE FROM {table} WHERE {where}", vals) cols = ", ".join(data.keys()) placeholders = ", ".join("?" for _ in data) self.conn.execute( f"INSERT INTO {table} ({cols}) VALUES ({placeholders})", tuple(data.values()), ) self.conn.commit() def get_customer_by_domain(self, domain: str) -> dict | None: row = self.conn.execute( "SELECT c.* FROM customers c " "JOIN service_subscriptions s ON s.customer_id = c.id " "JOIN voip_subscriptions v ON v.subscription_id = s.id " "WHERE v.ringlogix_domain = ?", (domain,), ).fetchone() return dict(row) if row else None def upsert_customer_for_domain(self, domain: str, company: str, email: str) -> str: """Create/find customer + subscription + voip_subscription for a domain.""" existing = self.get_customer_by_domain(domain) import uuid if existing: # Update fields self.conn.execute( "UPDATE customers SET company_name=?, contact_email=? WHERE id=?", (company, email, existing["id"]), ) cust_id = existing["id"] else: cust_id = str(uuid.uuid4()) self.conn.execute( "INSERT INTO customers (id, company_name, contact_email) VALUES (?, ?, ?)", (cust_id, company, email), ) # Create subscription sub_id = str(uuid.uuid4()) self.conn.execute( "INSERT INTO service_subscriptions (id, customer_id, service_type, plan_name) " "VALUES (?, ?, 'voip', 'Business Pro')", (sub_id, cust_id), ) # Create voip_subscription vs_id = str(uuid.uuid4()) self.conn.execute( "INSERT INTO voip_subscriptions (id, subscription_id, ringlogix_domain) " "VALUES (?, ?, ?)", (vs_id, sub_id, domain), ) self.conn.commit() return cust_id def get_voip_subscription_id(self, domain: str) -> str | None: row = self.conn.execute( "SELECT id FROM voip_subscriptions WHERE ringlogix_domain = ?", (domain,), ).fetchone() return row["id"] if row else None def close(self) -> None: self.conn.close() # ─── Sync Functions ───────────────────────────────────────────────── def _uid() -> str: import uuid return str(uuid.uuid4()) def sync_domain(auth: RingLogixAuth, db: Database, domain: str) -> dict[str, Any]: """Full sync of a single RingLogix domain.""" now = datetime.now(timezone.utc).isoformat() summary = {"domain": domain, "extensions": 0, "dids": 0, "auto_attendants": 0, "call_queues": 0} vs_id = db.get_voip_subscription_id(domain) if not vs_id: vs_id = str(_uid()) sub_id = str(_uid()) cust_id = str(_uid()) db.conn.execute( "INSERT INTO customers (id, company_name, contact_email) VALUES (?, ?, ?)", (cust_id, domain.split(".")[0].title() + " LLC", f"billing@{domain}"), ) db.conn.execute( "INSERT INTO service_subscriptions (id, customer_id, service_type) VALUES (?, ?, 'voip')", (sub_id, cust_id), ) db.conn.execute( "INSERT INTO voip_subscriptions (id, subscription_id, ringlogix_domain) VALUES (?, ?, ?)", (vs_id, sub_id, domain), ) db.conn.commit() # 1. Count subscribers try: count_resp = auth.api_call("subscriber", "count", {"domain": domain}) total_subs = int(count_resp.get("total", 0)) except Exception: total_subs = 0 # 2. Read all subscribers try: subs = auth.api_get("subscriber", "read", { "domain": domain, "scope": "all", "fields": "user,first_name,last_name,email,presence,account_status,time_zone,vmail_provisioned,voicemail_emergency,recording,cf_always,cf_busy,cf_no_answer", }) except Exception as e: print(f" ⚠️ Subscriber read failed: {e}") subs = [] for sub in subs: ext = sub.get("user", "") if not ext: continue features = { "voicemail": sub.get("vmail_provisioned", "no") == "yes", "recording": sub.get("recording", "no") == "yes", "call_forwarding": bool(sub.get("cf_always") or sub.get("cf_busy") or sub.get("cf_no_answer")), } db.upsert("voip_extensions", { "id": str(_uid()), "voip_subscription_id": vs_id, "extension": ext, "user_login": sub.get("login", ""), "full_name": f"{sub.get('first_name','')} {sub.get('last_name','')}".strip(), "email": sub.get("email", ""), "department": sub.get("group", ""), "presence": sub.get("presence", "offline"), "features": json.dumps(features), "cf_always": sub.get("cf_always", ""), "cf_busy": sub.get("cf_busy", ""), "cf_no_answer": sub.get("cf_no_answer", ""), "vm_enabled": 1 if features["voicemail"] else 0, "recording_enabled": 1 if features["recording"] else 0, "updated_at": now, }, ["voip_subscription_id", "extension"]) summary["extensions"] += 1 # 3. Read devices try: devices = auth.api_get("device", "read", {"domain": domain}) # We can match devices to extensions by user field for dev in devices: ext_user = dev.get("user", "") if ext_user: dev_mac = dev.get("mac", "") dev_model = dev.get("model", "") dev_status = "registered" if dev.get("registration_time") else "offline" # Update the matching extension's device info rows = db.conn.execute( "SELECT id FROM voip_extensions WHERE voip_subscription_id=? AND extension=?", (vs_id, ext_user), ).fetchall() for row in rows: db.conn.execute( "UPDATE voip_extensions SET device_mac=?, device_model=?, device_status=?, updated_at=? WHERE id=?", (dev_mac, dev_model, dev_status, now, row["id"]), ) db.conn.commit() except Exception as e: print(f" ⚠️ Device read failed: {e}") # 4. Read DIDs / phone numbers try: dids = auth.api_get("phonenumber", "read", {"domain": domain}) for did in dids: number = did.get("phonenumber", did.get("number", "")) if not number: continue db.upsert("voip_dids", { "id": str(_uid()), "voip_subscription_id": vs_id, "phone_number": number, "label": did.get("description", did.get("label", "")), "routing": did.get("route_type", ""), "routing_target": did.get("route_dest", ""), "caller_id_name": did.get("caller_id_name_emerge", ""), }, ["phone_number"]) summary["dids"] += 1 except Exception as e: print(f" ⚠️ DID read failed: {e}") # 5. Read auto attendants try: attendants = auth.api_get("autoattendant", "read", {"domain": domain}) for aa in attendants: name = aa.get("name", aa.get("user", "Main")) db.upsert("voip_auto_attendants", { "id": str(_uid()), "voip_subscription_id": vs_id, "name": name, "greeting_type": aa.get("greeting_type", ""), "greeting_text": aa.get("greeting_text", ""), }, ["voip_subscription_id", "name"]) summary["auto_attendants"] += 1 except Exception as e: print(f" ⚠️ Auto attendant read failed: {e}") # 6. Read call queues try: queues = auth.api_get("callqueue", "read", {"domain": domain}) for q in queues: name = q.get("name", q.get("queue_name", "Queue")) db.upsert("voip_call_queues", { "id": str(_uid()), "voip_subscription_id": vs_id, "name": name, "routing_strategy": q.get("routing_strategy", q.get("queue_type", "")), "max_wait_time_seconds": int(q.get("max_wait_time", q.get("queue_timeout", 300))), "agents": json.dumps(q.get("agents", [])), }, ["voip_subscription_id", "name"]) summary["call_queues"] += 1 except Exception as e: print(f" ⚠️ Call queue read failed: {e}") # Update last_synced_at db.conn.execute( "UPDATE voip_subscriptions SET last_synced_at=? WHERE id=?", (now, vs_id), ) db.conn.commit() return summary def sync_all(auth: RingLogixAuth, db: Database, target_domain: str | None = None) -> list[dict[str, Any]]: """Sync all domains under the reseller account, or a single domain.""" if target_domain: print(f"Syncing single domain: {target_domain}") return [sync_domain(auth, db, target_domain)] # List all domains try: domains = auth.api_get("domain", "read", {}) except Exception as e: print(f"Failed to list domains: {e}") return [] results = [] for d in domains: domain_name = d.get("domain", "") if domain_name: print(f"Syncing {domain_name}...") try: summary = sync_domain(auth, db, domain_name) results.append(summary) print(f" ✅ {summary['extensions']} ext, {summary['dids']} DIDs, " f"{summary['auto_attendants']} AA, {summary['call_queues']} queues") except Exception as e: print(f" ❌ {e}") return results # ─── Feature Definitions (seed data) ──────────────────────────────── DEFAULT_FEATURES = [ # (plan, key, label, category, included, max_qty, sort) ("Business Basic", "call_forwarding", "Call Forwarding (Always/Busy/No Answer)", "Call Management", True, None, 1), ("Business Basic", "call_transfer", "Call Transfer (Attended/Blind/Intercom)", "Call Management", True, None, 2), ("Business Basic", "three_way_calling", "Three-Way Calling", "Call Management", True, None, 3), ("Business Basic", "voicemail", "Voicemail-to-Email", "Voicemail", True, None, 4), ("Business Basic", "dnd", "Do Not Disturb", "Call Management", True, None, 5), ("Business Basic", "auto_attendant", "Auto Attendant", "Auto Attendant", True, 1, 6), ("Business Basic", "call_recording", "Call Recording", "Call Recording", False, None, 7), ("Business Basic", "call_queue", "Call Queue", "Call Queues", False, None, 8), ("Business Basic", "vm_transcription", "Voicemail Transcription", "Voicemail", False, None, 9), ("Business Pro", "call_forwarding", "Call Forwarding (Always/Busy/No Answer/Find Me Follow Me)", "Call Management", True, None, 1), ("Business Pro", "call_transfer", "Call Transfer (Attended/Blind/Intercom)", "Call Management", True, None, 2), ("Business Pro", "three_way_calling", "Three-Way Calling", "Call Management", True, None, 3), ("Business Pro", "voicemail", "Voicemail-to-Email", "Voicemail", True, None, 4), ("Business Pro", "vm_transcription", "Voicemail Transcription", "Voicemail", True, None, 5), ("Business Pro", "dnd", "Do Not Disturb", "Call Management", True, None, 6), ("Business Pro", "auto_attendant", "Auto Attendant", "Auto Attendant", True, 2, 7), ("Business Pro", "call_queue", "Call Queues", "Call Queues", True, 3, 8), ("Business Pro", "call_recording", "Call Recording", "Call Recording", True, None, 9), ("Business Pro", "call_parking", "Call Parking (up to 5)", "Call Management", True, None, 10), ("Business Pro", "crm_integration", "CRM Integration", "Integrations", False, None, 11), ("Enterprise", "call_forwarding", "Call Forwarding (All options)", "Call Management", True, None, 1), ("Enterprise", "call_transfer", "Call Transfer (All options)", "Call Management", True, None, 2), ("Enterprise", "three_way_calling", "Three-Way Calling", "Call Management", True, None, 3), ("Enterprise", "voicemail", "Voicemail-to-Email", "Voicemail", True, None, 4), ("Enterprise", "vm_transcription", "Voicemail Transcription", "Voicemail", True, None, 5), ("Enterprise", "dnd", "Do Not Disturb", "Call Management", True, None, 6), ("Enterprise", "auto_attendant", "Auto Attendant", "Auto Attendant", True, 5, 7), ("Enterprise", "call_queue", "Call Queues", "Call Queues", True, 10, 8), ("Enterprise", "call_recording", "Call Recording", "Call Recording", True, None, 9), ("Enterprise", "call_parking", "Call Parking (unlimited)", "Call Management", True, None, 10), ("Enterprise", "crm_integration", "CRM Integration", "Integrations", True, None, 11), ("Enterprise", "skill_based_routing", "Skill-Based Routing", "Call Queues", True, None, 12), ("Enterprise", "hot_desking", "Hot Desking / Hoteling", "Advanced", True, None, 13), ] def seed_features(db: Database) -> None: """Insert default feature definitions if empty.""" count = db.conn.execute("SELECT COUNT(*) as c FROM voip_plan_features").fetchone()["c"] if count > 0: return for plan, key, label, cat, included, max_qty, sort in DEFAULT_FEATURES: import uuid db.conn.execute( "INSERT INTO voip_plan_features (id, plan_name, feature_key, feature_label, " "feature_category, included, max_qty, sort_order) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (str(uuid.uuid4()), plan, key, label, cat, 1 if included else 0, max_qty, sort), ) db.conn.commit() print(f" ✅ Seeded {len(DEFAULT_FEATURES)} plan features") # ─── Main ──────────────────────────────────────────────────────────── def main() -> None: import argparse parser = argparse.ArgumentParser(description="RingLogix Sync Engine") parser.add_argument("--domain", help="Sync only this domain") parser.add_argument("--db", help="Database URL (overrides DATABASE_URL env)") parser.add_argument("--seed", action="store_true", help="Seed plan features only, no sync") args = parser.parse_args() db_url = args.db or DATABASE_URL if not db_url: print("❌ DATABASE_URL not set. Use --db or set env.") sys.exit(1) db = Database(db_url) print(f"📦 Database: {db_url}") if args.seed: seed_features(db) db.close() return # Validate RingLogix credentials missing = [] for var, name in [ (CLIENT_ID, "RINGLOGIX_CLIENT_ID"), (CLIENT_SECRET, "RINGLOGIX_CLIENT_SECRET"), (USERNAME, "RINGLOGIX_USERNAME"), (PASSWORD, "RINGLOGIX_PASSWORD"), ]: if not var: missing.append(name) if missing: print(f"❌ Missing env vars: {', '.join(missing)}") print(" Set them in .env or export them.") db.close() sys.exit(1) auth = RingLogixAuth() # Seed features if empty seed_features(db) # Sync results = sync_all(auth, db, args.domain) print(f"\n{'='*50}") print(f"Sync complete: {len(results)} domains processed") for r in results: print(f" {r['domain']}: {r['extensions']} ext, {r['dids']} DIDs, " f"{r['auto_attendants']} AA, {r['call_queues']} queues") db.close() if __name__ == "__main__": main()