Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix: Failed to download videos after upgrade #17

Merged
merged 9 commits into from
Nov 14, 2024
3 changes: 3 additions & 0 deletions SettingsTemplate.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ body:
- mkv
- avi
- flv
- mov
- webm
- type: dropdown
attributes:
Expand All @@ -39,4 +40,6 @@ body:
- m4a
- wav
- flac
- aac
- opus

180 changes: 76 additions & 104 deletions plugin/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,48 +4,41 @@
# Date: 2024-07-28

import os
import re
import subprocess
from datetime import datetime, timedelta
from pathlib import Path
from typing import Tuple

from pyflowlauncher import Plugin, ResultResponse, send_results
from pyflowlauncher.settings import settings
from utils import (
is_valid_url,
sort_by_resolution,
sort_by_tbr,
sort_by_fps,
sort_by_size,
verify_ffmpeg_binaries,
verify_ffmpeg,
extract_ffmpeg,
)
from results import (
init_results,
invalid_result,
error_result,
empty_result,
query_result,
download_ffmpeg_result,
ffmpeg_not_found_result,
)
from ytdlp import CustomYoutubeDL

EXE_PATH = os.path.join(os.path.dirname(__file__), "yt-dlp.exe")
CHECK_INTERVAL_DAYS = 10
PLUGIN_ROOT = os.path.dirname(__file__)
EXE_PATH = os.path.join(PLUGIN_ROOT, "yt-dlp.exe")
CHECK_INTERVAL_DAYS = 5
DEFAULT_DOWNLOAD_PATH = str(Path.home() / "Downloads")
URL_REGEX = (
"((http|https)://)(www.)?"
+ "[a-zA-Z0-9@:%._\\+~#?&//=]"
+ "{1,256}\\.[a-z]"
+ "{2,6}\\b([-a-zA-Z0-9@:%"
+ "._\\+~#?&//=]*)"
)

plugin = Plugin()


def is_valid_url(url: str) -> bool:
"""
Check if the given URL is valid based on a predefined regex pattern.

Args:
url (str): The URL string to validate.

Returns:
bool: True if the URL matches the regex pattern, False otherwise.
"""

return bool(re.match(URL_REGEX, url))
plugin = Plugin()


def fetch_settings() -> Tuple[str, str, str, str]:
Expand Down Expand Up @@ -76,74 +69,15 @@ def fetch_settings() -> Tuple[str, str, str, str]:
return download_path, sorting_order, pref_video_format, pref_audio_format


def sort_by_resolution(formats):
"""
Sorts a list of video formats by their resolution in descending order.

Returns:
list of dict: The list of video formats sorted by resolution in descending order.
Audio-only formats are considered to have the lowest resolution.
"""

def resolution_to_tuple(resolution):
if resolution == "audio only":
return (0, 0)
return tuple(map(int, resolution.split("x")))

return sorted(
formats, key=lambda x: resolution_to_tuple(x["resolution"]), reverse=True
)


def sort_by_tbr(formats):
"""
Sort a list of video formats by their 'tbr' (total bitrate) in descending order.

Returns:
list: The input list sorted by the 'tbr' value in descending order.
"""
return sorted(formats, key=lambda x: x["tbr"], reverse=True)


def sort_by_fps(formats):
"""
Sort a list of video formats by frames per second (fps) in descending order.

Returns:
list of dict: The list of video formats sorted by fps in descending order.
Formats with None fps values are placed at the end.
"""
return sorted(
formats,
key=lambda x: (
x["fps"] is None,
-x["fps"] if x["fps"] is not None else float("-inf"),
),
)


def sort_by_size(formats):
"""
Sort a list of video formats by their file size in descending order.

Returns:
list of dict: The list of video formats sorted by file size in
descending order. Formats with no file size information
(None) are placed at the end of the list.
"""
return sorted(
formats,
key=lambda x: (
x["filesize"] is None,
-x["filesize"] if x["filesize"] is not None else float("-inf"),
),
)


@plugin.on_method
def query(query: str) -> ResultResponse:
d_path, sort, pvf, paf = fetch_settings()

if verify_ffmpeg():
return send_results([download_ffmpeg_result(PLUGIN_ROOT)])

extract_ffmpeg()

if not query.strip():
return send_results([init_results(d_path)])

Expand Down Expand Up @@ -186,21 +120,37 @@ def query(query: str) -> ResultResponse:
elif sort == "FPS":
formats = sort_by_fps(formats)

results = [
query_result(
query,
str(info.get("thumbnail")),
str(info.get("title")),
format,
d_path,
pvf,
paf,
)
for format in formats
]
results = []
if verify_ffmpeg_binaries():
results.extend([ffmpeg_not_found_result()])
results.extend(
[
query_result(
query,
str(info.get("thumbnail")),
str(info.get("title")),
format,
d_path,
pvf,
paf,
)
for format in formats
]
)
return send_results(results)


@plugin.on_method
def download_ffmpeg_binaries(PLUGIN_ROOT) -> None:
BIN_URL = (
"https://github.com/z1nc0r3/ffmpeg-binaries/blob/main/ffmpeg-bin.zip?raw=true"
)
FFMPEG_ZIP = os.path.join(PLUGIN_ROOT, "ffmpeg.zip")
command = f'curl -L "{BIN_URL}" -o "{FFMPEG_ZIP}"'

subprocess.run(command)


@plugin.on_method
def download(
url: str,
Expand All @@ -211,11 +161,13 @@ def download(
is_audio: bool,
) -> None:
last_modified_time = datetime.fromtimestamp(os.path.getmtime(EXE_PATH))
exe_path = os.path.join(os.path.dirname(__file__), "yt-dlp.exe")
ffmpeg_path = os.path.dirname(__file__)

format = (
f"-f ba -x --audio-format {pref_audio_path}"
f"-f b -x --audio-format {pref_audio_path} --audio-quality 0"
if is_audio
else f"-f {format_id}+ba --remux-video {pref_video_path}"
else f"-f {format_id}+ba[ext=mp3]/{format_id}+ba[ext=aac]/{format_id}+ba[ext=m4a]/{format_id}+ba[ext=wav]/{format_id}+ba --remux-video {pref_video_path}"
)

update = (
Expand All @@ -224,9 +176,29 @@ def download(
else ""
)

command = f'yt-dlp "{url}" {format} -P {download_path} --windows-filenames --restrict-filenames --trim-filenames 50 --quiet --progress --no-mtime --force-overwrites --no-part {update}'
command = [
exe_path,
url,
*format.split(),
"-P",
download_path,
"--windows-filenames",
"--restrict-filenames",
"--trim-filenames",
"50",
"--quiet",
"--progress",
"--no-mtime",
"--force-overwrites",
"--no-part",
"--ffmpeg-location",
ffmpeg_path,
update,
]

command = [arg for arg in command if arg]

os.system(command)
subprocess.run(command)


if __name__ == "__main__":
Expand Down
18 changes: 17 additions & 1 deletion plugin/results.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,18 @@ def init_results(download_path) -> Result:
def invalid_result() -> Result:
return Result(Title="Please check the URL for errors.", IcoPath="Images/error.png")

def ffmpeg_not_found_result() -> Result:
return Result(
Title="FFmpeg is not installed!",
SubTitle="Some features may not work as expected.",
IcoPath="Images/error.png",
)


def error_result() -> Result:
return Result(
Title="Something went wrong!",
SubTitle="Couldn't extract video information.",
SubTitle=f"Couldn't extract video information.",
IcoPath="Images/error.png",
)

Expand All @@ -25,6 +32,15 @@ def empty_result() -> Result:
return Result(Title="Couldn't find any video formats.", IcoPath="Images/error.png")


def download_ffmpeg_result(dest_path) -> Result:
return Result(
Title="FFmpeg is not installed!",
SubTitle="Click this to download FFmpeg binaries.",
IcoPath="Images/error.png",
JsonRPCAction={"method": "download_ffmpeg_binaries", "parameters": [dest_path]},
)


def query_result(
query, thumbnail, title, format, download_path, pref_video_path, pref_audio_path
) -> Result:
Expand Down
Loading