robots-auto-action/douyin/douyin_live_comment.py

# 抖音直播间评论脚本
import random
import time

from func.action_func import del_key_vague
from task.task_job import callback_task
from tools import loggerKit, redis_client
from scene.oprator.atom_data import start_app, stop_app, get_text_by_control, single_click_by_control, \
    send_text_by_control, many_click_by_position, single_click_by_text

import yaml
import json

import httpx

with open('config.yaml', 'r') as file:
    config = yaml.load(file, Loader=yaml.FullLoader)

extern_domain = config['bmp-cp']['extern_domain']

# 任务执行回调url
task_callback_url = extern_domain + config['bmp-cp']['task_callback_url']

# 调用AIGC接口
post_ai_gc_url = extern_domain + config['bmp-content-center']['comment_local_url']


# 抖音互动 (根据视频链接 进行搜索获取到对应的视频进行点赞 收藏 评论)
# version 29.5.0
def douyin_spider(task_id, keyword, data):
    loggerKit.info('请求信息:{0}'.format(data))

    device_id = data.get("deviceID")
    perform_action_id = data.get("performActionId")
    result = data.get("result")
    extension_info = data['data']['extendInfo']
    extension_info = json.loads(extension_info)
    keyword = extension_info['liveURL']

    if result is not None:
        """
        非首个指令
        """
        perform_action_result = result.get("performActionResult")
        if perform_action_result is None:
            return_dict = {
                "data": "",
                "code": -2,
                "message": "fail, performActionResult is null"
            }

            # 回调任务中心

            del_key_vague(device_id)

            callback_task(500, '指令执行失败', task_id, device_id, 0, None)

            return return_dict

        if perform_action_result == "failure":
            # 回调任务中心
            return_dict = {
                "data": "",
                "code": -2,
                "message": "fail, performActionResult is not success"
            }

            del_key_vague(device_id)

            callback_task(500, '任务执行失败,可能当前无法再次执行看广告任务', task_id, device_id, 0, None)

            return return_dict

        if perform_action_result == "stop":
            # 回调任务中心
            return_dict = {
                "data": "",
                "code": -2,
                "message": "fail, performActionResult is not success"
            }

            del_key_vague(device_id)

            callback_task(500, '指令被用户终止', task_id, device_id, 0, None)

            return return_dict

        # 每次操作完成后会将对应的操作唯一id存储到redis，并且返回给手机端 手机端下次带着上个操作id来执行下一个操作
        last_action_id = redis_client.get(device_id + "operate")
        step0 = redis_client.get(f"{device_id}_step0")
        if (step0 is not None and int(step0) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id)
                and perform_action_result == "success"):
            action1_id = int(round(time.time() * 1000))
            """
            发送第1条指令
            点击：我(菜单)
            """

            action1_dict = single_click_by_text(action1_id, target_app="douyin", target_version="28.8.0",
                                                package_name="com.ss.android.ugc.aweme.lite",
                                                text="我")

            redis_client.set(device_id + "operate", action1_id)
            redis_client.set(f"{device_id}_step_me", "1")
            redis_client.delete(f"{device_id}_step0")

            loggerKit.info("taskId:{0}, action1_id:{1}", task_id, action1_id)

            return action1_dict

        # 获取到对应抖音账号并存入redis:
        step1 = redis_client.get(f"{device_id}_step_me")
        if step1 is not None and int(step1) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id):
            loggerKit.info("设备:{0}, last_action_id:{1}", device_id, int(last_action_id))
            action2_id = int(round(time.time() * 1000))
            """
            插播指令 获取当前的抖音账号
            获取当前抖音账号
            """
            action2_dict = get_text_by_control(action2_id,
                                               control_id="com.ss.android.ugc.aweme.lite:id/ib7",
                                               target_app="douyin",
                                               target_version="28.8.0",
                                               package_name="com.ss.android.ugc.aweme.lite",
                                               sleep_time=3)

            redis_client.set(device_id + "operate", action2_id)
            redis_client.set(f"{device_id}_step_get_account", "1")
            redis_client.delete(f"{device_id}_step_me")
            loggerKit.info("taskId:{0}, action2_id:{1}", task_id, action2_id)

            return action2_dict

        step1 = redis_client.get(f"{device_id}_step_get_account")
        if step1 is not None and int(step1) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id):
            loggerKit.info("设备:{0}, last_action_id:{1}", device_id, int(last_action_id))
            action2_id = int(round(time.time() * 1000))
            # 获取当前登陆账号并存入缓存
            account_name = result.get("performActionText")
            if account_name is None:
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "未获取到账号信息"
                }

                del_key_vague(device_id)
                callback_task(500, '未获取到账号信息', task_id, device_id, 0, None)

                return return_dict

            # 将当前设备操作的账号存入redis
            redis_client.set(device_id + 'account', account_name)

            """
            发送第2条指令
            点击：首页
            """
            action2_dict = single_click_by_text(action2_id, target_app="douyin", target_version="28.8.0",
                                                package_name="com.ss.android.ugc.aweme.lite",
                                                text="首页")

            redis_client.set(device_id + "operate", action2_id)
            redis_client.set(f"{device_id}_step_search", "1")
            redis_client.delete(f"{device_id}_step_get_account")
            loggerKit.info("taskId:{0}, action2_id:{1}", task_id, action2_id)

            return action2_dict

        step0 = redis_client.get(f"{device_id}_step_search")
        if step0 is not None and int(step0) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id):

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '回到首页失败', task_id, device_id, 0, None)
                return return_dict

            action1_id = int(round(time.time() * 1000))
            """
            发送第1条指令
            点击搜索框
            """

            action1_dict = single_click_by_control(action1_id, target_app="douyin", target_version="28.8.0",
                                                   package_name="com.ss.android.ugc.aweme.lite",
                                                   control_id="com.ss.android.ugc.aweme.lite:id/dxw", timeout=5)

            redis_client.set(device_id + "operate", action1_id)
            redis_client.set(f"{device_id}_step1", "1")
            redis_client.delete(f"{device_id}_step_search")

            loggerKit.info("设备:{0}, action1_id:{1}", device_id, action1_id)

            return action1_dict

        step1 = redis_client.get(f"{device_id}_step1")
        if step1 is not None and int(step1) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) and perform_action_result == "success":
            loggerKit.info("设备:{0}, action2_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '点击搜索按钮失败', task_id, device_id, 0, None)
                return return_dict
            action2_id = int(round(time.time() * 1000))
            """
            发送第2条指令
            对搜索框进行赋值
            """

            action2_dict = send_text_by_control(action2_id, target_app="douyin", target_version="28.8.0",
                                                package_name="com.ss.android.ugc.aweme.lite",
                                                control_id="com.ss.android.ugc.aweme.lite:id/et_search_kw",
                                                content=keyword, timeout=5)
            redis_client.set(device_id + "operate", action2_id)
            redis_client.set(f"{device_id}_step2", "1")
            redis_client.delete(f"{device_id}_step1")

            loggerKit.info("设备:{0}, action2_id:{1}", device_id, action2_id)

            return action2_dict

        step2 = redis_client.get(f"{device_id}_step2")
        if step2 is not None and int(step2) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) and perform_action_result == "success":
            loggerKit.info("设备:{0}, action3_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '搜索框赋值失败', task_id, device_id, 0, None)
                return return_dict

            action3_id = int(round(time.time() * 1000))
            """
            发送第3条指令
            点击搜索按钮
            """

            comment_count = random.randint(5, 8)
            redis_client.set(device_id + 'comment_count', comment_count)
            loggerKit.info("任务id:{0}, 评论次数:{1}", task_id, comment_count)

            action3_dict = single_click_by_control(action3_id, target_app="douyin", target_version="28.8.0",
                                                   package_name="com.ss.android.ugc.aweme.lite",
                                                   control_id="com.ss.android.ugc.aweme.lite:id/nou")
            redis_client.set(device_id + "operate", action3_id)
            redis_client.set(f"{device_id}_step3", "1")
            redis_client.delete(f"{device_id}_step2")

            loggerKit.info("设备:{0}, action3_id:{1}", device_id, action3_id)

            return action3_dict

        # 第四步 判断是否可分享 不可分享开始评论
        step3 = redis_client.get(f"{device_id}_step3")
        if step3 is not None and int(step3) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) and perform_action_result == "success":
            loggerKit.info("设备:{0}, action3_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '搜索框赋值失败', task_id, device_id, 0, None)
                return return_dict

            action4_id = int(round(time.time() * 1000))

            account_name = redis_client.get(device_id + 'account')

            # 判断该设备是否对该链接进行过转发
            done_link = redis_client.set_nx(account_name + keyword, 'done')
            if done_link:
                """
                发送第4条指令
                点击更多 开始分享
                """
                action6_dict = single_click_by_control(action4_id, target_app="douyin", target_version="28.8.0",
                                                       package_name="com.ss.android.ugc.aweme.lite",
                                                       control_id="m.l.live.plugin:id/more_toolbar_icon", timeout=5)
                redis_client.set(device_id + "operate", action4_id)
                redis_client.set(f"{device_id}_step4", "1")
                redis_client.delete(f"{device_id}_step3")

                loggerKit.info("设备:{0}, action7_id:{1}", device_id, action4_id)

                return action6_dict

            """
            发送第4条指令
            点开直播间对应的输入框
            """
            action6_dict = many_click_by_position(action4_id, target_app="douyin", target_version="28.8.0",
                                                  package_name="com.ss.android.ugc.aweme.lite",
                                                  count=2, timeout=5)
            redis_client.set(device_id + "operate", action4_id)
            redis_client.set(f"{device_id}_step6", "1")
            redis_client.delete(f"{device_id}_step3")

            loggerKit.info("设备:{0}, action7_id:{1}", device_id, action4_id)

            return action6_dict

        # 第5步
        # 点击分享
        step4 = redis_client.get(f"{device_id}_step4")
        if step4 is not None and int(step4) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) \
                and perform_action_result == "success":
            loggerKit.info("设备:{0}, action6_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '点击直播间右下角更多按钮失败', task_id, device_id, 0, None)
                return return_dict

            action5_id = int(round(time.time() * 1000))
            """
            发送第5条指令
            点击分享
            """
            action6_dict = single_click_by_text(action5_id, target_app="douyin", target_version="28.8.0",
                                                package_name="com.ss.android.ugc.aweme.lite",
                                                text="分享", timeout=5)
            redis_client.set(device_id + "operate", action5_id)
            redis_client.set(f"{device_id}_step5", "1")
            redis_client.delete(f"{device_id}_step4")

            loggerKit.info("设备:{0}, action7_id:{1}", device_id, action5_id)

            return action6_dict

        # 第6步
        # 点击复制链接
        step4 = redis_client.get(f"{device_id}_step5")
        if step4 is not None and int(step4) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) \
                and perform_action_result == "success":
            loggerKit.info("设备:{0}, action6_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '点击分享失败', task_id, device_id, 0, None)
                return return_dict

            action5_id = int(round(time.time() * 1000))
            """
            发送第5条指令
            点击分享
            """
            action6_dict = single_click_by_text(action5_id, target_app="douyin", target_version="28.8.0",
                                                package_name="com.ss.android.ugc.aweme.lite",
                                                text="复制链接", timeout=5)
            redis_client.set(device_id + "operate", action5_id)
            redis_client.set(f"{device_id}_step6", "1")
            redis_client.delete(f"{device_id}_step5")

            loggerKit.info("设备:{0}, action7_id:{1}", device_id, action5_id)

            return action6_dict

        # 第7步
        # 点击

        step6 = redis_client.get(f"{device_id}_step6")
        if step6 is not None and int(step6) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) \
                and perform_action_result == "success":
            loggerKit.info("设备:{0}, action6_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '点击搜索按钮/分享链接失败', task_id, device_id, 0, None)
                return return_dict

            action6_id = int(round(time.time() * 1000))
            """
            发送第7条指令
            点开直播间对应的输入框
            """
            action6_dict = single_click_by_control(action6_id, target_app="douyin", target_version="28.8.0",
                                                   package_name="com.ss.android.ugc.aweme.lite",
                                                   control_id="m.l.live.plugin:id/edit_btn_audience", timeout=5)
            redis_client.set(device_id + "operate", action6_id)
            redis_client.set(f"{device_id}_step8", "1")
            redis_client.delete(f"{device_id}_step6")

            loggerKit.info("设备:{0}, action7_id:{1}", device_id, action6_id)

            return action6_dict

        step8 = redis_client.get(f"{device_id}_step8")
        if step8 is not None and int(step8) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) \
                and perform_action_result == "success":
            loggerKit.info("设备:{0}, action6_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '唤醒评论输入框失败', task_id, device_id, 0, None)
                return return_dict

            action9_id = int(round(time.time() * 1000))
            """
            发送第9条指令
            根据视频内容 生成评论文案将文案缓存 并且点击评论
            如果没获取到对应的评论 任务结束 关闭
            """

            request_data = {
                "title": "",
                "comment": "",
                "platform": "douyin",
                "mode": "1"
            }

            loggerKit.info("任务id:{0},请求AIGC信息:{1}", task_id, request_data)
            response = httpx.post(post_ai_gc_url, json=request_data, timeout=120)

            if not response.is_success:
                #  调用AIGC获取评论失败
                loggerKit.info("任务id:{0}对应标题:{1},请求AIGC失败信息:{2},返回信息:{3}", task_id, keyword, request_data, response)

                reply = '不错👍'

            response_body = json.loads(response.text)

            reply_data = response_body.get('data')

            reply = reply_data.get('comment')

            if '内容太少' in reply or '对不起' in reply or reply is None or reply == '' or '提供' in reply:
                #  获取到的评论内容不符合
                loggerKit.info("任务id:{0}对应标题:{1},请求AIGC失败信息:{2},返回信息:{3}", task_id, keyword, request_data, response)

                reply = '不错👍'

            redis_client.set(device_id + "operate", action9_id)
            redis_client.set(f"{device_id}_step9", "1")
            redis_client.delete(f"{device_id}_step8")

            loggerKit.info("设备:{0}, action7_id:{1}", device_id, action9_id)

            # 评论框赋值
            action9_dict = send_text_by_control(action9_id, target_app="douyin", target_version="28.8.0",
                                                package_name="com.ss.android.ugc.aweme.lite",
                                                control_id="m.l.live.plugin:id/edit_text_container",
                                                content=reply, timeout=5, sleep_time=5, item_index=0)

            return action9_dict

        step9 = redis_client.get(f"{device_id}_step9")
        if step9 is not None and int(step9) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) \
                and perform_action_result == "success":
            loggerKit.info("设备:{0}, action9_id_mem:{1}", device_id, int(last_action_id))
            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '评论输入框赋值失败', task_id, device_id, 0, None)
                return return_dict

            action10_id = int(round(time.time() * 1000))
            """
            发送第10条指令
            发送评论
            """
            action10_dict = single_click_by_control(action10_id, target_app="douyin", target_version="28.8.0",
                                                    package_name="com.ss.android.ugc.aweme.lite",
                                                    control_id="m.l.live.plugin:id/input_panel_send_view", timeout=5,
                                                    sleep_time=5)
            redis_client.set(device_id + "operate", action10_id)
            redis_client.set(f"{device_id}_step10", "1")
            redis_client.delete(f"{device_id}_step9")

            loggerKit.info("设备:{0}, action10_id:{1}", device_id, action10_id)

            return action10_dict

        step10 = redis_client.get(f"{device_id}_step10")
        if step10 is not None and int(step10) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id) \
                and perform_action_result == "success":
            loggerKit.info("设备:{0}, action10_id_mem:{1}", device_id, int(last_action_id))

            # 元素未找到
            if perform_action_result == "invalid operation":
                # 回调任务中心
                return_dict = {
                    "data": "",
                    "code": -2,
                    "message": "fail, performActionResult is null"
                }

                del_key_vague(device_id)
                callback_task(500, '发送评论失败', task_id, device_id, 0, None)
                return return_dict

            action11_id = int(round(time.time() * 1000))
            """
            发送第11条指令
            点赞直播间
            """
            like_count = random.randint(10, 15)
            action11_dict = many_click_by_position(action11_id, target_app="douyin", target_version="28.8.0",
                                                   package_name="com.ss.android.ugc.aweme.lite",
                                                   count=like_count, timeout=5)
            redis_client.set(device_id + "operate", action11_id)
            redis_client.set(f"{device_id}_step11", "1")
            redis_client.delete(f"{device_id}_step10")

            loggerKit.info("设备:{0}, action10_id:{1}", device_id, action11_id)

            return action11_dict

        """
        停止app
        """
        step10 = redis_client.get(f"{device_id}_step11")
        if step10 is not None and int(step10) == 1 and last_action_id is not None and int(perform_action_id) == int(
                last_action_id):
            loggerKit.info("设备:{0}, action10_id_mem:{1}", device_id, int(last_action_id))

            # 已完成任务数
            done_comment_count = redis_client.incr(device_id + "done_count", 1)

            # 预期完成任务数
            anticipate_comment_count = redis_client.get(device_id + 'comment_count')

            action12_id = int(round(time.time() * 1000))

            # 如果已经达到预期评论次数 结束任务
            if int(done_comment_count) >= int(anticipate_comment_count):
                """
                停止指令
                停止app
                """

                action12_dict = stop_app(action12_id, target_app="douyin", target_version="28.8.0",
                                         package_name="com.ss.android.ugc.aweme.lite", timeout=5)

                del_key_vague(device_id)

                loggerKit.info("设备:{0}, action11_id:{1}", device_id, action12_id)

                # 回调任务中心修改任务状态
                callback_task(None, None, task_id, device_id, 1, None)

                return action12_dict

            # 未达到评论次数 重新开始评论
            """
            发送第7条指令
            点开直播间对应的输入框
            """
            action6_dict = single_click_by_control(action12_id, target_app="douyin", target_version="28.8.0",
                                                   package_name="com.ss.android.ugc.aweme.lite",
                                                   control_id="m.l.live.plugin:id/edit_btn_audience", timeout=5)
            redis_client.set(device_id + "operate", action12_id)
            redis_client.set(f"{device_id}_step8", "1")
            redis_client.delete(f"{device_id}_step11")

            loggerKit.info("设备:{0}, action7_id:{1}", device_id, action12_id)

            return action6_dict





    else:

        action0_id = int(round(time.time() * 1000))

        """
        启动指令
        启动app
        """

        action0_dict = start_app(action0_id, target_app="douyin", target_version="29.5.0",
                                 package_name="com.ss.android.ugc.aweme.lite")

        redis_client.set(device_id + "operate", action0_id)
        redis_client.set(f"{device_id}_step0", "1")
        redis_client.delete(f"{device_id}_step9")

        loggerKit.info("设备:{0}, action0_id:{1}", device_id, action0_id)

        return action0_dict

# 批量模糊删除缓存
# def del_key_vague(device_id):
#     # 批量模糊删除keys
#     keys = redis_client.match_pattern_prefix(device_id)
#     if len(keys) > 0:
#         # 需要判断是否有匹配的值, 没有的话会报错
#         for key in keys:
#             redis_client.delete(key)
#         loggerKit.info(f"clear {device_id} keys success...")
#     else:
#         loggerKit.info(f"{device_id} keys none ...")