path: root/backend.py

from flask import Flask
from pathlib import Path
from dotenv import load_dotenv
import json
import socket
import requests
import random
import subprocess
import threading
import os

MPV_SOCKET = "/tmp/remote-automation-mpv.sock"

app = Flask(__name__)
print_lock = threading.Lock()

FAVOURITE = "/srv/shares/public/Media/Music/Oxlo - Money Mouth ｜ Hyperpop ｜ NCS - Copyright Free Music [BElFoAIsIOk].mp3"

music_process = None

load_dotenv("/srv/automation/remote-automation/.env")

@app.get("/automate/hack_this_club")
def hack_this_club():
    print("i’M gOnNa HaCk\nThIs ClUb!!!1!1!1!!11!111!1!11!1")
    print("this club. ia m going to hack ,is hack club.")
    print("yes, i am going To hack, this club Is called hack club. i see no problem with This,")
    print("i will hack this club on October 21st, 2043 at 5:36 am")
    print("i will hack into, The hcb and it has money .also i have idv database downloaded and ,i will use it to hack this club ")
    os.system('lp -d Canon_G3060 /opsec.txt')
    return "ok"

@app.get("/automate/agentic")
def agentic():
    print("That's so agentic.")
    os.system('lp -d Canon_G3060 /srv/shares/public/thats-so-agentic.png')
    return "ok"

@app.get("/automate/print_queue")
def print_queue():
    if not print_lock.acquire(blocking=False):
        print("print queue is already running")
        return "print queue busy", 400
    try:
        print("printing file(s) in /srv/shares/public/printing")
        path = Path("/srv/shares/public/printing")
        files = [f for f in path.iterdir() if f.is_file()]

        for file in files:
            subprocess.run(["lp", "-d", "Canon_G3060", str(file)], check=True)
            subprocess.run(["mv", str(file), "/srv/shares/public/printed"], check=True)

        print(f"printed {len(files)} file(s)")
        return f"printed {len(files)} file(s)"
    except Exception as e:
        return f"exception: {(str(e))}", 500
    finally:
        print_lock.release()

def mpv_command(command):
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
        sock.connect(MPV_SOCKET)
        sock.sendall((json.dumps({"command": command}) + "\n").encode())

@app.get("/automate/music")
def music():
    global music_process

    if music_process and music_process.poll() is None:
        mpv_command(["cycle", "pause"])
        return {"status": "toggled"}

    NAVIDROME_URL = os.environ["NAVIDROME_URL"]
    USERNAME = os.environ["NAVIDROME_USER"]
    PASSWORD = os.environ["NAVIDROME_PASSWORD"]
    PLAYLIST_ID = os.environ["NAVIDROME_PLAYLIST_ID"]

    params = {
        "u": USERNAME,
        "p": PASSWORD,
        "s": "remote-automation",
        "v": "1.16.1",
        "c": "remote-automation",
        "f": "json",
        "id": PLAYLIST_ID,
    }

    response = requests.get(
        f"{NAVIDROME_URL}/rest/getPlaylist.view",
        params=params,
    )
    response.raise_for_status()

    data = response.json()["subsonic-response"]

    if data["status"] != "ok":
        return {"status": "error", "message": data.get("error", {})}, 500

    songs = data["playlist"]["entry"]
    song = random.choice(songs)

    stream_url = f"{NAVIDROME_URL}/rest/stream.view"

    stream_params = {
        "u": USERNAME,
        "p": PASSWORD,
        "s": "remote-automation",
        "v": "1.16.1",
        "c": "remote-automation",
        "id": song["id"],
    }

    stream_request = requests.Request(
        "GET",
        stream_url,
        params=stream_params,
    ).prepare()

    # Stop previous song (in case it somehow survived earlier)
    if music_process and music_process.poll() is None:
        music_process.terminate()

    music_process = subprocess.Popen([
        "mpv",
        "--no-video",
        "--audio-device=pipewire",
        f"--input-ipc-server={MPV_SOCKET}",
        "--",
        stream_request.url,
    ])

    print(f"Playing: {song['title']} - {song.get('artist', 'Unknown Artist')}")

    return {
        "status": "playing",
        "title": song["title"],
        "artist": song.get("artist", "Unknown Artist"),
    }

@app.get("/automate/music/next")
def music_next():
    global music_process

    if music_process and music_process.poll() is None:
        music_process.terminate()
        music_process = None

    return music()

@app.get("/automate/music/stop")
def music_stop():
    global music_process

    if music_process and music_process.poll() is None:
        music_process.terminate()
        music_process = None
        return {"status": "stopped"}

    return {"status": "not_playing"}

@app.get("/automate/volup")
def volup():
    os.system("wpctl set-volume @DEFAULT_AUDIO_SINK@ 5%+")
    return "ok"

@app.get("/automate/voldown")
def voldown():
    os.system("wpctl set-volume @DEFAULT_AUDIO_SINK@ 5%-")
    return "ok"

@app.get("/automate/music/fav")
def fav_song():
    global music_process
    if music_process and music_process.poll() is None:
        music_process.terminate()

    music_process = subprocess.Popen([
        "mpv",
        "--no-video",
        "--audio-device=pipewire",
        f"--input-ipc-server={MPV_SOCKET}",
        "--",
        FAVOURITE,
    ])

    return {
        "status": "playing",
        "title": "Money Mouth",
        "artist": "Oxlo"
    }

@app.get("/automate/zola")
def zola():
    result = subprocess.run(
        [
            "sh", "-c",
            "cd /srv/www/_src-v3 && "
            "git pull && "
            "~/.cargo/bin/zola build -o /srv/www/_site-v3 -f"
        ],
        capture_output=True,
        text=True,
    )

    if result.returncode != 0:
        print(result.stdout)
        print(result.stderr)
        return "deployment failed", 500

    subprocess.Popen([
        "mpv",
        "--no-video",
        "--audio-device=pipewire",
        "/srv/shares/public/Media/done.mp3",
    ])

    return "ok"

@app.get("/automate/cgit")
def cgit():
    result = subprocess.run(
        [
            "sh", "-c",
            "cd /srv/docker/cgit && "
            "docker compose down && "
            "docker compose up -d"
        ],
        capture_output=True,
        text=True,
    )

    if result.returncode != 0:
        print(result.stdout)
        print(result.stderr)
        return "cgit restart failed", 500

    subprocess.Popen([
        "mpv",
        "--no-video",
        "--audio-device=pipewire",
        "/srv/shares/public/Media/done.mp3",
    ])

    return "ok"

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=6732)