"""Markdown for the tickets Gandalf opens in Tinker Tickets, and the notes it adds to them. Pure formatting only: monitor.py gathers the facts and passes them in. """ from datetime import datetime, timezone from typing import List, Optional, Sequence, Tuple from urllib.parse import quote Section = Tuple[str, List[str]] def duration(seconds) -> str: seconds = max(0, int(seconds)) if seconds < 60: return f'{seconds} s' minutes = seconds // 60 if minutes < 60: return f'{minutes} min' hours, minutes = divmod(minutes, 60) if hours < 48: return f'{hours} h {minutes} min' if minutes else f'{hours} h' days, hours = divmod(hours, 24) return f'{days} d {hours} h' if hours else f'{days} d' def utc(dt: Optional[datetime]) -> str: """Database times are naive UTC.""" if dt is None: return 'unknown' if dt.tzinfo is not None: dt = dt.astimezone(timezone.utc).replace(tzinfo=None) return dt.strftime('%Y-%m-%d %H:%M UTC') def speed(mbps) -> str: if not mbps: return '' mbps = int(mbps) if mbps >= 1000: gbps = mbps / 1000 return f'{gbps:g} Gbps' return f'{mbps} Mbps' def device_url(base: str, name: str, port_idx=None) -> str: if not base or not name: return '' url = f'{base}/device/{quote(name, safe="")}' return f'{url}#port-{port_idx}' if port_idx not in (None, '') else url def history_url(base: str, target: str) -> str: return f'{base}/events?target={quote(target, safe="")}&range=90d' if base and target else '' def cutsheet_url(base: str, cable_id) -> str: return f'{base}/cutsheet#cable-{cable_id}' if base else '' def link(text: str, url: str) -> str: return f'[{text}]({url})' if url else text def port_text(switch: str, port_name: str, port: dict) -> str: """'8SFP-AGG port 7 "Large1": up at 10 Gbps, DAC'""" idx = port.get('port_idx') text = f'{switch} port {idx}' if idx is not None else f'{switch} {port_name}' if port_name and idx is not None and port_name != f'Port {idx}': text += f' "{port_name}"' state = [] if port.get('up'): rate = speed(port.get('speed_mbps')) state.append(f'up at {rate}' if rate else 'up') else: state.append('down') if port.get('poe_power'): state.append(f'PoE {float(port["poe_power"]):.1f} W') sfp = port.get('sfp') or {} if sfp.get('found'): kind = 'DAC' if sfp.get('kind') == 'dac' else sfp.get('kind', '') module = ' '.join(p for p in (kind, sfp.get('part', '')) if p) if module: state.append(f'module: {module}') if sfp.get('rx_fault') or sfp.get('tx_fault'): state.append('**module reports a fault**') return f'{text}: {", ".join(state)}' def description(heading: str, facts: Sequence[Tuple[str, str]], summary: str, checks: Sequence[str], sections: Sequence[Section] = (), recovers: bool = True) -> str: lines = [f'## {heading}', ''] lines += [f'- **{key}:** {value}' for key, value in facts if value not in (None, '')] lines += ['', summary.strip()] if checks: lines += ['', '### What to check'] lines += [f'{n}. {check}' for n, check in enumerate(checks, 1)] for title, body in sections: if body: lines += ['', f'### {title}'] + list(body) if recovers: lines += ['', '_Opened automatically by Gandalf. It adds a note here when the problem clears, ' 'but leaves closing the ticket to you._'] else: lines += ['', '_Reported automatically by Gandalf. This already happened, so there will be no ' 'recovery note._'] return '\n'.join(lines) def history_lines(stats: dict, now: datetime) -> List[str]: week, month = stats.get('last_7d', 0), stats.get('last_30d', 0) if not month: lines = ['- First time in the last 30 days.'] else: lines = [f'- Happened {week} time{"s" if week != 1 else ""} in the last 7 days ' f'and {month} in the last 30 (not counting this one).'] if week >= 3: lines.append('- **This keeps coming back**: look for a cause that comes and goes ' '(a loose cable, a failing optic, power, or a scheduled job).') first, resolved = stats.get('last_first_seen'), stats.get('last_resolved_at') if resolved: took = duration((resolved - first).total_seconds()) if first else 'unknown' ago = duration((now - resolved).total_seconds()) lines.append(f'- Last cleared {utc(resolved)} ({ago} ago), after {took}.') return lines ISSUE_FIXED = { 'cable_mismatch': 'The cable now matches the cutsheet (or the cutsheet was updated).', 'monitor_source_unavailable': 'Prometheus is returning data again; host and interface alerting has resumed.', } def recovery_comment(event: dict, now: datetime, url: str = '') -> str: """The note Gandalf adds to a ticket when its alert clears.""" first, resolved = event.get('first_seen'), event.get('resolved_at') or now target = event['target_name'] + (f' {event["target_detail"]}' if event.get('target_detail') else '') lines = [f'**✅ Cleared**: {event["event_type"].replace("_", " ")} on {target}.', ''] lines.append(f'- **Cleared:** {utc(resolved)}') if first: lines.append(f'- **Lasted:** {duration((resolved - first).total_seconds())} ' f'(first seen {utc(first)})') if event.get('consecutive_failures'): lines.append(f'- **Failed checks:** {event["consecutive_failures"]}') if url: lines.append(f'- **Gandalf:** {url}') lines += ['', ISSUE_FIXED.get(event['event_type'], 'Gandalf no longer sees the problem.')] lines += ['', '_Gandalf leaves this ticket open. Close it once you know why it happened, ' 'or leave it open if it needs follow-up._'] return '\n'.join(lines) def incident_comment(name: str, summary: str, checks: Sequence[str], when: str) -> str: """A further report that belongs to the incident this ticket is already about.""" lines = [f'**Also seen: {name}** ({when})', '', summary.strip()] if checks: lines += ['', '**What to check**'] + [f'{n}. {check}' for n, check in enumerate(checks, 1)] lines += ['', '_Gandalf added this here instead of opening another ticket, since it is part of ' 'the same incident._'] return '\n'.join(lines) def update_comment(label: str, when: str, lookback_min: int, hold_min: int) -> str: """Note on a ticket that a UniFi update finished around the time it was opened.""" return '\n'.join([ f'**UniFi update:** {label} finished at {when}.', '', f'This ticket was opened within {lookback_min} minutes of that update, which UniFi logs once ' f'it is done. Updating the Network app or the gateway firmware can drop the WAN, and Smart ' f'Power may then power-cycle the modem, so the update is the likely cause.', '', f'_Gandalf holds new modem and WAN tickets for {hold_min} minutes after an update; anything ' f'still wrong after that gets its own ticket._', ])