|
#!/usr/bin/env python3 |
|
import os |
|
import sys |
|
import json |
|
import subprocess |
|
import urllib.parse |
|
import urllib.request |
|
from pathlib import Path |
|
from concurrent.futures import ThreadPoolExecutor |
|
|
|
# --- Config --- |
|
APP_KEY = "your-api-key-from-klipy" |
|
API_BASE = f"https://api.klipy.com/api/v1/{APP_KEY}/gifs/search" |
|
LIMIT = 15 |
|
CACHE_DIR = Path.home() / ".cache" / "rofi-gifs" |
|
CACHE_DIR.mkdir(parents=True, exist_ok=True) |
|
|
|
# Detect environment |
|
IS_WAYLAND = os.environ.get("WAYLAND_DISPLAY") |
|
COPY_CMD = ["wl-copy"] if IS_WAYLAND else ["xclip", "-selection", "clipboard"] |
|
|
|
# Rofi variables |
|
RETV = int(os.environ.get("ROFI_RETV", "0")) |
|
INFO = os.environ.get("ROFI_INFO", "") |
|
QUERY = os.environ.get("ROFI_INPUT", sys.argv[1] if len(sys.argv) > 1 else "").strip() |
|
|
|
def rofi_print(text): |
|
sys.stdout.write(f"{text}\n") |
|
sys.stdout.flush() |
|
|
|
def download(url, path): |
|
if path.exists(): return True |
|
try: |
|
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) |
|
with urllib.request.urlopen(req, timeout=3) as r: |
|
path.write_bytes(r.read()) |
|
return True |
|
except: return False |
|
|
|
def process_gif(item): |
|
try: |
|
id = item.get("id") |
|
title = (item.get("title") or "GIF").replace("&", "&") |
|
files = item.get("file", {}) |
|
|
|
|
|
thumb = files.get("xs", {}).get("jpg", {}).get("url") |
|
url = files.get("hd", {}).get("gif", {}).get("url") or files.get("gif", {}).get("url") |
|
|
|
if not thumb or not url: return |
|
|
|
path = CACHE_DIR / f"{id}.jpg" |
|
if download(thumb, path): |
|
rofi_print(f"{title}\0icon\x1f{path}\x1finfo\x1f{url}") |
|
except: pass |
|
|
|
def main(): |
|
# 1. Handle selection |
|
if RETV == 1 and INFO: |
|
subprocess.run(COPY_CMD, input=INFO.encode()) |
|
subprocess.run(["notify-send", "-t", "1000", "GIF Picker", "URL Copied"]) |
|
rofi_print("\0keep-selection\x1ftrue") |
|
rofi_print("\0no-custom\x1ftrue") |
|
|
|
# 2. UI Setup |
|
rofi_print("\0prompt\x1fGIFs") |
|
rofi_print("\0markup-rows\x1ftrue") |
|
|
|
if not QUERY: |
|
rofi_print("\0message\x1fStart typing to search...") |
|
return |
|
|
|
# 3. Search logic |
|
try: |
|
url = f"{API_BASE}?q={urllib.parse.quote(QUERY)}&per_page={LIMIT}" |
|
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) |
|
with urllib.request.urlopen(req, timeout=5) as r: |
|
data = json.loads(r.read().decode()) |
|
items = data.get("data", []) |
|
if isinstance(items, dict): items = items.get("data", []) |
|
|
|
with ThreadPoolExecutor(max_workers=10) as exe: |
|
exe.map(process_gif, items) |
|
except Exception as e: |
|
rofi_print(f"\0message\x1fSearch failed: {e}") |
|
|
|
if __name__ == "__main__": |
|
main() |
did you like it ??