v5.0-B 手机测试API端点(12个)+CLI mobile命令+json导入修复

Author: zcj220

cli.py

@@ -238,6 +238,124 @@ def cmd_init(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_mobile(args: argparse.Namespace) -> int:
+    """手机测试相关命令。"""
+    import urllib.request
+    import urllib.error
+
+    base = f"http://127.0.0.1:{args.port}/api/v1"
+    sub = getattr(args, "mobile_command", None)
+
+    if not sub:
+        print("请指定子命令: devices | appium | connect | screenshot | sessions")
+        print("  python cli.py mobile devices     列出已连接设备")
+        print("  python cli.py mobile appium      检查Appium状态")
+        print("  python cli.py mobile connect      连接设备")
+        print("  python cli.py mobile sessions     列出活跃会话")
+        return 0
+
+    if sub == "devices":
+        try:
+            req = urllib.request.Request(f"{base}/mobile/devices")
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                data = json.loads(resp.read().decode())
+                devices = data.get("devices", [])
+                if not devices:
+                    print("📱 未检测到已连接的Android设备")
+                    if data.get("error"):
+                        print(f"   ⚠️ {data['error']}")
+                    print("   请确保: USB调试已开启 + 设备已连接 + adb已安装")
+                else:
+                    print(f"📱 检测到 {len(devices)} 台设备:")
+                    for d in devices:
+                        model = d.get("model", "未知型号")
+                        print(f"   • {d['serial']} ({model})")
+                return 0
+        except urllib.error.URLError:
+            print(f"❌ 无法连接引擎，请先运行: python cli.py serve")
+            return 2
+
+    elif sub == "appium":
+        try:
+            req = urllib.request.Request(f"{base}/mobile/appium/status")
+            with urllib.request.urlopen(req, timeout=5) as resp:
+                data = json.loads(resp.read().decode())
+                if data.get("running"):
+                    print("✅ Appium Server 运行中")
+                else:
+                    print("❌ Appium Server 未运行")
+                    print("   请执行: npm install -g appium && appium")
+                return 0
+        except urllib.error.URLError:
+            print(f"❌ 无法连接引擎，请先运行: python cli.py serve")
+            return 2
+
+    elif sub == "connect":
+        payload = {
+            "device_name": args.device,
+            "app_package": args.package,
+            "app_activity": args.activity,
+            "app_path": args.apk,
+        }
+        try:
+            req_data = json.dumps(payload).encode()
+            req = urllib.request.Request(
+                f"{base}/mobile/session/create",
+                data=req_data,
+                headers={"Content-Type": "application/json"},
+                method="POST",
+            )
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                data = json.loads(resp.read().decode())
+                print(f"✅ {data.get('message', '连接成功')}")
+                print(f"   会话ID: {data.get('session_id')}")
+                device = data.get("device", {})
+                print(f"   设备: {device.get('name', '未知')}")
+                return 0
+        except urllib.error.HTTPError as e:
+            body = e.read().decode() if e.fp else ""
+            print(f"❌ 连接失败: {body}")
+            return 1
+        except urllib.error.URLError:
+            print(f"❌ 无法连接引擎，请先运行: python cli.py serve")
+            return 2
+
+    elif sub == "screenshot":
+        session_id = args.session
+        try:
+            req = urllib.request.Request(f"{base}/mobile/session/{session_id}/screenshot")
+            with urllib.request.urlopen(req, timeout=15) as resp:
+                data = json.loads(resp.read().decode())
+                print(f"📸 截图已保存: {data.get('path', '未知')}")
+                return 0
+        except urllib.error.HTTPError as e:
+            print(f"❌ 截图失败: {e.read().decode() if e.fp else e}")
+            return 1
+        except urllib.error.URLError:
+            print(f"❌ 无法连接引擎")
+            return 2
+
+    elif sub == "sessions":
+        try:
+            req = urllib.request.Request(f"{base}/mobile/sessions")
+            with urllib.request.urlopen(req, timeout=5) as resp:
+                data = json.loads(resp.read().decode())
+                sessions = data.get("sessions", [])
+                if not sessions:
+                    print("📱 没有活跃的手机测试会话")
+                else:
+                    print(f"📱 {len(sessions)} 个活跃会话:")
+                    for s in sessions:
+                        device = s.get("device", {})
+                        print(f"   • {s['session_id']} → {device.get('name', '未知')}")
+                return 0
+        except urllib.error.URLError:
+            print(f"❌ 无法连接引擎")
+            return 2
+
+    return 0
+
+
 def _print_report(report: dict, duration: float) -> None:
     """在终端打印测试报告。"""
     pass_rate = report.get("pass_rate", 0)
@@ -430,6 +548,24 @@ def main() -> int:
     # health
     subparsers.add_parser("health", help="检查引擎健康状态")
 
+    # mobile (手机测试 v5.0)
+    p_mobile = subparsers.add_parser("mobile", help="手机测试相关命令")
+    mobile_sub = p_mobile.add_subparsers(dest="mobile_command", help="手机测试子命令")
+
+    mobile_sub.add_parser("devices", help="列出已连接的手机设备")
+    mobile_sub.add_parser("appium", help="检查 Appium Server 状态")
+
+    p_mobile_connect = mobile_sub.add_parser("connect", help="连接手机设备并创建测试会话")
+    p_mobile_connect.add_argument("--device", default="", help="设备名称")
+    p_mobile_connect.add_argument("--package", default="", help="Android 包名")
+    p_mobile_connect.add_argument("--activity", default="", help="启动 Activity")
+    p_mobile_connect.add_argument("--apk", default="", help="APK 文件路径")
+
+    p_mobile_screenshot = mobile_sub.add_parser("screenshot", help="截取手机屏幕")
+    p_mobile_screenshot.add_argument("--session", required=True, help="会话ID")
+
+    mobile_sub.add_parser("sessions", help="列出活跃的手机测试会话")
+
     args = parser.parse_args()
 
     if not args.command:
@@ -442,6 +578,7 @@ def main() -> int:
         "explore": cmd_explore,
         "init": cmd_init,
         "health": cmd_health,
+        "mobile": cmd_mobile,
     }
 
     handler = commands.get(args.command)

src/api/routes.py

@@ -7,6 +7,8 @@
 - /browser/*: 浏览器操作（启动/导航/点击/输入/截图/关闭）
 """
 
+import json
+
 from fastapi import APIRouter, HTTPException
 from loguru import logger
 
@@ -607,4 +609,197 @@ async def send_webhook(req: dict) -> dict:
         success = handler(webhook_url, report)
         return {"success": success, "type": notify_type}
 
+    # ── 手机测试（v5.0）─────────────────────────────
+
+    # 设备会话管理（内存中缓存）
+    _mobile_sessions: dict[str, "AndroidController"] = {}
+
+    @router.get("/mobile/devices", tags=["手机测试"])
+    async def list_mobile_devices() -> dict:
+        """列出已连接的 Android 设备（通过 adb devices）。"""
+        import subprocess
+        try:
+            result = subprocess.run(
+                ["adb", "devices", "-l"],
+                capture_output=True, text=True, timeout=5,
+            )
+            lines = result.stdout.strip().split("\n")[1:]  # 跳过header
+            devices = []
+            for line in lines:
+                parts = line.split()
+                if len(parts) >= 2 and parts[1] == "device":
+                    info = {"serial": parts[0], "status": "connected"}
+                    # 解析 model/device 等额外信息
+                    for part in parts[2:]:
+                        if ":" in part:
+                            k, v = part.split(":", 1)
+                            info[k] = v
+                    devices.append(info)
+            return {"devices": devices, "count": len(devices)}
+        except FileNotFoundError:
+            return {"devices": [], "count": 0, "error": "adb 未安装或不在PATH中"}
+        except Exception as e:
+            return {"devices": [], "count": 0, "error": str(e)}
+
+    @router.get("/mobile/appium/status", tags=["手机测试"])
+    async def appium_status() -> dict:
+        """检查 Appium Server 是否运行。"""
+        import urllib.request
+        import urllib.error
+
+        appium_url = "http://127.0.0.1:4723/status"
+        try:
+            req = urllib.request.Request(appium_url)
+            with urllib.request.urlopen(req, timeout=3) as resp:
+                data = json.loads(resp.read().decode())
+                return {"running": True, "data": data}
+        except (urllib.error.URLError, Exception):
+            return {"running": False, "message": "Appium Server 未运行。请执行: appium"}
+
+    @router.post("/mobile/session/create", tags=["手机测试"])
+    async def create_mobile_session(req: dict) -> dict:
+        """创建手机测试会话（连接设备）。
+
+        Body:
+            device_name: 设备名称（可选）
+            app_package: 应用包名（可选）
+            app_activity: 启动Activity（可选）
+            app_path: APK路径（可选，自动安装）
+        """
+        from src.controller.android import AndroidController, MobileConfig
+
+        config = MobileConfig(
+            device_name=req.get("device_name", ""),
+            app_package=req.get("app_package", ""),
+            app_activity=req.get("app_activity", ""),
+            app_path=req.get("app_path", ""),
+        )
+
+        controller = AndroidController(config)
+        try:
+            await controller.launch()
+            session_id = f"mobile_{len(_mobile_sessions) + 1}"
+            _mobile_sessions[session_id] = controller
+            return {
+                "session_id": session_id,
+                "device": controller.device_info.model_dump(),
+                "message": "手机会话创建成功",
+            }
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=f"连接设备失败: {e}")
+
+    @router.post("/mobile/session/{session_id}/tap", tags=["手机测试"])
+    async def mobile_tap(session_id: str, req: dict) -> dict:
+        """点击手机元素。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            await ctrl.tap(req.get("selector", ""))
+            return {"success": True}
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=str(e))
+
+    @router.post("/mobile/session/{session_id}/input", tags=["手机测试"])
+    async def mobile_input(session_id: str, req: dict) -> dict:
+        """在手机输入框中输入文本。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            await ctrl.input_text(req.get("selector", ""), req.get("text", ""))
+            return {"success": True}
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=str(e))
+
+    @router.post("/mobile/session/{session_id}/swipe", tags=["手机测试"])
+    async def mobile_swipe(session_id: str, req: dict) -> dict:
+        """手机滑动操作。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            await ctrl.swipe(
+                req.get("start_x", 0), req.get("start_y", 0),
+                req.get("end_x", 0), req.get("end_y", 0),
+                req.get("duration_ms", 300),
+            )
+            return {"success": True}
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=str(e))
+
+    @router.get("/mobile/session/{session_id}/screenshot", tags=["手机测试"])
+    async def mobile_screenshot(session_id: str, name: str = "") -> dict:
+        """截取手机屏幕。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            path = await ctrl.screenshot(name or "mobile_capture")
+            # 返回base64用于前端显示
+            import base64
+            b64 = base64.b64encode(path.read_bytes()).decode()
+            return {"path": str(path), "base64": b64}
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
+
+    @router.get("/mobile/session/{session_id}/source", tags=["手机测试"])
+    async def mobile_page_source(session_id: str) -> dict:
+        """获取手机UI层级XML。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            source = await ctrl.get_page_source()
+            return {"source": source}
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
+
+    @router.post("/mobile/session/{session_id}/navigate", tags=["手机测试"])
+    async def mobile_navigate(session_id: str, req: dict) -> dict:
+        """打开URL或Activity。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            await ctrl.navigate(req.get("target", ""))
+            return {"success": True}
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=str(e))
+
+    @router.post("/mobile/session/{session_id}/back", tags=["手机测试"])
+    async def mobile_back(session_id: str) -> dict:
+        """按返回键。"""
+        ctrl = _mobile_sessions.get(session_id)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            await ctrl.back()
+            return {"success": True}
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=str(e))
+
+    @router.delete("/mobile/session/{session_id}", tags=["手机测试"])
+    async def close_mobile_session(session_id: str) -> dict:
+        """关闭手机测试会话。"""
+        ctrl = _mobile_sessions.pop(session_id, None)
+        if not ctrl:
+            raise HTTPException(status_code=404, detail="会话不存在")
+        try:
+            await ctrl.close()
+            return {"message": "会话已关闭"}
+        except Exception as e:
+            return {"message": f"关闭时出错: {e}"}
+
+    @router.get("/mobile/sessions", tags=["手机测试"])
+    async def list_mobile_sessions() -> dict:
+        """列出所有活跃的手机测试会话。"""
+        sessions = []
+        for sid, ctrl in _mobile_sessions.items():
+            sessions.append({
+                "session_id": sid,
+                "device": ctrl.device_info.model_dump(),
+            })
+        return {"sessions": sessions, "count": len(sessions)}
+
     return router