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

import json
import time
import uiautomator2 as u2
import httpx
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):
        """

        :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

    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
        }


# 回复详情
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
# 签到操作

def simulated_operation(device_serial, keyword, task_id, media_channel):
    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()

    # 开始后台监控
    # 默认监控间隔2.0s
    d.watcher.start(1.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}", device_serial,
                    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()

    target_text = '签到'

    xpath = f'//*[contains(@text, "{target_text}")]'

    if d.xpath(xpath).exists:
        print("找到了")
        d.xpath(xpath).click()

    if d.xpath('//*[@text="签到"]').exists:
        d.xpath('//*[@text="签到"]').click()

    status = shell_status_content(1, 0, '', '', '')
    return status


# 查询到用户主页寻找帖子
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)

    # 获取帖子发布时间
    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')

    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)

    # 点击对话框
    d.xpath('//*[@resource-id="com.ss.android.auto:id/kl2"]').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))

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

    loggerKit.info(json.dumps(status, default=lambda o: o.__dict__, indent=4, ensure_ascii=False))