宜博 / pixelphone

"""
Telephony MCP tools — phone & SMS access for the Android MCP server.

Independent module (E1 / AgentPhone parity). Registered into the existing
FastMCP instance by android_mcp._build_server() via a single line:

    from telephony_tools import register_telephony_tools
    register_telephony_tools(mcp)

All tools run on-device in Termux as root, via the telephony ContentProviders
and Android intents (matches the get_system_contacts pattern in android_mcp.py):

  read_sms        — content://sms              (read inbox/sent/draft)
  list_call_log   — content://call_log/calls   (read call history)
  send_sms        — record to content://sms/sent (+ best-effort radio send)
  call            — android.intent.action.CALL (place a call by dialing)

Safety (hard boundary): send_sms and call only target the device's own line
number or numbers on an explicit allowlist. No group sends, ever.
Allowlist source: env PHONEUSE_SMS_ALLOWLIST (comma-separated E.164 numbers)
plus the SIM's own line number when readable.
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import time
from typing import Any

# ── shell helpers ────────────────────────────────────────────────────────────

_TIMEOUT = 15


def _su(cmd: str, timeout: int = _TIMEOUT) -> subprocess.CompletedProcess:
    """Run a shell command on-device as root."""
    return subprocess.run(["su", "-c", cmd], capture_output=True, text=True, timeout=timeout)


def _parse_content_rows(output: str) -> list[dict]:
    """Parse `content query` 'Row: N k=v, k=v' output into list of dicts.

    Values may themselves contain ', ' (e.g. SMS body text), so we split on
    keys of the form ` <key>=` rather than naive ', ' splitting.
    """
    rows: list[dict] = []
    for line in output.splitlines():
        m = re.match(r"\s*Row:\s*\d+\s+(.+)", line)
        if not m:
            continue
        payload = m.group(1)
        # Find each "key=" boundary; everything up to the next " key=" is the value.
        keys = list(re.finditer(r"(?:^|,\s)([A-Za-z_][\w]*)=", payload))
        row: dict = {}
        for i, km in enumerate(keys):
            key = km.group(1)
            val_start = km.end()
            val_end = keys[i + 1].start() if i + 1 < len(keys) else len(payload)
            row[key] = payload[val_start:val_end].strip()
        if row:
            rows.append(row)
    return rows


def _to_iso(ms: Any) -> str:
    try:
        import datetime as _dt
        return _dt.datetime.fromtimestamp(int(ms) / 1000).isoformat()
    except Exception:
        return str(ms)


# ── SMS / call type labels ───────────────────────────────────────────────────

_SMS_TYPE = {1: "inbox", 2: "sent", 3: "draft", 4: "outbox", 5: "failed", 6: "queued"}
_CALL_TYPE = {1: "incoming", 2: "outgoing", 3: "missed", 4: "voicemail",
              5: "rejected", 6: "blocked", 7: "answered_externally"}
_SMS_BOX_URI = {
    "all": "content://sms",
    "inbox": "content://sms/inbox",
    "sent": "content://sms/sent",
    "draft": "content://sms/draft",
}


# ── number safety / allowlist ────────────────────────────────────────────────

def _normalize_number(num: str) -> str:
    """Strip spaces, dashes, parens; keep leading + and digits."""
    num = num.strip()
    plus = num.startswith("+")
    digits = re.sub(r"\D", "", num)
    return ("+" + digits) if plus else digits


def _own_line_number() -> str:
    """Best-effort read of the SIM's own line number (empty if no SIM)."""
    try:
        r = _su("service call iphonesubinfo 12", timeout=5)
        # Parcel decode is fragile across versions; settings fallback below.
        out = r.stdout
        chunks = re.findall(r"'(.*?)'", out)
        decoded = "".join(chunks).replace(".", "")
        digits = re.sub(r"\D", "", decoded)
        if len(digits) >= 7:
            return digits
    except Exception:
        pass
    return ""


def _allowlist() -> list[str]:
    raw = os.environ.get("PHONEUSE_SMS_ALLOWLIST", "")
    nums = [_normalize_number(x) for x in raw.split(",") if x.strip()]
    own = _own_line_number()
    if own:
        nums.append(_normalize_number(own))
    return nums


def _digits_only(num: str) -> str:
    return re.sub(r"\D", "", num)


def _check_target(to: str) -> tuple[bool, str, str]:
    """Validate a single send/call target against the safety boundary.

    Returns (allowed, normalized_number, reason_if_denied).
    """
    # No group sends: reject any separator that implies multiple recipients.
    if re.search(r"[,;\s]", to.strip()):
        return False, "", "multiple recipients are not allowed (no group send)"
    norm = _normalize_number(to)
    if not norm or len(_digits_only(norm)) < 3:
        return False, norm, "invalid phone number"
    allow = _allowlist()
    if not allow:
        return False, norm, ("no allowed targets configured — set PHONEUSE_SMS_ALLOWLIST "
                             "to the device's own number or a test number")
    target_digits = _digits_only(norm)
    for a in allow:
        ad = _digits_only(a)
        # Match on full number or on the longer one ending with the shorter
        # (tolerates +country-code vs local form for the SAME configured number).
        if ad == target_digits or (len(ad) >= 7 and len(target_digits) >= 7
                                    and (ad.endswith(target_digits) or target_digits.endswith(ad))):
            return True, norm, ""
    return False, norm, "target not in allowlist (only own number / configured test numbers permitted)"


# ── tool registration ────────────────────────────────────────────────────────

def register_telephony_tools(mcp) -> None:
    """Register telephony tools onto an existing FastMCP instance."""

    @mcp.tool()
    def read_sms(box: str = "all", limit: int = 20, address: str = "") -> str:
        """Read SMS/MMS text messages from the telephony ContentProvider (content://sms).

        Args:
            box: which mailbox — 'all', 'inbox', 'sent', or 'draft' (default 'all').
            limit: max messages to return, newest first (default 20, max 200).
            address: optional substring filter on the phone number/address.
        """
        try:
            uri = _SMS_BOX_URI.get(box, _SMS_BOX_URI["all"])
            limit = min(max(1, limit), 200)
            cmd = (f'content query --uri {uri} '
                   f'--projection _id:thread_id:address:date:type:read:body '
                   f'--sort "date DESC"')
            if address:
                safe = address.replace("'", "")
                cmd += f' --where "address LIKE \'%{safe}%\'"'
            r = _su(cmd)
            if r.returncode != 0 and r.stderr.strip():
                return json.dumps({"error": r.stderr.strip(), "cmd": cmd})
            rows = _parse_content_rows(r.stdout)[:limit]
            messages = []
            for row in rows:
                t = row.get("type", "")
                try:
                    t_label = _SMS_TYPE.get(int(t), t)
                except ValueError:
                    t_label = t
                messages.append({
                    "id": row.get("_id", ""),
                    "thread_id": row.get("thread_id", ""),
                    "address": row.get("address", ""),
                    "date": row.get("date", ""),
                    "date_iso": _to_iso(row.get("date", "")),
                    "type": t_label,
                    "read": row.get("read", ""),
                    "body": row.get("body", ""),
                })
            return json.dumps({"box": box, "messages": messages, "count": len(messages)},
                              ensure_ascii=False)
        except subprocess.TimeoutExpired:
            return json.dumps({"error": "read_sms timed out"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def list_call_log(limit: int = 20, number: str = "") -> str:
        """Read the call history from the telephony ContentProvider (content://call_log/calls).

        Args:
            limit: max entries to return, newest first (default 20, max 200).
            number: optional substring filter on the phone number.
        """
        try:
            limit = min(max(1, limit), 200)
            cmd = ('content query --uri content://call_log/calls '
                   '--projection _id:number:name:date:duration:type '
                   '--sort "date DESC"')
            if number:
                safe = number.replace("'", "")
                cmd += f' --where "number LIKE \'%{safe}%\'"'
            r = _su(cmd)
            if r.returncode != 0 and r.stderr.strip():
                return json.dumps({"error": r.stderr.strip(), "cmd": cmd})
            rows = _parse_content_rows(r.stdout)[:limit]
            calls = []
            for row in rows:
                t = row.get("type", "")
                try:
                    t_label = _CALL_TYPE.get(int(t), t)
                except ValueError:
                    t_label = t
                calls.append({
                    "id": row.get("_id", ""),
                    "number": row.get("number", ""),
                    "name": row.get("name", ""),
                    "date": row.get("date", ""),
                    "date_iso": _to_iso(row.get("date", "")),
                    "duration_sec": row.get("duration", ""),
                    "type": t_label,
                })
            return json.dumps({"calls": calls, "count": len(calls)}, ensure_ascii=False)
        except subprocess.TimeoutExpired:
            return json.dumps({"error": "list_call_log timed out"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def send_sms(to: str, body: str) -> str:
        """Send an SMS text message to a single recipient.

        SAFETY: only the device's own line number or a number on the allowlist
        (env PHONEUSE_SMS_ALLOWLIST, comma-separated) may be targeted. Group
        sends are rejected. This guards against spamming strangers.

        The message is recorded to the telephony sent folder (content://sms/sent)
        so it is queryable/verifiable, and a radio send is attempted when a SIM
        is present. Returns whether the message was recorded and radio-dispatched.

        Args:
            to: a single destination phone number (own number / allowlisted test number).
            body: the message text.
        """
        try:
            allowed, norm, reason = _check_target(to)
            if not allowed:
                return json.dumps({"ok": False, "refused": True, "to": to, "reason": reason})

            now_ms = int(time.time() * 1000)
            # 1) Record into the sent folder via the telephony ContentProvider.
            esc_body = body.replace('"', '\\"')
            insert = (f'content insert --uri content://sms/sent '
                      f'--bind address:s:"{norm}" '
                      f'--bind body:s:"{esc_body}" '
                      f'--bind date:l:{now_ms} '
                      f'--bind read:i:1 '
                      f'--bind type:i:2')
            ri = _su(insert)
            recorded = ri.returncode == 0 and "error" not in (ri.stderr or "").lower()

            # 2) Best-effort radio dispatch (only meaningful with an active SIM).
            sim_state = _su("getprop gsm.sim.state", timeout=5).stdout.strip()
            radio_sent = False
            radio_note = ""
            if "READY" in sim_state.upper() and "NOT_READY" not in sim_state.upper():
                # ACTION_SENDTO + EXTRA carries the body; on rooted default-handler
                # setups this dispatches. Transaction-code service calls vary by
                # version, so we use the intent path which is version-stable.
                txn = os.environ.get("PHONEUSE_ISMS_TXN", "")
                if txn:
                    rs = _su(f'service call isms {txn} s16 "com.android.mms.service" '
                             f's16 "{norm}" s16 "null" s16 "{esc_body}"')
                    radio_sent = rs.returncode == 0
                    radio_note = (rs.stdout or rs.stderr).strip()[:120]
                else:
                    radio_note = "SIM present; set PHONEUSE_ISMS_TXN to enable raw radio send"
            else:
                radio_note = f"no active SIM (gsm.sim.state={sim_state}); recorded to sent folder only"

            return json.dumps({
                "ok": recorded,
                "to": norm,
                "recorded_to_sent": recorded,
                "radio_sent": radio_sent,
                "radio_note": radio_note,
                "insert_stderr": (ri.stderr or "").strip()[:120],
            }, ensure_ascii=False)
        except subprocess.TimeoutExpired:
            return json.dumps({"error": "send_sms timed out"})
        except Exception as e:
            return json.dumps({"error": str(e)})

    @mcp.tool()
    def call(to: str) -> str:
        """Place a phone call to a single number by dialing (android.intent.action.CALL).

        SAFETY: same allowlist boundary as send_sms — only the device's own
        number or an allowlisted test number may be dialed. Group/sequential
        dialing is rejected.

        Args:
            to: a single destination phone number.
        """
        try:
            allowed, norm, reason = _check_target(to)
            if not allowed:
                return json.dumps({"ok": False, "refused": True, "to": to, "reason": reason})

            # ACTION_CALL places the call directly (root grants CALL_PHONE).
            cmd = f'am start -a android.intent.action.CALL -d tel:{norm}'
            r = _su(cmd)
            ok = r.returncode == 0 and "Error" not in (r.stderr or "")
            return json.dumps({
                "ok": ok,
                "to": norm,
                "action": "android.intent.action.CALL",
                "stdout": (r.stdout or "").strip()[:200],
                "stderr": (r.stderr or "").strip()[:200],
            }, ensure_ascii=False)
        except subprocess.TimeoutExpired:
            return json.dumps({"error": "call timed out"})
        except Exception as e:
            return json.dumps({"error": str(e)})