21273905644

Committed 23 Jan 2026 03:51AM UTC coverage: 82.511% (-0.7%) from 83.2%

Build # 21273905644

Build Type

push

github

Committed by

x4dr

Commit Message

install script

Run Details

4 of 63 new or added lines in 1 file covered. (6.35%)

8 existing lines in 1 file now uncovered.

2708 of 3282 relevant lines covered (82.51%)

0.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

9.47

/scripts/install.py

#!/usr/bin/env python3
import argparse
import getpass
import json
import os
import re
import secrets
import subprocess
from pathlib import Path


def get_input(prompt, default=None):
    if default:
        res = input(f"{prompt} [{default}]: ").strip()
        return res if res else default
    return input(f"{prompt}: ").strip()


def get_current_user():
    try:
        return os.getlogin()
    except OSError:
        return getpass.getuser()


def detect_domains():
    domains = []
    sites_enabled = Path("/etc/nginx/sites-enabled")
    if not sites_enabled.exists():
        return domains

    for site in sites_enabled.iterdir():
        try:
            content = site.read_text()
            matches = re.findall(r"server_name\s+([^;]+);", content)
            for match in matches:
                for domain in match.split():
                    if domain and not domain.startswith("_"):
                        domains.append(domain)
        except Exception:
            continue
    return sorted(list(set(domains)))


def get_repo_nwo():
    try:
        res = subprocess.run(
            ["gh", "repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"],
            capture_output=True,
            text=True,
            check=True,
        )
        return res.stdout.strip()
    except subprocess.CalledProcessError:
        return None


def register_webhook(domain, secret):
    print("\n--- GitHub Webhook Registration ---")
    use_gh = get_input("Use 'gh' CLI to register/update webhook?", "y")
    if use_gh.lower() != "y":
        return False

    if subprocess.run(["which", "gh"], capture_output=True).returncode != 0:
        print("Error: 'gh' CLI not found.")
        return False

    # Check GH auth status
    auth_check = subprocess.run(["gh", "auth", "status"], capture_output=True)
    if auth_check.returncode != 0:
        print("GitHub CLI is not authenticated.")
        if get_input("Run 'gh auth login' now?", "y").lower() == "y":
            try:
                subprocess.run(["gh", "auth", "login"], check=True)
            except subprocess.CalledProcessError:
                return False
        else:
            return False

    nwo = get_repo_nwo()
    if not nwo:
        print("Could not determine repository name.")
        return False

    webhook_url = f"https://{domain}/webhook"

    # Check for existing webhook
    try:
        hooks_json = subprocess.run(
            ["gh", "api", f"repos/{nwo}/hooks"],
            capture_output=True,
            text=True,
            check=True,
        ).stdout
        hooks = json.loads(hooks_json)
        existing_hook = next(
            (h for h in hooks if h.get("config", {}).get("url") == webhook_url), None
        )

        if existing_hook:
            print(f"Found existing webhook (ID: {existing_hook['id']}). Updating...")
            cmd = [
                "gh",
                "api",
                f"repos/{nwo}/hooks/{existing_hook['id']}",
                "--method",
                "PATCH",
                "-f",
                "active=true",
                "-f",
                "events[]=workflow_run",
                "-f",
                f"config[url]={webhook_url}",
                "-f",
                "config[content_type]=json",
                "-f",
                f"config[secret]={secret}",
            ]
        else:
            print("Creating new webhook...")
            cmd = [
                "gh",
                "api",
                f"repos/{nwo}/hooks",
                "--method",
                "POST",
                "-f",
                "name=web",
                "-f",
                "active=true",
                "-f",
                "events[]=workflow_run",
                "-f",
                f"config[url]={webhook_url}",
                "-f",
                "config[content_type]=json",
                "-f",
                f"config[secret]={secret}",
            ]

        subprocess.run(cmd, check=True)
        print(f"Successfully registered/updated webhook: {webhook_url}")
        return True
    except Exception as e:
        print(f"Failed to manage webhook: {e}")
        return False


def uninstall(okysa_root):
    print("\n--- Uninstalling Okysa Deployment ---")

    # Stop and disable services
    services = ["okysa.service", "okysa-webhook.service"]
    for svc in services:
        print(f"Stopping and disabling {svc}...")
        subprocess.run(["sudo", "systemctl", "stop", svc], stderr=subprocess.DEVNULL)
        subprocess.run(["sudo", "systemctl", "disable", svc], stderr=subprocess.DEVNULL)
        subprocess.run(
            ["sudo", "rm", f"/etc/systemd/system/{svc}"], stderr=subprocess.DEVNULL
        )

    # Nginx
    sites_available = Path("/etc/nginx/sites-available")
    if sites_available.exists():
        for site in sites_available.iterdir():
            try:
                content = site.read_text()
                if str(okysa_root) in content:
                    print(f"Removing Nginx config: {site.name}")
                    subprocess.run(
                        ["sudo", "rm", f"/etc/nginx/sites-enabled/{site.name}"],
                        stderr=subprocess.DEVNULL,
                    )
                    subprocess.run(["sudo", "rm", str(site)], stderr=subprocess.DEVNULL)
            except:
                continue

    # Sudoers
    print("Removing sudoers entry...")
    subprocess.run(
        ["sudo", "rm", "/etc/sudoers.d/okysa-deploy"], stderr=subprocess.DEVNULL
    )

    subprocess.run(["sudo", "systemctl", "daemon-reload"])
    print("\nUninstallation complete.")


def main():
    parser = argparse.ArgumentParser(description="Okysa Deployment Installer")
    parser.add_argument(
        "--uninstall", action="store_true", help="Uninstall the deployment"
    )
    args = parser.parse_args()

    script_path = Path(__file__).resolve()
    okysa_root = script_path.parent.parent

    if args.uninstall:
        uninstall(okysa_root)
        return

    gamepack_root = okysa_root.parent / "GamePack"
    uv_path = subprocess.getoutput("which uv") or "uv"

    print("--- Path Detection ---")
    print(f"Okysa Root:    {okysa_root}")
    if gamepack_root.exists():
        print(f"GamePack Root: {gamepack_root}")
    else:
        print(f"WARNING: GamePack not found at {gamepack_root}")
        gamepack_root = Path(
            get_input("Enter absolute path to GamePack", str(gamepack_root))
        )

    # 2. Gather Configuration
    print("\n--- Configuration ---")
    user = get_input("System user to run the bot", get_current_user())

    detected_domains = detect_domains()
    if detected_domains:
        print("Detected domains from Nginx:")
        for i, d in enumerate(detected_domains):
            print(f"  {i + 1}. {d}")
        choice = get_input(
            f"Choose domain (1-{len(detected_domains)}) or enter new", "1"
        )
        if choice.isdigit() and 1 <= int(choice) <= len(detected_domains):
            domain = detected_domains[int(choice) - 1]
        else:
            domain = choice
    else:
        domain = get_input("Domain for Nginx", "nossinet.cc")

    webhook_secret = secrets.token_urlsafe(32)

    # 3. Update scripts
    deploy_sh_path = okysa_root / "scripts" / "deploy.sh"
    webhook_py_path = okysa_root / "scripts" / "webhook_listener.py"

    deploy_content = f"""#!/bin/bash
set -e
PROJECT_ROOT="{okysa_root}"
GAMEPACK_ROOT="{gamepack_root}"

echo "[$(date)] Starting redeployment..."
cd "$GAMEPACK_ROOT" && git pull
cd "$PROJECT_ROOT" && git pull
{uv_path} sync --all-extras
sudo systemctl restart okysa.service
echo "[$(date)] Deployment successful!"
"""
    with open(deploy_sh_path, "w") as f:
        f.write(deploy_content)
    os.chmod(deploy_sh_path, 0o755)

    with open(webhook_py_path, "r") as f:
        lines = f.readlines()
    with open(webhook_py_path, "w") as f:
        for line in lines:
            if line.startswith("DEPLOY_SCRIPT ="):
                f.write(f'DEPLOY_SCRIPT = "{deploy_sh_path}"\n')
            else:
                f.write(line)

    # 4. Generate Systemd Units
    bot_service = f"""[Unit]
Description=Okysa Discord Bot
After=network.target

[Service]
Type=simple
User={user}
WorkingDirectory={okysa_root}
ExecStart={uv_path} run Okysa.py
Restart=always

[Install]
WantedBy=multi-user.target
"""

    webhook_service = f"""[Unit]
Description=Okysa Webhook Listener
After=network.target

[Service]
Type=simple
User={user}
WorkingDirectory={okysa_root}
Environment="GITHUB_WEBHOOK_SECRET={webhook_secret}"
ExecStart=/usr/bin/python3 {webhook_py_path}
Restart=always

[Install]
WantedBy=multi-user.target
"""

    # 5. Generate Nginx Config
    nginx_conf = f"""server {{
    listen 80;
    server_name {domain};

    location /webhook {{
        proxy_pass http://localhost:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }}
}}
"""

    print("\n--- Proposed Actions ---")
    print("1. Write /etc/systemd/system/okysa.service")
    print("2. Write /etc/systemd/system/okysa-webhook.service")
    print(f"3. Write /etc/nginx/sites-available/{domain}")
    print("4. Add sudoers entry for systemctl restart")

    if get_input("Perform these actions? (requires sudo)", "n").lower() == "y":

        def sudo_write(content, path):
            subprocess.run(
                ["sudo", "tee", str(path)],
                input=content.encode(),
                stdout=subprocess.DEVNULL,
            )

        sudo_write(bot_service, "/etc/systemd/system/okysa.service")
        sudo_write(webhook_service, "/etc/systemd/system/okysa-webhook.service")
        sudo_write(nginx_conf, f"/etc/nginx/sites-available/{domain}")

        sudoers_line = (
            f"{user} ALL=(ALL) NOPASSWD: /usr/bin/systemctl restart okysa.service\n"
        )
        sudo_write(sudoers_line, "/etc/sudoers.d/okysa-deploy")

        if (
            get_input(f"Link /etc/nginx/sites-enabled/{domain}? (y/n)", "y").lower()
            == "y"
        ):
            subprocess.run(
                [
                    "sudo",
                    "ln",
                    "-sf",
                    f"/etc/nginx/sites-available/{domain}",
                    f"/etc/nginx/sites-enabled/{domain}",
                ]
            )

        subprocess.run(["sudo", "systemctl", "daemon-reload"])
        if get_input("Enable and start services now? (y/n)", "y").lower() == "y":
            subprocess.run(
                ["sudo", "systemctl", "enable", "--now", "okysa-webhook", "okysa"]
            )

        register_webhook(domain, webhook_secret)
        print(f"\nInstallation complete. Secret: {webhook_secret}")
    else:
        print("\nInstallation aborted.")


if __name__ == "__main__":
    main()

1	#!/usr/bin/env python3
2	import argparse	1✔
3	import getpass	1✔
4	import json	1✔
5	import os	1✔
6	import re	1✔
7	import secrets	1✔
8	import subprocess	1✔
9	from pathlib import Path	1✔
10
11
12	def get_input(prompt, default=None):	1✔
13	if default:	×
14	res = input(f"{prompt} [{default}]: ").strip()	×
15	return res if res else default	×
16	return input(f"{prompt}: ").strip()	×
17
18
19	def get_current_user():	1✔
20	try:	×
21	return os.getlogin()	×
22	except OSError:	×
23	return getpass.getuser()	×
24
25
26	def detect_domains():	1✔
27	domains = []	×
28	sites_enabled = Path("/etc/nginx/sites-enabled")	×
29	if not sites_enabled.exists():	×
30	return domains	×
31
32	for site in sites_enabled.iterdir():	×
33	try:	×
34	content = site.read_text()	×
35	matches = re.findall(r"server_name\s+([^;]+);", content)	×
36	for match in matches:	×
37	for domain in match.split():	×
38	if domain and not domain.startswith("_"):	×
39	domains.append(domain)	×
40	except Exception:	×
41	continue	×
42	return sorted(list(set(domains)))	×
43
44
45	def get_repo_nwo():	1✔
NEW 46	try:	×
NEW 47	res = subprocess.run(	×
48	["gh", "repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"],
49	capture_output=True,
50	text=True,
51	check=True,
52	)
NEW 53	return res.stdout.strip()	×
NEW 54	except subprocess.CalledProcessError:	×
NEW 55	return None	×
56
57
58	def register_webhook(domain, secret):	1✔
59	print("\n--- GitHub Webhook Registration ---")	×
NEW 60	use_gh = get_input("Use 'gh' CLI to register/update webhook?", "y")	×
61	if use_gh.lower() != "y":	×
62	return False	×
63
UNCOV 64	if subprocess.run(["which", "gh"], capture_output=True).returncode != 0:	×
NEW 65	print("Error: 'gh' CLI not found.")	×
66	return False	×
67
68	# Check GH auth status
69	auth_check = subprocess.run(["gh", "auth", "status"], capture_output=True)	×
70	if auth_check.returncode != 0:	×
71	print("GitHub CLI is not authenticated.")	×
NEW 72	if get_input("Run 'gh auth login' now?", "y").lower() == "y":	×
73	try:	×
UNCOV 74	subprocess.run(["gh", "auth", "login"], check=True)	×
75	except subprocess.CalledProcessError:	×
76	return False	×
77	else:
78	return False	×
79
NEW 80	nwo = get_repo_nwo()	×
NEW 81	if not nwo:	×
NEW 82	print("Could not determine repository name.")	×
NEW 83	return False	×
84
85	webhook_url = f"https://{domain}/webhook"	×
86
87	# Check for existing webhook
88	try:	×
NEW 89	hooks_json = subprocess.run(	×
90	["gh", "api", f"repos/{nwo}/hooks"],
91	capture_output=True,
92	text=True,
93	check=True,
94	).stdout
NEW 95	hooks = json.loads(hooks_json)	×
NEW 96	existing_hook = next(	×
97	(h for h in hooks if h.get("config", {}).get("url") == webhook_url), None
98	)
99
NEW 100	if existing_hook:	×
NEW 101	print(f"Found existing webhook (ID: {existing_hook['id']}). Updating...")	×
NEW 102	cmd = [	×
103	"gh",
104	"api",
105	f"repos/{nwo}/hooks/{existing_hook['id']}",
106	"--method",
107	"PATCH",
108	"-f",
109	"active=true",
110	"-f",
111	"events[]=workflow_run",
112	"-f",
113	f"config[url]={webhook_url}",
114	"-f",
115	"config[content_type]=json",
116	"-f",
117	f"config[secret]={secret}",
118	]
119	else:
NEW 120	print("Creating new webhook...")	×
NEW 121	cmd = [	×
122	"gh",
123	"api",
124	f"repos/{nwo}/hooks",
125	"--method",
126	"POST",
127	"-f",
128	"name=web",
129	"-f",
130	"active=true",
131	"-f",
132	"events[]=workflow_run",
133	"-f",
134	f"config[url]={webhook_url}",
135	"-f",
136	"config[content_type]=json",
137	"-f",
138	f"config[secret]={secret}",
139	]
140
NEW 141	subprocess.run(cmd, check=True)	×
NEW 142	print(f"Successfully registered/updated webhook: {webhook_url}")	×
143	return True	×
NEW 144	except Exception as e:	×
NEW 145	print(f"Failed to manage webhook: {e}")	×
UNCOV 146	return False	×
147
148
149	def uninstall(okysa_root):	1✔
NEW 150	print("\n--- Uninstalling Okysa Deployment ---")	×
151
152	# Stop and disable services
NEW 153	services = ["okysa.service", "okysa-webhook.service"]	×
NEW 154	for svc in services:	×
NEW 155	print(f"Stopping and disabling {svc}...")	×
NEW 156	subprocess.run(["sudo", "systemctl", "stop", svc], stderr=subprocess.DEVNULL)	×
NEW 157	subprocess.run(["sudo", "systemctl", "disable", svc], stderr=subprocess.DEVNULL)	×
NEW 158	subprocess.run(	×
159	["sudo", "rm", f"/etc/systemd/system/{svc}"], stderr=subprocess.DEVNULL
160	)
161
162	# Nginx
NEW 163	sites_available = Path("/etc/nginx/sites-available")	×
NEW 164	if sites_available.exists():	×
NEW 165	for site in sites_available.iterdir():	×
NEW 166	try:	×
NEW 167	content = site.read_text()	×
NEW 168	if str(okysa_root) in content:	×
NEW 169	print(f"Removing Nginx config: {site.name}")	×
NEW 170	subprocess.run(	×
171	["sudo", "rm", f"/etc/nginx/sites-enabled/{site.name}"],
172	stderr=subprocess.DEVNULL,
173	)
NEW 174	subprocess.run(["sudo", "rm", str(site)], stderr=subprocess.DEVNULL)	×
NEW 175	except:	×
NEW 176	continue	×
177
178	# Sudoers
NEW 179	print("Removing sudoers entry...")	×
NEW 180	subprocess.run(	×
181	["sudo", "rm", "/etc/sudoers.d/okysa-deploy"], stderr=subprocess.DEVNULL
182	)
183
NEW 184	subprocess.run(["sudo", "systemctl", "daemon-reload"])	×
NEW 185	print("\nUninstallation complete.")	×
186
187
188	def main():	1✔
NEW 189	parser = argparse.ArgumentParser(description="Okysa Deployment Installer")	×
NEW 190	parser.add_argument(	×
191	"--uninstall", action="store_true", help="Uninstall the deployment"
192	)
NEW 193	args = parser.parse_args()	×
194
195	script_path = Path(__file__).resolve()	×
196	okysa_root = script_path.parent.parent	×
197
NEW 198	if args.uninstall:	×
NEW 199	uninstall(okysa_root)	×
NEW 200	return	×
201
202	gamepack_root = okysa_root.parent / "GamePack"	×
203	uv_path = subprocess.getoutput("which uv") or "uv"	×
204
205	print("--- Path Detection ---")	×
206	print(f"Okysa Root: {okysa_root}")	×
207	if gamepack_root.exists():	×
208	print(f"GamePack Root: {gamepack_root}")	×
209	else:
210	print(f"WARNING: GamePack not found at {gamepack_root}")	×
NEW 211	gamepack_root = Path(	×
212	get_input("Enter absolute path to GamePack", str(gamepack_root))
213	)
214
215	# 2. Gather Configuration
216	print("\n--- Configuration ---")	×
217	user = get_input("System user to run the bot", get_current_user())	×
218
219	detected_domains = detect_domains()	×
220	if detected_domains:	×
221	print("Detected domains from Nginx:")	×
222	for i, d in enumerate(detected_domains):	×
223	print(f" {i + 1}. {d}")	×
224	choice = get_input(	×
225	f"Choose domain (1-{len(detected_domains)}) or enter new", "1"
226	)
227	if choice.isdigit() and 1 <= int(choice) <= len(detected_domains):	×
228	domain = detected_domains[int(choice) - 1]	×
229	else:
230	domain = choice	×
231	else:
NEW 232	domain = get_input("Domain for Nginx", "nossinet.cc")	×
233
234	webhook_secret = secrets.token_urlsafe(32)	×
235
236	# 3. Update scripts
237	deploy_sh_path = okysa_root / "scripts" / "deploy.sh"	×
238	webhook_py_path = okysa_root / "scripts" / "webhook_listener.py"	×
239
240	deploy_content = f"""#!/bin/bash	×
241	set -e
242	PROJECT_ROOT="{okysa_root}"
243	GAMEPACK_ROOT="{gamepack_root}"
244
245	echo "[$(date)] Starting redeployment..."
246	cd "$GAMEPACK_ROOT" && git pull
247	cd "$PROJECT_ROOT" && git pull
248	{uv_path} sync --all-extras
249	sudo systemctl restart okysa.service
250	echo "[$(date)] Deployment successful!"
251	"""
UNCOV 252	with open(deploy_sh_path, "w") as f:	×
253	f.write(deploy_content)	×
254	os.chmod(deploy_sh_path, 0o755)	×
255
256	with open(webhook_py_path, "r") as f:	×
257	lines = f.readlines()	×
UNCOV 258	with open(webhook_py_path, "w") as f:	×
259	for line in lines:	×
260	if line.startswith("DEPLOY_SCRIPT ="):	×
261	f.write(f'DEPLOY_SCRIPT = "{deploy_sh_path}"\n')	×
262	else:
263	f.write(line)	×
264
265	# 4. Generate Systemd Units
266	bot_service = f"""[Unit]	×
267	Description=Okysa Discord Bot
268	After=network.target
269
270	[Service]
271	Type=simple
272	User={user}
273	WorkingDirectory={okysa_root}
274	ExecStart={uv_path} run Okysa.py
275	Restart=always
276
277	[Install]
278	WantedBy=multi-user.target
279	"""
280
281	webhook_service = f"""[Unit]	×
282	Description=Okysa Webhook Listener
283	After=network.target
284
285	[Service]
286	Type=simple
287	User={user}
288	WorkingDirectory={okysa_root}
289	Environment="GITHUB_WEBHOOK_SECRET={webhook_secret}"
290	ExecStart=/usr/bin/python3 {webhook_py_path}
291	Restart=always
292
293	[Install]
294	WantedBy=multi-user.target
295	"""
296
297	# 5. Generate Nginx Config
298	nginx_conf = f"""server {{	×
299	listen 80;
300	server_name {domain};
301
302	location /webhook {{
303	proxy_pass http://localhost:5000;
304	proxy_set_header Host $host;
305	proxy_set_header X-Real-IP $remote_addr;
306	proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
307	proxy_set_header X-Forwarded-Proto $scheme;
308	}}
309	}}
310	"""
311
312	print("\n--- Proposed Actions ---")	×
313	print("1. Write /etc/systemd/system/okysa.service")	×
314	print("2. Write /etc/systemd/system/okysa-webhook.service")	×
315	print(f"3. Write /etc/nginx/sites-available/{domain}")	×
316	print("4. Add sudoers entry for systemctl restart")	×
317
NEW 318	if get_input("Perform these actions? (requires sudo)", "n").lower() == "y":	×
319
320	def sudo_write(content, path):	×
321	subprocess.run(	×
322	["sudo", "tee", str(path)],
323	input=content.encode(),
324	stdout=subprocess.DEVNULL,
325	)
326
327	sudo_write(bot_service, "/etc/systemd/system/okysa.service")	×
328	sudo_write(webhook_service, "/etc/systemd/system/okysa-webhook.service")	×
329	sudo_write(nginx_conf, f"/etc/nginx/sites-available/{domain}")	×
330
331	sudoers_line = (	×
332	f"{user} ALL=(ALL) NOPASSWD: /usr/bin/systemctl restart okysa.service\n"
333	)
334	sudo_write(sudoers_line, "/etc/sudoers.d/okysa-deploy")	×
335
NEW 336	if (	×
337	get_input(f"Link /etc/nginx/sites-enabled/{domain}? (y/n)", "y").lower()
338	== "y"
339	):
UNCOV 340	subprocess.run(	×
341	[
342	"sudo",
343	"ln",
344	"-sf",
345	f"/etc/nginx/sites-available/{domain}",
346	f"/etc/nginx/sites-enabled/{domain}",
347	]
348	)
349
UNCOV 350	subprocess.run(["sudo", "systemctl", "daemon-reload"])	×
NEW 351	if get_input("Enable and start services now? (y/n)", "y").lower() == "y":	×
352	subprocess.run(	×
353	["sudo", "systemctl", "enable", "--now", "okysa-webhook", "okysa"]
354	)
355
UNCOV 356	register_webhook(domain, webhook_secret)	×
NEW 357	print(f"\nInstallation complete. Secret: {webhook_secret}")	×
358	else:
NEW 359	print("\nInstallation aborted.")	×
360
361
362	if __name__ == "__main__":	1✔
363	main()	×

x4dr / Okysa / 21273905644

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous