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
|
"""
Sopel plugin: bot controller
Owner-only commands to start, stop, and restart IRC bots directly as
processes. Alfred manages them; no rc-service or sudo required.
Bots are discovered automatically from ~/.config/sopel/*.cfg - add a new
config file there and it will appear in !bot list on the next command.
!bot list - show status of all managed bots
!bot all <start|stop|restart> - control all bots at once
!bot <name> <start|stop|restart> - control a bot
!bot <name> status - show status of one bot
"""
import glob
import os
import subprocess
import time
from sopel import plugin
import os as _os, sys as _sys
_alfred_dir = _os.path.dirname(_os.path.abspath(__file__))
if _alfred_dir not in _sys.path:
_sys.path.insert(0, _alfred_dir)
import helpstrings as h
SOPEL = "/usr/bin/sopel"
CFG_DIR = "/home/ahmed/.config/sopel"
EXCLUDE = {"alfred"}
def _discover():
paths = glob.glob(os.path.join(CFG_DIR, "*.cfg"))
return {
os.path.splitext(os.path.basename(p))[0]: p
for p in sorted(paths)
if os.path.splitext(os.path.basename(p))[0] not in EXCLUDE
}
def _is_owner(bot, trigger):
return trigger.nick == bot.settings.core.owner
def _running(cfg_path):
r = subprocess.run(["pgrep", "-f", f"sopel.*{cfg_path}"], capture_output=True)
return r.returncode == 0
def _start(name, cfg_path):
if _running(cfg_path):
return f"{name}: already running"
subprocess.Popen(
[SOPEL, "--config", cfg_path],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
)
return f"{name}: started"
def _stop(name, cfg_path):
r = subprocess.run(["pkill", "-f", f"sopel.*{cfg_path}"], capture_output=True)
return f"{name}: stopped" if r.returncode == 0 else f"{name}: not running"
def _restart(name, cfg_path):
_stop(name, cfg_path)
for _ in range(10):
time.sleep(0.2)
if not _running(cfg_path):
break
return _start(name, cfg_path)
@plugin.command("bot")
def cmd_bot(bot, trigger):
if not _is_owner(bot, trigger):
return
bots = _discover()
args = (trigger.group(2) or "").split()
if not args:
bot.say(h.BOT_TOPIC)
return
if args[0] == "list":
if not bots:
bot.say("no bots found")
return
bot.say(" | ".join(
f"{n}:{'up' if _running(p) else 'down'}" for n, p in bots.items()
))
return
if args[0] == "all":
if len(args) < 2 or args[1] not in ("start", "stop", "restart"):
bot.say("usage: !bot all <start|stop|restart>")
return
action = args[1]
if not bots:
bot.say("no bots found")
return
fn = {"start": _start, "stop": _stop, "restart": _restart}[action]
results = [fn(n, p) for n, p in bots.items()]
bot.say(" | ".join(results))
return
name = args[0]
if name not in bots:
bot.say(f"unknown bot - known: {', '.join(bots)}")
return
if len(args) < 2:
bot.say(f"usage: !bot {name} <start|stop|restart|status>")
return
cfg_path = bots[name]
action = args[1]
if action == "start":
bot.say(_start(name, cfg_path))
elif action == "stop":
bot.say(_stop(name, cfg_path))
elif action == "restart":
bot.say(_restart(name, cfg_path))
elif action == "status":
bot.say(f"{name}: {'running' if _running(cfg_path) else 'stopped'}")
else:
bot.say(f"usage: !bot {name} <start|stop|restart|status>")
|
