piaoquan_crawler/kanyikan/kanyikan_recommend/kyk_recommend.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2023/6/21
"""
看一看推荐
207 服务器抓取模式
微信 session 从爬虫平台获取
"""

import json
import os
import random
import sys
import time
import requests
import urllib3
sys.path.append(os.getcwd())
from common.mq import MQ
from common.common import Common
from common.scheduling_db import MysqlHelper
from common.public import get_config_from_mysql, download_rule
proxies = {"http": None, "https": None}


class KanyikanRecommend:
    platform = "看一看"
    strategy = "推荐抓取策略"

    @classmethod
    def get_session(cls, log_type, crawler, env):
        session_sql = """ SELECT * FROM crawler_config WHERE `source` in ('kanyikan', '看一看'); """
        session_response = MysqlHelper.get_values(log_type, crawler, session_sql, env, action="")
        for config in session_response:
            if "token" in config["config"] and config["title"] == "看一看推荐":
                token_str = config["config"]
                token = json.loads(token_str)["token"]
                return token
        return None

    @classmethod
    def repeat_video(cls, log_type, crawler, video_id, env):
        sql = f""" select * from crawler_video where platform in ("{crawler}","{cls.platform}") and create_time>='2023-06-26' and out_video_id="{video_id}"; """
        repeat_video = MysqlHelper.get_values(log_type, crawler, sql, env)
        return len(repeat_video)

    @classmethod
    def get_videoList(cls, log_type, crawler, our_uid, rule_dict, env):
        mq = MQ(topic_name="topic_crawler_etl_" + env)
        try:
            Common.logger(log_type, crawler).info(f"正在抓取列表页")
            Common.logging(log_type, crawler, env, f"正在抓取列表页")
            session = cls.get_session(log_type, crawler, env)
            Common.logger(log_type, crawler).info(f"session:{session}")
            if session is None:
                Common.logger(log_type, crawler).info("session is None!")
                time.sleep(1)
                cls.get_videoList(log_type, crawler, our_uid, rule_dict, env)
            url = 'https://search.weixin.qq.com/cgi-bin/recwxa/recwxavideolist?'
            header = {
                "Connection": "keep-alive",
                "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.18(0x18001236) "
                              "NetType/WIFI Language/zh_CN",
                "Referer": "https://servicewechat.com/wxbb9a805eb4f9533c/234/page-frame.html",
            }
            params = {
                'session': session,
                "offset": 0,
                "wxaVersion": "3.9.2",
                "count": "10",
                "channelid": "208",
                "scene": '310',
                "subscene": '1089',
                "clientVersion": '8.0.18',
                "sharesearchid": '0',
                "nettype": 'wifi',
                "switchprofile": "0",
                "switchnewuser": "0",
            }
            urllib3.disable_warnings()
            response = requests.get(url=url, headers=header, params=params, proxies=proxies, verify=False)
            if "data" not in response.text:
                Common.logger(log_type, crawler).info("获取视频list时，session过期，随机睡眠 31-50 秒")
                Common.logging(log_type, crawler, env, "获取视频list时，session过期，随机睡眠 31-50 秒")
                # 如果返回空信息，则随机睡眠 31-40 秒
                time.sleep(random.randint(31, 40))
                cls.get_videoList(log_type, crawler, our_uid, rule_dict, env)
            elif "items" not in response.json()["data"]:
                Common.logger(log_type, crawler).info(f"get_feeds:{response.json()}，随机睡眠 1-3 分钟")
                Common.logging(log_type, crawler, env, f"get_feeds:{response.json()}，随机睡眠 1-3 分钟")
                # 如果返回空信息，则随机睡眠 1-3 分钟
                time.sleep(random.randint(60, 180))
                cls.get_videoList(log_type, crawler, our_uid, rule_dict, env)
            feeds = response.json().get("data", {}).get("items", "")
            if feeds == "":
                Common.logger(log_type, crawler).info(f"feeds:{feeds}")
                Common.logging(log_type, crawler, env, f"feeds:{feeds}")
                return
            for i in range(len(feeds)):
                try:
                    video_title = feeds[i].get("title", "").strip().replace("\n", "") \
                        .replace("/", "").replace("\\", "").replace("\r", "") \
                        .replace(":", "").replace("*", "").replace("？", "") \
                        .replace("?", "").replace('"', "").replace("<", "") \
                        .replace(">", "").replace("|", "").replace(" ", "") \
                        .replace("&NBSP", "").replace(".", "。").replace(" ", "") \
                        .replace("'", "").replace("#", "").replace("Merge", "")
                    publish_time_stamp = feeds[i].get("date", 0)
                    publish_time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(publish_time_stamp))
                    # 获取播放地址
                    if "videoInfo" not in feeds[i]:
                        video_url = ""
                    elif "mpInfo" in feeds[i]["videoInfo"]["videoCdnInfo"]:
                        if len(feeds[i]["videoInfo"]["videoCdnInfo"]["mpInfo"]["urlInfo"]) > 2:
                            video_url = feeds[i]["videoInfo"]["videoCdnInfo"]["mpInfo"]["urlInfo"][2]["url"]
                        else:
                            video_url = feeds[i]["videoInfo"]["videoCdnInfo"]["mpInfo"]["urlInfo"][0]["url"]
                    elif "ctnInfo" in feeds[i]["videoInfo"]["videoCdnInfo"]:
                        video_url = feeds[i]["videoInfo"]["videoCdnInfo"]["ctnInfo"]["urlInfo"][0]["url"]
                    else:
                        video_url = feeds[i]["videoInfo"]["videoCdnInfo"]["urlInfo"][0]["url"]
                    video_dict = {
                        "video_title": video_title,
                        "video_id": feeds[i].get("videoId", ""),
                        "play_cnt": feeds[i].get("playCount", 0),
                        "like_cnt": feeds[i].get("liked_cnt", 0),
                        "comment_cnt": feeds[i].get("comment_cnt", 0),
                        "share_cnt": feeds[i].get("shared_cnt", 0),
                        "duration": feeds[i].get("mediaDuration", 0),
                        "video_width": feeds[i].get("short_video_info", {}).get("width", 0),
                        "video_height": feeds[i].get("short_video_info", {}).get("height", 0),
                        "publish_time_stamp": publish_time_stamp,
                        "publish_time_str": publish_time_str,
                        "user_name": feeds[i].get("source", "").strip().replace("\n", ""),
                        "user_id": feeds[i].get("openid", ""),
                        "avatar_url": feeds[i].get("bizIcon", ""),
                        "cover_url": feeds[i].get("thumbUrl", ""),
                        "video_url": video_url,
                        "session": session,
                    }
                    for k, v in video_dict.items():
                        Common.logger(log_type, crawler).info(f"{k}:{v}")
                    Common.logging(log_type, crawler, env, f"video_dict:{video_dict}")

                    if video_dict["video_id"] == "" or video_dict["video_title"] == "" or video_dict["video_url"] == "":
                        Common.logger(log_type, crawler).info("无效视频\n")
                        Common.logging(log_type, crawler, env, "无效视频\n")
                    elif download_rule(log_type=log_type, crawler=crawler, video_dict=video_dict, rule_dict=rule_dict) is False:
                        Common.logger(log_type, crawler).info("不满足抓取规则\n")
                        Common.logging(log_type, crawler, env, "不满足抓取规则\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')
                        Common.logging(log_type, crawler, env, '已中过滤词\n')
                    elif cls.repeat_video(log_type, crawler, video_dict["video_id"], env) != 0:
                        Common.logger(log_type, crawler).info('视频已下载\n')
                        Common.logging(log_type, crawler, env, '视频已下载\n')
                    else:
                        video_dict["out_user_id"] = video_dict["user_id"]
                        video_dict["platform"] = crawler
                        video_dict["strategy"] = log_type
                        video_dict["out_video_id"] = video_dict["video_id"]
                        video_dict["width"] = video_dict["video_width"]
                        video_dict["height"] = video_dict["video_height"]
                        video_dict["crawler_rule"] = json.dumps(rule_dict)
                        video_dict["user_id"] = our_uid
                        video_dict["publish_time"] = video_dict["publish_time_str"]

                        mq.send_msg(video_dict)
                except Exception as e:
                    Common.logger(log_type, crawler).error(f"抓取单条视频异常:{e}\n")
                    Common.logging(log_type, crawler, env, f"抓取单条视频异常:{e}\n")
        except Exception as e:
            Common.logger(log_type, crawler).error(f"抓取列表页时异常:{e}\n")
            Common.logging(log_type, crawler, env, f"抓取列表页时异常:{e}\n")


if __name__ == "__main__":
    # print(get_config_from_mysql(log_type="recommend",
    #                             source="kanyikan",
    #                             env="dev",
    #                             text="filter",
    #                             action=""))
    KanyikanRecommend.get_session(log_type="recommend", crawler="kanyikan", env="dev")
    pass