优乐搞笑小视频，无法获取视频列表问题解决

common/__init__.py

@@ -1 +1,2 @@
-
+from .aliyun_log import AliyunLogger
+from .redirect_url import get_redirect_url

common/pipeline.py

@@ -133,6 +133,7 @@ class PiaoQuanPipeline:
                 message="重复的视频",
                 data=self.item,
             )
+            print("video_repeat")
             return False
         return True
 

common/redirect_url.py

@@ -0,0 +1,10 @@
+import requests
+
+
+def get_redirect_url(url):
+    res = requests.get(url, allow_redirects=False)
+    if res.status_code == 302 or res.status_code == 301:
+        new_url = res.headers['Location']
+        return new_url
+    else:
+        return url

youlegaoxiaoxiaoshipin/youlegaoxiaoxiaoshipin_main/run_ylgxxsp_recommend.py

@@ -96,12 +96,21 @@ def main(my_platform, mode, topic_name, group_id, env):
                     our_uid=our_uid,
                 )
                 for i in range(50):
-                    if YLGXXSP.download_count >= int(rule_dict.get("videos_cnt", {}).get("min", 10)):
-                        YLGXXSP.download_count = 0
-                        break
-                    else:
-                        YLGXXSP.get_videoList(page_id=i + 1)
-                        time.sleep(60)
+                    try:
+                        if YLGXXSP.download_count >= int(rule_dict.get("videos_cnt", {}).get("min", 10)):
+                            YLGXXSP.download_count = 0
+                            break
+                        else:
+                            YLGXXSP.get_videoList(page_id=i + 1)
+                            time.sleep(60)
+                    except:
+                        AliyunLogger.logging(
+                            code="2001",
+                            platform=my_platform,
+                            mode=mode,
+                            env=env,
+                            message="未在此页获取到视频列表"
+                        )
 
                 AliyunLogger.logging(
                     code="1004",

zhongmiaoyinxin/zhongmiaoyinxin_recommend/zhongmiaoyinxin_recommend_new.py

@@ -0,0 +1,231 @@
+# 众妙音信接入后台
+# @Author: luojunhui
+# @Time: 2023/10/26
+import json
+import os
+import sys
+import time
+from hashlib import md5
+
+from appium import webdriver
+from appium.webdriver.common.touch_action import TouchAction
+from appium.webdriver.extensions.android.nativekey import AndroidKey
+from appium.webdriver.webdriver import WebDriver
+from bs4 import BeautifulSoup
+from selenium.common.exceptions import NoSuchElementException
+from selenium.webdriver.common.by import By
+import multiprocessing
+
+sys.path.append(os.getcwd())
+from common.mq import MQ
+from common.aliyun_log import AliyunLogger
+from common.pipeline import PiaoQuanPipeline
+
+
+class ZMYXRecommend:
+    env = None
+    driver = None
+    log_type = None
+
+    def __init__(self, log_type, crawler, env, rule_dict, our_uid):
+        self.mq = None
+        self.platform = "众妙音信"
+        self.download_cnt = 0
+        self.element_list = []
+        self.count = 0
+        self.swipe_count = 0
+        self.log_type = log_type
+        self.crawler = crawler
+        self.env = env
+        self.rule_dict = rule_dict
+        self.our_uid = our_uid
+        if self.env == "dev":
+            chromedriverExecutable = "/Users/luojunhui/Downloads/chromedriver_V111/chromedriver"
+        else:
+            chromedriverExecutable = "/Users/luojunhui/Downloads/chromedriver_v111/chromedriver"
+        # 微信的配置文件
+        caps = {
+            "platformName": "Android",  # 手机操作系统 Android / iOS
+            "deviceName": "a0a65126",  # 连接的设备名（模拟器或真机），安卓可以随便写
+            # "udid": "emulator-5554",  # 指定 adb devices 中的哪一台设备
+            "platforVersion": "11",  # 手机对应的系统版本
+            "appPackage": "com.tencent.mm",  # 被测APP的包名，乐活圈 Android
+            "appActivity": ".ui.LauncherUI",  # 启动的Activity名
+            "autoGrantPermissions": "true",  # 让 appium 自动授权 base 权限，
+            # 如果 noReset 为 True，则该条不生效（该参数为 Android 独有），对应的值为 True 或 False
+            "unicodekeyboard": True,  # 使用自带输入法，输入中文时填True
+            "resetkeyboard": True,  # 执行完程序恢复原来输入法
+            "noReset": True,  # 不重置APP
+            "printPageSourceOnFailure": True,  # 找不到元素时，appium log 会完整记录当前页面的 pagesource
+            "newCommandTimeout": 6000,  # 初始等待时间
+            "automationName": "UiAutomator2",  # 使用引擎，默认为 Appium，
+            # 其中 Appium、UiAutomator2、Selendroid、Espresso 用于 Android，XCUITest 用于 iOS
+            "showChromedriverLog": True,
+            'enableWebviewDetailsCollection': True,
+            'setWebContentsDebuggingEnabled': True,
+            'recreateChromeDriverSessions': True,
+            'chromedriverExecutable': chromedriverExecutable,
+            "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"},
+            # "chromeOptions": {"androidProcess": "com.tencent.mm:tools"},
+            'browserName': ''
+        }
+        self.driver = webdriver.Remote("http://localhost:4723/wd/hub", caps)
+        self.driver.implicitly_wait(30)
+
+        for i in range(120):
+            try:
+                if self.driver.find_elements(By.ID, "com.tencent.mm:id/f2s"):
+                    print("启动微信成功")
+                    break
+                elif self.driver.find_element(By.ID, "com.android.systemui:id/dismiss_view"):
+                    print("发现并关闭系统下拉菜单")
+                    self.driver.find_element(By.ID, "com.android.system:id/dismiss_view").click()
+                else:
+                    pass
+            except NoSuchElementException:
+                time.sleep(1)
+
+        size = self.driver.get_window_size()
+        self.driver.swipe(int(size['width'] * 0.5), int(size['height'] * 0.2),
+                          int(size['width'] * 0.5), int(size['height'] * 0.8), 200)
+        time.sleep(1)
+        self.driver.find_elements(By.XPATH, '//*[@text="西瓜悦"]')[-1].click()
+        print("打开小程序成功")
+        time.sleep(5)
+        self.get_videoList()
+        time.sleep(100)
+        self.driver.quit()
+
+    def search_elements(self, xpath):
+        time.sleep(1)
+        windowHandles = self.driver.window_handles
+        for handle in windowHandles:
+            self.driver.switch_to.window(handle)
+            time.sleep(1)
+            try:
+                elements = self.driver.find_elements(By.XPATH, xpath)
+                if elements:
+                    return elements
+            except NoSuchElementException:
+                pass
+
+    def check_to_applet(self):
+        while True:
+            webview = self.driver.contexts
+            self.driver.switch_to.context(webview[1])
+            windowHandles = self.driver.window_handles
+            for handle in windowHandles:
+                self.driver.switch_to.window(handle)
+                time.sleep(1)
+                try:
+                    video_list = self.driver.find_element(By.XPATH, '//*[@class="index--navbar-list"]/*[1]')
+                    video_list.click()
+                    print("切换 webview 成功")
+                    return
+                except NoSuchElementException:
+                    time.sleep(1)
+            print("切换 webview 失败")
+            break
+
+    def swipe_up(self):
+        self.search_elements('//*[@class="list-list--list"]')
+        size = self.driver.get_window_size()
+        self.driver.swipe(int(size["width"] * 0.5), int(size["height"] * 0.8),
+                          int(size["width"] * 0.5), int(size["height"] * 0.442), 200)
+        self.swipe_count += 1
+
+    def close_ad(self):
+        window_size = self.driver.get_window_size()
+        TouchAction(self.driver).tap(x=int(window_size['width'] * 0.5), y=int(window_size['height'] * 0.1)).perform()
+
+    def get_videoList(self):
+        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)
+        self.driver.implicitly_wait(20)
+        print("关闭广告")
+        self.close_ad()
+        print("切换到 webview")
+        self.check_to_applet()
+        index = 0
+        while True:
+            if self.search_elements('//*[@id="scrollContainer"]') is None:
+                print("窗口已销毁")
+                return
+            print("获取视频列表")
+            video_elements = self.search_elements('//wx-view[@class="cover"]')
+            if video_elements is None:
+                print("视频列表为空列表")
+                return
+            video_element_temp = video_elements[index:]
+            if len(video_element_temp) == 0:
+                print("视频已经到底")
+                return
+            for i, video_element in enumerate(video_element_temp):
+                if video_element is None:
+                    return
+                self.download_cnt += 1
+                self.search_elements('//wx-view[@class="cover"]')
+                time.sleep(3)
+                self.driver.execute_script("arguments[0].scrollIntoView({block:'center',inline:'center'})",
+                                           video_element)
+                video_title = video_element.find_elements(By.XPATH, '//wx-view[@class="playImgs"]')[index + i].text
+                cover_url = video_element.find_elements(By.XPATH, '//wx-image[@class="coverImg"]')[
+                    index + i].get_attribute('src')
+                play_cnt = video_element.find_elements(By.XPATH, '//wx-image[@class="coverImg"]/span/*[2]')[
+                    index + i].text
+
+                if "万" in play_cnt:
+                    play_cnt = int(play_cnt.split("万")[0]) * 10000
+                out_video_id = md5(video_title.encode('utf8')).hexdigest()
+                video_dict = {
+                    'video_title': video_title,
+                    'video_id': out_video_id,
+                    'out_video_id': out_video_id,
+                    'play_cnt': play_cnt,
+                    'comment_cnt': 0,
+                    'like_cnt': 0,
+                    'share_cnt': 0,
+                    'publish_time_stamp': int(time.time()),
+                    'publish_time_str': time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()))),
+                    'update_time_stamp': int(time.time()),
+                    'user_name': "zhongmiaoyinxin",
+                    'user_id': "zhongmiaoyinxin",
+                    'avatar_url': cover_url,
+                    'cover_url': cover_url,
+                    'session': f"zhongmiaoyinxin-{int(time.time())}"
+                }
+                pipeline = PiaoQuanPipeline(
+                    platform=self.crawler,
+                    mode=self.log_type,
+                    item=video_dict,
+                    rule_dict=self.rule_dict,
+                    env=self.env
+                )
+                flag = pipeline.process_item()
+                if flag:
+                    print(video_dict)
+                else:
+                    print("被规则过滤")
+                # self.mq.send_msg(video_dict)
+                # print(video_dict)
+
+
+def run():
+    rule_dict1 = {}
+    ZMYXRecommend("recommend", "zhongmiaoyinxin", "dev", rule_dict1, 6267141)
+
+
+if __name__ == "__main__":
+    run()
+    # process = multiprocessing.Process(
+    #     target=run
+    # )
+    # process.start()
+    # while True:
+    #     if not process.is_alive():
+    #         print("正在重启")
+    #         process.terminate()
+    #         time.sleep(60)
+    #         os.system("adb forward --remove-all")
+    #         process = multiprocessing.Process(target=run)
+    #         process.start()
+    #     time.sleep(60)