Watches Windows clipboard for links, and checks attempts download of supported sites via `gallery-dl` or `yt-dlp`

download-gallery-links.py

import os
import sys
import time

from bs4 import BeautifulSoup
from urllib.parse import urlparse

import gallery_dl
from gallery_dl import extractor as gallery_dl_extractor
import yt_dlp
import ctypes
import threading 
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Union, List, Optional
import win32api, win32clipboard, win32con, win32gui

downloads = []
active=True

class Clipboard:
    count=0
    @dataclass
    class Clip:
        type: str
        value: Union[str, List[Path]]

    def __init__(
            self,
            trigger_at_start: bool = False,
            on_text: Callable[[str], None] = None,
            on_html: Callable[[str], None] = None,
            on_update: Callable[[Clip], None] = None
    ):
        self._trigger_at_start = trigger_at_start
        self._on_update = on_update
        self._on_text = on_text
        self._on_html = on_html

    def _create_window(self) -> int:
        """
        Create a window for listening to messages
        :return: window hwnd
        """
        wc = win32gui.WNDCLASS()
        wc.lpfnWndProc = self._process_message
        wc.lpszClassName = self.__class__.__name__
        wc.hInstance = win32api.GetModuleHandle(None)
        class_atom = win32gui.RegisterClass(wc)
        return win32gui.CreateWindow(class_atom, self.__class__.__name__, 0, 0, 0, 0, 0, 0, 0, wc.hInstance, None)

    def _process_message(self, hwnd: int, msg: int, wparam: int, lparam: int):
        WM_CLIPBOARDUPDATE = 0x031D
        if msg == WM_CLIPBOARDUPDATE:
            self._process_clip()
        return 0

    def _process_clip(self):
        clip = self.read_clipboard()
        if not clip:
            return

        if self._on_update:
            self._on_update(clip)
        if clip.type == 'html' and self._on_text:
            self._on_html(clip.value)
        if clip.type == 'text' and self._on_text:
            self._on_text(clip.value)

    @staticmethod
    def read_clipboard() -> Optional[Clip]:
        try:
            win32clipboard.OpenClipboard()

            def get_formatted(fmt):
                if win32clipboard.IsClipboardFormatAvailable(fmt):
                    return win32clipboard.GetClipboardData(fmt)
                return None

            if text_bytes := get_formatted(49472):
                return Clipboard.Clip('html', text_bytes.decode())
            elif text := get_formatted(win32con.CF_UNICODETEXT):
                return Clipboard.Clip('text', text)
            elif text_bytes := get_formatted(win32con.CF_TEXT):
                return Clipboard.Clip('text', text_bytes.decode())
            return None
        finally:
            try:
                win32clipboard.CloseClipboard()
            except BaseException as e:  # to catch pywintypes.error
                if e.winerror == 1418:
                    return

    def listen(self):
        if self._trigger_at_start:
            self._process_clip()

        def runner():
            hwnd = self._create_window()
            ctypes.windll.user32.AddClipboardFormatListener(hwnd)
            win32gui.PumpMessages()

        th = threading.Thread(target=runner, daemon=True)
        th.start()
        while th.is_alive() and active:
            th.join(0.25)

def should_download_html(html: str):
    soup = BeautifulSoup(html, 'html.parser')
    urls = []
    for link in soup.find_all('a'):
        url = link.get('href')
        if 'gallery' not in url:
            continue
        extr = gallery_dl_extractor.find(url)
        if not extr:
            return

        if url not in downloads:
            print(f"calling download thread for: {url}")
            th = threading.Thread(target=download, daemon=True, args=(url,))
            th.start()
        time.sleep(5)
            # while th.is_alive() and active:
            #     th.join(0.25)
def should_download(url: str):

    if '\r\n' in url:
        urls = url.split('\r\n')
        for url in urls:
            should_download(url)
        return
    
    o = urlparse(url)
    if not o.scheme in ['https', 'http']:
        return

    extr = gallery_dl_extractor.find(url)

    if url not in downloads:
        print(f"calling download thread for: {url}")
        if extr:
            th = threading.Thread(target=download, daemon=True, args=(url,"gallery"))
        else:
            th = threading.Thread(target=download, daemon=True, args=(url,"video"))
        th.start()
        # while th.is_alive() and active:
        #     th.join(0.25)

def download(url: str, type):
    downloads.append(url)
    sys.argv = [sys.argv[0], url]
    if type=="gallery":
        print(f"calling gallery-dl download for: {url}")
        gallery_dl.main()
    else:
        print(f"calling yt-dlp download for: {url}")
        sys.argv.extend(["--cookies-from-browser", "firefox"])
        yt_dlp.main()
    print(f"{url} Done!")
    downloads.remove(url)

def quit(self):
    os.system('taskkill /F /PID %d' % os.getpid())

if __name__ == '__main__':
    clipboard = Clipboard(on_text=should_download, on_html=should_download_html, trigger_at_start=False)
    clipboard.listen()