import uiautomator2
defmain():
d = uiautomator2.connect()print("Device has been connected. Device Info:")print(d.info)# Launch Bilibili APP
d.app_start('tv.danmaku.bili', stop=True)
d.wait_activity('.MainActivityV2')
d(text="我的").wait(timeout=10)# Wait for the splash AD to finish
d(text="我的").click()# Show the fans count
fans_count = d(resourceId="tv.danmaku.bili:id/fans_count").get_text()print(f"Fans count of my bilibili account: {fans_count}")
publicclassBinariesExporter{privatefinalApplication app;publicBinariesExporter(Application app){this.app = app;}// The libxxx.so for the architecture of the current platform will be linked to {linkingDir}/libxxx.so.// (Note that the linkingDir must be in the application's private directory!)// If there is a custom name mapping in fileNameMapping, the symbolic link will be created to the corresponding file name.publicvoidexportLinkTo(File linkingDir,Map<String,String> fileNameMapping)throwsErrnoException{if(linkingDir.exists())deleteDir(linkingDir);
linkingDir.mkdirs();File nativeLibsFolder =newFile(app.getApplicationInfo().nativeLibraryDir);File[] files = nativeLibsFolder.listFiles();if(files ==null)return;for(File srcFile : files){if(srcFile.isFile()){String destName = srcFile.getName();if(fileNameMapping.containsKey(destName))
destName = fileNameMapping.get(destName);File linkFile =newFile(linkingDir, destName);Os.symlink(srcFile.getAbsolutePath(), linkFile.getAbsolutePath());}}}privatestaticbooleandeleteDir(File f){if(f ==null)returntrue;if(f.isDirectory()){String[] children = f.list();if(children !=null){for(String child : children){boolean success =deleteDir(newFile(f, child));if(!success){returnfalse;}}}}return f.delete();}}
// Pair devicepublicbooleanpairDevice(int pairingPort,String pairingCode)throwsSecurityException{if(hasWriteSecureSettingsPermission()){// This permission is only granted after pairing once.// So if this perm is granted, we can confirm that the pairing had been finished.
logger.d("Device had been paired. Nothing to do here.");returntrue;}
logger.d("Pairing device at port "+ pairingPort +" with code "+ pairingCode +"...");if(!adbProcess.pairDevice(pairingPort, pairingCode)){
logger.e("Fail to pair the device!");thrownewSecurityException("Fail to pair the device");}
logger.d("Pairing succeeded.");// After pairing, grant the WRITE_SECURE_SETTINGS permission for current appgrantWriteSecureSettingsPermission();returntrue;}// Enable wireless ADB through secure settings@RequiresApi(Build.VERSION_CODES.TIRAMISU)protectedvoidenableWirelessAdb()throwsSecurityException{if(!hasWriteSecureSettingsPermission()){
logger.e("No permission for WRITE_SECURE_SETTINGS");thrownewSecurityException("No permission for WRITE_SECURE_SETTINGS");}
logger.d("WRITE_SECURE_SETTINGS permission granted.");
logger.d("Enabling wireless ADB...");finalContentResolver cr = context.getContentResolver();Settings.Global.putInt(cr,"adb_wifi_enabled",1);Settings.Global.putInt(cr,Settings.Global.ADB_ENABLED,1);Settings.Global.putLong(cr,"adb_allowed_connection_time",0L);if(Settings.Global.getInt(cr,"adb_wifi_enabled",0)!=1){
logger.e("Fail to enable wireless ADB");thrownewSecurityException("Fail to enable wireless ADB");}
logger.d("Wireless ADB enabled.");}// Discover ADB service on current device with mDNS@RequiresApi(Build.VERSION_CODES.R)protectedCompletableFuture<Integer>discoverAdbService(){finalContentResolver cr = context.getContentResolver();CompletableFuture<Integer> result =newCompletableFuture<>();ExecutorService executor =Executors.newSingleThreadExecutor();
executor.submit(()->{
logger.d("Finding wireless ADB service...");finalCountDownLatch latch =newCountDownLatch(1);AdbMdns adbMdns =newAdbMdns(context,AdbMdns.TLS_CONNECT, port ->{if(port <=0){
logger.e("Fail to find wireless ADB service.");thrownewSecurityException("Fail to find wireless ADB service");}
logger.d("Wireless ADB service found at port "+ port);
result.complete(port);
latch.countDown();});try{if(Settings.Global.getInt(cr,"adb_wifi_enabled",0)==1){
logger.d("Wireless ADB mDNS finding...");
adbMdns.start();
latch.await(3,TimeUnit.SECONDS);
adbMdns.stop();}}catch(InterruptedException e){
e.printStackTrace();}});
executor.shutdown();return result;}// Returns adb port numberpublicCompletableFuture<Integer>enableAndDiscoverAdbPort()throwsSecurityException,UnsupportedOperationException{if(Build.VERSION.SDK_INT>=Build.VERSION_CODES.TIRAMISU){enableWirelessAdb();returndiscoverAdbService();}else{thrownewUnsupportedOperationException("Android version is too old");}}
通过 Chaquopy 执行 uiautomator2 自动化脚本,将各种运行需要的参数配置传递进去,然后执行 main 函数运行其主要逻辑
6.2. 修改 Python 代码
回到最开始编写的 python/main.py 文件,将代码修改为如下形式:
import os
import sys
import warnings
import uiautomator2
warnings.filterwarnings("ignore", category=ResourceWarning)
context =None
adb_address =""# Load configs from Android layordefload_android_configs(app_context, adb_path:str, ld_dir:str, adb_port:int):global context
context = app_context
global adb_address
adb_address =f"127.0.0.1:{adb_port}"print("")
os.environ["ADBUTILS_ADB_PATH"]= adb_path
print(f"ADB Path set to: {adb_path}")
os.environ['LD_LIBRARY_PATH']= ld_dir
print(f"LD_LIBRARY_PATH set to:{ld_dir}\n")# Print something to python consoledefprint_to_console(text:str, end="\n"):print(text, end=end)defmain():print(f"Connecting to {adb_address}...\n")
d = uiautomator2.connect(adb_address)print(f"Device has been connected. Device Info: {d.info}\n")# Launch Bilibili APP
d.app_start('tv.danmaku.bili', stop=True)
d.wait_activity('.MainActivityV2')
d(text="我的").wait(timeout=10)# Wait for the splash AD to finish
d(text="我的").click()# Show the fans count
fans_count = d(resourceId="tv.danmaku.bili:id/fans_count").get_text()print(f"Fans count of my bilibili account: {fans_count}")