2619 lines
105 KiB
Python
2619 lines
105 KiB
Python
#!/usr/bin/env python3
|
|
import os
|
|
import sys
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import urllib.request
|
|
import urllib.error
|
|
import time
|
|
import mimetypes
|
|
import urllib.parse
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import ssl
|
|
import threading
|
|
import signal
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
|
|
from PySide6.QtCore import Qt, QThread, QObject, Signal, QUrl, QFile, QIODevice, QTimer, QMimeData, QProcess, QRunnable, QThreadPool
|
|
from PySide6.QtGui import QIcon, QAction, QActionGroup, QPixmap, QPainter, QColor, QFont, QPen, QImage, QDesktopServices, QDrag, QClipboard
|
|
from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QHttpMultiPart, QHttpPart, QNetworkReply, QLocalServer, QLocalSocket
|
|
from PySide6.QtWidgets import (
|
|
QApplication, QSystemTrayIcon, QMenu, QDialog, QVBoxLayout, QHBoxLayout,
|
|
QLabel, QLineEdit, QComboBox, QCheckBox, QPushButton, QGroupBox,
|
|
QFormLayout, QMessageBox, QFileDialog, QSlider, QScrollArea, QFrame, QGridLayout, QWidget, QSizePolicy, QInputDialog
|
|
)
|
|
|
|
def normalize_file_path(path_str: str) -> str:
|
|
if not path_str:
|
|
return ""
|
|
path_str = path_str.strip().strip("'\"")
|
|
if path_str.startswith("file://"):
|
|
url = QUrl(path_str)
|
|
if url.isLocalFile():
|
|
path_str = url.toLocalFile()
|
|
else:
|
|
path_str = urllib.parse.unquote(path_str[7:])
|
|
return os.path.abspath(path_str)
|
|
|
|
def normalize_api_url(url: str) -> str:
|
|
if not url:
|
|
return ""
|
|
url = url.strip().strip("'\"")
|
|
if not url:
|
|
return ""
|
|
if not (url.startswith("http://") or url.startswith("https://")):
|
|
url = "http://" + url
|
|
|
|
parsed = urllib.parse.urlparse(url)
|
|
path = parsed.path.rstrip('/')
|
|
if not path or path == "":
|
|
return f"{url.rstrip('/')}/api/v2/upload"
|
|
elif path.endswith("/api/v2"):
|
|
return f"{url.rstrip('/')}/upload"
|
|
elif not path.endswith("/upload"):
|
|
return f"{url.rstrip('/')}/api/v2/upload"
|
|
return url
|
|
|
|
def parse_upload_target(arg: str) -> tuple[str, bool]:
|
|
if not arg:
|
|
return ("", False)
|
|
|
|
arg = arg.strip().strip("'\"")
|
|
|
|
if arg.startswith("f0ckm://") or arg.startswith("f0ckm-uploader://"):
|
|
parsed = urllib.parse.urlparse(arg)
|
|
query_params = urllib.parse.parse_qs(parsed.query)
|
|
target_url = ""
|
|
if "url" in query_params and query_params["url"]:
|
|
target_url = query_params["url"][0]
|
|
else:
|
|
raw_path = parsed.netloc + parsed.path
|
|
if parsed.query:
|
|
raw_path += "?" + parsed.query
|
|
raw_path = raw_path.lstrip("/")
|
|
if raw_path.startswith("upload/"):
|
|
raw_path = raw_path[7:]
|
|
elif raw_path.startswith("upload?"):
|
|
raw_path = raw_path[7:]
|
|
target_url = raw_path
|
|
|
|
target_url = urllib.parse.unquote(target_url)
|
|
if target_url.startswith("http://") or target_url.startswith("https://"):
|
|
print(f"[f0ckm-gui] [PARSE] Protocol URL parsed: '{arg}' -> target='{target_url}' (is_url=True)", flush=True)
|
|
return (target_url, True)
|
|
|
|
if arg.startswith("http://") or arg.startswith("https://"):
|
|
print(f"[f0ckm-gui] [PARSE] Direct URL parsed: '{arg}' (is_url=True)", flush=True)
|
|
return (arg, True)
|
|
|
|
norm = normalize_file_path(arg)
|
|
print(f"[f0ckm-gui] [PARSE] File path parsed: '{arg}' -> norm='{norm}' (is_url=False)", flush=True)
|
|
return (norm, False)
|
|
|
|
def parse_cli_args(args: list[str]) -> tuple[str, list[str]]:
|
|
if not args:
|
|
return ("show", [])
|
|
|
|
targets = []
|
|
action = "show"
|
|
|
|
for arg in args:
|
|
arg_str = arg.strip()
|
|
if not arg_str:
|
|
continue
|
|
if arg_str in ("--spectacle", "-s"):
|
|
action = "spectacle"
|
|
elif arg_str.startswith("--upload="):
|
|
action = "upload"
|
|
t, _ = parse_upload_target(arg_str[9:])
|
|
if t:
|
|
targets.append(t)
|
|
elif arg_str == "--upload":
|
|
action = "upload"
|
|
else:
|
|
t, _ = parse_upload_target(arg_str)
|
|
if t:
|
|
action = "upload"
|
|
targets.append(t)
|
|
|
|
return (action, targets)
|
|
|
|
class LocalHTTPBridge(QObject):
|
|
request_upload = Signal(str)
|
|
|
|
local_http_bridge = LocalHTTPBridge()
|
|
|
|
class LocalHTTPHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
try:
|
|
parsed = urllib.parse.urlparse(self.path)
|
|
query = urllib.parse.parse_qs(parsed.query)
|
|
if parsed.path in ("/upload", "/upload/") and "url" in query:
|
|
target_url = query["url"][0]
|
|
print(f"[f0ckm-gui] [LOCAL HTTP API] Received direct upload request for URL: {target_url}", flush=True)
|
|
local_http_bridge.request_upload.emit(target_url)
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"success": true, "msg": "Queued upload"}')
|
|
elif parsed.path in ("/ping", "/ping/"):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"status": "ok", "app": "f0ckm-uploader"}')
|
|
else:
|
|
self.send_response(404)
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(b'{"error": "Not Found"}')
|
|
except Exception as e:
|
|
print(f"[f0ckm-gui] [LOCAL HTTP ERROR] {e}", flush=True)
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
def start_local_http_server(tray_app, port=18739):
|
|
try:
|
|
server = HTTPServer(("127.0.0.1", port), LocalHTTPHandler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
print(f"[f0ckm-gui] [LOCAL HTTP API] Started loopback server on http://127.0.0.1:{port}", flush=True)
|
|
return server
|
|
except Exception as e:
|
|
print(f"[f0ckm-gui] [LOCAL HTTP API WARNING] Could not start HTTP server on port {port}: {e}", flush=True)
|
|
return None
|
|
|
|
|
|
def copy_to_clipboard(text: str):
|
|
if not text:
|
|
return
|
|
try:
|
|
cb = QApplication.clipboard()
|
|
if cb:
|
|
cb.setText(text, QClipboard.Mode.Clipboard)
|
|
cb.setText(text, QClipboard.Mode.Selection)
|
|
except Exception as e:
|
|
print(f"Qt clipboard error: {e}")
|
|
|
|
# Fallback to system CLI clipboard utilities for Wayland/X11 persistence
|
|
if shutil.which("wl-copy"):
|
|
try:
|
|
subprocess.run(["wl-copy", text], check=False)
|
|
except Exception as e:
|
|
print(f"wl-copy error: {e}")
|
|
elif shutil.which("xclip"):
|
|
try:
|
|
subprocess.run(["xclip", "-selection", "clipboard"], input=text.encode("utf-8"), check=False)
|
|
except Exception as e:
|
|
print(f"xclip error: {e}")
|
|
elif shutil.which("xsel"):
|
|
try:
|
|
subprocess.run(["xsel", "--clipboard", "--input"], input=text.encode("utf-8"), check=False)
|
|
except Exception as e:
|
|
print(f"xsel error: {e}")
|
|
|
|
# Configuration paths
|
|
CONFIG_DIR = os.path.expanduser("~/.config/f0ckm-uploader")
|
|
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json")
|
|
HISTORY_PATH = os.path.join(CONFIG_DIR, "history.json")
|
|
|
|
def load_history():
|
|
if not os.path.exists(HISTORY_PATH):
|
|
return []
|
|
try:
|
|
with open(HISTORY_PATH, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return []
|
|
|
|
def save_history(history_list):
|
|
try:
|
|
os.makedirs(CONFIG_DIR, exist_ok=True)
|
|
with open(HISTORY_PATH, "w", encoding="utf-8") as f:
|
|
json.dump(history_list[:500], f, indent=2)
|
|
except Exception as e:
|
|
print(f"Error saving history: {e}")
|
|
|
|
def add_history_entry(entry):
|
|
if isinstance(entry, dict) and "instance_name" not in entry:
|
|
config = get_env_config()
|
|
instances = config.get("instances", [])
|
|
idx = config.get("active_instance_index", 0)
|
|
if 0 <= idx < len(instances):
|
|
entry["instance_name"] = instances[idx].get("name", "Default Instance")
|
|
else:
|
|
entry["instance_name"] = "Default Instance"
|
|
|
|
history = load_history()
|
|
history = [h for h in history if h.get("url") != entry.get("url")]
|
|
history.insert(0, entry)
|
|
save_history(history)
|
|
|
|
DEFAULT_CONFIG = {
|
|
"instances": [
|
|
{
|
|
"name": "Default Instance",
|
|
"api_url": "",
|
|
"api_key": ""
|
|
}
|
|
],
|
|
"active_instance_index": 0,
|
|
"api_url": "",
|
|
"api_key": "",
|
|
"default_rating": "",
|
|
"default_visibility": "0",
|
|
"default_tags": "screenshot",
|
|
"default_is_oc": False,
|
|
"enable_notifications": True,
|
|
"show_progress_dialog": True,
|
|
"autostart": False,
|
|
"icon_theme": "dark",
|
|
"clipboard_url_type": "post"
|
|
}
|
|
|
|
def load_config():
|
|
if not os.path.exists(CONFIG_PATH):
|
|
return DEFAULT_CONFIG.copy()
|
|
try:
|
|
with open(CONFIG_PATH, 'r') as f:
|
|
data = json.load(f)
|
|
config = DEFAULT_CONFIG.copy()
|
|
config.update(data)
|
|
|
|
instances = config.get("instances")
|
|
if not isinstance(instances, list) or not instances:
|
|
url = data.get("api_url") or config.get("api_url", "")
|
|
key = data.get("api_key") or config.get("api_key", "")
|
|
instances = [{"name": "Default Instance", "api_url": normalize_api_url(url), "api_key": key}]
|
|
config["instances"] = instances
|
|
config["active_instance_index"] = 0
|
|
else:
|
|
for inst in instances:
|
|
if "api_url" in inst:
|
|
inst["api_url"] = normalize_api_url(inst["api_url"])
|
|
|
|
idx = config.get("active_instance_index", 0)
|
|
if not isinstance(idx, int) or idx < 0 or idx >= len(config["instances"]):
|
|
idx = 0
|
|
config["active_instance_index"] = 0
|
|
|
|
active = config["instances"][idx]
|
|
config["api_url"] = normalize_api_url(active.get("api_url", ""))
|
|
config["api_key"] = active.get("api_key", "")
|
|
return config
|
|
except Exception:
|
|
return DEFAULT_CONFIG.copy()
|
|
|
|
def get_env_config():
|
|
config = DEFAULT_CONFIG.copy()
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
candidates = [
|
|
os.path.join(script_dir, ".env"),
|
|
"/home/kibi/Projects/f0ckm-uploader/.env",
|
|
os.path.expanduser("~/.config/f0ckm-uploader/.env"),
|
|
os.path.join(os.getcwd(), ".env")
|
|
]
|
|
dotenv_path = None
|
|
env_url = ""
|
|
env_key = ""
|
|
for path in candidates:
|
|
if os.path.exists(path):
|
|
dotenv_path = path
|
|
break
|
|
|
|
if dotenv_path:
|
|
try:
|
|
with open(dotenv_path, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
key = k.strip()
|
|
val = v.strip().strip("'\"")
|
|
if key == "F0CKM_URL" and val:
|
|
url = normalize_api_url(val)
|
|
env_url = url
|
|
config["api_url"] = url
|
|
elif key == "API_KEY" and val:
|
|
env_key = val
|
|
config["api_key"] = val
|
|
elif key == "RATING":
|
|
config["default_rating"] = val
|
|
elif key == "TAGS":
|
|
config["default_tags"] = val
|
|
elif key == "VISIBILITY":
|
|
config["default_visibility"] = val
|
|
elif key == "ICON_THEME":
|
|
config["icon_theme"] = val.lower()
|
|
except Exception as e:
|
|
print(f"Error reading .env from {dotenv_path}: {e}")
|
|
|
|
# Override with GUI config.json (User GUI Settings have priority over .env)
|
|
user_cfg = load_config()
|
|
if os.path.exists(CONFIG_PATH):
|
|
for k, v in user_cfg.items():
|
|
config[k] = v
|
|
|
|
instances = config.get("instances")
|
|
if not isinstance(instances, list) or not instances:
|
|
url = config.get("api_url") or env_url
|
|
key = config.get("api_key") or env_key
|
|
instances = [{"name": "Default Instance", "api_url": normalize_api_url(url), "api_key": key}]
|
|
config["instances"] = instances
|
|
config["active_instance_index"] = 0
|
|
|
|
idx = config.get("active_instance_index", 0)
|
|
if not isinstance(idx, int) or idx < 0 or idx >= len(instances):
|
|
idx = 0
|
|
config["active_instance_index"] = 0
|
|
|
|
active_inst = instances[idx]
|
|
config["api_url"] = normalize_api_url(active_inst.get("api_url", ""))
|
|
config["api_key"] = active_inst.get("api_key", "")
|
|
config["active_instance_name"] = active_inst.get("name", "Default Instance")
|
|
|
|
return config
|
|
|
|
def save_config(config):
|
|
try:
|
|
os.makedirs(CONFIG_DIR, exist_ok=True)
|
|
if "instances" in config and isinstance(config["instances"], list) and config["instances"]:
|
|
for inst in config["instances"]:
|
|
if "api_url" in inst:
|
|
inst["api_url"] = normalize_api_url(inst["api_url"])
|
|
idx = config.get("active_instance_index", 0)
|
|
if 0 <= idx < len(config["instances"]):
|
|
config["api_url"] = normalize_api_url(config["instances"][idx].get("api_url", ""))
|
|
config["api_key"] = config["instances"][idx].get("api_key", "")
|
|
with open(CONFIG_PATH, 'w') as f:
|
|
json.dump(config, f, indent=4)
|
|
return True
|
|
except Exception as e:
|
|
print(f"Failed to save config: {e}")
|
|
return False
|
|
|
|
def set_autostart(enabled):
|
|
autostart_dir = os.path.expanduser("~/.config/autostart")
|
|
autostart_path = os.path.join(autostart_dir, "f0ckm-uploader-gui.desktop")
|
|
|
|
if enabled:
|
|
try:
|
|
os.makedirs(autostart_dir, exist_ok=True)
|
|
# Use the installed path ~/.local/bin/f0ckm-uploader-gui if possible
|
|
exec_path = os.path.expanduser("~/.local/bin/f0ckm-uploader-gui")
|
|
if not os.path.exists(exec_path):
|
|
exec_path = os.path.abspath(sys.argv[0])
|
|
|
|
content = f"""[Desktop Entry]
|
|
Type=Application
|
|
Name=f0ckm Uploader GUI
|
|
Comment=System tray GUI and settings for f0ckm Uploader
|
|
Exec={exec_path}
|
|
Icon=f0ckm-uploader
|
|
Terminal=false
|
|
Categories=Utility;Network;
|
|
StartupNotify=false
|
|
X-GNOME-Autostart-enabled=true
|
|
"""
|
|
with open(autostart_path, "w") as f:
|
|
f.write(content)
|
|
os.chmod(autostart_path, 0o755)
|
|
print(f"[f0ckm-gui] Autostart desktop file written to {autostart_path}", flush=True)
|
|
except Exception as e:
|
|
print(f"Failed to create autostart entry: {e}", flush=True)
|
|
else:
|
|
if os.path.exists(autostart_path):
|
|
try:
|
|
os.remove(autostart_path)
|
|
print(f"[f0ckm-gui] Autostart desktop file removed from {autostart_path}", flush=True)
|
|
except Exception as e:
|
|
print(f"Failed to remove autostart entry: {e}", flush=True)
|
|
|
|
# ==========================================
|
|
# Connection Tester Worker
|
|
# ==========================================
|
|
class ConnectionTester(QObject):
|
|
finished = Signal(bool, str) # Success, Message
|
|
|
|
def __init__(self, url, api_key):
|
|
super().__init__()
|
|
self.url = normalize_api_url(url)
|
|
self.api_key = api_key
|
|
|
|
def run(self):
|
|
try:
|
|
# Send an empty POST request to test authorization and connection
|
|
req = urllib.request.Request(
|
|
self.url,
|
|
data=b"",
|
|
headers={
|
|
"X-Api-Key": self.api_key,
|
|
"User-Agent": "f0ckm-Uploader-GUI/1.0"
|
|
},
|
|
method="POST"
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as response:
|
|
self.finished.emit(True, "Connection successful! Server responded with HTTP 200.")
|
|
except urllib.error.HTTPError as e:
|
|
body = e.read().decode('utf-8', errors='ignore')
|
|
if e.code == 401:
|
|
self.finished.emit(False, "Unauthorized: Invalid API Key.")
|
|
elif e.code in (400, 422):
|
|
# Key is valid but payload is invalid (empty file parameter), which confirms authentication works
|
|
self.finished.emit(True, "Connection successful! (API key is valid)")
|
|
else:
|
|
self.finished.emit(False, f"HTTP Error {e.code}: {body or e.reason}")
|
|
except urllib.error.URLError as e:
|
|
self.finished.emit(False, f"Connection failed: {e.reason}")
|
|
except Exception as e:
|
|
self.finished.emit(False, f"Error: {str(e)}")
|
|
|
|
def get_app_icon() -> QIcon:
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
candidates = [
|
|
os.path.join(script_dir, "icon_app.svg"),
|
|
os.path.join(script_dir, "icon_app.png"),
|
|
"/home/kibi/Projects/f0ckm-uploader/icon_app.svg",
|
|
os.path.expanduser("~/.local/bin/icon_app.svg"),
|
|
os.path.expanduser("~/.local/share/icons/hicolor/scalable/apps/f0ckm-uploader.svg"),
|
|
os.path.expanduser("~/.local/share/icons/hicolor/128x128/apps/f0ckm-uploader.png")
|
|
]
|
|
for c in candidates:
|
|
if os.path.exists(c):
|
|
return QIcon(c)
|
|
return QIcon.fromTheme("f0ckm-uploader")
|
|
|
|
# ==========================================
|
|
# Settings Dialog Window
|
|
# ==========================================
|
|
class SettingsDialog(QDialog):
|
|
def __init__(self, parent=None, tray_app=None):
|
|
super().__init__(parent)
|
|
self.tray_app = tray_app
|
|
self.setWindowTitle("f0ckm Uploader Settings")
|
|
self.resize(560, 700)
|
|
|
|
app_icon = get_app_icon()
|
|
if not app_icon.isNull():
|
|
self.setWindowIcon(app_icon)
|
|
|
|
self.tester_thread = None
|
|
self.tester = None
|
|
|
|
self.instances = []
|
|
self.current_instance_index = 0
|
|
self._ignore_instance_signals = False
|
|
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
main_layout = QVBoxLayout(self)
|
|
main_layout.setContentsMargins(20, 20, 20, 20)
|
|
main_layout.setSpacing(15)
|
|
|
|
# Header
|
|
header_layout = QHBoxLayout()
|
|
header_icon = QLabel()
|
|
icon = QIcon.fromTheme("preferences-system", QIcon.fromTheme("network-server"))
|
|
header_icon.setPixmap(icon.pixmap(32, 32))
|
|
|
|
header_text_layout = QVBoxLayout()
|
|
header_title = QLabel("f0ckm Uploader")
|
|
header_title.setStyleSheet("font-size: 16px; font-weight: bold;")
|
|
header_subtitle = QLabel("Configure f0ckm instances, connection settings and default options")
|
|
header_subtitle.setStyleSheet("font-size: 11px; opacity: 0.7;")
|
|
header_text_layout.addWidget(header_title)
|
|
header_text_layout.addWidget(header_subtitle)
|
|
|
|
header_layout.addWidget(header_icon)
|
|
header_layout.addLayout(header_text_layout)
|
|
header_layout.addStretch()
|
|
main_layout.addLayout(header_layout)
|
|
|
|
# Section 1: API Connection & Instances Settings
|
|
grp_api = QGroupBox("f0ckm Instances & Connection")
|
|
api_layout = QFormLayout(grp_api)
|
|
api_layout.setContentsMargins(15, 20, 15, 15)
|
|
api_layout.setSpacing(10)
|
|
|
|
# Instance Selection & Management Row
|
|
inst_row_layout = QHBoxLayout()
|
|
self.cb_instance = QComboBox()
|
|
self.cb_instance.currentIndexChanged.connect(self.on_instance_changed)
|
|
inst_row_layout.addWidget(self.cb_instance, stretch=1)
|
|
|
|
self.btn_add_instance = QPushButton("+ Add")
|
|
self.btn_add_instance.setToolTip("Add a new f0ckm instance")
|
|
self.btn_add_instance.clicked.connect(self.on_add_instance)
|
|
inst_row_layout.addWidget(self.btn_add_instance)
|
|
|
|
self.btn_rename_instance = QPushButton("Rename")
|
|
self.btn_rename_instance.setToolTip("Rename currently selected instance")
|
|
self.btn_rename_instance.clicked.connect(self.on_rename_instance)
|
|
inst_row_layout.addWidget(self.btn_rename_instance)
|
|
|
|
self.btn_delete_instance = QPushButton("Delete")
|
|
self.btn_delete_instance.setToolTip("Delete currently selected instance")
|
|
self.btn_delete_instance.clicked.connect(self.on_delete_instance)
|
|
inst_row_layout.addWidget(self.btn_delete_instance)
|
|
|
|
api_layout.addRow("Instance:", inst_row_layout)
|
|
|
|
self.txt_url = QLineEdit()
|
|
self.txt_url.setPlaceholderText("https://example.com/api/v2/upload")
|
|
self.txt_url.textChanged.connect(self.on_url_or_key_edited)
|
|
api_layout.addRow("API URL:", self.txt_url)
|
|
|
|
key_layout = QHBoxLayout()
|
|
self.txt_key = QLineEdit()
|
|
self.txt_key.setEchoMode(QLineEdit.Password)
|
|
self.txt_key.setPlaceholderText("your_api_key_here")
|
|
self.txt_key.textChanged.connect(self.on_url_or_key_edited)
|
|
key_layout.addWidget(self.txt_key)
|
|
|
|
self.btn_toggle_key = QPushButton("Show")
|
|
self.btn_toggle_key.setFixedWidth(60)
|
|
self.btn_toggle_key.clicked.connect(self.toggle_key_visibility)
|
|
key_layout.addWidget(self.btn_toggle_key)
|
|
api_layout.addRow("API Key:", key_layout)
|
|
|
|
test_layout = QHBoxLayout()
|
|
test_layout.addStretch()
|
|
self.btn_test = QPushButton("Test Connection")
|
|
self.btn_test.setObjectName("test")
|
|
self.btn_test.clicked.connect(self.on_test_connection)
|
|
test_layout.addWidget(self.btn_test)
|
|
api_layout.addRow("", test_layout)
|
|
|
|
main_layout.addWidget(grp_api)
|
|
|
|
# Section 2: Default Parameters
|
|
grp_defaults = QGroupBox("Default Parameters")
|
|
defaults_layout = QFormLayout(grp_defaults)
|
|
defaults_layout.setContentsMargins(15, 20, 15, 15)
|
|
defaults_layout.setSpacing(10)
|
|
|
|
self.cb_rating = QComboBox()
|
|
self.cb_rating.addItems(["(None / Default)", "SFW", "NSFW", "NSFL"])
|
|
defaults_layout.addRow("Default Rating:", self.cb_rating)
|
|
|
|
self.cb_visibility = QComboBox()
|
|
self.cb_visibility.addItems(["Public", "Unlisted", "Private"])
|
|
defaults_layout.addRow("Default Visibility:", self.cb_visibility)
|
|
|
|
self.txt_tags = QLineEdit()
|
|
self.txt_tags.setPlaceholderText("comma-separated tags (e.g., meme, anime)")
|
|
defaults_layout.addRow("Default Tags:", self.txt_tags)
|
|
|
|
self.chk_oc = QCheckBox("Mark upload as Original Content (is_oc=1)")
|
|
defaults_layout.addRow("", self.chk_oc)
|
|
|
|
main_layout.addWidget(grp_defaults)
|
|
|
|
# Section 3: App Behavior
|
|
grp_behavior = QGroupBox("App Behavior")
|
|
behavior_layout = QVBoxLayout(grp_behavior)
|
|
behavior_layout.setContentsMargins(15, 20, 15, 15)
|
|
behavior_layout.setSpacing(10)
|
|
|
|
self.chk_notifications = QCheckBox("Enable Desktop Notifications")
|
|
self.chk_progress = QCheckBox("Show Live Progress Dialog")
|
|
self.chk_autostart = QCheckBox("Start automatically on system login")
|
|
|
|
behavior_layout.addWidget(self.chk_notifications)
|
|
behavior_layout.addWidget(self.chk_progress)
|
|
behavior_layout.addWidget(self.chk_autostart)
|
|
|
|
# Icon Theme Selection
|
|
theme_layout = QHBoxLayout()
|
|
theme_label = QLabel("Tray Icon Theme:")
|
|
self.cb_icon_theme = QComboBox()
|
|
self.cb_icon_theme.addItems(["Dark (Black Sun)", "Light (White Sun)"])
|
|
theme_layout.addWidget(theme_label)
|
|
theme_layout.addWidget(self.cb_icon_theme)
|
|
theme_layout.addStretch()
|
|
behavior_layout.addLayout(theme_layout)
|
|
|
|
# Copied Link Format Selection
|
|
url_type_layout = QHBoxLayout()
|
|
url_type_label = QLabel("Copied Link Format:")
|
|
self.cb_url_type = QComboBox()
|
|
self.cb_url_type.addItems(["Post Page URL (e.g. /v/123)", "Direct File URL (e.g. /files/123.png)"])
|
|
url_type_layout.addWidget(url_type_label)
|
|
url_type_layout.addWidget(self.cb_url_type)
|
|
url_type_layout.addStretch()
|
|
behavior_layout.addLayout(url_type_layout)
|
|
|
|
main_layout.addWidget(grp_behavior)
|
|
|
|
# Section 4: Sound Notification
|
|
grp_sound = QGroupBox("Audio Notification")
|
|
sound_layout = QVBoxLayout(grp_sound)
|
|
sound_layout.setContentsMargins(15, 20, 15, 15)
|
|
sound_layout.setSpacing(10)
|
|
|
|
self.chk_sound = QCheckBox("Play sound on successful upload")
|
|
sound_layout.addWidget(self.chk_sound)
|
|
|
|
file_picker_layout = QHBoxLayout()
|
|
self.txt_sound_path = QLineEdit()
|
|
self.txt_sound_path.setPlaceholderText("Select native audio file (.wav, .ogg, .flac)...")
|
|
file_picker_layout.addWidget(self.txt_sound_path)
|
|
|
|
self.btn_browse_sound = QPushButton("Browse...")
|
|
self.btn_browse_sound.clicked.connect(self.on_browse_sound)
|
|
file_picker_layout.addWidget(self.btn_browse_sound)
|
|
|
|
self.btn_test_sound = QPushButton("Test Sound")
|
|
self.btn_test_sound.clicked.connect(self.on_test_sound)
|
|
file_picker_layout.addWidget(self.btn_test_sound)
|
|
|
|
sound_layout.addLayout(file_picker_layout)
|
|
|
|
vol_layout = QHBoxLayout()
|
|
vol_label = QLabel("Volume:")
|
|
self.slider_volume = QSlider(Qt.Orientation.Horizontal)
|
|
self.slider_volume.setRange(0, 100)
|
|
self.slider_volume.setValue(15)
|
|
self.lbl_volume = QLabel("15%")
|
|
self.lbl_volume.setFixedWidth(40)
|
|
self.slider_volume.valueChanged.connect(lambda v: self.lbl_volume.setText(f"{v}%"))
|
|
vol_layout.addWidget(vol_label)
|
|
vol_layout.addWidget(self.slider_volume)
|
|
vol_layout.addWidget(self.lbl_volume)
|
|
sound_layout.addLayout(vol_layout)
|
|
|
|
main_layout.addWidget(grp_sound)
|
|
|
|
# Save / Cancel Buttons
|
|
btn_layout = QHBoxLayout()
|
|
btn_layout.addStretch()
|
|
|
|
btn_cancel = QPushButton("Cancel")
|
|
btn_cancel.clicked.connect(self.reject)
|
|
|
|
self.btn_save = QPushButton("Save Settings")
|
|
self.btn_save.setObjectName("primary")
|
|
self.btn_save.clicked.connect(self.save_settings)
|
|
|
|
btn_layout.addWidget(btn_cancel)
|
|
btn_layout.addWidget(self.btn_save)
|
|
main_layout.addLayout(btn_layout)
|
|
|
|
self.load_current_settings()
|
|
|
|
def toggle_key_visibility(self):
|
|
if self.txt_key.echoMode() == QLineEdit.Password:
|
|
self.txt_key.setEchoMode(QLineEdit.Normal)
|
|
self.btn_toggle_key.setText("Hide")
|
|
else:
|
|
self.txt_key.setEchoMode(QLineEdit.Password)
|
|
self.btn_toggle_key.setText("Show")
|
|
|
|
def refresh_instance_combo(self):
|
|
prev_ignore = self._ignore_instance_signals
|
|
self._ignore_instance_signals = True
|
|
try:
|
|
self.cb_instance.clear()
|
|
for inst in self.instances:
|
|
self.cb_instance.addItem(inst.get("name", "Unnamed Instance"))
|
|
|
|
if 0 <= self.current_instance_index < len(self.instances):
|
|
self.cb_instance.setCurrentIndex(self.current_instance_index)
|
|
self.load_instance_fields(self.current_instance_index)
|
|
finally:
|
|
self._ignore_instance_signals = prev_ignore
|
|
|
|
def load_instance_fields(self, index):
|
|
if 0 <= index < len(self.instances):
|
|
prev_ignore = self._ignore_instance_signals
|
|
self._ignore_instance_signals = True
|
|
try:
|
|
inst = self.instances[index]
|
|
self.txt_url.setText(inst.get("api_url", ""))
|
|
self.txt_key.setText(inst.get("api_key", ""))
|
|
finally:
|
|
self._ignore_instance_signals = prev_ignore
|
|
|
|
def on_instance_changed(self, new_index):
|
|
if self._ignore_instance_signals or new_index < 0 or new_index >= len(self.instances):
|
|
return
|
|
|
|
if 0 <= self.current_instance_index < len(self.instances):
|
|
self.instances[self.current_instance_index]["api_url"] = self.txt_url.text().strip()
|
|
self.instances[self.current_instance_index]["api_key"] = self.txt_key.text().strip()
|
|
|
|
self.current_instance_index = new_index
|
|
self.load_instance_fields(new_index)
|
|
|
|
def on_url_or_key_edited(self):
|
|
if self._ignore_instance_signals:
|
|
return
|
|
if 0 <= self.current_instance_index < len(self.instances):
|
|
self.instances[self.current_instance_index]["api_url"] = self.txt_url.text().strip()
|
|
self.instances[self.current_instance_index]["api_key"] = self.txt_key.text().strip()
|
|
|
|
def on_add_instance(self):
|
|
name, ok = QInputDialog.getText(self, "Add Instance", "Enter name for new f0ckm instance:")
|
|
if ok and name.strip():
|
|
name = name.strip()
|
|
self.on_url_or_key_edited()
|
|
new_inst = {
|
|
"name": name,
|
|
"api_url": "",
|
|
"api_key": ""
|
|
}
|
|
self.instances.append(new_inst)
|
|
self.current_instance_index = len(self.instances) - 1
|
|
self.refresh_instance_combo()
|
|
|
|
def on_rename_instance(self):
|
|
if not (0 <= self.current_instance_index < len(self.instances)):
|
|
return
|
|
curr_name = self.instances[self.current_instance_index].get("name", "")
|
|
name, ok = QInputDialog.getText(self, "Rename Instance", "Enter new instance name:", text=curr_name)
|
|
if ok and name.strip():
|
|
name = name.strip()
|
|
self.instances[self.current_instance_index]["name"] = name
|
|
self.refresh_instance_combo()
|
|
|
|
def on_delete_instance(self):
|
|
if len(self.instances) <= 1:
|
|
QMessageBox.information(self, "Cannot Delete", "You must keep at least one f0ckm instance.")
|
|
return
|
|
|
|
curr_name = self.instances[self.current_instance_index].get("name", "")
|
|
reply = QMessageBox.question(
|
|
self,
|
|
"Delete Instance",
|
|
f"Are you sure you want to delete instance '{curr_name}'?",
|
|
QMessageBox.Yes | QMessageBox.No
|
|
)
|
|
if reply == QMessageBox.Yes:
|
|
self.instances.pop(self.current_instance_index)
|
|
self.current_instance_index = max(0, self.current_instance_index - 1)
|
|
self.refresh_instance_combo()
|
|
|
|
def load_current_settings(self):
|
|
config = get_env_config()
|
|
|
|
self.instances = [dict(inst) for inst in config.get("instances", [])]
|
|
if not self.instances:
|
|
self.instances = [{
|
|
"name": "Default Instance",
|
|
"api_url": config.get("api_url", ""),
|
|
"api_key": config.get("api_key", "")
|
|
}]
|
|
|
|
self.current_instance_index = config.get("active_instance_index", 0)
|
|
if self.current_instance_index < 0 or self.current_instance_index >= len(self.instances):
|
|
self.current_instance_index = 0
|
|
|
|
self.refresh_instance_combo()
|
|
|
|
rating = config.get("default_rating", "").lower()
|
|
rating_map = {"": 0, "s": 1, "sfw": 1, "safe": 1, "q": 2, "nsfw": 2, "questionable": 2, "e": 3, "nsfl": 3, "explicit": 3}
|
|
self.cb_rating.setCurrentIndex(rating_map.get(rating, 0))
|
|
|
|
vis = str(config.get("default_visibility", "0")).lower()
|
|
vis_map = {"0": 0, "public": 0, "1": 1, "unlisted": 1, "2": 2, "private": 2}
|
|
self.cb_visibility.setCurrentIndex(vis_map.get(vis, 0))
|
|
|
|
self.txt_tags.setText(config.get("default_tags", ""))
|
|
self.chk_oc.setChecked(config.get("default_is_oc", False))
|
|
self.chk_notifications.setChecked(config.get("enable_notifications", True))
|
|
self.chk_progress.setChecked(config.get("show_progress_dialog", True))
|
|
self.chk_autostart.setChecked(config.get("autostart", False))
|
|
|
|
icon_theme = config.get("icon_theme", "dark").lower()
|
|
self.cb_icon_theme.setCurrentIndex(1 if icon_theme == "light" else 0)
|
|
|
|
url_type = config.get("clipboard_url_type", "post").lower()
|
|
self.cb_url_type.setCurrentIndex(1 if url_type == "direct" else 0)
|
|
|
|
self.chk_sound.setChecked(config.get("play_success_sound", False))
|
|
self.txt_sound_path.setText(config.get("success_audio_path", ""))
|
|
|
|
vol = config.get("success_audio_volume", 15)
|
|
try:
|
|
vol = int(vol)
|
|
except Exception:
|
|
vol = 15
|
|
vol = min(100, max(0, vol))
|
|
self.slider_volume.setValue(vol)
|
|
self.lbl_volume.setText(f"{vol}%")
|
|
|
|
def on_test_connection(self):
|
|
self.btn_test.setEnabled(False)
|
|
self.btn_test.setText("Testing...")
|
|
|
|
raw_url = self.txt_url.text().strip()
|
|
url = normalize_api_url(raw_url)
|
|
if url != raw_url:
|
|
self.txt_url.setText(url)
|
|
key = self.txt_key.text().strip()
|
|
|
|
self.tester_thread = QThread()
|
|
self.tester = ConnectionTester(url, key)
|
|
self.tester.moveToThread(self.tester_thread)
|
|
|
|
self.tester_thread.started.connect(self.tester.run)
|
|
self.tester.finished.connect(self.on_test_finished)
|
|
self.tester.finished.connect(self.tester_thread.quit)
|
|
self.tester.finished.connect(self.tester.deleteLater)
|
|
self.tester_thread.finished.connect(self.tester_thread.deleteLater)
|
|
|
|
self.tester_thread.start()
|
|
|
|
def on_test_finished(self, success, message):
|
|
self.btn_test.setEnabled(True)
|
|
self.btn_test.setText("Test Connection")
|
|
|
|
if success:
|
|
QMessageBox.information(self, "Connection Test Success", message)
|
|
else:
|
|
QMessageBox.warning(self, "Connection Test Failed", message)
|
|
|
|
def on_browse_sound(self):
|
|
file_path, _ = QFileDialog.getOpenFileName(
|
|
self,
|
|
"Select Native Audio File for Success Notification",
|
|
"",
|
|
"Supported Audio Files (*.wav *.ogg *.oga *.flac *.aiff *.aif *.au);;WAV Audio (*.wav);;OGG Audio (*.ogg *.oga);;FLAC Audio (*.flac)"
|
|
)
|
|
if file_path:
|
|
self.txt_sound_path.setText(file_path)
|
|
|
|
def on_test_sound(self):
|
|
sound_path = self.txt_sound_path.text().strip()
|
|
if not sound_path or not os.path.exists(sound_path):
|
|
QMessageBox.warning(self, "Invalid File", "Please select a valid audio file first.")
|
|
return
|
|
ext = os.path.splitext(sound_path)[1].lower()
|
|
supported = [".wav", ".ogg", ".oga", ".flac", ".aiff", ".aif", ".au"]
|
|
if ext not in supported:
|
|
QMessageBox.warning(
|
|
self,
|
|
"Unsupported Format",
|
|
f"The format '{ext}' is not supported natively by PipeWire/PulseAudio.\nPlease select a .wav, .ogg, or .flac file."
|
|
)
|
|
return
|
|
play_success_sound(sound_path, volume=self.slider_volume.value())
|
|
|
|
def save_settings(self):
|
|
self.on_url_or_key_edited()
|
|
|
|
active_idx = self.cb_instance.currentIndex()
|
|
if active_idx < 0 or active_idx >= len(self.instances):
|
|
active_idx = 0
|
|
|
|
active_inst = self.instances[active_idx] if self.instances else {"api_url": "", "api_key": ""}
|
|
|
|
config = {
|
|
"instances": self.instances,
|
|
"active_instance_index": active_idx,
|
|
"api_url": active_inst.get("api_url", ""),
|
|
"api_key": active_inst.get("api_key", ""),
|
|
"default_rating": ["", "s", "q", "e"][self.cb_rating.currentIndex()],
|
|
"default_visibility": ["0", "1", "2"][self.cb_visibility.currentIndex()],
|
|
"default_tags": self.txt_tags.text().strip(),
|
|
"default_is_oc": self.chk_oc.isChecked(),
|
|
"enable_notifications": self.chk_notifications.isChecked(),
|
|
"show_progress_dialog": self.chk_progress.isChecked(),
|
|
"autostart": self.chk_autostart.isChecked(),
|
|
"icon_theme": "light" if self.cb_icon_theme.currentIndex() == 1 else "dark",
|
|
"clipboard_url_type": "direct" if self.cb_url_type.currentIndex() == 1 else "post",
|
|
"play_success_sound": self.chk_sound.isChecked(),
|
|
"success_audio_path": self.txt_sound_path.text().strip(),
|
|
"success_audio_volume": self.slider_volume.value()
|
|
}
|
|
|
|
if save_config(config):
|
|
set_autostart(config["autostart"])
|
|
if self.tray_app:
|
|
if hasattr(self.tray_app, "load_tray_icon"):
|
|
self.tray_app.load_tray_icon()
|
|
if hasattr(self.tray_app, "rebuild_instance_menu"):
|
|
self.tray_app.rebuild_instance_menu()
|
|
self.accept()
|
|
else:
|
|
QMessageBox.critical(self, "Error", "Failed to save configuration file.")
|
|
|
|
# ==========================================
|
|
# Draggable File Label Widget
|
|
# ==========================================
|
|
# ==========================================
|
|
# Draggable File Label Widget
|
|
# ==========================================
|
|
class DraggableImageLabel(QLabel):
|
|
def __init__(self, item, parent=None):
|
|
super().__init__(parent)
|
|
self.item = item if isinstance(item, dict) else {"file_path": str(item)}
|
|
self.file_path = self.item.get("file_path", "")
|
|
self.drag_start_pos = None
|
|
|
|
def mousePressEvent(self, event):
|
|
if event.button() == Qt.MouseButton.LeftButton:
|
|
self.drag_start_pos = event.position().toPoint()
|
|
super().mousePressEvent(event)
|
|
|
|
def mouseReleaseEvent(self, event):
|
|
if event.button() == Qt.MouseButton.LeftButton and self.drag_start_pos is not None:
|
|
delta = (event.position().toPoint() - self.drag_start_pos).manhattanLength()
|
|
if delta < QApplication.startDragDistance():
|
|
file_path = self.item.get("file_path", "")
|
|
url = self.item.get("url", "")
|
|
if file_path and os.path.exists(file_path):
|
|
QDesktopServices.openUrl(QUrl.fromLocalFile(file_path))
|
|
elif url:
|
|
QDesktopServices.openUrl(QUrl(url))
|
|
self.drag_start_pos = None
|
|
super().mouseReleaseEvent(event)
|
|
|
|
def mouseMoveEvent(self, event):
|
|
if not (event.buttons() & Qt.MouseButton.LeftButton):
|
|
return
|
|
if self.drag_start_pos is None:
|
|
return
|
|
if (event.position().toPoint() - self.drag_start_pos).manhattanLength() < QApplication.startDragDistance():
|
|
return
|
|
|
|
if self.file_path and os.path.exists(self.file_path):
|
|
drag = QDrag(self)
|
|
mime_data = QMimeData()
|
|
mime_data.setUrls([QUrl.fromLocalFile(self.file_path)])
|
|
drag.setMimeData(mime_data)
|
|
|
|
if self.pixmap() and not self.pixmap().isNull():
|
|
drag.setPixmap(self.pixmap().scaled(80, 80, Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.SmoothTransformation))
|
|
|
|
drag.exec(Qt.DropAction.CopyAction)
|
|
self.drag_start_pos = None
|
|
|
|
# ==========================================
|
|
# Upload Gallery Window & Components
|
|
# ==========================================
|
|
THUMBNAIL_CACHE = {}
|
|
|
|
class ThumbnailWorkerSignals(QObject):
|
|
finished = Signal(dict, QImage)
|
|
|
|
class ThumbnailWorker(QRunnable):
|
|
def __init__(self, item, callback):
|
|
super().__init__()
|
|
self.item = item
|
|
self.signals = ThumbnailWorkerSignals()
|
|
self.signals.finished.connect(callback)
|
|
|
|
def run(self):
|
|
try:
|
|
thumb_path = self.item.get("thumbnail_path", "")
|
|
file_path = self.item.get("file_path", "")
|
|
qimg = None
|
|
if thumb_path and os.path.exists(thumb_path):
|
|
qimg = QImage(thumb_path)
|
|
elif file_path and os.path.exists(file_path):
|
|
gen_thumb = generate_thumbnail(file_path)
|
|
if gen_thumb and os.path.exists(gen_thumb):
|
|
qimg = QImage(gen_thumb)
|
|
self.item["thumbnail_path"] = gen_thumb
|
|
|
|
if qimg is None or qimg.isNull():
|
|
self.signals.finished.emit(self.item, QImage())
|
|
else:
|
|
self.signals.finished.emit(self.item, qimg)
|
|
except Exception:
|
|
try:
|
|
self.signals.finished.emit(self.item, QImage())
|
|
except Exception:
|
|
pass
|
|
|
|
class GalleryCard(QFrame):
|
|
def __init__(self, item, parent=None, on_delete=None):
|
|
super().__init__(parent)
|
|
self.item = item
|
|
self.on_delete = on_delete
|
|
self._is_destroyed = False
|
|
self.setFrameShape(QFrame.StyledPanel)
|
|
self.setFixedHeight(190)
|
|
self.setMinimumWidth(160)
|
|
self.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
|
self.setStyleSheet("""
|
|
GalleryCard {
|
|
background-color: #1e1e24;
|
|
border: 1px solid #333340;
|
|
border-radius: 8px;
|
|
}
|
|
GalleryCard:hover {
|
|
border: 1px solid #555575;
|
|
background-color: #242430;
|
|
}
|
|
""")
|
|
|
|
layout = QVBoxLayout(self)
|
|
layout.setContentsMargins(8, 8, 8, 8)
|
|
layout.setSpacing(6)
|
|
|
|
# Thumbnail with Drag & Drop & Click support
|
|
self.thumb_label = DraggableImageLabel(item)
|
|
self.thumb_label.setFixedHeight(95)
|
|
self.thumb_label.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
|
|
self.thumb_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
self.thumb_label.setCursor(Qt.CursorShape.PointingHandCursor)
|
|
self.thumb_label.setToolTip("Click to open file | Drag to drop into other apps")
|
|
self.thumb_label.setStyleSheet("background-color: #121216; border-radius: 4px;")
|
|
|
|
layout.addWidget(self.thumb_label)
|
|
|
|
# Filename
|
|
fname = item.get("filename", "File")
|
|
lbl_name = QLabel(fname)
|
|
lbl_name.setStyleSheet("font-weight: bold; font-size: 11px; color: #e0e0e0;")
|
|
lbl_name.setToolTip(item.get("filename", ""))
|
|
layout.addWidget(lbl_name)
|
|
|
|
# Metadata Row: Filetype badge + Timestamp + Instance badge
|
|
meta_layout = QHBoxLayout()
|
|
meta_layout.setContentsMargins(0, 0, 0, 0)
|
|
meta_layout.setSpacing(4)
|
|
|
|
# Filetype badge
|
|
ext = os.path.splitext(item.get("filename", "") or item.get("file_path", "") or item.get("url", ""))[1].replace(".", "").upper()
|
|
if not ext:
|
|
ext = "FILE"
|
|
lbl_ext = QLabel(ext)
|
|
lbl_ext.setStyleSheet("background-color: #2a2a38; color: #7f9cf5; border-radius: 3px; font-weight: bold; font-size: 9px; padding: 1px 4px;")
|
|
meta_layout.addWidget(lbl_ext)
|
|
|
|
# Timestamp
|
|
ts = item.get("timestamp", 0)
|
|
dt_str = time.strftime("%Y-%m-%d %H:%M", time.localtime(ts)) if ts else ""
|
|
lbl_time = QLabel(dt_str)
|
|
lbl_time.setStyleSheet("font-size: 9px; color: #8888a0;")
|
|
meta_layout.addWidget(lbl_time)
|
|
|
|
meta_layout.addStretch()
|
|
|
|
# Instance badge if available
|
|
inst_name = item.get("instance_name", "")
|
|
if inst_name:
|
|
lbl_inst = QLabel(inst_name)
|
|
lbl_inst.setStyleSheet("background-color: #262630; color: #a0a0b0; border-radius: 3px; font-size: 9px; padding: 1px 4px;")
|
|
lbl_inst.setToolTip(f"Instance: {inst_name}")
|
|
meta_layout.addWidget(lbl_inst)
|
|
|
|
layout.addLayout(meta_layout)
|
|
|
|
# Action Buttons
|
|
btn_layout = QHBoxLayout()
|
|
btn_layout.setSpacing(4)
|
|
|
|
btn_copy = QPushButton("Copy")
|
|
btn_copy.setToolTip("Copy URL to clipboard")
|
|
btn_copy.setStyleSheet("font-size: 10px; padding: 4px 6px;")
|
|
btn_copy.clicked.connect(self.on_copy_link)
|
|
|
|
btn_open = QPushButton("Open")
|
|
btn_open.setToolTip("Open link in browser")
|
|
btn_open.setStyleSheet("font-size: 10px; padding: 4px 6px;")
|
|
btn_open.clicked.connect(self.on_open_url)
|
|
|
|
btn_del = QPushButton("✕")
|
|
btn_del.setToolTip("Remove from history")
|
|
btn_del.setStyleSheet("font-size: 10px; padding: 4px 6px; color: #ff6666;")
|
|
btn_del.clicked.connect(self.on_remove_item)
|
|
|
|
btn_layout.addWidget(btn_copy)
|
|
btn_layout.addWidget(btn_open)
|
|
btn_layout.addWidget(btn_del)
|
|
|
|
layout.addLayout(btn_layout)
|
|
|
|
self.load_thumbnail_async()
|
|
|
|
def load_thumbnail_async(self):
|
|
cache_key = self.item.get("url") or self.item.get("file_path") or self.item.get("thumbnail_path")
|
|
if cache_key and cache_key in THUMBNAIL_CACHE:
|
|
self.thumb_label.setPixmap(THUMBNAIL_CACHE[cache_key])
|
|
else:
|
|
self.thumb_label.setText("Loading...")
|
|
self.thumb_label.setStyleSheet("background-color: #121216; color: #666; font-size: 11px;")
|
|
worker = ThumbnailWorker(self.item, self.on_thumbnail_loaded)
|
|
QThreadPool.globalInstance().start(worker)
|
|
|
|
def on_thumbnail_loaded(self, item, qimg):
|
|
try:
|
|
if getattr(self, "_is_destroyed", False):
|
|
return
|
|
if not qimg or qimg.isNull():
|
|
self.thumb_label.setText("No Preview")
|
|
self.thumb_label.setStyleSheet("background-color: #121216; color: #666; font-size: 11px;")
|
|
return
|
|
|
|
scaled = qimg.scaled(160, 100, Qt.AspectRatioMode.KeepAspectRatioByExpanding, Qt.TransformationMode.SmoothTransformation)
|
|
x = max(0, (scaled.width() - 160) // 2)
|
|
y = max(0, (scaled.height() - 100) // 2)
|
|
cropped = scaled.copy(x, y, 160, 100)
|
|
pixmap = QPixmap.fromImage(cropped)
|
|
|
|
cache_key = item.get("url") or item.get("file_path") or item.get("thumbnail_path")
|
|
if cache_key:
|
|
THUMBNAIL_CACHE[cache_key] = pixmap
|
|
|
|
self.thumb_label.setPixmap(pixmap)
|
|
except Exception:
|
|
try:
|
|
self.thumb_label.setText("No Preview")
|
|
self.thumb_label.setStyleSheet("background-color: #121216; color: #666; font-size: 11px;")
|
|
except Exception:
|
|
pass
|
|
|
|
def deleteLater(self):
|
|
self._is_destroyed = True
|
|
super().deleteLater()
|
|
|
|
def on_copy_link(self):
|
|
url = self.item.get("url", "")
|
|
if url:
|
|
copy_to_clipboard(url)
|
|
|
|
def on_open_url(self):
|
|
url = self.item.get("url", "")
|
|
if url:
|
|
QDesktopServices.openUrl(QUrl(url))
|
|
|
|
def on_remove_item(self):
|
|
if self.on_delete:
|
|
self.on_delete(self.item)
|
|
|
|
class GalleryWindow(QDialog):
|
|
BATCH_SIZE = 16
|
|
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.setWindowTitle("f0ckm Upload Gallery")
|
|
self.resize(780, 580)
|
|
self.filtered_history = []
|
|
self.loaded_count = 0
|
|
self.cards = []
|
|
self.current_cols = 0
|
|
self.is_loading_batch = False
|
|
|
|
app_icon = get_app_icon()
|
|
if not app_icon.isNull():
|
|
self.setWindowIcon(app_icon)
|
|
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
main_layout = QVBoxLayout(self)
|
|
main_layout.setContentsMargins(15, 15, 15, 15)
|
|
main_layout.setSpacing(12)
|
|
|
|
# Header bar
|
|
header_layout = QHBoxLayout()
|
|
|
|
lbl_title = QLabel("Recent Uploads")
|
|
lbl_title.setStyleSheet("font-size: 16px; font-weight: bold; color: #ffffff;")
|
|
header_layout.addWidget(lbl_title)
|
|
|
|
header_layout.addStretch()
|
|
|
|
# Instance Filter Dropdown
|
|
self.cb_instance = QComboBox()
|
|
self.cb_instance.setToolTip("Filter uploads by target instance")
|
|
self.cb_instance.setStyleSheet("padding: 3px 6px;")
|
|
self.cb_instance.currentIndexChanged.connect(self.load_items)
|
|
header_layout.addWidget(self.cb_instance)
|
|
|
|
# Search bar
|
|
self.txt_search = QLineEdit()
|
|
self.txt_search.setPlaceholderText("Search uploads...")
|
|
self.txt_search.setFixedWidth(180)
|
|
self.txt_search.textChanged.connect(self.load_items)
|
|
header_layout.addWidget(self.txt_search)
|
|
|
|
btn_clear = QPushButton("Clear History")
|
|
btn_clear.clicked.connect(self.on_clear_history)
|
|
header_layout.addWidget(btn_clear)
|
|
|
|
main_layout.addLayout(header_layout)
|
|
|
|
# Scroll Area for Gallery Cards
|
|
self.scroll_area = QScrollArea()
|
|
self.scroll_area.setWidgetResizable(True)
|
|
self.scroll_area.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
|
self.scroll_area.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
|
|
self.scroll_area.setStyleSheet("QScrollArea { border: 1px solid #333340; background: #141418; border-radius: 6px; }")
|
|
|
|
self.scroll_content = QWidget()
|
|
self.grid_layout = QGridLayout(self.scroll_content)
|
|
self.grid_layout.setContentsMargins(12, 12, 12, 12)
|
|
self.grid_layout.setSpacing(12)
|
|
self.grid_layout.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignLeft)
|
|
|
|
self.scroll_area.setWidget(self.scroll_content)
|
|
main_layout.addWidget(self.scroll_area)
|
|
|
|
# Connect scroll listener for lazy loading next batch
|
|
self.scroll_area.verticalScrollBar().valueChanged.connect(self.on_scroll)
|
|
|
|
self.populate_instance_filter()
|
|
self.load_items()
|
|
|
|
def populate_instance_filter(self):
|
|
prev_signal_state = self.cb_instance.blockSignals(True)
|
|
self.cb_instance.clear()
|
|
self.cb_instance.addItem("All Instances")
|
|
|
|
config = get_env_config()
|
|
configured = [inst.get("name") for inst in config.get("instances", []) if inst.get("name")]
|
|
|
|
history = load_history()
|
|
from_history = [h.get("instance_name") for h in history if h.get("instance_name")]
|
|
|
|
all_names = sorted(list(set([n for n in (configured + from_history) if n])))
|
|
for name in all_names:
|
|
self.cb_instance.addItem(name)
|
|
|
|
self.cb_instance.blockSignals(prev_signal_state)
|
|
|
|
def get_column_count(self):
|
|
w = self.scroll_area.viewport().width()
|
|
if w <= 0:
|
|
w = self.width() - 40
|
|
card_min_w = 160
|
|
spacing = 12
|
|
margin_padding = 40
|
|
cols = max(1, (w - margin_padding + spacing) // (card_min_w + spacing))
|
|
return cols
|
|
|
|
def load_items(self):
|
|
# Clear existing cards
|
|
for card in self.cards:
|
|
card.deleteLater()
|
|
self.cards.clear()
|
|
|
|
while self.grid_layout.count():
|
|
child = self.grid_layout.takeAt(0)
|
|
if child.widget():
|
|
child.widget().deleteLater()
|
|
|
|
# Reset column stretches
|
|
for c in range(20):
|
|
self.grid_layout.setColumnStretch(c, 0)
|
|
|
|
history = load_history()
|
|
|
|
# Filter by selected Instance
|
|
selected_instance = self.cb_instance.currentText()
|
|
if selected_instance and selected_instance != "All Instances":
|
|
history = [
|
|
h for h in history
|
|
if h.get("instance_name", "Default Instance") == selected_instance or selected_instance.lower() in h.get("url", "").lower()
|
|
]
|
|
|
|
# Filter by Search Query
|
|
query = self.txt_search.text().strip().lower()
|
|
if query:
|
|
history = [
|
|
h for h in history
|
|
if query in h.get("filename", "").lower() or query in h.get("url", "").lower() or query in h.get("instance_name", "").lower()
|
|
]
|
|
|
|
self.filtered_history = history
|
|
self.loaded_count = 0
|
|
|
|
if not history:
|
|
lbl_empty = QLabel("No uploads found.")
|
|
lbl_empty.setAlignment(Qt.AlignmentFlag.AlignCenter)
|
|
lbl_empty.setStyleSheet("color: #8888a0; font-size: 14px; margin: 40px;")
|
|
self.grid_layout.addWidget(lbl_empty, 0, 0)
|
|
return
|
|
|
|
cols = self.get_column_count()
|
|
self.current_cols = cols
|
|
|
|
for c in range(cols):
|
|
self.grid_layout.setColumnStretch(c, 1)
|
|
|
|
self.load_next_batch()
|
|
|
|
def load_next_batch(self):
|
|
if self.is_loading_batch or self.loaded_count >= len(self.filtered_history):
|
|
return
|
|
|
|
self.is_loading_batch = True
|
|
try:
|
|
cols = self.get_column_count()
|
|
self.current_cols = cols
|
|
|
|
start_idx = self.loaded_count
|
|
end_idx = min(len(self.filtered_history), start_idx + self.BATCH_SIZE)
|
|
|
|
for idx in range(start_idx, end_idx):
|
|
item = self.filtered_history[idx]
|
|
card = GalleryCard(item, on_delete=self.delete_item)
|
|
self.cards.append(card)
|
|
row = idx // cols
|
|
col = idx % cols
|
|
self.grid_layout.addWidget(card, row, col)
|
|
|
|
self.loaded_count = end_idx
|
|
finally:
|
|
self.is_loading_batch = False
|
|
|
|
QTimer.singleShot(50, self.check_fill_viewport)
|
|
|
|
def check_fill_viewport(self):
|
|
vbar = self.scroll_area.verticalScrollBar()
|
|
if not vbar.isVisible() and self.loaded_count < len(self.filtered_history):
|
|
self.load_next_batch()
|
|
|
|
def on_scroll(self, value):
|
|
vbar = self.scroll_area.verticalScrollBar()
|
|
if value >= vbar.maximum() - 150:
|
|
self.load_next_batch()
|
|
|
|
def relayout_cards(self):
|
|
cols = self.get_column_count()
|
|
self.current_cols = cols
|
|
|
|
for c in range(20):
|
|
self.grid_layout.setColumnStretch(c, 1 if c < cols else 0)
|
|
|
|
# Remove cards from grid without destroying them
|
|
for card in self.cards:
|
|
self.grid_layout.removeWidget(card)
|
|
|
|
# Re-add cards at new grid positions
|
|
for idx, card in enumerate(self.cards):
|
|
row = idx // cols
|
|
col = idx % cols
|
|
self.grid_layout.addWidget(card, row, col)
|
|
|
|
def resizeEvent(self, event):
|
|
super().resizeEvent(event)
|
|
cols = self.get_column_count()
|
|
if cols != getattr(self, "current_cols", 0) and self.cards:
|
|
self.relayout_cards()
|
|
|
|
def delete_item(self, item):
|
|
cache_key = item.get("url") or item.get("file_path") or item.get("thumbnail_path")
|
|
if cache_key and cache_key in THUMBNAIL_CACHE:
|
|
del THUMBNAIL_CACHE[cache_key]
|
|
|
|
history = load_history()
|
|
history = [h for h in history if h.get("url") != item.get("url")]
|
|
save_history(history)
|
|
self.populate_instance_filter()
|
|
self.load_items()
|
|
|
|
def on_clear_history(self):
|
|
if QMessageBox.question(self, "Clear History", "Are you sure you want to clear your upload history?", QMessageBox.Yes | QMessageBox.No) == QMessageBox.Yes:
|
|
THUMBNAIL_CACHE.clear()
|
|
save_history([])
|
|
self.populate_instance_filter()
|
|
self.load_items()
|
|
|
|
# ==========================================
|
|
# Dynamic Tray Digit Icon Painter
|
|
# ==========================================
|
|
def create_digit_icon(percent: int) -> QIcon:
|
|
size = 64
|
|
pixmap = QPixmap(size, size)
|
|
pixmap.fill(Qt.GlobalColor.transparent)
|
|
|
|
painter = QPainter(pixmap)
|
|
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
|
painter.setRenderHint(QPainter.RenderHint.TextAntialiasing)
|
|
|
|
# Sharp-edged solid black badge
|
|
painter.setBrush(QColor(0, 0, 0, 255))
|
|
painter.setPen(QPen(QColor(0, 0, 0), 2))
|
|
painter.drawRect(2, 2, size - 4, size - 4)
|
|
|
|
# Crisp white digits (0 - 99)
|
|
val_str = str(min(99, max(0, percent)))
|
|
font_size = 32 if len(val_str) == 1 else 26
|
|
font = QFont("Sans-Serif", font_size, QFont.Weight.Bold)
|
|
painter.setFont(font)
|
|
painter.setPen(QColor(255, 255, 255))
|
|
|
|
painter.drawText(pixmap.rect(), Qt.AlignmentFlag.AlignCenter, val_str)
|
|
painter.end()
|
|
|
|
return QIcon(pixmap)
|
|
|
|
# ==========================================
|
|
# Thumbnail Generator Helper
|
|
# ==========================================
|
|
def generate_thumbnail(file_path: str) -> str:
|
|
if not file_path or not os.path.exists(file_path):
|
|
return ""
|
|
|
|
thumb_path = os.path.join(tempfile.gettempdir(), f"f0ckm_thumb_{int(time.time())}.png")
|
|
ext = os.path.splitext(file_path)[1].lower()
|
|
|
|
# 1. Video files -> Extract thumbnail frame using ffmpeg
|
|
video_exts = {".mp4", ".webm", ".mkv", ".avi", ".mov", ".flv", ".m4v", ".wmv"}
|
|
if ext in video_exts:
|
|
try:
|
|
res = subprocess.run(
|
|
["ffmpeg", "-y", "-ss", "00:00:01", "-i", file_path, "-vframes", "1",
|
|
"-vf", "scale=160:80:force_original_aspect_ratio=increase,crop=160:80", thumb_path],
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5
|
|
)
|
|
if res.returncode == 0 and os.path.exists(thumb_path) and os.path.getsize(thumb_path) > 0:
|
|
return thumb_path
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Image files -> Native fast PySide6 scaling
|
|
image_exts = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".avif"}
|
|
if ext in image_exts:
|
|
try:
|
|
img = QImage(file_path)
|
|
if not img.isNull():
|
|
scaled = img.scaled(160, 80, Qt.AspectRatioMode.KeepAspectRatioByExpanding, Qt.TransformationMode.SmoothTransformation)
|
|
x = max(0, (scaled.width() - 160) // 2)
|
|
y = max(0, (scaled.height() - 80) // 2)
|
|
cropped = scaled.copy(x, y, 160, 80)
|
|
if cropped.save(thumb_path, "PNG"):
|
|
return thumb_path
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. Non-image / Non-video -> Create sleek placeholder badge
|
|
try:
|
|
pixmap = QPixmap(160, 80)
|
|
pixmap.fill(QColor(24, 24, 30))
|
|
painter = QPainter(pixmap)
|
|
painter.setRenderHint(QPainter.RenderHint.Antialiasing)
|
|
|
|
painter.setPen(QPen(QColor(60, 60, 75), 2))
|
|
painter.drawRect(1, 1, 158, 78)
|
|
|
|
display_ext = ext.replace(".", "").upper()[:5] or "FILE"
|
|
painter.setFont(QFont("Sans-Serif", 16, QFont.Weight.Bold))
|
|
painter.setPen(QColor(220, 220, 240))
|
|
painter.drawText(pixmap.rect(), Qt.AlignmentFlag.AlignCenter, display_ext)
|
|
painter.end()
|
|
|
|
if pixmap.save(thumb_path, "PNG"):
|
|
return thumb_path
|
|
except Exception:
|
|
pass
|
|
|
|
return ""
|
|
|
|
def format_size(bytes_val: int) -> str:
|
|
if bytes_val < 1024:
|
|
return f"{bytes_val} B"
|
|
elif bytes_val < 1024 * 1024:
|
|
return f"{bytes_val / 1024:.1f} KB"
|
|
elif bytes_val < 1024 * 1024 * 1024:
|
|
return f"{bytes_val / (1024 * 1024):.1f} MB"
|
|
else:
|
|
return f"{bytes_val / (1024 * 1024 * 1024):.2f} GB"
|
|
|
|
def format_speed(bytes_per_sec: float) -> str:
|
|
if bytes_per_sec < 1024:
|
|
return f"{int(bytes_per_sec)} B/s"
|
|
elif bytes_per_sec < 1024 * 1024:
|
|
return f"{bytes_per_sec / 1024:.1f} KB/s"
|
|
else:
|
|
return f"{bytes_per_sec / (1024 * 1024):.1f} MB/s"
|
|
|
|
def play_success_sound(audio_path=None, volume=None):
|
|
config = get_env_config()
|
|
if not audio_path:
|
|
audio_path = config.get("success_audio_path", "")
|
|
if not audio_path or not os.path.exists(audio_path):
|
|
return
|
|
|
|
if volume is None:
|
|
volume = config.get("success_audio_volume", 15)
|
|
|
|
try:
|
|
volume = min(100, max(0, int(volume)))
|
|
except Exception:
|
|
volume = 15
|
|
|
|
ext = os.path.splitext(audio_path)[1].lower()
|
|
supported = [".wav", ".ogg", ".oga", ".flac", ".aiff", ".aif", ".au"]
|
|
if ext not in supported:
|
|
return
|
|
|
|
vol_float = volume / 100.0
|
|
pa_vol_int = int(65536 * vol_float)
|
|
|
|
# Use native Linux audio daemons (pw-play, paplay, canberra-gtk-play, aplay)
|
|
for player in ["pw-play", "paplay", "canberra-gtk-play", "aplay"]:
|
|
if shutil.which(player):
|
|
try:
|
|
if player == "pw-play":
|
|
cmd = [player, "--volume", f"{vol_float:.2f}", audio_path]
|
|
elif player == "paplay":
|
|
cmd = [player, f"--volume={pa_vol_int}", audio_path]
|
|
elif player == "canberra-gtk-play":
|
|
cmd = [player, "-f", audio_path]
|
|
else:
|
|
cmd = [player, audio_path]
|
|
|
|
subprocess.Popen(
|
|
cmd,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback to Qt QSoundEffect for WAV files
|
|
if ext == ".wav":
|
|
try:
|
|
from PySide6.QtMultimedia import QSoundEffect
|
|
effect = QSoundEffect()
|
|
effect.setSource(QUrl.fromLocalFile(audio_path))
|
|
effect.setVolume(vol_float)
|
|
effect.play()
|
|
app = QApplication.instance()
|
|
if app:
|
|
if not hasattr(app, "_sound_effects"):
|
|
app._sound_effects = []
|
|
app._sound_effects.append(effect)
|
|
effect.playingChanged.connect(
|
|
lambda: app._sound_effects.remove(effect) if not effect.isPlaying() and effect in getattr(app, "_sound_effects", []) else None
|
|
)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
class StreamingMultipartEncoder:
|
|
def __init__(self, fields, file_key, file_path):
|
|
self.boundary = f"----WebKitFormBoundary{time.time_ns():x}"
|
|
self.file_path = file_path
|
|
self.file_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
|
|
self.fields = fields
|
|
self.file_key = file_key
|
|
|
|
body_parts = []
|
|
for key, val in fields.items():
|
|
if val is not None and str(val) != "":
|
|
body_parts.append(f"--{self.boundary}\r\nContent-Disposition: form-data; name=\"{key}\"\r\n\r\n{val}\r\n".encode('utf-8'))
|
|
|
|
filename = os.path.basename(file_path)
|
|
mime_type, _ = mimetypes.guess_type(file_path)
|
|
if not mime_type:
|
|
ext = os.path.splitext(file_path)[1].lower()
|
|
if ext in ['.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp']:
|
|
mime_type = f"image/{ext.lstrip('.')}"
|
|
if mime_type == "image/jpg": mime_type = "image/jpeg"
|
|
elif ext in ['.mp4', '.webm', '.mkv', '.mov', '.avi']:
|
|
mime_type = f"video/{ext.lstrip('.')}"
|
|
else:
|
|
mime_type = "image/png"
|
|
|
|
file_header = f"--{self.boundary}\r\nContent-Disposition: form-data; name=\"{self.file_key}\"; filename=\"{filename}\"\r\nContent-Type: {mime_type}\r\n\r\n".encode('utf-8')
|
|
file_footer = f"\r\n--{self.boundary}--\r\n".encode('utf-8')
|
|
|
|
self.header_bytes = b"".join(body_parts) + file_header
|
|
self.footer_bytes = file_footer
|
|
self.total_size = len(self.header_bytes) + self.file_size + len(self.footer_bytes)
|
|
self.content_type = f"multipart/form-data; boundary={self.boundary}"
|
|
|
|
class URLPostThread(QThread):
|
|
finished = Signal(bool, str) # success, response_text_or_error
|
|
|
|
def __init__(self, api_url, api_key, url_param, rating, tags, visibility, is_oc, parent=None):
|
|
super().__init__(parent)
|
|
self.api_url = api_url
|
|
self.api_key = api_key
|
|
self.url_param = url_param
|
|
self.rating = rating
|
|
self.tags = tags
|
|
self.visibility = visibility
|
|
self.is_oc = is_oc
|
|
|
|
def run(self):
|
|
print(f"[f0ckm-gui] [URL POST START] Sending JSON URL payload '{self.url_param}' to '{self.api_url}'", flush=True)
|
|
try:
|
|
payload = {
|
|
"url": self.url_param,
|
|
"tags": self.tags if self.tags else "url,upload",
|
|
"visibility": str(self.visibility) if self.visibility is not None else "0"
|
|
}
|
|
if self.rating and str(self.rating).lower() not in ["", "default", "none"]:
|
|
rating_map = {"s": "sfw", "q": "nsfw", "e": "nsfl", "sfw": "sfw", "nsfw": "nsfw", "nsfl": "nsfl"}
|
|
payload["rating"] = rating_map.get(str(self.rating).lower(), str(self.rating).lower())
|
|
if self.is_oc:
|
|
payload["is_oc"] = "1"
|
|
|
|
json_bytes = json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
self.api_url,
|
|
data=json_bytes,
|
|
headers={
|
|
"User-Agent": "f0ckm-uploader/2.0",
|
|
"X-Api-Key": self.api_key,
|
|
"Content-Type": "application/json"
|
|
},
|
|
method="POST"
|
|
)
|
|
with urllib.request.urlopen(req, timeout=300) as resp:
|
|
resp_text = resp.read().decode("utf-8")
|
|
print(f"[f0ckm-gui] [URL POST OK] Server Response: {resp_text}", flush=True)
|
|
self.finished.emit(True, resp_text)
|
|
except Exception as e:
|
|
print(f"[f0ckm-gui] [URL POST ERROR] {e}", flush=True)
|
|
self.finished.emit(False, str(e))
|
|
|
|
class StreamingDownloadThread(QThread):
|
|
progress = Signal(int, int) # bytes_received, total_bytes
|
|
finished = Signal(bool, str) # success, file_path_or_error_msg
|
|
|
|
def __init__(self, url, parent=None):
|
|
super().__init__(parent)
|
|
self.url = url
|
|
self.aborted = False
|
|
|
|
def abort(self):
|
|
self.aborted = True
|
|
|
|
def run(self):
|
|
print(f"[f0ckm-gui] [DOWNLOAD START] Requesting remote URL: {self.url}", flush=True)
|
|
try:
|
|
req = urllib.request.Request(
|
|
self.url,
|
|
headers={
|
|
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
}
|
|
)
|
|
with urllib.request.urlopen(req, timeout=120) as resp:
|
|
total_size = int(resp.headers.get("Content-Length", 0))
|
|
content_type = resp.headers.get("Content-Type", "")
|
|
print(f"[f0ckm-gui] [DOWNLOAD HTTP OK] Content-Type: '{content_type}', Length: {total_size} bytes", flush=True)
|
|
|
|
url_path = urllib.parse.urlparse(self.url).path
|
|
ext = os.path.splitext(url_path)[1].lower()
|
|
if not ext or len(ext) > 6:
|
|
mime_base = content_type.split(";")[0].strip() if content_type else ""
|
|
ext = mimetypes.guess_extension(mime_base) or ".png"
|
|
if ext == ".jpe":
|
|
ext = ".jpg"
|
|
|
|
clean_name = os.path.basename(url_path) or f"remote_{int(time.time())}"
|
|
if not os.path.splitext(clean_name)[1]:
|
|
clean_name += ext
|
|
|
|
temp_dir = tempfile.gettempdir()
|
|
temp_path = os.path.join(temp_dir, f"f0ckm_remote_{int(time.time())}_{clean_name}")
|
|
|
|
received = 0
|
|
chunk_size = 131072
|
|
last_emit = 0.0
|
|
with open(temp_path, "wb") as f:
|
|
while True:
|
|
if self.aborted:
|
|
print("[f0ckm-gui] [DOWNLOAD ABORTED]", flush=True)
|
|
self.finished.emit(False, "ABORTED")
|
|
return
|
|
chunk = resp.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
f.write(chunk)
|
|
received += len(chunk)
|
|
now = time.time()
|
|
if now - last_emit >= 0.05 or (total_size > 0 and received == total_size):
|
|
self.progress.emit(received, total_size)
|
|
last_emit = now
|
|
|
|
print(f"[f0ckm-gui] [DOWNLOAD COMPLETE] Saved remote file to: {temp_path} ({received} bytes)", flush=True)
|
|
self.finished.emit(True, temp_path)
|
|
except Exception as e:
|
|
print(f"[f0ckm-gui] [DOWNLOAD ERROR] Failed to fetch {self.url}: {e}", flush=True)
|
|
if self.aborted:
|
|
self.finished.emit(False, "ABORTED")
|
|
else:
|
|
self.finished.emit(False, str(e))
|
|
|
|
class StreamingUploadThread(QThread):
|
|
progress = Signal(int, int) # bytes_sent, bytes_total
|
|
finished = Signal(bool, str) # success, response_or_error_msg
|
|
|
|
def __init__(self, api_url, api_key, file_path, rating, tags, visibility, is_oc, parent=None):
|
|
super().__init__(parent)
|
|
self.api_url = api_url
|
|
self.api_key = api_key
|
|
self.file_path = file_path
|
|
self.rating = rating
|
|
self.tags = tags
|
|
self.visibility = visibility
|
|
self.is_oc = is_oc
|
|
self.aborted = False
|
|
self.sock = None
|
|
|
|
def abort(self):
|
|
self.aborted = True
|
|
if self.sock:
|
|
try:
|
|
self.sock.close()
|
|
except Exception:
|
|
pass
|
|
|
|
def run(self):
|
|
print(f"[f0ckm-gui] [UPLOAD START] File: '{self.file_path}' -> Target API: '{self.api_url}'", flush=True)
|
|
fields = {
|
|
"rating": self.rating,
|
|
"tags": self.tags,
|
|
"visibility": self.visibility,
|
|
"is_oc": self.is_oc
|
|
}
|
|
|
|
try:
|
|
encoder = StreamingMultipartEncoder(fields, "file", self.file_path)
|
|
parsed_url = urllib.parse.urlparse(self.api_url)
|
|
|
|
host = parsed_url.hostname
|
|
port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80)
|
|
path = parsed_url.path or "/"
|
|
if parsed_url.query:
|
|
path += "?" + parsed_url.query
|
|
|
|
raw_sock = socket.create_connection((host, port), timeout=30)
|
|
raw_sock.settimeout(1800.0)
|
|
|
|
if parsed_url.scheme == "https":
|
|
ctx = ssl.create_default_context()
|
|
self.sock = ctx.wrap_socket(raw_sock, server_hostname=host)
|
|
else:
|
|
self.sock = raw_sock
|
|
|
|
req_headers = (
|
|
f"POST {path} HTTP/1.1\r\n"
|
|
f"Host: {host}\r\n"
|
|
f"User-Agent: f0ckm-uploader/2.0\r\n"
|
|
f"X-Api-Key: {self.api_key}\r\n"
|
|
f"Content-Type: {encoder.content_type}\r\n"
|
|
f"Content-Length: {encoder.total_size}\r\n"
|
|
f"Connection: close\r\n\r\n"
|
|
).encode('utf-8')
|
|
|
|
self.sock.sendall(req_headers)
|
|
|
|
total_sent = 0
|
|
self.sock.sendall(encoder.header_bytes)
|
|
total_sent += len(encoder.header_bytes)
|
|
self.progress.emit(total_sent, encoder.total_size)
|
|
|
|
chunk_size = 131072
|
|
last_progress_time = 0.0
|
|
|
|
with open(self.file_path, "rb") as f:
|
|
while True:
|
|
if self.aborted:
|
|
self.finished.emit(False, "ABORTED")
|
|
return
|
|
chunk = f.read(chunk_size)
|
|
if not chunk:
|
|
break
|
|
|
|
self.sock.sendall(chunk)
|
|
total_sent += len(chunk)
|
|
|
|
now = time.time()
|
|
if now - last_progress_time >= 0.05 or total_sent >= encoder.total_size:
|
|
pct = min(99, int((total_sent / encoder.total_size) * 100))
|
|
bytes_sent_calc = int(encoder.total_size * (pct / 100.0))
|
|
self.progress.emit(bytes_sent_calc, encoder.total_size)
|
|
last_progress_time = now
|
|
|
|
self.sock.sendall(encoder.footer_bytes)
|
|
total_sent += len(encoder.footer_bytes)
|
|
self.progress.emit(encoder.total_size, encoder.total_size)
|
|
|
|
response_data = bytearray()
|
|
while not self.aborted:
|
|
data = self.sock.recv(4096)
|
|
if not data:
|
|
break
|
|
response_data.extend(data)
|
|
|
|
if self.aborted:
|
|
self.finished.emit(False, "ABORTED")
|
|
return
|
|
|
|
parts = response_data.split(b"\r\n\r\n", 1)
|
|
headers_str = parts[0].decode('utf-8', errors='replace') if parts else ""
|
|
raw_body = parts[1].decode('utf-8', errors='replace') if len(parts) > 1 else response_data.decode('utf-8', errors='replace')
|
|
status_line = headers_str.split("\r\n")[0] if headers_str else ""
|
|
|
|
# De-chunk HTTP/1.1 response body if Transfer-Encoding: chunked or chunk-prefixed
|
|
body = raw_body
|
|
if "transfer-encoding: chunked" in headers_str.lower() or re.match(r'^[0-9a-fA-F]+\r\n', raw_body.lstrip()):
|
|
decoded_chunks = []
|
|
rest = raw_body.lstrip()
|
|
try:
|
|
while rest:
|
|
lines = rest.split("\r\n", 1)
|
|
if len(lines) < 2:
|
|
break
|
|
hex_str = lines[0].strip().split(";")[0]
|
|
try:
|
|
chunk_len = int(hex_str, 16)
|
|
except ValueError:
|
|
break
|
|
if chunk_len == 0:
|
|
break
|
|
chunk_content = lines[1][:chunk_len]
|
|
decoded_chunks.append(chunk_content)
|
|
rest = lines[1][chunk_len:]
|
|
if rest.startswith("\r\n"):
|
|
rest = rest[2:]
|
|
if decoded_chunks:
|
|
body = "".join(decoded_chunks)
|
|
except Exception:
|
|
pass
|
|
|
|
# Extra safety fallback: extract JSON substring if there's any hex prefix left
|
|
if not body.strip().startswith("{") and "{" in body and "}" in body:
|
|
match = re.search(r'\{.*\}', body, re.DOTALL)
|
|
if match:
|
|
body = match.group(0)
|
|
|
|
if "200" in status_line or "201" in status_line:
|
|
self.finished.emit(True, body)
|
|
else:
|
|
self.finished.emit(False, body or status_line)
|
|
|
|
except Exception as e:
|
|
if self.aborted:
|
|
self.finished.emit(False, "ABORTED")
|
|
else:
|
|
err_msg = str(e)
|
|
if isinstance(e, (socket.timeout, TimeoutError)) or "timed out" in err_msg.lower():
|
|
err_msg = f"Upload timed out: socket read timed out ({err_msg})"
|
|
self.finished.emit(False, err_msg)
|
|
finally:
|
|
if self.sock:
|
|
try:
|
|
self.sock.close()
|
|
except Exception:
|
|
pass
|
|
|
|
# ==========================================
|
|
# System Tray Application Manager
|
|
# ==========================================
|
|
class SystemTrayApp(QObject):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.tray = QSystemTrayIcon()
|
|
|
|
self.load_tray_icon()
|
|
self.tray.setToolTip("f0ckm Uploader (Click to capture & upload)")
|
|
|
|
self.network_manager = QNetworkAccessManager(self)
|
|
self.current_reply = None
|
|
|
|
# Queue system state
|
|
self.upload_queue = [] # List of tuples: (target_str, is_url)
|
|
self.is_uploading = False
|
|
self.active_thread = None
|
|
self._running_threads = set()
|
|
self.current_target = None
|
|
self.total_queue_count = 0
|
|
self.current_item_index = 0
|
|
|
|
self.menu = QMenu()
|
|
|
|
# Header entry
|
|
title_action = QAction("f0ckm Uploader", self.menu)
|
|
title_action.setEnabled(False)
|
|
self.menu.addAction(title_action)
|
|
|
|
# Instance Submenu
|
|
self.instance_menu = QMenu("Switch Instance", self.menu)
|
|
self.menu.addMenu(self.instance_menu)
|
|
|
|
self.menu.addSeparator()
|
|
|
|
# Actions
|
|
capture_action = QAction("Capture Region & Upload", self.menu)
|
|
capture_action.triggered.connect(self.on_capture_and_upload)
|
|
self.menu.addAction(capture_action)
|
|
|
|
upload_action = QAction("Upload File...", self.menu)
|
|
upload_action.triggered.connect(self.on_upload_file)
|
|
self.menu.addAction(upload_action)
|
|
|
|
upload_url_action = QAction("Upload URL...", self.menu)
|
|
upload_url_action.triggered.connect(self.on_upload_url)
|
|
self.menu.addAction(upload_url_action)
|
|
|
|
clipboard_action = QAction("Upload from Clipboard", self.menu)
|
|
clipboard_action.triggered.connect(self.on_upload_clipboard)
|
|
self.menu.addAction(clipboard_action)
|
|
|
|
self.abort_action = QAction("Abort Upload", self.menu)
|
|
self.abort_action.setEnabled(False)
|
|
self.abort_action.triggered.connect(self.on_abort_upload)
|
|
self.menu.addAction(self.abort_action)
|
|
|
|
settings_action = QAction("Settings...", self.menu)
|
|
settings_action.triggered.connect(lambda checked=False: self.on_open_settings())
|
|
self.menu.addAction(settings_action)
|
|
|
|
gallery_action = QAction("Recent Uploads Gallery", self.menu)
|
|
gallery_action.triggered.connect(lambda checked=False: self.on_open_gallery())
|
|
self.menu.addAction(gallery_action)
|
|
|
|
self.menu.addSeparator()
|
|
|
|
quit_action = QAction("Quit", self.menu)
|
|
quit_action.triggered.connect(QApplication.quit)
|
|
self.menu.addAction(quit_action)
|
|
|
|
self.tray.setContextMenu(self.menu)
|
|
self.tray.activated.connect(self.on_tray_activated)
|
|
self.last_uploaded_url = ""
|
|
self.tray.messageClicked.connect(self.on_notification_clicked)
|
|
|
|
self.settings_dialog = None
|
|
self.gallery_window = None
|
|
|
|
self.rebuild_instance_menu()
|
|
|
|
def _start_thread(self, thread):
|
|
self._running_threads.add(thread)
|
|
thread.finished.connect(thread.deleteLater)
|
|
thread.finished.connect(lambda: self._running_threads.discard(thread))
|
|
self.active_thread = thread
|
|
thread.start()
|
|
|
|
def rebuild_instance_menu(self):
|
|
self.instance_menu.clear()
|
|
self.instance_actions = []
|
|
|
|
config = get_env_config()
|
|
instances = config.get("instances", [])
|
|
active_idx = config.get("active_instance_index", 0)
|
|
|
|
active_name = config.get("active_instance_name", "Default Instance")
|
|
self.instance_menu.setTitle(f"Instance: {active_name}")
|
|
|
|
self.instance_action_group = QActionGroup(self.instance_menu)
|
|
self.instance_action_group.setExclusive(True)
|
|
|
|
for idx, inst in enumerate(instances):
|
|
name = inst.get("name", f"Instance {idx + 1}")
|
|
url = inst.get("api_url", "")
|
|
action = QAction(name, self.instance_menu)
|
|
action.setCheckable(True)
|
|
self.instance_action_group.addAction(action)
|
|
if url:
|
|
action.setToolTip(url)
|
|
if idx == active_idx:
|
|
action.setChecked(True)
|
|
|
|
action.triggered.connect(lambda checked=False, i=idx: self.switch_active_instance(i))
|
|
self.instance_menu.addAction(action)
|
|
self.instance_actions.append(action)
|
|
|
|
self.instance_menu.addSeparator()
|
|
manage_action = QAction("Manage Instances...", self.instance_menu)
|
|
manage_action.triggered.connect(lambda checked=False: self.on_open_settings())
|
|
self.instance_menu.addAction(manage_action)
|
|
|
|
def switch_active_instance(self, index):
|
|
config = get_env_config()
|
|
instances = config.get("instances", [])
|
|
if 0 <= index < len(instances):
|
|
config["active_instance_index"] = index
|
|
active_inst = instances[index]
|
|
config["api_url"] = active_inst.get("api_url", "")
|
|
config["api_key"] = active_inst.get("api_key", "")
|
|
save_config(config)
|
|
|
|
name = active_inst.get("name", "Instance")
|
|
print(f"[f0ckm-gui] Switched active instance to '{name}' ({active_inst.get('api_url', '')})", flush=True)
|
|
self.show_message("f0ckm Instance Switched", f"Active instance changed to: {name}", QSystemTrayIcon.Information)
|
|
|
|
self.instance_menu.setTitle(f"Instance: {name}")
|
|
if hasattr(self, "instance_actions") and 0 <= index < len(self.instance_actions):
|
|
self.instance_actions[index].setChecked(True)
|
|
|
|
if self.settings_dialog:
|
|
self.settings_dialog.load_current_settings()
|
|
|
|
def enqueue_targets(self, targets):
|
|
if isinstance(targets, str):
|
|
targets = [targets]
|
|
valid_targets = []
|
|
for t in targets:
|
|
target_str, is_url = parse_upload_target(t)
|
|
if target_str:
|
|
valid_targets.append((target_str, is_url))
|
|
|
|
if not valid_targets:
|
|
return
|
|
|
|
if not self.is_uploading and len(self.upload_queue) == 0:
|
|
self.total_queue_count = len(valid_targets)
|
|
self.current_item_index = 0
|
|
self.upload_queue.extend(valid_targets)
|
|
self.process_next_in_queue()
|
|
else:
|
|
self.upload_queue.extend(valid_targets)
|
|
self.total_queue_count += len(valid_targets)
|
|
print(f"[f0ckm-gui] [QUEUE] Enqueued {len(valid_targets)} items. Total in queue: {len(self.upload_queue)} (Batch total: {self.total_queue_count})", flush=True)
|
|
|
|
def process_next_in_queue(self):
|
|
if len(self.upload_queue) == 0:
|
|
self.is_uploading = False
|
|
self.current_target = None
|
|
self.active_thread = None
|
|
self.total_queue_count = 0
|
|
self.current_item_index = 0
|
|
self.abort_action.setEnabled(False)
|
|
self.tray.setIcon(self.default_icon)
|
|
self.tray.setToolTip("f0ckm Uploader (Click to capture & upload)")
|
|
return
|
|
|
|
target_tuple = self.upload_queue.pop(0)
|
|
target_str, is_url = target_tuple
|
|
self.is_uploading = True
|
|
self.current_target = target_str
|
|
self.current_item_index += 1
|
|
self.abort_action.setEnabled(True)
|
|
|
|
print(f"[f0ckm-gui] [QUEUE ITEM {self.current_item_index}/{self.total_queue_count}] Processing target: '{target_str}' (is_url={is_url})", flush=True)
|
|
|
|
if is_url:
|
|
self.download_and_upload_url(target_str)
|
|
else:
|
|
self.upload_file_direct(target_str)
|
|
|
|
def _on_single_task_complete(self):
|
|
self.active_thread = None
|
|
QTimer.singleShot(50, self.process_next_in_queue)
|
|
|
|
def on_notification_clicked(self):
|
|
url = getattr(self, "last_uploaded_url", "")
|
|
if url:
|
|
print(f"[f0ckm-gui] Notification clicked -> Opening URL in browser: '{url}'", flush=True)
|
|
QDesktopServices.openUrl(QUrl(url))
|
|
|
|
def on_open_gallery(self, checked=False):
|
|
if self.gallery_window is None:
|
|
self.gallery_window = GalleryWindow()
|
|
self.gallery_window.load_items()
|
|
if self.gallery_window.isMinimized():
|
|
self.gallery_window.showNormal()
|
|
self.gallery_window.show()
|
|
self.gallery_window.raise_()
|
|
self.gallery_window.activateWindow()
|
|
|
|
def show(self):
|
|
self.tray.show()
|
|
|
|
def on_tray_activated(self, reason):
|
|
if reason == QSystemTrayIcon.Trigger:
|
|
# Single click on tray icon starts Spectacle capture & upload
|
|
self.on_capture_and_upload()
|
|
elif reason == QSystemTrayIcon.DoubleClick:
|
|
self.on_open_settings()
|
|
|
|
def load_tray_icon(self):
|
|
config = get_env_config()
|
|
theme = config.get("icon_theme", "dark").lower()
|
|
filename = "icon_light.svg" if theme == "light" else "icon.svg"
|
|
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
candidates = [
|
|
os.path.join(script_dir, filename),
|
|
f"/home/kibi/Projects/f0ckm-uploader/{filename}",
|
|
os.path.expanduser(f"~/.local/bin/{filename}")
|
|
]
|
|
icon_path = None
|
|
for c in candidates:
|
|
if os.path.exists(c):
|
|
icon_path = c
|
|
break
|
|
|
|
if icon_path:
|
|
self.default_icon = QIcon(icon_path)
|
|
else:
|
|
self.default_icon = QIcon.fromTheme("cloud-upload-symbolic", QIcon.fromTheme("network-server"))
|
|
|
|
self.tray.setIcon(self.default_icon)
|
|
|
|
def on_open_settings(self, checked=False):
|
|
if not self.settings_dialog:
|
|
self.settings_dialog = SettingsDialog(tray_app=self)
|
|
self.settings_dialog.load_current_settings()
|
|
if self.settings_dialog.isMinimized():
|
|
self.settings_dialog.showNormal()
|
|
self.settings_dialog.show()
|
|
self.settings_dialog.raise_()
|
|
self.settings_dialog.activateWindow()
|
|
|
|
def on_capture_and_upload(self):
|
|
screenshot_dir = os.path.expanduser("~/Pictures/Screenshots")
|
|
os.makedirs(screenshot_dir, exist_ok=True)
|
|
ts = time.strftime('%Y%m%d_%H%M%S')
|
|
ms = int(time.time() * 1000) % 1000
|
|
save_path = os.path.join(screenshot_dir, f"screenshot_{ts}_{ms:03d}.png")
|
|
|
|
try:
|
|
proc = QProcess(self)
|
|
proc.finished.connect(lambda exit_code, exit_status, p=proc, sp=save_path: self._on_spectacle_finished(p, sp, exit_code))
|
|
proc.start("spectacle", ["-r", "-b", "-n", "-o", save_path])
|
|
except Exception as e:
|
|
self.show_message("Upload Error", f"Failed to capture screenshot: {e}", QSystemTrayIcon.Critical)
|
|
|
|
def _on_spectacle_finished(self, proc, save_path, exit_code):
|
|
proc.deleteLater()
|
|
if os.path.exists(save_path) and os.path.getsize(save_path) > 0:
|
|
self.enqueue_targets([save_path])
|
|
|
|
def on_abort_upload(self):
|
|
print(f"[f0ckm-gui] [QUEUE] Abort requested. Clearing queue ({len(self.upload_queue)} remaining items) and stopping active upload.", flush=True)
|
|
self.upload_queue.clear()
|
|
self.total_queue_count = 0
|
|
self.current_item_index = 0
|
|
if self.active_thread and hasattr(self.active_thread, "abort"):
|
|
self.active_thread.abort()
|
|
if self.current_reply and self.current_reply.isRunning():
|
|
self.current_reply.abort()
|
|
|
|
def upload_target_direct(self, target):
|
|
self.enqueue_targets([target])
|
|
|
|
def download_and_upload_url(self, url):
|
|
config = get_env_config()
|
|
api_url = config.get("api_url", "")
|
|
api_key = config.get("api_key", "")
|
|
rating = config.get("default_rating", "")
|
|
tags = config.get("default_tags", "")
|
|
visibility = config.get("default_visibility", "0")
|
|
is_oc = config.get("default_is_oc", False)
|
|
|
|
print(f"[f0ckm-gui] [URL UPLOAD] Initiating URL upload: '{url}' (API URL: '{api_url}')", flush=True)
|
|
if not api_url or not api_key:
|
|
print(f"[f0ckm-gui] [CONFIG ERROR] API URL or API Key missing in configuration!", flush=True)
|
|
self.show_message("Upload Error", "API URL and API Key must be set in Settings.", QSystemTrayIcon.Critical)
|
|
QTimer.singleShot(100, self._on_single_task_complete)
|
|
return
|
|
|
|
self.download_url = url
|
|
self.download_start_time = time.time()
|
|
self.tray.setIcon(create_digit_icon(50))
|
|
|
|
batch_prefix = f"[{self.current_item_index}/{self.total_queue_count}] " if self.total_queue_count > 1 else ""
|
|
rem_str = f"\n({len(self.upload_queue)} waiting in queue)" if self.upload_queue else ""
|
|
self.tray.setToolTip(f"Uploading URL {batch_prefix}to f0ckm...\n{url[:45]}{rem_str}")
|
|
|
|
thread = URLPostThread(api_url, api_key, url, rating, tags, visibility, is_oc)
|
|
thread.finished.connect(lambda ok, res: self.on_url_post_finished(ok, res, url))
|
|
self._start_thread(thread)
|
|
|
|
def on_url_post_finished(self, ok, result, url):
|
|
if ok:
|
|
try:
|
|
data = json.loads(result)
|
|
if data.get("success") and (data.get("url") or data.get("file_url") or data.get("direct_url") or data.get("post_url") or data.get("file")):
|
|
post_url = data.get("url") or data.get("post_url") or ""
|
|
direct_url = data.get("file_url") or data.get("direct_url") or data.get("file") or ""
|
|
config = get_env_config()
|
|
url_type = config.get("clipboard_url_type", "post").lower()
|
|
final_url = direct_url if (url_type == "direct" and direct_url) else (post_url or direct_url)
|
|
|
|
copy_to_clipboard(final_url)
|
|
play_success_sound(config.get("sound_file"), config.get("sound_volume"))
|
|
|
|
entry = {
|
|
"timestamp": int(time.time()),
|
|
"file_path": url,
|
|
"file_name": url.split("/")[-1] or "url_upload",
|
|
"file_size": 0,
|
|
"url": final_url,
|
|
"post_url": post_url,
|
|
"file_url": direct_url
|
|
}
|
|
add_history_entry(entry)
|
|
|
|
self.last_uploaded_url = final_url
|
|
self.show_success_notification(url, final_url)
|
|
self._on_single_task_complete()
|
|
return
|
|
except Exception as e:
|
|
print(f"[f0ckm-gui] JSON parse notice: {e}", flush=True)
|
|
|
|
print(f"[f0ckm-gui] Direct URL API notice: Falling back to local download of '{url}'", flush=True)
|
|
thread = StreamingDownloadThread(url)
|
|
thread.progress.connect(self.on_download_progress)
|
|
thread.finished.connect(lambda ok_dl, res_dl: self.on_download_finished(ok_dl, res_dl, url))
|
|
self._start_thread(thread)
|
|
|
|
def on_download_progress(self, bytes_received, bytes_total):
|
|
batch_prefix = f"[{self.current_item_index}/{self.total_queue_count}] " if self.total_queue_count > 1 else ""
|
|
rem_str = f"\n({len(self.upload_queue)} waiting in queue)" if self.upload_queue else ""
|
|
if bytes_total > 0:
|
|
pct = min(99, max(0, int((bytes_received / bytes_total) * 100)))
|
|
self.tray.setIcon(create_digit_icon(pct))
|
|
rec_str = format_size(bytes_received)
|
|
tot_str = format_size(bytes_total)
|
|
url_str = getattr(self, "download_url", "")
|
|
short_url = (url_str[:35] + "...") if len(url_str) > 35 else url_str
|
|
self.tray.setToolTip(f"Downloading {batch_prefix}: {short_url}\n{pct}% ({rec_str} / {tot_str}){rem_str}")
|
|
else:
|
|
rec_str = format_size(bytes_received)
|
|
self.tray.setToolTip(f"Downloading remote file {batch_prefix}...\n{rec_str}{rem_str}")
|
|
|
|
def on_download_finished(self, success, result, url):
|
|
if not success:
|
|
if result == "ABORTED":
|
|
self.show_message("Download Aborted", "Remote URL download was canceled.", QSystemTrayIcon.Information)
|
|
else:
|
|
self.show_message("Download Failed", f"Failed to download remote URL: {result}", QSystemTrayIcon.Critical)
|
|
self._on_single_task_complete()
|
|
return
|
|
|
|
self.upload_file_direct(result)
|
|
|
|
def upload_file_direct(self, file_path):
|
|
file_path = normalize_file_path(file_path)
|
|
if not file_path or not os.path.exists(file_path):
|
|
self.show_message("Upload Error", f"File not found: {file_path}", QSystemTrayIcon.Critical)
|
|
QTimer.singleShot(100, self._on_single_task_complete)
|
|
return
|
|
|
|
config = get_env_config()
|
|
api_url = config.get("api_url", "")
|
|
api_key = config.get("api_key", "")
|
|
if not api_url or not api_key:
|
|
self.show_message("Upload Error", "API URL and API Key must be set in Settings.", QSystemTrayIcon.Critical)
|
|
QTimer.singleShot(100, self._on_single_task_complete)
|
|
return
|
|
|
|
rating = config.get("default_rating", "")
|
|
tags = config.get("default_tags", "")
|
|
if tags:
|
|
tag_list = [t.strip() for t in re.split(r'[,]+', tags) if t.strip()]
|
|
tags = ", ".join(tag_list)
|
|
|
|
visibility = str(config.get("default_visibility", "0"))
|
|
is_oc = "1" if config.get("default_is_oc", False) else "0"
|
|
|
|
filename = os.path.basename(file_path)
|
|
self.upload_filename = filename
|
|
self.upload_start_time = time.time()
|
|
|
|
self.tray.setIcon(create_digit_icon(0))
|
|
|
|
batch_prefix = f"[{self.current_item_index}/{self.total_queue_count}] " if self.total_queue_count > 1 else ""
|
|
rem_str = f"\n({len(self.upload_queue)} waiting in queue)" if self.upload_queue else ""
|
|
self.tray.setToolTip(f"Uploading {batch_prefix}{filename}...\n0% (0 B){rem_str}")
|
|
|
|
thread = StreamingUploadThread(api_url, api_key, file_path, rating, tags, visibility, is_oc)
|
|
thread.progress.connect(self.on_upload_progress)
|
|
thread.finished.connect(lambda ok, resp: self.on_upload_finished(ok, resp, file_path))
|
|
self._start_thread(thread)
|
|
|
|
def on_upload_progress(self, bytes_sent, bytes_total):
|
|
if bytes_total > 0:
|
|
pct = min(100, max(0, int((bytes_sent / bytes_total) * 100)))
|
|
self.tray.setIcon(create_digit_icon(pct))
|
|
|
|
elapsed = time.time() - getattr(self, "upload_start_time", time.time())
|
|
speed = (bytes_sent / elapsed) if elapsed > 0.1 else 0
|
|
|
|
sent_str = format_size(bytes_sent)
|
|
total_str = format_size(bytes_total)
|
|
filename = getattr(self, "upload_filename", "file")
|
|
|
|
batch_prefix = f"[{self.current_item_index}/{self.total_queue_count}] " if self.total_queue_count > 1 else ""
|
|
remaining = len(self.upload_queue)
|
|
rem_str = f"\n({remaining} waiting in queue)" if remaining > 0 else ""
|
|
|
|
if pct >= 99:
|
|
tooltip = f"Uploading {batch_prefix}: {filename}\n99% ({total_str})\nProcessing upload on server...{rem_str}"
|
|
else:
|
|
speed_str = format_speed(speed)
|
|
remaining_bytes = bytes_total - bytes_sent
|
|
eta_sec = int(remaining_bytes / speed) if speed > 0 else 0
|
|
if eta_sec < 60:
|
|
eta_str = f"{eta_sec}s"
|
|
elif eta_sec < 3600:
|
|
eta_str = f"{eta_sec // 60}m {eta_sec % 60}s"
|
|
else:
|
|
eta_str = f"{eta_sec // 3600}h {(eta_sec % 3600) // 60}m"
|
|
|
|
tooltip = f"Uploading {batch_prefix}: {filename}\n{pct}% ({sent_str} / {total_str})\nSpeed: {speed_str} | ETA: {eta_str}{rem_str}"
|
|
|
|
self.tray.setToolTip(tooltip)
|
|
|
|
def on_upload_finished(self, success, response_text, file_path):
|
|
if not success:
|
|
if response_text == "ABORTED":
|
|
self.show_message("Upload Aborted", "The active upload was canceled.", QSystemTrayIcon.Information)
|
|
else:
|
|
self.show_message("Upload Failed", response_text, QSystemTrayIcon.Critical)
|
|
self._on_single_task_complete()
|
|
return
|
|
|
|
try:
|
|
res = json.loads(response_text)
|
|
if res.get("success") and (res.get("url") or res.get("file_url") or res.get("direct_url") or res.get("target_url") or res.get("file")):
|
|
config = get_env_config()
|
|
url_type = config.get("clipboard_url_type", "post").lower()
|
|
|
|
post_url = res.get("url") or res.get("post_url") or ""
|
|
direct_url = res.get("file_url") or res.get("direct_url") or res.get("target_url") or res.get("file") or ""
|
|
|
|
if url_type == "direct" and direct_url:
|
|
item_url = direct_url
|
|
else:
|
|
item_url = post_url or direct_url or res.get("url", "")
|
|
|
|
copy_to_clipboard(item_url)
|
|
|
|
# Add to history gallery
|
|
thumb = generate_thumbnail(file_path)
|
|
add_history_entry({
|
|
"timestamp": time.time(),
|
|
"filename": os.path.basename(file_path),
|
|
"file_path": file_path,
|
|
"url": item_url,
|
|
"thumbnail_path": thumb if thumb and os.path.exists(thumb) else ""
|
|
})
|
|
if hasattr(self, "gallery_window") and self.gallery_window and self.gallery_window.isVisible():
|
|
self.gallery_window.load_items()
|
|
|
|
self.show_success_notification(file_path, item_url)
|
|
else:
|
|
msg = res.get("msg", "Upload failed")
|
|
self.show_message("Upload Failed", msg, QSystemTrayIcon.Critical)
|
|
except Exception as e:
|
|
self.show_message("Upload Error", f"Error parsing response: {e}\nResponse: {response_text}", QSystemTrayIcon.Critical)
|
|
finally:
|
|
self._on_single_task_complete()
|
|
|
|
def show_success_notification(self, file_path, item_url):
|
|
self.last_uploaded_url = item_url
|
|
config = get_env_config()
|
|
if config.get("play_success_sound", False) or config.get("success_audio_path"):
|
|
audio_path = config.get("success_audio_path", "")
|
|
volume = config.get("success_audio_volume", 15)
|
|
if audio_path:
|
|
play_success_sound(audio_path, volume=volume)
|
|
|
|
if not config.get("enable_notifications", True):
|
|
return
|
|
|
|
abs_path = os.path.abspath(file_path)
|
|
ext = os.path.splitext(abs_path)[1].lower()
|
|
if ext in [".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".avif"]:
|
|
img_path = abs_path
|
|
else:
|
|
img_path = generate_thumbnail(abs_path) or abs_path
|
|
|
|
file_url = "file://" + abs_path
|
|
|
|
body_text = f'<a href="{item_url}">{item_url}</a>'
|
|
|
|
# Direct DBus Notification for Spectacle-style centered image preview & file drag-and-drop
|
|
try:
|
|
import dbus
|
|
bus = dbus.SessionBus()
|
|
notify_obj = bus.get_object('org.freedesktop.Notifications', '/org/freedesktop/Notifications')
|
|
notify_iface = dbus.Interface(notify_obj, 'org.freedesktop.Notifications')
|
|
|
|
hints = {
|
|
'urls': dbus.Array([dbus.String(file_url)], signature='s', variant_level=1),
|
|
'x-kde-urls': dbus.Array([dbus.String(file_url)], signature='s', variant_level=1),
|
|
'desktop-entry': dbus.String('', variant_level=1),
|
|
'x-kde-app-icon': dbus.String('', variant_level=1)
|
|
}
|
|
notify_iface.Notify('f0ckm Uploader', dbus.UInt32(0), '', '', body_text, [], hints, -1)
|
|
return
|
|
except Exception as e:
|
|
print(f"DBus notification error: {e}")
|
|
|
|
# Fallback to notify-send
|
|
try:
|
|
cmd = [
|
|
"notify-send",
|
|
"-a", "f0ckm Uploader",
|
|
"-h", f"string:x-kde-urls:{file_url}",
|
|
"-h", f"string:urls:{file_url}",
|
|
"-h", "string:desktop-entry:",
|
|
"-h", "string:x-kde-app-icon:",
|
|
"",
|
|
body_text
|
|
]
|
|
subprocess.run(cmd, check=False)
|
|
except Exception as e:
|
|
print(f"Notification error: {e}")
|
|
|
|
def on_upload_file(self):
|
|
file_paths, _ = QFileDialog.getOpenFileNames(
|
|
None, "Select Files to Upload", "", "All Files (*)"
|
|
)
|
|
if file_paths:
|
|
self.enqueue_targets(file_paths)
|
|
|
|
def on_upload_url(self):
|
|
url, ok = QInputDialog.getText(None, "Upload URL", "Enter image or file URL:")
|
|
if ok and url.strip():
|
|
self.enqueue_targets([url.strip()])
|
|
|
|
def on_upload_clipboard(self):
|
|
clipboard = QApplication.clipboard()
|
|
mime_data = clipboard.mimeData()
|
|
|
|
if mime_data.hasImage():
|
|
image = clipboard.image()
|
|
temp_dir = tempfile.gettempdir()
|
|
ts = time.strftime('%Y%m%d_%H%M%S')
|
|
ms = int(time.time() * 1000) % 1000
|
|
temp_path = os.path.join(temp_dir, f"f0ckm_clipboard_{ts}_{ms:03d}.png")
|
|
if image.save(temp_path, "PNG"):
|
|
self.enqueue_targets([temp_path])
|
|
else:
|
|
self.show_message("Upload Error", "Failed to save clipboard image to temporary file.", QSystemTrayIcon.Critical)
|
|
elif mime_data.hasUrls():
|
|
urls = mime_data.urls()
|
|
valid_paths = []
|
|
for url in urls:
|
|
if url.isLocalFile():
|
|
path = url.toLocalFile()
|
|
if os.path.exists(path):
|
|
valid_paths.append(path)
|
|
if valid_paths:
|
|
self.enqueue_targets(valid_paths)
|
|
else:
|
|
self.show_message("Upload Error", "No valid local files in clipboard.", QSystemTrayIcon.Warning)
|
|
elif mime_data.hasText():
|
|
text = mime_data.text().strip()
|
|
lines = [l.strip() for l in text.split('\n') if l.strip()]
|
|
valid_targets = []
|
|
for line in lines:
|
|
if line.startswith("file://"):
|
|
line = line[7:]
|
|
t, _ = parse_upload_target(line)
|
|
if t:
|
|
valid_targets.append(t)
|
|
if valid_targets:
|
|
self.enqueue_targets(valid_targets)
|
|
else:
|
|
self.show_message("Upload Error", "Clipboard text does not contain valid file paths or URLs.", QSystemTrayIcon.Warning)
|
|
else:
|
|
self.show_message("Upload Error", "Clipboard does not contain an image, files, or URLs.", QSystemTrayIcon.Warning)
|
|
|
|
def show_message(self, title, message, icon=QSystemTrayIcon.Information):
|
|
self.tray.showMessage(title, message, icon, 5000)
|
|
|
|
# ==========================================
|
|
# Application Entry Point & IPC
|
|
# ==========================================
|
|
class SingleInstanceApp:
|
|
def __init__(self, name):
|
|
self.name = name
|
|
self.server = QLocalServer()
|
|
|
|
def start(self, callback_on_message):
|
|
if not self.server.listen(self.name):
|
|
socket = QLocalSocket()
|
|
socket.connectToServer(self.name)
|
|
if socket.waitForConnected(500):
|
|
action, targets = parse_cli_args(sys.argv[1:])
|
|
payload = json.dumps({"action": action, "targets": targets})
|
|
print(f"[f0ckm-gui] [IPC CLIENT] Daemon running. Forwarding payload via socket: '{payload}'", flush=True)
|
|
socket.write(payload.encode("utf-8"))
|
|
socket.waitForBytesWritten(500)
|
|
socket.disconnectFromServer()
|
|
return False
|
|
else:
|
|
QLocalServer.removeServer(self.name)
|
|
if not self.server.listen(self.name):
|
|
return False
|
|
|
|
print(f"[f0ckm-gui] [IPC SERVER] Started single-instance IPC listener socket server: '{self.name}'", flush=True)
|
|
self.server.newConnection.connect(lambda: self._handle_connection(callback_on_message))
|
|
return True
|
|
|
|
def _handle_connection(self, callback):
|
|
socket = self.server.nextPendingConnection()
|
|
if socket:
|
|
socket.readyRead.connect(lambda: self._read_message(socket, callback))
|
|
|
|
def _read_message(self, socket, callback):
|
|
data = socket.readAll().data().decode("utf-8").strip()
|
|
print(f"[f0ckm-gui] [IPC SERVER] Received message from socket: '{data}'", flush=True)
|
|
callback(data)
|
|
socket.disconnectFromServer()
|
|
|
|
def main():
|
|
app = QApplication(sys.argv)
|
|
app.setApplicationName("f0ckm Uploader")
|
|
app.setDesktopFileName("f0ckm-uploader-gui")
|
|
app.setQuitOnLastWindowClosed(False)
|
|
|
|
app_icon = get_app_icon()
|
|
if not app_icon.isNull():
|
|
app.setWindowIcon(app_icon)
|
|
|
|
instance_name = "f0ckm-uploader-gui-lock"
|
|
single_instance = SingleInstanceApp(instance_name)
|
|
|
|
tray_app = None
|
|
|
|
def on_activate(msg_str):
|
|
nonlocal tray_app
|
|
if tray_app:
|
|
if msg_str.startswith("{"):
|
|
try:
|
|
data = json.loads(msg_str)
|
|
act = data.get("action", "show")
|
|
targets = data.get("targets", [])
|
|
if act == "spectacle":
|
|
tray_app.on_capture_and_upload()
|
|
elif act == "upload" and targets:
|
|
tray_app.enqueue_targets(targets)
|
|
else:
|
|
tray_app.on_open_settings()
|
|
return
|
|
except Exception as e:
|
|
print(f"[f0ckm-gui] IPC JSON parse notice: {e}", flush=True)
|
|
|
|
if msg_str == "spectacle":
|
|
tray_app.on_capture_and_upload()
|
|
elif msg_str.startswith("upload:"):
|
|
target = msg_str[7:]
|
|
tray_app.enqueue_targets([target])
|
|
else:
|
|
tray_app.on_open_settings()
|
|
|
|
if not single_instance.start(on_activate):
|
|
sys.exit(0)
|
|
|
|
if not os.path.exists(CONFIG_PATH):
|
|
save_config(DEFAULT_CONFIG)
|
|
|
|
cfg = get_env_config()
|
|
set_autostart(cfg.get("autostart", False))
|
|
|
|
tray_app = SystemTrayApp()
|
|
local_http_bridge.request_upload.connect(lambda url: tray_app.enqueue_targets([url]))
|
|
tray_app.show()
|
|
start_local_http_server(tray_app)
|
|
|
|
# Check if started with action argument
|
|
if len(sys.argv) > 1:
|
|
action, targets = parse_cli_args(sys.argv[1:])
|
|
if action == "spectacle":
|
|
tray_app.on_capture_and_upload()
|
|
elif action == "upload" and targets:
|
|
tray_app.enqueue_targets(targets)
|
|
else:
|
|
if not cfg.get("api_url") or not cfg.get("api_key"):
|
|
print("[f0ckm-gui] First launch detected (missing API credentials). Opening settings dialog...", flush=True)
|
|
signal.signal(signal.SIGINT, signal.SIG_DFL)
|
|
sig_timer = QTimer()
|
|
sig_timer.timeout.connect(lambda: None)
|
|
sig_timer.start(500)
|
|
|
|
sys.exit(app.exec())
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|