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
|
"""
Jeeves core helpers. Not loaded directly by Sopel (not in enable list).
All jeeves plugins import this via sys.path insertion.
Provides: ensure_setup, save, cfg, cfg_val, chan_db, CONFIG_KEYS.
"""
import json
import os
import threading
from sopel.config.types import FilenameAttribute, StaticSection
# (type, default, global_only, description)
CONFIG_KEYS = {
"dj_track_duration": (int, 120, True, "seconds per track before auto-advance"),
"dj_queue_cap": (int, 50, True, "max tracks in queue"),
"art_height": (int, 6, True, "default lines for !draw (4-12)"),
"duck_spawn_min": (int, 60, False, "min seconds between duck spawns"),
"duck_spawn_max": (int, 300, False, "max seconds between duck spawns"),
"duck_flee_time": (int, 45, False, "seconds before duck flees"),
"duck_idle_rounds": (int, 3, False, "empty rounds before auto-stop (0=off)"),
"news_interval": (int, 90, False, "minutes between news broadcasts"),
"news_count": (int, 1, False, "headlines per broadcast cycle"),
"quotes_interval": (int, 60, False, "minutes between quote auto-posts"),
}
_SEEN_MAX = 1000
class JeevesSection(StaticSection):
db_path = FilenameAttribute("db_path", relative=False, default="/var/jeeves/db.json")
duck_db_path = FilenameAttribute("duck_db_path", relative=False, default="/var/jeeves/scores.db")
recipes_path = FilenameAttribute("recipes_path", relative=False, default="/var/jeeves/recipes.json")
feeds_path = FilenameAttribute("feeds_path", relative=False, default="/var/jeeves/feeds.json")
def _load_feeds(path):
try:
with open(path) as f:
return json.load(f)
except Exception:
return []
def _load(path, feeds_path):
feeds = _load_feeds(feeds_path)
if os.path.exists(path):
try:
with open(path) as f:
data = json.load(f)
data.setdefault("config", {})
data.setdefault("channels", {})
data.setdefault("quotes", [])
data.setdefault("news_feeds", feeds)
data.setdefault("news_seen", [])
return data
except Exception:
pass
data = {
"config": {},
"channels": {},
"quotes": [],
"news_feeds": feeds,
"news_seen": [],
}
_save_raw(path, data)
return data
def _save_raw(path, data):
os.makedirs(os.path.dirname(path), exist_ok=True)
seen = data.get("news_seen", [])
if len(seen) > _SEEN_MAX:
data["news_seen"] = seen[-_SEEN_MAX:]
with open(path, "w") as f:
json.dump(data, f, indent=2)
def save(bot):
_save_raw(bot.memory["jv_db_path"], bot.memory["jv_db"])
def ensure_setup(bot):
if "jv_db" in bot.memory:
return
try:
bot.config.define_section("jeeves", JeevesSection, validate=False)
db_path = bot.config.jeeves.db_path
duck_db_path = bot.config.jeeves.duck_db_path
recipes_path = bot.config.jeeves.recipes_path
feeds_path = bot.config.jeeves.feeds_path
except Exception:
db_path = "/var/jeeves/db.json"
duck_db_path = "/var/jeeves/scores.db"
recipes_path = "/var/jeeves/recipes.json"
feeds_path = "/var/jeeves/feeds.json"
bot.memory["jv_db_path"] = db_path
bot.memory["jv_duck_db_path"] = duck_db_path
bot.memory["jv_recipes_path"] = recipes_path
bot.memory["jv_feeds_path"] = feeds_path
bot.memory["jv_db"] = _load(db_path, feeds_path)
bot.memory["jv_lock"] = threading.Lock()
def cfg(bot, channel=None):
db = bot.memory.get("jv_db", {})
global_cfg = db.get("config", {})
if channel:
chan_cfg = (db.get("channels", {})
.get(channel.lower(), {})
.get("config", {}))
return {**global_cfg, **chan_cfg}
return dict(global_cfg)
def cfg_val(bot, key, channel=None):
merged = cfg(bot, channel)
default = CONFIG_KEYS[key][1]
return type(CONFIG_KEYS[key][0])(merged.get(key, default))
def chan_db(bot, channel):
db = bot.memory["jv_db"]
db.setdefault("channels", {})
ch = db["channels"].setdefault(channel.lower(), {})
ch.setdefault("config", {})
ch.setdefault("news", {"enabled": True, "last_broadcast": 0})
ch.setdefault("quotes", {"enabled": False, "last_broadcast": 0})
return ch
def is_owner(bot, trigger):
return trigger.nick == bot.settings.core.owner
def is_op(bot, trigger):
chan = bot.channels.get(str(trigger.sender))
if not chan:
return False
from sopel import plugin as _p
return chan.privileges.get(trigger.nick, 0) >= _p.OP
def is_authorized(bot, trigger):
return is_owner(bot, trigger) or is_op(bot, trigger)
|
