diff --git a/.gitignore b/.gitignore index 6115fe4..4c4dcce 100644 --- a/.gitignore +++ b/.gitignore @@ -12,6 +12,7 @@ data.json.bak settings*.conf NOTES.md sanity/ +.offline_data.json # Tools wrapper/ diff --git a/assets/docs/commands/__commands__.json b/assets/docs/commands/__commands__.json new file mode 100644 index 0000000..9e26dfe --- /dev/null +++ b/assets/docs/commands/__commands__.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/assets/docs/no.svg b/assets/docs/no.svg new file mode 100644 index 0000000..008ada5 --- /dev/null +++ b/assets/docs/no.svg @@ -0,0 +1,7 @@ + + + Flag of Norway + + + + \ No newline at end of file diff --git a/bot.py b/bot.py index 8e3c746..71a607e 100644 --- a/bot.py +++ b/bot.py @@ -9,7 +9,7 @@ from modules.common.boot_notice import post_boot_notice # Version consists of: # Major.Enhancement.Minor.Patch.Test (Test is alphanumeric; doesn’t trigger auto update) -VERSION = "0.3.9.8.a1" +VERSION = "0.4.0.0.a1" # ---------- Env loading ---------- diff --git a/dev/offline_preview.py b/dev/offline_preview.py new file mode 100644 index 0000000..780fb47 --- /dev/null +++ b/dev/offline_preview.py @@ -0,0 +1,148 @@ +# offline_preview.py +""" +ShaiWatcher offline preview (Discord-less) + +Run from anywhere: + # optional (forces root if your layout is odd) + # export SHAI_PROJECT_ROOT=/absolute/path/to/shaiwatcher + # export SHAI_DOCS_HOST=127.0.0.1 + # export SHAI_DOCS_PORT=8911 + # export SHAI_OFFLINE=1 + python3 offline_preview.py +""" +import os +import sys +import asyncio +import pathlib +import traceback + +VERSION = "offline-preview-3" + +# ---------- repo root discovery ---------- +def _find_project_root() -> pathlib.Path: + cand = [] + env = os.environ.get("SHAI_PROJECT_ROOT") + if env: + cand.append(pathlib.Path(env).resolve()) + here = pathlib.Path(__file__).resolve().parent + cand.extend([ + pathlib.Path.cwd().resolve(), # current working dir + here, # folder containing this file + here.parent, # one level up + here.parent.parent, # two levels up + ]) + # Also walk upwards from CWD a few levels to be forgiving + cur = pathlib.Path.cwd().resolve() + for _ in range(5): + cand.append(cur) + cur = cur.parent + + tried = [] + for c in cand: + tried.append(str(c)) + if (c / "modules").is_dir() and (c / "modules" / "common").is_dir(): + return c + + raise FileNotFoundError( + "Could not locate project root with a 'modules/common' folder.\n" + f"Tried:\n - " + "\n - ".join(tried) + + "\nTip: set SHAI_PROJECT_ROOT=/absolute/path/to/repo" + ) + +PROJECT_ROOT = _find_project_root() +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +# ---------- now safe to import project modules ---------- +import discord # type: ignore +from discord.ext import commands # type: ignore + +# Optional: your config helper if cogs expect it to exist +try: + from modules.common.settings import cfg as cfg_helper # noqa: F401 +except Exception as e: + print("[OFFLINE] Warning: couldn't import cfg helper:", repr(e)) + +def _discover_extensions(project_root: pathlib.Path): + modules_path = project_root / "modules" + exts = [] + for folder in modules_path.iterdir(): + if not folder.is_dir(): + continue + if folder.name == "common": # match your prod loader + continue + for file in folder.glob("*.py"): + if file.name == "__init__.py": + continue + exts.append(f"modules.{folder.name}.{file.stem}") + return exts + +async def main(): + print(f"[OFFLINE] ShaiWatcher offline preview v{VERSION}") + print(f"[OFFLINE] Project root -> {PROJECT_ROOT}") + + # Keep intents minimal; we never connect anyway + intents = discord.Intents.none() + intents.guilds = True + + bot = commands.Bot(command_prefix="!", intents=intents) + + # Mark environment as offline for any cogs that check it + os.environ.setdefault("SHAI_OFFLINE", "1") + # Bind docs to localhost by default while testing + os.environ.setdefault("SHAI_DOCS_HOST", "127.0.0.1") + os.environ.setdefault("SHAI_DOCS_PORT", "8911") + os.environ.setdefault("SHAI_DOCS_TITLE", "ShaiWatcher (Offline Preview)") + + # Optional: isolate data file so we don't touch prod paths + data_file = os.environ.get("SHAI_DATA", str(PROJECT_ROOT / ".offline_data.json")) + try: + from data_manager import DataManager # if your project has this at root + os.makedirs(os.path.dirname(data_file) or ".", exist_ok=True) + if not os.path.exists(data_file): + with open(data_file, "w", encoding="utf-8") as f: + f.write("{}") + bot.data_manager = DataManager(data_file) + print(f"[OFFLINE] DATA_FILE -> {data_file}") + except Exception as e: + print("[OFFLINE] DataManager unavailable/failed:", repr(e)) + + os.environ.setdefault("SHAI_OFFLINE", "1") # before loading cogs + + # Load extensions exactly like prod + failures = 0 + for ext in _discover_extensions(PROJECT_ROOT): + try: + await bot.load_extension(ext) + print(f"[Modules] Loaded: {ext}") + except Exception as e: + failures += 1 + print(f"[Modules] Failed to load {ext}: {e}") + traceback.print_exc() + + if failures: + print(f"[OFFLINE] Loaded with {failures} module error(s). See logs above.") + + docs = bot.get_cog("DocsSite") + if docs and hasattr(docs, "force_ready"): + docs.force_ready(True) + + # Make is_ready() == True so DocsSite serves immediately + try: + # discord.py sets this in login/READY; we emulate it + if not hasattr(bot, "_ready") or bot._ready is None: # type: ignore[attr-defined] + bot._ready = asyncio.Event() # type: ignore[attr-defined] + bot._ready.set() # type: ignore[attr-defined] + except Exception: + pass + + print("[OFFLINE] Docs: http://%s:%s/" + % (os.environ.get("SHAI_DOCS_HOST", "127.0.0.1"), + os.environ.get("SHAI_DOCS_PORT", "8911"))) + print("[OFFLINE] This runner does NOT connect to Discord.") + + # Idle forever; DocsSite runs in its own daemon thread + await asyncio.Event().wait() + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/modules/docs_site/docs_site.py b/modules/docs_site/docs_site.py index f3b9214..a16d5fa 100644 --- a/modules/docs_site/docs_site.py +++ b/modules/docs_site/docs_site.py @@ -1,6 +1,12 @@ # modules/docs_site/docs_site.py import json import threading +import traceback +import os +import re +import time +import importlib +from pathlib import Path from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler from urllib.parse import urlparse from typing import Any, Dict, List, Optional, Tuple @@ -10,46 +16,124 @@ from discord.ext import commands from discord import app_commands from modules.common.settings import cfg -import mod_perms # for name/qualname detection only (no calls) -# ----------------------------- -# Helpers: command introspection -# ----------------------------- +_START_TS = time.time() -def _is_perm_check(fn) -> Tuple[bool, List[str]]: - """ - Detects if a prefix/hybrid command check enforces guild/admin permissions. - Returns (has_perm_check, [perm_names]). - """ +# ============================= +# Safe JSON helpers +# ============================= + +_PRIMITIVES = (str, int, float, bool, type(None)) + +def _to_primitive(obj: Any, depth: int = 0) -> Any: + if depth > 6: + return str(obj) + if isinstance(obj, _PRIMITIVES): + return obj + if isinstance(obj, (list, tuple, set)): + return [_to_primitive(x, depth + 1) for x in obj] + if isinstance(obj, dict): + out = {} + for k, v in obj.items(): + try: + out[str(k)] = _to_primitive(v, depth + 1) + except Exception: + out[str(k)] = str(v) + return out + if hasattr(obj, "value"): + try: + return _to_primitive(getattr(obj, "value"), depth + 1) + except Exception: + pass + return str(obj) + +def _json_dumps_safe(payload: Any) -> bytes: + return json.dumps(_to_primitive(payload), ensure_ascii=False, separators=(",", ":")).encode("utf-8") + +# ============================= +# File serving helpers +# ============================= + +def _static_root() -> Path: + return _project_root() / "assets" / "docs" + +def _guess_mime(p: Path) -> str: + ext = p.suffix.lower() + return { + ".svg": "image/svg+xml; charset=utf-8", + ".png": "image/png", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".webp": "image/webp", + ".gif": "image/gif", + ".css": "text/css; charset=utf-8", + ".js": "application/javascript; charset=utf-8", + ".json": "application/json; charset=utf-8", + ".md": "text/markdown; charset=utf-8", + ".txt": "text/plain; charset=utf-8", + }.get(ext, "application/octet-stream") + +# ============================= +# Version / uptime helpers +# ============================= + +def _project_root() -> Path: + return Path(__file__).resolve().parents[2] + +def _read_version_from_file() -> Optional[str]: + """Strictly parse VERSION from bot.py: VERSION = "x.y.z".""" try: - name = getattr(fn, "__qualname__", "") or getattr(fn, "__name__", "") + bot_py = _project_root() / "bot.py" + if not bot_py.is_file(): + return None + text = bot_py.read_text(encoding="utf-8", errors="ignore") + m = re.search(r'^\s*VERSION\s*=\s*["\']([^"\']+)["\']', text, re.M) + if m: + return m.group(1) except Exception: - name = "" + traceback.print_exc() + return None - # discord.py wraps has_permissions/has_guild_permissions checks; they carry attrs sometimes. - perms = [] +def _get_version_from_botpy() -> Optional[str]: + # Secondary fallback: import try: - # Some check predicates store desired perms on closure/cell vars; best-effort: - code = getattr(fn, "__code__", None) - if code and code.co_freevars and getattr(fn, "__closure__", None): - for cell in fn.__closure__ or []: - val = getattr(cell, "cell_contents", None) - if isinstance(val, dict): - # likely {'manage_messages': True, ...} - for k, v in val.items(): - if v: - perms.append(str(k)) + m = importlib.import_module("bot") + v = getattr(m, "VERSION", None) + if isinstance(v, (str, int, float)): + return str(v) except Exception: pass + return None - has_perm = any(p for p in perms) - return has_perm, perms +def _get_boot_state(bot: commands.Bot) -> Dict[str, Any]: + dm = getattr(bot, "data_manager", None) + if dm and dm.get("boot_state"): + try: + return (dm.get("boot_state") or [{}])[-1] + except Exception: + return {} + return {} +def _status_payload(bot: commands.Bot) -> Dict[str, Any]: + st = _get_boot_state(bot) + now = time.time() + boot_ts = float(st.get("last_boot_ts", 0.0)) if st else 0.0 + if boot_ts > 0: + uptime = max(0, int(now - boot_ts)) + else: + uptime = max(0, int(now - _START_TS)) + + ver = st.get("last_version") if st else None + if not ver: + ver = _read_version_from_file() or _get_version_from_botpy() or os.getenv("SHAI_VERSION", "unknown") + return {"uptime_seconds": uptime, "version": str(ver)} + + +# ============================= +# Mod detection (heuristics) +# ============================= def _looks_like_mod_check(fn) -> bool: - """ - Heuristic: identify checks referring to your moderator helpers. - """ try: qn = getattr(fn, "__qualname__", "") or "" mod_names = ( @@ -60,7 +144,6 @@ def _looks_like_mod_check(fn) -> bool: ) if any(m in qn for m in mod_names): return True - # Also check module path/name: mod = getattr(fn, "__module__", "") or "" if "mod_perms" in mod: return True @@ -68,181 +151,402 @@ def _looks_like_mod_check(fn) -> bool: pass return False +def _is_perm_check(fn) -> Tuple[bool, List[str]]: + perms = [] + try: + code = getattr(fn, "__code__", None) + if code and code.co_freevars and getattr(fn, "__closure__", None): + for cell in fn.__closure__ or []: + val = getattr(cell, "cell_contents", None) + if isinstance(val, dict): + for k, v in val.items(): + if v: + perms.append(str(k)) + except Exception: + pass + return (len(perms) > 0), perms + + +# ============================= +# Introspection +# ============================= + +def _bot_prefix(bot: commands.Bot) -> str: + p = getattr(bot, "command_prefix", "!") + try: + if callable(p): + val = p(bot, None) + if isinstance(val, (list, tuple)) and val: + return str(val[0]) + return str(val) + return str(p) + except Exception: + return "!" def _is_mod_command_prefix(cmd: commands.Command) -> Tuple[bool, List[str]]: - """ - For prefix/hybrid commands: inspect .checks for moderator checks or admin perms. - """ is_mod = False perms: List[str] = [] try: for chk in getattr(cmd, "checks", []) or []: - m = _looks_like_mod_check(chk) - p_flag, p_list = _is_perm_check(chk) - if m: + if _looks_like_mod_check(chk): is_mod = True + p_flag, p_list = _is_perm_check(chk) if p_flag: perms.extend(p_list) except Exception: pass - - # If command declares directly required perms (rare), capture them too: - # No official attribute on commands.Command; we rely on checks only. return is_mod, sorted(set(perms)) - def _is_mod_command_slash(cmd: app_commands.Command) -> Tuple[bool, List[str]]: - """ - For slash commands: inspect .checks (app_commands) if present. - """ is_mod = False - perms: List[str] = [] try: for chk in getattr(cmd, "checks", []) or []: - m = _looks_like_mod_check(chk) - # app_commands has no built-in has_permissions; users often wrap their own. - if m: + if _looks_like_mod_check(chk): is_mod = True - except Exception: - pass - # Best‑effort: look into ._checks (private) if available - try: - for chk in getattr(cmd, "_checks", []) or []: + for chk in getattr(cmd, "_checks", []) or []: # private best-effort if _looks_like_mod_check(chk): is_mod = True except Exception: pass - return is_mod, perms + return is_mod, [] - -def _command_usage_prefix(cmd: commands.Command) -> str: - # Prefer .usage, else generate "/help"-like usage. +def _command_usage_prefix(cmd: commands.Command, prefix: str) -> str: if cmd.usage: return cmd.usage try: - # Build basic placeholder usage params = [] - for k, p in cmd.clean_params.items(): - if p.kind.name.lower().startswith("var"): + for k, p in (cmd.clean_params or {}).items(): + if getattr(p, "kind", None) and str(p.kind).lower().startswith("var"): params.append(f"[{k}...]") elif p.default is p.empty: params.append(f"<{k}>") else: params.append(f"[{k}]") - if params: - return f"!{cmd.qualified_name} " + " ".join(params) - return f"!{cmd.qualified_name}" + return f"{prefix}{cmd.qualified_name}" + ((" " + " ".join(params)) if params else "") except Exception: - return f"!{cmd.name}" + return f"{prefix}{cmd.name}" - -def _command_usage_slash(cmd: app_commands.Command) -> str: - # Slash usage is mostly the path; options aren’t trivial to enumerate without ._params. - # Do a safe best‑effort: +def _command_usage_slash_like(cmd_name: str, options: Optional[List[Any]] = None) -> str: try: - parts = [f"/{cmd.name}"] - # options (private API in discord.py); keep it safe and minimal: + parts = [f"/{cmd_name}"] opts = [] - for opt in getattr(cmd, "options", []) or []: - n = getattr(opt, "name", "arg") - req = getattr(opt, "required", False) + seq = options or [] + for opt in seq: + n = getattr(opt, "name", None) or getattr(opt, "display_name", None) or "arg" + req = bool(getattr(opt, "required", False)) opts.append(f"<{n}>" if req else f"[{n}]") if opts: parts.append(" " + " ".join(opts)) return "".join(parts) except Exception: - return f"/{cmd.name}" + return f"/{cmd_name}" +def _command_usage_slash(cmd: app_commands.Command) -> str: + try: + options = getattr(cmd, "options", None) or getattr(cmd, "parameters", None) or getattr(cmd, "_params", None) + return _command_usage_slash_like((cmd.name or "").replace("/", " "), options) + except Exception: + return f"/{cmd.name}" def _gather_prefix_and_hybrid(bot: commands.Bot) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] seen = set() - for cmd in bot.commands: - # Skip hidden - if getattr(cmd, "hidden", False): - continue + prefix = _bot_prefix(bot) - ctype = "hybrid" if isinstance(cmd, commands.HybridCommand) else "prefix" - is_mod, perms = _is_mod_command_prefix(cmd) - row = { - "type": ctype, - "name": cmd.qualified_name, - "help": (cmd.help or "").strip(), - "brief": (cmd.brief or "").strip(), - "usage": _command_usage_prefix(cmd), - "cog": getattr(cmd.cog, "qualified_name", None) if getattr(cmd, "cog", None) else None, - "module": getattr(cmd.callback, "__module__", None), - "moderator_only": bool(is_mod), - "required_permissions": perms, - } - key = ("px", row["name"]) - if key not in seen: - rows.append(row) - seen.add(key) + for cmd in getattr(bot, "commands", []) or []: + try: + if getattr(cmd, "hidden", False): + continue + is_hybrid = isinstance(cmd, commands.HybridCommand) + ctype = "hybrid" if is_hybrid else "prefix" + + is_mod, perms = _is_mod_command_prefix(cmd) + + usage_slash = None + if is_hybrid: + try: + app_cmd = getattr(cmd, "app_command", None) + if isinstance(app_cmd, app_commands.Command): + usage_slash = _command_usage_slash(app_cmd) + s_mod, _ = _is_mod_command_slash(app_cmd) + if s_mod: + is_mod = True + desc = (getattr(app_cmd, "description", "") or "") + if "[mod]" in desc.lower(): + is_mod = True + except Exception: + pass + + usage_prefix = _command_usage_prefix(cmd, prefix) + + row = { + "type": ctype, + "name": cmd.qualified_name, + "display_name": cmd.qualified_name, # bare name, no slash/prefix + "help": (cmd.help or "").strip(), + "brief": (cmd.brief or "").strip(), + "usage": usage_prefix if not is_hybrid else None, + "usage_prefix": usage_prefix, + "usage_slash": usage_slash, + "cog": getattr(cmd.cog, "qualified_name", None) if getattr(cmd, "cog", None) else None, + "module": getattr(getattr(cmd, "callback", None), "__module__", None), + "moderator_only": bool(is_mod), + "required_permissions": perms, + } + key = ("px", row["name"]) + if key not in seen: + rows.append(row) + seen.add(key) + except Exception: + traceback.print_exc() + continue return rows - def _walk_app_tree(node: Any, prefix: str = "") -> List[Tuple[str, app_commands.Command]]: - """ - Flatten app command tree (includes groups/subcommands) into (path, command) pairs. - """ out: List[Tuple[str, app_commands.Command]] = [] if isinstance(node, app_commands.Command): out.append((f"{prefix}/{node.name}", node)) return out if isinstance(node, app_commands.Group): base = f"{prefix}/{node.name}" - # Subcommands or nested groups for sub in list(getattr(node, "commands", []) or []): out.extend(_walk_app_tree(sub, base)) return out +def _safe_extras(obj: Any) -> Optional[Dict[str, Any]]: + d = getattr(obj, "extras", None) + if not d: + return None + if not isinstance(d, dict): + return {"value": _to_primitive(d)} + return _to_primitive(d) def _gather_slash(bot: commands.Bot) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] try: - # Local view of registered commands (no network call) cmds = bot.tree.get_commands() except Exception: + traceback.print_exc() cmds = [] - - for cmd in cmds: + for cmd in cmds or []: for path, leaf in _walk_app_tree(cmd, prefix=""): - is_mod, perms = _is_mod_command_slash(leaf) - row = { - "type": "slash", - "name": path, # includes group path - "help": (leaf.description or "").strip(), - "brief": "", - "usage": _command_usage_slash(leaf), - "cog": getattr(leaf, "binding", None).__class__.__name__ if getattr(leaf, "binding", None) else None, - "module": getattr(leaf.callback, "__module__", None) if getattr(leaf, "callback", None) else None, - "moderator_only": bool(is_mod), - "required_permissions": perms, - # Extras (discord.py 2.x): curation knobs per-command if authors set them - "extras": dict(getattr(leaf, "extras", {}) or {}), - # DM permission when available (discord.py exposes default permissions in command tree) - "dm_permission": getattr(leaf, "dm_permission", None), - } - rows.append(row) + try: + is_mod, perms = _is_mod_command_slash(leaf) + binding = getattr(leaf, "binding", None) + callback = getattr(leaf, "callback", None) + + display = "/" + path.lstrip("/").replace("/", " ") + options = getattr(leaf, "options", None) or getattr(leaf, "parameters", None) or getattr(leaf, "_params", None) + usage_full = _command_usage_slash_like(path.lstrip("/").replace("/", " "), options) + + row = { + "type": "slash", + "name": path, # canonical '/group/sub' + "display_name": display, # shown without the leading '/' + "help": (getattr(leaf, "description", "") or "").strip(), + "brief": "", + "usage": usage_full, + "usage_prefix": None, + "usage_slash": usage_full, + "cog": binding.__class__.__name__ if binding else None, + "module": getattr(callback, "__module__", None) if callback else None, + "moderator_only": bool(is_mod), + "required_permissions": perms, + "extras": _safe_extras(leaf), + "dm_permission": getattr(leaf, "dm_permission", None), + } + rows.append(row) + except Exception: + traceback.print_exc() + continue return rows +# ============================= +# Details loader & master JSON +# ============================= + +_DETAILS_DIR = _project_root() / "assets" / "docs" / "commands" +_MASTER_PATH = _DETAILS_DIR / "__commands__.json" + +_DOCS_CACHE: Dict[str, Any] = {"map": {}, "sig": ""} + +def _row_key_candidates(row: Dict[str, Any]) -> List[str]: + c = str(row.get("cog") or "").strip() + if row.get("type") == "slash": + base = str(row.get("name", "")).lstrip("/").split("/")[-1] + else: + base = str(row.get("name", "")).split(" ")[0] + keys = [] + if c: + keys.append(f"{c}.{base}") + if row.get("type") == "slash": + keys.append(str(row.get("name", "")).lstrip("/")) # e.g., power/restart + keys.append(base) + return keys + +def _scan_md_files() -> Dict[str, Dict[str, Any]]: + out: Dict[str, Dict[str, Any]] = {} + if not _DETAILS_DIR.is_dir(): + return out + for p in _DETAILS_DIR.glob("*.md"): + if p.name.startswith("_"): + continue + try: + out[p.stem] = {"details_md": p.read_text(encoding="utf-8")} + except Exception: + traceback.print_exc() + continue + return out + +def _load_master_json() -> Dict[str, Dict[str, Any]]: + if not _MASTER_PATH.is_file(): + return {} + try: + raw_text = _MASTER_PATH.read_text(encoding="utf-8") + if not raw_text.strip(): + return {} + raw_any = json.loads(raw_text) or {} + out: Dict[str, Dict[str, Any]] = {} + for k, v in raw_any.items(): + if isinstance(v, dict): + out[str(k)] = {kk: vv for kk, vv in v.items()} + return out + except Exception: + traceback.print_exc() + return {} + +def _write_master_json(mapping: Dict[str, Dict[str, Any]]) -> None: + try: + _DETAILS_DIR.mkdir(parents=True, exist_ok=True) + with _MASTER_PATH.open("w", encoding="utf-8") as f: + json.dump(mapping, f, ensure_ascii=False, indent=2) + except Exception: + traceback.print_exc() + +def _dir_signature() -> str: + parts = [] + if _DETAILS_DIR.is_dir(): + for p in sorted(_DETAILS_DIR.glob("*")): + try: + parts.append(f"{p.name}:{int(p.stat().st_mtime)}") + except Exception: + continue + try: + if _MASTER_PATH.exists(): + parts.append(f"__master__:{int(_MASTER_PATH.stat().st_mtime)}") + except Exception: + pass + return "|".join(parts) + +def _load_external_docs() -> Dict[str, Dict[str, Any]]: + sig = _dir_signature() + if _DOCS_CACHE.get("sig") == sig: + return _DOCS_CACHE["map"] + + master = _load_master_json() + md_map = _scan_md_files() + + merged: Dict[str, Dict[str, Any]] = {k: dict(v) for k, v in master.items()} + for k, v in md_map.items(): + if k not in merged: + merged[k] = {} + merged[k]["details_md"] = v.get("details_md", "") + + _write_master_json(merged) + _DOCS_CACHE["map"] = merged + _DOCS_CACHE["sig"] = sig + return merged + +def _augment_with_external_docs(rows: List[Dict[str, Any]]) -> None: + mapping = _load_external_docs() + for r in rows: + if not r.get("details_md") and isinstance(r.get("extras"), dict): + dm = r["extras"].get("details_md") + if isinstance(dm, str) and dm.strip(): + r["details_md"] = dm + if not r.get("details_md"): + for key in _row_key_candidates(r): + if key in mapping and isinstance(mapping[key], dict): + dm = mapping[key].get("details_md") + if isinstance(dm, str) and dm.strip(): + r["details_md"] = dm + meta = {kk: vv for kk, vv in mapping[key].items() if kk != "details_md"} + if meta: + r["doc_meta"] = meta + break + + +# ============================= +# Merge hybrids with their slash twins +# ============================= + +def _merge_hybrid_slash(rows: List[Dict[str, Any]]) -> None: + idx_by_hybrid: Dict[Tuple[str, str], int] = {} + for i, r in enumerate(rows): + if r.get("type") == "hybrid": + cog = (r.get("cog") or "").strip() + base = str(r.get("name") or "").split(" ")[-1].lower() + idx_by_hybrid[(cog, base)] = i + + to_remove: List[int] = [] + for i, r in enumerate(rows): + if r.get("type") != "slash": + continue + base = str(r.get("name") or "").lstrip("/").split("/")[-1].lower() + cog = (r.get("cog") or "").strip() + key = (cog, base) + hi = idx_by_hybrid.get(key) + if hi is None: + continue + h = rows[hi] + if not h.get("usage_slash") and r.get("usage_slash"): + h["usage_slash"] = r["usage_slash"] + if not h.get("help") and r.get("help"): + h["help"] = r["help"] + if r.get("moderator_only"): + h["moderator_only"] = True + if r.get("required_permissions"): + h["required_permissions"] = sorted(set((h.get("required_permissions") or []) + r["required_permissions"])) + if not h.get("extras") and r.get("extras"): + h["extras"] = r["extras"] + if r.get("details_md") and not h.get("details_md"): + h["details_md"] = r["details_md"] + to_remove.append(i) + + for i in sorted(to_remove, reverse=True): + rows.pop(i) + + +# ============================= +# Schema builder +# ============================= + def build_command_schema(bot: commands.Bot) -> Dict[str, Any]: px = _gather_prefix_and_hybrid(bot) sl = _gather_slash(bot) all_rows = px + sl - # Derive “moderator” flag if extras hint exists (in case checks missed it) + # Additional moderator checks & sanitization for row in all_rows: + try: + helptext = f"{row.get('help') or ''} {row.get('brief') or ''}" + if "[mod]" in helptext.lower(): + row["moderator_only"] = True + except Exception: + pass + if row.get("required_permissions"): + row["moderator_only"] = True try: ex = row.get("extras") or {} - if isinstance(ex, dict) and ex.get("category", "").lower() in {"mod", "moderator", "staff"}: + if isinstance(ex, dict) and str(ex.get("category", "")).lower() in {"mod", "moderator", "staff"}: row["moderator_only"] = True except Exception: pass - # Basic sections + _augment_with_external_docs(all_rows) + _merge_hybrid_slash(all_rows) + mods = [r for r in all_rows if r.get("moderator_only")] users = [r for r in all_rows if not r.get("moderator_only")] @@ -257,199 +561,479 @@ def build_command_schema(bot: commands.Bot) -> Dict[str, Any]: } -# ----------------------------- -# HTTP server + request handler -# ----------------------------- +# ============================= +# HTTP + UI +# ============================= _HTML = """ - - -{title} + +__TITLE__ -

{title}

+

__TITLE__

-
-
-
- -
Sections: All · Moderator · User
-
-
-
-
-
-
-

Notes

-
    -
  • No authentication; info is read-only and non-sensitive.
    • -
  • Moderator-only is inferred from checks (best-effort).
    • -
  • Slash commands show full path if grouped, e.g. /group/sub.
    • -
-

JSON: /api/commands · Health: /healthz

-
-
-

Moderator commands

-
-
+ +
+ +
+ Sections: User · Moderator · All + +
+ + +
+ +
+ +
+
+
+ + +
+
+
+
+ + + +
""" class _DocsHandler(BaseHTTPRequestHandler): - # We attach bot + title on the class at server startup bot: commands.Bot = None title: str = "ShaiWatcher Commands" + force_ready: bool = False def _set(self, code=200, content_type="text/html; charset=utf-8"): self.send_response(code) self.send_header("Content-Type", content_type) - # Tiny bit of sanity for browser caches: self.send_header("Cache-Control", "no-store") self.end_headers() def log_message(self, fmt, *args): - # Silence default noisy logs return def do_GET(self): try: + ready = bool(self.force_ready) or (getattr(self, "bot", None) and self.bot.is_ready()) + if not ready: + self._set(503, "text/plain; charset=utf-8") + self.wfile.write(b"warming up") + return + path = urlparse(self.path).path if path == "/": self._set() - self.wfile.write(_HTML.format(title=self.title).encode("utf-8")) + html = _HTML.replace("__TITLE__", self.title) + try: + schema = build_command_schema(self.bot) + inline = json.dumps(_to_primitive(schema), ensure_ascii=False, separators=(",", ":")) + html = html.replace("", f"") + except Exception: + traceback.print_exc() + self.wfile.write(html.encode("utf-8")) return + + path = urlparse(self.path).path + + # Static assets: /assets/docs/... + if path.startswith("/assets/docs/"): + try: + root = _static_root().resolve() + rel = path[len("/assets/docs/"):] + fs_path = (root / rel).resolve() + # Prevent path traversal + try: + # Python 3.10+: Path.is_relative_to + if not fs_path.is_relative_to(root): + raise ValueError("outside root") + except AttributeError: + # Fallback for very old Python (not needed on 3.10+) + if str(root) not in str(fs_path): + raise ValueError("outside root") + if fs_path.is_file(): + mime = _guess_mime(fs_path) + self.send_response(200) + self.send_header("Content-Type", mime) + self.send_header("Cache-Control", "public, max-age=86400, immutable") + self.end_headers() + with fs_path.open("rb") as f: + self.wfile.write(f.read()) + return + self._set(404, "text/plain; charset=utf-8") + self.wfile.write(b"not found") + return + except Exception: + traceback.print_exc() + self._set(500, "text/plain; charset=utf-8") + self.wfile.write(b"internal error") + return + + if path == "/api/status": + payload = _status_payload(self.bot) + self._set(200, "application/json; charset=utf-8") + self.wfile.write(_json_dumps_safe(payload)) + return + if path == "/healthz": self._set(200, "text/plain; charset=utf-8") self.wfile.write(b"ok") return + if path == "/api/commands": schema = build_command_schema(self.bot) - payload = json.dumps(schema, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + payload = _json_dumps_safe(schema) self._set(200, "application/json; charset=utf-8") self.wfile.write(payload) return + self._set(404, "text/plain; charset=utf-8") self.wfile.write(b"not found") except Exception: + traceback.print_exc() try: self._set(500, "text/plain; charset=utf-8") self.wfile.write(b"internal error") except Exception: pass - def _start_server(bot: commands.Bot, host: str, port: int, title: str): _DocsHandler.bot = bot _DocsHandler.title = title + _DocsHandler.force_ready = os.getenv("SHAI_OFFLINE", "").lower() in {"1", "true", "yes"} httpd = ThreadingHTTPServer((host, port), _DocsHandler) def _run(): try: httpd.serve_forever(poll_interval=0.5) except Exception: - pass + traceback.print_exc() t = threading.Thread(target=_run, name="DocsSite", daemon=True) t.start() - print(f"[DocsSite] Listening on http://{host}:{port} (title='{title}')") - - -# ----------------------------- -# Cog setup -# ----------------------------- + print(f"[DocsSite] Listening on http://{host}:{port} (title='{title}', offline={_DocsHandler.force_ready})") class DocsSite(commands.Cog): - """Tiny Swagger‑like docs website for bot commands.""" + """Tiny Swagger-like docs site for bot commands.""" def __init__(self, bot: commands.Bot): self.bot = bot @@ -457,12 +1041,10 @@ class DocsSite(commands.Cog): self.host = r.get("docs_host", "0.0.0.0") # SHAI_DOCS_HOST self.port = r.int("docs_port", 8910) # SHAI_DOCS_PORT self.title = r.get("docs_title", "ShaiWatcher Commands") # SHAI_DOCS_TITLE - - # Start server immediately; safe: ThreadingHTTPServer _start_server(self.bot, self.host, self.port, self.title) - # No commands; this cog only serves HTTP. - + def force_ready(self, value: bool = True): + _DocsHandler.force_ready = bool(value) async def setup(bot: commands.Bot): await bot.add_cog(DocsSite(bot))