Migrate to pyflowlauncher package & Breaking changes

plugin.json

@@ -8,5 +8,5 @@
     "Language": "python",
     "Website": "https://github.com/z1nc0r3/AnyVideo-Downloader-Flow-Plugin",
     "IcoPath": "Images\\app.png",
-    "ExecuteFileName": "main.py"
+    "ExecuteFileName": "run.py"
 }

plugin/main.py

@@ -0,0 +1,99 @@
+# Author: Lasith Manujitha
+# Github: @z1nc0r3
+# Description: A plugin to download videos from multiple websites
+# Date: 2024-07-28
+
+import os
+import re
+import shutil
+from datetime import datetime, timedelta
+from pathlib import Path
+
+from pyflowlauncher import Plugin, ResultResponse, send_results
+from pyflowlauncher.settings import settings
+from results import (
+    init_results,
+    invalid_result,
+    error_result,
+    empty_result,
+    query_result,
+)
+from ytdlp import CustomYoutubeDL
+
+parent_folder_path = os.path.abspath(os.path.dirname(__file__))
+EXE_PATH = os.path.join(parent_folder_path, "yt-dlp.exe")
+CHECK_INTERVAL_DAYS = 10
+download_path = str(Path.home()) + "\\Downloads"
+
+plugin = Plugin()
+
+
+def is_valid_url(url: str) -> bool:
+    regex = (
+        "((http|https)://)(www.)?"
+        + "[a-zA-Z0-9@:%._\\+~#?&//=]"
+        + "{1,256}\\.[a-z]"
+        + "{2,6}\\b([-a-zA-Z0-9@:%"
+        + "._\\+~#?&//=]*)"
+    )
+
+    p = re.compile(regex)
+
+    return re.match(p, url)
+
+
+@plugin.on_method
+def query(query: str) -> ResultResponse:
+    download_path = settings().get("download_path", str(Path.home()) + "\\Downloads")
+
+    if not query.strip():
+        return send_results([init_results()])
+
+    if not is_valid_url(query):
+        return send_results([invalid_result()])
+
+    if query.startswith("https://"):
+        query = query.replace("https://", "http://")
+
+    ydl_opts = {
+        "quiet": True,
+        "no_warnings": True,
+        "format-sort": "res,tbr",
+    }
+    ydl = CustomYoutubeDL(params=ydl_opts)
+    info = ydl.extract_info(query)
+
+    if ydl.error_message:
+        return send_results([error_result()])
+
+    formats = [
+        format
+        for format in info["formats"]
+        if format.get("resolution") and format.get("tbr")
+    ]
+
+    if not formats:
+        return send_results([empty_result()])
+
+    results = [query_result(query, info, format) for format in reversed(formats)]
+
+    return send_results(results)
+
+
+@plugin.on_method
+def download(url: str, format_id: str) -> None:
+    last_modified_time = datetime.fromtimestamp(os.path.getmtime(EXE_PATH))
+
+    base_command = f'yt-dlp "{url}" -f {format_id}+ba -P {download_path} --windows-filenames --restrict-filenames --trim-filenames 50 --quiet --progress --no-mtime --force-overwrites --no-part'
+
+    command = (
+        f"{base_command} -U"
+        if datetime.now() - last_modified_time >= timedelta(days=CHECK_INTERVAL_DAYS)
+        else base_command
+    )
+
+    os.system(command)
+
+
+if __name__ == "__main__":
+    plugin.run()

plugin/results.py

@@ -0,0 +1,33 @@
+from pyflowlauncher import Result
+
+
+def init_results() -> Result:
+    return Result(Title="Please input the URL of the video", IcoPath="Images/app.png")
+
+
+def invalid_result() -> Result:
+    return Result(Title="Please check the URL for errors.", IcoPath="Images/error.png")
+
+
+def error_result() -> Result:
+    return Result(
+        Title="Something went wrong!",
+        SubTitle="Couldn't extract video information.",
+        IcoPath="Images/error.png",
+    )
+
+
+def empty_result() -> Result:
+    return Result(Title="Couldn't find any video formats.", IcoPath="Images/error.png")
+
+
+def query_result(query, info, format) -> Result:
+    return Result(
+        Title=info["title"],
+        SubTitle=f"{format['resolution']} ({round(format['tbr'])} kbps) {'┃ Format: ' + str(format['ext']) if format.get('ext') else ''} {'┃ FPS: ' + str(format['fps']) if format.get('fps') else ''}",
+        IcoPath=info.get("thumbnail") or "Images/app.png",
+        JsonRPCAction={
+            "method": "download",
+            "parameters": [query, f"{format['format_id']}"],
+        },
+    )

plugin/ytdlp.py

@@ -0,0 +1,29 @@
+from yt_dlp import YoutubeDL
+
+
+# Custom YoutubeDL class to exception handling
+class CustomYoutubeDL(YoutubeDL):
+    def __init__(self, params=None, auto_init=True):
+        super().__init__(params, auto_init)
+        self.error_message = None
+
+    def report_error(self, message, *args, **kwargs):
+        self.error_message = message
+
+    def extract_info(
+        self,
+        url,
+        download=False,
+        ie_key=None,
+        extra_info=None,
+        process=True,
+        force_generic_extractor=False,
+    ):
+        try:
+            result = super().extract_info(
+                url, download, ie_key, extra_info, process, force_generic_extractor
+            )
+            return result
+        except Exception as e:
+            self.error_message = f"Unexpected error: {str(e)}"
+            return None

requirements.txt

@@ -1,2 +1,2 @@
-flowlauncher
+pyflowlauncher[all]
 yt-dlp

run.py

@@ -0,0 +1,12 @@
+import sys
+import os
+
+plugindir = os.path.abspath(os.path.dirname(__file__))
+sys.path.append(plugindir)
+sys.path.append(os.path.join(plugindir, "lib"))
+sys.path.append(os.path.join(plugindir, "plugin"))
+
+
+if __name__ == "__main__":
+    from plugin.main import plugin
+    plugin.run()