import fnmatch import io import os import pathlib import requests WEBHOOK_URL = os.environ.get("RANDOM_FILE_PICKER_WEBHOOK_URL", None) MOVE_FOLDER = os.environ.get("RANDOM_FILE_PICKER_MOVE_FOLDER", None) SUPPORTED_EXTENSIONS = [ ".jpg", ".jpeg", ".webp", ".png", ".gif", ".mp4", ".mov", ".webm", ] def collect_files( *, search_path: pathlib.Path, search_subdirectories: bool, file_size_limit: int, blacklist_patterns: list[str] | None = None, whitelist_patterns: list[str] | None = None, ) -> list[pathlib.Path]: """ Collect files from the specified search_path, applying filters based on file size, blacklist patterns, and whitelist patterns. :param search_path: The directory path to search for files. :param search_subdirectories: If True, search in subdirectories as well. :param file_size_limit: The maximum file size (in MB) for files to be included in the results. :param blacklist_patterns: A list of patterns to exclude files from the results. If None, no files will be excluded based on patterns. :param whitelist_patterns: A list of patterns to include files in the results. If None, all files will be included regardless of patterns. :return: A list of file paths that meet the specified criteria. """ collected_files: list[pathlib.Path] = [] # Set default values for blacklist and whitelist patterns if they are None blacklist_patterns = blacklist_patterns or [] whitelist_patterns = whitelist_patterns or [] search_path = search_path.resolve() for directory, _, files in search_path.walk(): if directory != search_path and not search_subdirectories: continue for f in files: file_path = directory / f # Check against whitelist patterns if whitelist_patterns and not any(fnmatch.fnmatch(str(file_path), pattern) for pattern in whitelist_patterns): continue # Check against blacklist patterns if any(fnmatch.fnmatch(str(file_path), pattern) for pattern in blacklist_patterns): continue # Make sure the file has a supported extension if not any(f.lower().endswith(ext) for ext in SUPPORTED_EXTENSIONS): continue stat_res = file_path.stat() if stat_res.st_size < 1024 * 1024 * file_size_limit: collected_files.append(file_path) return collected_files def upload_file(file_path: pathlib.Path) -> None: """ Upload the specified file to a Discord webhook. :param file_path: The path of the file to be uploaded. """ if WEBHOOK_URL is None: raise ValueError("RANDOM_FILE_PICKER_WEBHOOK_URL environment variable is not set.") filename = file_path.name with file_path.open("rb") as image_file: image_data = image_file.read() files = {"file": (filename, io.BytesIO(image_data))} response = requests.post(WEBHOOK_URL, files=files, data={}, timeout=60*5) response.raise_for_status() # Raise HTTPError for bad responses def move_file_to_move_folder(file_path: pathlib.Path) -> None: """ Move the specified file to the move folder defined by the RANDOM_FILE_PICKER_MOVE_FOLDER environment variable. :param file_path: The path of the file to be moved. """ if MOVE_FOLDER is None: raise ValueError("RANDOM_FILE_PICKER_MOVE_FOLDER environment variable is not set.") move_dir_path = pathlib.Path(MOVE_FOLDER) # Make sure the move folder exists if not move_dir_path.exists(): raise RuntimeError(f"Move folder '{move_dir_path}' does not exist.") # Make sure the file being moved is not already in the move folder if file_path.parent.resolve() == move_dir_path.resolve(): raise RuntimeError("File is already in the move folder.") # Formulate the destination path and avoid overwriting existing files base_name = file_path.name destination_path = move_dir_path / base_name if destination_path.exists(): name, ext = destination_path.stem, destination_path.suffix counter = 1 while destination_path.exists(): destination_path = move_dir_path / f"{name}_{counter}{ext}" counter += 1 _ = file_path.rename(destination_path)