# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2023/9/13 import json import os import random import re import shutil import sys import time from hashlib import md5 from appium import webdriver from appium.webdriver.extensions.android.nativekey import AndroidKey from appium.webdriver.webdriver import WebDriver from bs4 import BeautifulSoup from selenium.common import NoSuchElementException from selenium.webdriver.common.by import By sys.path.append(os.getcwd()) from common.common import Common from common.feishu import Feishu from common.publish import Publish from common.scheduling_db import MysqlHelper class HTZFRecommend: platform = "海豚祝福" element_list = [] @classmethod def today_download_cnt(cls, log_type, crawler, env): select_sql = """ SELECT COUNT(*) FROM crawler_video WHERE platform IN ("haitunzhufu", "海豚祝福") AND DATE(create_time) = CURDATE(); """ today_download_cnt = MysqlHelper.get_values(log_type, crawler, select_sql, env, action="")[0]['COUNT(*)'] return today_download_cnt @classmethod def start_wechat(cls, log_type, crawler, videos_cnt, env): if env == "dev": chromedriverExecutable = "/Users/wangkun/Downloads/chromedriver/chromedriver_v111/chromedriver" else: chromedriverExecutable = "/Users/piaoquan/Downloads/chromedriver" Common.logger(log_type, crawler).info("启动微信") caps = { "platformName": "Android", "platformVersion": "11", "devicesName": "Android", "appPackage": "com.tencent.mm", "appActivity": ".ui.LauncherUI", "noReset": True, "resetkeyboard": True, "unicodekeyboard": True, "showChromedriverLog": True, "autoGrantPermissions": True, "printPageSourceOnFailure": True, "recreateChromeDriverSessions": True, "enableWebviewDetailsCollention": True, "newCommandTimeout": 6000, "automationName": "UiAutomator2", "chromedriverExecutable": chromedriverExecutable, "chromeOptions": {"androidProcess": "com.tencent.mm:appbrand0"}, } driver = webdriver.Remote("http://localhost:4723/wd/hub", caps) driver.implicitly_wait(20) for i in range(120): try: if driver.find_element(By.ID, "com.tencent.mm:id/f2s"): break elif driver.find_element(By.ID, "com.android.system:id/dismiss_view"): Common.logger(log_type, crawler).info("发现并关闭系统下拉菜单栏") else: pass except NoSuchElementException: pass Common.logger(log_type, crawler).info("下滑，展示小程序选择面板") size = driver.get_window_size() 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(3) Common.logger(log_type, crawler).info('打开小程序"海豚祝福"') driver.find_elements(By.XPATH, '//*[@text="海豚祝福"]')[-1].click() time.sleep(5) cls.get_videoList(log_type=log_type, crawler=crawler, driver=driver, videos_cnt=videos_cnt, env=env) time.sleep(1) driver.quit() @classmethod def search_elements(cls, driver: WebDriver, xpath): time.sleep(1) windowHandles = driver.window_handles for handle in windowHandles: driver.switch_to.window(handle) time.sleep(1) try: elements = driver.find_elements(By.XPATH, xpath) if elements: return elements except NoSuchElementException: pass @classmethod def repeat_out_video_id(cls, log_type, crawler, out_video_id, env): sql = f""" select * from crawler_video where platform in ("众妙音信", "刚刚都传", "吉祥幸福", "知青天天看", "zhufuquanzi", "祝福圈子", "haitunzhufu", "海豚祝福") and out_video_id="{out_video_id}"; """ repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env) return len(repeat_video) @classmethod def get_video_url(cls, driver: WebDriver, video_title_element): for i in range(3): cls.search_elements(driver, '//*[@class="list"]') video_title_element[0].click() time.sleep(5) video_url_elements = cls.search_elements(driver, '//*[@id="myVideo"]') if video_url_elements: return video_url_elements[0].get_attribute("src") @classmethod def swipe_up(cls, driver: WebDriver): cls.search_elements(driver, '//*[@class="list"]') size = driver.get_window_size() driver.swipe(int(size["width"] * 0.5), int(size["height"] * 0.8), int(size["width"] * 0.5), int(size["height"] * 0.55), 200) @classmethod def get_videoList(cls, log_type, crawler, driver: WebDriver, videos_cnt, env): driver.implicitly_wait(20) webviews = driver.contexts Common.logger(log_type, crawler).info(f"webviews:{webviews}") driver.switch_to.context(webviews[1]) windowHandles = driver.window_handles for handle in windowHandles: driver.switch_to.window(handle) time.sleep(1) try: if cls.search_elements(driver, '//*[@class="bottom_scroll"]'): Common.logger(log_type, crawler).info("切换到小程序") break except NoSuchElementException: time.sleep(1) cls.search_elements(driver, '//*[@class="nav cur"]')[-1].click() Common.logger(log_type, crawler).info('点击"推荐"列表成功\n') # while True: for page in range(500): Common.logger(log_type, crawler).info(f"正在抓取第{page+1}页") if cls.search_elements(driver, '//*[@class="list"]') is None: Common.logger(log_type, crawler).info("列表页窗口已销毁\n") return for i in range(1): cls.swipe_up(driver) time.sleep(0.5) page_source = driver.page_source soup = BeautifulSoup(page_source, 'html.parser') soup.prettify() video_list_elements = soup.findAll("wx-view", class_="img_bf") # video_list_elements 有，cls.element_list 中没有的元素 video_list_elements = list(set(video_list_elements).difference(set(cls.element_list))) # video_list_elements 与 cls.element_list 的并集 cls.element_list = list(set(video_list_elements) | set(cls.element_list)) Common.logger(log_type, crawler).info(f"第{page+1}页共:{len(video_list_elements)}条视频\n") if len(video_list_elements) == 0: for i in range(10): Common.logger(log_type, crawler).info(f"向上滑动第{i + 1}次") cls.swipe_up(driver) time.sleep(0.5) continue for i, video_element in enumerate(video_list_elements): try: today_download = cls.today_download_cnt(log_type, crawler, env) if today_download >= videos_cnt: Common.logger(log_type, crawler).info(f"今日已下载视频数:{today_download}") return Common.logger(log_type, crawler).info(f"第{i+1}条视频") video_title = video_element.find("wx-view", class_="title").text play_str = video_element.find("wx-view", class_="wan").text play_cnt = int(re.sub(r"\D", "", play_str)) * 10000 if "万" in play_str else play_str cover_url = video_element.find("wx-image", class_="img")["src"] out_video_id = md5(video_title.encode('utf8')).hexdigest() video_dict = { "video_title": video_title, 'video_id': out_video_id, "plat_cnt_str": play_str, "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()))), 'user_name': "haitunzhufu", 'user_id': "haitunzhufu", "cover_url": cover_url, 'avatar_url': cover_url, 'session': f"haitunzhufu-{int(time.time())}" } for k, v in video_dict.items(): Common.logger(log_type, crawler).info(f"{k}:{v}") if video_title is None or cover_url is None: Common.logger(log_type, crawler).info("无效视频\n") cls.swipe_up(driver) time.sleep(1) elif cls.repeat_out_video_id(log_type=log_type, crawler=crawler, out_video_id=out_video_id, env=env) != 0: Common.logger(log_type, crawler).info('视频已下载\n') cls.swipe_up(driver) time.sleep(1) else: video_title_element = cls.search_elements(driver, f'//*[contains(text(), "{video_title}")]') if video_title_element is None: Common.logger(log_type, crawler).warning(f"未找到该视频标题的element:{video_title_element}") continue Common.logger(log_type, crawler).info("点击标题，进入视频详情页") video_url = cls.get_video_url(driver, video_title_element) if video_url is None: Common.logger(log_type, crawler).info("未获取到视频播放地址\n") driver.press_keycode(AndroidKey.BACK) time.sleep(3) continue video_dict['video_url'] = video_url Common.logger(log_type, crawler).info(f"video_url:{video_url}\n") cls.download_publish(log_type=log_type, crawler=crawler, video_dict=video_dict, env=env) driver.press_keycode(AndroidKey.BACK) time.sleep(3) except Exception as e: Common.logger(log_type, crawler).error(f"抓取单条视频异常:{e}\n") Common.logger(log_type, crawler).info('已抓取完一组视频，休眠5秒\n') time.sleep(5) @classmethod def get_our_uid(cls, log_type, crawler, env): select_sql = f""" SELECT uid FROM crawler_user_v3 WHERE `source`="{crawler}"; """ uids = MysqlHelper.get_values(log_type, crawler, select_sql, env, action="") uid_list = [] for uid_dict in uids: uid_list.append(uid_dict["uid"]) return random.choice(uid_list) @classmethod def download_publish(cls, log_type, crawler, video_dict, env): Common.download_method(log_type=log_type, crawler=crawler, text='video', title=video_dict['video_title'], url=video_dict['video_url']) ffmpeg_dict = Common.ffmpeg(log_type, crawler, f"./{crawler}/videos/{video_dict['video_title']}/video.mp4") if ffmpeg_dict is None: md_title = md5(video_dict['video_title'].encode('utf8')).hexdigest() shutil.rmtree(f"./{crawler}/videos/{md_title}/") Common.logger(log_type, crawler).info("视频size=0，删除成功\n") return video_dict["duration"] = ffmpeg_dict["duration"] video_dict["video_width"] = ffmpeg_dict["width"] video_dict["video_height"] = ffmpeg_dict["height"] # 下载封面 Common.download_method(log_type=log_type, crawler=crawler, text='cover', title=video_dict['video_title'], url=video_dict['cover_url']) # 保存视频信息至txt Common.save_video_info(log_type=log_type, crawler=crawler, video_dict=video_dict) # 上传视频 Common.logger(log_type, crawler).info("开始上传视频...") our_video_id = Publish.upload_and_publish(log_type=log_type, crawler=crawler, strategy="推荐榜爬虫策略", our_uid=cls.get_our_uid(log_type, crawler, env), env=env, oss_endpoint="out") if env == 'dev': our_video_link = f"https://testadmin.piaoquantv.com/cms/post-detail/{our_video_id}/info" else: our_video_link = f"https://admin.piaoquantv.com/cms/post-detail/{our_video_id}/info" Common.logger(log_type, crawler).info("视频上传完成") if our_video_id is None: # 删除视频文件夹 shutil.rmtree(f"./{crawler}/videos/{video_dict['video_title']}") return # 视频信息保存至飞书 Feishu.insert_columns(log_type, crawler, "d51d20", "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(upload_time)), "推荐榜爬虫策略", video_dict["video_title"], video_dict["video_id"], video_dict["play_cnt"], video_dict["duration"], f'{video_dict["video_width"]}*{video_dict["video_height"]}', our_video_link, video_dict["cover_url"], video_dict["video_url"]]] time.sleep(1) Feishu.update_values(log_type, crawler, "d51d20", "F2:V2", values) Common.logger(log_type, crawler).info(f"视频已保存至飞书文档\n") rule_dict = {} # 视频信息保存数据库 insert_sql = f""" insert into crawler_video(video_id, out_user_id, platform, strategy, out_video_id, video_title, cover_url, video_url, duration, publish_time, play_cnt, crawler_rule, width, height) values({our_video_id}, "{video_dict['user_id']}", "{cls.platform}", "推荐榜爬虫策略", "{video_dict['video_id']}", "{video_dict['video_title']}", "{video_dict['cover_url']}", "{video_dict['video_url']}", {int(video_dict['duration'])}, "{video_dict['publish_time_str']}", {int(video_dict['play_cnt'])}, '{json.dumps(rule_dict)}', {int(video_dict['video_width'])}, {int(video_dict['video_height'])}) """ Common.logger(log_type, crawler).info(f"insert_sql:{insert_sql}") MysqlHelper.update_values(log_type, crawler, insert_sql, env, action='') Common.logger(log_type, crawler).info('视频信息写入数据库成功！\n') if __name__ == "__main__": HTZFRecommend.start_wechat("recommend", "haitunzhufu", 5, "dev") # HTZFRecommend.today_download_cnt("recommend", "haitunzhufu", "dev") # HTZFRecommend.get_play_cnt() pass