90 lines
3.4 KiB
Python
90 lines
3.4 KiB
Python
|
# coding=utf-8
|
|||
|
|
import uiautomator2 as u2
|
|||
|
|
import time
|
|||
|
|
import logging
|
|||
|
|
import sys
|
|||
|
|
import os
|
|||
|
|
import asyncio
|
|||
|
|
|
|||
|
|
# 添加项目根目录到 sys.path 以便导入 Util
|
|||
|
|
project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|||
|
|
if project_root not in sys.path:
|
|||
|
|
sys.path.append(project_root)
|
|||
|
|
|
|||
|
|
from WeiXin.WxUtil import find_input_box_center, perform_input_action
|
|||
|
|
from Util.LlmUtil import get_llm_response
|
|||
|
|
|
|||
|
|
# 配置日志
|
|||
|
|
log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Logs")
|
|||
|
|
if not os.path.exists(log_dir):
|
|||
|
|
os.makedirs(log_dir)
|
|||
|
|
|
|||
|
|
logging.basicConfig(
|
|||
|
|
level=logging.INFO,
|
|||
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
|||
|
|
handlers=[
|
|||
|
|
logging.FileHandler(os.path.join(log_dir, "T4_InputLlmText.log"), encoding='utf-8'),
|
|||
|
|
logging.StreamHandler()
|
|||
|
|
]
|
|||
|
|
)
|
|||
|
|
logger = logging.getLogger("T4_InputLlmText")
|
|||
|
|
|
|||
|
|
async def generate_and_input():
|
|||
|
|
logger.info("开始执行 T4: 生成 LLM 回复并输入...")
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# 1. 连接设备
|
|||
|
|
d = u2.connect()
|
|||
|
|
logger.info(f"设备连接成功: {d.info.get('serial')}")
|
|||
|
|
|
|||
|
|
# 2. 调用 LLM 生成回复
|
|||
|
|
# 模拟一个简单的场景：针对之前的对话生成回复
|
|||
|
|
prompt = "对方说：'AI 助手我现在可以开始和你聊天了！'，请给出一个简短、友好且自然的回复，不超过 20 字。不要包含任何解释性文字。"
|
|||
|
|
logger.info(f"正在请求 LLM 生成回复, Prompt: {prompt}")
|
|||
|
|
|
|||
|
|
full_response = ""
|
|||
|
|
async for chunk in get_llm_response(prompt, stream=False):
|
|||
|
|
full_response += chunk
|
|||
|
|
|
|||
|
|
llm_text = full_response.strip().strip('"').strip('“').strip('”')
|
|||
|
|
logger.info(f"LLM 生成的回复内容: {llm_text}")
|
|||
|
|
|
|||
|
|
if not llm_text:
|
|||
|
|
logger.error("LLM 生成内容为空，停止执行。")
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
# 3. 识别输入框位置
|
|||
|
|
# 先截个图供识别使用
|
|||
|
|
screenshot_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Screenshots")
|
|||
|
|
if not os.path.exists(screenshot_dir):
|
|||
|
|
os.makedirs(screenshot_dir)
|
|||
|
|
|
|||
|
|
tmp_shot = os.path.join(screenshot_dir, "t4_temp_input_check.jpg")
|
|||
|
|
d.screenshot(tmp_shot)
|
|||
|
|
|
|||
|
|
center_point, rect_box = find_input_box_center(tmp_shot)
|
|||
|
|
|
|||
|
|
if center_point:
|
|||
|
|
cx, cy = center_point
|
|||
|
|
logger.info(f"识别到输入框中心: ({cx}, {cy})")
|
|||
|
|
|
|||
|
|
# 4. 点击输入框并输入文字
|
|||
|
|
# 使用 WxUtil 中的 perform_input_action
|
|||
|
|
# 该函数内部会先点击坐标，然后输入内容
|
|||
|
|
# 设置 auto_send=True，输入文字后点击发送
|
|||
|
|
logger.info(f"正在点击输入框并输入文字并发送: {llm_text}")
|
|||
|
|
success = perform_input_action(d, center_point, llm_text, auto_send=True)
|
|||
|
|
|
|||
|
|
if success:
|
|||
|
|
logger.info("✅ T4 执行完成：文字已成功输入并点击发送。")
|
|||
|
|
else:
|
|||
|
|
logger.error("❌ T4 执行失败：输入动作未成功完成。")
|
|||
|
|
else:
|
|||
|
|
logger.error("❌ T4 执行失败：未能识别到输入框位置。")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"❌ T4 执行出错: {e}", exc_info=True)
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
asyncio.run(generate_and_input())
|