codex-oauth-plus-onboarding-skill/scripts/luban_sms.py

#!/usr/bin/env python3
"""Luban SMS API client for Codex/OpenAI phone verification.

Reads secrets from ~/.hermes/env/codex_oauth_onboarding.env.
Never prints API keys, full phone numbers, or SMS codes unless explicitly requested by caller.

Docs: https://lubansms.com/api_docs/
"""
from __future__ import annotations

import argparse
import json
import os
import pathlib
import re
import shlex
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from typing import Any, Dict, Iterable, List, Optional, Tuple

ENV_PATH = pathlib.Path.home() / ".hermes/env/codex_oauth_onboarding.env"
DEFAULT_BASE = "https://lubansms.com/v2/api"


def load_env(path: pathlib.Path = ENV_PATH) -> Dict[str, str]:
    env: Dict[str, str] = {}
    if not path.exists():
        return env
    for line in path.read_text().splitlines():
        s = line.strip()
        if not s or s.startswith("#") or "=" not in s:
            continue
        if s.startswith("export "):
            s = s[7:].strip()
        k, v = s.split("=", 1)
        k = k.strip()
        v = v.strip()
        if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", k):
            continue
        try:
            if v and v[0] in "'\"":
                v = shlex.split(v)[0]
        except Exception:
            pass
        env[k] = v.strip("'\"")
    return env


def redact_phone(phone: Any) -> Optional[str]:
    if phone is None:
        return None
    s = re.sub(r"\D", "", str(phone))
    if len(s) <= 4:
        return "[REDACTED]"
    return "[REDACTED]" + s[-4:]


def strict_code(text: str) -> Optional[str]:
    m = re.search(r"(?<!\d)(\d{6})(?!\d)", text or "")
    return m.group(1) if m else None


class LubanSms:
    def __init__(self, apikey: str, base: str = DEFAULT_BASE, timeout: int = 30):
        self.apikey = apikey
        self.base = base.rstrip("/")
        self.timeout = timeout

    @classmethod
    def from_env(cls) -> "LubanSms":
        env = load_env()
        key = env.get("LUBAN_SMS_API_KEY") or env.get("LUBAN_API_KEY")
        if not key:
            raise SystemExit("missing LUBAN_SMS_API_KEY in ~/.hermes/env/codex_oauth_onboarding.env")
        return cls(key, env.get("LUBAN_SMS_API_BASE", DEFAULT_BASE))

    def get(self, endpoint: str, **params: Any) -> Tuple[int, Dict[str, Any], str]:
        params = {k: v for k, v in params.items() if v is not None}
        params["apikey"] = self.apikey
        url = f"{self.base}/{endpoint.lstrip('/')}?{urllib.parse.urlencode(params)}"
        req = urllib.request.Request(url, headers={
            "Accept": "application/json",
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36",
        })
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                raw = r.read().decode(errors="ignore")
                status = r.status
        except urllib.error.HTTPError as e:
            raw = e.read().decode(errors="ignore")
            status = e.code
        except Exception as e:
            return 0, {"code": 0, "msg": f"{type(e).__name__}: {e}"}, ""
        try:
            data = json.loads(raw)
        except Exception:
            data = {"raw": raw[:500]}
        return status, data, raw

    def balance(self) -> Tuple[int, Dict[str, Any], str]:
        return self.get("getBalance")

    def services(self, country: Optional[str] = None, service: Optional[str] = None, language: str = "en", page: int = 1):
        return self.get("List", country=country, service=service, language=language, page=page)

    def countries(self):
        return self.get("countries")

    def buy(self, service_id: str) -> Tuple[int, Dict[str, Any], str]:
        return self.get("getNumber", service_id=service_id)

    def sms(self, request_id: str) -> Tuple[int, Dict[str, Any], str]:
        return self.get("getSms", request_id=request_id)

    def reject(self, request_id: str) -> Tuple[int, Dict[str, Any], str]:
        return self.get("setStatus", request_id=request_id, status="reject")

    def again(self, request_id: str) -> Tuple[int, Dict[str, Any], str]:
        return self.get("getAgainNmuber", request_id=request_id)

    def history(self, language: str = "en", page: int = 1) -> Tuple[int, Dict[str, Any], str]:
        return self.get("smsHistory", language=language, page=page)


def summarize_services(data: Dict[str, Any], limit: int = 20) -> List[Dict[str, Any]]:
    msg = data.get("msg")
    rows = msg if isinstance(msg, list) else []
    out: List[Dict[str, Any]] = []
    for r in rows[:limit]:
        if not isinstance(r, dict):
            continue
        out.append({
            "service_id": str(r.get("service_id", "")),
            "country_en": r.get("country_name_en") or r.get("country_name") or r.get("country"),
            "country_zh": r.get("country_name_zh"),
            "service_name": r.get("service_name"),
            "provider": r.get("provider"),
            "cost": r.get("cost"),
        })
    return out


def resolve_service_id(api: LubanSms, country: Optional[str], service: str, language: str = "en", pages: int = 3) -> Optional[Dict[str, Any]]:
    # Exact service_name match preferred, then substring match. Keep first cheapest by numeric cost when possible.
    candidates: List[Dict[str, Any]] = []
    for page in range(1, pages + 1):
        st, data, _ = api.services(country=country, service=service, language=language, page=page)
        if st != 200 or data.get("code") != 0:
            continue
        for row in summarize_services(data, limit=500):
            name = str(row.get("service_name") or "")
            if service.lower() in name.lower() or name.lower() in service.lower():
                candidates.append(row)
    if not candidates:
        return None
    def cost_key(r: Dict[str, Any]) -> float:
        try: return float(r.get("cost"))
        except Exception: return 999999.0
    exact = [r for r in candidates if str(r.get("service_name", "")).lower() == service.lower()]
    return sorted(exact or candidates, key=cost_key)[0]


def cmd_balance(args: argparse.Namespace) -> int:
    api = LubanSms.from_env()
    st, data, _ = api.balance()
    print(json.dumps({"http": st, "code": data.get("code"), "ok": st == 200 and data.get("code") == 0, "balance": data.get("balance"), "msg": data.get("msg")}, ensure_ascii=False))
    return 0 if st == 200 and data.get("code") == 0 else 1


def cmd_services(args: argparse.Namespace) -> int:
    api = LubanSms.from_env()
    st, data, _ = api.services(country=args.country, service=args.service, language=args.language, page=args.page)
    print(json.dumps({"http": st, "code": data.get("code"), "ok": st == 200 and data.get("code") == 0, "services": summarize_services(data, args.limit), "msg_type": type(data.get("msg")).__name__}, ensure_ascii=False))
    return 0 if st == 200 and data.get("code") == 0 else 1


def cmd_resolve(args: argparse.Namespace) -> int:
    api = LubanSms.from_env()
    row = resolve_service_id(api, args.country, args.service, args.language, args.pages)
    print(json.dumps({"ok": bool(row), "selected": row}, ensure_ascii=False))
    return 0 if row else 2


def cmd_buy(args: argparse.Namespace) -> int:
    api = LubanSms.from_env()
    service_id = args.service_id
    selected = None
    if not service_id:
        selected = resolve_service_id(api, args.country, args.service, args.language, args.pages)
        if not selected:
            print(json.dumps({"phase": "resolve_service_fail", "country": args.country, "service": args.service}, ensure_ascii=False))
            return 2
        service_id = selected["service_id"]
    st, data, _ = api.buy(service_id)
    ok = st == 200 and data.get("code") == 0 and data.get("request_id") and data.get("number")
    print(json.dumps({"http": st, "ok": bool(ok), "code": data.get("code"), "msg": data.get("msg"), "request_id": data.get("request_id"), "phone_redacted": redact_phone(data.get("number")), "selected": selected}, ensure_ascii=False))
    return 0 if ok else 1


def cmd_sms(args: argparse.Namespace) -> int:
    api = LubanSms.from_env()
    st, data, _ = api.sms(args.request_id)
    code = data.get("sms_code") or strict_code(json.dumps(data, ensure_ascii=False))
    print(json.dumps({"http": st, "ok": st == 200 and data.get("code") == 0, "msg": data.get("msg"), "has_code": bool(code), "code_len": len(str(code)) if code else 0}, ensure_ascii=False))
    return 0


def cmd_reject(args: argparse.Namespace) -> int:
    api = LubanSms.from_env()
    st, data, _ = api.reject(args.request_id)
    print(json.dumps({"http": st, "ok": st == 200 and data.get("code") == 0, "code": data.get("code"), "msg": data.get("msg")}, ensure_ascii=False))
    return 0 if st == 200 and data.get("code") == 0 else 1


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="Luban SMS API helper")
    sub = p.add_subparsers(dest="cmd", required=True)
    sub.add_parser("balance")
    s = sub.add_parser("services"); s.add_argument("--country"); s.add_argument("--service", default="OpenAI"); s.add_argument("--language", default="en"); s.add_argument("--page", type=int, default=1); s.add_argument("--limit", type=int, default=30)
    r = sub.add_parser("resolve"); r.add_argument("--country"); r.add_argument("--service", default="OpenAI"); r.add_argument("--language", default="en"); r.add_argument("--pages", type=int, default=3)
    b = sub.add_parser("buy"); b.add_argument("--service-id"); b.add_argument("--country"); b.add_argument("--service", default="OpenAI"); b.add_argument("--language", default="en"); b.add_argument("--pages", type=int, default=3)
    g = sub.add_parser("sms"); g.add_argument("request_id")
    x = sub.add_parser("reject"); x.add_argument("request_id")
    return p


def main(argv: Optional[List[str]] = None) -> int:
    args = build_parser().parse_args(argv)
    return globals()[f"cmd_{args.cmd}"](args)


if __name__ == "__main__":
    raise SystemExit(main())