1. 项目概述与核心价值

最近在折腾智能家居的自动化场景,发现一个挺有意思的需求:如何让家里的智能音箱或者手机助手,每天准时提醒我“今天该扔哪种垃圾”?这听起来是个小事,但真要做起来,涉及到的环节还真不少。从获取官方的垃圾回收日历数据,到解析处理,再到接入主流的智能家居平台,每一步都有不少门道。正好,我在GitHub上看到了一个名为 openclaw-skill-abfallkalender-rv 的项目,它本质上是一个为开源语音助手 OpenClaw 开发的技能(Skill),专门用于查询德国地区(RV通常指特定区域)的垃圾回收日历。

这个项目标题拆开来看, openclaw 是语音助手平台, skill 是技能, abfallkalender 是德语“垃圾日历”, rv 可能指某个区域或垃圾处理公司。它的核心价值在于,将本地化、碎片化的公共服务信息(垃圾回收时间表)通过一个标准化的接口整合起来,让用户可以通过自然语音交互来查询,这比去翻纸质日历或者记在手机备忘录里要方便和智能得多。对于生活在德国,或者任何有类似固定垃圾回收制度的地区的朋友来说,如果能把这个功能集成到日常的智能家居流里,会大大提升生活便利性。接下来,我就结合这个项目的思路,以及我自己的实践,来详细拆解一下从零构建这样一个“语音垃圾日历”的完整流程、技术选型和避坑指南。

2. 整体方案设计与技术选型考量

2.1 核心需求与架构拆解

要实现“语音查询垃圾日历”,我们不能只盯着这一个技能本身,得把它放到一个完整的智能家居交互链条里来看。核心需求可以分解为以下几点:

  1. 数据源 :稳定、准确地获取目标区域的垃圾回收计划数据。这是所有功能的基石。
  2. 数据处理与存储 :将获取到的原始数据(可能是PDF、ICS日历文件、HTML或API JSON)进行解析、清洗,并转换成结构化的、易于查询的格式(如JSON),可能还需要本地缓存以避免频繁请求。
  3. 服务接口 :提供一个标准的、对内的数据查询接口。例如,一个RESTful API,接收地址或区域参数,返回今日/明日/本周的垃圾回收类型。
  4. 语音技能集成 :让语音助手平台(如OpenClaw、Home Assistant的Nabu Casa云、或自建的Rhasspy)能够调用上述接口,理解用户的查询意图(如“今天扔什么垃圾?”),并给出语音回复。
  5. 通知与自动化 :除了主动查询,更高级的需求是自动化的提醒。例如,在回收日的前一天晚上,通过智能音箱播报提醒,或在手机APP上发送推送通知。

基于这些需求,一个典型的架构会分为三层:

  • 数据层 :负责数据抓取、解析和持久化。可能涉及网络请求、PDF解析、日历文件处理等。
  • 服务层 :提供业务逻辑和API接口。用Python的Flask/FastAPI,或Node.js的Express等轻量框架实现。
  • 交互层 :即语音技能本身,负责意图识别、会话管理和调用服务层API。这需要遵循特定语音平台(如OpenClaw)的技能开发规范。

openclaw-skill-abfallkalender-rv 项目主要聚焦在“交互层”,即作为OpenClaw的一个技能模块。但我们要实现完整功能,必须补全数据层和服务层。

2.2 关键技术选型与理由

  1. 数据获取方式

    • 首选官方API :许多先进的垃圾处理公司或市政部门会提供标准的ICS(iCalendar)订阅链接或开放的JSON API。这是最稳定、最规范的方式。你需要去当地垃圾处理公司的官网(如 abfallkalender.rvxx.de 这类域名)寻找“Kalender herunterladen”或“iCal”之类的选项。
    • 次选网页抓取 :如果官方没有提供API,但有一个发布了日历的网页,那么可以考虑使用Python的 requests 库获取HTML,再用 BeautifulSoup lxml 进行解析。这种方式脆弱,一旦网页改版就需要调整代码。
    • 下策PDF解析 :有些地区只提供PDF版本的日历。这需要使用像 pdfplumber PyPDF2 这样的库来提取文本和表格,再通过规则匹配日期和垃圾类型。复杂度最高,稳定性最差。
    • 我的选择理由 :优先寻找ICS订阅。因为ICS是标准日历格式,可以被绝大多数日历应用和库(如Python的 icalendar 库)直接解析,且通常由官方维护,数据准确可靠。在德国,很多 Abfallwirtschaft (垃圾管理)公司都提供此服务。
  2. 后端服务框架

    • Python (FastAPI/Flask) :生态丰富,数据处理和解析库多(如 icalendar , beautifulsoup4 , pdfplumber ),开发速度快。FastAPI天生支持异步,适合IO密集型的网络请求和API响应,自动生成API文档也是一大优点。
    • Node.js (Express) :对于熟悉JavaScript生态的开发者也很友好,有 node-ical 这样的库可以解析ICS。
    • 我的选择理由 :我选择 Python + FastAPI 。主要考虑到后续的数据处理复杂性可能较高,Python在数据抓取、文本处理方面的库更成熟。FastAPI的异步特性在处理多个用户请求或同时抓取多个数据源时更有优势,而且其类型提示和自动文档对后期维护友好。
  3. 数据存储

    • 简单文件缓存 :如果数据更新频率低(如每月或每季度更新一次),可以将解析后的JSON数据直接保存为本地文件,并设置一个缓存过期时间。
    • 轻量级数据库 :如SQLite。适合需要存储多个区域数据、或需要记录历史查询的情况。
    • 我的选择理由 :对于个人或家庭使用, 文件缓存 足够了。我们可以在服务启动时检查缓存文件是否过期(比如是否超过24小时),如果过期则重新从源头抓取并解析数据,更新缓存文件。这样既避免了每次查询都去请求外部源(可能慢或有频率限制),实现也最简单。
  4. 语音平台集成

    • OpenClaw Skill :需要按照OpenClaw的技能开发规范来写,通常包括意图定义(Intents)、话语样本(Utterances)和技能处理逻辑。
    • Home Assistant :可以通过其“Conversation”组件集成自定义命令,或者开发一个完整的集成(Integration)。对于提醒,可以利用HA强大的自动化(Automation)和通知(Notify)系统。
    • 自建Rhasspy :完全本地化、隐私友好的方案,技能开发也有其特定格式(通常为Jinja2模板和Python脚本)。
    • 我的选择理由 :由于项目标题指向OpenClaw,我们将以 OpenClaw Skill 的形式作为主要交互入口进行讲解。但核心的服务层(API)是独立的,这意味着你可以用同一套API去适配Home Assistant或其它平台,只需编写对应的“客户端”即可,实现了关注点分离。

3. 数据层构建:获取与解析垃圾日历

3.1 寻找并确认数据源

这是最关键也是最容易卡住的一步。以德国为例,通常你需要知道你所在地的垃圾处理负责公司(如 AWB , RSAG , AVR 等)或其对应的行政区划代码。然后访问其官网。

实操步骤:

  1. 打开浏览器,搜索“ [你所在城市名] Abfallkalender ”或“ [垃圾处理公司名] Abfuhrtermine ”。
  2. 在官网上寻找“Abfuhrtermine”、“Abfallkalender”、“Kalender download”或“iCal-/ICS-Kalender”等字样的链接。
  3. 如果能找到直接的ICS下载链接(通常以 .ics 结尾),这就是最好的情况。复制这个链接地址。
  4. 重要验证 :你可以尝试将这个ICS链接添加到你的手机日历(如Google Calendar或Apple Calendar)或桌面日历应用(如Thunderbird)。如果能成功订阅并显示正确的垃圾回收事件,说明这个数据源是可靠且格式标准的。

注意 :有些网站可能需要你输入邮编、街道门牌号来生成个性化的日历链接。这种情况下,你需要模拟这个交互过程,在代码中构造出包含你个人地址参数的最终ICS链接。这可能涉及到分析网页表单或查看网络请求。

3.2 使用Python解析ICS日历数据

假设我们已经获得了一个可靠的ICS链接: https://www.example-abfall.de/ical/your-unique-calendar-id.ics

我们将使用Python的 icalendar 库和 requests 库来获取并解析数据。

首先,安装必要的库:

pip install icalendar requests

接下来,编写数据获取与解析的核心函数:

import requests
from icalendar import Calendar
from datetime import datetime, date, timedelta
import json
import os
from pathlib import Path

CACHE_FILE = Path("./cache/abfall_data.json")
CACHE_DURATION = timedelta(hours=12) # 缓存12小时

def fetch_and_parse_ics(ics_url: str):
    """
    从给定的ICS URL获取并解析垃圾回收事件。
    返回一个按日期分类的事件列表。
    """
    try:
        resp = requests.get(ics_url, timeout=10)
        resp.raise_for_status() # 检查HTTP错误
        ics_content = resp.content

        cal = Calendar.from_ical(ics_content)
        events_by_date = {}

        for component in cal.walk('vevent'):
            # 获取事件摘要(通常是垃圾类型,如'Restmüll', 'Gelbe Tonne', 'Papier')
            summary = str(component.get('summary', ''))
            # 获取事件开始日期(通常是回收日)
            dtstart = component.get('dtstart').dt
            # 确保dtstart是date对象,如果是datetime,则取日期部分
            if isinstance(dtstart, datetime):
                event_date = dtstart.date()
            else:
                event_date = dtstart

            # 将事件按日期分组
            date_str = event_date.isoformat() # 转换为'YYYY-MM-DD'字符串作为键
            if date_str not in events_by_date:
                events_by_date[date_str] = []
            events_by_date[date_str].append(summary)

        return events_by_date
    except requests.exceptions.RequestException as e:
        print(f"网络请求失败: {e}")
        return None
    except Exception as e:
        print(f"解析ICS数据失败: {e}")
        return None

def get_cached_or_fresh_data(ics_url: str):
    """
    获取数据,优先使用缓存。
    如果缓存不存在或已过期,则重新抓取并更新缓存。
    """
    # 检查缓存是否存在且未过期
    if CACHE_FILE.exists():
        cache_mtime = datetime.fromtimestamp(CACHE_FILE.stat().st_mtime)
        if datetime.now() - cache_mtime < CACHE_DURATION:
            try:
                with open(CACHE_FILE, 'r', encoding='utf-8') as f:
                    print("使用缓存数据。")
                    return json.load(f)
            except json.JSONDecodeError:
                print("缓存文件损坏,将重新抓取。")

    # 缓存无效,重新抓取
    print("缓存失效或不存在,从源抓取数据...")
    fresh_data = fetch_and_parse_ics(ics_url)
    if fresh_data:
        # 确保缓存目录存在
        CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
        # 写入缓存
        with open(CACHE_FILE, 'w', encoding='utf-8') as f:
            json.dump(fresh_data, f, ensure_ascii=False, indent=2)
        print("数据已更新至缓存。")
    return fresh_data

代码解读与注意事项:

  • 缓存机制 :我们引入了简单的文件缓存。 CACHE_DURATION 定义了缓存的有效期(例如12小时)。这既尊重了数据源的潜在请求限制,也极大地提升了查询响应速度。
  • 错误处理 :网络请求和解析过程都可能出错,使用 try...except 进行包裹,并打印有意义的错误信息,便于后续排查。
  • 日期处理 :ICS中的 dtstart 属性可能是 datetime 类型(包含具体时间),也可能是 date 类型。我们统一转换为 date 类型,因为我们只关心回收日是哪一天。
  • 数据结构 :我们将数据解析为字典 events_by_date ,其键是日期字符串(如‘2023-10-27’),值是该日期所有回收事件的列表(如 ['Restmüll', 'Bioabfall'] )。这个结构非常利于后续的日期查询。

3.3 处理非标准数据源(网页/PDF)

如果找不到ICS,只能退而求其次。这里简要说明思路:

网页抓取(BeautifulSoup):

import requests
from bs4 import BeautifulSoup
# 分析目标网页结构,找到包含日历数据的HTML元素(通常是<table>或<ul>)
# 使用soup.select()或soup.find()定位元素,提取文本,再用正则表达式或字符串处理提取日期和垃圾类型。

难点 :网页结构可能复杂,且随时可能变动,需要编写健壮的选择器,并做好解析失败的备用方案。

PDF解析(pdfplumber):

import pdfplumber
# 打开PDF,逐页提取文本和表格。
with pdfplumber.open('kalender.pdf') as pdf:
    for page in pdf.pages:
        text = page.extract_text()
        tables = page.extract_tables()
        # 分析文本和表格结构,匹配月份、日期和垃圾类型图标或文字。

难点 :PDF格式千差万别,提取的文本可能错位,表格可能合并单元格。这通常需要针对特定PDF文件编写定制化的、脆弱的解析逻辑,维护成本最高。

实操心得 :在项目初期,花最多时间在寻找 稳定、标准的数据源 上。一个可靠的ICS链接抵得上1000行脆弱的网页抓取代码。如果官方实在不提供,可以尝试在社区论坛(如德国本地的MyDealz, City-Forum)搜索,看是否有其他技术爱好者已经找到了稳定的数据接口或方法。

4. 服务层构建:提供标准化查询API

有了结构化的数据,我们需要创建一个Web服务来提供查询接口。这里使用FastAPI,因为它轻量、快速,并且自动生成交互式API文档。

4.1 搭建FastAPI应用骨架

首先,安装FastAPI和Uvicorn(ASGI服务器):

pip install fastapi uvicorn

创建主应用文件 main.py

from fastapi import FastAPI, HTTPException
from datetime import date, datetime, timedelta
from typing import List, Optional
import json
from pathlib import Path
# 导入我们之前写的数据获取函数
from data_fetcher import get_cached_or_fresh_data

app = FastAPI(title="Abfallkalender API", description="垃圾回收日历查询服务")

# 配置你的ICS_URL,可以从环境变量读取,更安全
import os
ICS_URL = os.getenv("ICS_URL", "https://www.example-abfall.de/ical/your-calendar.ics")

@app.get("/")
def read_root():
    return {"message": "Abfallkalender API is running. Check /docs for API documentation."}

@app.get("/api/abfall/heute", summary="获取今日垃圾回收类型")
def get_today():
    """查询今天需要回收哪些垃圾。"""
    return query_for_date(date.today())

@app.get("/api/abfall/morgen", summary="获取明日垃圾回收类型")
def get_tomorrow():
    """查询明天需要回收哪些垃圾。"""
    return query_for_date(date.today() + timedelta(days=1))

@app.get("/api/abfall/datum/{target_date}", summary="获取指定日期垃圾回收类型")
def get_by_date(target_date: str):
    """
    查询指定日期的垃圾回收类型。
    - **target_date**: 日期字符串,格式为 YYYY-MM-DD,例如 2023-10-27
    """
    try:
        parsed_date = date.fromisoformat(target_date)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD.")
    return query_for_date(parsed_date)

@app.get("/api/abfall/woche", summary="获取本周剩余回收计划")
def get_week():
    """查询从明天开始到本周日(或自定义天数)的所有回收计划。"""
    today = date.today()
    end_of_week = today + timedelta(days=(6 - today.weekday())) # 计算本周日
    return query_for_date_range(today + timedelta(days=1), end_of_week) # 从明天开始

def query_for_date(query_date: date):
    """内部函数:查询特定日期的垃圾类型"""
    data = get_cached_or_fresh_data(ICS_URL)
    if not data:
        raise HTTPException(status_code=503, detail="Service temporarily unavailable. Failed to fetch data.")
    date_str = query_date.isoformat()
    events = data.get(date_str, [])
    return {
        "date": date_str,
        "abfall_types": events,
        "is_pickup_day": len(events) > 0
    }

def query_for_date_range(start_date: date, end_date: date):
    """内部函数:查询一个日期范围内的垃圾回收计划"""
    data = get_cached_or_fresh_data(ICS_URL)
    if not data:
        raise HTTPException(status_code=503, detail="Service temporarily unavailable. Failed to fetch data.")
    result = []
    current_date = start_date
    while current_date <= end_date:
        date_str = current_date.isoformat()
        events = data.get(date_str, [])
        if events: # 只返回有回收事件的日期
            result.append({
                "date": date_str,
                "abfall_types": events
            })
        current_date += timedelta(days=1)
    return result

4.2 运行与测试API

在终端中,进入项目目录,运行:

uvicorn main:app --reload --host 0.0.0.0 --port 8000
  • --reload :开发模式,代码修改后自动重启。
  • --host 0.0.0.0 :允许局域网内其他设备访问(方便从手机或另一台电脑测试)。
  • --port 8000 :指定端口。

服务启动后,打开浏览器访问 http://localhost:8000/docs ,你会看到自动生成的Swagger UI交互式文档。你可以直接在这里点击尝试各个接口,如 /api/abfall/heute

测试示例:

  • 访问 http://localhost:8000/api/abfall/heute ,会返回类似以下的JSON:
    {
        "date": "2023-10-27",
        "abfall_types": ["Restmüll", "Bioabfall"],
        "is_pickup_day": true
    }
    
  • 如果当天没有回收事件, abfall_types 会是空数组, is_pickup_day false

4.3 部署考虑

对于家庭使用,你可以在树莓派、旧笔记本或家里的NAS(如果支持Docker)上运行这个服务。

  • 简单部署 :使用 systemd 创建一个服务单元文件,让Uvicorn在后台常驻运行。
  • 容器化部署(推荐) :编写一个 Dockerfile ,将应用打包成Docker镜像。这样部署和迁移都非常方便。
    FROM python:3.11-slim
    WORKDIR /app
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt
    COPY . .
    CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
    
    然后构建并运行: docker build -t abfall-api . docker run -d -p 8000:8000 --env ICS_URL=你的链接 --name abfall-api abfall-api

注意事项 :务必通过环境变量(如 ICS_URL )来传递敏感或可变的配置,而不是硬编码在代码中。这符合十二要素应用原则,也便于在不同环境(开发、生产)中切换。

5. 交互层实现:开发OpenClaw语音技能

现在,我们有了一个健壮的、提供标准JSON API的后端服务。接下来,就是让OpenClaw能够与之对话。OpenClaw的技能开发有其特定范式,通常包括三个核心文件: intents.yaml (定义意图)、 utterances.yaml (定义话语模板)和技能的主逻辑文件(如 __init__.py )。

5.1 定义技能意图与话语

我们需要定义用户可能怎么问,以及技能需要理解什么。

intents.yaml

intents:
  Abfallkalender:
    keywords:
      - abfall
      - müll
      - tonne
      - leerung
      - kalender
    actions:
      - query

utterances.yaml

utterances:
  Abfallkalender:
    - was kommt heute
    - was wird heute abgeholt
    - muss heute die tonne raus
    - kommt heute der müll
    - was kommt morgen
    - muss morgen die tonne raus
    - was kommt am {date}
    - muss am {date} die tonne raus
    - zeig mir den abfallkalender für diese woche

这里我们定义了一个名为 Abfallkalender 的意图。话语样本覆盖了“今天/明天/某天扔什么”以及“本周计划”的常见问法。 {date} 是一个槽位(slot),用于捕获用户说的具体日期,OpenClaw会尝试将其解析为日期实体。

5.2 实现技能处理逻辑

技能的主逻辑文件(例如 skill.py )需要处理匹配到的意图,调用我们的后端API,并生成语音回复。

import requests
from datetime import datetime, date
from typing import Dict, Any
import logging

_logger = logging.getLogger(__name__)

# 配置你的后端API地址
ABFALL_API_BASE = "http://localhost:8000/api/abfall" # 如果OpenClaw和API不在同一主机,需改为IP地址

def handle_abfallkalender(intent: Dict[str, Any]) -> str:
    """
    处理Abfallkalender意图。
    """
    slots = intent.get('slots', {})
    # 判断用户询问的是今天、明天、特定日期还是本周
    if 'date' in slots:
        # 用户指定了日期,例如“was kommt am freitag”
        target_date_str = slots['date'].get('value') # 假设OpenClaw已解析为'YYYY-MM-DD'
        if target_date_str:
            api_url = f"{ABFALL_API_BASE}/datum/{target_date_str}"
            date_display = target_date_str
        else:
            # 解析失败,默认查询今天
            api_url = f"{ABFALL_API_BASE}/heute"
            date_display = "heute"
    else:
        # 检查是否有关键词暗示“明天”或“本周”
        raw_utterance = intent.get('raw_utterance', '').lower()
        if 'morgen' in raw_utterance:
            api_url = f"{ABFALL_API_BASE}/morgen"
            date_display = "morgen"
        elif 'woche' in raw_utterance:
            # 处理本周查询
            return handle_week_query()
        else:
            # 默认查询今天
            api_url = f"{ABFALL_API_BASE}/heute"
            date_display = "heute"

    try:
        response = requests.get(api_url, timeout=5)
        response.raise_for_status()
        data = response.json()
    except requests.exceptions.RequestException as e:
        _logger.error(f"API请求失败: {e}")
        return "Entschuldigung, der Abfallkalender Dienst ist momentan nicht erreichbar."

    abfall_types = data.get('abfall_types', [])
    if not abfall_types:
        return f"Am {date_display} wird kein Müll abgeholt."
    else:
        types_str = ", ".join(abfall_types)
        # 根据垃圾类型数量调整回复语句
        if len(abfall_types) == 1:
            return f"Am {date_display} wird {types_str} abgeholt."
        else:
            return f"Am {date_display} werden {types_str} abgeholt."

def handle_week_query() -> str:
    """处理本周查询"""
    try:
        response = requests.get(f"{ABFALL_API_BASE}/woche", timeout=5)
        response.raise_for_status()
        week_data = response.json()
    except requests.exceptions.RequestException as e:
        _logger.error(f"API请求失败: {e}")
        return "Entschuldigung, der Abfallkalender Dienst ist momentan nicht erreichbar."

    if not week_data:
        return "Für den Rest der Woche sind keine weiteren Abholungen geplant."

    messages = []
    for day in week_data:
        date_str = day['date']
        # 将YYYY-MM-DD转换为更友好的格式,如“Freitag”
        try:
            dt = datetime.strptime(date_str, '%Y-%m-%d')
            day_name = dt.strftime('%A') # 获取星期名,可能需要本地化
        except ValueError:
            day_name = date_str
        types_str = ", ".join(day['abfall_types'])
        messages.append(f"Am {day_name}: {types_str}")

    reply = "Die Abholungen für den Rest der Woche sind: " + "; ".join(messages)
    return reply

5.3 在OpenClaw中注册技能

你需要将技能文件放到OpenClaw的技能目录下(例如 /opt/openclaw/skills/abfallkalender/ ),并确保 __init__.py 或相应的注册文件正确导出了你的意图处理函数。具体的注册方式取决于你使用的OpenClaw版本和架构,请参考其官方文档。

核心逻辑流程:

  1. 用户 :对OpenClaw说“Was kommt heute?”
  2. OpenClaw :语音识别(ASR)转文本,自然语言理解(NLU)模块匹配到 Abfallkalender 意图,并调用注册的 handle_abfallkalender 函数,传入意图数据。
  3. 技能逻辑 :函数分析意图中的槽位和原始话语,决定调用哪个API端点( /heute , /morgen , /datum/... , /woche )。
  4. 调用后端API :向本地运行的FastAPI服务发送HTTP GET请求。
  5. 处理响应 :接收JSON,判断是否有回收事件。
  6. 生成回复 :根据结果构造德语文本回复,例如“Heute wird Restmüll und Bioabfall abgeholt.”
  7. OpenClaw :将文本回复通过文本转语音(TTS)模块读出来给用户听。

6. 进阶功能与系统集成

6.1 实现自动化提醒

语音查询很方便,但自动化提醒才是“懒人”的终极目标。我们可以利用Home Assistant(HA)强大的自动化引擎来实现。

思路 :我们的后端API已经提供了数据。我们可以在HA中创建一个传感器(Sensor),定期(例如每天凌晨5点)调用我们的API,获取当天的回收信息。然后,基于这个传感器的状态,触发自动化。

在HA的 configuration.yaml 中添加RESTful传感器:

sensor:
  - platform: rest
    name: "Heutige Abfuhr"
    resource: "http://你的API地址:8000/api/abfall/heute"
    scan_interval: 3600 # 每小时更新一次,避免频繁请求
    value_template: >
      {% if value_json.is_pickup_day %}
        {{ value_json.abfall_types | join(', ') }}
      {% else %}
        Keine Abfuhr
      {% endif %}
    json_attributes:
      - abfall_types
      - date
      - is_pickup_day

然后创建一个自动化(Automation):

automation:
  - alias: "Abfall Erinnerung am Vorabend"
    trigger:
      - platform: time
        at: '20:00:00' # 晚上8点触发
    condition:
      - condition: template
        value_template: >
          {% set tomorrow = now().date() + timedelta(days=1) %}
          {% set tomorrow_str = tomorrow.isoformat() %}
          {% set response = states('sensor.heutige_abfuhr') %}
          {# 这里需要更复杂的逻辑,最好直接调用API查询明天 #}
          {# 简化示例:检查传感器属性或调用另一个服务 #}
          {{ false }} # 占位,实际需要实现条件判断
    action:
      - service: tts.google_translate_say
        data:
          entity_id: media_player.wohnzimmer_lautsprecher
          message: >
            Morgen wird {{ states('sensor.morgen_abfuhr') }} abgeholt.
            Bitte die Tonne rausstellen.

注意 :上述HA配置仅为概念展示。更健壮的做法是创建一个自定义集成(Custom Component),或者使用HA的 command_line 传感器配合脚本,每天查询明天的数据并设置状态。也可以直接在运行后端API的服务器上写一个Python脚本,在特定时间通过HA的REST API或MQTT来发送通知。

6.2 多区域/多用户支持

如果你的服务需要支持多个家庭(不同地址),就需要扩展后端API。

  • 数据库 :引入SQLite或PostgreSQL,存储 用户/家庭 ICS_URL 区域代码 的映射关系。
  • API改造 :查询接口需要增加身份验证(如简单的API Key)或会话管理,以便区分不同用户的数据源。例如: GET /api/abfall/heute?api_key=xxx
  • 技能改造 :OpenClaw技能需要能够识别用户(如果OpenClaw支持多用户),或者通过语音询问用户地址(如“Für welche Adresse?”),然后将地址参数传递给后端API。

6.3 数据持久化与历史记录

将解析后的日历数据存储到SQLite数据库中,不仅可以缓存,还能实现历史查询和数据分析。

  • 表设计 :可以设计 pickup_events 表,包含 id , date , type , region_id 等字段。
  • 定期任务 :使用 apscheduler 等库创建一个后台任务,每天定时执行,抓取未来一段时间(如下个月)的数据并存入数据库。
  • 优势 :查询速度极快(直接从本地DB读取),可以轻松实现“下个月哪天收纸垃圾?”这类复杂查询。

7. 常见问题排查与优化技巧

在实际部署和运行中,你可能会遇到以下问题:

7.1 数据获取失败

  • 症状 :API返回“Service unavailable”或缓存数据一直不更新。
  • 排查
    1. 检查网络 :确保运行后端服务的主机可以访问外部的ICS URL。在主机上执行 curl -I [你的ICS_URL] 看看是否返回200 OK。
    2. 检查URL有效性 :ICS链接有时效性吗?有些个性化链接可能过期。尝试手动在浏览器中打开该链接,看是否能下载到有效的ICS文件。
    3. 检查解析逻辑 :官方ICS格式可能有微调。打印出解析后的原始组件信息,检查 summary dtstart 字段是否还能正确获取。
    4. 查看日志 :后端服务的日志(如果你配置了,比如Uvicorn的访问日志和错误日志)是首要排查点。

7.2 语音技能无响应或理解错误

  • 症状 :对OpenClaw说话没反应,或者说“今天收垃圾”但技能没触发。
  • 排查
    1. 检查技能注册 :确认技能文件放对了位置,并且 __init__.py 正确导出了意图处理函数。重启OpenClaw服务。
    2. 检查意图匹配 :在OpenClaw的管理界面或日志中,查看用户的原始话语是否被正确识别为文本,以及NLU是否将其匹配到了 Abfallkalender 意图。可能你需要补充更多的话语样本。
    3. 检查技能逻辑日志 :在 handle_abfallkalender 函数开始处添加日志,打印接收到的 intent 字典,确认参数是否正确传入。
    4. 检查内部API调用 :确保技能中配置的 ABFALL_API_BASE 地址是正确的,并且从OpenClaw主机可以访问到该地址(注意localhost和网络IP的区别)。

7.3 日期解析错误

  • 症状 :用户说“am Freitag”,但技能查询了错误的日期。
  • 解决 :这通常是OpenClaw的NLU日期实体解析问题。你需要测试OpenClaw对各种日期表达式的解析结果。在技能代码中,不要完全依赖解析出的 date 槽位,可以结合原始话语中的关键词(如“heute”, “morgen”, “übermorgen”)做后备判断。对于复杂的相对日期(如“nächsten Dienstag”),可能需要更复杂的逻辑或依赖NLU提供更精确的解析。

7.4 性能优化

  • 缓存策略 :对于ICS数据,缓存时间( CACHE_DURATION )可以设置得长一些,比如24小时甚至更长,因为垃圾回收日历很少频繁变动。
  • 异步处理 :在FastAPI中,如果数据获取函数 fetch_and_parse_ics 是同步的且比较耗时,可能会阻塞其他请求。可以考虑将其改为异步函数( async def ),并使用 httpx 等异步HTTP客户端。
  • 健康检查端点 :为后端API添加一个 /health 端点,返回服务的状态(如缓存是否新鲜、数据源是否可达)。这便于监控。

7.5 安全考虑

  • API暴露 :如果你的后端API运行在家庭网络中,并需要从公网访问(例如手机在外查询),务必通过反向代理(如Nginx)设置HTTPS,并考虑添加简单的API Key认证。
  • 环境变量 :永远不要将ICS_URL等敏感信息硬编码在代码中。使用环境变量或配置文件。
  • 输入验证 :FastAPI已经通过类型提示做了基础验证。对于日期参数,我们使用了 date.fromisoformat ,它能有效过滤非法格式。
Logo

码道开发者社区,聚焦华为云码道 CodeArts 代码智能体,沉淀 Agent、Skill、鸿蒙开发实战内容,供开发者查阅资料、交流技术、分享工程实践

更多推荐