workphone-sdk/sdk/phone

#!/usr/bin/env python3
"""
📱 phone - 工作手机命令行控制工具
用自然语言或命令控制 Android 手机

用法:
  phone <自然语言指令>       AI 智能控制（带执行）
  phone ai <指令>            AI 智能控制（带执行）
  phone parse <指令>         AI 纯解析（不执行）
  phone cmd <命令>           模式匹配命令
  phone home                 回主页
  phone back                 返回
  phone screenshot           截图
  phone apps                 查看已安装APP
  phone status               查看系统状态
  phone chat                 进入交互对话模式

示例:
  phone 打开微信
  phone ai "打开Chrome搜索今天天气"
  phone cmd "打开微信；然后点击发现"
  phone chat
"""

import sys
import json
import urllib.request
import urllib.error
import readline

SDK_URL = "http://localhost:8899/api/v3"
DEVICE_ID = "emulator-5554"

# 颜色
GREEN = "\033[92m"
RED = "\033[91m"
CYAN = "\033[96m"
YELLOW = "\033[93m"
BOLD = "\033[1m"
DIM = "\033[2m"
RESET = "\033[0m"

def api_call(method, path, data=None, timeout=30):
    """调用 SDK API"""
    url = f"http://localhost:8899{path}"
    headers = {"Content-Type": "application/json"}
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return json.loads(resp.read())
    except urllib.error.URLError as e:
        return {"error": f"SDK 未连接: {e}"}
    except Exception as e:
        return {"error": str(e)}

def ai_chat(message, execute=True):
    """AI 智能控制"""
    data = {"message": message}
    if execute:
        data["device_id"] = DEVICE_ID
    result = api_call("POST", "/api/v3/ai/chat", data, timeout=30)
    if "error" in result:
        print(f"  {RED}✗ {result['error']}{RESET}")
        return
    d = result.get("data", {})
    actions = d.get("actions", [])
    executed = d.get("executed", [])
    
    # 显示 AI 解析结果
    print(f"  {CYAN}🧠 {d.get('message', '')}{RESET}")
    if actions:
        for i, a in enumerate(actions):
            detail = a.get("app", a.get("text", a.get("key", a.get("direction", a.get("seconds", "")))))
            print(f"  {DIM}   {i+1}. {a['action']}: {detail}{RESET}")
    
    # 显示执行结果
    if executed:
        print()
        for e in executed:
            icon = f"{GREEN}✅" if e.get("success") else f"{RED}❌"
            print(f"  {icon} {e.get('step', '')}{RESET}")

def agent_execute(task):
    """模式匹配命令"""
    result = api_call("POST", "/api/v3/agent/execute", {
        "device_id": DEVICE_ID, "task": task
    }, timeout=30)
    if "error" in result:
        print(f"  {RED}✗ {result['error']}{RESET}")
        return
    d = result.get("data", {})
    engine = " [AI]" if d.get("engine") == "ai" else ""
    steps = d.get("steps", [])
    if d.get("success"):
        print(f"  {GREEN}✓{engine} {' → '.join(steps)}{RESET}")
    else:
        print(f"  {RED}✗{engine} {d.get('error', ' → '.join(steps))}{RESET}")

def show_status():
    """显示系统状态"""
    # 健康检查
    health = api_call("GET", "/health")
    if "error" in health:
        print(f"  {RED}SDK 离线: {health['error']}{RESET}")
        return
    print(f"  {GREEN}SDK 在线{RESET}")
    print(f"  设备: {health.get('adb_devices', 0)} ADB / {health.get('devices_online', 0)} WebSocket")
    
    # AI 状态
    ai = api_call("GET", "/api/v3/ai/status")
    ai_data = ai.get("data", {})
    if ai_data.get("available"):
        print(f"  AI: {GREEN}{ai_data['backend']} · {ai_data['model']}{RESET}")
    else:
        print(f"  AI: {RED}不可用{RESET}")

def interactive_chat():
    """交互式对话模式"""
    print(f"\n{BOLD}📱 工作手机 AI 对话控制{RESET}")
    print(f"{DIM}输入自然语言控制手机，输入 quit/exit 退出{RESET}")
    print(f"{DIM}前缀 ! 使用模式匹配，前缀 ? 纯解析不执行{RESET}\n")
    
    show_status()
    print()
    
    while True:
        try:
            msg = input(f"{YELLOW}📱 > {RESET}").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n再见！")
            break
        
        if not msg:
            continue
        if msg.lower() in ("quit", "exit", "q"):
            print("再见！")
            break
        if msg == "status":
            show_status()
            continue
        if msg == "screenshot":
            print("  截图中...")
            agent_execute("截图")
            continue
        
        if msg.startswith("!"):
            # 模式匹配
            agent_execute(msg[1:].strip())
        elif msg.startswith("?"):
            # 纯解析
            ai_chat(msg[1:].strip(), execute=False)
        else:
            # AI 智能控制
            ai_chat(msg)
        print()

def main():
    args = sys.argv[1:]
    
    if not args:
        print(__doc__)
        return
    
    cmd = args[0].lower()
    rest = " ".join(args[1:]) if len(args) > 1 else ""
    
    if cmd == "chat":
        interactive_chat()
    elif cmd == "status":
        show_status()
    elif cmd == "home":
        agent_execute("回主页")
    elif cmd == "back":
        agent_execute("返回")
    elif cmd == "screenshot":
        agent_execute("截图")
    elif cmd == "apps":
        result = api_call("GET", f"/api/v3/adb/devices/{DEVICE_ID}/apps")
        if "error" not in result:
            packages = result.get("data", {}).get("packages", [])
            print(f"  已安装 {len(packages)} 个第三方APP:")
            for p in packages:
                print(f"    {p}")
        else:
            print(f"  {RED}{result['error']}{RESET}")
    elif cmd == "ai":
        if not rest:
            print("用法: phone ai <自然语言指令>")
            return
        ai_chat(rest)
    elif cmd == "parse":
        if not rest:
            print("用法: phone parse <自然语言指令>")
            return
        ai_chat(rest, execute=False)
    elif cmd == "cmd":
        if not rest:
            print("用法: phone cmd <命令>")
            return
        agent_execute(rest)
    else:
        # 默认：整个参数作为自然语言指令
        full_cmd = " ".join(args)
        ai_chat(full_cmd)

if __name__ == "__main__":
    main()