import json import time import uiautomator2 as u2 import yaml import httpx from datetime import datetime, timedelta import subprocess from tools import loggerKit from difflib import SequenceMatcher # 读取 YAML 文件 with open('config.yaml', 'r') as file: config = yaml.load(file, Loader=yaml.FullLoader) extern_domain = config['bmp-cp']['extern_domain'] post_ai_gc_url = config['bmp-cp']['post_ai_gc_url'] class shell_status_content: def __init__(self, execute_status, error_code, error_msg, result, content, nickname): """ :rtype: object """ # 0失败 1成功 self.execute_status = execute_status self.error_code = error_code self.error_msg = error_msg self.result = result self.content = content self.nickname = nickname def to_dict(self): return { 'execute_status': self.execute_status, 'error_code': self.error_code, 'error_msg': self.error_msg, 'result': self.result, 'content': self.content, 'nickname': self.nickname } # 回复详情 class content_detail: def __init__(self, keyword, title, reply, account_name): """ :rtype: object """ # 搜索关键词 self.keyword = keyword self.title = title self.reply = reply self.account_name = account_name def to_dict(self): return { 'keyword': self.keyword, 'title': self.title, 'reply': self.reply, 'accountName': self.account_name } # 懂车帝 v7.8.3 # 3-5 天的互动操作 def simulated_operation(device_serial, keyword, task_id, media_channel, sub_resource_name): cmd_conn = f"adb connect {device_serial}" conn_output = subprocess.check_output(cmd_conn, shell=True) print(f"{conn_output}") d = u2.connect(device_serial) d.healthcheck() d.screen_on() d.unlock() d.debug = False # 跳过开屏广告 d.watcher.when("跳过广告").click() # 跳过升级弹框 d.watcher.when("以后再说").click() # 位置授权 d.watcher.when("本次运行允许").click() # 稍后 d.watcher.when("稍后").click() # 开始后台监控 # 默认监控间隔2.0s d.watcher.start(2.0) d.app_stop("com.ss.android.auto") d.app_start('com.ss.android.auto', use_monkey=True) loggerKit.info("懂车帝养号" + d.device_info['serial'] + ", 开始操作:{0}, {1},{2},{3}", device_serial, keyword, task_id, media_channel) time.sleep(5) # 进入我的获取账号信息 d.xpath('//*[@resource-id="android:id/tabs"]/android.widget.RelativeLayout[5]').click() time.sleep(2) account_name = d.xpath('//*[@resource-id="com.ss.android.auto:id/goi"]').get_text() # 点击到首页 d.xpath('//*[@resource-id="android:id/tabs"]/android.widget.RelativeLayout[1]').click() # 点击开始搜索 d.xpath('//*[@resource-id="com.ss.android.auto:id/d0r"]').click(timeout=0.25) time.sleep(2) replace_word = keyword.replace('|', '') # 输入框赋值 d.xpath('//*[@resource-id="com.ss.android.auto:id/h1k"]').set_text(replace_word) time.sleep(1) # 点击搜索 d.xpath('//*[@resource-id="com.ss.android.auto:id/g61"]').click() # 等待元素加载 time.sleep(5) xpath = '//*[@text="{}"]' # d.xpath(xpath.format(keyword)).all()[1].click() if d.xpath(xpath.format(keyword)).all().__len__() > 0: if d.xpath(xpath.format(keyword)).all().__len__() == 1: d.xpath(xpath.format(keyword)).click() else: d.xpath(xpath.format(keyword)).all()[1].click() else: # 第一次存在元素加载问题 if d.xpath(xpath.format(keyword)).all().__len__() == 1: d.xpath(xpath.format(keyword)).click() elif d.xpath(xpath.format(keyword)).all().__len__() >= 1: d.xpath(xpath.format(keyword)).all()[1].click() else: # 返回未找到帖子 if sub_resource_name is None or sub_resource_name == '': status = shell_status_content(0, 501, '该帖子未获取到，且无对应发帖人，无法进行深层次搜索', '', '', '') d.app_stop("com.ss.android.auto") return status else: return find_user(d, sub_resource_name, keyword, account_name, task_id) # if not d.xpath('//*[@resource-id="com.ss.android.auto:id/hzw"]').exists and not d.xpath( # '//*[@resource-id="com.ss.android.auto:id/iv_more"]').exists and (not d.xpath( # '//*[@resource-id="com.ss.android.auto:id/ff6"]').exists or not d.xpath( # '//*[@resource-id="com.ss.android.auto:id/d6l"]').exists): if not d.xpath('//*[@resource-id="com.ss.android.auto:id/hzw"]').exists and not d.xpath( '//*[@resource-id="com.ss.android.auto:id/iv_more"]').exists and not d.xpath( '//*[@resource-id="com.ss.android.auto:id/ff6"]').exists and not d.xpath( '//*[@resource-id="com.ss.android.auto:id/d6l"]').exists: # 返回未找到帖子 if sub_resource_name is None or sub_resource_name == '': status = shell_status_content(0, 501, '该帖子未获取到，且无对应发帖人，无法进行深层次搜索', '', '', '') d.app_stop("com.ss.android.auto") return status else: return find_user(d, sub_resource_name, keyword, account_name, task_id) return title_action(d, keyword, account_name, task_id) # 查询到用户主页寻找帖子 def find_user(d, user_name, keyword, account_name, task_id): time.sleep(3) if not d.xpath('//*[@resource-id="com.ss.android.auto:id/h1k"]').exists: status = shell_status_content(0, 501, '该帖子类型不支持/搜索用户页面错误', '', '', '') d.app_stop("com.ss.android.auto") return status # 赋值搜索框 用户名称 d.xpath('//*[@resource-id="com.ss.android.auto:id/h1k"]').set_text(user_name) # 点击搜索 d.xpath('//*[@resource-id="com.ss.android.auto:id/g61"]').click() time.sleep(3) # time.sleep(2) # d.xpath('//*[@text="小视频"]').click_exists() if d.xpath('//*[@text="二手车"]').exists: d(text='二手车').drag_to(text='综合', timeout=3) else: d(text='图片').drag_to(text='综合', timeout=3) # time.sleep(1) # d.xpath('//*[@text="车友圈"]').click_exists() # # time.sleep(1) # d.xpath('//*[@text="问答"]').click_exists() time.sleep(1) d.xpath('//*[@text="用户"]').click() time.sleep(3) xpath = '//*[@text="{}"]' user_name_list = d.xpath(xpath.format(user_name)).all() # 根据用户名获取元素 搜索框一定有一个 所以<=1 则表示未搜索到用户 if user_name_list.__len__() <= 1: # 当前用户没有获取到任何帖子 status = shell_status_content(0, 501, '未搜索到当前用户', '', '', '') return status else: # 懂车帝可能根据搜索算法 推出 是否仍要搜索目标选项， 这种情况对下标为2的进行点击。 正常情况对下标为1的进行点击 if not d.xpath('//*[@text="”的搜索结果。仍然搜索："]').exists: user_name_list[1].click() elif user_name_list.__len__() >= 3: user_name_list[2].click() # if user_name_list.__len__() > 1: # user_name_list[user_name_list.__len__() - 1].click() # else: # d.xpath(xpath.format(user_name)).click() # 点击第一个用户进入首页 # d.xpath('//*[@resource-id="root"]/android.view.View[2]/android.view.View[1]/android.view.View[1]').click_exists() time.sleep(2) # 如果存在关注用户标签 进行关闭 check_follow_button(d) time.sleep(1) # 获取帖子发布时间 if not d.xpath('//*[@resource-id="com.ss.android.auto:id/tv_time"]').exists: # 当前用户没有获取到任何帖子 status = shell_status_content(0, 501, '当前用户没有获取到任何帖子', '', '', '') return status # 预期时间 transfer_date = datetime.strptime('2023-10-01', '%Y-%m-%d').date() while 1: # 如果存在关注用户标签 进行关闭 check_follow_button(d) publish_time = d.xpath('//*[@resource-id="com.ss.android.auto:id/tv_time"]').get_text() publish_time = __formate_time(publish_time) # 发布时间 publish_date = datetime.strptime(publish_time, '%Y-%m-%d').date() # 获取到所有帖子标题进行模糊匹配 text_list = d.xpath('//*[@resource-id="com.ss.android.auto:id/s"]').all() for text in text_list: matcher = SequenceMatcher(None, text.text, keyword) similarity_ratio = matcher.ratio() if similarity_ratio >= 0.8: text.click() loggerKit.info("任务{0},用户{1}下搜索到{2}对应帖子 进行点击并操作", task_id, account_name, keyword) return title_action(d, keyword, account_name, task_id) # 如果能够根据标题搜索到帖子 # if d.xpath(xpath.format(keyword)).exists: # d.xpath(xpath.format(keyword)).click() # return title_action(d, keyword, account_name) # 先判断是否超过10月1号 if transfer_date > publish_date: # 已经抓取到10月1号帖子 仍未获取到该帖子 print('已经获取到10月1号帖子') status = shell_status_content(0, 501, '已经抓取到10月1号帖子 仍未获取到该帖子,不进行下一步获取', '', '', '') return status elif d.xpath('//*[@text="没有更多了"]').exists: # 没有更多帖子 print('没有更多了展示') status = shell_status_content(0, 501, "用户拉取到'没有更多了'标识，仍未获取到帖子", '', '', '') return status else: d.swipe_ext("up", 1) def check_follow_button(d): # 如果存在关注用户标签 进行关闭 if d.xpath('//*[@resource-id="com.ss.android.auto:id/byg"]').exists: d.xpath('//*[@resource-id="com.ss.android.auto:id/byg"]').click() def __formate_time(time_str): now = datetime.now() if '刚刚' in time_str: p_time = now elif '分钟' in time_str: minutes = int(time_str[:time_str.index('分')]) p_time = now - timedelta(minutes=minutes) elif '小时' in time_str: hours = int(time_str[:time_str.index('小')]) p_time = now - timedelta(hours=hours) elif '昨天' in time_str: p_time = now - timedelta(days=1) elif '前天' in time_str: p_time = now - timedelta(days=2) elif '天' in time_str: days = int(time_str[:time_str.index('天')]) p_time = now - timedelta(days=days) elif '周' in time_str: weeks = int(time_str[:time_str.index('周') - 1]) p_time = now - timedelta(weeks=weeks) elif '年' not in time_str and '月' in time_str and '日' in time_str: p_time = datetime.strptime(f'{now.year}年{time_str}', "%Y年%m月%d日").date() else: time_str = time_str.split()[0] items = time_str.split("-") if len(items) == 2: p_time = datetime.strptime(f'{now.year}年{items[0]}月{items[1]}日', "%Y年%m月%d日").date() elif len(items) == 3: p_time = datetime.strptime(f'{items[0]}年{items[1]}月{items[2]}日', "%Y年%m月%d日").date() else: return None # p_time = p_time.strftime('%Y-%m-%d') return p_time # 进入帖子进行操作 def title_action(d, keyword, account_name, task_id): time.sleep(3) # 目前只支持对帖子进行操作 if not d.xpath('//*[@resource-id="com.ss.android.auto:id/ff6"]').exists and not d.xpath( '//*[@resource-id="com.ss.android.auto:id/d6l"]').exists: # 可能受网络/帖子内容影响 元素未加载出来 等待三秒再重试加载 time.sleep(3) if not retry(lambda: d.xpath('//*[@resource-id="com.tencent.wework:id/ff6"]').exists, task_id): # if not d.xpath('//*[@resource-id="com.ss.android.auto:id/ff6"]').exists: # 获取评论内容，首先获取原文内容如果没有获取到，则获取标题 status = shell_status_content(0, 501, "该文章类型不支持操作", '', '', '') d.app_stop("com.ss.android.auto") return status if d.xpath('//*[@text="视频"]').exists: # 返回未找到帖子 status = shell_status_content(0, 501, '该帖子为视频类型不支持评论', '', '', '') d.app_stop("com.ss.android.auto") return status comment_request_text = '' if d.xpath('//*[@resource-id="com.ss.android.auto:id/s"]').exists: comment_request_text = d.xpath('//*[@resource-id="com.ss.android.auto:id/s"]').get_text() elif d.xpath('//*[@resource-id="com.ss.android.auto:id/izg"]').exists: comment_request_text = d.xpath('//*[@resource-id="com.ss.android.auto:id/izg"]').get_text() elif d.xpath('//*[@resource-id="com.ss.android.auto:id/p"]').exists: comment_request_text = d.xpath('//*[@resource-id="com.ss.android.auto:id/p"]').get_text() else: d.swipe_ext("up", 0.3) text_view_text = '' view_view_text = '' # 获取文章 article_list = d.xpath('//android.widget.TextView').all() if article_list is None or article_list.__len__() == 0: text_view_text = keyword else: # if article_list.__len__() > 3: # article_list = article_list[3:] for article in article_list: # if article.text.contains('说点什么'): # break text_view_text += article.text # 获取文章 article_list = d.xpath('//android.view.View').all() if article_list is None or article_list.__len__() == 0: view_view_text = keyword else: # if article_list.__len__() > 3: # article_list = article_list[3:] for article in article_list: # if article.text.contains('说点什么'): # break view_view_text += article.text if len(view_view_text) > len(text_view_text): comment_request_text = view_view_text else: comment_request_text = text_view_text request_data = { "templateId": "dcd_comment", "content": comment_request_text } # response = httpx.post('http://47.116.62.124:3000/comment', json=request_data, timeout=120) # # if not response.is_success: # # 调用AIGC获取评论失败 # status = shell_status_content(0, 500, "调用AIGC获取评论失败", '', '','') # d.app_stop("com.ss.android.auto") # return status # # data = json.loads(response.text) # # reply_text = data.get('text') # # reply_text_json = json.loads(reply_text) # # reply = reply_text_json.get('comment') response = httpx.post(post_ai_gc_url, json=request_data, timeout=120) if not response.is_success: # 调用AIGC获取评论失败 status = shell_status_content(0, 500, "调用AIGC获取评论失败", '', '', '') d.app_stop("com.ss.android.auto") return status response_body = json.loads(response.text) data = response_body.get('data') reply_text = data.get('answer') reply_text_json = json.loads(reply_text) reply = reply_text_json.get('comment') if '内容太少' in reply or '对不起' in reply or reply is None or reply == '': # 返回未获取合适回复内容 status = shell_status_content(0, 500, "无效评论,请求为:" + comment_request_text + "回复为:" + reply, '', '', '') d.app_stop("com.ss.android.auto") return status # 点赞 # if d.xpath('//*[@resource-id="com.ss.android.auto:id/d6l"]').exists: # like_list = d.xpath('//*[@resource-id="com.ss.android.auto:id/d6l"]').all() # if like_list.__len__() == 1: # like_list[0].click() # else: # like_list[like_list.__len__() - 1].click() # else: # like_list = d.xpath('//*[@resource-id="com.ss.android.auto:id/ff6"]').all() # if like_list.__len__() == 1: # like_list[0].click() # else: # like_list[like_list.__len__() - 1].click() # # time.sleep(2) # 点击对话框 if d.xpath('//*[@resource-id="com.ss.android.auto:id/kl2"]').exists: d.xpath('//*[@resource-id="com.ss.android.auto:id/kl2"]').click() elif d.xpath('//*[@resource-id="com.ss.android.auto:id/dib"]').exists: d.xpath('//*[@resource-id="com.ss.android.auto:id/dib"]').click() time.sleep(2) d.xpath('//*[@resource-id="com.ss.android.auto:id/iit"]').click() elif d.xpath('//*[@resource-id="com.ss.android.auto:id/die"]').exists: d.xpath('//*[@resource-id="com.ss.android.auto:id/die"]').click() else: d.xpath('//*[@resource-id="com.ss.android.auto:id/dvq"]').click() time.sleep(2) # 评论框赋值 d.xpath('//*[@resource-id="com.ss.android.auto:id/daj"]').set_text(reply) time.sleep(2) # 发布 d.xpath('//*[@resource-id="com.ss.android.auto:id/fx_"]').click() time.sleep(2) if d.xpath('//*[@text="发布评论"]').exists: d.xpath('//*[@text="发布评论"]').click() content_result = content_detail(keyword, comment_request_text, reply, account_name) status = shell_status_content(1, 0, "", '', json.dumps(content_result.to_dict(), ensure_ascii=False), account_name) return status # 机器人重复查找元素 针对网络不顺畅的情况 或手机性能问题 def retry(operation, device_serial, retries=11, retry_interval=3): for i in range(retries): loggerKit.info("懂车帝互动获取多次获取元素" + device_serial + "开始操作") if operation(): return True if i == retries - 1: return False time.sleep(retry_interval) # 等待2秒再重试 return False if __name__ == "__main__": # search_test('QXNUT21905001550','新车 售18.99万元，定位纯电中大型轿车，飞凡F7都市版正式上市') # while 1: # json_str = '''{ "data": [ { "taskId": 1163119, "mediaChannel": "dongchedi", "taskType": "testPOC", "taskSubType": "dongchediAppointInteraction", "taskSequenceId": "", "taskDesc": "懂车帝指定账号互动", "actionType": "content", "resourceName": "为何飞凡广州车展带来的两道“开胃菜”，让人越吃越“舒适”？", "subResourceName": null, "executeRobotAccount": "testRobotAccount", "executeRobotName": "testRobotAccount", "deviceId": "1", "content": "", "answerType": null, "materialUri": null, "spiderType": null, "sequence": null, "demoTask": false }, { "taskId": 1163119, "mediaChannel": "dongchedi", "taskType": "testPOC", "taskSubType": "dongchediAppointInteraction", "taskSequenceId": "", "taskDesc": "懂车帝指定账号互动", "actionType": "content", "resourceName": "为何飞凡广州车展带来的两道“开胃菜”，让人越吃越“舒适”？", "subResourceName": null, "executeRobotAccount": "testRobotAccount", "executeRobotName": "testRobotAccount", "deviceId": "1", "content": "", "answerType": null, "materialUri": null, "spiderType": null, "sequence": null, "demoTask": false } ], "code": "00000000", "message": "success" }''' data_list = json.loads(json_str).get('data') for data in data_list: print(data.get('taskId')) status = simulated_operation("7HX5T19911019170", "感受飞凡F7最快的一次充电", 199, 'dongchedi', 'Lightman317') print(json.dumps(status, default=lambda o: o.__dict__, indent=4, ensure_ascii=False))