aboutsummaryrefslogtreecommitdiff
path: root/backend.py
blob: 9ef65de2eff189f06833d32f73f3620d653b134c (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
from flask import Flask
from pathlib import Path
from dotenv import load_dotenv
import requests
import random
import subprocess
import threading
import os

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/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()

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

    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
    if music_process and music_process.poll() is None:
        music_process.terminate()

    music_process = subprocess.Popen([
        "mpv",
        "--no-video",
        "--audio-device=pipewire",
        "--",
        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():
    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():
    music_process = subprocess.Popen([
        "mpv",
        "--no-video",
        "--audio-device=pipewire",
        "--",
        FAVOURITE,
    ])

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

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