干货，用UI自动化方式获取视频信息，请勿恶意爬虫

谁借莪１个温暖的怀抱￠ 2023-09-26 10:33 20阅读 0赞

这一期我们来分享利用UI自动化方式获取视频信息

**1.分析请求数据（包括解析文章请求、内容请求、视频请求、token等解析）：**

# -*- coding = utf-8 -*-
    # ------------------------------
    # @time: 2022/5/5 17:56
    # @Author: drew_gg
    # @File: analyze_data_news.py
    # @Software: wei_chat_news
    # ------------------------------
    
    import json
    import gzip
    from io import BytesIO
    from bs4 import BeautifulSoup
    from urllib.parse import urlparse, parse_qs
    from qn import common_qiniu as qn_url
    
    
    def analyze_news(driver):
        """
        分析账号内容
        :param driver:
        :return:
        """
        freq = False
        news_data = []
        for re in driver.requests:
            if "https://mp.weixin.qq.com/cgi-bin/appmsg?action=list_ex&begin=" in re.url:
                try:
                    if "freq control" in str(re.response.body):
                        freq = True
                    else:
                        buff = BytesIO(re.response.body)
                        f = gzip.GzipFile(fileobj=buff)
                        htmls = f.read().decode('utf-8')
                        news_data.append(json.loads(htmls))
                except Exception as e:
                    print(e)
                    # return news_data
            if "https://mp.weixin.qq.com/cgi-bin/searchbiz?action=search_biz&begin=0&count=5" in re.url:
                if "freq control" in str(re.response.body):
                    freq = True
        return news_data, freq
    
    
    def analyze_video(driver, url=''):
        """
        分析账号视频
        :param driver:
         :param url:
        :return:
        """
        news_data = []
        video_data = []
        video_cover = []
        vid = ''
        cover = ''
        tx_vid = ''
        tx_cover = ''
        for re in driver.requests:
            if url and url.split('#')[0].replace('http', 'https') in re.url:
                buff = BytesIO(re.response.body)
                f = gzip.GzipFile(fileobj=buff)
                htmls = f.read().decode('utf-8')
                ht = BeautifulSoup(htmls, features="lxml")
                video = ht.findAll('iframe', {
        'class': 'video_iframe'})
                if video:
                    for v in video:
                        if "vid" in v.attrs['data-src']:
                            tx_vid = parse_qs(urlparse(v.attrs['data-src']).query)['vid'][0]
            if tx_vid + '.png' in re.url:
                tx_cover = qn_url.url_to_qiniu(re.url, 'png')
    
            if "https://mp.weixin.qq.com/mp/videoplayer?action=get_mp_video_play_url" in re.url:
                pa = re.url.split('&')
                for i in pa:
                    if 'VID' in i.upper():
                        vid = i.split('=')[1]
                buff = BytesIO(re.response.body)
                f = gzip.GzipFile(fileobj=buff)
                htmls = f.read().decode('utf-8')
                html_dic = json.loads(htmls)
                html_dic['vid'] = vid
                news_data.append(html_dic)
                # print(json.loads(htmls))
            if "https://mp.weixin.qq.com/mp/videoplayer?action=get_mp_video_cover&vid=" in re.url:
                pa = re.url.split('&')
                for i in pa:
                    if 'VID' in i.upper():
                        vid = i.split('=')[1]
                news_cover = json.loads(re.response.body)['url']
                video_cover.append(vid + '@@@@' + news_cover)
            # 获取TX视频的视频
            if "https://vd6.l.qq.com/proxyhttp" in re.url:
                buf = BytesIO(re.response.body)
                f = gzip.GzipFile(fileobj=buf)  # // gzip 文件498行做了修改，遇到异常中断执行，而不是抛出异常
                data = f.read().decode('utf-8')
                video_data.append(json.loads(data))
        video_dic = {
        }
        video_url = []
        # 视频路径 //json.loads(news_data[0]['vinfo'])['vl']['vi'][0]['ul']['ui'][3]['url']
        if video_data:
            for n1 in video_data:
                vi_dic = {
        }
                try:
                    for n2 in (json.loads(n1['vinfo']))['vl']['vi']:
                        vi_l = []
                        vid = n2['vid']
                        for n3 in n2['ul']['ui']:
                            vi_l.append(n3['url'])
                        vi_dic['url'] = vi_l
                except Exception as e:
                    print("解析无vinfo！")
                    print(e)
                video_url.append(vi_dic)
        video_dic[vid] = video_url
        if news_data:
            try:
                # 如果vid相同，把cover加入news_data
                for i, v1 in enumerate(news_data):
                    if video_cover:
                        for v2 in video_cover:
                            if v1['vid'] == v2.split('@@@@')[0]:
                                cover = v2.split('@@@@')[1]
                    else:
                        cover = ''
                    news_data[i]['cover'] = cover
            except Exception as e:
                print(e)
        return news_data, video_dic, tx_cover
    
    
    def analyze_token(driver):
        """
        分析账号token
        :param driver:
        :return:
        """
        news_data = []
        for re in driver.requests:
            if "https://mp.weixin.qq.com/cgi-bin/home?t=home/index&lang=zh_CN&token=" in re.url:
                pa = re.url.split('=')[-1]
                news_data.append(pa)
        return news_data

2.获取视频主方法：

def get_video_main():
        """
        爬取文章视频
        :return:
        """
    
        news_link = read_sql.deal_mysql("get_news_link.sql")
        if news_link:
            for i in news_link:
                link_url = i[4]
                print(" get video ing !")
                driver = video.get_video(link_url)
                try:
                    time.sleep(1)
                    # 获取文章数据
                    news_data, video_dic, tx_cover = analyze_news.analyze_video(driver, link_url)
                    # 如果有TX视频
                    if video_dic:
                        for k, y in video_dic.items():
                            if k != '':
                                for v in y:
                                    video_url = m3u8_mp4.m3u8_mp4_main(v['url'][0])
                                    vid = k
                                    if video_url:
                                        in_video.insert_video_tx(i, vid, video_url, tx_cover)
                                        un.update_news(i[0])
                            else:
                                print("无视频数据")
                    # 数据插入数据库
                    in_video.insert_video(i, news_data)
                    if news_data:
                        un.update_news(i[0])
                except Exception as e:
                    print(e)
                finally:
                    driver.quit()
        else:
            print("暂无数据处理！")

3.内容详情处理方法：

def get_content_main():
        """
        获取文章详情
        :return:
        """
        video_format = ' controlslist="nodownload"' \
                       ' style="margin-right: auto;margin-left: auto;outline: 0px;' \
                       'box-sizing: border-box;' \
                       'vertical-align: inherit;' \
                       'display: block;clear: both;overflow-wrap: break-word !important;' \
                       'visibility: visible !important; max-width: 100%;' \
                       '" controls="controls" src="'
        news_link_l = read_sql.deal_mysql("get_news_for_content.sql")
        if news_link_l:
            for i in news_link_l:
                news_id = i[0]
                news_title = i[1]
                news_link = i[2]
                have_video = i[3]
                print(datetime.datetime.now(), '*********** get content ing !')
                # 获取内容
                news_content, news_video, img_l, source = content.get_content(news_link)
                # # 更新视频表封面图
                # for c, k in cover.items():
                #     in_video.update_video(vid=c, is_cover=1, cover=k)
                # data-src替换成src，把隐藏参数去掉
                news_content_all = news_content.replace('data-src', 'src').replace(' style="visibility: hidden;"', '')
                # url图片转存七牛并替换
                for im in img_l:
                    time.sleep(0.1)
                    new_img = qn_url.url_to_qiniu(im, 'png')
                    news_content_all = news_content_all.replace(im, new_img)
                news_dic = [{
        'news_id': str(news_id)}]
                if have_video == 1:
                    sql_result = read_sql.deal_more_mysql("get_video_for_content.sql", news_dic)
                    if sql_result:
                        for v1 in sql_result:
                            video_html = '<video poster="' + v1[4] + '"' + video_format + v1[1] + '"></video>'
                            for v2 in news_video:
                                if v1[0] in str(v2):
                                    news_content_all = news_content_all.replace(str(v2).replace('data-src', 'src'), video_html)
                    else:
                        print('视频还未转存')
                        continue
                # 插入内容
                try:
                    in_content.insert_content(news_id, news_title, news_content_all, source)
                except Exception as e:
                    print(e)
                # 更新weichat_news表的is_content字段
                un.update_news(news_id, 1)
    
        else:
            print("暂无数据处理！")

4.遇到m3u8视频，处理方法：

# -*- coding = utf-8 -*-
    # ------------------------------
    # @time: 2022/5/23 11:21
    # @Author: drew_gg
    # @File: get_m3u8.py
    # @Software: wei_chat_news
    # ------------------------------
    
    import os
    import time
    import random
    import requests
    import subprocess
    import urllib3
    from qn import common_upload as qiniu
    
    urllib3.disable_warnings()
    
    project_path = os.path.abspath(os.path.dirname(__file__))
    video_path = os.path.abspath(os.path.join(os.path.dirname(__file__), 'video'))
    
    
    user_agent_list = [
        "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/22.0.1207.1 Safari/537.1",
        "Mozilla/5.0 (X11; CrOS i686 2268.111.0) AppleWebKit/536.11 (KHTML, like Gecko) Chrome/20.0.1132.57 Safari/536.11",
        "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1092.0 Safari/536.6",
        "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.6 (KHTML, like Gecko) Chrome/20.0.1090.0 Safari/536.6",
        "Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.1 (KHTML, like Gecko) Chrome/19.77.34.5 Safari/537.1",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.9 Safari/536.5",
        "Mozilla/5.0 (Windows NT 6.0) AppleWebKit/536.5 (KHTML, like Gecko) Chrome/19.0.1084.36 Safari/536.5",
        "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3",
        "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_0) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1063.0 Safari/536.3",
        "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3",
        "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1062.0 Safari/536.3",
        "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3",
        "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3",
        "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.1 Safari/536.3",
        "Mozilla/5.0 (Windows NT 6.2) AppleWebKit/536.3 (KHTML, like Gecko) Chrome/19.0.1061.0 Safari/536.3",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/535.24 (KHTML, like Gecko) Chrome/19.0.1055.1 Safari/535.24",
        "Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/535.24 (KHTML, like Gecko) Chrome/19.0.1055.1 Safari/535.24"
    ]
    
    
    def get_user_agent():
        header = {
        
            'Accept': 'application/json, text/javascript, */*; q=0.01',
            'Accept-Encoding': 'gzip, deflate',
            'content-type': 'application/json',
            'x-requested-with': 'XMLHttpRequest',
            'Accept-Language': 'zh-CN,zh;q=0.8',
            'User-Agent': random.choice(user_agent_list)}
        return header
    
    
    def get_m3u8_file(url_m3u8, timeout=(10, 30), max_retry_time=3):
        """
        获取m3u8信息
        :param url_m3u8: m3u8的视频地址
        :param timeout: 超时时间
        :param max_retry_time: 尝试次数
        :return:
        """
        # 获取主域名的完整路径
        host_path = get_path(url_m3u8)
        # 随机获取headers，防止被认为机器刷
        headers = get_user_agent()
        i = 1
        while i <= max_retry_time:
            try:
                res = requests.get(url=(url_m3u8.rstrip()).strip(), headers=headers, timeout=timeout, verify=False)
                if res.status_code != 200:
                    return None
                # res.text 返回ts信息目录
                return host_path, res.text
            except Exception as e:
                print(e)
                i += 1
        return None
    
    
    def deal_m3u8_file(host_path, content_m3u8):
        """
        处理m3u8记录，获取单个的ts文件
        :param host_path: 主目录路径
        :param content_m3u8: m3u8记录
        :return:
        """
    
        ts_file = {
        }
        content_m3u8 = content_m3u8.split('\n')
        for i, m in enumerate(content_m3u8):
            if "#EXTINF:" in m:
                ts_file[content_m3u8[i+1].split('?')[0]] = host_path + content_m3u8[i+1]
            if "#EXT-X-ENDLIST" in m:
                break
            # ts_file 返回ts字典{'ts名称':'ts连接'}
        return ts_file
    
    
    def get_path(url):
        """
        获取域名目录
        :param url: m3u8连接
        :return:
        """
        if url.rfind("/") != -1:
            return url[0:url.rfind("/")] + "/"
        else:
            return url[0:url.rfind("\\")] + "\\"
    
    
    def download_ts_video(ts_url):
        """
        下载ts文件
        :param ts_url:
        :return:
        """
        # 通过ts链接下载ts文件
        ts_name = str(int(time.time() * 1000000))
        save_ts_path = video_path + '\\' + ts_name + '\\'
        # 创建存放ts文件的目录
        cmd = "md %s" % save_ts_path
        os.popen(cmd)
        time.sleep(1)
        # 存放ts顺序与名称文件的txt文件名称
        ts_record_file = save_ts_path + ts_name + '.txt'
        # 构建copy /b 字符串 ts_str
        ts_str = ''
        # ts写入ts_record_file
        with open(ts_record_file, "w+") as f:
            for k, y in ts_url.items():
                try:
                    response = requests.get(y, stream=True, verify=False)
                except Exception as e:
                    print("异常请求：%s" % e.args)
                # 下载后ts文件的名称
                ts_path_file = save_ts_path + k
                # 下载ts文件
                with open(ts_path_file, "wb+") as file:
                    for chunk in response.iter_content(chunk_size=1024):
                        if chunk:
                            file.write(chunk)
                f.write("file '%s'\n" % k)
                ts_str += k + '+'
        return ts_record_file, save_ts_path, ts_str.strip('+')
    
    
    def ts_to_mp4(ts_txt, save_ts_path, ts_str):
        """
        合并ts文件生成mp4文件
        :param ts_txt: ts记录表txt文件 // 用于ffmpeg合并所用
        :param save_ts_path:  存放ts文件路径
        :param ts_str: ts顺序串 // 1.ts + 2.ts + 3.ts + ……
        :return:
        """
        # 存放txt文件的名称
        txt = ts_txt.split(save_ts_path)[1]
        # 合并后存放视频的名称
        video_name = txt.split('.')[0] + '.mp4'
        # 生成copy /b 命令
        copy_cmd = "cd /d " + save_ts_path + "&copy /b " + ts_str + ' ' + video_name
        # 生成ffmpeg命令  //由于windows下python无法用管理员执行ffmpeg，只能舍弃这种合并方式
        # ffmpeg_cmd = "cd /d %s & ffmpeg -f concat -safe 0 -i %s -c copy %s" % (save_ts_path, txt, video_name)
        # print(copy_cmd)
        # print(ffmpeg_cmd)
        try:
            # 管道cmd模式执行cmd命令，可以收到命令执行结果
            subprocess.check_output(copy_cmd, shell=True, stderr=subprocess.PIPE)
            # ffmpeg的命令
            # subprocess.Popen(ffmpeg_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        except subprocess.CalledProcessError as err:
            raise AssertionError('合并文件失败!')
        return save_ts_path + video_name
    
    
    def m3u8_mp4_main(url):
        """
        获取m3u8信息下载并上传七牛并转码   // 暂时没有处理加密的ts文件
        :param url:
        :return:
        """
        # 获取m3u8信息  // ts文件信息和主目录路径
        video_download_url = ''
        try:
            host_path, text_m3u8 = get_m3u8_file(url)
            # 获取具体ts信息，ts_file字典{'ts名称':'ts连接'}
            ts_file = deal_m3u8_file(host_path, text_m3u8)
            # 下载ts文件，获取ts文件存放路径，copy /b命令需要的ts字符串、ffmpeg执行所需要的ts文件txt信息
            record_file_ts, ts_path_save, str_ts = download_ts_video(ts_file)
            # 获取合并后的文件
            video_file = ts_to_mp4(record_file_ts, ts_path_save, str_ts)
            # 视频上传七牛并转码
            video_download_url = qiniu.upload_qiniu(video_file)
        except Exception as e:
            print(e)
        finally:
            return video_download_url
    
    
    if __name__ == '__main__':
    
        m3u8_url = "https://apd-vlive.apdcdn.tc.qq.com/defaultts.tc.qq.com/svp_50001/D9WbuxKxEK_3ORIxMGPmLkb_Hf-9EYaHERGcUiwKAObEk-a" \
                   "C0oWDQrKibK2oJu9pTk6cCYTLBJ7Ya_890oNNu3dwYOVwrsvU9m0A7bjXVSnVUWQ8BAb_6iyUrfHJdj6JLB9ORCEBpj6FY8Ew97akeKokNWBTMwN2Cmy" \
                   "4xm72amqgqHo3zn0lSA/szg_6565_50001_0bc3waaaiaaagyamd24b4jrfdmgdasyaabca.f341003.ts.m3u8?ver=4"
        m3u8_mp4_main(m3u8_url)

常规视频分析如图（一般都是三个视频，超清、高清、流畅，至少有两个）：

![图片][2e695f3b39104c8d954e9684b0656a5c.png]

m3u8的视频：

![在这里插入图片描述][1ca56f4499f64e88a2d9503d36acf64b.png]

解析vinfo后一般有3个m3u8地址：

{
        'url':'https://apd-ab3bfda6e3c7f17903673bc29aa63be3.v.smtcdns.com/omts.tc.qq.com/AwfBnjY0sy7CusqUlUD8AyDJttC4uz-symHTokgF7fts/uwMROfz2r57EIaQXGdGnCmdeOm5ieD5vLVPWVxBgVS8DGd80/svp_50001/6JH7xBzZnXGE6Alc3apP8D4zuLTz0odtRlLJgtXefYU36iQhSBOLVoZp2S5N-SNIZ3x1oed-eIUVs_G-fLiqfq0LcwjmKOI353EmSEI_qltjWg0pnPvzVL-SjsSf5jD83PHEkaXv7JjRWPdNPPbt18hkmeDySTlIhMZkKdCn8X7P-Zz-5Iku_Q/szg_4751_50001_0bc3puaamaaa5eapln4k4nrfc7odaz6qabsa.f304110.ts.m3u8?ver=4','vt':2806,'dtc':0,'dt':2},
    
    {
        'url':'https://apd-19bad465f8d944562bdc26c7556258de.v.smtcdns.com/omts.tc.qq.com/AwfBnjY0sy7CusqUlUD8AyDJttC4uz-symHTokgF7fts/uwMROfz2r57EIaQXGdGnCWdeOm4EnYbsz8uMYzCQBiqsbQpy/svp_50001/6JH7xBzZnXGE6Alc3apP8D4zuLTz0odtRlLJgtXefYU36iQhSBOLVoZp2S5N-SNIZ3x1oed-eIUVs_G-fLiqfq0LcwjmKOI353EmSEI_qltjWg0pnPvzVL-SjsSf5jD83PHEkaXv7JjRWPdNPPbt18hkmeDySTlIhMZkKdCn8X7P-Zz-5Iku_Q/szg_4751_50001_0bc3puaamaaa5eapln4k4nrfc7odaz6qabsa.f304110.ts.m3u8?ver=4','vt':2806,'dtc':0,'dt':2},
    
    {
        'url':'https://apd-vlive.apdcdn.tc.qq.com/defaultts.tc.qq.com/svp_50001/6JH7xBzZnXGE6Alc3apP8D4zuLTz0odtRlLJgtXefYU36iQhSBOLVoZp2S5N-SNIZ3x1oed-eIUVs_G-fLiqfq0LcwjmKOI353EmSEI_qltjWg0pnPvzVL-SjsSf5jD83PHEkaXv7JjRWPdNPPbt18hkmeDySTlIhMZkKdCn8X7P-Zz-5Iku_Q/szg_4751_50001_0bc3puaamaaa5eapln4k4nrfc7odaz6qabsa.f304110.ts.m3u8?ver=4','vt':12800,'dtc':0,'dt':2}

目前看TX视频都是m3u8格式，好在ts文件都是没有加密的，所以也就没做解密处理。

**强调几点：**

1.视频的封面图和视频，都是有vid的，通过vid相同来处理视频的封面图，也是根据vid来替换视频，比如一篇文章中有多个视频。

2.账号执行表是为了处理异常执行情况，比如执行到一半，被封禁了，下次执行又得所有账号在来一遍，浪费时间，所以加了一个patch来标志，没执行完的pitch继续执行，执行完了下次执行时最大的patch加1，这样就保障了这些任务。

![在这里插入图片描述][b62a9ad6754b484bac741a77aeb12345.png]

视频记录：

![图片][ac72e629195e40e79ff512ed9eeaeff2.png]

钉钉处理登录情况：

![图片][9ad03d3092ad415c88ff91ea1978e2c0.png]

再次说明：以上仅供分享学习，请不要用于恶意爬虫！！

--------------------

### 资源分享 ###

**最后感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的，虽然不是什么很值钱的东西，如果你用得到的话可以直接拿走…**

> ![在这里插入图片描述][cd053d4700c84446ae949c6966d929a5.png_pic_center]

**喜欢软件测试的小伙伴们，如果我的博客对你有帮助、如果你喜欢我的博客内容，请 “点赞” “评论” “收藏” 一 键三连哦！**  
![在这里插入图片描述][e5839dcedca14117aff9fd1945058bd6.gif_pic_center]

[2e695f3b39104c8d954e9684b0656a5c.png]: https://img-blog.csdnimg.cn/2e695f3b39104c8d954e9684b0656a5c.png
[1ca56f4499f64e88a2d9503d36acf64b.png]: https://img-blog.csdnimg.cn/1ca56f4499f64e88a2d9503d36acf64b.png
[b62a9ad6754b484bac741a77aeb12345.png]: https://img-blog.csdnimg.cn/b62a9ad6754b484bac741a77aeb12345.png
[ac72e629195e40e79ff512ed9eeaeff2.png]: https://img-blog.csdnimg.cn/ac72e629195e40e79ff512ed9eeaeff2.png
[9ad03d3092ad415c88ff91ea1978e2c0.png]: https://img-blog.csdnimg.cn/9ad03d3092ad415c88ff91ea1978e2c0.png
[cd053d4700c84446ae949c6966d929a5.png_pic_center]: https://img-blog.csdnimg.cn/cd053d4700c84446ae949c6966d929a5.png#pic_center
[e5839dcedca14117aff9fd1945058bd6.gif_pic_center]: https://img-blog.csdnimg.cn/e5839dcedca14117aff9fd1945058bd6.gif#pic_center