asterisk-voicemail/src/vm_contacts.py

#!/usr/bin/env python3
"""
Caller-ID -> contact name resolution for Asterisk voicemail notifications.

Backends (contacts are resolved exclusively from the MySQL `contacts` table
on the asterisk DB):
  mysql   - MySQL contacts table on the asterisk DB (primary, fast, offline).

Everything here is best-effort: a lookup failure returns None and the caller
falls back to the raw caller-ID string.  Results (hits AND misses) are cached
to avoid per-call API traffic.

NOTE: app-password / basic auth does NOT work against Google - Google disabled
it for CardDAV/CalDAV/IMAP on 2024-09-30 and now requires OAuth for CardDAV.
The carddav backend is for Nextcloud / Fastmail / iCloud / Radicale etc.
"""

import base64
import configparser
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

CONF_PATH = os.environ.get("VM_CONTACTS_CONF", "/opt/vm-transcribe/contacts.conf")


# ------------------------------------------------------------------- helpers
def digits_of(s):
    return re.sub(r"\D", "", s or "")


def normalize_uk(d):
    """Canonicalise a UK number so +44 and 0 forms collide.

    Asterisk stores caller IDs as '+441273003930' while a Google export holds
    '01273003930' (and vice-versa). Both are the same number; strip the leading
    '44' (from +44) and prepend '0', and turn a '00' intl prefix into '00'.
    Non-UK numbers are returned unchanged.
    """
    if not d:
        return d
    if d.startswith("44") and len(d) == 12:      # +441273003930
        return "0" + d[2:]
    if d.startswith("0044") and len(d) == 14:    # 00441273003930
        return "0" + d[4:]
    return d


def extract_number(caller):
    """Pull a dialable number out of a CallerID string like 'Dave <079...>'."""
    if not caller:
        return ""
    m = re.search(r"<([^>]+)>", caller)
    cand = m.group(1) if m else caller
    return normalize_uk(digits_of(cand))


def _key(num_digits, n):
    d = digits_of(num_digits)
    return d[-n:] if len(d) >= n else d


_EMOJI_RE = re.compile(
    "[\U0001F000-\U0001FAFF\U00002600-\U000027BF\U0001F1E6-\U0001F1FF]")
_PUNCT_RE = re.compile(r"[\W_]+", re.UNICODE)


def _is_junk_name(name):
    """True for empty, emoji-only, or punctuation-only names that should not
    shadow a real contact name when numbers collide."""
    if not name:
        return True
    s = name.strip()
    if not s:
        return True
    stripped = _EMOJI_RE.sub("", s)
    return not _PUNCT_RE.sub("", stripped)


class Cache:
    def __init__(self, path, ttl):
        self.path, self.ttl, self.data = path, ttl, {}
        try:
            with open(path) as fh:
                self.data = json.load(fh)
        except Exception:
            self.data = {}

    def get(self, key):
        e = self.data.get(key)
        if not e:
            return None  # unknown -> caller should look up
        if self.ttl and time.time() - e.get("t", 0) > self.ttl:
            return None
        return e  # {"t":..., "name": <str or None>}

    def put(self, key, result):
        # result is the (name, email) tuple from resolve().
        self.data[key] = {"t": time.time(), "name": result[0], "email": result[1]}
        try:
            os.makedirs(os.path.dirname(self.path), exist_ok=True)
            tmp = self.path + ".tmp"
            with open(tmp, "w") as fh:
                json.dump(self.data, fh)
            os.replace(tmp, self.path)
        except Exception:
            pass


# ------------------------------------------------------------- file backend
def _parse_vcf(text):
    """Return list of (name, [numbers])."""
    out = []
    name, nums, emails = None, [], []
    for raw in text.splitlines():
        line = raw.strip()
        u = line.upper()
        if u == "BEGIN:VCARD":
            name, nums, emails = None, [], []
        elif u.startswith("FN"):
            name = line.split(":", 1)[1].strip() if ":" in line else None
        elif u.startswith("TEL"):
            if ":" in line:
                nums.append(line.split(":", 1)[1].strip())
        elif u.startswith("EMAIL"):
            if ":" in line:
                # Prefer a WORK-typed address when one exists; callers usually
                # want the business reply address, not a personal/home one.
                is_work = "WORK" in line.upper()
                addr = line.split(":", 1)[1].strip()
                if is_work:
                    emails.insert(0, addr)   # work goes first
                else:
                    emails.append(addr)
        elif u == "END:VCARD":
            if name and nums:
                out.append((name, nums, emails))
    return out


def _parse_csv(path):
    import csv
    out = []
    with open(path, newline="", encoding="utf-8", errors="replace") as fh:
        r = csv.DictReader(fh)
        cols = r.fieldnames or []
        name_cols = [c for c in cols if c and ("Name" == c or c.endswith("Name"))]
        phone_cols = [c for c in cols if c and "Phone" in c and "Value" in c] or \
                     [c for c in cols if c and "Phone" in c]
        email_cols = [c for c in cols if c and "E-mail" in c] or \
                     [c for c in cols if c and "Email" in c]
        for row in r:
            name = ""
            if "Name" in row and row["Name"]:
                name = row["Name"]
            else:
                parts = [row.get(c, "") for c in ("Given Name", "Family Name") if row.get(c)]
                name = " ".join(parts) or (row.get(name_cols[0], "") if name_cols else "")
            nums = []
            for c in phone_cols:
                v = row.get(c, "")
                if v:
                    nums.extend(re.split(r"\s*:::\s*|\s*;", v))
            emails = []
            for c in email_cols:
                v = row.get(c, "")
                if v:
                    emails.extend(re.split(r"\s*:::\s*|\s*;", v))
            if name and nums:
                out.append((name.strip(), nums, emails))
    return out


def _index(entries, n):
    """Index contact numbers -> (name, email) by full digit string.

    See resolve()/lookup_file() for the collision rules. Returns a dict
    keyed by the last n digits of each number, value (name, email_or_None).
    """
    idx = {}
    for name, nums, emails in entries:
        if not name or not nums:
            continue
        if _is_junk_name(name):
            continue  # don't let an emoji/placeholder card win a slot
        d = digits_of(nums[0])
        if not d:
            continue
        k = d[-n:] if len(d) >= n else d
        if k not in idx:
            idx[k] = (name, (emails[0] if emails else None))
    return idx


def lookup_file(cfg, num_digits, n, log):
    path = cfg.get("path", "").strip()
    if not path or not os.path.exists(path):
        return None
    try:
        if path.lower().endswith(".csv"):
            entries = _parse_csv(path)
        else:
            with open(path, encoding="utf-8", errors="replace") as fh:
                entries = _parse_vcf(fh.read())
        index = _index(entries, n)
        d = digits_of(num_digits)
        if not d:
            return None
        for length in range(min(len(d), n), max(0, n - 6), -1):
            hit = index.get(d[-length:])
            if hit:
                return hit  # (name, email)
        return None
    except Exception as e:
        log("contacts file backend error: %s" % e)
        return None


# ----------------------------------------------------------- google backend
def _google_access_token(token_path, log):
    try:
        with open(token_path) as fh:
            tok = json.load(fh)
    except Exception as e:
        log("google token unreadable (%s): %s" % (token_path, e))
        return None
    # try existing token first; refresh if People API 401s
    at = tok.get("token") or tok.get("access_token")
    refresh = tok.get("refresh_token")
    cid = tok.get("client_id")
    secret = tok.get("client_secret")
    if at:
        return at, (refresh, cid, secret, token_path, tok)
    return _google_refresh((refresh, cid, secret, token_path, tok), log)


def _google_refresh(ctx, log):
    refresh, cid, secret, token_path, tok = ctx
    if not (refresh and cid and secret):
        log("google token missing refresh_token/client_id/client_secret")
        return None
    try:
        data = urllib.parse.urlencode({
            "client_id": cid, "client_secret": secret,
            "refresh_token": refresh, "grant_type": "refresh_token",
        }).encode()
        req = urllib.request.Request("https://oauth2.googleapis.com/token", data=data)
        with urllib.request.urlopen(req, timeout=20) as r:
            new = json.loads(r.read().decode())
        at = new.get("access_token")
        if at:
            tok["token"] = at
            try:
                with open(token_path, "w") as fh:
                    json.dump(tok, fh)
            except Exception:
                pass
            return at, ctx
    except Exception as e:
        log("google token refresh failed: %s" % e)
    return None


def lookup_google(cfg, num_digits, n, log):
    token_path = cfg.get("token_path", "").strip()
    if not token_path or not os.path.exists(token_path):
        return None
    got = _google_access_token(token_path, log)
    if not got:
        return None
    at, ctx = got
    want = _key(num_digits, n)

    def query(access):
        url = ("https://people.googleapis.com/v1/people:searchContacts"
               "?query=%s&readMask=names,phoneNumbers"
               % urllib.parse.quote(num_digits[-7:] or num_digits))
        req = urllib.request.Request(url, headers={"Authorization": "Bearer %s" % access})
        with urllib.request.urlopen(req, timeout=20) as r:
            return json.loads(r.read().decode())

    try:
        try:
            d = query(at)
        except urllib.error.HTTPError as he:
            if he.code == 401:  # refresh once
                got = _google_refresh(ctx if isinstance(ctx, tuple) and len(ctx) == 5
                                      else ctx, log)
                if not got:
                    return None
                d = query(got[0])
            else:
                raise
        for res in d.get("results", []):
            person = res.get("person", {})
            for ph in person.get("phoneNumbers", []):
                if _key(ph.get("value", ""), n) == want:
                    names = person.get("names", [])
                    if names:
                        emails = [e.get("value") for e in person.get("emailAddresses", [])
                                  if e.get("value")]
                        return (names[0].get("displayName"),
                                (emails[0] if emails else None))
    except Exception as e:
        log("google People API error: %s" % e)
    return None


# ----------------------------------------------------------- mysql backend
def lookup_mysql(cfg, num_digits, n, log):
    """Look up a contact in the asterisk MySQL DB contacts table.

    Uses vm_store.connect() so it shares the same asterisk DB config and
    PyMySQL dependency already required by the portal.  If the table does
    not not exist yet, run vm_import_contacts.py to create and populate it.
    """
    try:
        sys_path = os.path.dirname(os.path.abspath(__file__))
        if sys_path not in sys.path:
            sys.path.insert(0, sys_path)
        import vm_store  # noqa: E402  (local import to avoid hard dep at import time)
    except Exception as e:
        log("contacts mysql backend: vm_store unavailable (%s)" % e)
        return None

    num = extract_number(num_digits)
    if not num:
        return None
    try:
        con = vm_store.connect()
        row = con.execute(
            "SELECT name, email FROM contacts "
            "WHERE number_e164 = ? AND name IS NOT NULL AND name != '' "
            "ORDER BY id LIMIT 1",
            (num,)).fetchone()
        con.close()
        if row:
            return (row["name"], row.get("email") or None)
    except Exception as e:
        log("contacts mysql backend error: %s" % e)
    return None


# ---------------------------------------------------------- carddav backend
def lookup_carddav(cfg, num_digits, n, log):
    import ssl
    url = cfg.get("url", "").strip()
    user = cfg.get("username", "").strip()
    pw = cfg.get("app_password", "").strip()
    if not (url and user and pw):
        return None
    if "google.com" in url:
        log("carddav backend points at Google, which rejects app passwords "
            "since 2024-09-30; use the 'google' or 'file' backend instead")
        return None

    path = cfg.get("addressbook_path", "").strip()
    base = url.rstrip("/") + (path if path.startswith("/") else "/" + path if path else "")
    ctx = None
    if cfg.get("verify_tls", "yes").lower() in ("no", "false", "0"):
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
    timeout = int(cfg.get("timeout", "20") or 20)
    auth = base64.b64encode(("%s:%s" % (user, pw)).encode()).decode()

    body = ('<?xml version="1.0"?>'
            '<C:addressbook-query xmlns:D="DAV:" '
            'xmlns:C="urn:ietf:params:xml:ns:carddav">'
            '<D:prop><C:address-data/></D:prop></C:addressbook-query>')
    try:
        req = urllib.request.Request(
            base, data=body.encode(), method="REPORT",
            headers={"Authorization": "Basic %s" % auth, "Depth": "1",
                     "Content-Type": "application/xml; charset=utf-8"})
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
            xml = r.read().decode("utf-8", "replace")
        vcards = re.findall(r"BEGIN:VCARD.*?END:VCARD", xml, re.S | re.I)
        entries = []
        for v in vcards:
            entries.extend(_parse_vcf(v.replace("&#13;", "").replace("\r", "")))
        return _index(entries, n).get(_key(num_digits, n))  # (name, email)
    except Exception as e:
        log("carddav backend error: %s" % e)
        return None


# ---------------------------------------------------------------- entrypoint
_BACKENDS = {"mysql": lookup_mysql}


def resolve(caller_id, log=print):
    """Return (name, email) for a CallerID string, or (None, None).

    `name` is the resolved contact name (or None); `email` is the contact's
    email when the matched number also has one in the address book (or None).
    """
    num = extract_number(caller_id)
    if not num or not os.path.exists(CONF_PATH):
        return (None, None)
    try:
        cp = configparser.ConfigParser(inline_comment_prefixes=("#", ";"))
        cp.read(CONF_PATH)
        if not cp.has_section("contacts") or not cp["contacts"].getboolean("enabled", False):
            return (None, None)
        g = cp["contacts"]
        n = int(g.get("match_digits", "15") or 15)
        cache = Cache(g.get("cache_path", "/var/lib/vm-transcribe/contacts_cache.json"),
                      int(g.get("cache_ttl", "86400") or 86400))

        key = _key(num, n)
        cached = cache.get(key)
        if cached is not None:
            # The cache stores {"t":..., "name": name_str, "email": email_str}.
            if isinstance(cached, dict):
                return (cached.get("name") or None, cached.get("email") or None)
            # legacy: stored the name directly (no dict wrapper)
            return (cached, None)

        order = [b.strip() for b in g.get("backends", "file").split(",") if b.strip()]
        result = (None, None)
        for b in order:
            fn = _BACKENDS.get(b)
            if not fn:
                log("unknown contacts backend: %s" % b)
                continue
            sect = cp[b] if cp.has_section(b) else {}
            try:
                hit = fn(sect, num, n, log)
            except Exception as e:
                log("contacts backend %s crashed: %s" % (b, e))
                hit = None
            if isinstance(hit, tuple):
                name, email = hit
            else:
                name, email = hit, None  # legacy backends returned name only
            if name:
                result = (name, email)
                log("contacts: %s -> %s (via %s)%s"
                    % (num, name, b, (" email=%s" % email if email else "")))
                break

        cache.put(key, result)  # cache the (name, email) tuple
        return result
    except Exception as e:
        log("contacts resolve error: %s" % e)
        return (None, None)


if __name__ == "__main__":
    import sys
    print(resolve(sys.argv[1] if len(sys.argv) > 1 else "", log=lambda m: print("[log]", m)))