feat: add support for opening images in geeqie
This commit is contained in:
@@ -9,3 +9,5 @@
|
|||||||
!random_file_picker/
|
!random_file_picker/
|
||||||
!random_file_picker/__init__.py
|
!random_file_picker/__init__.py
|
||||||
!random_file_picker/cli.py
|
!random_file_picker/cli.py
|
||||||
|
!random_file_picker/core.py
|
||||||
|
!random_file_picker/openers.py
|
||||||
|
|||||||
+27
-114
@@ -7,30 +7,16 @@ Here's an example usage:
|
|||||||
./rand.py -fl 5 -d /path/to/directory --descend --blacklist "*.png" "*.mp4"
|
./rand.py -fl 5 -d /path/to/directory --descend --blacklist "*.png" "*.mp4"
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import io
|
|
||||||
import os
|
|
||||||
import random
|
|
||||||
import requests
|
|
||||||
import subprocess
|
|
||||||
import argparse
|
import argparse
|
||||||
|
import os
|
||||||
|
import pathlib
|
||||||
|
import random
|
||||||
import sys
|
import sys
|
||||||
import tty
|
|
||||||
import termios
|
import termios
|
||||||
import fnmatch
|
import tty
|
||||||
|
|
||||||
MEME_FOLDER = os.environ.get("RANDOM_FILE_PICKER_MEME_FOLDER", None)
|
from random_file_picker.core import collect_files, move_file_to_meme_folder, upload_file
|
||||||
WEBHOOK_URL = os.environ.get("RANDOM_FILE_PICKER_WEBHOOK_URL", None)
|
from random_file_picker.openers import open_file_with_geeqie, open_path_in_dolphin, open_path_with_default_app
|
||||||
|
|
||||||
ALLOWED_EXTENSIONS = [
|
|
||||||
".jpg",
|
|
||||||
".jpeg",
|
|
||||||
".webp",
|
|
||||||
".png",
|
|
||||||
".gif",
|
|
||||||
".mp4",
|
|
||||||
".mov",
|
|
||||||
".webm",
|
|
||||||
]
|
|
||||||
|
|
||||||
IMG_EXTENSIONS = [
|
IMG_EXTENSIONS = [
|
||||||
".jpg",
|
".jpg",
|
||||||
@@ -40,20 +26,6 @@ IMG_EXTENSIONS = [
|
|||||||
".gif",
|
".gif",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
def send_file(file_path: str):
|
|
||||||
if WEBHOOK_URL is None:
|
|
||||||
print("Error: RANDOM_FILE_PICKER_WEBHOOK_URL environment variable is not set.")
|
|
||||||
return
|
|
||||||
|
|
||||||
filename = os.path.basename(file_path)
|
|
||||||
with open(file_path, "rb") as image_file:
|
|
||||||
image_data = image_file.read()
|
|
||||||
files = {'file': (filename, io.BytesIO(image_data))}
|
|
||||||
response = requests.post(WEBHOOK_URL, files=files, data={})
|
|
||||||
response.raise_for_status() # Raise HTTPError for bad responses
|
|
||||||
print(f"\nSuccessfully uploaded {filename}")
|
|
||||||
|
|
||||||
def getch() -> str:
|
def getch() -> str:
|
||||||
fd = sys.stdin.fileno()
|
fd = sys.stdin.fileno()
|
||||||
old_settings = termios.tcgetattr(fd)
|
old_settings = termios.tcgetattr(fd)
|
||||||
@@ -64,25 +36,12 @@ def getch() -> str:
|
|||||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||||
return ch
|
return ch
|
||||||
|
|
||||||
|
|
||||||
def open_with_gwenview(maybe_process: subprocess.Popen[bytes] | None, file_path: str) -> subprocess.Popen[bytes]:
|
|
||||||
if maybe_process is not None:
|
|
||||||
maybe_process.terminate()
|
|
||||||
maybe_process.wait()
|
|
||||||
|
|
||||||
process = subprocess.Popen(["gwenview", file_path])
|
|
||||||
return process
|
|
||||||
|
|
||||||
def open_in_folder(file_path: str):
|
|
||||||
subprocess.Popen(["dolphin", "--select", file_path])
|
|
||||||
#subprocess.run(["dolphin", "--select", file_path])
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
parser = argparse.ArgumentParser(description="Random File Sender")
|
parser = argparse.ArgumentParser(description="Random File Sender")
|
||||||
parser.add_argument('-fl', '--file-limit', type=int, default=10, help='Maximum file size in MB')
|
parser.add_argument('-fl', '--file-limit', type=int, default=10, help='Maximum file size in MB')
|
||||||
parser.add_argument('-d', '--directory', type=str, default='.', help='Directory to search for files')
|
parser.add_argument('-d', '--directory', type=str, default='.', help='Directory to search for files')
|
||||||
parser.add_argument('--descend', action='store_true', help='Whether to descend into subdirectories')
|
parser.add_argument('--descend', action='store_true', help='Whether to descend into subdirectories')
|
||||||
#parser.add_argument('--direct', action='store_true', help='Whether to open files directly without xdg-open')
|
parser.add_argument('--use-geeqie', action='store_true', help='Whether to use Geeqie as the image viewer')
|
||||||
|
|
||||||
parser._positionals.title = 'Positional arguments'
|
parser._positionals.title = 'Positional arguments'
|
||||||
parser._optionals.title = 'Optional arguments'
|
parser._optionals.title = 'Optional arguments'
|
||||||
@@ -94,65 +53,44 @@ def main():
|
|||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
file_limit: int = args.file_limit
|
file_limit: int = args.file_limit
|
||||||
#open_directly: bool = args.direct
|
|
||||||
directory: str = args.directory
|
directory: str = args.directory
|
||||||
descend: bool = args.descend
|
descend: bool = args.descend
|
||||||
blacklist_patterns: list[str] = args.blacklist if args.blacklist else []
|
blacklist_patterns: list[str] = args.blacklist if args.blacklist else []
|
||||||
whitelist_patterns: list[str] = args.whitelist if args.whitelist else []
|
whitelist_patterns: list[str] = args.whitelist if args.whitelist else []
|
||||||
|
use_geeqie: bool = args.use_geeqie
|
||||||
print(f"File size limit set to {file_limit} MB")
|
print(f"File size limit set to {file_limit} MB")
|
||||||
#print(f"Open files directly: {open_directly}")
|
|
||||||
if not os.path.exists(directory) or not os.path.isdir(directory):
|
search_path = pathlib.Path(directory)
|
||||||
|
if not search_path.exists() or not search_path.is_dir():
|
||||||
print(f"Directory '{directory}' does not exist or is not a directory.")
|
print(f"Directory '{directory}' does not exist or is not a directory.")
|
||||||
return
|
return
|
||||||
|
|
||||||
search_path = os.path.abspath(directory)
|
|
||||||
|
|
||||||
# Collect files from the current directory
|
|
||||||
usable_files: list[str] = []
|
|
||||||
for directory, _, files in os.walk(search_path):
|
|
||||||
if directory != search_path and not descend:
|
|
||||||
continue
|
|
||||||
|
|
||||||
for f in files:
|
|
||||||
file_path = os.path.join(directory, f)
|
|
||||||
|
|
||||||
# Check against whitelist patterns
|
|
||||||
if whitelist_patterns and not any(fnmatch.fnmatch(file_path, pattern) for pattern in whitelist_patterns):
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Check against blacklist patterns
|
|
||||||
if any(fnmatch.fnmatch(file_path, pattern) for pattern in blacklist_patterns):
|
|
||||||
continue
|
|
||||||
|
|
||||||
if not any(f.lower().endswith(ext) for ext in ALLOWED_EXTENSIONS):
|
|
||||||
continue
|
|
||||||
|
|
||||||
stat_res = os.stat(file_path)
|
|
||||||
if stat_res.st_size < 1024 * 1024 * file_limit:
|
|
||||||
usable_files.append(file_path)
|
|
||||||
|
|
||||||
|
usable_files = collect_files(
|
||||||
|
search_path=search_path,
|
||||||
|
search_subdirectories=descend,
|
||||||
|
file_size_limit=file_limit,
|
||||||
|
blacklist_patterns=blacklist_patterns,
|
||||||
|
whitelist_patterns=whitelist_patterns,
|
||||||
|
)
|
||||||
if not usable_files:
|
if not usable_files:
|
||||||
print("No usable files found.")
|
print("No usable files found.")
|
||||||
return
|
return
|
||||||
|
|
||||||
pickFile = True
|
pickFile = True
|
||||||
gwenview_process = None
|
|
||||||
while pickFile:
|
while pickFile:
|
||||||
file = random.choice(usable_files)
|
file = random.choice(usable_files)
|
||||||
|
|
||||||
print(f"Random file: {file}")
|
print(f"Random file: {file}")
|
||||||
|
if use_geeqie and str(file).lower().endswith(tuple(IMG_EXTENSIONS)):
|
||||||
if file.lower().endswith(tuple(IMG_EXTENSIONS)):
|
open_file_with_geeqie(file)
|
||||||
gwenview_process = open_with_gwenview(gwenview_process, file)
|
|
||||||
else:
|
else:
|
||||||
subprocess.run(["xdg-open", file])
|
open_path_with_default_app(file)
|
||||||
|
|
||||||
|
|
||||||
selectionTime = True
|
selectionTime = True
|
||||||
while selectionTime:
|
while selectionTime:
|
||||||
print("Options: [u]pload, [n]ext, [d]elete, [f]older, [q]uit")
|
print("Options: [u]pload, [n]ext, [d]elete, [f]older, [q]uit")
|
||||||
choice = getch().lower()
|
choice = getch().lower()
|
||||||
|
|
||||||
|
|
||||||
# Skip escape sequences (arrow keys)
|
# Skip escape sequences (arrow keys)
|
||||||
if choice.encode("utf-8") == b'\x1b':
|
if choice.encode("utf-8") == b'\x1b':
|
||||||
@@ -164,7 +102,8 @@ def main():
|
|||||||
|
|
||||||
if choice == "u":
|
if choice == "u":
|
||||||
print("\nUploading...")
|
print("\nUploading...")
|
||||||
send_file(file)
|
upload_file(file)
|
||||||
|
print(f"\nSuccessfully uploaded {file}")
|
||||||
print("Upload complete.")
|
print("Upload complete.")
|
||||||
#selectionTime = False
|
#selectionTime = False
|
||||||
elif choice == "d":
|
elif choice == "d":
|
||||||
@@ -177,40 +116,14 @@ def main():
|
|||||||
selectionTime = False
|
selectionTime = False
|
||||||
usable_files.remove(file)
|
usable_files.remove(file)
|
||||||
elif choice == "f":
|
elif choice == "f":
|
||||||
open_in_folder(file)
|
open_path_in_dolphin(file)
|
||||||
elif choice == "q":
|
elif choice == "q":
|
||||||
print("\nQuitting...")
|
print("\nQuitting...")
|
||||||
pickFile = False
|
pickFile = False
|
||||||
break
|
break
|
||||||
elif choice == "m":
|
elif choice == "m":
|
||||||
print("\nMoving to meme folder...")
|
print("\nMoving to meme folder...")
|
||||||
|
move_file_to_meme_folder(file)
|
||||||
if MEME_FOLDER is None:
|
|
||||||
print("Error: RANDOM_FILE_PICKER_MEME_FOLDER environment variable is not set.")
|
|
||||||
continue
|
|
||||||
|
|
||||||
# First, make sure we're not in the meme folder already
|
|
||||||
if os.path.abspath(os.path.dirname(file)) == os.path.abspath(MEME_FOLDER):
|
|
||||||
print("File is already in the meme folder.")
|
|
||||||
continue
|
|
||||||
|
|
||||||
# Also, ensure the meme folder exists
|
|
||||||
if not os.path.exists(MEME_FOLDER):
|
|
||||||
print(f"Meme folder '{MEME_FOLDER}' does not exist.")
|
|
||||||
continue
|
|
||||||
|
|
||||||
base_name = os.path.basename(file)
|
|
||||||
new_path = os.path.join(MEME_FOLDER, base_name)
|
|
||||||
|
|
||||||
# Avoid overwriting existing files
|
|
||||||
if os.path.exists(new_path):
|
|
||||||
name, ext = os.path.splitext(base_name)
|
|
||||||
counter = 1
|
|
||||||
while os.path.exists(new_path):
|
|
||||||
new_path = os.path.join(MEME_FOLDER, f"{name}_{counter}{ext}")
|
|
||||||
counter += 1
|
|
||||||
|
|
||||||
os.rename(file, new_path)
|
|
||||||
usable_files.remove(file)
|
usable_files.remove(file)
|
||||||
selectionTime = False
|
selectionTime = False
|
||||||
else:
|
else:
|
||||||
@@ -218,4 +131,4 @@ def main():
|
|||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
|
|||||||
@@ -0,0 +1,123 @@
|
|||||||
|
import fnmatch
|
||||||
|
import io
|
||||||
|
import os
|
||||||
|
import pathlib
|
||||||
|
|
||||||
|
import requests
|
||||||
|
|
||||||
|
WEBHOOK_URL = os.environ.get("RANDOM_FILE_PICKER_WEBHOOK_URL", None)
|
||||||
|
MEME_FOLDER = os.environ.get("RANDOM_FILE_PICKER_MEME_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:
|
||||||
|
print("Error: RANDOM_FILE_PICKER_WEBHOOK_URL environment variable is not set.")
|
||||||
|
return
|
||||||
|
|
||||||
|
filename = file_path.name
|
||||||
|
with open(file_path, "rb") as image_file:
|
||||||
|
image_data = image_file.read()
|
||||||
|
files = {"file": (filename, io.BytesIO(image_data))}
|
||||||
|
response = requests.post(WEBHOOK_URL, files=files, data={})
|
||||||
|
response.raise_for_status() # Raise HTTPError for bad responses
|
||||||
|
|
||||||
|
def move_file_to_meme_folder(file_path: pathlib.Path) -> None:
|
||||||
|
"""
|
||||||
|
Move the specified file to the meme folder defined by the RANDOM_FILE_PICKER_MEME_FOLDER environment variable.
|
||||||
|
|
||||||
|
:param file_path: The path of the file to be moved.
|
||||||
|
"""
|
||||||
|
if MEME_FOLDER is None:
|
||||||
|
print("Error: RANDOM_FILE_PICKER_MEME_FOLDER environment variable is not set.")
|
||||||
|
return
|
||||||
|
meme_dir_path = pathlib.Path(MEME_FOLDER)
|
||||||
|
|
||||||
|
# Make sure the meme folder exists
|
||||||
|
if not meme_dir_path.exists():
|
||||||
|
print(f"Error: Meme folder '{meme_dir_path}' does not exist.")
|
||||||
|
return
|
||||||
|
|
||||||
|
# Make sure the file being moved is not already in the meme folder
|
||||||
|
if file_path.parent.resolve() == meme_dir_path.resolve():
|
||||||
|
print("File is already in the meme folder.")
|
||||||
|
return
|
||||||
|
|
||||||
|
# Formulate the destination path and avoid overwriting existing files
|
||||||
|
base_name = file_path.name
|
||||||
|
destination_path = meme_dir_path / base_name
|
||||||
|
if destination_path.exists():
|
||||||
|
name, ext = destination_path.stem, destination_path.suffix
|
||||||
|
counter = 1
|
||||||
|
while destination_path.exists():
|
||||||
|
destination_path = meme_dir_path / f"{name}_{counter}{ext}"
|
||||||
|
counter += 1
|
||||||
|
|
||||||
|
ok = file_path.rename(destination_path)
|
||||||
|
if not ok:
|
||||||
|
print(f"Error moving {file_path} to {destination_path}")
|
||||||
|
else:
|
||||||
|
print(f"Moved {file_path} to {destination_path}")
|
||||||
@@ -0,0 +1,40 @@
|
|||||||
|
import os
|
||||||
|
import pathlib
|
||||||
|
import subprocess
|
||||||
|
|
||||||
|
|
||||||
|
def open_path_with_default_app(file_path: pathlib.Path):
|
||||||
|
"""
|
||||||
|
Open the specified file path with the default application associated with its file type.
|
||||||
|
|
||||||
|
:param file_path: The path of the file to be opened.
|
||||||
|
"""
|
||||||
|
subprocess.run(["/usr/bin/xdg-open", str(file_path)])
|
||||||
|
|
||||||
|
def open_file_with_geeqie(file_path: pathlib.Path):
|
||||||
|
"""
|
||||||
|
Open the specified file path with the Geeqie image viewer.
|
||||||
|
|
||||||
|
:param file_path: The path of the file to be opened.
|
||||||
|
"""
|
||||||
|
print("Running subprocess to open file with Geeqie:", file_path)
|
||||||
|
my_env = os.environ.copy()
|
||||||
|
my_env["GQ_DISABLE_CLUTTER"] = "y"
|
||||||
|
subprocess.Popen(
|
||||||
|
[
|
||||||
|
"/usr/bin/geeqie",
|
||||||
|
# "-t",
|
||||||
|
# "--without-tools",
|
||||||
|
"--file",
|
||||||
|
str(file_path),
|
||||||
|
],
|
||||||
|
env=my_env,
|
||||||
|
)
|
||||||
|
|
||||||
|
def open_path_in_dolphin(file_path: pathlib.Path):
|
||||||
|
"""
|
||||||
|
Open the specified file path in the Dolphin file manager and select the file.
|
||||||
|
|
||||||
|
:param file_path: The path of the file to be opened in Dolphin.
|
||||||
|
"""
|
||||||
|
subprocess.Popen(["/usr/bin/dolphin", "--select", str(file_path)])
|
||||||
Reference in New Issue
Block a user