Crawling & Scraping

[python] pymysql과 pytube를 이용한 Audio 파일 다운로드

JasonHan 2022. 3. 16. 04:07

python에서 데이터베이스에 있는 값을 가져와 자동으로 pytube에서 검색하여 다운로드하는 과정을 진행하려 합니다.

 

아래 그림은 현재 진행하는 프로젝트에서 현재 진행하려는 부분만 도식화하였습니다.

DB → Make Search Terms → Download Audio Files

 

그리고 원활한 테스트를 위하여 개발 환경은 google colabotory를 사용하였습니다.

먼저 Mysql 데이터베이스 연결을 위한 "pymysql" 설치와 라이브러리를 import 하고,

DB 접속 정보를 입력하여 연결

!pip install pymysql # mysql database connection library
# pymysql 설치

import pymysql.cursors  
# For Request Query String
import pandas as pd 
# Querying results to make dataframe(to divide datas)

conn = pymysql.connect(host='test', port=3306, user='root', password='test', \
db='test', charset='utf8', autocommit=True, cursorclass=pymysql.cursors.DictCursor)        
# mysql connect information

 

연결 정보와 쿼리문 실행을 위한 커서를 생성하고, 사용할 쿼리문을 실행합니다.

여기서 cur.execute(sql)를 실행하면 조회된 결과 개수가 출력됩니다.

INSERT, "CREATE" 등의 쿼리문은 파이썬에서 실행시키면 그 결과 값이 없으므로, fetch함수를 사용하여 결과를 저장

fetchall()하여 결과를 저장하고, 데이터 베이스 연결을 종료

(cursor를 사용하지 않고 with문을 사용하면 따로 연결을 해제하지 않을 수 있다.)

cur = conn.cursor() # open to query
sql = "SELECT sg_name, sg_artists FROM bf_sg;"  # select data in DB
cur.execute(sql)    # excute query

################################################

result = cur.fetchall() # merging result
conn.close()    # DB disconnect
print(result)   # Check Status
# An error appeared, but it doesn't matter because just too many search results.

 

위의 결과 값을 데이터 프레임으로 변환하여 확인

df = pd.DataFrame(result)   # 전체 음악리스트 확인(playlist_id 기준이기 때문에 중복값 들어있음)
df

Result

If (Lp Ver.) Bread
All By Myself Eric Carmen
Time After Time Cyndi Lauper
Nothing's Gonna Change My Love For You George Benson
Reality Vladimir Cosma, Richard Sanderson
... ...
Spectre Rob Simonsen
Debussy : Prelude A L'apres-Midi D'un Faune L.... Tal & Groethuysen
Yesterday The Piano Guys
Coeur Rob Simonsen
Debussy : 3 Esquisses Symphoniques L.109 'La M... Tal & Groethuysen

109480 rows × 2 columns

 

 

현재 조회한 데이터는 DB상에 여러 가지 플레이리스트에 포함된 음악 리스트이므로, 중복 값이 매우 많이 있을 것.

음원파일을 다운로드하기 위한 것이므로 중복 제거합니다.

df.drop_duplicates(subset=None, keep='first', inplace=True, ignore_index=True)  # 중복 처리
df

Result

If (Lp Ver.) Bread
All By Myself Eric Carmen
Time After Time Cyndi Lauper
Nothing's Gonna Change My Love For You George Benson
Reality Vladimir Cosma, Richard Sanderson
... ...
Spectre Rob Simonsen
Debussy : Prelude A L'apres-Midi D'un Faune L.... Tal & Groethuysen
Yesterday The Piano Guys
Coeur Rob Simonsen
Debussy : 3 Esquisses Symphoniques L.109 'La M... Tal & Groethuysen

43038 rows × 2 columns

 

중복 제거 후 데이터의 양이 반이상 줄어들었습니다. 이 중 테스트를 위해 10개의 데이터만 제한적으로 사용할 것입니다. 한 번에 너무 많은 양의 데이터를 처리하면 Youtube에서 IP 차단을 24시간 동안 하기 때문에 슬라이싱 된 데이터를 기준으로 나눠서 요청할 것입니다.

df1 = df[0:10]
df1

ext_lists = df1.values.tolist() # pytube에서 검색어로 사용하기 위한 리스트 변환

그리고 pytube에서 검색어로 사용하기 위하여 데이터 프레임을 리스트로 변환

 

Result

If (Lp Ver.) Bread
All By Myself Eric Carmen
Time After Time Cyndi Lauper
Nothing's Gonna Change My Love For You George Benson
Reality Vladimir Cosma, Richard Sanderson
Raindrops Keep Falling on My Head B.J. Thomas
Take On Me A-ha
September Earth, Wind & Fire
(They Long To Be) Close To You (Album Ver.) (드... The Carpenters
Let It Be (Remastered 2015) The Beatles

이제 음원을 받기 위해 pytube를 설치하고, 필요한 라이브러리들을 import.

그리고 pytube의 "Search"는 문자열을 입력받아 Youtube 검색 결과 리스트를 반환합니다.

 

우리가 필요한 음원은 대부분 1~2번째 있으므로 인덱싱으로 1개의 목록만 가져올 것이고, 간혹 영상 자체가 없는 경우가 있는 걸로 보아 (한/미 곡 외에도 중국, 일본 등등 제3 언어로 된 데이터 때문으로 보임) 에러가 난 검색 리스트는 확인용이나 DB에 적용하기 위해 반환하여 리스트에 저장하도록 합니다.

 

반복문에서 검색 결과만 이후에 전달하다 보니, 저장할 때 파일명을 어쩔 수 없이 영상 제목으로 하게 되어 나중에 라벨링을 하거나 DB에 연관 지으려면 골치 아플 것 같아서 리스트에 추가할 때 검색어를 같이 딕셔너리 형태로 append 하였습니다. 

!python3 -m pip install --upgrade "git+https://github.com/nficano/pytube.git"
# pytube-12.0.0 설치

import os
import pytube
from pytube import Search


audio_lists = []    # 검색결과가 나온 비디오 넣는 리스트
forbidden_lists = []    # 검색결과 없는 비디오 리스트 - 추후 DB 적용하기 위하여

for ext in ext_lists:
    try:
        StrClean = "-".join(ext)    
        # 노래 제목과 가수를 "-"를 구분자로 문자열 합치기
        yt = Search(StrClean).results[0]    
        # Search 메소드를 사용하여 검색하고, 그 결과의 1번째 결과 사용(가장 관련있는 영상 사용)
    except:
        print("=================== Error Occured :  " + StrClean + " ===================")  # 에러 로그
        forbidden_lists.append(StrClean)    
        # 검색결과 없는 비디오 리스트 삽입
    audio_lists.append({StrClean : yt})     
    # 성공적으로 검색되는 비디오의 Youtube 객체 리스트 삽입
    # 추후 파일명으로 사용하여 라벨링 및 DB 적용을 수월하게 하기 위하여 딕셔너리 형태로 전달

 

반복문으로 만들어진 리스트 확인과 성공적으로 추가된 영상 목록을 확인

아래 Result에서 볼 수 있듯이 딕셔너리 형태로 리스트에 들어갔으며, 구분자 "-"만 넣어 추후에 가공이 쉽도록 하였습니다. 추가로 pytube의 "__main__"  함수에서 Youtube 객체로 다운로드 링크를 대신할 수 있는 역할을 합니다. 

audio_lists, len(audio_lists)   # 성공 목록 확인 및 개수 확인

# forbidden_lists, len(forbidden_lists)   # 실패 목록 확인 및 개수 확인

#################################  Result #################################
([{'If (Lp Ver.)-Bread': <pytube.__main__.YouTube object: videoId=SQGSR88WZtw>},
  {'All By Myself-Eric Carmen': <pytube.__main__.YouTube object: videoId=iN9CjAfo5n0>},
  {'Time After Time-Cyndi Lauper': <pytube.__main__.YouTube object: videoId=VdQY7BusJNU>},
  {"Nothing's Gonna Change My Love For You-George Benson": <pytube.__main__.YouTube object: videoId=Tr97MQiqW38>},
  {'Reality-Vladimir Cosma, Richard Sanderson': <pytube.__main__.YouTube object: videoId=5zq0_x77aAE>},
  {'Raindrops Keep Falling on My Head-B.J. Thomas': <pytube.__main__.YouTube object: videoId=sySlY1XKlhM>},
  {'Take On Me-A-ha': <pytube.__main__.YouTube object: videoId=djV11Xbc914>},
  {'September-Earth, Wind & Fire': <pytube.__main__.YouTube object: videoId=Gs069dndIYk>},
  {"(They Long To Be) Close To You (Album Ver.) (드라마 '그녀는 예뻤다' 삽입곡)-The Carpenters": <pytube.__main__.YouTube object: videoId=fBuNYi0Bdjw>},
  {'Let It Be (Remastered 2015)-The Beatles': <pytube.__main__.YouTube object: videoId=HzvDofigTKQ>}],
 10)

아래 코드 블록은 음원을 다운로드하기 위한 가장 긴 메인 함수입니다.

unfounded_lists = []    # 검색된 영상 중 필터링된 관련성 떨어지는 목록용 리스트

def youtube_download(var):
    keys = str(list(var.items())[0][0])  
    # 딕셔너리에서 파일 이름으로 쓸 string 언패킹
    link = var.get(keys)            
    # 딕셔너리에서 파일이름이 key값이므로, values 값인  (pytube.__main__.YouTube) 자료형 언패킹
    audio_streams = link.streams.filter(only_audio=True)    
    # pytube에서 지원하는 다양한 stream 중 Audio 속성을 가진 stream만
    print(audio_streams)  
    # audio streams 처리 전체 정보

    flt1a = (int(link.length)) > 130    # 영상 길이 2분 10초 초과
    flt1b = (int(link.length)) < 150    # 영상 길이 10분 미만
    flt1 = flt1a and flt1b              # 위의 두 조건 동시에 만족하는 필터
    flt2 = ("official" in str(link.title).lower() or \
    "audio" in str(link.title).lower() or "lylic" in str(link.title).lower())  
    # "official", "audio", "lylic" 중 제목에 하나라도 들어가는 영상 제목 통과

    # 내용 체크
    print(f'영상제목: {link.title},  영상 조회수: {link.views}, \
    영상 길이: {link.length} sec. [{str(link.length // 60).zfill(2)}\
    :{str(link.length % 60).zfill(2)}]')
    print(f'영상 URL: {link.watch_url}')
    print(f"* 1번 체크포인트 - 음원 길이(초):  {int(link.length)},\
    타입은? : {type(int(link.length))}")
    print(f"* 2번 체크포인트 - 음원 제목(소문자):  {str(link.title).lower()}") 
    print(f"* 3번 체크포인트 - 2분 이상 10분 이하 길이:  {flt1}  ,\
    음원에 자주 붙은 문자열 포함 여부: {flt2}")

    # 저장 폴더 만들기 / Local 혹은 Server 환경에서 돌릴 경우 사용
    # if not os.path.exists('aurigin'):
    #     os.mkdir('aurigin')
    # else:
    #     pass
    
    ## filter()를 활용하여 파일 확장자 타입 등을 포함한 streams 처리
    if flt1 or flt2 == True:    
    # 위의 1, 2 조건 중 한가지라도 만족한다면 다운로드
        audio_streams = link.streams.filter(file_extension='mp4').get_by_itag(140)  
        # 여러 개의 다운로드 스트림 선택해서 링크 하나 가져오기
        # print(audio_streams) # 가능한 stream 표시
        
        title = keys  
        # 영상 제목으로 파일명이 하고 싶다면? audio_streams.title 또는 link.title 사용

        # special_char = '\/:*?"<>|.'
        # for c in special_char:
        #     if c in title:
        #         title = title.replace(c, '')
        # print(title)  
        # 파일명에 혹시라도 특수문자 들어갈까봐 정규식처럼 문자치환(추후 정규식으로 변경)
    
        audio_streams.download(filename= title + '.mp3', output_path='test')     
        # 설정한 파일명, 경로에 다운로드
        print(f"#@# {keys} 다운로드 성공!")           
        # 성공 로그
       
    else:
        print(f"^^^ Related Video Not Found : {keys}")  
        # 실패 로그
        unfounded_lists.append(keys)  							
        # 실패 목록 리스트에 저장

함수 내부 필터링 조건문 이전까지는 딕셔너리 형태로 받은 변수를 다시 리스트 삽입 이전의 자료형으로 출력하기 위한 언패킹 작업과 조건문에 사용될 변수 선언이 대부분입니다.

그 외 중요한 pytube의 스트림 설정이 있습니다.

https://pytube.io/en/latest/user/streams.html#filtering-streams

 

Working with Streams and StreamQuery — pytube 12.0.0 documentation

Working with Streams and StreamQuery The next section will explore the various options available for working with media streams, but before we can dive in, we need to review a new-ish streaming technique adopted by YouTube. It assumes that you have already

pytube.io

위 링크는 pytube 매뉴얼 페이지 스트림 설명이 되어있고,

그중 오디오 스트림만 설명하자면, 각 tag값으로 다운로드할 수 있는 확장자 및 코덱, 대역폭을 선택할 수 있습니다.

>>> yt.streams.filter(only_audio=True)
[<Stream: itag="140" mime_type="audio/mp4" abr="128kbps" acodec="mp4a.40.2" progressive="False" type="audio">,
<Stream: itag="249" mime_type="audio/webm" abr="50kbps" acodec="opus" progressive="False" type="audio">,
<Stream: itag="250" mime_type="audio/webm" abr="70kbps" acodec="opus" progressive="False" type="audio">,
<Stream: itag="251" mime_type="audio/webm" abr="160kbps" acodec="opus" progressive="False" type="audio">]

스트림을 선택하게 되어 [영상제목, 영상 조회수, 영상 길이, 영상 설명, 영상 평점, 영상 썸네일 링크, 영상 나이 제한, 영상 제작자, 영상 아이디, 영상 채널 URL, 영상 링크, 영상 URL] 과 같은 정보들을 확인할 수 있습니다.

 

그 아래 조건문은 영상 제목과 길이를 통하여 인덱스 0번의 영상 중에서 음원파일로 보기 어려운 영상들을 필터링하여 추후 반환하기 위한 리스트에 삽입하였다.

그리고 현재는 코랩 노트북 가상 환경에서 작업 중이므로, 폴더 생성 및 로컬 경로 저장 부분은 모두 주석처리하였습니다. 기존에 설정한 mp4 타입의 오디오 파일을 일단 경로에 저장 시 mp3 확장자로 설정

(이후에 wav파일로 변환하기 위해 pydub 라이브러리를 통해 mp4 to wav 할 예정입니다.)

for audio in audio_lists:
    try:
        youtube_download(audio)
    except:
        print(f"no error keep going, {keys}")

youtube_download 함수에 링크 정보가 담겨있는 리스트로 호출하여 하나씩 다운로드 시도하도록 하였으며, 다운로드 중 에러 발생 시 로그를 남기도록만 설정하였습니다. 

 

아래 결과는 로그 중 1개의 결과를 가져왔으며, 영상길이가 130 초과 150 미만인 조건과 "official"과 같은 문자가 영상 제목에 없으므로 관련 없는 영상으로 에러를 출력하였습니다.

(실제 동작 시 영상길이 필터는 150초 미만이 아닌 600초 미만으로 설정)

#################################  Result #################################
[<Stream: itag="139" mime_type="audio/mp4" abr="48kbps" 
acodec="mp4a.40.5" progressive="False" type="audio">, 
<Stream: itag="140" mime_type="audio/mp4" abr="128kbps" 
acodec="mp4a.40.2" progressive="False" type="audio">, 
<Stream: itag="249" mime_type="audio/webm" abr="50kbps" 
acodec="opus" progressive="False" type="audio">, 
<Stream: itag="250" mime_type="audio/webm" abr="70kbps" 
acodec="opus" progressive="False" type="audio">, 
<Stream: itag="251" mime_type="audio/webm" abr="160kbps" 
acodec="opus" progressive="False" type="audio">]

영상제목: Bread - If,  영상 조회수: 4423,  영상 길이: 203 sec. [03:23]
영상 URL: https://youtube.com/watch?v=SQGSR88WZtw
* 1번 체크포인트 - 음원 길이(초):  203, 타입은? : <class 'int'>
* 2번 체크포인트 - 음원 제목(소문자):  bread - if
* 3번 체크포인트 - 2분 이상 10분 이하 길이:  False  ,  
음원에 자주 붙은 문자열 포함 여부: False
^^^ Related Video Not Found : If (Lp Ver.)-Bread

아래 코드는 필터링에서 걸러진 데이터들을 DataFrame에 나눠 담을 수 있게 만든 함수입니다. 

def unfounded_df(ufs):   # unfounded_lists를 DB에 적용하기 위한 Dataframe 만들기
    df = pd.DataFrame(ufs, columns = ['col'])
    df['sg_name'] = df.col.str.split('-').str[0]
    df['sg_artists'] = df.col.str.split('-').str[1]
    df.drop(['col'], axis=1, inplace=True)
    return df
    
=================================================

unfounded_df(unfounded_lists)   # 데이터프레임 확인

실패한 리스트를 변수로 넣고 실행하게 되면 아래와 같이 출력이 됩니다.

처음 DB에서 데이터를 뽑아 데이터 프레임화 했던 데이터와 똑같지 않은가요?

이후 데이터베이스에 수월하게 적용 가능할 것으로 생각됩니다.

Result

If (Lp Ver.) Bread
Nothing's Gonna Change My Love For You George Benson
Reality Vladimir Cosma, Richard Sanderson
Raindrops Keep Falling on My Head B.J. Thomas
(They Long To Be) Close To You (Album Ver.) (드... The Carpenters
Let It Be (Remastered 2015) The Beatles

여기까지 Pytube 라이브러리를 사용하여 유튜브에서 음원을 다운로드하는 방법이었습니다.