wacli.py

#!/usr/bin/env python3
"""
wacli — WhatsApp Business Cloud API from the command line.

Direct Meta Cloud API — no BSP middleman. Create templates, send messages,
and run ops from the terminal or Claude Code. Designed as a thin, fast,
deterministic CLI that pairs with LLM-driven slash commands for natural-
language workflows.

Setup (see .env.example):
    META_WHATSAPP_TOKEN=<system-user token with whatsapp_business_management +
                         whatsapp_business_messaging>
    META_APP_ID=<your Meta app id — needed only for IMAGE header uploads>
    WHATSAPP_WABA_ID=<your WhatsApp Business Account id>
    WHATSAPP_PHONE_ID=<your phone number id on the WABA>

Common examples:
    wacli templates                          # list approved/pending/rejected
    wacli matrix                             # grid by language + status
    wacli template hello_v1                  # template detail
    wacli send 14155551234 hello_v1          # send to a phone number
    wacli send alice hello_v1                # send by recipients.json alias
    wacli test hello_v1 --to alice bob       # send to many
    wacli send 14155551234 --text "hi"       # free-form (session window)
    wacli create hello_v2 --body "Hi {{1}}" --example Alice --wait 30
    wacli create-image promo_v1 --body "..." --image promo.png --wait 30
    wacli delete hello_v1
    wacli insights hello_v1 --lifetime
    wacli phone
    wacli token-check
"""

import argparse
import json
import os
import sys
import time
from collections import Counter
from pathlib import Path

import requests

GRAPH = "https://graph.facebook.com/v20.0"
ROOT = Path(__file__).resolve().parent


def _load_env():
    """Load a .env file from the current working directory or the script dir."""
    for candidate in (Path.cwd() / ".env", ROOT / ".env"):
        if not candidate.exists():
            continue
        for line in candidate.read_text().splitlines():
            line = line.strip()
            if "=" in line and not line.startswith("#"):
                k, v = line.split("=", 1)
                os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
        break


_load_env()

TOKEN = os.environ.get("META_WHATSAPP_TOKEN", "")
WABA_ID = os.environ.get("WHATSAPP_WABA_ID", "")
PHONE_ID = os.environ.get("WHATSAPP_PHONE_ID", "")
APP_ID = os.environ.get("META_APP_ID", "")
RECIPIENTS_FILE = os.environ.get("WACLI_RECIPIENTS", "recipients.json")


# ---------------------------------------------------------------------------
# recipients.json resolver
# ---------------------------------------------------------------------------

def _load_recipients():
    """Read the recipients JSON from CWD or script dir. Accepts a list of
    objects with at minimum a 'whatsapp' field and one of: 'alias', 'name'."""
    for candidate in (Path.cwd() / RECIPIENTS_FILE, ROOT / RECIPIENTS_FILE):
        if candidate.exists():
            try:
                return json.loads(candidate.read_text())
            except Exception:
                return []
    return []


def resolve_recipient(value: str) -> tuple[str, str]:
    """Resolve a recipient to (wa_id, display_label).

    Accepts:
      - raw phone digits (with or without country code)
      - recipient alias (case-insensitive)
      - name substring (case-insensitive)
    """
    v = value.strip()
    digits = "".join(c for c in v if c.isdigit())
    if digits and digits == v.lstrip("+"):
        return digits, f"+{digits}"

    people = _load_recipients()
    vlow = v.lower()
    for m in people:
        if (m.get("alias", "").lower() == vlow or
                m.get("name", "").lower() == vlow or
                (m.get("name") and vlow in m["name"].lower())):
            phone = m.get("whatsapp")
            if not phone:
                print(f"ERROR: '{m.get('name', v)}' has no whatsapp number in {RECIPIENTS_FILE}")
                sys.exit(1)
            label = m.get("name") or m.get("alias") or phone
            return phone, f"{label} ({phone})"
    print(f"ERROR: Could not resolve '{value}' as a phone or recipient.")
    print("Try: raw digits (14155551234), alias, or a name substring.")
    sys.exit(1)


# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------

def _check_token():
    missing = []
    if not TOKEN:
        missing.append("META_WHATSAPP_TOKEN")
    if not WABA_ID:
        missing.append("WHATSAPP_WABA_ID")
    if not PHONE_ID:
        missing.append("WHATSAPP_PHONE_ID")
    if missing:
        print(f"ERROR: missing env var(s): {', '.join(missing)}")
        print("Copy .env.example to .env and fill in values. See README.md.")
        sys.exit(1)


def _headers():
    return {"Authorization": f"Bearer {TOKEN}"}


def _api_get(url, params=None):
    r = requests.get(url, params=params, headers=_headers(), timeout=30)
    data = r.json()
    if "error" in data:
        print(f"API Error: {data['error'].get('message', data['error'])}")
        sys.exit(1)
    return data


def _api_post(url, payload=None):
    r = requests.post(url, json=payload,
                      headers={**_headers(), "Content-Type": "application/json"}, timeout=30)
    data = r.json()
    if "error" in data:
        print(f"API Error: {data['error'].get('message', data['error'])}")
        sys.exit(1)
    return data


def _api_delete(url):
    r = requests.delete(url, headers=_headers(), timeout=30)
    data = r.json()
    if "error" in data:
        print(f"API Error: {data['error'].get('message', data['error'])}")
        sys.exit(1)
    return data


# ---------------------------------------------------------------------------
# Templates — list / detail / matrix
# ---------------------------------------------------------------------------

def _fetch_all_templates():
    url = f"{GRAPH}/{WABA_ID}/message_templates"
    params = {"fields": "name,status,category,language,id,quality_score", "limit": 100}
    all_t = []
    while url:
        r = requests.get(url, params=params, headers=_headers(), timeout=30)
        data = r.json()
        if "error" in data:
            print(f"API Error: {data['error'].get('message')}")
            sys.exit(1)
        all_t.extend(data.get("data", []))
        url = data.get("paging", {}).get("next")
        params = {}
    return all_t


def cmd_templates(args):
    """List all templates on the WABA."""
    all_t = _fetch_all_templates()
    if args.status:
        all_t = [t for t in all_t if t["status"] == args.status.upper()]
    if args.category:
        all_t = [t for t in all_t if t["category"] == args.category.upper()]
    all_t.sort(key=lambda t: (t["status"], t["name"]))

    if args.json:
        print(json.dumps(all_t, indent=2))
        return

    counts = Counter(t["status"] for t in all_t)
    print(f"Total: {len(all_t)} templates")
    for s in ["APPROVED", "PENDING", "PAUSED", "DISABLED", "REJECTED"]:
        if counts.get(s):
            print(f"  {s}: {counts[s]}")
    print()

    icons = {"APPROVED": "+", "REJECTED": "x", "DISABLED": "-", "PAUSED": "~", "PENDING": "?"}
    print(f"{'#':<3} {'':2} {'Name':<55} {'Lang':<6} {'Category':<16} {'Status':<12}")
    print("-" * 96)
    for i, t in enumerate(all_t, 1):
        icon = icons.get(t["status"], " ")
        print(f"{i:<3} [{icon}] {t['name']:<55} {t['language']:<6} {t['category']:<16} {t['status']:<12}")


def cmd_matrix(args):
    """Grid view: template name × language × status."""
    all_t = _fetch_all_templates()
    by_name: dict[str, dict[str, str]] = {}
    langs: set[str] = set()
    for t in all_t:
        by_name.setdefault(t["name"], {})[t["language"]] = t["status"]
        langs.add(t["language"])
    lang_cols = sorted(langs)

    icons = {"APPROVED": "+", "REJECTED": "x", "DISABLED": "-", "PAUSED": "~", "PENDING": "?"}
    header = f"{'Name':<50} " + " ".join(f"{l:<6}" for l in lang_cols)
    print(header)
    print("-" * len(header))
    for name in sorted(by_name):
        row = by_name[name]
        cells = " ".join(f"{icons.get(row.get(l, ''), ' '):<6}" for l in lang_cols)
        print(f"{name:<50} {cells}")
    print()
    print("Legend: + approved  ? pending  x rejected  ~ paused  - disabled")


def cmd_template(args):
    """Show details for a specific template."""
    data = _api_get(f"{GRAPH}/{WABA_ID}/message_templates",
                    {"name": args.name, "fields": "name,status,category,language,components,quality_score,id"})
    templates = data.get("data", [])
    if not templates:
        print(f"Template '{args.name}' not found.")
        sys.exit(1)
    if args.json:
        print(json.dumps(templates, indent=2))
        return
    for t in templates:
        print(f"Name:     {t['name']}")
        print(f"ID:       {t['id']}")
        print(f"Status:   {t['status']}")
        print(f"Category: {t['category']}")
        print(f"Language: {t['language']}")
        if t.get("quality_score"):
            print(f"Quality:  {t['quality_score']}")
        print()
        for comp in t.get("components", []):
            ctype = comp["type"]
            if ctype == "HEADER":
                fmt = comp.get("format", "TEXT")
                print(f"  HEADER ({fmt}): {comp.get('text', fmt)}")
            elif ctype == "BODY":
                print(f"  BODY: {comp.get('text', '')}")
                ex = comp.get("example", {}).get("body_text", [])
                if ex:
                    print(f"  EXAMPLE: {ex[0]}")
            elif ctype == "FOOTER":
                print(f"  FOOTER: {comp.get('text', '')}")
            elif ctype == "BUTTONS":
                for btn in comp.get("buttons", []):
                    btype = btn.get("type", "?")
                    target = btn.get("url", btn.get("phone_number", ""))
                    print(f"  BUTTON ({btype}): {btn.get('text', '')} -> {target}")
        print()


# ---------------------------------------------------------------------------
# Send
# ---------------------------------------------------------------------------

def _build_template_payload(to, name, lang, params, button_param):
    tpl = {"name": name, "language": {"code": lang or "en"}}
    components = []
    if params:
        components.append({
            "type": "body",
            "parameters": [{"type": "text", "text": p} for p in params],
        })
    if button_param:
        components.append({
            "type": "button", "sub_type": "url", "index": "0",
            "parameters": [{"type": "text", "text": button_param}],
        })
    if components:
        tpl["components"] = components
    return {"messaging_product": "whatsapp", "to": to, "type": "template", "template": tpl}


def _send(payload, label: str) -> dict:
    data = _api_post(f"{GRAPH}/{PHONE_ID}/messages", payload)
    msg = (data.get("messages") or [{}])[0]
    status = msg.get("message_status", "accepted")
    msg_id = msg.get("id", "?")
    print(f"  {label:<35}  status={status}  id={msg_id[:50]}")
    return data


def cmd_send(args):
    """Send a text or template to a phone or recipient alias."""
    wa_id, label = resolve_recipient(args.recipient)
    if args.text:
        payload = {"messaging_product": "whatsapp", "to": wa_id, "type": "text",
                   "text": {"body": args.text}}
    else:
        if not args.template:
            print("ERROR: provide --text or a template name")
            sys.exit(1)
        payload = _build_template_payload(wa_id, args.template, args.lang, args.params, args.button_param)
    _send(payload, label)


def cmd_test(args):
    """Send a template (or text) to multiple recipients at once."""
    if not args.to:
        print("ERROR: --to required with at least one recipient")
        sys.exit(1)
    targets = [resolve_recipient(x) for x in args.to]
    what = f"'{args.template}'" if args.template else "text"
    print(f"Sending {what} to {len(targets)} recipient(s):\n")
    for wa_id, label in targets:
        if args.text:
            payload = {"messaging_product": "whatsapp", "to": wa_id, "type": "text",
                       "text": {"body": args.text}}
        else:
            payload = _build_template_payload(wa_id, args.template, args.lang, args.params, args.button_param)
        try:
            _send(payload, label)
        except SystemExit:
            pass


# ---------------------------------------------------------------------------
# Create template (text / buttons / url / image header)
# ---------------------------------------------------------------------------

def _upload_image_to_meta(image_path: str) -> str:
    """Resumable upload. Returns a media handle to use in the IMAGE header."""
    if not APP_ID:
        print("ERROR: META_APP_ID env var required for image uploads.")
        sys.exit(1)
    p = Path(image_path)
    if not p.exists():
        print(f"ERROR: image not found: {image_path}")
        sys.exit(1)
    size = p.stat().st_size
    mime = "image/png" if p.suffix.lower() == ".png" else "image/jpeg"

    r = requests.post(
        f"{GRAPH}/{APP_ID}/uploads",
        params={"file_name": p.name, "file_length": size, "file_type": mime,
                "access_token": TOKEN},
        timeout=30,
    )
    r.raise_for_status()
    session_id = r.json()["id"]

    r2 = requests.post(
        f"{GRAPH}/{session_id}",
        headers={"Authorization": f"OAuth {TOKEN}", "file_offset": "0"},
        data=p.read_bytes(),
        timeout=120,
    )
    r2.raise_for_status()
    return r2.json()["h"]


def _create_template_payload(name, body, lang, category, header=None, header_image_handle=None,
                             footer=None, buttons=None, url_button=None, body_example=None):
    components = []
    if header_image_handle:
        components.append({
            "type": "HEADER", "format": "IMAGE",
            "example": {"header_handle": [header_image_handle]},
        })
    elif header:
        components.append({"type": "HEADER", "format": "TEXT", "text": header})

    body_comp = {"type": "BODY", "text": body}
    if body_example:
        body_comp["example"] = {"body_text": [body_example]}
    components.append(body_comp)

    if footer:
        components.append({"type": "FOOTER", "text": footer})

    btns = []
    if buttons:
        btns.extend({"type": "QUICK_REPLY", "text": b} for b in buttons)
    if url_button:
        btn_text, btn_url = url_button
        btns.append({"type": "URL", "text": btn_text, "url": btn_url})
    if btns:
        components.append({"type": "BUTTONS", "buttons": btns})

    return {"name": name, "language": lang or "en",
            "category": (category or "MARKETING").upper(), "components": components}


def cmd_create(args):
    """Create a text-only / button / url-button template."""
    payload = _create_template_payload(
        name=args.name, body=args.body, lang=args.lang, category=args.category,
        header=args.header, footer=args.footer, buttons=args.buttons,
        url_button=args.url_button, body_example=args.example,
    )
    data = _api_post(f"{GRAPH}/{WABA_ID}/message_templates", payload)
    print(f"Created: {data.get('id')} | Status: {data.get('status')} | Category: {data.get('category')}")
    if args.wait:
        _poll_approval(args.name, max_minutes=args.wait)


def cmd_create_image(args):
    """Create a template with an IMAGE header (uploads the image first)."""
    print(f"Uploading image: {args.image}")
    handle = _upload_image_to_meta(args.image)
    print(f"  handle: {handle[:60]}...")
    payload = _create_template_payload(
        name=args.name, body=args.body, lang=args.lang, category=args.category,
        header_image_handle=handle, footer=args.footer, buttons=args.buttons,
        url_button=args.url_button, body_example=args.example,
    )
    data = _api_post(f"{GRAPH}/{WABA_ID}/message_templates", payload)
    print(f"Created: {data.get('id')} | Status: {data.get('status')} | Category: {data.get('category')}")
    if args.wait:
        _poll_approval(args.name, max_minutes=args.wait)


def _poll_approval(name: str, max_minutes: int = 30, interval_s: int = 60):
    """Poll Meta every interval_s until APPROVED/REJECTED or max_minutes elapses.

    Prints a one-line progress tick on each cycle (flushed) so the caller sees
    live status. Returns the final template dict (or None on timeout).
    """
    start = time.time()
    print(f"  Polling approval (up to {max_minutes}m, every {interval_s}s)...", flush=True)
    while (time.time() - start) < max_minutes * 60:
        r = requests.get(
            f"{GRAPH}/{WABA_ID}/message_templates",
            params={"name": name, "fields": "name,status,id,rejected_reason"},
            headers=_headers(), timeout=15,
        )
        data = r.json().get("data", [])
        for t in data:
            elapsed = int((time.time() - start) / 60)
            print(f"  [{elapsed}m] status={t['status']}", flush=True)
            if t["status"] == "APPROVED":
                print(f"  ✓ Approved: {name}", flush=True)
                return t
            if t["status"] in ("REJECTED", "DISABLED"):
                reason = t.get("rejected_reason", "")
                print(f"  ✗ {t['status']}: {name}" + (f" — {reason}" if reason else ""), flush=True)
                return t
        time.sleep(interval_s)
    print(f"  ⌛ Still pending after {max_minutes}m — check later with `wacli template {name}`", flush=True)
    return None


# ---------------------------------------------------------------------------
# Delete / insights / phone / token / recipients
# ---------------------------------------------------------------------------

def cmd_delete(args):
    url = f"{GRAPH}/{WABA_ID}/message_templates?name={args.name}&access_token={TOKEN}"
    data = _api_delete(url)
    if data.get("success"):
        print(f"Deleted template: {args.name}")
    else:
        print(f"Failed to delete: {data}")


def cmd_insights(args):
    data = _api_get(f"{GRAPH}/{WABA_ID}/message_templates",
                    {"name": args.name, "fields": "name,id,status"})
    templates = data.get("data", [])
    if not templates:
        print(f"Template '{args.name}' not found.")
        sys.exit(1)
    tid = templates[0]["id"]
    print(f"Template: {args.name} (ID: {tid}, Status: {templates[0]['status']})")
    print()
    params = {
        "template_ids": f'["{tid}"]',
        "granularity": "LIFETIME" if args.lifetime else "DAILY",
    }
    if args.start:
        params["start"] = args.start
    if args.end:
        params["end"] = args.end
    r = requests.get(f"{GRAPH}/{WABA_ID}/template_analytics",
                     params=params, headers=_headers(), timeout=30)
    analytics = r.json()
    if "error" in analytics:
        print(f"Analytics not available: {analytics['error'].get('message', '?')}")
        return
    if args.json:
        print(json.dumps(analytics, indent=2))
        return
    for dp in analytics.get("data", []):
        for p in dp.get("data_points", []):
            print(f"  Sent: {p.get('sent', 0)} | Delivered: {p.get('delivered', 0)} | "
                  f"Read: {p.get('read', 0)} | Start: {p.get('start', '?')} | End: {p.get('end', '?')}")


def cmd_phone(args):
    fields = ("id,verified_name,display_phone_number,quality_rating,platform_type,"
              "throughput,code_verification_status,name_status,status,account_mode")
    data = _api_get(f"{GRAPH}/{PHONE_ID}", {"fields": fields})
    if args.json:
        print(json.dumps(data, indent=2))
        return
    print(f"  Phone:       {data.get('display_phone_number', '?')}")
    print(f"  Name:        {data.get('verified_name', '?')}")
    print(f"  Name status: {data.get('name_status', '?')}")
    print(f"  ID:          {data.get('id', PHONE_ID)}")
    print(f"  Status:      {data.get('status', '?')}")
    print(f"  Platform:    {data.get('platform_type', '?')}")
    print(f"  Mode:        {data.get('account_mode', '?')}")
    print(f"  Quality:     {data.get('quality_rating', '?')}")
    tp = data.get("throughput", {})
    if tp:
        print(f"  Throughput:  {tp.get('level', '?')}")


def cmd_token_check(args):
    data = _api_get(f"{GRAPH}/debug_token", {"input_token": TOKEN, "access_token": TOKEN})
    d = data.get("data", {})
    if args.json:
        print(json.dumps(d, indent=2))
        return
    print(f"App:     {d.get('application', '?')} ({d.get('app_id', '?')})")
    print(f"Type:    {d.get('type', '?')}")
    print(f"Valid:   {d.get('is_valid', '?')}")
    expires = d.get("expires_at", 0)
    print(f"Expires: {'Never' if expires == 0 else expires}")
    print(f"Scopes:  {', '.join(d.get('scopes', []))}")
    print()
    wa_scopes = [s for s in d.get("granular_scopes", []) if "whatsapp" in s.get("scope", "")]
    if wa_scopes:
        print("WhatsApp permissions:")
        for s in wa_scopes:
            targets = s.get("target_ids", [])
            status = f"WABAs: {targets}" if targets else "(business-wide access)"
            print(f"  {s['scope']}: {status}")
    else:
        print("WARNING: No WhatsApp scopes on this token.")


def cmd_recipients(args):
    """List resolvable recipients from recipients.json."""
    people = _load_recipients()
    if not people:
        print(f"No recipients loaded. Create a {RECIPIENTS_FILE} file — see recipients.example.json.")
        return
    print(f"{'Alias':<8} {'Name':<25} {'WhatsApp':<15}")
    print("-" * 50)
    for m in people:
        wa = m.get("whatsapp") or "-"
        print(f"{m.get('alias','?'):<8} {m.get('name','?'):<25} {wa:<15}")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        prog="wacli",
        description="WhatsApp Business Cloud API from the command line",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__.split("Common examples:")[1] if "Common examples:" in __doc__ else "",
    )
    parser.add_argument("--json", action="store_true", help="Raw JSON output")
    sub = parser.add_subparsers(dest="command", required=True)

    p = sub.add_parser("templates", help="List templates")
    p.add_argument("--status", help="approved/pending/rejected/paused/disabled")
    p.add_argument("--category", help="marketing/utility/authentication")
    p.set_defaults(func=cmd_templates)

    p = sub.add_parser("matrix", help="Template × language × status grid")
    p.set_defaults(func=cmd_matrix)

    p = sub.add_parser("template", help="Show template details")
    p.add_argument("name")
    p.set_defaults(func=cmd_template)

    p = sub.add_parser("send", help="Send to a phone or recipient alias")
    p.add_argument("recipient", help="phone digits, alias, or name substring")
    p.add_argument("template", nargs="?", help="Template name")
    p.add_argument("--text", help="Free-form text (session window only)")
    p.add_argument("--params", nargs="+", help="Body variables in order")
    p.add_argument("--button-param", help="Dynamic URL button parameter")
    p.add_argument("--lang", default="en", help="Template language (default en)")
    p.set_defaults(func=cmd_send)

    p = sub.add_parser("test", help="Send (template or text) to multiple recipients")
    p.add_argument("template", nargs="?", help="Template name (omit with --text)")
    p.add_argument("--to", nargs="+", required=True, help="Recipients (phone/alias/name)")
    p.add_argument("--text", help="Free-form text instead of template")
    p.add_argument("--params", nargs="+", help="Body variables in order")
    p.add_argument("--button-param", help="Dynamic URL button parameter")
    p.add_argument("--lang", default="en", help="Template language")
    p.set_defaults(func=cmd_test)

    p = sub.add_parser("create", help="Create a new template")
    p.add_argument("name")
    p.add_argument("--body", required=True, help="Body ({{1}} for variables)")
    p.add_argument("--header", help="Header text")
    p.add_argument("--footer")
    p.add_argument("--category", default="MARKETING")
    p.add_argument("--buttons", nargs="+", help="Quick-reply labels")
    p.add_argument("--url-button", nargs=2, metavar=("TEXT", "URL"))
    p.add_argument("--example", nargs="+", help="Example values for body vars")
    p.add_argument("--lang", default="en")
    p.add_argument("--wait", type=int, default=0,
                   help="Poll for approval, max minutes (0 = don't poll)")
    p.set_defaults(func=cmd_create)

    p = sub.add_parser("create-image", help="Create template with IMAGE header")
    p.add_argument("name")
    p.add_argument("--body", required=True)
    p.add_argument("--image", required=True, help="Path to header image (png/jpg)")
    p.add_argument("--footer")
    p.add_argument("--category", default="MARKETING")
    p.add_argument("--buttons", nargs="+")
    p.add_argument("--url-button", nargs=2, metavar=("TEXT", "URL"))
    p.add_argument("--example", nargs="+")
    p.add_argument("--lang", default="en")
    p.add_argument("--wait", type=int, default=0,
                   help="Poll for approval, max minutes (0 = don't poll)")
    p.set_defaults(func=cmd_create_image)

    p = sub.add_parser("delete", help="Delete a template")
    p.add_argument("name")
    p.set_defaults(func=cmd_delete)

    p = sub.add_parser("insights", help="Template analytics")
    p.add_argument("name")
    p.add_argument("--lifetime", action="store_true")
    p.add_argument("--start")
    p.add_argument("--end")
    p.set_defaults(func=cmd_insights)

    p = sub.add_parser("phone", help="Phone number health")
    p.set_defaults(func=cmd_phone)

    p = sub.add_parser("token-check", help="Verify token + scopes")
    p.set_defaults(func=cmd_token_check)

    p = sub.add_parser("recipients", help="List resolvable recipients")
    p.set_defaults(func=cmd_recipients)

    args = parser.parse_args()
    _check_token()
    args.func(args)


if __name__ == "__main__":
    main()