flake/scripts/zipline

#! /usr/bin/env nix-shell
#! nix-shell -i python3 -p python312 python312Packages.tkinter python312Packages.requests libnotify

import argparse
import mimetypes
import os
import subprocess
from pathlib import Path
from shutil import which
from tkinter import Tk
from typing import Any

import requests  # type: ignore


def read_secret_file(secret: str) -> str:
    with open(f"/run/secrets/{secret}", "r") as f:
        return f.read().strip()


def does_desktop_entry_exist(desktop_entry: str) -> bool:
    if not desktop_entry:
        raise ValueError("Please provide the full filename of the desktop entry.")

    if not desktop_entry.endswith(".desktop"):
        desktop_entry += ".desktop"

    entry_paths = []

    # Check if qtpaths is available
    if which("qtpaths"):
        result = subprocess.run(
            ["qtpaths", "--paths", "ApplicationsLocation"],
            stdout=subprocess.PIPE,
            text=True,
        )
        entry_paths = result.stdout.strip().split(":")
    else:
        print("qtpaths is not installed, falling back to XDG_DATA_DIRS.")
        xdg_data_dirs = os.getenv("XDG_DATA_DIRS", "/usr/share:/usr/local/share").split(
            ":"
        )
        entry_paths = [os.path.join(path, "applications") for path in xdg_data_dirs]
        entry_paths.append(os.path.expanduser("~/.local/share/applications"))

    print(f"Checking the following paths for {desktop_entry}:\n{entry_paths}\n{'-'*20}")

    # Search for the desktop entry file
    for entry_path in entry_paths:
        entry_file = Path(entry_path) / f"{desktop_entry}"
        print(f"Checking for {entry_file}")
        if entry_file.is_file():
            print(f"{desktop_entry} found in {entry_path}")
            return True

    print(f"Desktop entry {desktop_entry} does not exist.")
    return False


def copy_to_clipboard(text: str) -> None:
    root = Tk()
    root.withdraw()
    root.clipboard_clear()
    root.clipboard_append(text)
    root.update()
    root.destroy()


def notify(
    application_name: str,
    title: str,
    message: str,
    urgency: str = "low",
    category: str | None = None,
    icon: Path | None = None,
    desktop_entry: str | None = None,
) -> None:
    args = ["notify-send" "-a", application_name, "-u", urgency]
    if category:
        args.append("-c")
        args.append(category)
    if icon:
        args.append("-i")
        args.append(str(icon))
    if desktop_entry:
        args.append("-h")
        args.append(f"string:desktop-entry:{desktop_entry}")
    args.append(title)
    args.append(message)
    subprocess.run(args)


def zipline(
    file_path: Path,
    instance_url: str,
    application_name: str = None,
    desktop_entry: str = None,
) -> Any:
    token = read_secret_file("zipline")
    if not token:
        print("Secret file at /run/secrets/zipline either does not exist or is empty.")
        raise FileNotFoundError(
            "Secret file at /run/secrets/zipline either does not exist or is empty."
        )

    if not os.path.isfile(file_path):
        print(f"File at {file_path} does not exist.")
        raise FileNotFoundError(f"File at {file_path} does not exist.")

    use_send_notify = False
    if application_name and desktop_entry:
        if not does_desktop_entry_exist(desktop_entry=desktop_entry):
            print("Desktop entry does not exist.")
            raise FileNotFoundError("Desktop entry does not exist.")

        if not which("notify-send"):
            print("notify-send is not installed.")
            raise FileNotFoundError("notify-send is not installed.")

        use_send_notify = True

    content_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"

    try:
        headers = {"authorization": token}
        files = {
            "file": (os.path.basename(file_path), open(file_path, "rb"), content_type)
        }
        response = requests.post(
            f"{instance_url.rstrip('/')}/api/upload", headers=headers, files=files
        )

        if response.status_code == 200:
            response_data = response.json()
            link = response_data.get("files", [None])[0]

            if link:
                copy_to_clipboard(text=link)

                if use_send_notify:
                    notify(
                        application_name=application_name,
                        title="Upload Successful",
                        message=f"Link copied to clipboard: {link}",
                        urgency="low",
                        category="transfer.complete",
                        icon=file_path,
                    )
            else:
                raise ValueError("Invalid response format.")
        else:
            error_message = response.text
            raise Exception(error_message)

    except Exception as e:
        if use_send_notify:
            notify(
                application_name=application_name,
                title="Upload Failed",
                message=f"An error occurred: {e}",
                urgency="critical",
                category="transfer.error",
                icon=file_path,
            )
        raise e


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        prog="zipline",
        description="Upload a file to a Zipline instance.",
        epilog="Example usage: zipline /path/to/file.txt",
    )
    parser.add_argument("file", help="The file to upload.")
    parser.add_argument(
        "--url",
        help="The URL of the Zipline instance. Defaults to 'https://csw.im'.",
        default="https://csw.im",
    )
    parser.add_argument(
        "--application-name",
        help="The name of the application that is uploading the file. Defaults to 'Zipline'.",
        default="Zipline",
    )
    parser.add_argument(
        "--desktop-entry",
        help="The desktop entry file for the application that is uploading the file. If this is provided, send-notify will be invoked to display a notification if the upload succeeds or fails.",
        default=None,
    )

    args = parser.parse_args()

    zipline(
        file_path=args.file,
        instance_url=args.url,
        application_name=args.application_name,
        desktop_entry=args.desktop_entry,
    )