🐝 蜂巢云控 Python SDK
简单、强大的手机自动化控制SDK，支持 无障碍服务、ADB USB、ADB WiFi、HID/OTG 多种控制方式。
✨ 特性
- 🚀 简单易用 - 几行代码即可控制手机
- 📶 无障碍服务 - 无需ROOT，稳定可靠
- 🔄 批量控制 - 同时控制50+台设备
- ⚡ 并行操作 - 多线程同时执行，效率翻倍
- 🔍 OCR识别 - 屏幕文字识别与点击
- 🖼️ 图像识别 - 模板匹配查找图标
- 📱 元素查找 - 按ID/类名/文本查找元素
- 🌐 远程控制 - 通过中继服务器远程控制
📦 安装
pip install fengchao-sdk
安装全部功能：
pip install fengchao-sdk[all] # 包含OCR和图像识别
📱 无障碍模式（推荐）
无障碍模式无需ROOT，通过安卓无障碍服务实现控制，支持远程操作。
1️⃣ WiFi直连模式
手机和电脑在同一局域网时使用：
from fengchao import Device
# 连接设备（手机IP:端口）
device = Device('192.168.1.100', port=8888, mode='wifi')
# 基础操作
device.screenshot() # 截图
device.click(500, 800) # 点击
device.swipe(100, 800, 100, 200) # 滑动
device.input_text('Hello') # 输入文字
# 按键操作
device.press_home() # Home键
device.press_back() # 返回键
device.press_menu() # 菜单键
# APP操作
device.start_app('com.android.settings') # 启动APP
device.stop_app('com.android.settings') # 停止APP
device.get_current_app() # 获取当前APP
2️⃣ 中继服务器模式
通过中继服务器远程控制设备：
from fengchao import Device
# 连接中继服务器
device = Device(
device_id='设备ID',
api_key='你的API密钥',
relay_server='http://服务器地址:9999',
mode='relay'
)
# 所有操作与WiFi直连一致
device.screenshot()
device.click(500, 800)
device.press_home()
3️⃣ 元素查找（无障碍核心功能）
通过无障碍服务可以精确查找和操作UI元素：
# 按文本查找
elem = device.find_element(text='登录')
if elem:
elem.click() # 点击元素
# 按资源ID查找
elem = device.find_element(resource_id='com.example:id/btn_login')
# 按类名查找
elems = device.find_elements(class_name='android.widget.Button')
print(f'找到 {len(elems)} 个按钮')
# 按内容描述查找
elem = device.find_element(content_desc='返回')
# 组合查找
elem = device.find_element(
class_name='android.widget.TextView',
text='确定'
)
# 检查元素是否存在
if device.element_exists(text='登录成功'):
print('登录成功！')
# 等待元素出现
elem = device.wait_element(text='加载完成', timeout=10)
# 直接点击元素
device.click_element(text='确定')
device.click_element(resource_id='com.example:id/btn')
元素属性
elem = device.find_element(text='登录')
print(elem.text) # 文本内容
print(elem.resource_id) # 资源ID
print(elem.class_name) # 类名
print(elem.content_desc) # 内容描述
print(elem.bounds) # 坐标范围
print(elem.clickable) # 是否可点击
4️⃣ 图像识别
# 截图并裁剪模板
img = device.screenshot()
template = img.crop((100, 100, 200, 200))
template.save('button.png')
# 查找图片
result = device.find_image('button.png', threshold=0.8)
if result:
print(f"找到位置: {result['center']}")
print(f"置信度: {result['confidence']}")
# 点击图片
device.click_image('button.png')
# 等待图片出现
device.wait_image('loading_done.png', timeout=10)
# 检查图片是否存在
if device.image_exists('success.png'):
print('操作成功！')
5️⃣ OCR文字识别
需要安装：pip install rapidocr-onnxruntime
# OCR识别屏幕文字
texts = device.ocr()
for t in texts:
print(f"{t['text']} @ {t['center']}")
# 查找文字
result = device.find_text('登录')
if result:
print(f"找到: {result['text']} @ {result['center']}")
# 点击文字
device.click_text('登录')
device.click_text('确定', timeout=5)
# 检查文字是否存在
if device.text_exists('登录成功'):
print('成功！')
# 等待文字出现
device.wait_text('加载完成', timeout=10)
6️⃣ 完整API列表
基础操作
| 方法 |
说明 |
示例 |
screenshot() |
截图，返回PIL Image |
img = device.screenshot() |
click(x, y) |
点击坐标 |
device.click(500, 800) |
long_press(x, y, duration) |
长按 |
device.long_press(500, 800, 1000) |
swipe(x1, y1, x2, y2, duration) |
滑动 |
device.swipe(500, 1500, 500, 500) |
input_text(text) |
输入文字 |
device.input_text('Hello') |
按键操作
| 方法 |
说明 |
press_home() |
Home键 |
press_back() |
返回键 |
press_menu() |
菜单键 |
press_key(keycode) |
发送按键码 |
APP操作
| 方法 |
说明 |
示例 |
start_app(package) |
启动APP |
device.start_app('com.android.settings') |
stop_app(package) |
停止APP |
device.stop_app('com.android.settings') |
get_current_app() |
获取当前APP |
app = device.get_current_app() |
元素查找
| 方法 |
说明 |
find_element(text, resource_id, class_name, content_desc) |
查找单个元素 |
find_elements(...) |
查找多个元素 |
click_element(...) |
查找并点击元素 |
element_exists(...) |
检查元素是否存在 |
wait_element(..., timeout) |
等待元素出现 |
图像识别
| 方法 |
说明 |
find_image(template, threshold) |
查找图片 |
click_image(template) |
点击图片 |
image_exists(template) |
检查图片是否存在 |
wait_image(template, timeout) |
等待图片出现 |
OCR识别
| 方法 |
说明 |
ocr(region) |
OCR识别 |
find_text(text) |
查找文字 |
click_text(text, timeout) |
点击文字 |
text_exists(text) |
检查文字是否存在 |
wait_text(text, timeout) |
等待文字出现 |
🔌 ADB 连接方式
1️⃣ USB 连接（最简单）
from fengchao import ADBDevice
# 自动连接第一台USB设备
device = ADBDevice()
# 或指定设备序列号
device = ADBDevice(serial="A9GVVB2B30026001")
# 开始操作
device.click(500, 800)
device.swipe_up()
2️⃣ WiFi 连接（无需USB线）
方式A：USB转WiFi（推荐首次使用）
from fengchao import ADBDevice
# 1. 先用USB连接
device = ADBDevice()
# 2. 开启WiFi调试模式
device.enable_tcpip(5555)
# 3. 获取手机IP
ip = device.get_ip_address()
print(f"手机IP: {ip}")
# 4. 现在可以拔掉USB线了！
# 5. 通过WiFi连接
wifi_device = ADBDevice.connect_wifi(ip, 5555)
# 6. WiFi控制手机
wifi_device.click(500, 800)
wifi_device.swipe_up()
方式B：直接WiFi连接（手机已开启tcpip）
from fengchao import ADBDevice
# 直接用IP连接（手机需要已开启 adb tcpip 5555）
device = ADBDevice.connect_wifi("192.168.1.100", 5555)
# 操作手机
device.start_app("com.smile.gifmaker")
device.swipe_up()
WiFi连接数量
| 限制因素 |
推荐数量 |
| 单台电脑 |
20-50台 |
| 极限测试 |
100台+ |
主要受限于：电脑性能、网络带宽
🔄 批量操作（重点！）
串行批量（简单但慢）
from fengchao import ADBDevice
# 获取所有设备
serials = ADBDevice.list_devices()
devices = [ADBDevice(s) for s in serials]
# 逐个操作（一台一台来）
for d in devices:
d.swipe_up()
⚡ 并行批量（同时操作，超快！）
from fengchao import ADBDevice
from concurrent.futures import ThreadPoolExecutor
# 连接所有设备
serials = ADBDevice.list_devices()
devices = [ADBDevice(s) for s in serials]
# 🚀 同时操作所有设备（不是for循环！）
with ThreadPoolExecutor(max_workers=50) as pool:
# 所有设备同时上滑
list(pool.map(lambda d: d.swipe_up(), devices))
# 所有设备同时打开快手
list(pool.map(lambda d: d.start_app("com.smile.gifmaker"), devices))
🎯 封装成一行代码
from fengchao import ADBDevice
from concurrent.futures import ThreadPoolExecutor
# 连接所有设备
devices = [ADBDevice(s) for s in ADBDevice.list_devices()]
def batch(func):
"""批量并行执行"""
with ThreadPoolExecutor(max_workers=50) as pool:
return list(pool.map(func, devices))
# 使用示例 - 一行代码控制所有设备！
batch(lambda d: d.swipe_up()) # 全部上滑
batch(lambda d: d.press_home()) # 全部回桌面
batch(lambda d: d.click(500, 800)) # 全部点击
batch(lambda d: d.start_app("com.smile.gifmaker")) # 全部打开快手
batch(lambda d: d.screenshot(f"screen_{d.serial}.png")) # 全部截图
📊 批量获取信息
# 获取所有设备电量
results = batch(lambda d: {"serial": d.serial, "battery": d.get_battery_level()})
for r in results:
print(f"{r['serial']}: {r['battery']}%")
# 获取所有设备信息
infos = batch(lambda d: d.get_device_info())
for info in infos:
print(f"{info['brand']} {info['model']} - Android {info['android_version']}")
📚 完整API文档
设备连接
| 方法 |
说明 |
示例 |
ADBDevice() |
自动连接第一台USB设备 |
device = ADBDevice() |
ADBDevice(serial) |
连接指定序列号设备 |
device = ADBDevice("ABC123") |
ADBDevice.connect_wifi(ip, port) |
WiFi连接设备 |
device = ADBDevice.connect_wifi("192.168.1.100", 5555) |
ADBDevice.list_devices() |
列出所有已连接设备 |
serials = ADBDevice.list_devices() |
device.enable_tcpip(port) |
开启WiFi调试模式 |
device.enable_tcpip(5555) |
device.disconnect() |
断开WiFi连接 |
device.disconnect() |
触摸操作
| 方法 |
参数 |
说明 |
click(x, y) |
x, y: 坐标 |
点击屏幕 |
double_click(x, y) |
x, y: 坐标 |
双击屏幕 |
long_press(x, y, duration) |
duration: 毫秒，默认1000 |
长按屏幕 |
swipe(x1, y1, x2, y2, duration) |
duration: 毫秒，默认300 |
滑动 |
swipe_up(duration) |
duration: 毫秒，默认300 |
上滑（刷视频） |
swipe_down(duration) |
duration: 毫秒，默认300 |
下滑 |
swipe_left(duration) |
duration: 毫秒，默认300 |
左滑 |
swipe_right(duration) |
duration: 毫秒，默认300 |
右滑 |
device.click(500, 800) # 点击
device.double_click(500, 800) # 双击
device.long_press(500, 800, duration=2000) # 长按2秒
device.swipe(100, 800, 900, 800, duration=500) # 从左滑到右
device.swipe_up() # 上滑
device.swipe_down() # 下滑
按键操作
| 方法 |
说明 |
press_back() |
返回键 |
press_home() |
Home键 |
press_menu() |
菜单键 |
press_recent() |
最近任务键 |
press_key(keycode) |
发送任意按键码 |
device.press_back() # 返回
device.press_home() # 回到桌面
device.press_menu() # 菜单
device.press_recent() # 最近任务
device.press_key(24) # 音量+ (KEYCODE_VOLUME_UP)
device.press_key(25) # 音量- (KEYCODE_VOLUME_DOWN)
device.press_key(26) # 电源键 (KEYCODE_POWER)
文本输入
| 方法 |
参数 |
说明 |
input_text(text) |
text: 要输入的文字 |
输入文字（仅英文数字） |
clear_text(length) |
length: 删除字符数，默认50 |
清空输入框 |
set_clipboard(text) |
text: 文字内容 |
设置剪贴板 |
input_from_clipboard() |
- |
粘贴剪贴板内容 |
device.input_text("hello123") # 输入英文数字
device.clear_text() # 清空
device.set_clipboard("中文内容") # 支持中文
device.input_from_clipboard() # 粘贴
屏幕操作
| 方法 |
参数 |
说明 |
screenshot(path) |
path: 保存路径（可选） |
截图 |
get_screen_size() |
- |
获取屏幕尺寸，返回 (width, height) |
wake_up() |
- |
唤醒屏幕 |
lock_screen() |
- |
锁屏 |
is_screen_on() |
- |
屏幕是否亮着 |
is_screen_locked() |
- |
屏幕是否锁定 |
device.screenshot("screen.png") # 保存截图
width, height = device.get_screen_size() # 获取尺寸
device.wake_up() # 唤醒
device.lock_screen() # 锁屏
print(device.is_screen_on()) # True/False
APP管理
| 方法 |
参数 |
说明 |
start_app(package) |
package: 包名 |
启动APP |
stop_app(package) |
package: 包名 |
停止APP |
get_current_app() |
- |
获取当前APP信息 |
is_app_installed(package) |
package: 包名 |
检查是否安装 |
install_app(apk_path) |
apk_path: APK路径 |
安装APP |
uninstall_app(package) |
package: 包名 |
卸载APP |
list_packages() |
- |
列出所有已安装包 |
clear_app_data(package) |
package: 包名 |
清除APP数据 |
# 常用APP包名
KUAISHOU = "com.smile.gifmaker" # 快手
DOUYIN = "com.ss.android.ugc.aweme" # 抖音
WECHAT = "com.tencent.mm" # 微信
TAOBAO = "com.taobao.taobao" # 淘宝
device.start_app(KUAISHOU) # 打开快手
device.stop_app(KUAISHOU) # 关闭快手
app = device.get_current_app() # 获取当前APP
print(app["package"]) # 包名
print(app["activity"]) # Activity
设备信息
| 方法 |
返回值 |
说明 |
get_device_info() |
dict |
设备完整信息 |
get_battery_level() |
int |
电量百分比 |
get_battery_status() |
dict |
电池详细状态 |
get_ip_address() |
str |
设备IP地址 |
get_mac_address() |
str |
MAC地址 |
get_brightness() |
int |
屏幕亮度 |
get_density() |
int |
屏幕密度 |
get_cpu_usage() |
float |
CPU使用率 |
get_memory_info() |
dict |
内存信息 |
info = device.get_device_info()
print(f"品牌: {info['brand']}")
print(f"型号: {info['model']}")
print(f"Android版本: {info['android_version']}")
print(f"分辨率: {info['screen_resolution']}")
print(f"电量: {device.get_battery_level()}%")
print(f"IP: {device.get_ip_address()}")
文件操作
| 方法 |
参数 |
说明 |
push(local, remote) |
local: 本地路径, remote: 手机路径 |
推送文件到手机 |
pull(remote, local) |
remote: 手机路径, local: 本地路径 |
从手机拉取文件 |
list_files(path) |
path: 目录路径 |
列出目录文件 |
file_exists(path) |
path: 文件路径 |
检查文件是否存在 |
delete_file(path) |
path: 文件路径 |
删除文件 |
mkdir(path) |
path: 目录路径 |
创建目录 |
device.push("local.txt", "/sdcard/remote.txt") # 上传
device.pull("/sdcard/remote.txt", "local.txt") # 下载
files = device.list_files("/sdcard/") # 列出文件
device.mkdir("/sdcard/mydir") # 创建目录
device.delete_file("/sdcard/mydir") # 删除
OCR文字识别
需要安装：pip install rapidocr-onnxruntime
| 方法 |
参数 |
说明 |
init_ocr(engine) |
engine: "rapidocr"/"easyocr"/"paddleocr" |
初始化OCR引擎 |
ocr_screen() |
- |
识别屏幕所有文字 |
find_text(text) |
text: 要查找的文字 |
查找文字位置 |
find_all_text(text) |
text: 要查找的文字 |
查找所有匹配 |
click_text(text) |
text: 要点击的文字 |
找到文字并点击 |
text_exists(text) |
text: 要查找的文字 |
检查文字是否存在 |
wait_for_text(text, timeout) |
timeout: 超时秒数 |
等待文字出现 |
# 初始化OCR（首次会慢，后续复用）
device.init_ocr(engine="rapidocr")
# 识别屏幕所有文字
results = device.ocr_screen()
for r in results:
print(f"{r['text']} @ {r['position']}")
# 查找并点击文字
device.click_text("登录")
device.click_text("确定")
device.click_text("关闭")
# 检查文字是否存在
if device.text_exists("登录成功"):
print("登录成功！")
# 等待文字出现
device.wait_for_text("加载完成", timeout=10)
UI元素分析
| 方法 |
参数 |
说明 |
dump_ui() |
- |
获取UI XML |
get_ui_elements() |
- |
获取所有UI元素 |
find_element_by_text(text) |
text: 文字 |
通过文字查找元素 |
find_element_by_id(resource_id) |
resource_id: 资源ID |
通过ID查找元素 |
find_element_by_desc(desc) |
desc: 描述 |
通过描述查找元素 |
click_by_text(text) |
text: 文字 |
点击文字元素 |
click_by_id(resource_id) |
resource_id: 资源ID |
点击ID元素 |
click_by_desc(desc) |
desc: 描述 |
点击描述元素 |
# 查找元素
elem = device.find_element_by_text("登录")
if elem:
print(f"找到: {elem['text']} @ {elem['bounds']}")
device.click(*elem["center"]) # 点击元素中心
# 通过ID查找
elem = device.find_element_by_id("com.xxx:id/btn_login")
# 通过描述查找
elem = device.find_element_by_desc("点赞")
# 直接点击
device.click_by_text("确定")
device.click_by_id("com.xxx:id/btn")
device.click_by_desc("返回")
# 打印UI树
device.print_ui_tree()
图像识别
需要安装：pip install opencv-python
| 方法 |
参数 |
说明 |
find_image(template, threshold) |
template: 模板图片路径, threshold: 阈值0-1 |
查找图片位置 |
find_all_images(template, threshold) |
同上 |
查找所有匹配 |
click_image(template, threshold) |
同上 |
找到图片并点击 |
image_exists(template, threshold) |
同上 |
检查图片是否存在 |
wait_for_image(template, timeout) |
timeout: 超时秒数 |
等待图片出现 |
# 查找并点击图标
device.click_image("heart.png") # 点击心形图标
device.click_image("button.png", threshold=0.8)
# 检查图标是否存在
if device.image_exists("success.png"):
print("操作成功！")
# 等待图片出现
device.wait_for_image("loading_done.png", timeout=10)
系统设置
| 方法 |
参数 |
说明 |
set_brightness(level) |
level: 0-255 |
设置亮度 |
set_auto_brightness(enable) |
enable: True/False |
自动亮度 |
enable_wifi() / disable_wifi() |
- |
WiFi开关 |
enable_mobile_data() / disable_mobile_data() |
- |
移动数据开关 |
enable_airplane_mode() / disable_airplane_mode() |
- |
飞行模式开关 |
set_screen_timeout(ms) |
ms: 毫秒 |
屏幕超时时间 |
set_density(dpi) |
dpi: 密度值 |
设置屏幕密度 |
reset_density() |
- |
恢复默认密度 |
device.set_brightness(128) # 设置亮度50%
device.set_auto_brightness(True) # 开启自动亮度
device.enable_wifi() # 打开WiFi
device.disable_wifi() # 关闭WiFi
device.set_screen_timeout(60000) # 屏幕1分钟后关闭
其他功能
| 方法 |
说明 |
open_url(url) |
打开网址 |
call_phone(number) |
拨打电话 |
send_sms(number, text) |
发送短信 |
open_settings() |
打开设置 |
get_logcat(lines) |
获取日志 |
clear_logcat() |
清除日志 |
reboot() |
重启设备 |
shell(cmd) |
执行Shell命令 |
device.open_url("https://www.baidu.com")
device.open_settings()
device.shell("pm list packages")
🎬 实战示例
批量刷快手视频
from fengchao import ADBDevice
from concurrent.futures import ThreadPoolExecutor
import time
def kuaishou_script(device):
"""快手刷视频脚本"""
print(f"[{device.serial}] 开始")
# 打开快手
device.start_app("com.smile.gifmaker")
time.sleep(3)
# 关闭弹窗
for text in ["关闭", "跳过", "我知道了"]:
try:
device.click_text(text)
time.sleep(0.5)
except:
pass
# 刷10个视频
for i in range(10):
time.sleep(5) # 看5秒
device.swipe_up() # 下一个
print(f"[{device.serial}] 视频 {i+1}/10")
print(f"[{device.serial}] 完成")
# 连接所有设备
devices = [ADBDevice(s) for s in ADBDevice.list_devices()]
print(f"共 {len(devices)} 台设备")
# 并行执行
with ThreadPoolExecutor(max_workers=len(devices)) as pool:
pool.map(kuaishou_script, devices)
print("全部完成！")
WiFi批量控制
from fengchao import ADBDevice
from concurrent.futures import ThreadPoolExecutor
# WiFi设备IP列表
DEVICE_IPS = [
"192.168.1.101",
"192.168.1.102",
"192.168.1.103",
"192.168.1.104",
"192.168.1.105",
]
# 连接所有WiFi设备
devices = []
for ip in DEVICE_IPS:
try:
d = ADBDevice.connect_wifi(ip, 5555)
devices.append(d)
print(f"✅ 连接成功: {ip}")
except:
print(f"❌ 连接失败: {ip}")
print(f"共连接 {len(devices)} 台设备")
# 批量操作
def batch(func):
with ThreadPoolExecutor(max_workers=50) as pool:
return list(pool.map(func, devices))
# 全部回桌面
batch(lambda d: d.press_home())
# 全部打开快手
batch(lambda d: d.start_app("com.smile.gifmaker"))
# 全部刷视频
for i in range(10):
print(f"第 {i+1} 次上滑")
batch(lambda d: d.swipe_up())
time.sleep(5)
📋 常用APP包名
| APP |
包名 |
| 快手 |
com.smile.gifmaker |
| 抖音 |
com.ss.android.ugc.aweme |
| 微信 |
com.tencent.mm |
| 淘宝 |
com.taobao.taobao |
| 拼多多 |
com.xunmeng.pinduoduo |
| 京东 |
com.jingdong.app.mall |
| 支付宝 |
com.eg.android.AlipayGphone |
| QQ |
com.tencent.mobileqq |
| 微博 |
com.sina.weibo |
| 小红书 |
com.xingin.xhs |
| B站 |
tv.danmaku.bili |
| 设置 |
com.android.settings |
🔧 常见问题
Q: WiFi连接失败？
# 1. 确保手机和电脑在同一WiFi
# 2. 先用USB连接，执行：
adb tcpip 5555
# 3. 获取手机IP后连接
Q: OCR识别慢？
# 预加载OCR引擎
device.init_ocr(engine="rapidocr")
# 后续识别会很快
Q: 如何查看设备序列号？
adb devices
Q: 批量操作有数量限制吗？
- 推荐：20-50台/电脑
- 极限：100台+
- 取决于电脑性能和网络
📄 许可证
MIT License
