🐝 蜂巢云控 Python SDK
简单、强大的手机自动化控制SDK，支持 无障碍服务、ADB USB、ADB WiFi、HID/OTG 多种控制方式。
✨ 特性
- 🚀 简单易用 - 几行代码即可控制手机
- 📶 无障碍服务 - 无需ROOT，稳定可靠
- 🔄 批量控制 - 同时控制50+台设备
- ⚡ 并行操作 - 多线程同时执行，效率翻倍
- 🔍 OCR识别 - 屏幕文字识别与点击
- 🖼️ 图像识别 - 模板匹配查找图标
- 📱 元素查找 - 按ID/类名/文本查找元素
- 🌐 远程控制 - 通过中继服务器远程控制
📦 安装
pip install fengchao-sdk
安装全部功能：
pip install fengchao-sdk[all] # 包含OCR和图像识别
📱 无障碍模式（推荐）
无障碍模式是最推荐的控制方式！无需ROOT、无需ADB、无需电脑连接，通过安卓无障碍服务实现稳定控制。
🔧 准备工作
- 安装蜂巢云控APP - 在手机上安装APP
- 开启无障碍服务 - 设置 → 无障碍 → 蜂巢云控 → 开启
- 授予截图权限 - 首次截图时会弹窗，点击允许
- 获取连接信息：
- WiFi直连：获取手机IP地址（APP主界面显示）
- 中继模式：获取设备ID和API密钥
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) |
等待文字出现 |
7️⃣ 常用按键码
| 按键 |
Keycode |
说明 |
| Home |
3 |
回到桌面 |
| Back |
4 |
返回 |
| Menu |
82 |
菜单键 |
| Power |
26 |
电源键 |
| Volume Up |
24 |
音量+ |
| Volume Down |
25 |
音量- |
| Enter |
66 |
确认键 |
| Delete |
67 |
删除键 |
| Tab |
61 |
Tab键 |
| Space |
62 |
空格键 |
device.press_key(3) # Home
device.press_key(4) # Back
device.press_key(24) # Volume Up
device.press_key(25) # Volume Down
device.press_key(26) # Power
8️⃣ 无障碍模式实战示例
示例1：自动登录APP
from fengchao import Device
import time
device = Device(
device_id='你的设备ID',
api_key='你的API密钥',
relay_server='http://服务器:9999',
mode='relay'
)
# 启动APP
device.start_app('com.example.app')
time.sleep(3)
# 等待登录按钮出现
if device.wait_element(text='登录', timeout=10):
device.click_element(text='登录')
time.sleep(1)
# 输入账号
device.click_element(resource_id='com.example.app:id/username')
device.input_text('myusername')
# 输入密码
device.click_element(resource_id='com.example.app:id/password')
device.input_text('mypassword')
# 点击登录按钮
device.click_element(text='登录')
# 检查是否登录成功
if device.wait_element(text='首页', timeout=10):
print('登录成功！')
示例2：刷短视频
from fengchao import Device
import time
device = Device(
device_id='设备ID',
api_key='API密钥',
relay_server='http://服务器:9999',
mode='relay'
)
# 打开抖音
device.start_app('com.ss.android.ugc.aweme')
time.sleep(5)
# 关闭可能的弹窗
for text in ['我知道了', '关闭', '跳过', '以后再说']:
if device.text_exists(text):
device.click_text(text)
time.sleep(0.5)
# 刷视频
for i in range(100):
print(f'观看第 {i+1} 个视频')
time.sleep(8) # 观看8秒
device.swipe(540, 1500, 540, 500, 300) # 上滑
示例3：OCR识别验证码
from fengchao import Device
device = Device('192.168.1.100', port=8888, mode='wifi')
# 截图
img = device.screenshot()
# OCR识别指定区域（验证码区域）
texts = device.ocr(region=(100, 500, 300, 550))
if texts:
code = texts[0]['text']
print(f'识别到验证码: {code}')
# 输入验证码
device.click(400, 600) # 点击输入框
device.input_text(code)
示例4：图像识别点击
from fengchao import Device
device = Device('192.168.1.100', port=8888, mode='wifi')
# 等待某个图标出现并点击
if device.wait_image('like_button.png', timeout=10):
device.click_image('like_button.png')
print('点击成功')
# 检查是否存在某个状态图标
if device.image_exists('success_icon.png'):
print('操作成功')
else:
print('操作失败')
示例5：批量控制多台设备
from fengchao import Device
from concurrent.futures import ThreadPoolExecutor
import time
# 设备列表
DEVICES = [
{'device_id': 'device1', 'api_key': 'key1'},
{'device_id': 'device2', 'api_key': 'key2'},
{'device_id': 'device3', 'api_key': 'key3'},
]
RELAY_SERVER = 'http://服务器:9999'
def run_script(config):
"""单台设备的脚本"""
device = Device(
device_id=config['device_id'],
api_key=config['api_key'],
relay_server=RELAY_SERVER,
mode='relay'
)
device.press_home()
device.start_app('com.smile.gifmaker')
time.sleep(3)
for i in range(10):
time.sleep(5)
device.swipe(540, 1500, 540, 500, 300)
print(f"设备 {config['device_id']} 完成")
# 并行执行
with ThreadPoolExecutor(max_workers=len(DEVICES)) as pool:
pool.map(run_script, DEVICES)
print('全部完成！')
🔌 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']}")
📚 ADBDevice 完整API文档
以下是 ADBDevice 类（ADB模式）的完整API。
ADB设备连接
| 方法 |
说明 |
示例 |
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() |
enable_tcpip(port) |
开启WiFi调试模式 |
device.enable_tcpip(5555) |
disconnect() |
断开WiFi连接 |
device.disconnect() |
ADB触摸操作
| 方法 |
参数 |
说明 |
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() |
- |
上滑（刷视频） |
swipe_down() |
- |
下滑 |
swipe_left() |
- |
左滑 |
swipe_right() |
- |
右滑 |
ADB按键操作
| 方法 |
说明 |
press_back() |
返回键 |
press_home() |
Home键 |
press_menu() |
菜单键 |
press_recent() |
最近任务键 |
press_key(keycode) |
发送任意按键码 |
ADB文本输入
| 方法 |
参数 |
说明 |
input_text(text) |
text: 要输入的文字 |
输入文字 |
clear_text(length) |
length: 删除字符数 |
清空输入框 |
set_clipboard(text) |
text: 文字内容 |
设置剪贴板 |
input_from_clipboard() |
- |
粘贴剪贴板内容 |
ADB屏幕操作
| 方法 |
说明 |
screenshot(path) |
截图并保存 |
get_screen_size() |
获取屏幕尺寸 |
wake_up() |
唤醒屏幕 |
lock_screen() |
锁屏 |
is_screen_on() |
屏幕是否亮着 |
is_screen_locked() |
屏幕是否锁定 |
ADB APP管理
| 方法 |
说明 |
start_app(package) |
启动APP |
stop_app(package) |
停止APP |
get_current_app() |
获取当前APP |
is_app_installed(package) |
检查是否安装 |
install_app(apk_path) |
安装APP |
uninstall_app(package) |
卸载APP |
list_packages() |
列出所有已安装包 |
clear_app_data(package) |
清除APP数据 |
ADB设备信息
| 方法 |
说明 |
get_device_info() |
设备完整信息 |
get_battery_level() |
电量百分比 |
get_ip_address() |
设备IP地址 |
get_mac_address() |
MAC地址 |
get_brightness() |
屏幕亮度 |
get_cpu_usage() |
CPU使用率 |
get_memory_info() |
内存信息 |
ADB文件操作
| 方法 |
说明 |
push(local, remote) |
推送文件到手机 |
pull(remote, local) |
从手机拉取文件 |
list_files(path) |
列出目录文件 |
file_exists(path) |
检查文件是否存在 |
delete_file(path) |
删除文件 |
mkdir(path) |
创建目录 |
ADB系统设置
| 方法 |
说明 |
set_brightness(level) |
设置亮度 |
enable_wifi() / disable_wifi() |
WiFi开关 |
enable_airplane_mode() / disable_airplane_mode() |
飞行模式 |
reboot() |
重启设备 |
shell(cmd) |
执行Shell命令 |
ADB OCR/UI分析
| 方法 |
说明 |
init_ocr(engine) |
初始化OCR引擎 |
ocr_screen() |
识别屏幕文字 |
click_text(text) |
找到文字并点击 |
dump_ui() |
获取UI XML |
find_element_by_text(text) |
通过文字查找元素 |
find_element_by_id(id) |
通过ID查找元素 |
click_by_text(text) |
点击文字元素 |
from fengchao import ADBDevice
# ADB模式示例
device = ADBDevice()
device.start_app("com.smile.gifmaker")
device.swipe_up()
device.click_text("关注")
device.screenshot("screen.png")
device.push("local.txt", "/sdcard/test.txt")
info = device.get_device_info()
print(info)
# 识别屏幕所有文字
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)
🎬 ADB实战示例
批量刷快手视频
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
