From 0f9b6fcc5d3d60b8119b8eaec7baf3c86ec35e01 Mon Sep 17 00:00:00 2001 From: shuiyu1123 Date: Fri, 26 Jun 2026 21:25:12 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=20Windows=20?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E5=AE=9E=E7=8E=B0=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 _windows_impl.py，包含 14 个 Windows 实现类 - 支持网络、蓝牙、传感器、音频、剪贴板、电源、位置、通知、显示、UI自动化、摄像头、进程管理、Shell 等功能 - 更新 README.md 添加 Windows 支持说明 - 更新 setup.py 添加 Windows 专用依赖 --- README.md | 34 +- android_pyapi/_windows_impl.py | 1380 ++++++++++++++++++++++++++++++++ setup.py | 15 + 3 files changed, 1424 insertions(+), 5 deletions(-) create mode 100644 android_pyapi/_windows_impl.py diff --git a/README.md b/README.md index 60a9bfa..3282744 100644 --- a/README.md +++ b/README.md @@ -47,10 +47,31 @@ Android-pyApi 将完整的 Android SDK API 翻译为纯 Python 实现，让你 | Camera | OpenCV / V4L2 | 真实拍照 | | UI 自动化 | `xdotool` | 真实点击、输入、按键 | +### Windows 真实实现 + +在 Windows 环境下，以下 API 使用真实系统调用而非存根： + +| API | 实现方式 | 真实行为 | +|-----|---------|---------| +| ConnectivityManager | `ipconfig` + `netsh` + `socket` | 真实 IP、接口、DNS、网关、WiFi 信息 | +| LocationManager | IP 地理定位 | 真实经纬度、城市、ISP | +| SensorManager | `wmi` + `psutil` | 真实温度、CPU 频率、风扇转速 | +| AudioManager | `pycaw` / PowerShell | 真实音量控制、静音 | +| BluetoothAdapter | PowerShell Bluetooth cmdlets | 真实设备列表、配对状态 | +| ClipboardManager | `pyperclip` | 真实剪贴板读写 | +| PowerManager / BatteryManager | `psutil` | 真实 CPU、内存、电池状态 | +| NotificationManager | `win10toast` / PowerShell | 真实桌面通知 (Toast) | +| Activity / Display | `pyautogui` + `screeninfo` + `win32api` | 真实屏幕尺寸、截屏 | +| ActivityManager | `psutil` + `tasklist` | 真实进程列表 | +| Shell | `subprocess` + PowerShell | 真实命令执行 | +| Camera | OpenCV | 真实拍照 | +| UI 自动化 | `pyautogui` + `win32gui` | 真实点击、输入、按键、窗口操作 | + ### 跨平台 -- **Linux** — 系统级 API 使用真实实现 -- **macOS / Windows** — 系统级 API 使用存根（后续支持） +- **Linux** — 系统级 API 使用真实实现 (`_linux_impl.py`) +- **Windows** — 系统级 API 使用真实实现 (`_windows_impl.py`) +- **macOS** — 系统级 API 使用存根（后续支持） - **API 形状** — 所有平台一致，方法签名完全对齐 Android SDK ## 安装 @@ -68,11 +89,14 @@ export PYTHONPATH=/path/to/Android-pyApi:$PYTHONPATH ### 可选依赖 ```bash +# Python 库 (所有平台通用) +pip install psutil requests Pillow opencv-python numpy + # Linux 系统工具 sudo apt install xclip xdotool libnotify-bin alsa-utils bluez lm-sensors -# Python 库 -pip install psutil requests Pillow opencv-python numpy +# Windows 专用库 (推荐安装以启用完整功能) +pip install pyautogui pyperclip screeninfo pywin32 wmi pycaw comtypes win10toast ``` ## 快速开始 @@ -116,6 +140,7 @@ print(loc) # 真实经纬度 android_pyapi/ ├── _core.py # Stub 基类、全局上下文 ├── _linux_impl.py # Linux 真实实现层 (14个实现类) +├── _windows_impl.py # Windows 真实实现层 (14个实现类) ├── __init__.py # 统一导出 ├── android/ # 167 个 Android 模块 │ ├── app.py # Activity, Service, Fragment, Notification @@ -173,7 +198,6 @@ Android-pyApi 不是 Android 模拟器。它不包含 ART 虚拟机、SurfaceFli 欢迎贡献！特别需要： - macOS 真实实现层 (`_macos_impl.py`) -- Windows 真实实现层 (`_windows_impl.py`) - 更多 API 的真实行为实现 - 测试用例 diff --git a/android_pyapi/_windows_impl.py b/android_pyapi/_windows_impl.py new file mode 100644 index 0000000..83e173f --- /dev/null +++ b/android_pyapi/_windows_impl.py @@ -0,0 +1,1380 @@ +""" +Windows 真实实现层 +================== +在 Windows 环境下，将 Android API 的存根行为替换为真实的系统调用。 +使 Android-pyApi 从"存根库"变为"可用的生产力工具"。 + +覆盖能力: + - 网络: ipconfig, netsh, socket, wmi + - 蓝牙: PowerShell Bluetooth cmdlets + - 传感器: wmi, psutil + - 音频: pycaw / PowerShell + - 剪贴板: pyperclip / win32clipboard + - 电源/电池: psutil, wmi + - 位置: IP 地理定位 (requests) + - 通知: win10toast / plyer + - 显示: pyautogui, screeninfo, win32api + - UI自动化: pyautogui + - 摄像头: opencv + - 进程管理: psutil, subprocess + - 文件: 原生 Python os + - 媒体: ffmpeg/ffprobe +""" + +from __future__ import annotations + +import os +import sys +import re +import json +import time +import socket +import shutil +import subprocess +import platform +import hashlib +from pathlib import Path +from datetime import datetime +from typing import Any, Dict, List, Optional, Tuple + +# ═══════════════════════════════════════════════════════════════ +# 能力检测 +# ═══════════════════════════════════════════════════════════════ + +IS_WINDOWS = platform.system() == "Windows" + +def _which(name: str) -> Optional[str]: + return shutil.which(name) + +HAS_POWERSHELL = _which("powershell") is not None or _which("pwsh") is not None +HAS_FFMPEG = _which("ffmpeg") is not None +HAS_FFPROBE = _which("ffprobe") is not None + +try: + import psutil + HAS_PSUTIL = True +except ImportError: + HAS_PSUTIL = False + +try: + import requests + HAS_REQUESTS = True +except ImportError: + HAS_REQUESTS = False + +try: + import cv2 + HAS_CV2 = True +except ImportError: + HAS_CV2 = False + +try: + import numpy as np + HAS_NUMPY = True +except ImportError: + HAS_NUMPY = False + +try: + from PIL import Image, ImageDraw, ImageGrab + HAS_PIL = True +except ImportError: + HAS_PIL = False + +try: + import pyautogui + HAS_PYAUTOGUI = True +except ImportError: + HAS_PYAUTOGUI = False + +try: + import pyperclip + HAS_PYPERCLIP = True +except ImportError: + HAS_PYPERCLIP = False + +try: + from win10toast import ToastNotifier + HAS_WIN10TOAST = True +except ImportError: + HAS_WIN10TOAST = False + +try: + import wmi + HAS_WMI = True +except ImportError: + HAS_WMI = False + +try: + import win32api + import win32con + import win32gui + HAS_WIN32 = True +except ImportError: + HAS_WIN32 = False + +try: + from pycaw.pycaw import AudioUtilities, IAudioEndpointVolume + from comtypes import CLSCTX_ALL + HAS_PYCAW = True +except ImportError: + HAS_PYCAW = False + +try: + from screeninfo import get_monitors + HAS_SCREENINFO = True +except ImportError: + HAS_SCREENINFO = False + + +def _run(cmd: List[str] or str, timeout: int = 10, capture: bool = True) -> Tuple[int, str, str]: + """执行命令，返回 (exit_code, stdout, stderr)""" + try: + if isinstance(cmd, str): + result = subprocess.run(cmd, shell=True, capture_output=capture, + text=True, timeout=timeout, encoding='utf-8', errors='replace') + else: + result = subprocess.run(cmd, capture_output=capture, + text=True, timeout=timeout, encoding='utf-8', errors='replace') + return result.returncode, result.stdout.strip(), result.stderr.strip() + except subprocess.TimeoutExpired: + return -1, "", "timeout" + except Exception as e: + return -1, "", str(e) + + +def _run_ps(script: str, timeout: int = 10) -> Tuple[int, str, str]: + """执行 PowerShell 脚本""" + ps_exe = "pwsh" if _which("pwsh") else "powershell" + cmd = [ps_exe, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script] + return _run(cmd, timeout=timeout) + + +def _read_file(path: str) -> Optional[str]: + try: + with open(path, "r", encoding='utf-8', errors='replace') as f: + return f.read().strip() + except Exception: + return None + + +def _read_file_int(path: str) -> Optional[int]: + val = _read_file(path) + if val is not None: + try: + return int(val) + except ValueError: + return None + return None + + +# ═══════════════════════════════════════════════════════════════ +# 网络 — 真实实现 +# ═══════════════════════════════════════════════════════════════ + +class WindowsNetwork: + """真实网络信息 — 通过 ipconfig, netsh, socket""" + + @staticmethod + def get_interfaces() -> List[dict]: + """获取网络接口列表""" + interfaces = [] + # 使用 PowerShell 获取网络适配器信息 + code, out, _ = _run_ps( + "Get-NetAdapter | Select-Object Name, Status, MacAddress, InterfaceDescription | ConvertTo-Json" + ) + if code == 0 and out: + try: + data = json.loads(out) + if isinstance(data, dict): + data = [data] + for iface in data: + interfaces.append({ + "ifname": iface.get("Name", ""), + "operstate": iface.get("Status", ""), + "address": iface.get("MacAddress", ""), + "description": iface.get("InterfaceDescription", ""), + }) + return interfaces + except json.JSONDecodeError: + pass + # 回退到 ipconfig + code, out, _ = _run("ipconfig /all") + if code == 0: + current = {} + for line in out.split("\n"): + line = line.strip() + if "adapter" in line.lower() and ":" in line: + if current: + interfaces.append(current) + current = {"ifname": line.split(":")[0].strip(), "operstate": "up", "address": ""} + elif "Physical Address" in line or "物理地址" in line: + current["address"] = line.split(":")[-1].strip() + if current: + interfaces.append(current) + return interfaces + + @staticmethod + def get_ip_address(interface: str = None) -> str: + """获取IP地址""" + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("8.8.8.8", 80)) + ip = s.getsockname()[0] + s.close() + return ip + except Exception: + pass + try: + return socket.gethostbyname(socket.gethostname()) + except Exception: + return "127.0.0.1" + + @staticmethod + def get_wifi_info() -> dict: + """获取WiFi信息""" + code, out, _ = _run("netsh wlan show interfaces") + if code == 0: + info = {"ssid": "", "signal": 0, "status": "disconnected"} + for line in out.split("\n"): + line = line.strip() + if "SSID" in line and "BSSID" not in line: + parts = line.split(":", 1) + if len(parts) >= 2: + info["ssid"] = parts[1].strip() + elif "Signal" in line: + parts = line.split(":", 1) + if len(parts) >= 2: + signal_str = parts[1].strip().rstrip("%") + try: + info["signal"] = int(signal_str) + except ValueError: + pass + elif "State" in line: + parts = line.split(":", 1) + if len(parts) >= 2: + info["status"] = parts[1].strip().lower() + return info + return {"ssid": "", "signal": 0, "status": "unknown"} + + @staticmethod + def get_connectivity() -> bool: + """检测网络连通性""" + for host, port in [("8.8.8.8", 53), ("1.1.1.1", 53), ("114.114.114.114", 53)]: + try: + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.settimeout(3) + s.connect((host, port)) + s.close() + return True + except Exception: + continue + if HAS_REQUESTS: + try: + requests.get("http://www.baidu.com", timeout=3) + return True + except Exception: + pass + return False + + @staticmethod + def get_dns_servers() -> List[str]: + """获取DNS服务器""" + code, out, _ = _run_ps( + "Get-DnsClientServerAddress | Select-Object -ExpandProperty ServerAddresses" + ) + if code == 0 and out: + return [s.strip() for s in out.split("\n") if s.strip()] + # 回退到 ipconfig + code, out, _ = _run("ipconfig /all") + if code == 0: + dns = [] + for line in out.split("\n"): + if "DNS" in line and ":" in line: + parts = line.split(":", 1) + if len(parts) >= 2: + addr = parts[1].strip() + if addr and not addr.startswith("("): + dns.append(addr) + return dns + return [] + + @staticmethod + def get_gateway() -> str: + """获取默认网关""" + code, out, _ = _run("ipconfig") + if code == 0: + for line in out.split("\n"): + if ("Default Gateway" in line or "默认网关" in line) and ":" in line: + parts = line.split(":", 1) + if len(parts) >= 2: + gw = parts[1].strip() + if gw and gw != "0.0.0.0": + return gw + return "" + + @staticmethod + def get_mac_address(interface: str = None) -> str: + """获取MAC地址""" + code, out, _ = _run("getmac /fo csv /v") + if code == 0: + lines = out.split("\n") + for line in lines[1:]: # 跳过标题行 + parts = line.split(",") + if len(parts) >= 3: + mac = parts[2].strip('"').replace("-", ":") + if mac and mac != "N/A": + return mac + return "" + + @staticmethod + def get_rx_tx_bytes(interface: str = None) -> dict: + """获取网络收发字节数""" + if HAS_PSUTIL: + counters = psutil.net_io_counters(pernic=True) + if interface and interface in counters: + c = counters[interface] + return {"rx_bytes": c.bytes_recv, "tx_bytes": c.bytes_sent, "interface": interface} + elif counters: + # 取第一个非回环接口 + for name, c in counters.items(): + if name.lower() != "lo": + return {"rx_bytes": c.bytes_recv, "tx_bytes": c.bytes_sent, "interface": name} + return {"rx_bytes": 0, "tx_bytes": 0, "interface": ""} + + +# ═══════════════════════════════════════════════════════════════ +# 蓝牙 — 真实实现 (PowerShell) +# ═══════════════════════════════════════════════════════════════ + +class WindowsBluetooth: + """真实蓝牙操作 — 通过 PowerShell Bluetooth cmdlets""" + + STATE_OFF = 10 + STATE_TURNING_ON = 11 + STATE_ON = 12 + STATE_TURNING_OFF = 13 + + @staticmethod + def get_state() -> int: + code, out, _ = _run_ps( + "Get-PnpDevice -Class Bluetooth | Where-Object {$_.FriendlyName -like '*Radio*' -or $_.FriendlyName -like '*Adapter*'} | Select-Object Status" + ) + if code == 0 and "OK" in out: + return WindowsBluetooth.STATE_ON + return WindowsBluetooth.STATE_OFF + + @staticmethod + def get_address() -> Optional[str]: + code, out, _ = _run_ps( + "Get-PnpDevice -Class Bluetooth -Status OK | Select-Object InstanceId | ConvertTo-Json" + ) + if code == 0 and out: + try: + data = json.loads(out) + if isinstance(data, dict): + data = [data] + for dev in data: + instance_id = dev.get("InstanceId", "") + # 从 InstanceId 提取 MAC 地址 + mac_match = re.search(r'([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}', instance_id) + if mac_match: + return mac_match.group().replace("-", ":") + except json.JSONDecodeError: + pass + return None + + @staticmethod + def get_name() -> Optional[str]: + code, out, _ = _run_ps( + "Get-PnpDevice -Class Bluetooth -Status OK | Where-Object {$_.FriendlyName -like '*Radio*' -or $_.FriendlyName -like '*Adapter*'} | Select-Object FriendlyName | ConvertTo-Json" + ) + if code == 0 and out: + try: + data = json.loads(out) + if isinstance(data, dict): + return data.get("FriendlyName") + elif isinstance(data, list) and data: + return data[0].get("FriendlyName") + except json.JSONDecodeError: + pass + return None + + @staticmethod + def enable() -> bool: + code, _, err = _run_ps( + "Get-PnpDevice -Class Bluetooth | Enable-PnpDevice -Confirm:$false" + ) + return code == 0 + + @staticmethod + def disable() -> bool: + code, _, err = _run_ps( + "Get-PnpDevice -Class Bluetooth | Disable-PnpDevice -Confirm:$false" + ) + return code == 0 + + @staticmethod + def get_bonded_devices() -> List[dict]: + code, out, _ = _run_ps( + "Get-PnpDevice -Class Bluetooth -Status OK | Select-Object FriendlyName, InstanceId | ConvertTo-Json" + ) + devices = [] + if code == 0 and out: + try: + data = json.loads(out) + if isinstance(data, dict): + data = [data] + for dev in data: + name = dev.get("FriendlyName", "") + if name and "Enumerator" not in name: + devices.append({ + "address": "", + "name": name, + "bonded": True, + }) + except json.JSONDecodeError: + pass + return devices + + @staticmethod + def scan(timeout: int = 10) -> List[dict]: + """扫描蓝牙设备 (Windows 限制，只能列出已配对设备)""" + return WindowsBluetooth.get_bonded_devices() + + @staticmethod + def pair(address: str) -> bool: + # Windows 需要通过设置界面配对 + code, _, _ = _run_ps("Start-Process ms-settings:bluetooth") + return code == 0 + + @staticmethod + def connect(address: str) -> bool: + return False # Windows 无命令行直接连接方式 + + @staticmethod + def disconnect(address: str) -> bool: + return False + + @staticmethod + def unpair(address: str) -> bool: + return False + + +# ═══════════════════════════════════════════════════════════════ +# 传感器 — 真实实现 (wmi, psutil) +# ═══════════════════════════════════════════════════════════════ + +class WindowsSensors: + """真实传感器数据 — 通过 WMI 和 psutil""" + + TYPE_ACCELEROMETER = 1 + TYPE_GYROSCOPE = 4 + TYPE_LIGHT = 5 + TYPE_PRESSURE = 6 + TYPE_PROXIMITY = 8 + TYPE_TEMPERATURE = 7 + TYPE_RELATIVE_HUMIDITY = 12 + TYPE_AMBIENT_TEMPERATURE = 13 + + @staticmethod + def get_temperatures() -> List[dict]: + """获取温度传感器读数""" + temps = [] + # 通过 WMI 获取 CPU 温度 + if HAS_WMI: + try: + w = wmi.WMI(namespace="root\\wmi") + for sensor in w.MSAcpi_ThermalZoneTemperature(): + temp_celsius = (sensor.CurrentTemperature / 10.0) - 273.15 + temps.append({ + "name": sensor.InstanceName, + "type": "thermal_zone", + "value": temp_celsius, + "unit": "°C", + }) + except Exception: + pass + # 通过 psutil 获取 (如果支持) + if HAS_PSUTIL: + try: + if hasattr(psutil, "sensors_temperatures"): + for name, entries in psutil.sensors_temperatures().items(): + for entry in entries: + temps.append({ + "name": f"{name}/{entry.label or 'unknown'}", + "type": name, + "value": entry.current, + "unit": "°C", + }) + except Exception: + pass + return temps + + @staticmethod + def get_fan_speeds() -> List[dict]: + """获取风扇转速""" + fans = [] + if HAS_PSUTIL: + try: + if hasattr(psutil, "sensors_fans"): + for name, entries in psutil.sensors_fans().items(): + for entry in entries: + fans.append({ + "name": f"{name}/{entry.label or 'unknown'}", + "rpm": entry.current, + }) + except Exception: + pass + return fans + + @staticmethod + def get_cpu_frequency() -> List[dict]: + """获取CPU频率""" + freqs = [] + if HAS_PSUTIL: + try: + cpu_freq = psutil.cpu_freq(percpu=True) + if cpu_freq: + for i, freq in enumerate(cpu_freq): + freqs.append({ + "cpu": i, + "freq_mhz": freq.current, + }) + else: + # 单个频率值 + freq = psutil.cpu_freq() + if freq: + freqs.append({ + "cpu": 0, + "freq_mhz": freq.current, + }) + except Exception: + pass + return freqs + + @staticmethod + def get_light_level() -> Optional[float]: + """获取环境光传感器 (Windows 通常无直接访问)""" + return None + + @staticmethod + def get_all_sensors() -> dict: + """获取所有传感器数据""" + return { + "temperatures": WindowsSensors.get_temperatures(), + "fans": WindowsSensors.get_fan_speeds(), + "cpu_freq": WindowsSensors.get_cpu_frequency(), + "light": WindowsSensors.get_light_level(), + } + + +# ═══════════════════════════════════════════════════════════════ +# 音频 — 真实实现 (pycaw / PowerShell) +# ═══════════════════════════════════════════════════════════════ + +class WindowsAudio: + """真实音频控制 — 通过 pycaw 或 PowerShell""" + + @staticmethod + def _get_pycaw_volume(): + """获取 pycaw 音量接口""" + if not HAS_PYCAW: + return None + try: + devices = AudioUtilities.GetSpeakers() + interface = devices.Activate(IAudioEndpointVolume._iid_, CLSCTX_ALL, None) + return interface.QueryInterface(IAudioEndpointVolume) + except Exception: + return None + + @staticmethod + def get_volume() -> int: + """获取当前音量 (0-100)""" + # 方法1: pycaw + vol = WindowsAudio._get_pycaw_volume() + if vol: + try: + return int(vol.GetMasterVolumeLevelScalar() * 100) + except Exception: + pass + # 方法2: PowerShell + code, out, _ = _run_ps( + "(Get-AudioDevice -PlaybackVolume) 2>$null; if($?) {$out} else {(Get-WmiObject -Class Win32_SoundDevice | Select-Object -First 1).Volume}" + ) + if code == 0 and out: + try: + return int(out.strip()) + except ValueError: + pass + return 50 # 默认值 + + @staticmethod + def set_volume(volume: int) -> bool: + """设置音量 (0-100)""" + volume = max(0, min(100, volume)) + # 方法1: pycaw + vol = WindowsAudio._get_pycaw_volume() + if vol: + try: + vol.SetMasterVolumeLevelScalar(volume / 100.0, None) + return True + except Exception: + pass + # 方法2: PowerShell + code, _, _ = _run_ps(f"(Get-AudioDevice -PlaybackVolume {volume}) 2>$null") + return code == 0 + + @staticmethod + def get_mute() -> bool: + """获取静音状态""" + vol = WindowsAudio._get_pycaw_volume() + if vol: + try: + return vol.GetMute() + except Exception: + pass + return False + + @staticmethod + def set_mute(mute: bool) -> bool: + """设置静音""" + vol = WindowsAudio._get_pycaw_volume() + if vol: + try: + vol.SetMute(mute, None) + return True + except Exception: + pass + return False + + @staticmethod + def get_audio_devices() -> List[dict]: + """获取音频设备列表""" + devices = [] + code, out, _ = _run_ps( + "Get-AudioDevice -List 2>$null; if(-not $?) { Get-WmiObject Win32_SoundDevice | Select-Object Name, Status }" + ) + if code == 0 and out: + for line in out.split("\n"): + if line.strip(): + devices.append({"name": line.strip(), "type": "windows"}) + return devices + + +# ═══════════════════════════════════════════════════════════════ +# 剪贴板 — 真实实现 (pyperclip) +# ═══════════════════════════════════════════════════════════════ + +class WindowsClipboard: + """真实剪贴板 — 通过 pyperclip 或 tkinter""" + + @staticmethod + def read() -> str: + if HAS_PYPERCLIP: + try: + return pyperclip.paste() + except Exception: + pass + # 回退到 tkinter + try: + import tkinter + root = tkinter.Tk() + root.withdraw() + text = root.clipboard_get() + root.destroy() + return text + except Exception: + return "" + + @staticmethod + def write(text: str) -> bool: + if HAS_PYPERCLIP: + try: + pyperclip.copy(text) + return True + except Exception: + pass + # 回退到 tkinter + try: + import tkinter + root = tkinter.Tk() + root.withdraw() + root.clipboard_clear() + root.clipboard_append(text) + root.update() + root.destroy() + return True + except Exception: + return False + + +# ═══════════════════════════════════════════════════════════════ +# 电源/电池 — 真实实现 (psutil) +# ═══════════════════════════════════════════════════════════════ + +class WindowsPower: + """真实电源/电池信息""" + + BATTERY_STATUS_UNKNOWN = 1 + BATTERY_STATUS_CHARGING = 2 + BATTERY_STATUS_DISCHARGING = 3 + BATTERY_STATUS_NOT_CHARGING = 4 + BATTERY_STATUS_FULL = 5 + + @staticmethod + def get_battery_info() -> dict: + """获取电池信息""" + info = {"present": False, "level": -1, "status": "unknown", + "temperature": 0, "voltage": 0, "technology": ""} + if HAS_PSUTIL: + try: + battery = psutil.sensors_battery() + if battery: + info["present"] = True + info["level"] = battery.percent + if battery.power_plugged: + info["status"] = "charging" if battery.percent < 100 else "full" + else: + info["status"] = "discharging" + if battery.secsleft and battery.secsleft > 0: + info["seconds_left"] = battery.secsleft + except Exception: + pass + return info + + @staticmethod + def get_power_status() -> dict: + """获取电源状态""" + info = WindowsPower.get_battery_info() + return { + "battery_present": info["present"], + "battery_level": info["level"], + "battery_status": info["status"], + "is_charging": "charging" in info["status"], + "temperature": info["temperature"], + "voltage": info["voltage"], + } + + @staticmethod + def get_cpu_usage() -> float: + """获取CPU使用率""" + if HAS_PSUTIL: + return psutil.cpu_percent(interval=0.5) + return 0.0 + + @staticmethod + def get_memory_info() -> dict: + """获取内存信息""" + if HAS_PSUTIL: + mem = psutil.virtual_memory() + return { + "total": mem.total, + "available": mem.available, + "used": mem.used, + "free": mem.free, + "percent": mem.percent, + } + return {"total": 0, "available": 0, "used": 0, "free": 0, "percent": 0} + + +# ═══════════════════════════════════════════════════════════════ +# 位置 — 真实实现 (IP 地理定位) +# ═══════════════════════════════════════════════════════════════ + +class WindowsLocation: + """真实位置信息 — 通过 IP 地理定位""" + + _cached_location = None + _cache_time = 0 + + @staticmethod + def get_location() -> dict: + """通过 IP 获取地理位置""" + if not HAS_REQUESTS: + return {"latitude": 0.0, "longitude": 0.0, "accuracy": 0, "source": "none"} + + now = time.time() + if WindowsLocation._cached_location and now - WindowsLocation._cache_time < 300: + return WindowsLocation._cached_location + + services = [ + ("http://ip-api.com/json/", "ip-api"), + ("https://ipapi.co/json/", "ipapi"), + ("https://freeipapi.com/api/json", "freeipapi"), + ] + + for url, source in services: + try: + resp = requests.get(url, timeout=5) + if resp.status_code == 200: + data = resp.json() + lat = data.get("lat") or data.get("latitude") or 0.0 + lon = data.get("lon") or data.get("longitude") or 0.0 + location = { + "latitude": float(lat), + "longitude": float(lon), + "accuracy": 50000, + "source": source, + "city": data.get("city", ""), + "region": data.get("regionName", data.get("region", "")), + "country": data.get("country", data.get("country_name", "")), + "country_code": data.get("countryCode", data.get("country", "")), + "timezone": data.get("timezone", ""), + "isp": data.get("isp", ""), + } + WindowsLocation._cached_location = location + WindowsLocation._cache_time = now + return location + except Exception: + continue + + return {"latitude": 0.0, "longitude": 0.0, "accuracy": 0, "source": "failed"} + + +# ═══════════════════════════════════════════════════════════════ +# 通知 — 真实实现 (win10toast / PowerShell) +# ═══════════════════════════════════════════════════════════════ + +class WindowsNotify: + """真实桌面通知 — 通过 win10toast 或 PowerShell""" + + @staticmethod + def send(title: str, text: str, icon: str = None, urgency: str = "normal", + timeout: int = 5) -> bool: + # 方法1: win10toast + if HAS_WIN10TOAST: + try: + toaster = ToastNotifier() + toaster.show_toast(title, text, duration=timeout, threaded=True) + return True + except Exception: + pass + # 方法2: PowerShell BurntToast 模块 + script = f''' + try {{ + Import-Module BurntToast -ErrorAction Stop + New-BurntToastNotification -Text "{title}", "{text}" + }} catch {{ + [System.Reflection.Assembly]::LoadWithPartialName("System.Windows.Forms") + [System.Windows.Forms.MessageBox]::Show("{text}", "{title}") + }} + ''' + code, _, _ = _run_ps(script) + if code == 0: + return True + # 方法3: PowerShell 原生方式 + script = f''' + [Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType = WindowsRuntime] | Out-Null + [Windows.Data.Xml.Dom.XmlDocument, Windows.Data.Xml.Dom.XmlDocument, ContentType = WindowsRuntime] | Out-Null + $template = @" + + + + {title} + {text} + + + +"@ + $xml = New-Object Windows.Data.Xml.Dom.XmlDocument + $xml.LoadXml($template) + $toast = [Windows.UI.Notifications.ToastNotification]::new($xml) + [Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier("Android-pyApi").Show($toast) + ''' + code, _, _ = _run_ps(script) + return code == 0 + + +# ═══════════════════════════════════════════════════════════════ +# 显示/屏幕 — 真实实现 (pyautogui, screeninfo, win32api) +# ═══════════════════════════════════════════════════════════════ + +class WindowsDisplay: + """真实显示/屏幕信息""" + + @staticmethod + def get_screen_size() -> Tuple[int, int]: + """获取屏幕尺寸""" + # 方法1: screeninfo + if HAS_SCREENINFO: + try: + monitors = get_monitors() + if monitors: + m = monitors[0] + return m.width, m.height + except Exception: + pass + # 方法2: win32api + if HAS_WIN32: + try: + width = win32api.GetSystemMetrics(0) + height = win32api.GetSystemMetrics(1) + return width, height + except Exception: + pass + # 方法3: pyautogui + if HAS_PYAUTOGUI: + try: + size = pyautogui.size() + return size.width, size.height + except Exception: + pass + # 方法4: tkinter + try: + import tkinter + root = tkinter.Tk() + w = root.winfo_screenwidth() + h = root.winfo_screenheight() + root.destroy() + return w, h + except Exception: + return 1920, 1080 + + @staticmethod + def screenshot(path: str = None) -> Optional[str]: + """截屏""" + if HAS_PYAUTOGUI: + try: + img = pyautogui.screenshot() + if path: + img.save(path) + return path + else: + buf = os.path.join(os.environ.get("TEMP", "C:\\Temp"), + f"screenshot_{int(time.time())}.png") + img.save(buf) + return buf + except Exception: + pass + if HAS_PIL: + try: + img = ImageGrab.grab() + if path: + img.save(path) + return path + else: + buf = os.path.join(os.environ.get("TEMP", "C:\\Temp"), + f"screenshot_{int(time.time())}.png") + img.save(buf) + return buf + except Exception: + pass + return None + + @staticmethod + def get_mouse_position() -> Tuple[int, int]: + """获取鼠标位置""" + if HAS_PYAUTOGUI: + try: + pos = pyautogui.position() + return pos.x, pos.y + except Exception: + pass + if HAS_WIN32: + try: + pos = win32api.GetCursorPos() + return pos[0], pos[1] + except Exception: + pass + return 0, 0 + + +# ═══════════════════════════════════════════════════════════════ +# UI 自动化 — 真实实现 (pyautogui) +# ═══════════════════════════════════════════════════════════════ + +class WindowsUIAutomation: + """真实 UI 自动化 — 通过 pyautogui""" + + @staticmethod + def click(x: int, y: int) -> bool: + if not HAS_PYAUTOGUI: + return False + try: + pyautogui.click(x, y) + return True + except Exception: + return False + + @staticmethod + def double_click(x: int, y: int) -> bool: + if not HAS_PYAUTOGUI: + return False + try: + pyautogui.doubleClick(x, y) + return True + except Exception: + return False + + @staticmethod + def right_click(x: int, y: int) -> bool: + if not HAS_PYAUTOGUI: + return False + try: + pyautogui.rightClick(x, y) + return True + except Exception: + return False + + @staticmethod + def type_text(text: str) -> bool: + if not HAS_PYAUTOGUI: + return False + try: + pyautogui.typewrite(text, interval=0.02) + return True + except Exception: + return False + + @staticmethod + def press_key(key: str) -> bool: + if not HAS_PYAUTOGUI: + return False + key_map = { + "enter": "enter", "return": "enter", + "tab": "tab", "escape": "escape", "esc": "escape", + "backspace": "backspace", "delete": "delete", + "space": "space", "up": "up", "down": "down", + "left": "left", "right": "right", + "home": "home", "end": "end", + "page_up": "pageup", "page_down": "pagedown", + } + pyautogui_key = key_map.get(key.lower(), key) + try: + pyautogui.press(pyautogui_key) + return True + except Exception: + return False + + @staticmethod + def hotkey(*keys) -> bool: + if not HAS_PYAUTOGUI: + return False + key_map = { + "ctrl": "ctrl", "control": "ctrl", + "alt": "alt", "shift": "shift", + "super": "win", "win": "win", "meta": "win", + } + mapped = [key_map.get(k.lower(), k) for k in keys] + try: + pyautogui.hotkey(*mapped) + return True + except Exception: + return False + + @staticmethod + def scroll(amount: int) -> bool: + if not HAS_PYAUTOGUI: + return False + try: + pyautogui.scroll(amount) + return True + except Exception: + return False + + @staticmethod + def find_window(title: str) -> Optional[int]: + """查找窗口""" + if HAS_WIN32: + try: + hwnd = win32gui.FindWindow(None, title) + if hwnd: + return hwnd + except Exception: + pass + # 模糊搜索 + if HAS_WIN32: + found = [] + def callback(hwnd, _): + if win32gui.IsWindowVisible(hwnd): + window_title = win32gui.GetWindowText(hwnd) + if title.lower() in window_title.lower(): + found.append(hwnd) + return True + try: + win32gui.EnumWindows(callback, None) + if found: + return found[0] + except Exception: + pass + return None + + @staticmethod + def activate_window(window_id: int) -> bool: + if HAS_WIN32: + try: + win32gui.SetForegroundWindow(window_id) + return True + except Exception: + pass + return False + + @staticmethod + def close_window(window_id: int) -> bool: + if HAS_WIN32: + try: + win32gui.PostMessage(window_id, win32con.WM_CLOSE, 0, 0) + return True + except Exception: + pass + return False + + +# ═══════════════════════════════════════════════════════════════ +# 摄像头 — 真实实现 (opencv) +# ═══════════════════════════════════════════════════════════════ + +class WindowsCamera: + """真实摄像头 — 通过 OpenCV""" + + @staticmethod + def list_cameras() -> List[int]: + """列出可用摄像头""" + cameras = [] + if HAS_CV2: + for i in range(10): + cap = cv2.VideoCapture(i) + if cap.isOpened(): + cameras.append(i) + cap.release() + else: + break + return cameras + + @staticmethod + def capture(camera_id: int = 0, output_path: str = None) -> Optional[str]: + """拍照""" + if not HAS_CV2: + return None + cap = cv2.VideoCapture(camera_id) + if not cap.isOpened(): + return None + ret, frame = cap.read() + cap.release() + if not ret: + return None + if not output_path: + output_path = os.path.join(os.environ.get("TEMP", "C:\\Temp"), + f"camera_{int(time.time())}.jpg") + cv2.imwrite(output_path, frame) + return output_path + + @staticmethod + def get_camera_info(camera_id: int = 0) -> dict: + """获取摄像头信息""" + info = {"id": camera_id, "available": False, "width": 0, "height": 0} + if HAS_CV2: + cap = cv2.VideoCapture(camera_id) + if cap.isOpened(): + info["available"] = True + info["width"] = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)) + info["height"] = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) + info["fps"] = cap.get(cv2.CAP_PROP_FPS) + cap.release() + return info + + +# ═══════════════════════════════════════════════════════════════ +# 进程/应用管理 — 真实实现 (psutil) +# ═══════════════════════════════════════════════════════════════ + +class WindowsProcess: + """真实进程/应用管理""" + + @staticmethod + def list_processes() -> List[dict]: + """列出所有进程""" + if HAS_PSUTIL: + procs = [] + for p in psutil.process_iter(['pid', 'name', 'username', 'memory_percent', + 'cpu_percent', 'status', 'cmdline']): + try: + info = p.info + procs.append({ + "pid": info['pid'], + "name": info['name'], + "user": info['username'], + "memory": info.get('memory_percent', 0), + "cpu": info.get('cpu_percent', 0), + "status": info['status'], + "cmdline": " ".join(info.get('cmdline') or []), + }) + except (psutil.NoSuchProcess, psutil.AccessDenied): + pass + return procs + # 回退到 tasklist + code, out, _ = _run("tasklist /fo csv /v") + if code == 0: + procs = [] + lines = out.split("\n") + for line in lines[1:]: + parts = line.split('","') + if len(parts) >= 8: + procs.append({ + "name": parts[0].strip('"'), + "pid": int(parts[1].strip('"')), + "user": parts[6].strip('"') if len(parts) > 6 else "", + "status": parts[7].strip('"') if len(parts) > 7 else "", + "memory": 0, + "cpu": 0, + "cmdline": "", + }) + return procs + return [] + + @staticmethod + def get_running_apps() -> List[dict]: + """获取运行中的应用 (GUI 窗口)""" + apps = [] + if HAS_WIN32: + def callback(hwnd, _): + if win32gui.IsWindowVisible(hwnd): + title = win32gui.GetWindowText(hwnd) + if title and title not in ("Default IME", "MSCTFIME UI"): + apps.append({"window_id": hwnd, "name": title}) + return True + try: + win32gui.EnumWindows(callback, None) + except Exception: + pass + return apps + + @staticmethod + def kill_process(pid: int) -> bool: + """终止进程""" + if HAS_PSUTIL: + try: + psutil.Process(pid).terminate() + return True + except Exception: + return False + code, _, _ = _run(f"taskkill /pid {pid} /f") + return code == 0 + + @staticmethod + def start_app(name: str) -> bool: + """启动应用""" + try: + os.startfile(name) + return True + except Exception: + pass + try: + subprocess.Popen(name, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + return True + except Exception: + return False + + @staticmethod + def get_system_info() -> dict: + """获取系统信息""" + info = { + "platform": platform.system(), + "hostname": socket.gethostname(), + "kernel": platform.release(), + "arch": platform.machine(), + "python": platform.python_version(), + "windows_edition": platform.win32_edition() if hasattr(platform, 'win32_edition') else "", + "windows_version": platform.version(), + } + if HAS_PSUTIL: + info["cpu_count"] = psutil.cpu_count() + cpu_freq = psutil.cpu_freq() + info["cpu_freq"] = cpu_freq.current if cpu_freq else 0 + boot_time = psutil.boot_time() + info["uptime"] = time.time() - boot_time + mem = psutil.virtual_memory() + info["memory_total"] = mem.total + info["memory_used"] = mem.used + info["memory_percent"] = mem.percent + disk = psutil.disk_usage('C:\\') + info["disk_total"] = disk.total + info["disk_used"] = disk.used + info["disk_percent"] = disk.percent + return info + + +# ═══════════════════════════════════════════════════════════════ +# Shell — 真实实现 +# ═══════════════════════════════════════════════════════════════ + +class WindowsShell: + """真实 Shell 执行""" + + @staticmethod + def exec(command: str, timeout: int = 30, root: bool = False) -> dict: + """执行 Shell 命令""" + if root: + # 以管理员权限运行 + script = f'Start-Process cmd -ArgumentList "/c {command}" -Verb RunAs -Wait' + code, out, err = _run_ps(script, timeout=timeout) + else: + code, out, err = _run(command, timeout=timeout) + return { + "success": code == 0, + "output": out + ("\n" + err if err else ""), + "exit_code": code, + "stdout": out, + "stderr": err, + } + + @staticmethod + def exec_async(command: str) -> int: + """异步执行命令，返回 PID""" + try: + proc = subprocess.Popen(command, shell=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + return proc.pid + except Exception: + return -1 + + +# ═══════════════════════════════════════════════════════════════ +# 统一入口 +# ═══════════════════════════════════════════════════════════════ + +def get_windows_capabilities() -> dict: + """获取当前 Windows 环境的可用能力""" + return { + "is_windows": IS_WINDOWS, + "network": True, + "bluetooth": HAS_POWERSHELL, + "audio": HAS_PYCAW or HAS_POWERSHELL, + "clipboard": HAS_PYPERCLIP, + "ui_automation": HAS_PYAUTOGUI, + "notifications": HAS_WIN10TOAST or HAS_POWERSHELL, + "sensors": HAS_WMI or HAS_PSUTIL, + "display": True, + "camera": HAS_CV2, + "ffmpeg": HAS_FFMPEG, + "ffprobe": HAS_FFPROBE, + "psutil": HAS_PSUTIL, + "requests": HAS_REQUESTS, + "PIL": HAS_PIL, + "opencv": HAS_CV2, + "numpy": HAS_NUMPY, + "wmi": HAS_WMI, + "pyautogui": HAS_PYAUTOGUI, + "pyperclip": HAS_PYPERCLIP, + "pycaw": HAS_PYCAW, + "win32": HAS_WIN32, + "screeninfo": HAS_SCREENINFO, + "win10toast": HAS_WIN10TOAST, + } + + +# 导出所有实现类 +__all__ = [ + "IS_WINDOWS", "get_windows_capabilities", + "WindowsNetwork", "WindowsBluetooth", "WindowsSensors", + "WindowsAudio", "WindowsClipboard", "WindowsPower", + "WindowsLocation", "WindowsNotify", "WindowsDisplay", + "WindowsUIAutomation", "WindowsCamera", "WindowsProcess", + "WindowsShell", + # 能力标志 + "HAS_POWERSHELL", "HAS_FFMPEG", "HAS_FFPROBE", "HAS_PSUTIL", + "HAS_REQUESTS", "HAS_CV2", "HAS_NUMPY", "HAS_PIL", + "HAS_PYAUTOGUI", "HAS_PYPERCLIP", "HAS_WIN10TOAST", + "HAS_WMI", "HAS_WIN32", "HAS_PYCAW", "HAS_SCREENINFO", +] diff --git a/setup.py b/setup.py index b6361b7..be42b24 100644 --- a/setup.py +++ b/setup.py @@ -22,6 +22,21 @@ setup( "opencv-python", "numpy", ], + "windows": [ + "psutil", + "requests", + "Pillow", + "opencv-python", + "numpy", + "pyautogui", + "pyperclip", + "screeninfo", + "pywin32", + "wmi", + "pycaw", + "comtypes", + "win10toast", + ], "dev": [ "pytest", "pytest-cov",