robots-auto-action/scene/dongchedi/user_like_poc.py

import json
import time
import uiautomator2 as u2
from datetime import datetime, timedelta
import subprocess

from tools import loggerKit
from difflib import SequenceMatcher


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()

    # 点击到首页
    if d.xpath('//*[@text="首页"]').exists:
        d.xpath('//*[@text="首页"]').click()
    else:
        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('http://47.116.62.124:3000/comment-saic-pv', 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, "", '', '', 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))