mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekay
d696a015aa
The rename runs on the way in and nothing checked it afterwards, so paths and content drifted apart in three places, and each one broke something quietly. The script language had three names at once. Files on disk carried `.iql`/`.iqle`, the source matchers said `.script`, and the bytecode matchers said `.hoge`, so the test harness globbed zero files and the STL compiler regenerated an empty table. It is `script` now, in one place: `.script` source, `.scriptc` bytecode, `bin/script` to run one and `bin/scriptc` to compile one. 190 TypeScript specifiers named files that had been renamed away, which is the whole CDP consumer graph -- the event worker, not the web boot. `--check` only ever read Python imports; it reads both module systems now, and repoints both. `-hog-` became `-mascot-` in paths while content said `script`, so the code review product's skill directories stopped matching the prefix that discovers them and it ran with no perspectives at all. PATH_REGEX is CONTENT_REGEX -- the same list, not a copy -- so a rule can no longer land in one and miss the other, and the four disagreeing `hog` path rules collapse into the one word rule content already had. `HOGLI_*` becomes `INSIGHTSCLI_*` at every reader and writer, with no alias: an env var read under two names is two names. The seven casings of `hogql` and five of `hogvm` collapse into one rule per prefix casing, which leaves the suffix alone because `QL` is SQL's and `VM` is a virtual machine's. `--check` now refuses every retired spelling, in content and in paths, and draws the line the rules draw: `insights_hogfunction` is a Postgres table so it is left alone, and a migration filename is its identity in `django_migrations` so it is frozen rather than failed. Third-party names we cannot rename are reported as pending, never exempted.
161 lines
4.4 KiB
Python
161 lines
4.4 KiB
Python
"""Post-command contextual hints for insightscli CLI.
|
|
|
|
Occasionally reminds developers to run maintenance commands like
|
|
``insightscli doctor``, ``insightscli doctor:disk``, etc.
|
|
|
|
Suppression:
|
|
INSIGHTSCLI_NO_HINTS=1 -> disable hints
|
|
CI=* -> disable hints (same as telemetry)
|
|
|
|
State file: ~/.config/insights/insightscli_hints.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import TypedDict
|
|
|
|
import click
|
|
|
|
# Commands that should never trigger a trailing hint
|
|
_SKIP_COMMANDS = frozenset(
|
|
{
|
|
"doctor",
|
|
"doctor:disk",
|
|
"doctor:zombies",
|
|
"doctor:report",
|
|
"telemetry:off",
|
|
"telemetry:on",
|
|
"telemetry:status",
|
|
"meta:check",
|
|
"meta:concepts",
|
|
"help",
|
|
"quickstart",
|
|
}
|
|
)
|
|
|
|
|
|
class _HintDef:
|
|
__slots__ = ("command", "threshold_days", "message")
|
|
|
|
def __init__(self, command: str, threshold_days: int, message: str) -> None:
|
|
self.command = command
|
|
self.threshold_days = threshold_days
|
|
self.message = message
|
|
|
|
|
|
_HINT_DEFS: tuple[_HintDef, ...] = (
|
|
_HintDef("doctor:disk", 7, "Run `insightscli doctor:disk` to check for disk bloat"),
|
|
_HintDef("doctor:zombies", 7, "Run `insightscli doctor:zombies` to find orphaned processes"),
|
|
_HintDef("doctor", 14, "Run `insightscli doctor` for a quick health check"),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# State persistence (mirrors telemetry.py pattern)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class HintsState(TypedDict, total=False):
|
|
last_hint_shown: str # ISO 8601
|
|
last_runs: dict[str, str] # command -> ISO 8601
|
|
|
|
|
|
def _get_state_path() -> Path:
|
|
return Path.home() / ".config" / "insights" / "insightscli_hints.json"
|
|
|
|
|
|
def _load_state() -> HintsState:
|
|
try:
|
|
return json.loads(_get_state_path().read_text())
|
|
except Exception:
|
|
return HintsState()
|
|
|
|
|
|
def _save_state(state: HintsState) -> None:
|
|
try:
|
|
path = _get_state_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(state, indent=2) + "\n")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def record_check_run(command: str) -> None:
|
|
"""Record that a doctor command was run (updates hint cooldown)."""
|
|
state = _load_state()
|
|
last_runs = state.get("last_runs", {})
|
|
last_runs[command] = datetime.now(UTC).isoformat()
|
|
state["last_runs"] = last_runs
|
|
_save_state(state)
|
|
|
|
|
|
def maybe_show_hint(command: str | None) -> None:
|
|
"""Evaluate triggers and maybe print a hint to stderr. Never raises."""
|
|
try:
|
|
_maybe_show_hint(command)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _maybe_show_hint(command: str | None) -> None:
|
|
if os.environ.get("INSIGHTSCLI_NO_HINTS") == "1":
|
|
return
|
|
if os.environ.get("CI"):
|
|
return
|
|
if command in _SKIP_COMMANDS:
|
|
return
|
|
|
|
state = _load_state()
|
|
now = datetime.now(UTC)
|
|
|
|
# Rate limit: max 1 hint per 24 hours
|
|
last_shown = state.get("last_hint_shown")
|
|
if last_shown:
|
|
try:
|
|
last_dt = datetime.fromisoformat(last_shown)
|
|
if now - last_dt < timedelta(hours=24):
|
|
return
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
hint = _pick_hint(state, now)
|
|
if hint is None:
|
|
return
|
|
|
|
click.secho(f"\n Hint: {hint}", fg="bright_black", err=True)
|
|
|
|
state["last_hint_shown"] = now.isoformat()
|
|
_save_state(state)
|
|
|
|
|
|
def _pick_hint(state: HintsState, now: datetime) -> str | None:
|
|
"""Return the first applicable hint message, or None."""
|
|
last_runs = state.get("last_runs", {})
|
|
|
|
for hint_def in _HINT_DEFS:
|
|
last_run = last_runs.get(hint_def.command)
|
|
if last_run is None:
|
|
return hint_def.message
|
|
try:
|
|
last_dt = datetime.fromisoformat(last_run)
|
|
if now - last_dt > timedelta(days=hint_def.threshold_days):
|
|
return hint_def.message
|
|
except (ValueError, TypeError):
|
|
return hint_def.message
|
|
|
|
return None
|