# -*- coding: utf-8 -*- # @Author: wangkun # @Time: 2023/3/16 import json import os import random import shutil import sys import time import requests import urllib3 sys.path.append(os.getcwd()) from common.common import Common from common.feishu import Feishu from common.publish import Publish from common.public import get_config_from_mysql from common.scheduling_db import MysqlHelper proxies = {"http": None, "https": None} class XiaoniangaoPlay: platform = "小年糕" words = "abcdefghijklmnopqrstuvwxyz0123456789" uid = f"""{"".join(random.sample(words, 8))}-{"".join(random.sample(words, 4))}-{"".join(random.sample(words, 4))}-{"".join(random.sample(words, 4))}-{"".join(random.sample(words, 12))}""" token = "".join(random.sample(words, 32)) uid_token_dict = { "uid": uid, "token": token } # 生成 uid、token @classmethod def get_uid_token(cls): words = "abcdefghijklmnopqrstuvwxyz0123456789" uid = f"""{"".join(random.sample(words, 8))}-{"".join(random.sample(words, 4))}-{"".join(random.sample(words, 4))}-{"".join(random.sample(words, 4))}-{"".join(random.sample(words, 12))}""" token = "".join(random.sample(words, 32)) uid_token_dict = { "uid": uid, "token": token } return uid_token_dict # 基础门槛规则 @classmethod def download_rule(cls, video_dict): """ 下载视频的基本规则 :param video_dict: 视频信息，字典格式 :return: 满足规则，返回 True；反之，返回 False """ # 视频时长 if int(float(video_dict['duration'])) >= 40: # 宽或高 if int(video_dict['video_width']) >= 0 or int(video_dict['video_height']) >= 0: # 播放量 if int(video_dict['play_cnt']) >= 20000: # 点赞量 if int(video_dict['like_cnt']) >= 0: # 分享量 if int(video_dict['share_cnt']) >= 0: # 发布时间 <= 60 天 if int(time.time()) - int(video_dict['publish_time_stamp']) <= 3600 * 24 * 60: return True else: return False else: return False else: return False else: return False return False return False # 获取表情及符号 @classmethod def get_expression(cls): # 表情列表 expression_list = ['📍', '⭕️', '🔥', '📣', '🎈', '⚡', '🔔', '🚩', '💢', '💎', '👉', '💓', '❗️', '🔴', '🔺', '♦️', '♥️', '👉', '👈', '🏆', '❤️\u200d🔥'] # 符号列表 char_list = ['...', '~~'] return expression_list, char_list # 获取列表 @classmethod def get_videoList(cls, log_type, crawler, strategy, oss_endpoint, env): uid_token_dict = cls.uid_token_dict url = "https://kapi.xiaoniangao.cn/trends/get_recommend_trends" headers = { "x-b3-traceid": '1dc0a6d0929a2b', "X-Token-Id": 'ae99a4953804085ebb0ae36fa138031d-1146052582', "uid": uid_token_dict['uid'], "content-type": "application/json", "Accept-Encoding": "gzip,compress,br,deflate", "User-Agent": 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_7_1 like Mac OS X)' ' AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 ' 'MicroMessenger/8.0.20(0x18001432) NetType/WIFI Language/zh_CN', "Referer": 'https://servicewechat.com/wxd7911e4c177690e4/620/page-frame.html' } data = { "log_params": { "page": "discover_rec", "common": { "brand": "iPhone", "device": "iPhone 11", "os": "iOS 14.7.1", "weixinver": "8.0.20", "srcver": "2.24.2", "net": "wifi", "scene": 1089 } }, "qs": "imageMogr2/gravity/center/rotate/$/thumbnail/!750x500r/crop/750x500/interlace/1/format/jpg", "h_qs": "imageMogr2/gravity/center/rotate/$/thumbnail/!80x80r/crop/80x80/interlace/1/format/jpg", "share_width": 625, "share_height": 500, "ext": { "fmid": 0, "items": {} }, "app": "xng", "rec_scene": "discover_rec", "log_common_params": { "e": [{ "data": { "page": "discoverIndexPage", "topic": "recommend" }, "ab": {} }], "ext": { "brand": "iPhone", "device": "iPhone 11", "os": "iOS 14.7.1", "weixinver": "8.0.20", "srcver": "2.24.3", "net": "wifi", "scene": "1089" }, "pj": "1", "pf": "2", "session_id": "7bcce313-b57d-4305-8d14-6ebd9a1bad29" }, "refresh": False, "token": uid_token_dict['token'], "uid": uid_token_dict['uid'], "proj": "ma", "wx_ver": "8.0.20", "code_ver": "3.62.0" } urllib3.disable_warnings() r = requests.post(url=url, headers=headers, json=data, proxies=proxies, verify=False) if "data" not in r.text or r.status_code != 200: Common.logger(log_type, crawler).warning(f"get_videoList:{r.text}") return elif "data" not in r.json(): Common.logger(log_type, crawler).info(f"get_videoList:{r.json()}") return elif "list" not in r.json()["data"]: Common.logger(log_type, crawler).warning(f"get_videoList:{r.json()['data']}") return elif len(r.json()["data"]["list"]) == 0: Common.logger(log_type, crawler).warning(f"get_videoList:{r.json()['data']['list']}") return else: # 视频列表数据 feeds = r.json()["data"]["list"] for i in range(len(feeds)): # 标题，表情随机加在片头、片尾，或替代句子中间的标点符号 if "title" in feeds[i]: befor_video_title = feeds[i]["title"].strip().replace("\n", "") \ .replace("/", "").replace("\r", "").replace("#", "") \ .replace(".", "。").replace("\\", "").replace("&NBSP", "") \ .replace(":", "").replace("*", "").replace("？", "") \ .replace("?", "").replace('"', "").replace("<", "") \ .replace(">", "").replace("|", "").replace(" ", "").replace("#表情", "").replace("#符号", "").replace('"' ,'').replace("'", '') expression = cls.get_expression() expression_list = expression[0] char_list = expression[1] # 随机取一个表情 expression = random.choice(expression_list) # 生成标题list[表情+title, title+表情] expression_title_list = [expression + befor_video_title, befor_video_title + expression] # 从标题list中随机取一个标题 title_list1 = random.choice(expression_title_list) # 生成标题:原标题+符号 title_list2 = befor_video_title + random.choice(char_list) # 表情和标题组合，与标题和符号组合，汇总成待使用的标题列表 title_list4 = [title_list2, title_list1] # 最终标题 video_title = random.choice(title_list4) else: video_title = 0 # 视频 ID if "vid" in feeds[i]: video_id = feeds[i]["vid"] else: video_id = 0 # 播放量 if "play_pv" in feeds[i]: video_play_cnt = feeds[i]["play_pv"] else: video_play_cnt = 0 # 评论量 if "comment_count" in feeds[i]: video_comment_cnt = feeds[i]["comment_count"] else: video_comment_cnt = 0 # 点赞量 if "favor" in feeds[i]: video_like_cnt = feeds[i]["favor"]["total"] else: video_like_cnt = 0 # 分享量 if "share" in feeds[i]: video_share_cnt = feeds[i]["share"] else: video_share_cnt = 0 # 时长 if "du" in feeds[i]: video_duration = int(feeds[i]["du"] / 1000) else: video_duration = 0 # 宽和高 if "w" or "h" in feeds[i]: video_width = feeds[i]["w"] video_height = feeds[i]["h"] else: video_width = 0 video_height = 0 # 发布时间 if "t" in feeds[i]: video_send_time = feeds[i]["t"] else: video_send_time = 0 publish_time_stamp = int(int(video_send_time)/1000) publish_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(publish_time_stamp)) # 用户名 / 头像 if "user" in feeds[i]: user_name = feeds[i]["user"]["nick"].strip().replace("\n", "") \ .replace("/", "").replace("快手", "").replace(" ", "") \ .replace(" ", "").replace("&NBSP", "").replace("\r", "") head_url = feeds[i]["user"]["hurl"] else: user_name = 0 head_url = 0 # 用户 ID profile_id = feeds[i]["id"] # 用户 mid profile_mid = feeds[i]["user"]["mid"] # 视频封面 if "url" in feeds[i]: cover_url = feeds[i]["url"] else: cover_url = 0 # 视频播放地址 if "v_url" in feeds[i]: video_url = feeds[i]["v_url"] else: video_url = 0 video_dict = { "video_title": video_title, "video_id": video_id, "duration": video_duration, "play_cnt": video_play_cnt, "like_cnt": video_like_cnt, "comment_cnt": video_comment_cnt, "share_cnt": video_share_cnt, "user_name": user_name, "publish_time_stamp": publish_time_stamp, "publish_time_str": publish_time_str, "video_width": video_width, "video_height": video_height, "avatar_url": head_url, "profile_id": profile_id, "profile_mid": profile_mid, "cover_url": cover_url, "video_url": video_url, "session": f"xiaoniangao-play-{int(time.time())}" } for k, v in video_dict.items(): Common.logger(log_type, crawler).info(f"{k}:{v}") cls.download_publish(log_type=log_type, crawler=crawler, video_dict=video_dict, strategy=strategy, oss_endpoint=oss_endpoint, env=env) @classmethod def repeat_video(cls, log_type, crawler, video_id, env): sql = f""" select * from crawler_video where platform="小年糕" and out_video_id="{video_id}"; """ repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env) return len(repeat_video) @classmethod def download_publish(cls, log_type, crawler, video_dict, strategy, oss_endpoint, env): # 过滤无效视频 if video_dict["video_id"] == 0 \ or video_dict["video_url"] == 0\ or video_dict["cover_url"] == 0: Common.logger(log_type, crawler).warning("无效视频\n") # 抓取规则 elif cls.download_rule(video_dict) is False: Common.logger(log_type, crawler).info("不满足抓取规则\n") # 去重 elif cls.repeat_video(log_type, crawler, video_dict['video_id'], env) != 0: Common.logger(log_type, crawler).info("视频已下载\n") elif any(str(word) if str(word) in video_dict['video_title'] else False for word in get_config_from_mysql(log_type=log_type, source=crawler, env=env, text="filter", action="")) is True: Common.logger(log_type, crawler).info("视频已中过滤词\n") else: # 下载封面 Common.download_method(log_type=log_type, crawler=crawler, text="cover", title=video_dict["video_title"], url=video_dict["cover_url"]) # 下载视频 Common.download_method(log_type=log_type, crawler=crawler, text="video", title=video_dict["video_title"], url=video_dict["video_url"]) # 保存视频信息至 "./videos/{download_video_title}/info.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=strategy, our_uid="play", env=env, oss_endpoint=oss_endpoint) 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 # 视频信息保存数据库 rule_dict = { "duration": {"min": 40}, "play_cnt": {"min": 80000}, "min_publish_day": {"min": 60} } 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['profile_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) Common.logger(log_type, crawler).info('视频信息插入数据库成功！') # 视频写入飞书 Feishu.insert_columns(log_type, crawler, "c85k1C", "ROWS", 1, 2) # 视频ID工作表，首行写入数据 upload_time = int(time.time()) values = [[time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(upload_time)), "播放量榜爬虫策略", str(video_dict['video_id']), str(video_dict['video_title']), our_video_link, video_dict['play_cnt'], video_dict['comment_cnt'], video_dict['like_cnt'], video_dict['share_cnt'], video_dict['duration'], f"{video_dict['video_width']}*{video_dict['video_height']}", str(video_dict['publish_time_str']), str(video_dict['user_name']), str(video_dict['profile_id']), str(video_dict['profile_mid']), str(video_dict['avatar_url']), str(video_dict['cover_url']), str(video_dict['video_url'])]] time.sleep(1) Feishu.update_values(log_type, crawler, "c85k1C", "F2:Z2", values) Common.logger(log_type, crawler).info('视频信息写入飞书成功\n') if __name__ == '__main__': XiaoniangaoPlay.get_videoList("play", "xiaoniangao", "播放量榜爬虫策略", "out", "dev") pass