collect-data-pipeline/src/providers/sns/api_youtube_downloader.py
2025-09-17 07:03:06 +09:00

84 lines
3.1 KiB
Python

import os
import yt_dlp
from lib.custom_logger import get_logger
logger = get_logger()
class ApiYoutubeDownloader:
"""
YouTube動画ダウンロードクラス
Notes:
- yt_dlpライブラリを使用してYouTube動画をダウンロードするためのクラス
- `pip install yt_dlp` が必要です
- 著作権に注意して使用してください
"""
@classmethod
def download_video(cls, video_url:str, output_dir:str="downloads"):
"""
YouTube動画をダウンロードする
Args:
video_url (str): ダウンロードするYouTube動画のURLまたはVideo_ID
output_dir (str): ダウンロードした動画の保存先パス(ディレクトリ)。デフォルトは "downloads"
Returns:
str: ダウンロードした動画のファイルパス
"""
logger.info(f"Downloading video from URL: {video_url}")
os.makedirs(output_dir, exist_ok=True)
if not video_url.startswith("http"):
video_url = f"https://www.youtube.com/watch?v={video_url}"
ydl_opts = {
"outtmpl": os.path.join(output_dir, "%(title)s [%(id)s].%(ext)s"),
# 最良の映像+音声を結合。必要に応じて 'mp4' 固定に再エンコード可
"format": "bv*+ba/b",
"merge_output_format": "mp4",
"noprogress": False,
"quiet": False,
"restrictfilenames": True,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
ydl.download([video_url])
return ydl.prepare_filename(ydl.extract_info(video_url, download=False))
@classmethod
def download_audio(cls, video_url: str, output_dir: str = "downloads") -> str:
"""元の音声を変換せず保存(最良の音声トラックをそのまま)"""
os.makedirs(output_dir, exist_ok=True)z
if not video_url.startswith("http"):
video_url = f"https://www.youtube.com/watch?v={video_url}"
ydl_opts = {
"outtmpl": os.path.join(output_dir, "%(title)s [%(id)s].%(ext)s"),
# 音声トラックを優先的に取る
"format": "bestaudio/best",
"postprocessors": [
{
"key": "FFmpegExtractAudio",
"preferredcodec": "mp3",
"preferredquality": "192", # 128/192/320 kbps
}
],
# 互換性重視で 44.1kHz / ステレオにする
"postprocessor_args": ["-ar", "44100", "-ac", "2"],
"prefer_ffmpeg": True,
"restrictfilenames": True,
"noprogress": False,
"quiet": False,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
# 最終ファイルパスを安全に取得
path = (
(info.get("requested_downloads") or [{}])[0].get("filepath")
or info.get("filepath")
or info.get("_filename")
)
return path