first commit

This commit is contained in:
Jefferyhcool
2025-04-13 17:44:54 +08:00
commit 0e0b8da317
112 changed files with 4397 additions and 0 deletions
View File
+38
View File
@@ -0,0 +1,38 @@
import enum
from abc import ABC, abstractmethod
from typing import Optional, Union
from app.enmus.note_enums import DownloadQuality
from app.models.notes_model import AudioDownloadResult
from os import getenv
QUALITY_MAP = {
"fast": "32",
"medium": "64",
"slow": "128"
}
class Downloader(ABC):
def __init__(self):
#TODO 需要修改为可配置
self.quality = QUALITY_MAP.get('fast')
self.cache_data=getenv('DATA_DIR')
@abstractmethod
def download(self, video_url: str, output_dir: str = None,
quality: DownloadQuality = "fast", need_video: Optional[bool] = False) -> AudioDownloadResult:
'''
:param need_video:
:param video_url: 资源链接
:param output_dir: 输出路径 默认根目录data
:param quality: 音频质量 fast | medium | slow
:return:返回一个 AudioDownloadResult 类
'''
pass
@staticmethod
def download_video(self, video_url: str,
output_dir: Union[str, None] = None) -> str:
pass
@@ -0,0 +1,104 @@
import os
from abc import ABC
from typing import Union, Optional
import yt_dlp
from app.downloaders.base import Downloader, DownloadQuality, QUALITY_MAP
from app.models.notes_model import AudioDownloadResult
from app.utils.path_helper import get_data_dir
class BilibiliDownloader(Downloader, ABC):
def __init__(self):
super().__init__()
def download(
self,
video_url: str,
output_dir: Union[str, None] = None,
quality: DownloadQuality = "fast",
need_video:Optional[bool]=False
) -> AudioDownloadResult:
if output_dir is None:
output_dir = get_data_dir()
if not output_dir:
output_dir=self.cache_data
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
ydl_opts = {
'format': 'bestaudio[ext=m4a]/bestaudio/best',
'outtmpl': output_path,
'postprocessors': [
{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '64',
}
],
'noplaylist': True,
'quiet': False,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
video_id = info.get("id")
title = info.get("title")
duration = info.get("duration", 0)
cover_url = info.get("thumbnail")
audio_path = os.path.join(output_dir, f"{video_id}.mp3")
return AudioDownloadResult(
file_path=audio_path,
title=title,
duration=duration,
cover_url=cover_url,
platform="bilibili",
video_id=video_id,
raw_info=info,
video_path=None # ❗音频下载不包含视频路径
)
def download_video(
self,
video_url: str,
output_dir: Union[str, None] = None,
) -> str:
"""
下载视频,返回视频文件路径
"""
if output_dir is None:
output_dir = get_data_dir()
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
ydl_opts = {
'format': 'bv*[ext=mp4]/bestvideo+bestaudio/best',
'outtmpl': output_path,
'noplaylist': True,
'quiet': False,
'merge_output_format': 'mp4', # 确保合并成 mp4
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
video_id = info.get("id")
video_path = os.path.join(output_dir, f"{video_id}.mp4")
if not os.path.exists(video_path):
raise FileNotFoundError(f"视频文件未找到: {video_path}")
return video_path
def delete_video(self, video_path: str) -> str:
"""
删除视频文件
"""
if os.path.exists(video_path):
os.remove(video_path)
return f"视频文件已删除: {video_path}"
else:
return f"视频文件未找到: {video_path}"
+1
View File
@@ -0,0 +1 @@
# def download():
@@ -0,0 +1,90 @@
import os
from abc import ABC
from typing import Union, Optional
import yt_dlp
from app.downloaders.base import Downloader, DownloadQuality
from app.models.notes_model import AudioDownloadResult
from app.utils.path_helper import get_data_dir
class DouyinDownloader(Downloader, ABC):
def download(
self,
video_url: str,
output_dir: Union[str, None] = None,
quality: DownloadQuality = "fast",
need_video:Optional[bool]=False
) -> AudioDownloadResult:
if output_dir is None:
output_dir = get_data_dir()
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
ydl_opts = {
'format': 'bestaudio[ext=m4a]/bestaudio/best',
'outtmpl': output_path,
'postprocessors': [
{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '64',
}
],
'noplaylist': True,
'quiet': False,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
video_id = info.get("id")
title = info.get("title")
duration = info.get("duration", 0)
cover_url = info.get("thumbnail")
audio_path = os.path.join(output_dir, f"{video_id}.mp3")
return AudioDownloadResult(
file_path=audio_path,
title=title,
duration=duration,
cover_url=cover_url,
platform="douyin",
video_id=video_id,
raw_info={'tags':info.get('tags')}, #全部返回会报错
video_path=None # ❗音频下载不包含视频路径
)
def download_video(
self,
video_url: str,
output_dir: Union[str, None] = None,
) -> str:
"""
下载视频,返回视频文件路径
"""
if output_dir is None:
output_dir = get_data_dir()
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
ydl_opts = {
'format': 'worst[ext=mp4]/worst',
'outtmpl': output_path,
'noplaylist': True,
'quiet': False,
'merge_output_format': 'mp4', # 确保合并成 mp4
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
video_id = info.get("id")
video_path = os.path.join(output_dir, f"{video_id}.mp4")
if not os.path.exists(video_path):
raise FileNotFoundError(f"视频文件未找到: {video_path}")
return video_path
@@ -0,0 +1,95 @@
import os
from abc import ABC
from typing import Union, Optional
import yt_dlp
from app.downloaders.base import Downloader, DownloadQuality
from app.models.notes_model import AudioDownloadResult
from app.utils.path_helper import get_data_dir
class YoutubeDownloader(Downloader, ABC):
def __init__(self):
super().__init__()
def download(
self,
video_url: str,
output_dir: Union[str, None] = None,
quality: DownloadQuality = "fast",
need_video:Optional[bool]=False
) -> AudioDownloadResult:
if output_dir is None:
output_dir = get_data_dir()
if not output_dir:
output_dir=self.cache_data
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
ydl_opts = {
'format': 'bestaudio[ext=m4a]/bestaudio/best',
'outtmpl': output_path,
'postprocessors': [
{
'key': 'FFmpegExtractAudio',
'preferredcodec': 'mp3',
'preferredquality': '64',
}
],
'noplaylist': True,
'quiet': False,
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
video_id = info.get("id")
title = info.get("title")
duration = info.get("duration", 0)
cover_url = info.get("thumbnail")
audio_path = os.path.join(output_dir, f"{video_id}.mp3")
return AudioDownloadResult(
file_path=audio_path,
title=title,
duration=duration,
cover_url=cover_url,
platform="youtube",
video_id=video_id,
raw_info={'tags':info.get('tags')}, #全部返回会报错
video_path=None # ❗音频下载不包含视频路径
)
def download_video(
self,
video_url: str,
output_dir: Union[str, None] = None,
) -> str:
"""
下载视频,返回视频文件路径
"""
if output_dir is None:
output_dir = get_data_dir()
os.makedirs(output_dir, exist_ok=True)
output_path = os.path.join(output_dir, "%(id)s.%(ext)s")
ydl_opts = {
'format': 'worst[ext=mp4]/worst',
'outtmpl': output_path,
'noplaylist': True,
'quiet': False,
'merge_output_format': 'mp4', # 确保合并成 mp4
}
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
info = ydl.extract_info(video_url, download=True)
video_id = info.get("id")
video_path = os.path.join(output_dir, f"{video_id}.mp4")
if not os.path.exists(video_path):
raise FileNotFoundError(f"视频文件未找到: {video_path}")
return video_path