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
|
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()
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"}
if __name__ == "__main__":
app.run(host="0.0.0.0", port=6732)
|
