meshcore-bot/modules/commands/evac_command.py

#!/usr/bin/env python3
"""List Watch Duty evacuation orders and warnings for a fire."""

from __future__ import annotations

import asyncio
from typing import Any, Optional

from .. import watchduty_poll
from ..models import MeshMessage
from .base_command import BaseCommand


class EvacCommand(BaseCommand):
    """Interactive Watch Duty evacuation orders/warnings command."""

    name = "evac"
    keywords = ["evac", "evacs"]
    description = (
        "List evacuation orders and warnings "
        "(usage: evac [<# from evac list|Watch Duty id|name>] [item #])"
    )
    category = "info"
    requires_internet = True
    cooldown_seconds = 5

    short_description = "Show Watch Duty evacuation orders and warnings"
    usage = "evac [<# from evac list|Watch Duty id|name>] [item #]"
    examples = ["evac Woods Fire", "evac 1", "evac 93683", "evac 93817 2"]
    parameters = [
        {
            "name": "query",
            "description": "Optional list number, Watch Duty id, or fire name",
        },
        {
            "name": "item",
            "description": "Optional evacuation item number from the detail list",
        },
    ]

    settings_schema = [
        {
            "key": "enabled",
            "label": "Enabled",
            "type": "bool",
            "default": False,
            "help": "Enable the evac command (opt-in).",
        },
        {
            "key": "include_prescribed",
            "label": "Include prescribed burns",
            "type": "bool",
            "default": False,
            "help": "When true, include prescribed burns when resolving fires.",
        },
    ]

    def __init__(self, bot: Any) -> None:
        super().__init__(bot)
        self._enabled = self.get_config_value(
            "Evac_Command", "enabled", fallback=False, value_type="bool"
        )
        include = self.get_config_value(
            "Evac_Command", "include_prescribed", fallback=None, value_type="bool"
        )
        if include is None:
            include = self.get_config_value(
                "WatchDuty_Service",
                "include_prescribed",
                fallback=False,
                value_type="bool",
            )
        self._include_prescribed = bool(include)

    def can_execute(self, message: MeshMessage) -> bool:
        if not self._enabled:
            return False
        return super().can_execute(message)

    def _args_tail(self, message: MeshMessage) -> str:
        parts = message.content.strip().split(None, 1)
        if not parts:
            return ""
        kws = {x.lower() for x in self.keywords}
        if parts[0].lower() not in kws:
            return ""
        return parts[1].strip() if len(parts) > 1 else ""

    @staticmethod
    def _parse_event_and_item_query(tail: str) -> tuple[str, Optional[int]]:
        """Parse ``evac`` args into event query and optional item number."""
        text = (tail or "").strip()
        if not text:
            return "", None
        parts = text.split()
        if len(parts) >= 2 and parts[-1].isdigit():
            item_n = int(parts[-1])
            if item_n >= 1:
                return " ".join(parts[:-1]).strip(), item_n
        return text, None

    async def execute(self, message: MeshMessage) -> bool:
        tail = self._args_tail(message)
        try:
            events = await asyncio.to_thread(
                watchduty_poll.fetch_active_geo_events_for_user_query,
                self.bot.config,
                include_prescribed=self._include_prescribed,
            )
        except Exception as e:
            self.logger.error("evac command: fetch failed: %s", e)
            return await self.send_response(
                message, "Could not load fires (Watch Duty)."
            )

        evac_events = [e for e in events if watchduty_poll.incident_has_evac_info(e)]
        event_query, item_n = self._parse_event_and_item_query(tail)
        if not event_query:
            max_len = self.get_max_message_length(message)
            if not evac_events:
                return await self.send_response(
                    message,
                    "No active fires with evacuation info right now.",
                )
            lines = [f"Fires with evacuations ({len(evac_events)}):"]
            for i, event in enumerate(evac_events, start=1):
                name = (event.get("name") or f"Event {event.get('id')}").strip()
                loc = watchduty_poll.format_location_short(event)
                eid = event.get("id")
                id_part = f" · {eid}" if eid is not None else ""
                lines.append(f"{i}. {name} ({loc}){id_part}")
            chunks = watchduty_poll.mesh_pack_lines(lines, max_len)
            if len(chunks) == 1:
                return await self.send_response(message, chunks[0])
            return await self.send_response_chunked(message, chunks)

        event, err = await asyncio.to_thread(
            watchduty_poll.resolve_active_event_by_query,
            events,
            event_query,
            config=self.bot.config,
            include_prescribed=self._include_prescribed,
            numeric_index_list=evac_events,
        )
        if err:
            if err == "usage":
                return await self.send_response(
                    message,
                    "Usage: evac [<# from evac list|Watch Duty id|name>] [item #] — "
                    "list #s match evac with no args, not fire.",
                )
            return await self.send_response(message, err)

        assert event is not None
        eid = event.get("id")
        if eid is None:
            return await self.send_response(message, "Invalid event (missing id).")

        detail = await asyncio.to_thread(watchduty_poll.fetch_event_detail, int(eid))
        if not detail:
            detail = event
        name = (detail.get("name") or f"Event {eid}").strip()

        lines_body = watchduty_poll.evacuation_display_lines(detail)
        max_len = self.get_max_message_length(message)

        if not lines_body:
            msg = f"No evacuation info listed for {name} on Watch Duty."
            return await self.send_response(message, msg[:max_len])

        if item_n is not None:
            if item_n > len(lines_body):
                return await self.send_response(
                    message,
                    f"Only {len(lines_body)} evacuation item(s) for {name}. "
                    f"Try: evac {eid}",
                )
            full_text = lines_body[item_n - 1]
            lines = [f"Evac — {name}:", f"{item_n}. {full_text}"]
        else:
            lines = [f"Evac — {name}:"]
            for i, line in enumerate(lines_body, start=1):
                snippet = watchduty_poll.first_sentence(line)
                if not snippet:
                    continue
                if snippet != line and len(snippet) < len(line):
                    snippet = snippet.rstrip() + " [...]"
                lines.append(f"{i}. {snippet}")

        chunks = watchduty_poll.mesh_pack_lines(lines, max_len)
        if len(chunks) == 1:
            return await self.send_response(message, chunks[0])
        return await self.send_response_chunked(message, chunks)