347 lines
11 KiB
Python
347 lines
11 KiB
Python
import subprocess
|
||
import re
|
||
import time
|
||
import requests
|
||
import json
|
||
from appium import webdriver
|
||
from appium.webdriver.common.appiumby import AppiumBy
|
||
from appium.options.android import UiAutomator2Options
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
from selenium.webdriver.support import expected_conditions as EC
|
||
from urllib3.connection import port_by_scheme
|
||
|
||
|
||
# =======================
|
||
# 基础工具函数
|
||
# =======================
|
||
|
||
def run_command(command):
|
||
"""执行系统命令并返回输出"""
|
||
result = subprocess.run(command, shell=True, capture_output=True, text=True)
|
||
return result.stdout.strip()
|
||
|
||
|
||
# =======================
|
||
# API请求函数
|
||
# =======================
|
||
|
||
def get_new_port(yh_id):
|
||
"""从服务器获取新的端口号"""
|
||
url = "https://engineering.yuxindazhineng.com/index/index/get_yh_port"
|
||
headers = {
|
||
"Content-Type": "application/x-www-form-urlencoded"
|
||
}
|
||
data = {
|
||
"yh_id": yh_id
|
||
}
|
||
|
||
try:
|
||
print(f"🔍 查询服务器新端口号，用户ID: {yh_id}")
|
||
response = requests.post(url, headers=headers, data=data, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 0:
|
||
print(f"✅ 查询成功，新端口号: {result.get('data', '未知')}")
|
||
return result.get("data", None)
|
||
else:
|
||
print(f"❌ 查询失败: {result.get('msg', '未知错误')}")
|
||
return None
|
||
else:
|
||
print(f"❌ 服务器响应错误: {response.status_code}")
|
||
return None
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 网络请求失败: {e}")
|
||
return None
|
||
|
||
def get_accounts_from_server(yh_id):
|
||
"""从服务器获取账户信息"""
|
||
url = "http://www.yuxindazhineng.com:3002/api/accounts/get_uplaod_data"
|
||
headers = {
|
||
"Content-Type": "application/json"
|
||
}
|
||
data = {
|
||
"yh_id": yh_id
|
||
}
|
||
|
||
try:
|
||
print(f"🔍 查询服务器账户信息，用户ID: {yh_id}")
|
||
response = requests.post(url, headers=headers, json=data, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 0:
|
||
print(f"✅ 查询成功，找到 {result.get('total', 0)} 个账户")
|
||
return result.get("data", [])
|
||
else:
|
||
print(f"❌ 查询失败: {result.get('message', '未知错误')}")
|
||
return []
|
||
else:
|
||
print(f"❌ 服务器响应错误: {response.status_code}")
|
||
return []
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 网络请求失败: {e}")
|
||
return []
|
||
except json.JSONDecodeError as e:
|
||
print(f"❌ JSON解析失败: {e}")
|
||
return []
|
||
|
||
|
||
def update_device_info(account_id, device_name, device_port, device_ip):
|
||
"""更新设备信息到服务器"""
|
||
url = "http://www.yuxindazhineng.com:3002/api/accounts/update"
|
||
headers = {
|
||
"Content-Type": "application/json"
|
||
}
|
||
data = {
|
||
"account_id": str(account_id),
|
||
"account_data": {
|
||
"device_name": str(device_name),
|
||
"device_port": str(device_port),
|
||
"device_ip": str(device_ip)
|
||
}
|
||
}
|
||
|
||
try:
|
||
print(f"🔄 更新设备信息，账户ID: {account_id}")
|
||
print(f" 设备信息: 名称={device_name}, 端口={device_port}, IP={device_ip}")
|
||
|
||
response = requests.post(url, headers=headers, json=data, timeout=10)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 0:
|
||
print(f"✅ 更新成功: {result.get('message', '未知信息')}")
|
||
return True
|
||
else:
|
||
print(f"❌ 更新失败: {result.get('message', '未知错误')}")
|
||
return False
|
||
else:
|
||
print(f"❌ 服务器响应错误: {response.status_code}")
|
||
return False
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"❌ 网络请求失败: {e}")
|
||
return False
|
||
|
||
|
||
# =======================
|
||
# Appium 启动
|
||
# =======================
|
||
|
||
def start_appium():
|
||
print("🚀 启动 Appium Server ...")
|
||
subprocess.Popen(
|
||
["appium.cmd", "-a", "127.0.0.1", "-p", "4723"],
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL
|
||
)
|
||
time.sleep(5) # 给 Appium 启动时间
|
||
print("✅ Appium Server 已启动")
|
||
|
||
|
||
# =======================
|
||
# 启动沉降观测 App
|
||
# =======================
|
||
|
||
def start_settlement_app(device_id, device_ip, device_port):
|
||
print(f"📱 使用 Appium 连接设备: {device_id}")
|
||
|
||
options = UiAutomator2Options()
|
||
options.platform_name = "Android"
|
||
options.device_name = device_id
|
||
options.udid = device_id
|
||
|
||
# ⚠️ TODO：替换为你的真实信息
|
||
options.app_package = "com.bjjw.cjgc"
|
||
options.app_activity = ".activity.LoginActivity"
|
||
|
||
options.automation_name = "UiAutomator2"
|
||
options.no_reset = True
|
||
options.auto_grant_permissions = True
|
||
options.new_command_timeout = 28800
|
||
|
||
# 超时增强（无线 ADB 必须）
|
||
options.set_capability("uiautomator2ServerLaunchTimeout", 60000)
|
||
options.set_capability("adbExecTimeout", 120000)
|
||
|
||
driver = webdriver.Remote(
|
||
"http://127.0.0.1:4723",
|
||
options=options
|
||
)
|
||
|
||
# 使用ADB命令启动Activity
|
||
try:
|
||
adb_command = f"adb -s {device_id} shell am start -n com.bjjw.cjgc/.activity.LoginActivity"
|
||
result = subprocess.run(adb_command, shell=True, capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
time.sleep(1) # 等待Activity启动
|
||
except Exception:
|
||
return False
|
||
|
||
print("✅ 沉降观测 App 已成功启动")
|
||
|
||
# =======================
|
||
# 获取用户名文本框内容
|
||
# =======================
|
||
app_username = None
|
||
try:
|
||
print("🔍 尝试获取用户名文本框内容...")
|
||
|
||
# 创建显式等待对象
|
||
wait = WebDriverWait(driver, 15) # 最多等待15秒
|
||
|
||
# 等待用户名文本框可点击
|
||
username_field = wait.until(
|
||
EC.element_to_be_clickable((AppiumBy.ID, "com.bjjw.cjgc:id/et_user_name"))
|
||
)
|
||
|
||
# 获取文本框中的文本内容
|
||
app_username = username_field.text
|
||
|
||
# 如果文本框是空的，尝试使用get_attribute获取文本
|
||
if not app_username:
|
||
app_username = username_field.get_attribute("text")
|
||
|
||
# 如果还是空的，尝试获取其他属性
|
||
if not app_username:
|
||
app_username = username_field.get_attribute("content-desc") or username_field.get_attribute("label")
|
||
|
||
if app_username:
|
||
print(f"✅ 成功获取到用户名: {app_username}")
|
||
else:
|
||
print("⚠️ 用户名文本框为空")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 获取用户名失败: {e}")
|
||
|
||
return driver, app_username
|
||
|
||
|
||
# =======================
|
||
# 无线 ADB 建链主流程
|
||
# =======================
|
||
|
||
def setup_adb_wireless(yh_id="68c0dbfdb7cbcd616e7c5ab5"):
|
||
port = get_new_port(yh_id)
|
||
# port = 3435
|
||
print(f"🚀 开始无线 ADB 建链（端口 {port}）")
|
||
print(f"📋 用户ID: {yh_id}")
|
||
|
||
# 从服务器获取账户信息
|
||
accounts = get_accounts_from_server(yh_id)
|
||
if not accounts:
|
||
print("❌ 未从服务器获取到账户信息，终止流程")
|
||
return
|
||
|
||
devices_output = run_command("adb devices")
|
||
lines = devices_output.splitlines()[1:]
|
||
|
||
usb_devices = []
|
||
|
||
for line in lines:
|
||
if not line.strip():
|
||
continue
|
||
|
||
device_id = line.split()[0]
|
||
|
||
# 跳过已经是无线的
|
||
if ":" in device_id:
|
||
continue
|
||
|
||
usb_devices.append(device_id)
|
||
|
||
if not usb_devices:
|
||
print("❌ 未检测到 USB 设备")
|
||
return
|
||
|
||
for serial in usb_devices:
|
||
print(f"\n🔎 处理设备: {serial}")
|
||
|
||
# 获取 WLAN IP
|
||
ip_info = run_command(f"adb -s {serial} shell ip addr show wlan0")
|
||
ip_match = re.search(r'inet\s+(\d+\.\d+\.\d+\.\d+)', ip_info)
|
||
|
||
if not ip_match:
|
||
print("⚠️ 获取 IP 失败，请确认已连接 WiFi")
|
||
continue
|
||
|
||
device_ip = ip_match.group(1)
|
||
print(f"📍 设备 IP: {device_ip}")
|
||
|
||
# 切 TCP 模式
|
||
run_command(f"adb -s {serial} tcpip {port}")
|
||
time.sleep(2)
|
||
|
||
# 无线连接
|
||
connect_result = run_command(f"adb connect {device_ip}:{port}")
|
||
|
||
if "connected" not in connect_result.lower():
|
||
print(f"❌ 无线连接失败: {connect_result}")
|
||
continue
|
||
|
||
wireless_id = f"{device_ip}:{port}"
|
||
print(f"✅ 无线 ADB 成功: {wireless_id}")
|
||
|
||
# ===== 后续自动化 =====
|
||
start_appium()
|
||
driver, app_username = start_settlement_app(wireless_id, device_ip, port)
|
||
|
||
if not app_username:
|
||
print("⚠️ 未获取到App中的用户名，跳过服务器更新")
|
||
continue
|
||
|
||
# 在账户列表中查找匹配的用户名
|
||
matched_account = None
|
||
for account in accounts:
|
||
if account.get("username") == app_username:
|
||
matched_account = account
|
||
break
|
||
|
||
if not matched_account:
|
||
print(f"❌ 未找到与用户名 '{app_username}' 匹配的账户")
|
||
continue
|
||
|
||
print(f"✅ 找到匹配账户: {matched_account.get('cl_name')} ({matched_account.get('username')})")
|
||
print(f" account_id: {matched_account.get('account_id')}")
|
||
|
||
# 更新设备信息到服务器
|
||
device_name = serial # 使用设备序列号作为设备名称
|
||
|
||
# 构建更新数据
|
||
update_data = {
|
||
"account_id": matched_account.get("account_id"),
|
||
"device_name": device_name,
|
||
"device_port": port,
|
||
"device_ip": device_ip
|
||
}
|
||
|
||
success = update_device_info(
|
||
account_id=matched_account.get("account_id"),
|
||
device_name=device_name,
|
||
device_port=port,
|
||
device_ip=device_ip
|
||
)
|
||
|
||
if success:
|
||
print(f"🎉 所有操作完成! 账户 {matched_account.get('username')} 的设备信息已更新")
|
||
else:
|
||
print(f"⚠️ 设备信息更新失败，但无线连接和App启动已完成")
|
||
|
||
# 关闭Appium连接
|
||
if driver:
|
||
print("🔄 关闭Appium连接...")
|
||
driver.quit()
|
||
|
||
break # 处理完第一个设备后退出，如需处理多个设备可移除此行
|
||
|
||
|
||
# =======================
|
||
# 程序入口
|
||
# =======================
|
||
|
||
if __name__ == "__main__":
|
||
# 配置参数
|
||
|
||
USER_ID = "68c0dbfdb7cbcd616e7c5ab5" # 替换为实际的用户ID
|
||
|
||
setup_adb_wireless(USER_ID) |