何去何從？
原計劃第三篇要寫機器學習的內容，然而文章還沒寫完，酷Q沒了。
可是，真的沒有辦法了嗎？
我們相信冬天總會過去，不過，在春天到來之前，生一叢篝火取暖，也算是聊勝於無吧。
這篇文章，我將介紹
如何使用 Airtest（Poco） 框架進行簡單的 QQ 訊息自動收發處理
。
關於 Airtest
Airtest
是一款網易出品的自動化測試套件
，包括 Airtest（影象識別）、Poco（UI控制元件搜尋）和 Airtest IDE。Airtest 提供了一系列自動操作應用程式的介面，我們利用這些介面，就可以實現 QQ 訊息的自動收發處理。
Airtest 也是基於 Python 開發的，因此可以和之前兩篇文章中介紹的 chatterbot （以及沒來得及寫的 pytorch）很好地協作。
Airtest 可以在官網下載，其中 Airtest IDE 需要登入賬號才能使用。（可以使用網易賬號或者 Github 賬號登陸。）如果需要不依賴於 Airtest IDE，直接從命令列啟動指令碼，還需在本地的 Python 環境中安裝 Airtest 庫：
pip install airtest pocoui
一些構想
自 QQ 的第三方協議關停以來，我一直在尋求替代方案。從 Github 上一個名叫 FoolQQ 的專案得到啟發，我想出了自動化操作的方法。
首先在安卓模擬器上執行QQ並登陸賬號（
為什麼不用PC版？因為 Airtest 對安卓的支援相對更成熟
），然後使用 Airtest 這樣的框架來讀取其中的內容，或者傳送訊息。目前這種方法可以實現單個聊天中的訊息收發。
在模擬器的選擇上，我使用的是
逍遙模擬器
，它對 QQ 和 Airtest 的相容性都不錯。為了節省空間，我選擇了
QQ 極速版
（原輕聊版），以下內容均以此版本為依據。
使用 Poco
Poco 是 Airtest 中的 UI 控制元件搜尋框架
，基於 uiautomatior2，支援對 Android 原生應用以及一些遊戲引擎中 UI 元素層級的解析。有一個好訊息，在 QQ 的介面中，
所有的控制元件都是採用了繼承修改 Android 原生控制元件的模式
，這使得我們能夠直接利用 Poco 讀取訊息內容，避免了複雜的 OCR 實現。
在閱讀這一章節之前，可以先去看看 Airtest 的文件，瞭解一下 Airtest 的基礎知識以及 IDE 的使用，並掌握在 IDE 中連線安卓模擬器的方法。
連線模擬器之後，執行 QQ，然後在 IDE 的「Poco 輔助穿」的下拉列表中選擇「Android」，可以看到這樣的 UI 渲染樹：
雙擊其中的元素，就可以在指令碼編輯器中生成選擇這個元素的程式碼。
在 QQ 輕聊版的聊天介面，透過這種方式，我們分析出了幾個關鍵的 UI 控制元件：
poco（“com。tencent。qqlite：id/listView1”）
這個控制元件是當前顯示的聊天內容，它有若干個包名為
com。tencent。qqlite：id/base_chat_item_layout
的子節點，每個子結點的包名為
com。tencent。qqlite：id/chat_item_content_layout
的子節點的
text
屬性就是訊息內容。
poco（“com。tencent。qqlite：id/input”）
底部輸入框。
從這些出發，我們得到了檢視當前顯示的訊息的程式碼：
def
get_message_list
（）：
def
get_message_list_iter
（）：
try
：
msg_list
=
poco
（
“com。tencent。qqlite：id/listView1”
）
for
msg_container
in
msg_list
。
child
（
“com。tencent。qqlite：id/base_chat_item_layout”
）：
msg
=
msg_container
。
child
（
“com。tencent。qqlite：id/chat_item_content_layout”
）
if
msg
。
get_position
（）［
0
］
<
0。5
：
# 排除自己傳送的訊息
txt
=
msg
。
get_text
（）
if
txt
：
yield
txt
。
strip
（）
finally
：
pass
return
［
msg
for
msg
in
get_message_list_iter
（）］
以及傳送訊息的程式碼：
def
send_message
（
msg_text
）：
try
：
input_box
=
poco
（
“com。tencent。qqlite：id/input”
）
input_box
。
click
（）
input_box
。
set_text
（
msg_text
）
keyevent
（
“ENTER”
）
# 請在 QQ 的「設定-通用」中選中「回車鍵傳送訊息」
return
True
except
：
return
False
但是 Poco 只能獲取控制元件的基本資訊，這就導致了我們
無法得到訊息的傳送時間
。為了能獲取“新訊息”，我們採用類似於輪詢的機制，
每隔一段時間獲取一下訊息列表，對比最新訊息的內容有無變化，如果有變化，就認為之後的訊息都是“新訊息”
。
這種辦法不能識別有人復讀的情況，不過問題不大。
這個過程可以抽象為一個類：
class
MessageListener
（
object
）：
def
__init__
（
self
）：
message_list
=
get_message_list
（）
if
message_list
：
self
。
latest_message
=
message_list
［
-
1
］
else
：
self
。
latest_message
=
None
def
check_latest_message
（
self
）：
message_list
=
get_message_list
（）
if
message_list
：
if
self
。
latest_message
is
None
：
self
。
latest_message
=
message_list
［
-
1
］
return
［］
def
check_latest_message_iter
（）：
for
msg
in
reversed
（
message_list
）：
msg
=
msg
if
msg
==
self
。
latest_message
：
break
yield
msg
result
=
reversed
（
list
（
check_latest_message_iter
（）））
self
。
latest_message
=
message_list
［
-
1
］
return
result
else
：
return
［］
這三段程式碼是實現訊息自動收發的基礎。
綜合一下
這裡給出完整的程式碼（
core。py
）：
# -*- encoding=utf8 -*-
# core。py
__author__
=
“忘憂北萱草”
from
poco。drivers。android。uiautomation
import
AndroidUiautomationPoco
from
airtest。core。api
import
*
import
logging
logger
=
logging
。
getLogger
（
“airtest”
）
logger
。
setLevel
（
logging
。
ERROR
）
# 遮蔽多餘的日誌輸出
# 以下是連線安卓模擬器用的程式碼。如果你是在 Airtest IDE 中執行這些程式碼，可以刪去下一行以及 auto_setup 中的 devices 引數
# 21503 埠是逍遙模擬器的預設埠，如果你使用的不是逍遙，請按照 Airtest 的文件說明更改此處
simulator
=
“Android：//127。0。0。1：5037/127。0。0。1：21503？cap_method=JAVACAP^&^&ori_method=ADBORI”
auto_setup
（
__file__
，
devices
=
［
simulator
］）
poco
=
AndroidUiautomationPoco
（
use_airtest_input
=
True
，
screenshot_each_action
=
False
）
# 目前還不支援自動進入某個聊天，所以請手動啟動 QQ 並開啟聊天介面
# start_app（“com。tencent。qqlite”）
# 獲取當前訊息
def
get_message_list
（）：
def
get_message_list_iter
（）：
try
：
msg_list
=
poco
（
“com。tencent。qqlite：id/listView1”
）
for
msg_container
in
msg_list
。
child
（
“com。tencent。qqlite：id/base_chat_item_layout”
）：
msg
=
msg_container
。
child
（
“com。tencent。qqlite：id/chat_item_content_layout”
）
if
msg
。
get_position
（）［
0
］
<
0。5
：
txt
=
msg
。
get_text
（）
if
txt
：
yield
txt
。
strip
（）
finally
：
pass
return
［
msg
for
msg
in
get_message_list_iter
（）］
# 傳送訊息
def
send_message
（
msg_text
）：
try
：
input_box
=
poco
（
“com。tencent。qqlite：id/input”
）
input_box
。
click
（）
input_box
。
set_text
（
msg_text
）
keyevent
（
“ENTER”
）
return
True
except
：
return
False
# 訊息輪詢
class
MessageListener
（
object
）：
def
__init__
（
self
）：
message_list
=
get_message_list
（）
if
message_list
：
self
。
latest_message
=
message_list
［
-
1
］
else
：
self
。
latest_message
=
None
def
check_latest_message
（
self
）：
message_list
=
get_message_list
（）
if
message_list
：
if
self
。
latest_message
is
None
：
self
。
latest_message
=
message_list
［
-
1
］
return
［］
def
check_latest_message_iter
（）：
for
msg
in
reversed
（
message_list
）：
msg
=
msg
if
msg
==
self
。
latest_message
：
break
yield
msg
result
=
reversed
（
list
（
check_latest_message_iter
（）））
self
。
latest_message
=
message_list
［
-
1
］
return
result
else
：
return
［］
使用方法參看以下的 demo：
import
random
import
core
if
__name__
==
‘__main__’
：
messageListener
=
core
。
MessageListener
（）
（
‘初始化完成。’
）
while
True
：
messages
=
messageListener
。
check_latest_message
（）
for
msg
in
messages
：
（
‘>>> ’
+
msg
）
if
msg
==
‘報告狀態’
：
core
。
send_message
（
‘執行自檢……連線不穩定……’
）
core
。
sleep
（
1
）
本文所提到的技術已在 Github 上開源：
結語
這篇文章寫的有些倉促，不像之前兩篇那麼詳細，因為我想盡快把我的經驗分享給大家。
以上。
（封面PID：75286867）
