#!/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
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)
from PySide6.QtCore import Qt, QThread, QObject, Signal, QUrl, QFile, QIODevice
from PySide6.QtGui import QIcon, QAction, QPixmap, QPainter, QColor, QFont, QPen, QImage
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
)
# Configuration paths
CONFIG_DIR = os.path.expanduser("~/.config/f0ckm-uploader")
CONFIG_PATH = os.path.join(CONFIG_DIR, "config.json")
DEFAULT_CONFIG = {
"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"
}
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)
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
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 = val if val.endswith("/api/v2/upload") else f"{val.rstrip('/')}/api/v2/upload"
config["api_url"] = url
elif key == "API_KEY" and 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
return config
def save_config(config):
try:
os.makedirs(CONFIG_DIR, exist_ok=True)
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, "kde-uploader-gui.desktop")
if enabled:
try:
os.makedirs(autostart_dir, exist_ok=True)
# Use the installed path ~/.local/bin/kde-uploader-gui if possible
exec_path = os.path.expanduser("~/.local/bin/kde-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=cloud-upload-symbolic
Terminal=false
Categories=Utility;Network;
X-GNOME-Autostart-enabled=true
"""
with open(autostart_path, "w") as f:
f.write(content)
os.chmod(autostart_path, 0o755)
except Exception as e:
print(f"Failed to create autostart entry: {e}")
else:
if os.path.exists(autostart_path):
try:
os.remove(autostart_path)
except Exception as e:
print(f"Failed to remove autostart entry: {e}")
# ==========================================
# Connection Tester Worker
# ==========================================
class ConnectionTester(QObject):
finished = Signal(bool, str) # Success, Message
def __init__(self, url, api_key):
super().__init__()
self.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(540, 680)
app_icon = get_app_icon()
if not app_icon.isNull():
self.setWindowIcon(app_icon)
self.tester_thread = None
self.tester = None
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 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 Settings
grp_api = QGroupBox("API Connection")
api_layout = QFormLayout(grp_api)
api_layout.setContentsMargins(15, 20, 15, 15)
api_layout.setSpacing(10)
self.txt_url = QLineEdit()
self.txt_url.setPlaceholderText("https://example.com/api/v2/upload")
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")
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)
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 load_current_settings(self):
config = load_config()
self.txt_url.setText(config.get("api_url", ""))
self.txt_key.setText(config.get("api_key", ""))
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)
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...")
url = self.txt_url.text().strip()
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):
config = {
"api_url": self.txt_url.text().strip(),
"api_key": self.txt_key.text().strip(),
"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",
"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 and hasattr(self.tray_app, "load_tray_icon"):
self.tray_app.load_tray_icon()
self.accept()
else:
QMessageBox.critical(self, "Error", "Failed to save configuration file.")
# ==========================================
# 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 CurlUploadThread(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.process = None
self.file_size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
self.aborted = False
def abort(self):
self.aborted = True
if self.process:
try:
self.process.kill()
except Exception:
pass
def run(self):
mime_type, _ = mimetypes.guess_type(self.file_path)
if not mime_type:
ext = os.path.splitext(self.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_param = f"file=@{self.file_path}"
if mime_type:
file_param += f";type={mime_type}"
cmd = [
"curl", "-#",
"-F", file_param
]
if self.api_key:
cmd.extend(["-H", f"X-Api-Key: {self.api_key}"])
if self.rating:
cmd.extend(["-F", f"rating={self.rating}"])
if self.tags:
cmd.extend(["-F", f"tags={self.tags}"])
if self.visibility != "":
cmd.extend(["-F", f"visibility={self.visibility}"])
if self.is_oc != "0":
cmd.extend(["-F", f"is_oc={self.is_oc}"])
cmd.append(self.api_url)
try:
self.process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
buffer = ""
while True:
char = self.process.stderr.read(1)
if not char and self.process.poll() is not None:
break
if char in ("\r", "\n"):
matches = re.findall(r'(\d+(?:\.\d+)?)\s*%', buffer)
if matches:
try:
pct = float(matches[-1])
bytes_sent = int(self.file_size * (pct / 100.0))
self.progress.emit(bytes_sent, self.file_size)
except Exception:
pass
buffer = ""
else:
buffer += char
stdout_data, stderr_data = self.process.communicate()
exit_code = self.process.returncode
if self.aborted:
self.finished.emit(False, "ABORTED")
elif exit_code == 0:
self.finished.emit(True, stdout_data)
else:
self.finished.emit(False, stderr_data or f"curl exited with code {exit_code}")
except Exception as e:
if self.aborted:
self.finished.emit(False, "ABORTED")
else:
self.finished.emit(False, str(e))
# ==========================================
# 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
self.menu = QMenu()
# Header entry
title_action = QAction("f0ckm Uploader", self.menu)
title_action.setEnabled(False)
self.menu.addAction(title_action)
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)
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(self.on_open_settings)
self.menu.addAction(settings_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.settings_dialog = None
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):
if not self.settings_dialog:
self.settings_dialog = SettingsDialog(tray_app=self)
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)
save_path = os.path.join(screenshot_dir, f"screenshot_{time.strftime('%Y%m%d_%H%M%S')}.png")
try:
res = subprocess.run(["spectacle", "-r", "-b", "-n", "-o", save_path])
if os.path.exists(save_path) and os.path.getsize(save_path) > 0:
self.upload_file_direct(save_path)
except Exception as e:
self.show_message("Upload Error", f"Failed to capture screenshot: {e}", QSystemTrayIcon.Critical)
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)
return
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", "")
if tags:
# Format tags as clean comma-separated string ("tag1, tag2") for f0ckm API tagsRaw.split(',') compatibility
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))
self.tray.setToolTip(f"Uploading {filename}...\n0% (0 B)")
self.upload_thread = CurlUploadThread(api_url, api_key, file_path, rating, tags, visibility, is_oc)
self.upload_thread.progress.connect(self.on_upload_progress)
self.upload_thread.finished.connect(lambda ok, resp: self.on_upload_finished(ok, resp, file_path))
self.upload_thread.start()
self.abort_action.setEnabled(True)
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")
if pct >= 99:
tooltip = f"Uploading: {filename}\n99% ({total_str})\nProcessing upload on server..."
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: {filename}\n{pct}% ({sent_str} / {total_str})\nSpeed: {speed_str} | ETA: {eta_str}"
self.tray.setToolTip(tooltip)
def on_abort_upload(self):
if hasattr(self, "upload_thread") and self.upload_thread:
self.upload_thread.abort()
def on_upload_finished(self, success, response_text, file_path):
self.abort_action.setEnabled(False)
self.tray.setIcon(self.default_icon)
self.tray.setToolTip("f0ckm Uploader (Click to capture & upload)")
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)
return
try:
res = json.loads(response_text)
if res.get("success") and res.get("url"):
item_url = res.get("url")
QApplication.clipboard().setText(item_url)
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)
def show_success_notification(self, file_path, 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
thumb_path = generate_thumbnail(file_path)
if thumb_path and os.path.exists(thumb_path):
body_text = f'
{item_url}'
else:
body_text = f'Upload Successful: {item_url}'
try:
subprocess.run(["notify-send", "-a", "f0ckm", "Upload Successful", body_text], check=False)
except Exception as e:
print(f"Notification error: {e}")
def on_upload_file(self):
file_path, _ = QFileDialog.getOpenFileName(
None, "Select File to Upload", "", "All Files (*)"
)
if file_path:
self.upload_file_direct(file_path)
def on_upload_clipboard(self):
clipboard = QApplication.clipboard()
mime_data = clipboard.mimeData()
if mime_data.hasImage():
image = clipboard.image()
temp_dir = tempfile.gettempdir()
temp_path = os.path.join(temp_dir, f"f0ckm_clipboard_{int(time.time())}.png")
if image.save(temp_path, "PNG"):
self.upload_file_direct(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()
uploaded_any = False
for url in urls:
if url.isLocalFile():
path = url.toLocalFile()
if os.path.exists(path):
self.upload_file_direct(path)
uploaded_any = True
if not uploaded_any:
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()]
uploaded_any = False
for line in lines:
if line.startswith("file://"):
line = line[7:]
if os.path.exists(line):
self.upload_file_direct(line)
uploaded_any = True
if not uploaded_any:
self.show_message("Upload Error", "Clipboard text is not a valid local file path.", QSystemTrayIcon.Warning)
else:
self.show_message("Upload Error", "Clipboard does not contain an image or file paths.", 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):
msg = "show"
if len(sys.argv) > 1:
arg = sys.argv[1]
if arg in ("--spectacle", "-s"):
msg = "spectacle"
elif arg.startswith("--upload="):
msg = f"upload:{arg[9:]}"
elif arg.startswith("--upload"):
if len(sys.argv) > 2:
msg = f"upload:{sys.argv[2]}"
else:
norm = normalize_file_path(arg)
if os.path.exists(norm):
msg = f"upload:{norm}"
socket.write(msg.encode("utf-8"))
socket.waitForBytesWritten(500)
socket.disconnectFromServer()
return False
else:
QLocalServer.removeServer(self.name)
if not self.server.listen(self.name):
return False
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()
callback(data)
socket.disconnectFromServer()
def main():
app = QApplication(sys.argv)
app.setApplicationName("f0ckm Uploader")
app.setDesktopFileName("kde-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 == "spectacle":
tray_app.on_capture_and_upload()
elif msg_str.startswith("upload:"):
file_path = normalize_file_path(msg_str[7:])
if os.path.exists(file_path):
tray_app.upload_file_direct(file_path)
else:
tray_app.show_message("Upload Error", f"File not found: {file_path}", QSystemTrayIcon.Critical)
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)
tray_app = SystemTrayApp()
tray_app.show()
# Check if started with action argument
if len(sys.argv) > 1:
arg = sys.argv[1]
if arg in ("--spectacle", "-s"):
tray_app.on_capture_and_upload()
elif arg.startswith("--upload="):
file_path = normalize_file_path(arg[9:])
tray_app.upload_file_direct(file_path)
elif arg.startswith("--upload"):
if len(sys.argv) > 2:
file_path = normalize_file_path(sys.argv[2])
tray_app.upload_file_direct(file_path)
else:
norm = normalize_file_path(arg)
if os.path.exists(norm):
tray_app.upload_file_direct(norm)
sys.exit(app.exec())
if __name__ == "__main__":
main()