piaoquan_crawler/shipinhao/shipinhao_author/shipinhao_scheduling.py

import os
import json
import random
import sys
import datetime
import time
import uuid

import requests

sys.path.append(os.getcwd())
from common import PiaoQuanPipeline, AliyunLogger
from common.feishu import Feishu
from common.db import MysqlHelper
from common.mq import MQ
from common.public import clean_title


def find_target_user(name, user_list):
    """
    在搜索到到账号列表中找目标列表
    """
    for obj in user_list:
        if obj["nickname"] == name:
            return obj
        else:
            continue
    return False


class ShiPinHaoAccount(object):
    """
    视频号账号爬虫
    """
    def __init__(self, platform, mode, rule_dict, user_dict, env):
        self.cookie = None
        self.token = None
        self.account_name = user_dict["link"]
        self.platform = platform
        self.mode = mode
        self.rule_dict = rule_dict
        self.user_dict = user_dict
        self.env = env
        self.download_cnt = 0
        self.mq = MQ(topic_name="topic_crawler_etl_" + self.env)

    def get_token_from_mysql(self):
        """
        从mysql中读取token和cookie
        """
        select_sql = (
            f"""SELECT config from crawler_config where source = '{self.platform}'; """
        )
        # print(select_sql)
        configs = MysqlHelper.get_values(
            log_type=self.mode,
            crawler=self.platform,
            sql=select_sql,
            env=self.env,
            machine="",
        )
        token_config = configs[0][0]
        token_info = json.loads(token_config)
        self.token = token_info["token"]
        self.cookie = token_info["cookie"]

    def get_history_id(self):
        """
        从数据库表中读取 id
        """
        select_user_sql = f"""select name_id from accounts where name = "{self.account_name}" and platform = "{self.platform}" and useful = 1 limit 1"""
        name_id = MysqlHelper.get_values(
            log_type=self.mode,
            crawler=self.platform,
            sql=select_user_sql,
            env=self.env,
            machine="",
        )
        if name_id:
            return name_id[0][0]
        else:
            return False

    def get_account_id(self):
        """
        读历史数据，如果存在 id，则直接返回 id
        """
        history_id = self.get_history_id()
        if history_id:
            return history_id
        else:
            url = "https://mp.weixin.qq.com/cgi-bin/videosnap"
            self.get_token_from_mysql()
            params = {
                "action": "search",
                "scene": "1",
                "buffer": "",
                "query": self.account_name,
                "count": "21",
                "token": self.token,
                "lang": "zh_CN",
                "f": "json",
                "ajax": "1",
            }
            headers = {
                "authority": "mp.weixin.qq.com",
                "accept": "*/*",
                "accept-language": "en,zh-CN;q=0.9,zh;q=0.8",
                "cookie": self.cookie,
                "referer": "https://mp.weixin.qq.com/cgi-bin/appmsg?t=media/appmsg_edit_v2&action=edit&isNew=1&type=77&createType=0&token={}&lang=zh_CN".format(
                    self.token
                ),
                "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36",
                "x-requested-with": "XMLHttpRequest",
            }
            response = requests.request("GET", url, headers=headers, params=params)
            user_list = response.json()["acct_list"]
            target_user = find_target_user(name=self.account_name, user_list=user_list)
            # 写入 MySql 数据库
            if target_user:
                update_sql = f"""INSERT INTO accounts (name, name_id, platform, useful) values ("{self.account_name}", "{target_user['username']}", "{self.platform}", 1 )"""
                # print(update_sql)
                MysqlHelper.update_values(
                    log_type=self.mode,
                    crawler=self.platform,
                    sql=update_sql,
                    env=self.env,
                    machine="",
                )
                return target_user["username"]
            else:
                return False

    def get_account_videos(self):
        """
        # 一个账号最多抓取 30 条数据
        """
        user_id = self.get_account_id()
        if user_id:
            url = "https://mp.weixin.qq.com/cgi-bin/videosnap"
            buffer = ""  # 翻页指示器
            while True:
                if self.download_cnt >= int(
                    self.rule_dict.get("videos_cnt", {}).get("min", 30)
                ):
                    return
                self.get_token_from_mysql()
                headers = {
                    "authority": "mp.weixin.qq.com",
                    "accept": "*/*",
                    "accept-language": "en,zh-CN;q=0.9,zh;q=0.8",
                    "cookie": self.cookie,
                    "referer": "https://mp.weixin.qq.com/cgi-bin/appmsg?t=media/appmsg_edit_v2&action=edit&isNew=1&type=77&createType=0&token={}&lang=zh_CN".format(
                        self.token
                    ),
                    "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36",
                    "x-requested-with": "XMLHttpRequest",
                }
                params = {
                    "action": "get_feed_list",
                    "username": user_id,
                    "buffer": buffer,
                    "count": "15",
                    "scene": "1",
                    "token": self.token,
                    "lang": "zh_CN",
                    "f": "json",
                    "ajax": "1",
                }
                response = requests.request("GET", url, headers=headers, params=params)
                time.sleep(random.randint(10, 30))
                res_json = response.json()
                # 开始判断视频是否有信息，是否频控
                if res_json["base_resp"]["err_msg"] == "invalid session":
                    AliyunLogger.logging(
                        code="2000",
                        platform=self.platform,
                        mode=self.mode,
                        env=self.env,
                        message=f"status_code:{response.status_code}, get_videoList:{response.text}\n",
                    )
                    if 20 >= datetime.datetime.now().hour >= 10:
                        Feishu.bot(
                            log_type=self.mode,
                            crawler=self.platform,
                            text="视频号Token 过期啦"
                            # text=f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n频控啦，请扫码更换其他公众号token\nhttps://mp.weixin.qq.com/"
                        )
                    time.sleep(60 * 15)
                    continue
                if res_json["base_resp"]["err_msg"] == "freq control":
                    AliyunLogger.logging(
                        code="2000",
                        platform=self.platform,
                        mode=self.mode,
                        env=self.env,
                        message=f"status_code:{response.status_code}, get_videoList:{response.text}\n",
                    )
                    if 20 >= datetime.datetime.now().hour >= 10:
                        Feishu.bot(
                            log_type=self.mode,
                            crawler=self.platform,
                            text="视频号Token 过期啦"
                            # text=f"{token_dict['title']}\n操作人:{token_dict['operator']}\n更换日期:{token_dict['update_time']} \n频控啦，请扫码更换其他公众号token\nhttps://mp.weixin.qq.com/"
                        )
                    time.sleep(60 * 15)
                    continue
                if not res_json.get("list"):
                    AliyunLogger.logging(
                        code="2000",
                        platform=self.platform,
                        mode=self.mode,
                        env=self.env,
                        message="没有更多视频了",
                    )
                    return
                else:
                    buffer = res_json["last_buff"]
                    for obj in res_json["list"]:
                        try:
                            AliyunLogger.logging(
                                code="1001",
                                platform=self.platform,
                                mode=self.mode,
                                message="扫描到一条视频",
                                env=self.env,
                                data=obj,
                            )
                            repeat_flag = self.process_video_obj(obj)
                            if not repeat_flag:
                                return
                        except Exception as e:
                            AliyunLogger.logging(
                                code="3000",
                                platform=self.platform,
                                mode=self.mode,
                                env=self.env,
                                message=f"抓取单条视频异常:{e}\n",
                            )
        else:
            AliyunLogger.logging(
                code="3000",
                platform=self.platform,
                mode=self.mode,
                env=self.env,
                message="{}\t获取 id 失败".format(self.account_name),
            )

    def process_video_obj(self, video_obj):
        trace_id = self.platform + str(uuid.uuid1())
        video_dict = {
            "video_id": video_obj["nonce_id"],
            "video_title": clean_title(video_obj["desc"].split("\n")[0].split("#")[0]),
            "out_video_id": video_obj["nonce_id"],
            "publish_time_stamp": int(time.time()),
            "publish_time_str": time.strftime(
                "%Y-%m-%d %H:%M:%S", time.localtime(int(time.time()))
            ),
            "play_cnt": 0,
            "comment_cnt": 0,
            "like_cnt": 0,
            "share_cnt": 0,
            "user_id": self.user_dict["uid"],
            "cover_url": video_obj["media"][0]["cover_url"] if video_obj['media'][0]['cover_url'] else video_obj['media'][0]['thumb_url'],
            "video_url": video_obj["media"][0]["url"],
            "avatar_url": video_obj["head_url"],
            "width": video_obj["media"][0]["width"],
            "height": video_obj["media"][0]["height"],
            "duration": video_obj["media"][0]["video_play_len_s"],
            "platform": self.platform,
            "strategy": self.mode,
            "crawler_rule": self.rule_dict,
            "session": f"shipinhao-author-{int(time.time())}",
        }
        # video_dict["out_user_id"] = video_dict["user_id"]
        # 无更新时间，去重即可
        pipeline = PiaoQuanPipeline(
            platform=self.platform,
            mode=self.mode,
            item=video_dict,
            rule_dict=self.rule_dict,
            env=self.env,
            trace_id=trace_id,
        )
        if not pipeline.repeat_video():
            return False
        else:
            video_dict["publish_time"] = video_dict["publish_time_str"]
            self.mq.send_msg(video_dict)
            self.download_cnt += 1
            AliyunLogger.logging(
                code="1002",
                platform=self.platform,
                mode=self.mode,
                env=self.env,
                data=video_dict,
                trace_id=trace_id,
                message="成功发送 MQ 至 ETL",
            )
        return True


# if __name__ == "__main__":
#     temp_token = "2080949641"
#     temp_cookie = "ua_id=bw4VuFJr6fAuSkwdAAAAAClaW0m9Aua-6IfHaXU_zpo=; wxuin=95302180931488; mm_lang=zh_CN; RK=kreEMgtMMJ; ptcz=8fd1b267c98a1185bbe6455a081f1264048ee388363ca305d9ef4812892c7900; qq_domain_video_guid_verify=2ba78a5010233582; poc_sid=HOinP2Wj322Ex737kV651Zqy6y8fSprOUUvaegBg; _qimei_q36=; _qimei_h38=9eea33ea92afe8a922333fce03000001317916; pgv_pvid=9056371236; _clck=3930572231|1|fgk|0; uuid=6562bbd8859230ce4120dfa063c76997; rand_info=CAESIGAatjSIjvxVJVDxRDN7F/CNFWMifvAVqje98rd++8UY; slave_bizuin=3236647229; data_bizuin=3236647229; bizuin=3236647229; data_ticket=qm3i6jRhObs1yKHttGh0gVI02Mz7FTPfatn0RMLdaWyD7Ukcokm5Dc3mmYLQUZPg; slave_sid=UWxjZnhBREZRRTNKZ3dYZTlYRE9Db2lxQUhOM3lZUlRoMkV0MG1wdVVudGpQTWxnVkxzYW5pV2c3NjB3bnAyQ2lPaXBBVVRPazEybWtKSVEzTnUyazZ6WEJsdnFaWWVDaUFrM3pTTXRkeUNJS3RNVTc2NFRBWkZiVGQzYllacEFRalBBZ2tXZlltblJYS2VS; slave_user=gh_d284c09295eb; xid=cb96e6ba4b4960d74a22869b1bb21406; _clsk=z77guf|1699532621466|4|1|mp.weixin.qq.com/weheat-agent/payload/record"
#     SP = ShiPinHaoAccount(
#         token=temp_token,
#         cookie=temp_cookie,
#         account_name="心煤",
#         platform="shipinhao",
#         mode="author",
#         rule_dict={},
#         env="prod",
#     )
#     SP.get_account_videos()