From 6c1c1a11ed698050a8a86aeafceed5497dce76e8 Mon Sep 17 00:00:00 2001 From: host Date: Wed, 29 Jul 2026 20:06:17 +0300 Subject: [PATCH] =?UTF-8?q?v2.5.1=20=E2=80=94=20=D1=84=D0=B8=D0=BA=D1=81?= =?UTF-8?q?=20=D1=84=D0=B8=D0=BB=D1=8C=D1=82=D1=80=D0=B0=20=D0=BE=D0=BA?= =?UTF-8?q?=D0=BE=D0=BD/=D0=B4=D0=B2=D0=B5=D1=80=D0=B5=D0=B9:=20=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=BE=D0=B3=D0=BE=20=D0=BF=D0=BE=20device=5Fclass?= =?UTF-8?q?=20(door/window/garage=5Fdoor/lock)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- home_assistant_tools.py | 197 +++++++++------------------------------- 1 file changed, 41 insertions(+), 156 deletions(-) diff --git a/home_assistant_tools.py b/home_assistant_tools.py index 2fecf0a..102d13b 100644 --- a/home_assistant_tools.py +++ b/home_assistant_tools.py @@ -2,7 +2,7 @@ title: Home Assistant Smart Control description: Управление умным домом через Home Assistant REST API с интерпретацией естественного языка. author: ChatGPT (refactored by MiniMax-M3) -version: 2.5.0 +version: 2.5.1 license: MIT """ @@ -17,41 +17,20 @@ logger = logging.getLogger(__name__) class Tools: - # ======================================================================== - # Конфигурация (Valves - задаётся в UI Open WebUI) - # ======================================================================== class Valves(BaseModel): - ha_url: str = Field( - default="https://ha.nasweb.ru", - description="URL Home Assistant (например https://ha.nasweb.ru)", - ) - ha_token: str = Field( - default="", - description="Long-Lived Access Token из профиля HA", - ) - verify_ssl: bool = Field( - default=False, - description="Проверять SSL-сертификат HA (false для самоподписанных)", - ) + ha_url: str = Field(default="https://ha.nasweb.ru", description="URL Home Assistant") + ha_token: str = Field(default="", description="Long-Lived Access Token из профиля HA") + verify_ssl: bool = Field(default=False, description="Проверять SSL-сертификат HA") request_timeout: int = Field(default=10, description="Таймаут HTTP-запроса, сек") max_retries: int = Field(default=2, description="Количество ретраев при сетевых сбоях") cache_ttl: int = Field(default=5, description="TTL кэша состояний HA, сек") - confirm_global_actions: bool = Field( - default=True, - description="Спрашивать подтверждение для команд на ВСЕ сущности домена", - ) - dry_run: bool = Field( - default=False, - description="Не выполнять команды, только показывать что БУДЕТ сделано", - ) + confirm_global_actions: bool = Field(default=True, description="Спрашивать подтверждение для команд на ВСЕ сущности домена") + dry_run: bool = Field(default=False, description="Не выполнять команды, только показывать что БУДЕТ сделано") def __init__(self): self.valves = self.Valves() self._states_cache: Optional[Tuple[float, List[Dict[str, Any]]]] = None - # ======================================================================== - # Стоп-слова и ключевые слова - # ======================================================================== _STOP_WORDS = { "включи", "выключи", "поставь", "установи", "сделай", "запусти", "на", "в", "во", "мне", "пожалуйста", "давай", "ну", @@ -99,9 +78,11 @@ class Tools: _MONITOR_DOMAINS = { "window_door": { "domains": ["binary_sensor"], - "keywords": ["окн", "двер", "форточ", "калитк", "ворота", "гараж"], "open_states": ("open", "on", "unlocked", "detected"), "closed_states": ("closed", "off", "locked", "clear"), + # Допустимые device_class для окон/дверей/замков. + # Если в HA не задан device_class — сущность игнорируется. + "valid_device_classes": ("door", "window", "garage_door", "lock"), }, "light_status": { "domains": ["light"], @@ -136,9 +117,6 @@ class Tools: }, } - # ======================================================================== - # HTTP / API - # ======================================================================== def _headers(self) -> Dict[str, str]: return { "Authorization": "Bearer " + self.valves.ha_token, @@ -146,7 +124,6 @@ class Tools: } def _request(self, method: str, path: str, **kwargs) -> requests.Response: - """HTTP-запрос с ретраями и логированием.""" url = self.valves.ha_url + path last_exc: Optional[Exception] = None kwargs.setdefault("verify", self.valves.verify_ssl) @@ -171,7 +148,6 @@ class Tools: raise RuntimeError("Home Assistant недоступен: " + str(last_exc)) def _get_states(self, force: bool = False) -> List[Dict[str, Any]]: - """Получить состояния HA с TTL-кэшем.""" now = time.time() if ( not force @@ -186,7 +162,6 @@ class Tools: return states def _call_service(self, domain: str, service: str, data: Dict[str, Any]) -> bool: - """Вызов сервиса HA. Возвращает True если выполнено, False если dry-run.""" logger.info( "%s -> %s.%s %s", "[DRY] " if self.valves.dry_run else "", @@ -197,17 +172,12 @@ class Tools: self._request("POST", "/api/services/" + domain + "/" + service, json=data) return True - # ======================================================================== - # Поиск сущностей - # ======================================================================== def _tokenize(self, text: str) -> List[str]: - """Токенизация: фильтр стоп-слов, приведение к нижнему регистру.""" text = text.lower().strip().replace("ё", "е") tokens = re.findall(r"[а-яa-z0-9]+", text) return [t for t in tokens if t not in self._STOP_WORDS and len(t) >= 2] def _entity_text(self, s: Dict[str, Any]) -> str: - """Полный текст сущности: id + friendly_name + area.""" attrs = s.get("attributes", {}) or {} return " ".join([ s.get("entity_id", ""), @@ -222,10 +192,8 @@ class Tools: domain: Optional[str] = None, require_all: bool = False, ) -> List[str]: - """Поиск сущностей по ключевым словам.""" states = self._get_states() result: List[str] = [] - for s in states: eid = s["entity_id"] if domain and not eid.startswith(domain + "."): @@ -238,22 +206,18 @@ class Tools: matches = sum(1 for kw in keywords if kw in haystack) if (require_all and matches == len(keywords)) or (not require_all and matches > 0): result.append(eid) - return result def _get_state(self, entity_id: str) -> Optional[str]: - """Получить текущее состояние конкретной сущности.""" for s in self._get_states(): if s["entity_id"] == entity_id: return s.get("state") return None def _attr(self, s: Dict[str, Any], key: str, default=None) -> Any: - """Безопасное извлечение атрибута.""" return (s.get("attributes", {}) or {}).get(key, default) def _resolve_domain(self, cmd: str) -> Optional[str]: - """Определить домен HA по русской команде.""" for domain, aliases in self._DOMAIN_ALIASES.items(): for a in aliases: if a in cmd: @@ -261,7 +225,6 @@ class Tools: return None def _find_entity_by_name(self, name: str, domain: Optional[str] = None) -> Optional[str]: - """Найти первый entity_id, у которого friendly_name или id совпадает с name.""" states = self._get_states(force=True) name_lower = name.lower().strip() for s in states: @@ -271,16 +234,11 @@ class Tools: fn = (self._attr(s, "friendly_name", "") or "").lower() if fn == name_lower: return eid - # Частичное совпадение if name_lower in fn or name_lower in eid.lower(): return eid return None - # ======================================================================== - # Парсинг чисел - # ======================================================================== def _extract_number(self, text: str, default: Optional[float] = None) -> Optional[float]: - """Извлечь число из текста: арабские цифры или русские слова.""" m = re.search(r"-?\d+(?:[.,]\d+)?", text) if m: return float(m.group(0).replace(",", ".")) @@ -290,11 +248,7 @@ class Tools: return float(self._RU_NUMS[word]) return default - # ======================================================================== - # Форматирование ответа - # ======================================================================== def _format_entity(self, s: Dict[str, Any]) -> str: - """Красиво отформатировать одну сущность для вывода.""" attrs = s.get("attributes", {}) or {} fn = attrs.get("friendly_name", s["entity_id"]) state = s.get("state", "?") @@ -307,17 +261,24 @@ class Tools: # Мониторинг # ======================================================================== def _monitor_windows_doors(self, tokens: List[str]) -> List[str]: + """ + Какие окна/двери открыты. + + Строгий фильтр: только binary_sensor с device_class в + (door, window, garage_door, lock). Это исключает ложные + срабатывания на датчики движения, батареи штор, шторы Aqara и т.п. + """ cfg = self._MONITOR_DOMAINS["window_door"] + valid_dc = cfg["valid_device_classes"] states = self._get_states() + items: List[Tuple[Dict[str, Any], bool, bool]] = [] for s in states: eid = s["entity_id"] - is_relevant_domain = eid.startswith("binary_sensor.") + if not eid.startswith("binary_sensor."): + continue device_class = self._attr(s, "device_class", "") - is_door_window = device_class in ("door", "window", "garage_door", "lock") - text = self._entity_text(s) - has_keyword = any(k in text for k in cfg["keywords"]) - if not ((is_relevant_domain and is_door_window) or has_keyword): + if device_class not in valid_dc: continue cur = s.get("state", "") is_open = cur in cfg["open_states"] @@ -325,7 +286,7 @@ class Tools: items.append((s, is_open, is_closed)) if not items: - return ["датчики окон/дверей не найдены"] + return ["датчики окон/дверей не найдены (проверьте device_class в HA)"] open_items = [s for s, is_open, _ in items if is_open] closed_items = [s for s, _, is_closed in items if is_closed] @@ -340,6 +301,7 @@ class Tools: lines.append("Закрыто: " + str(len(closed_items)) + " шт.") if unknown: lines.append("Неизвестно: " + str(len(unknown)) + " шт.") + if tokens: filtered = [] for line in lines: @@ -347,6 +309,7 @@ class Tools: filtered.append(line) if filtered: return filtered + return lines def _monitor_lights(self, tokens: List[str]) -> List[str]: @@ -549,7 +512,6 @@ class Tools: # Главная точка входа # ======================================================================== def smart(self, command: str) -> str: - """Принимает команду на естественном языке и управляет домом / отвечает на вопросы.""" if not self.valves.ha_token: return ( "Не задан HA Token. Укажите его в настройках инструмента (Valves) " @@ -562,7 +524,6 @@ class Tools: warnings: List[str] = [] try: - # Маршрутизация 'покажи все / список устройств' -> list_devices if any(w in cmd for w in [ "покажи", "список устройств", "какие устройств", "что есть", "инвентар", "все устройств", @@ -570,14 +531,10 @@ class Tools: target = self._resolve_domain(cmd) return self.list_devices(domain=target or "", limit=80) - # Общий отчёт if "отчет" in cmd or "отчёт" in cmd or "обзор" in cmd or "все вместе" in cmd: return "\n".join(self._monitor_full_report()) - # Переименование if "переимен" in cmd: - # Ожидаемый формат: "переименуй в <новое имя>" - # или: "переименуй как <новое имя>" m = re.search(r"переимен\w*\s+(.+?)\s+(?:в|как|на)\s+(.+)$", cmd) if m: target = m.group(1).strip().strip('"').strip("'") @@ -592,7 +549,6 @@ class Tools: "Пример: `переименуй automation.сертификат в Обновление SSL`" ) - # Включение/выключение автоматизации if "автоматизац" in cmd and any(w in cmd for w in ("включ", "выключ", "активир", "деактивир")): target = "" for s in self._get_states(force=True): @@ -694,7 +650,6 @@ class Tools: ) -> List[str]: actions: List[str] = [] - # Свет if "свет" in cmd or "лампочк" in cmd: entities = self._find_entities(tokens, "light") if not entities: @@ -742,7 +697,6 @@ class Tools: label = "выключен" if turn_off else "включён" actions.append("свет " + label + " (" + str(count) + " шт.)") - # Розетки / реле / выключатели if any(k in cmd for k in ["розетк", "выключател", "устройств", "реле"]): entities = self._find_entities(tokens, "switch") if not entities: @@ -766,7 +720,6 @@ class Tools: label = "выключены" if turn_off else "включены" actions.append("розетки " + label + " (" + str(count) + " шт.)") - # Климат if any(k in cmd for k in ["температур", "градус", "тепл", "холод", "климат"]): temp = self._extract_number(cmd, default=22) entities = self._find_entities(tokens, "climate") @@ -801,7 +754,6 @@ class Tools: ) actions.append("режим охлаждения") - # Медиа if any(k in cmd for k in ["музык", "медиа", "плеер", "колонк"]): entities = self._find_entities(tokens, "media_player") if not entities: @@ -845,7 +797,6 @@ class Tools: if count: actions.append("воспроизведение (" + str(count) + " шт.)") - # Шторы if any(k in cmd for k in ["штор", "жалюз", "роллет", "гардин"]): entities = self._find_entities(tokens, "cover") if not entities: @@ -863,7 +814,6 @@ class Tools: label = "закрыты" if is_close else "открыты" actions.append("шторы " + label + " (" + str(count) + " шт.)") - # Замки if any(k in cmd for k in ["замок", "заблокир", "закрыт"]): entities = self._find_entities(tokens, "lock") if entities: @@ -878,7 +828,6 @@ class Tools: "замки " + ("закрыты" if is_lock else "открыты") + " (" + str(len(entities)) + " шт.)" ) - # Вентиляторы if "вентилятор" in cmd or "кулер" in cmd: entities = self._find_entities(tokens, "fan") if entities: @@ -893,7 +842,6 @@ class Tools: "вентиляторы " + ("выключены" if is_off else "включены") + " (" + str(len(entities)) + " шт.)" ) - # Пылесосы if "пылесос" in cmd or "робот" in cmd: entities = self._find_entities(tokens, "vacuum") if entities: @@ -910,7 +858,6 @@ class Tools: self._call_service("vacuum", "start", {"entity_id": e}) actions.append("пылесос уборка") - # Кнопки if "кнопк" in cmd and any(w in cmd for w in ("нажм", "кликн", "тапн", "триггерн", "активир")): entities = self._find_entities(tokens, "button") if entities: @@ -922,7 +869,6 @@ class Tools: else: warnings.append("кнопки не найдены") - # Сцены / скрипты if any(k in cmd for k in ["сцен", "режим", "запуст"]): all_scripts = self._find_entities([], "script") matched: List[str] = [] @@ -942,7 +888,7 @@ class Tools: return actions # ======================================================================== - # Диагностика / справочник устройств (отдельный инструмент) + # Диагностика / справочник устройств # ======================================================================== def list_devices( self, @@ -952,30 +898,8 @@ class Tools: show_unavailable: bool = True, limit: int = 50, ) -> str: - """ - Получить список устройств из Home Assistant с фильтрами. - - Используется для инвентаризации: какие устройства есть, в каком состоянии, - какие из них offline. Полезно для отладки и проверки интеграций. - - Args: - domain: фильтр по домену (light, switch, binary_sensor, sensor, climate, ...). - Спецзначение "window_door" — окна и двери. - Пусто — все домены. - device_class: фильтр по device_class атрибуту - (door, window, motion, battery, temperature, ...). - Пусто — все классы. - only_with_state: оставить только с указанным состоянием (on/off/open/closed/...). - Пусто — любые. - show_unavailable: показывать ли устройства со state="unavailable"/"unknown". - limit: максимум устройств в ответе. - - Returns: - Список устройств в формате markdown, сгруппированный по доменам. - """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." - try: states = self._get_states(force=True) except Exception as e: @@ -994,9 +918,10 @@ class Tools: if domain: if domain == "window_door": - if dc not in ("door", "window", "garage_door", "lock") and \ - not any(k in (eid + " " + (attrs.get("friendly_name", "") or "").lower()) - for k in ["окн", "двер", "форточ", "балкон", "калитк", "ворота", "гараж"]): + # Строгий фильтр: только binary_sensor с правильным device_class + if not eid.startswith("binary_sensor."): + continue + if dc not in ("door", "window", "garage_door", "lock"): continue else: if not eid.startswith(domain + "."): @@ -1038,7 +963,7 @@ class Tools: return "\n".join(lines) # ======================================================================== - # 🏷 Переименование сущностей + # Переименование # ======================================================================== def rename_entity( self, @@ -1046,21 +971,6 @@ class Tools: name: str = "", friendly_name: str = "", ) -> str: - """ - Переименовать сущность в Home Assistant (меняет friendly_name). - Работает для automation, script, light, switch, scene и др. - - Можно указать либо entity_id (точное имя), либо friendly_name (для поиска). - - Args: - entity_id: точный id сущности (например "automation.сертификат"). - Если пусто — ищем по friendly_name. - name: новое имя. - friendly_name: текущее friendly_name для поиска (если entity_id не указан). - - Returns: - Подтверждение или сообщение об ошибке. - """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." if not name: @@ -1068,14 +978,13 @@ class Tools: target = entity_id if not target or "." not in target: - # Поиск по friendly_name search_name = friendly_name or entity_id target = self._find_entity_by_name(search_name) if not target: return "Не нашёл сущность с именем: " + search_name if self.valves.dry_run: - return f"[DRY-RUN] Переименовал бы {target} -> {name}" + return "[DRY-RUN] Переименовал бы " + target + " -> " + name try: self._request( @@ -1083,13 +992,13 @@ class Tools: "/api/services/homeassistant/update_entity", json={"entity_id": target, "name": name}, ) - self._get_states(force=True) # сброс кэша - return f"Переименовано: {target} -> '{name}'" + self._get_states(force=True) + return "Переименовано: " + target + " -> '" + name + "'" except Exception as e: - return f"Ошибка переименования {target}: {e}" + return "Ошибка переименования " + target + ": " + str(e) # ======================================================================== - # ⚙️ Управление автоматизациями + # Управление автоматизациями # ======================================================================== def toggle_automation( self, @@ -1097,18 +1006,6 @@ class Tools: friendly_name: str = "", turn_off: bool = False, ) -> str: - """ - Включить или выключить автоматизацию. - - Args: - entity_id: точный id (например "automation.сертификат"). - Если пусто — ищем по friendly_name. - friendly_name: имя для поиска. - turn_off: True — выключить, False — включить. - - Returns: - Подтверждение или сообщение об ошибке. - """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." @@ -1123,7 +1020,7 @@ class Tools: action_label = "выключена" if turn_off else "включена" if self.valves.dry_run: - return f"[DRY-RUN] Автоматизация {target} была бы {action_label}" + return "[DRY-RUN] Автоматизация " + target + " была бы " + action_label try: self._request( @@ -1132,9 +1029,9 @@ class Tools: json={"entity_id": target}, ) self._get_states(force=True) - return f"Автоматизация {target} {action_label}" + return "Автоматизация " + target + " " + action_label except Exception as e: - return f"Ошибка: {e}" + return "Ошибка: " + str(e) def trigger_automation( self, @@ -1142,18 +1039,6 @@ class Tools: friendly_name: str = "", skip_condition: bool = True, ) -> str: - """ - Запустить автоматизацию вручную (триггернуть). - - Args: - entity_id: точный id (например "automation.сертификат"). - Если пусто — ищем по friendly_name. - friendly_name: имя для поиска. - skip_condition: пропустить ли условия (по умолчанию True). - - Returns: - Подтверждение или сообщение об ошибке. - """ if not self.valves.ha_token: return "Не задан HA Token (Valves)." @@ -1165,7 +1050,7 @@ class Tools: return "Не нашёл автоматизацию: " + search_name if self.valves.dry_run: - return f"[DRY-RUN] Триггернул бы {target}" + return "[DRY-RUN] Триггернул бы " + target try: data: Dict[str, Any] = {"entity_id": target} @@ -1176,6 +1061,6 @@ class Tools: "/api/services/automation/trigger", json=data, ) - return f"Триггер запущен: {target}" + return "Триггер запущен: " + target except Exception as e: - return f"Ошибка: {e}" \ No newline at end of file + return "Ошибка: " + str(e) \ No newline at end of file