133 lines
4.6 KiB
Python
133 lines
4.6 KiB
Python
import fnmatch
|
|
import io
|
|
import os
|
|
import pathlib
|
|
import shutil
|
|
|
|
import requests
|
|
import send2trash
|
|
|
|
WEBHOOK_URL = os.environ.get("RANDOM_FILE_PICKER_WEBHOOK_URL", None)
|
|
MOVE_FOLDER = os.environ.get("RANDOM_FILE_PICKER_MOVE_FOLDER", None)
|
|
|
|
IMG_EXTENSIONS = [
|
|
".png",
|
|
".jpg", ".jpeg",
|
|
".webp",
|
|
".avif",
|
|
".jxl",
|
|
# kind-of, can be opened with image viewer
|
|
".gif",
|
|
]
|
|
|
|
SUPPORTED_EXTENSIONS = [
|
|
*IMG_EXTENSIONS,
|
|
".mp4",
|
|
".m4v",
|
|
".mov",
|
|
".mkv",
|
|
".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
|
|
_ = shutil.move(file_path, destination_path)
|
|
|
|
def trash_file(file_path: pathlib.Path) -> None:
|
|
"""
|
|
Put the specified file into the system trash using the send2trash library.
|
|
|
|
:param file_path: The path of the file to be trashed.
|
|
"""
|
|
send2trash.send2trash(file_path)
|