crawler_gzh/main/demo.py

# -*- coding: utf-8 -*-
# @Author: wangkun
# @Time: 2022/8/3
import datetime
import json
import os
import time
# import urllib.parse
# from urllib import parse
import requests
import urllib3
from main.feishu_lib import Feishu


class Demo:
    # 获取 token，保存至飞书云文档
    @classmethod
    def get_token(cls):
        # charles 抓包文件保存目录
        charles_file_dir = "../chlsfiles/"

        if int(len(os.listdir(charles_file_dir))) == 0:
            print("未找到chlsfile文件，等待60s")
            time.sleep(60)
        else:
            # try:
            # 目标文件夹下所有文件
            all_file = sorted(os.listdir(charles_file_dir))

            # 获取到目标文件
            old_file = all_file[-1]

            # 分离文件名与扩展名
            new_file = os.path.splitext(old_file)

            # 重命名文件后缀
            os.rename(os.path.join(charles_file_dir, old_file),
                      os.path.join(charles_file_dir, new_file[0] + ".txt"))

            with open(charles_file_dir + new_file[0] + ".txt", encoding='utf-8-sig', errors='ignore') as f:
                contents = json.load(f, strict=False)
            for content in contents:
                if "mp.weixin.qq.com" in content['host']:
                    if content["path"] == r"/mp/getappmsgext":
                        # query
                        query = content["query"]
                        Feishu.update_values("recommend", "gzh", "VzrN7E", "B9:B9", [[query]])

                        # body
                        headers = content["request"]["header"]["headers"]
                        body = content["request"]["body"]["text"]
                        # time.sleep(1)
                        Feishu.update_values("recommend", "gzh", "VzrN7E", "B8:B8", [[body]])

                        # title / vid
                        title = content["request"]["body"]["text"].split("title=")[-1].split("&ct=")[0]
                        vid = content["request"]["body"]["text"].split("vid=")[-1].split("&is_pay_subscribe")[0]
                        # time.sleep(1)
                        Feishu.update_values("recommend", "gzh", "VzrN7E", "B1:B1", [[title]])
                        # time.sleep(1)
                        Feishu.update_values("recommend", "gzh", "VzrN7E", "B2:B2", [[vid]])

                        for h in headers:
                            if h["name"] == "cookie" and "pass_ticket" in h["value"]:
                                pass_ticket = h["value"].split("pass_ticket=")[-1]
                                # print(f"pass_ticket:{pass_ticket}")
                                Feishu.update_values("recommend", "gzh", "VzrN7E", "B5:B5", [[pass_ticket]])

                            if h["name"] == "referer":
                                referer = h["value"]
                                # print(f"__biz:{referer}")
                                Feishu.update_values("recommend", "gzh", "VzrN7E", "B7:B7", [[referer]])

                            if h["name"] == "referer":
                                __biz = h["value"].split("__biz=")[-1].split("&mid=")[0]
                                # print(f"__biz:{__biz}")
                                Feishu.update_values("recommend", "gzh", "VzrN7E", "B3:B3", [[__biz]])

                            if h["name"] == "cookie" and "appmsg_token" in h["value"]:
                                appmsg_token = h["value"].split("appmsg_token=")[-1]
                                # print(f"appmsg_token:{appmsg_token}")
                                Feishu.update_values("recommend", "gzh", "VzrN7E", "B4:B4", [[appmsg_token]])

                            if h["name"] == "cookie" and "wap_sid2" in h["value"]:
                                wap_sid2 = h["value"].split("wap_sid2=")[-1]
                                # print(f"wap_sid2:{wap_sid2}")
                                Feishu.update_values("recommend", "gzh", "VzrN7E", "B6:B6", [[wap_sid2]])

    # 获取视频下载链接
    @classmethod
    def get_url(cls, url):
        try:
            payload = {}
            headers = {
                'Cookie': 'rewardsn=; wxtokenkey=777'
            }
            urllib3.disable_warnings()
            response = requests.get(url=url, headers=headers, data=payload, verify=False)
            response_list = response.text.splitlines()
            video_url_list = []
            for m in response_list:
                if "mpvideo.qpic.cn" in m:
                    video_url = m.split("url: '")[1].split("',")[0].replace(r"\x26amp;", "&")
                    video_url_list.append(video_url)
            video_url = video_url_list[0]
            return video_url
        except Exception as e:
            print(f"get_url异常:{e}")

    @classmethod
    def recommend(cls):
        token_sheet = Feishu.get_values_batch("recommend", "gzh", "VzrN7E")
        title = token_sheet[0][1]
        vid = token_sheet[1][1]
        __biz = token_sheet[2][1]
        appmsg_token = token_sheet[3][1]
        pass_ticket = token_sheet[4][1]
        wap_sid2 = token_sheet[5][1]
        referer = token_sheet[6][1]
        body = token_sheet[7][1]
        query = token_sheet[8][1]

        # title = "%25E8%25BF%2599%25E4%25B8%25AA%25E5%25BC%259F%25E6%2591%258A%25E4%25B8%258A%25E8%25BF%2599%25E4%25B8%25AA%25E5%2593%25A5%25E4%25B8%258D%25E7%259F%25A5%25E9%2581%2593%25E6%2598%25AF%25E5%25B9%25B8%25E8%25BF%2590%25E8%25BF%2598%25E6%2598%25AF%25E5%2599%25A9%25E6%25A2%25A6%25E7%259A%2584%25E5%25BC%2580%25E5%25A7%258B"
        # vid = "wxv_2512850656278560768"
        # __biz = "MzkwMjM4OTYyMA%3D%3D"
        # appmsg_token = "1177_uEP3JZFhlaEpfuX2xZkF7a5mr7y0GIw2Xtm6WmawNEjFdmTFLWIMx6XbC5z-mFgwQ6mWBq8Wh9EvcaVI"
        # pass_ticket = "1VEraA4AwklrT95KI+7WK09zHoybkPgSDgYX/Fw3ArRKYh+QnkI1NXGFHvZ0naeD"
        # wap_sid2 = "CNSn5r4HEooBeV9ISXFxZ1FWX1JPazhCVXlfWW5UU0dSbXpza0lHZnlKSDVUVF9aUndjeDhvbDctc0tfUGlWd09uOTVwcTVxWGVkenB5Nld2YkRKNFBKVk9jZVpJZTZGU2hXUkpnZlB5OTMwSEJXQVpTNS13NnJXY0hiTjJibkJ6U3Z3WllaRkNmcEwzWVNBQUF+MOzurZcGOA1AAQ=="

        # print(f"title:{title}")
        # print(f"vid:{vid}")
        # print(f"__biz:{__biz}")
        # print(f"appmsg_token:{appmsg_token}")
        # print(f"pass_ticket:{pass_ticket}")
        # print(f"wap_sid2:{wap_sid2}\n")

        url = "https://mp.weixin.qq.com/mp/getappmsgext?"
        headers = {
            # "content-type": "application/x-www-form-urlencoded; charset=UTF-8",
            "content-type": 'text/plain',
            "accept": "*/*",
            "x-requested-with": "XMLHttpRequest",
            "accept-language": "zh-cn",
            "accept-encoding": "gzip, deflate, br",
            "origin": "https://mp.weixin.qq.com",
            "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.26(0x18001a29)"
                          " NetType/WIFI Language/zh_CN",
            "referer": referer,
        }
        # query_string = {
        #     "f": "json",
        #     "mock": "",
        #     "uin": "777",
        #     "key": "777",
        #     "pass_ticket": pass_ticket,
        #     "wxtoken": "",
        #     "devicetype": "iOS14.7.1",
        #     "clientversion": "18001a29",
        #     "__biz": __biz,
        #     "appmsg_token": appmsg_token,
        #     "x5": "0",
        #     # "f": "json"
        # }
        cookies = {
            "appmsg_token": appmsg_token,
            "devicetype": "iOS14.7.1",
            "lang": "zh_CN",
            "pass_ticket": pass_ticket,
            "rewardsn": "",
            "version": "18001a29",
            "wap_sid2": wap_sid2,
            "wxtokenkey": "777",
            "wxuin": "2010747860"
        }
        # form = {
        #     "r": "0.7399955678811934",
        #     "__biz": __biz,
        #     "appmsg_type": "9",
        #     "mid": "2247483669",
        #     "sn": "b290af5fe2d28a02bf93721a10ce5106",
        #     "idx": "1",
        #     "scene": "136",
        #     "title": title,
        #     "ct": "1659387788",
        #     "abtest_cookie": "AAACAA==",
        #     "devicetype": "iOS14.7.1",
        #     "version": "18001a29",
        #     "is_need_ticket": "0",
        #     "is_need_ad": "1",
        #     "comment_id": "0",
        #     "is_need_reward": "0",
        #     "both_ad": "0",
        #     "reward_uin_count": "0",
        #     "send_time": "",
        #     "msg_daily_idx": "1",
        #     "is_original": "0",
        #     "is_only_read": "1",
        #     "req_id": "0415gcfdToRrDu1wXSkK8VcK",
        #     "pass_ticket": pass_ticket,
        #     "is_temp_url": "0",
        #     "item_show_type": "5",
        #     "tmp_version": "1",
        #     "more_read_type": "0",
        #     "appmsg_like_type": "2",
        #     "related_video_sn": "",
        #     "related_video_num": "5",
        #     "vid": vid,
        #     "is_pay_subscribe": "0",
        #     "pay_subscribe_uin_count": "0",
        #     "has_red_packet_cover": "0",
        #     "album_id": "1296223588617486300",
        #     "album_video_num": "5",
        #     "cur_album_id": "",
        #     "is_public_related_video": "0",
        #     "encode_info_by_base64": "0",
        #     "exptype": ""
        # }

        response = requests.post(url=url, headers=headers, cookies=cookies, params=query, data=body)
        if "related_tag_video" not in response.json():
            print(f"response:{response.text}\n")
        elif len(response.json()["related_tag_video"]) == 0:
            print(f"response:{response.text}\n")
        else:
            feeds = response.json()["related_tag_video"]
            for m in range(len(feeds)):
                # video_title
                if "title" not in feeds[m]:
                    video_title = 0
                else:
                    video_title = feeds[m]["title"]
                    # video_title = base64.b64decode(video_title).decode("utf-8")

                # video_id
                if "vid" not in feeds[m]:
                    video_id = 0
                else:
                    video_id = feeds[m]["vid"]

                # play_cnt
                if "read_num" not in feeds[m]:
                    play_cnt = 0
                else:
                    play_cnt = feeds[m]["read_num"]

                # like_cnt
                if "like_num" not in feeds[m]:
                    like_cnt = 0
                else:
                    like_cnt = feeds[m]["like_num"]

                # duration
                if "duration" not in feeds[m]:
                    duration = 0
                else:
                    duration = feeds[m]["duration"]

                # video_width / video_height
                if "videoWidth" not in feeds[m] or "videoHeight" not in feeds[m]:
                    video_width = 0
                    video_height = 0
                else:
                    video_width = feeds[m]["videoWidth"]
                    video_height = feeds[m]["videoHeight"]

                # send_time
                if "pubTime" not in feeds[m]:
                    send_time = 0
                else:
                    send_time = feeds[m]["pubTime"]

                # user_name
                if "srcDisplayName" not in feeds[m]:
                    user_name = 0
                else:
                    user_name = feeds[m]["srcDisplayName"]
                    # user_name = base64.b64decode(user_name).decode("utf-8")

                # user_id
                if "srcUserName" not in feeds[m]:
                    user_id = 0
                else:
                    user_id = feeds[m]["srcUserName"]

                # head_url
                if "head_img_url" not in feeds[m]:
                    head_url = 0
                else:
                    head_url = feeds[m]["head_img_url"]

                # cover_url
                if "cover" not in feeds[m]:
                    cover_url = 0
                else:
                    cover_url = feeds[m]["cover"]

                # video_url
                if "url" not in feeds[m]:
                    video_url = 0
                else:
                    video_url = feeds[m]["url"]

                # 下载链接
                download_url = cls.get_url(video_url)

                print(f"video_title:{video_title}")
                print(f"download_url:{download_url}")
                # print(f"video_id:{video_id}")
                # print(f"play_cnt:{play_cnt}")
                # print(f"like_cnt:{like_cnt}")
                # print(f"duration:{duration}")
                # print(f"video_width:{video_width}")
                # print(f"video_height:{video_height}")
                # print(f"send_time:{send_time}")
                # print(f"user_name:{user_name}")
                # print(f"user_id:{user_id}")
                # print(f"head_url:{head_url}")
                # print(f"cover_url:{cover_url}")
                # print(f"video_url:{video_url}")
                print("\n")

    @classmethod
    def get_sheet(cls):
        sheet = Feishu.get_values_batch("recommend", "gzh", "VzrN7E")
        print(sheet)
        print(sheet[0][1])
        print(sheet[1][1])
        print(sheet[2][1])
        print(sheet[3][1])
        print(sheet[4][1])
        print(sheet[5][1])

    @classmethod
    def demo1(cls):
        t = 3600 * (24 - datetime.datetime.now().hour)
        print(24 - datetime.datetime.now().hour)
        print(t)
        print(3600*(24-datetime.datetime.now().hour))

    @classmethod
    def demo2(cls):
        for i in range(3):
            if '我们' in '我们今天吃什么？':
                print('no')
                print('休眠 2s\n')
                time.sleep(2)
            else:
                print('yes')
                print('休眠 2s\n')
                time.sleep(2)


if __name__ == "__main__":
    # Demo.get_token()
    # Demo.recommend()
    # Demo.get_sheet()
    Demo.demo2()