【Day03】用爬蟲抓影片播放清單的所有影片連結


前言

在 Day2 的進度中,已經成功的利用pytube測試下載到 1080p 解析度的 Youtube 影片,而這篇會透過爬蟲抓影片播放清單的所有影片連結,再把所有的影片連結丟到 Day2 的程式碼中執行測試下載整個播放清單中的所有影片。

安裝 requests 和 Beautiful Soup 模組

因為這裡會用到兩個第三方套件,所以在測試之前記得要先在命令提示字元(cmd)中使用 pip 來安裝 requests 和 Beautiful Soup 模組,不然在後面 import 模組的時候會報錯。

pip3 install requests
pip3 install beautifulsoup4

測試爬蟲下載播放清單所有影片連結


首先隨便挑一個帶有播放清單的影片,對播放清單的其中一個影片,右鍵選擇檢查,就可以找到播放清單的其中一個影片連結,如下圖。

但是這樣不太方便,所以讓我們寫程式去抓,這裡我們引入 requests 模組、 Beautiful Soup 模組和 re 模組,寫一個叫 playlist_urls 的函式,用 if 判斷這個影片網址是否為單一影片,判斷為 True 就直接回傳網址,判斷為 False 就用 requests 模組底下的get 方法,得到網頁原始碼,並用 Beautiful Soup 模組來解析網頁,從網頁的 a 標籤得到需要的部分連結跟Youtube 主網址結合,來取得播放清單中所有影片的正確網址,然後因為 playlist_urls 函式所抓到的影片網址,並沒有按照順序排列,所以我用urls.sort(key = lambda s:int(re.search('index=\d+',s).group()[6:]))這行代碼,來對所有影片網址做排序,再用 for 迴圈搭配 print 輸出到命令提示字元(cmd)中,程式碼如下所示:

# 引入 requests 模組
import requests as req
# 引入 Beautiful Soup 模組
from bs4 import BeautifulSoup
# 引入 re 模組
import re

def playlist_urls(url):  # 取得播放清單所有影片網址的自訂函式
    urls = []   # 播放清單網址
    if '&list=' not in url : return urls    # 單一影片
    response = req.get(url)    # 得到網頁原始碼
    if response.status_code != 200:
        print('請求失敗')
        return
    #請求發送成功, 解析網頁
    soup = BeautifulSoup(response.text, 'lxml')
    a_list = soup.find_all('a')
    base = 'https://www.youtube.com/'    # Youtube 主網址
    for i in a_list:
        href = i.get('href')
        url = base + href  # 主網址結合 href 才是正確的影片網址
        if ('&index=' in url) and (url not in urls):
            urls.append(url)
    return urls

playlist_link = 'https://www.youtube.com/watch?v=n7KpZoJy_j4&list=PLliocbKHJNwvnlL9xkwhdkaqmPbI9LU0m' #影片播放清單連結

urls = playlist_urls(playlist_link)   #執行 playlist_urls 函式,取得播放清單所有影片網址

urls.sort(key = lambda s:int(re.search('index=\d+',s).group()[6:]))

for url in urls:
    print(url)

測試下載播放清單所有影片

在取得所有影片的連結後,讓我們把 Day2 的用 python 搭配 FFmpeg 將下載到的 1080p 高解析度影片和聲音檔合併的程式碼和上面的程式碼結合,來測試下載播放清單中所有影片 1080p 解析度的版本。
兩個程式碼合併後的變化不大,主要是在最底下的 for 迴圈中要將 download_count 變數的數值,重新設定為 1,這樣下一輪的影片下載,才不會出錯。

完整程式碼如下:

from pytube import YouTube
import os
import subprocess
# 引入 requests 模組
import requests as req
# 引入 Beautiful Soup 模組
from bs4 import BeautifulSoup
# 引入 re 模組
import re

fileobj = {}
download_count = 1

# 檢查影片檔是否包含聲音
def check_media(filename):
    r = subprocess.Popen([".\\bin\\ffprobe", filename],
                         stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    out, err = r.communicate()

    if (out.decode('utf-8').find('Audio') == -1):
        return -1  # 沒有聲音
    else:
        return 1

# 合併影片檔
def merge_media():
    temp_video = os.path.join(fileobj['dir'], 'temp_video.mp4')
    temp_audio = os.path.join(fileobj['dir'], 'temp_audio.mp4')
    temp_output = os.path.join(fileobj['dir'], 'output.mp4')

    cmd = f'".\\bin\\ffmpeg" -i "{temp_video}" -i "{temp_audio}" \
        -map 0:v -map 1:a -c copy -y "{temp_output}"'
    try:
        subprocess.call(cmd, shell=True)
        # 視訊檔重新命名
        os.rename(temp_output, os.path.join(fileobj['dir'], fileobj['name']))
        os.remove(temp_audio)
        os.remove(temp_video)
        print('視訊和聲音合併完成')
        # fileobj = {}
    except:
        print('視訊和聲音合併失敗')

def onProgress(stream, chunk, remains):
    total = stream.filesize
    percent = (total-remains) / total * 100
    print('下載中… {:05.2f}%'.format(percent), end='\r')

def download_sound():
    try:
        yt.streams.filter(type="audio").first().download()
    except:
        print('下載影片時發生錯誤,請確認網路連線和YouTube網址無誤。')
        return

# 檔案下載的回呼函式
def onComplete(stream, file_path):
    global download_count, fileobj
    fileobj['name'] = os.path.basename(file_path)
    fileobj['dir'] = os.path.dirname(file_path)
    print('\r')

    if download_count == 1:
        if check_media(file_path) == -1:
            print('此影片沒有聲音')
            download_count += 1
            try:
                # 視訊檔重新命名
                os.rename(file_path, os.path.join(
                    fileobj['dir'], 'temp_video.mp4'))
            except:
                print('視訊檔重新命名失敗')
                return

            print('準備下載聲音檔')
            download_sound()          # 下載聲音
        else:
            print('此影片有聲音,下載完畢!')
    else:
        try:
            # 聲音檔重新命名
            os.rename(file_path, os.path.join(
                fileobj['dir'], 'temp_audio.mp4'))
        except:
            print("聲音檔重新命名失敗")
        # 合併聲音檔
        merge_media()
#--------------------------------------------------------------
def playlist_urls(url):  # 取得播放清單所有影片網址的自訂函式
    urls = []   # 播放清單網址
    if '&list=' not in url : return urls    # 單一影片
    response = req.get(url)    # 發送 GET 請求
    if response.status_code != 200:
        print('請求失敗')
        return
    #-----↓ 請求成功, 解析網頁 ↓------#
    soup = BeautifulSoup(response.text, 'lxml')
    a_list = soup.find_all('a')
    base = 'https://www.youtube.com/'    # Youtube 網址
    for a in a_list:
        href = a.get('href')
        url = base + href  # 主網址結合 href 才是完整的影片網址
        if ('&index=' in url) and (url not in urls):
            urls.append(url)
    return urls
#--------------------------------------------------------------

playlist_link = 'https://www.youtube.com/watch?v=n7KpZoJy_j4&list=PLliocbKHJNwvnlL9xkwhdkaqmPbI9LU0m' #影片播放清單連結

urls = playlist_urls(playlist_link)   #執行 playlist_urls 函式
#對所有影片網址做排序
urls.sort(key = lambda s:int(re.search("index=\d+",s).group()[6:]))

for url in urls:
    download_count = 1 #改回 1
    print(url) #印出影片網址
    yt = YouTube(url, on_progress_callback=onProgress,on_complete_callback=onComplete)
    try:
        print(yt.streams.filter(subtype='mp4',resolution="1080p")[0].download())
    except:
        print(yt.streams.filter(subtype='mp4',resolution="1080p")[1].download())
    print(fileobj)

如下圖所示,沒問題的話,應該可以在跟程式碼相同位置的資料夾找到所有完整的影片檔。

小結

Day03的進度:
用爬蟲測試下載到了播放清單的所有影片連結,在整合了 Day2 的程式碼之後,成功下載到播放清單的所有影片。
如果文章中的程式碼有錯誤或是有其他想法,請麻煩在下方留言告訴我。

參考資料

Requests: HTTP for Humans™
GitHub/psf/requests
requests 2.23.0
Python 初學第十講 — 排序

#requests #Beautiful Soup #pytube






Related Posts

JavaScript 中的同步與非同步 & event loop

JavaScript 中的同步與非同步 & event loop

漫談傳輸介面-I2C

漫談傳輸介面-I2C

[MTR04] W1 D3 Git 進階指令

[MTR04] W1 D3 Git 進階指令



Sponsored