PL Uncle

---
name: model-config-check
description: 校验模型配置是否正确、模型是否可以正常连接和返回内容。当用户说"检查模型"、"测试模型"、"模型能不能用"、"模型配置"、"诊断模型问题"时使用。**每次修改模型配置（config.patch/config.apply涉及models.providers）后必须自动执行校验。** 用户只给模型名+API key时，自动识别provider、查找配置、写入并校验。
---

# 模型配置校验 Skill

## ⚡ 自动触发规则

**当以下情况发生时，必须自动运行校验脚本，无需用户请求：**

1. 使用 `gateway config.patch` 修改了 `models` 相关配置（新增/修改 provider 或 model）
2. 使用 `gateway config.apply` 替换了整个配置且包含 models 变更
3. 使用 `gateway update.run` 更新后（模型接口可能有变化）

**自动校验流程：**
1. 配置写入完成 + gateway 重启后，等待 5 秒让服务就绪
2. 执行 `bash ~/.openclaw/workspace/skills/model-config-check/scripts/check_models.sh`
3. 解析输出，向用户汇报校验结果
4. 如果新模型不可用，立即告知用户具体原因和修复建议

**手动触发：** 用户说"检查模型"/"测试模型"/"模型能不能用"/"诊断模型"时也执行同样流程。

## 🤖 自动配置流程

当用户只提供「模型名 + API key」时，按以下流程自动完成配置：

### Step 1: 识别 Provider

根据模型名前缀匹配已知 provider：

| 模型名前缀 | Provider | Base URL | API 类型 |
|-----------|----------|----------|----------|
| `gpt-*`, `o1-*`, `o3-*`, `o4-*`, `chatgpt-*` | OpenAI | `https://api.openai.com` | `openai-completions` → `/v1/chat/completions` |
| `claude-*` | Anthropic | `https://api.anthropic.com` | `anthropic-messages` → `/v1/messages` |
| `deepseek-*` | DeepSeek | `https://api.deepseek.com` | `openai-completions` → `/v1/chat/completions` |
| `glm-*`, `chatglm-*` | 智谱 (Zhipu) | `https://open.bigmodel.cn/api/paas/v4` | `openai-completions` → `/v1/chat/completions` |
| `qwen-*`, `qwq-*`, `qvq-*` | 阿里通义 (DashScope) | `https://dashscope.aliyuncs.com/compatible-mode/v1` | `openai-completions` → `/v1/chat/completions` |
| `moonshot-*`, `kimi-*` | 月之暗面 (Moonshot) | `https://api.moonshot.cn/v1` | `openai-completions` → `/v1/chat/completions` |
| `doubao-*`, `ep-*` | 火山引擎 (豆包) | `https://ark.cn-beijing.volces.com/api/v3` | `openai-completions` → `/v1/chat/completions` |
| `minimax-*`, `abab-*` | MiniMax | `https://api.minimax.chat/v1` | `openai-completions` → `/v1/chat/completions` |
| `yi-*` | 零一万物 (01.AI) | `https://api.lingyiwanwu.com/v1` | `openai-completions` → `/v1/chat/completions` |
| `ernie-*`, `baidu-*` | 百度文心 | `https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop` | `openai-completions` |
| `grok-*` | xAI (Grok) | `https://api.x.ai/v1` | `openai-completions` → `/v1/chat/completions` |
| `gemini-*` | Google Gemini | `https://generativelanguage.googleapis.com/v1beta` | 特殊接口，需单独处理 |
| `mistral-*`, `codestral-*` | Mistral | `https://api.mistral.ai/v1` | `openai-completions` → `/v1/chat/completions` |
| `mimo-*` | 小米 (MiMo) | `https://api.xiaomimimo.com/anthropic` | `anthropic-messages` → `/v1/messages` |

**未匹配时：** 自动搜索 "[model_name] API documentation base_url" 确认配置。

### Step 2: 生成配置

根据匹配结果生成 config.patch 格式的配置，包含：
- provider 名称（用 provider 小写名作为 key）
- baseUrl
- apiKey（用户提供的）
- api 类型
- models 列表（至少包含用户提到的模型）
- 每个 model 的 contextWindow 和 maxTokens（根据已知信息填写默认值）

### Step 3: 应用配置

使用 `gateway config.patch` 写入配置并重启。

### Step 4: 自动校验

重启完成后执行校验脚本，汇报结果。

### 模型上下文/输出默认值

常见模型的 contextWindow 和 maxTokens 参考值：

| 模型 | contextWindow | maxTokens |
|------|-------------|-----------|
| gpt-4o | 128000 | 16384 |
| gpt-4-turbo | 128000 | 4096 |
| o1/o1-mini | 128000 | 100000 |
| claude-3.5-sonnet | 200000 | 8192 |
| claude-3-opus | 200000 | 4096 |
| deepseek-chat | 64000 | 8192 |
| deepseek-reasoner | 64000 | 8192 |
| glm-4 | 128000 | 4096 |
| qwen-max | 32000 | 8192 |
| qwen-long | 1000000 | 65536 |
| kimi-latest | 128000 | 4096 |
| doubao-pro | 4096 (volcengine) | 4096 |
| minimax-abab6.5 | 245760 | 4096 |
| mimo-v2-pro | 262144 | 8192 |

**未知模型：** contextWindow 默认 128000，maxTokens 默认 4096。搜文档确认后更新。

按以下顺序逐项检查，每项给出 ✅/❌ 结果：

### 1. 读取配置

使用 `read` 工具读取 `~/.openclaw/openclaw.json`，提取所有 `models.providers` 下的模型配置。

### 2. 配置完整性检查

对每个 provider 检查：
- `baseUrl` 是否存在且格式正确（http/https 开头）
- `apiKey` 是否存在且非空
- `api` 类型是否正确（`openai-completions` / `anthropic-messages` / `anthropic-completions`）
- 每个 model 的 `id` 是否存在
- `contextWindow` 和 `maxTokens` 是否设置

### 3. 网络连通性检查

使用 `exec` 执行 curl 测试每个 provider 的 baseUrl 是否可达：

```bash
curl -s -o /dev/null -w "%{http_code}" --connect-timeout 5 "<baseUrl>"
```

### 4. API 实际调用测试

对每个 provider，使用 `exec` 执行实际 API 调用测试：

**OpenAI 兼容接口 (`openai-completions`)**：
```bash
curl -s -X POST "<baseUrl>" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <apiKey>" \
  -d '{"model":"<modelId>","messages":[{"role":"user","content":"reply 1"}],"max_tokens":10}' \
  --connect-timeout 10 --max-time 30
```

**Anthropic 兼容接口 (`anthropic-messages`)**：
```bash
curl -s -X POST "<baseUrl>/v1/messages" \
  -H "Content-Type: application/json" \
  -H "x-api-key: <apiKey>" \
  -H "anthropic-version: 2023-06-01" \
  -d '{"model":"<modelId>","max_tokens":10,"messages":[{"role":"user","content":"reply 1"}]}' \
  --connect-timeout 10 --max-time 30
```

注意：对于 anthropic-messages，baseUrl 通常不带 `/v1/messages`，需要拼接。有些 baseUrl 已经包含路径，则直接用。

### 5. 结果解析

检查返回结果：
- HTTP 状态码是否为 200
- 响应体是否包含有效内容（非空）
- 对于 OpenAI 接口：检查 `choices[0].message.content` 是否非空
- 对于 Anthropic 接口：检查 `content[0].text` 是否非空
- 如果 content 为空但 reasoning_content 有值，说明模型把输出放到了思考字段，需标注配置问题

### 6. 生成报告

汇总输出格式：

```
## 模型配置校验报告

### Provider: <name>
- 配置完整性: ✅/❌
- 网络连通: ✅/❌ (HTTP <code>)
- API 认证: ✅/❌
- 模型返回: ✅/❌
- 模型: <modelId> → 状态: ✅/❌ [备注]

### 总结
- 可用模型: X/Y
- 不可用模型: [列表]
- 建议: [修复建议]
```

## URL 路径处理规则

不同 provider 的 baseUrl 结构不同，脚本自动处理：

| API 类型 | 处理规则 |
|----------|----------|
| `anthropic-messages` | 自动追加 `/v1/messages`（除非 URL 已包含） |
| `openai-completions` | 自动追加 `/chat/completions`（除非 URL 已以 `/chat/completions` 或 `/completions` 结尾） |

## 常见问题及修复

| 问题 | 原因 | 修复方式 |
|------|------|----------|
| HTTP 401/403 | API Key 无效或过期 | 更新 apiKey |
| HTTP 500 | 服务端错误/账户耗尽 | 检查账户余额或联系服务商 |
| 连接超时 | baseUrl 不可达 | 检查网络或更换 baseUrl |
| content 为空 | 接口类型不匹配 | 检查 api 字段是否正确（openai vs anthropic） |
| content 为空但 reasoning 有值 | 模型输出到了思考字段 | 增加 max_tokens 或切换接口类型 |
| Relay service error | 中转服务异常 | 检查中转服务状态和账户 |
| THINKING_ONLY | 推理模型思考未完成 | 正常现象，模型可用，增加 max_tokens 可获取完整输出 |

FILE:scripts/check_models.sh
#!/bin/bash
# check_models.sh - 快速检查所有模型配置和连通性
# 用法: bash check_models.sh [config_path]

CONFIG="-$HOME/.openclaw/openclaw.json"
ISSUES=()
OK_COUNT=0
TOTAL=0

GREEN='\033[0;32m'
RED='\033[0;31m'
YELLOW='\033[1;33m'
NC='\033[0m'

echo "=========================================="
echo " 🔍 OpenClaw 模型配置校验"
echo "=========================================="
echo ""

if [ ! -f "$CONFIG" ]; then
    echo -e "RED❌ 配置文件不存在: CONFIGNC"
    exit 1
fi
echo "✅ 配置文件: CONFIG"
echo ""

# Extract providers
TMPEXTRACT=$(mktemp)
cat > "$TMPEXTRACT" << 'PYEOF'
import json, sys
config_path = sys.argv[1]
with open(config_path) as f:
    config = json.load(f)
providers = config.get("models", {}).get("providers", {})
for name, p in providers.items():
    models = [m["id"] for m in p.get("models", [])]
    base = p.get("baseUrl", "")
    key = p.get("apiKey", "")
    api = p.get("api", "")
    print(f"{name}\t{base}\t{key}\t{api}\t{','.join(models)}")
PYEOF

PROVIDERS=$(python3 "$TMPEXTRACT" "$CONFIG" 2>/dev/null)
rm -f "$TMPEXTRACT"

if [ -z "$PROVIDERS" ]; then
    echo -e "RED❌ 未找到任何模型 provider 配置NC"
    exit 1
fi

# Parser for Anthropic responses
TMPPARSE=$(mktemp)
cat > "$TMPPARSE" << 'PYPARSE'
import json, sys
resp_file = sys.argv[1]
with open(resp_file) as f:
    raw = f.read().strip()
if not raw:
    print("EMPTY")
    sys.exit(0)
try:
    r = json.loads(raw)
    # Check for text content
    for c in r.get("content", []):
        if c.get("type") == "text" and c.get("text", "").strip():
            print(c["text"].strip())
            sys.exit(0)
    # Check for thinking content (model is reasoning but didn't produce text yet)
    for c in r.get("content", []):
        if c.get("type") == "thinking" and c.get("thinking", "").strip():
            print("THINKING_ONLY: " + c["thinking"].strip()[:80])
            sys.exit(0)
    if "error" in r:
        print("ERROR: " + json.dumps(r["error"]))
    else:
        print("EMPTY")
except Exception as e:
    print("PARSE_FAIL: " + str(e))
PYPARSE

# Parser for OpenAI responses
TMPPARSE_OAI=$(mktemp)
cat > "$TMPPARSE_OAI" << 'PYPARSE'
import json, sys
resp_file = sys.argv[1]
with open(resp_file) as f:
    raw = f.read().strip()
if not raw:
    print("EMPTY")
    sys.exit(0)
try:
    r = json.loads(raw)
    choices = r.get("choices", [])
    if choices:
        msg = choices[0].get("message", {})
        content = msg.get("content", "")
        if content and content.strip():
            print(content.strip())
            sys.exit(0)
        reasoning = msg.get("reasoning_content", "")
        if reasoning and reasoning.strip():
            print("REASONING_ONLY: " + reasoning.strip()[:80])
            sys.exit(0)
    if "error" in r:
        print("ERROR: " + json.dumps(r["error"]))
    else:
        print("EMPTY")
except Exception as e:
    print("PARSE_FAIL: " + str(e))
PYPARSE

while IFS=$'\t' read -r NAME BASEURL APIKEY API_TYPE MODELS; do
    echo "----------------------------------------"
    echo "📦 Provider: NAME"
    echo "   API 类型: API_TYPE"
    echo "   Base URL: BASEURL"

    if [ -z "$BASEURL" ]; then
        echo -e "   RED❌ baseUrl 为空NC"
        ISSUES+=("NAME: baseUrl 为空")
        continue
    fi
    if [ -z "$APIKEY" ]; then
        echo -e "   RED❌ apiKey 为空NC"
        ISSUES+=("NAME: apiKey 为空")
        continue
    fi
    echo -e "   GREEN✅ 配置完整NC"

    # Network check
    HTTP_CODE=$(curl -s -o /dev/null -w "%{http_code}" --connect-timeout 5 "$BASEURL" 2>/dev/null)
    if [ "$HTTP_CODE" = "000" ]; then
        echo -e "   RED❌ 网络不可达NC"
        ISSUES+=("NAME: 网络不可达")
        continue
    fi
    echo -e "   GREEN✅ 网络连通NC (HTTP HTTP_CODE)"

    IFS=',' read -ra MODEL_LIST <<< "$MODELS"
    for MODEL_ID in "MODEL_LIST[@]"; do
        TOTAL=$((TOTAL + 1))
        echo -n "   🧪 MODEL_ID ... "

        TMPRESP=$(mktemp)

        if [[ "$API_TYPE" == *"anthropic"* ]]; then
            # Build URL: append /v1/messages if not already in the path
            if [[ "$BASEURL" == *"/v1/messages"* ]]; then
                FULL_URL="$BASEURL"
            else
                FULL_URL="BASEURL%//v1/messages"
            fi

            curl -s -X POST "$FULL_URL" \
                -H "Content-Type: application/json" \
                -H "x-api-key: APIKEY" \
                -H "anthropic-version: 2023-06-01" \
                -d "{\"model\":\"MODEL_ID\",\"max_tokens\":512,\"messages\":[{\"role\":\"user\",\"content\":\"reply ok\"}]}" \
                --connect-timeout 10 --max-time 60 \
                -o "$TMPRESP" 2>/dev/null

            TEXT=$(python3 "$TMPPARSE" "$TMPRESP" 2>/dev/null)

        else
            # OpenAI: construct the full endpoint URL
            # If baseUrl doesn't end with /chat/completions or /completions, append /chat/completions
            if [[ "$BASEURL" == *"/chat/completions" ]] || [[ "$BASEURL" == *"/completions" ]]; then
                OAI_URL="$BASEURL"
            else
                OAI_URL="BASEURL%//chat/completions"
            fi

            curl -s -X POST "$OAI_URL" \
                -H "Content-Type: application/json" \
                -H "Authorization: Bearer APIKEY" \
                -d "{\"model\":\"MODEL_ID\",\"messages\":[{\"role\":\"user\",\"content\":\"reply ok\"}],\"max_tokens\":512}" \
                --connect-timeout 10 --max-time 60 \
                -o "$TMPRESP" 2>/dev/null

            TEXT=$(python3 "$TMPPARSE_OAI" "$TMPRESP" 2>/dev/null)
        fi

        rm -f "$TMPRESP"

        if [[ "$TEXT" == "THINKING_ONLY"* ]] || [[ "$TEXT" == "REASONING_ONLY"* ]]; then
            echo -e "YELLOW⚠️  模型工作正常，输出在 reasoning 字段NC"
            OK_COUNT=$((OK_COUNT + 1))
        elif [[ "$TEXT" == "ERROR"* ]]; then
            echo -e "RED❌ TEXTNC"
            ISSUES+=("MODEL_ID: TEXT")
        elif [[ "$TEXT" == "EMPTY" ]] || [[ "$TEXT" == "PARSE_FAIL"* ]] || [ -z "$TEXT" ]; then
            echo -e "RED❌ 无返回内容NC"
            ISSUES+=("MODEL_ID: 无返回内容")
        else
            echo -e "GREEN✅ 返回: TEXTNC"
            OK_COUNT=$((OK_COUNT + 1))
        fi
    done
    echo ""
done <<< "$PROVIDERS"

rm -f "$TMPPARSE" "$TMPPARSE_OAI"

echo "=========================================="
echo "📊 校验总结"
echo "=========================================="
echo "可用模型: OK_COUNT/TOTAL"
echo ""

if [ #ISSUES[@] -eq 0 ]; then
    echo -e "GREEN🎉 所有模型配置正常！NC"
else
    echo -e "RED⚠️  发现 #ISSUES[@] 个问题:NC"
    for issue in "ISSUES[@]"; do
        echo "   • issue"
    done
    echo ""
    echo "💡 修复建议:"
    echo "   - apiKey 无效 → 更新配置中的 apiKey"
    echo "   - HTTP 500 → 检查账户余额或联系服务商"
    echo "   - 连接超时 → 检查网络或 baseUrl"
    echo "   - 无返回内容 → 检查 model id 是否正确、max_tokens 是否充足"
fi

---
name: mijia-control
description: |
  Control Xiaomi Mi Home (米家) smart devices via Xiaomi Cloud API.
  Use when: user wants to control smart home devices (lights, AC, heater, bath heater, switches, etc.),
  check device status, or create automation scenes.
  Triggers: "开灯", "关灯", "开空调", "关空调", "开浴霸", "我要洗澡", "设备状态",
  "创建场景", "自动化", "米家", "smart home", "turn on/off", any home device control request.
  NOT for: non-Xiaomi devices, HomeKit-only devices.
---

# Mi Home (米家) Control

Control Xiaomi smart home devices and create automations via cloud API.

## Setup

**Prerequisites:** `pip3 install micloud`

**Credentials:** `~/.mijia_creds.json` (chmod 600)
```json
{"userId":"...", "serviceToken":"...", "ssecurity":"...", "cUserId":"..."}
```

If login requires 2FA (device trust verification), obtain credentials via browser:
1. Open `https://account.xiaomi.com/pass/serviceLogin?sid=xiaomiio&_json=true` in browser with existing cookies
2. POST to `serviceLoginAuth2` with hash password to get `ssecurity` + `location`
3. Follow `location` URL to get `serviceToken` from cookies

## Device Control

### List devices
```bash
python3 scripts/mijia.py devices
```

### Read property
```bash
python3 scripts/mijia.py get <did> <siid> <piid>
```

### Set property
```bash
python3 scripts/mijia.py set <did> <siid> <piid> <value>
# value: true/false for bool, integer for uint8, etc.
```

**Note:** Mesh devices (BLE switches) return code:1 on set — this is normal (async confirmation). The command still executes. Verify by reading the property after a 1-2 second delay.

### Batch control
Create a JSON file with commands:
```json
[
  {"did": "946635824", "siid": 2, "piid": 1, "value": true},
  {"did": "946633803", "siid": 2, "piid": 1, "value": false}
]
```
```bash
python3 scripts/mijia.py batch /tmp/commands.json
```

## MIoT Spec Reference

Each device has services (siid) → properties (piid). Look up specs at:
`https://home.miot-spec.com/spec/<model>`

Common patterns:
- **Switch/Light:** siid:2 piid:1 = on/off (bool)
- **3-way switch:** siid:2/3/4 piid:1 = channel 1/2/3 (bool)
- **AC:** siid:2 piid:1 = on/off, piid:2 = mode, piid:3 = target temp
- **Bath heater:** siid:2 = light, siid:3 = PTC heater, siid:10 = environment (temp sensor)
- **Water heater:** siid:2 piid:1 = on/off, piid:2 = target temp, piid:3 = current temp

## Create Automation Scenes

```bash
python3 scripts/mijia.py scene_create scene.json
```

Scene JSON format (see `references/scene_template.json`):
```json
{
  "name": "回家自动开灯",
  "identify": "homelight_1234",
  "st_id": 8,
  "setting": {"enable": 1, "enable_push": 0},
  "trigger": {
    "key": "event.<did>.<siid>.<eiid>",
    "did": "<trigger_device_did>",
    "model": "<model>",
    "extra": "{\"siid\":2,\"eiid\":1}"
  },
  "action_list": [
    {
      "did": "<target_device_did>",
      "model": "<model>",
      "extra": "{\"props\":[{\"siid\":2,\"piid\":1,\"value\":true}]}",
      "type": "device_ctrl"
    }
  ]
}
```

Key fields:
- `trigger.key`: format `event.<did>.<siid>.<eiid>` for device events
- `launch.attr_filter`: optional conditions (e.g., only if light is off)
- `action_list[].extra`: JSON string with `props` array
- `st_id`: 8 = user automation
- `home_id` and `uid` are auto-filled if omitted

## Predefined Scenes

### 🚿 Bath Mode (洗澡模式)
When user says "我要洗澡":
1. Read bathroom temp: `get 927361805 10 1`
2. If temp < 28°C → turn on bath heater: `set 927361805 3 1 1` (暖风), target 30°C
3. If temp ≥ 28°C → skip heater
4. Turn on water heater: `set 965879649 2 1 1`, set 43°C: `set 965879649 2 2 43`
5. Turn on bathroom light: `set 946635824 2 1 true`

### 🐟 Auto Fish Feeding (出门自动喂鱼)
Feeds fish once per day when BOSS leaves home.

**Architecture:**
1. **OpenClaw cron** (every 5min): polls door lock event log via `/user/get_user_device_data`
2. If today has an auto-lock event (action=1) AND not fed yet → feed fish → record state
3. After feeding, all subsequent checks skip immediately (daily dedup)
4. **State file** `~/.fish_feed_state.json`: tracks last feed date

**Components:**
- Door lock events: 智能门锁2 Pro (did: 1175215651), event key `2.1020`, action=1 = auto-lock (leaving home)
- Fish feeder: 智能鱼缸 (did: 2026943875), action siid:2 aiid:1, in:[1] (1 portion)
  - ⚠️ Action 参数格式: `"in": [1]` 而不是 `[{"piid":5,"value":1}]`
- Script: `scripts/fish_auto_feed.py`

**Manual commands:**
```bash
python3 scripts/fish_auto_feed.py              # Normal run
python3 scripts/fish_auto_feed.py --dry-run    # Check without feeding
python3 scripts/fish_auto_feed.py --force      # Feed regardless
python3 scripts/fish_auto_feed.py --status     # Show state
```

**Trigger logic:**
- Polls lock event log for auto-lock (action=1) today → feed fish
- Once fed, skips all further checks until next day
- No米家自动化 needed, pure OpenClaw polling

## 小爱音箱 TTS

Make the XiaoAI speaker say something aloud.

```bash
python3 scripts/mijia.py tts <did> "<text>"
# Example: python3 scripts/mijia.py tts 100783118 "你好"
```

Flow: unmute → play-text → wait → pause playback → re-mute.
This prevents the speaker from auto-playing music after TTS.

**Important:**
- Speaker may be muted (default state at night) — script handles unmute/re-mute automatically
- After TTS, always pause playback to prevent auto-music
- Cloud API cannot control screen UI — returning to clock display requires physical tap or voice command

## Token Refresh

Tokens expire periodically. If API returns decode errors or auth failures:
1. Re-login via browser (same flow as initial setup)
2. Update `~/.mijia_creds.json` with new `serviceToken` and `ssecurity`

The `ssecurity` changes each login. The `serviceToken` is session-bound.

FILE:references/devices.json
{
  "_doc": "BOSS home device registry. Lookup by Chinese name to find did/siid/piid.",
  "devices": {
    "厕所灯": {"did": "946635824", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
    "客厅灯": {"did": "946633803", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
    "入户门开关": {"did": "946796488", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
    "客厅中间开关": {"did": "946360216", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
    "次卧门口": {"did": "946625036", "model": "xiaomi.switch.2wpro3", "type": "3-way-switch"},
    "主卧左开关": {"did": "946293001", "model": "xiaomi.switch.2pro3", "type": "3-way-switch"},
    "主卧右开关": {"did": "946658849", "model": "xiaomi.switch.2pro3", "type": "3-way-switch"},
    "小米智能开关Pro3": {"did": "946271120", "model": "xiaomi.switch.2pro3", "type": "3-way-switch"},
    "家政间灯": {"did": "1034720433", "model": "zimi.switch.dhkg01", "type": "switch"},
    "浴霸": {"did": "927361805", "model": "xiaomi.bhf_light.s1", "type": "bath-heater",
      "notes": "siid:2=照明, siid:3=PTC暖风(piid:1=mode 0关/1暖风/2干燥/3吹风/4换气, piid:5=温度档), siid:10 piid:1=浴室温度"},
    "热水器": {"did": "965879649", "model": "xiaomi.waterheater.ymm6", "type": "water-heater",
      "notes": "siid:2 piid:1=开关(uint8 0/1), piid:2=目标温度, piid:3=当前水温"},
    "主卧空调": {"did": "871747012", "model": "xiaomi.airc.arf1r1", "type": "ac"},
    "客厅空调": {"did": "871759920", "model": "xiaomi.airc.arf1r1", "type": "ac"},
    "次卧空调": {"did": "871783853", "model": "xiaomi.airc.arf1r1", "type": "ac"},
    "客厅景观灯": {"did": "2042397495", "model": "lemesh.light.wy0d02", "type": "light"},
    "小米电视EA": {"did": "613985952", "model": "xiaomi.tv.eaffh1", "type": "tv"},
    "小爱触屏音箱": {"did": "100783118", "model": "xiaomi.wifispeaker.l04m", "type": "speaker"},
    "智能门锁2 Pro": {"did": "1175215651", "model": "loock.lock.t3pul", "type": "lock"},
    "智能鱼缸": {"did": "2026943875", "model": "hfjh.fishbowl.m100", "type": "fishbowl",
      "notes": "siid:2=鱼缸(piid:1=开关, piid:2=水泵, piid:4=水温, piid:5=喂食量1-3), action siid:2 aiid:1=喂食(in:[份数], 不是{piid:5,value:N}格式)"},
    "净烟机P1": {"did": "917614267", "model": "cykj.hood.jyj22", "type": "hood"},
    "路由器": {"did": "miwifi.4405b803-d717-7957-d8a3-2807e4477844", "model": "xiaomi.router.ra72", "type": "router"},
    "燃气热水器P10": {"did": "965879649", "model": "xiaomi.waterheater.ymm6", "type": "water-heater"}
  },
  "home_id": 159001331072,
  "scenes": {
    "洗澡模式": {
      "temp_threshold": 28,
      "bath_heater_target": 30,
      "water_heater_target": 43
    },
    "出门自动喂鱼": {
      "method": "OpenClaw cron polling lock events",
      "trigger": "门锁自动锁门(action=1)",
      "cron": "fish-auto-feed (every 5min)",
      "state_file": "~/.fish_feed_state.json"
    }
  }
}

FILE:references/scene_template.json
{
  "_doc": "Mi Home automation scene template. Copy and modify.",
  "name": "场景名称",
  "identify": "unique_id_timestamp",
  "st_id": 8,
  "setting": {"enable": 1, "enable_push": 0},
  "trigger": {
    "_doc": "Trigger: device event. key format: event.<did>.<siid>.<eiid>",
    "key": "event.<did>.<siid>.<eiid>",
    "did": "<device_did>",
    "model": "<device_model>",
    "extra": "{\"siid\":2,\"eiid\":1}"
  },
  "launch": {
    "_doc": "Optional conditions (only execute if conditions met)",
    "attr_filter": {
      "conditions": [
        {
          "did": "<device_did>",
          "model": "<device_model>",
          "extra": "{\"siid\":2,\"piid\":1,\"value\":false,\"operator\":\"=\"}"
        }
      ],
      "logic": "and"
    }
  },
  "action_list": [
    {
      "_doc": "Action: control a device",
      "did": "<device_did>",
      "model": "<device_model>",
      "extra": "{\"props\":[{\"siid\":2,\"piid\":1,\"value\":true}]}",
      "type": "device_ctrl"
    }
  ],
  "home_id": 0,
  "uid": 0
}

FILE:scripts/fish_auto_feed.py
#!/usr/bin/env python3
"""
Auto Fish Feeder v2 - polls door lock events and feeds fish once per day.

Flow:
1. OpenClaw cron (every 5min) runs this script
2. Script checks door lock event log for auto-lock events (action=1) today
3. If found and not fed today → feed fish → record state → done
4. If already fed today → skip immediately

Usage:
  python3 fish_auto_feed.py              # Normal run
  python3 fish_auto_feed.py --dry-run    # Check without feeding
  python3 fish_auto_feed.py --force      # Feed regardless of state
  python3 fish_auto_feed.py --status     # Show current state
"""

import sys, os, json, time, datetime
from pathlib import Path

STATE_FILE = Path.home() / ".fish_feed_state.json"
MIJIA_SCRIPT = Path(__file__).parent.parent / "skills" / "mijia-control" / "scripts" / "mijia.py"

# Config
LOCK_DID = "1175215651"          # 智能门锁2 Pro
LOCK_EVENT_KEY = "2.1020"        # Lock Event
FISH_DID = "2026943875"          # 智能鱼缸
FISH_FEED_SIID = 2
FISH_FEED_AIID = 1
FEED_AMOUNT = 1                  # 1-3 portions

def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {}

def save_state(state):
    STATE_FILE.write_text(json.dumps(state, indent=2))

def get_cloud():
    """Import and init micloud from mijia.py."""
    import importlib.util
    spec = importlib.util.spec_from_file_location("mijia_mod", str(MIJIA_SCRIPT))
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod.get_cloud()

def check_lock_events_today(mc):
    """Check if there's an auto-lock event (leaving home) today."""
    now_ts = int(time.time())
    now_dt = datetime.datetime.now()
    # Cycle starts at 06:00 today, or 06:00 yesterday if before 06:00
    if now_dt.hour < 6:
        cycle_start = (now_dt - datetime.timedelta(days=1)).replace(hour=6, minute=0, second=0, microsecond=0)
    else:
        cycle_start = now_dt.replace(hour=6, minute=0, second=0, microsecond=0)
    ts_start = int(cycle_start.timestamp())

    params = {"data": json.dumps({
        "did": LOCK_DID,
        "key": LOCK_EVENT_KEY,
        "type": "event",
        "time_start": ts_start,
        "time_end": now_ts,
        "limit": 20
    })}
    r = mc.request_country("/user/get_user_device_data", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())

    events = data.get("result", [])
    for evt in events:
        try:
            props = json.loads(evt["value"])
            prop_map = {p["piid"]: p["value"] for p in props}
            action = prop_map.get(3)   # piid:3 = Lock Action
            # action=1 means Lock (auto-lock after leaving)
            if action == 1:
                evt_time = datetime.datetime.fromtimestamp(evt["time"])
                print(f"[FOUND] Auto-lock event at {evt_time.strftime('%H:%M:%S')}")
                return True
        except:
            continue

    return False

def feed_fish(mc):
    """Trigger fish feeder via miotspec action."""
    params = {"data": json.dumps({"params": {
        "did": FISH_DID,
        "siid": FISH_FEED_SIID,
        "aiid": FISH_FEED_AIID,
        "in": [FEED_AMOUNT]
    }})}
    r = mc.request_country("/miotspec/action", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    code = data.get("result", {}).get("code", -1)
    print(f"[INFO] Fish feed API response code: {code}")
    return code in (0, 1)

def main():
    dry_run = "--dry-run" in sys.argv
    force = "--force" in sys.argv
    status = "--status" in sys.argv

    # "Today" runs from 06:00 to next day 06:00
    now = datetime.datetime.now()
    if now.hour < 6:
        feed_day = (now.date() - datetime.timedelta(days=1)).isoformat()
    else:
        feed_day = now.date().isoformat()

    state = load_state()

    if status:
        print(json.dumps(state, indent=2, ensure_ascii=False))
        return

    # Already fed this cycle? Skip entirely.
    if state.get("last_feed_date") == feed_day and not force:
        print(f"[OK] Already fed this cycle ({feed_day}). Done.")
        return

    mc = get_cloud()

    # Check for auto-lock event today
    has_left = check_lock_events_today(mc)
    if not has_left and not force:
        print(f"[OK] No auto-lock event today. BOSS hasn't left home yet.")
        return

    print(f"[TRIGGER] BOSS has left home. Feeding fish...")

    if dry_run:
        print("[DRY-RUN] Would feed fish now. Skipping.")
        return

    success = feed_fish(mc)

    # Record state regardless of success — only attempt once per day
    state["last_feed_date"] = feed_day
    state["last_feed_time"] = datetime.datetime.now().isoformat()
    state["last_feed_success"] = success
    state["total_feeds"] = state.get("total_feeds", 0) + (1 if success else 0)
    save_state(state)

    if success:
        print(f"[DONE] Fed fish. Total feeds: {state['total_feeds']}")
    else:
        print(f"[WARN] Feeding may have failed. Will NOT retry today.")

if __name__ == "__main__":
    main()

FILE:scripts/mijia.py
#!/usr/bin/env python3
"""
Mi Home (米家) Cloud Controller
Controls Xiaomi smart home devices via Xiaomi Cloud API.

Usage:
  python3 mijia.py login                         # Login and save credentials
  python3 mijia.py devices                        # List all devices
  python3 mijia.py status <did>                   # Read device properties
  python3 mijia.py set <did> <siid> <piid> <val>  # Set a device property
  python3 mijia.py get <did> <siid> <piid>        # Get a device property
  python3 mijia.py scene_create <json_file>       # Create automation scene
  python3 mijia.py scene_list                     # List automation scenes
  python3 mijia.py batch <json_file>              # Execute batch commands

Credentials are stored in ~/.mijia_creds.json (auto-refreshed via browser cookies).
"""

import sys, os, json, hashlib, time, argparse
from pathlib import Path

# Ensure micloud is available
try:
    from micloud import MiCloud
    from micloud.miutils import get_session
except ImportError:
    print("ERROR: micloud not installed. Run: pip3 install micloud", file=sys.stderr)
    sys.exit(1)

CREDS_PATH = Path.home() / ".mijia_creds.json"
DEVICES_CACHE = Path.home() / ".mijia_devices.json"

def load_creds():
    if not CREDS_PATH.exists():
        print(f"ERROR: No credentials found at {CREDS_PATH}", file=sys.stderr)
        print("Run 'python3 mijia.py login' first, or create the file manually.", file=sys.stderr)
        sys.exit(1)
    return json.loads(CREDS_PATH.read_text())

def save_creds(creds):
    CREDS_PATH.write_text(json.dumps(creds, indent=2))
    CREDS_PATH.chmod(0o600)

def get_cloud():
    creds = load_creds()
    mc = MiCloud.__new__(MiCloud)
    mc.user_id = int(creds["userId"])
    mc.service_token = creds["serviceToken"]
    mc.ssecurity = creds["ssecurity"]
    mc.cuser_id = creds.get("cUserId", "")
    mc.session = None
    mc.locale = "zh_CN"
    mc.timezone = "GMT+08:00"
    mc.default_server = "cn"
    mc.failed_logins = 0
    return mc

def cmd_login(args):
    """Interactive login - saves credentials."""
    username = args.username or input("Xiaomi username (phone): ")
    password = args.password or input("Xiaomi password: ")
    
    mc = MiCloud(username, password)
    try:
        mc.login()
        creds = {
            "userId": str(mc.user_id),
            "serviceToken": mc.service_token,
            "ssecurity": mc.ssecurity,
            "cUserId": mc.cuser_id or ""
        }
        save_creds(creds)
        print(f"Login successful! Credentials saved to {CREDS_PATH}")
    except Exception as e:
        print(f"Login failed: {e}", file=sys.stderr)
        print("If 2FA is required, obtain credentials via browser and save manually.", file=sys.stderr)
        sys.exit(1)

def cmd_devices(args):
    """List all devices."""
    mc = get_cloud()
    devices = mc.get_devices(country="cn")
    if not devices:
        print("No devices found or API error.")
        return
    
    # Cache devices
    DEVICES_CACHE.write_text(json.dumps(devices, ensure_ascii=False, indent=2))
    
    print(f"Found {len(devices)} devices:\n")
    for d in devices:
        online = "🟢" if d.get("isOnline") else "🔴"
        print(f"{online} {d.get('name')} | model: {d.get('model')} | did: {d.get('did')} | ip: {d.get('localip','N/A')}")

def cmd_get(args):
    """Get a device property."""
    mc = get_cloud()
    params = {"data": json.dumps({"params": [
        {"did": args.did, "siid": int(args.siid), "piid": int(args.piid)}
    ]})}
    r = mc.request_country("/miotspec/prop/get", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    result = data.get("result", [{}])[0]
    if result.get("code") == 0:
        print(json.dumps({"value": result["value"]}, ensure_ascii=False))
    else:
        print(json.dumps({"error": result.get("code")}, ensure_ascii=False))

def cmd_set(args):
    """Set a device property."""
    mc = get_cloud()
    # Auto-detect value type
    val = args.value
    if val.lower() == "true": val = True
    elif val.lower() == "false": val = False
    else:
        try: val = int(val)
        except:
            try: val = float(val)
            except: pass
    
    params = {"data": json.dumps({"params": [
        {"did": args.did, "siid": int(args.siid), "piid": int(args.piid), "value": val}
    ]})}
    r = mc.request_country("/miotspec/prop/set", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    result = data.get("result", [{}])[0]
    code = result.get("code", -1)
    # code:0 = confirmed success, code:1 = sent (mesh async), both are OK
    if code in (0, 1):
        print(json.dumps({"ok": True, "code": code}, ensure_ascii=False))
    else:
        print(json.dumps({"ok": False, "code": code}, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)

def cmd_status(args):
    """Read common properties of a device."""
    mc = get_cloud()
    # Read siid 2-10, piid 1-5
    props = []
    for siid in range(2, 11):
        for piid in range(1, 6):
            props.append({"did": args.did, "siid": siid, "piid": piid})
    
    params = {"data": json.dumps({"params": props})}
    r = mc.request_country("/miotspec/prop/get", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    
    results = []
    for p in data.get("result", []):
        if p.get("code") == 0:
            results.append({"siid": p["siid"], "piid": p["piid"], "value": p["value"]})
    print(json.dumps(results, ensure_ascii=False, indent=2))

def cmd_batch(args):
    """Execute batch set commands from JSON file."""
    mc = get_cloud()
    commands = json.loads(Path(args.json_file).read_text())
    
    params_list = []
    for cmd in commands:
        params_list.append({
            "did": cmd["did"],
            "siid": cmd["siid"],
            "piid": cmd["piid"],
            "value": cmd["value"]
        })
    
    params = {"data": json.dumps({"params": params_list})}
    r = mc.request_country("/miotspec/prop/set", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    
    for result in data.get("result", []):
        code = result.get("code", -1)
        ok = "✅" if code in (0, 1) else "❌"
        print(f"{ok} did:{result['did']} siid:{result['siid']} piid:{result['piid']} code:{code}")

def cmd_tts(args):
    """Make a XiaoAI speaker say text aloud."""
    import time as _time
    mc = get_cloud()
    did = args.did
    text = args.text

    def _action(siid, aiid, ins):
        params = {"data": json.dumps({"params": {"did": did, "siid": siid, "aiid": aiid, "in": ins}})}
        r = mc.request_country("/miotspec/action", "cn", params)
        return json.loads(r if isinstance(r, str) else r.decode())

    def _set(props):
        params = {"data": json.dumps({"params": [{"did": did, **p} for p in props]})}
        mc.request_country("/miotspec/prop/set", "cn", params)

    # 1. Unmute + ensure volume
    _set([{"siid": 2, "piid": 2, "value": False}, {"siid": 2, "piid": 1, "value": int(args.volume)}])
    _time.sleep(0.5)

    # 2. Play text (siid:5 aiid:1, in: piid:1=text)
    result = _action(5, 1, [{"piid": 1, "value": text}])
    code = result.get("result", {}).get("code", -1)
    print(json.dumps({"ok": code == 0, "code": code}, ensure_ascii=False))

    # 3. Wait for speech to finish (~0.15s per char + 2s buffer)
    wait = max(2, len(text) * 0.15 + 2)
    _time.sleep(wait)

    # 4. Pause playback to prevent auto-music
    _action(3, 2, [])

    # 5. Re-mute if requested
    if args.remute:
        _set([{"siid": 2, "piid": 2, "value": True}])

def cmd_scene_create(args):
    """Create an automation scene from JSON definition."""
    mc = get_cloud()
    scene = json.loads(Path(args.json_file).read_text())
    
    # Ensure required fields
    if "home_id" not in scene:
        scene["home_id"] = _get_home_id(mc)
    if "uid" not in scene:
        scene["uid"] = mc.user_id
    if "st_id" not in scene:
        scene["st_id"] = 8
    if "setting" not in scene:
        scene["setting"] = {"enable": 1, "enable_push": 0}
    if "identify" not in scene:
        scene["identify"] = f"scene_{int(time.time())}"
    
    params = {"data": json.dumps(scene)}
    r = mc.request_country("/scene/edit", "cn", params)
    if isinstance(r, bytes):
        r = r.decode()
    data = json.loads(r) if isinstance(r, str) else r
    print(json.dumps(data, ensure_ascii=False, indent=2))

def cmd_scene_list(args):
    """List existing scenes."""
    mc = get_cloud()
    home_id = _get_home_id(mc)
    
    params = {"data": json.dumps({"home_id": home_id, "req_type": 1})}
    r = mc.request_country("/appgateway/miot/appsceneservice/AppSceneService/GetSceneList", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    
    if data.get("code") == 0:
        scenes = data.get("result", {}).get("scene_info_list", [])
        print(f"Found {len(scenes)} scenes:")
        for s in scenes:
            print(f"  - {s.get('name','')} | id:{s.get('scene_id','')}")
    else:
        print(f"Error: {data}")

def _get_home_id(mc):
    """Get the first home ID."""
    params = {"data": json.dumps({"fg": True, "fetch_share": True, "limit": 10, "app_ver": 7})}
    r = mc.request_country("/v2/homeroom/gethome", "cn", params)
    data = json.loads(r if isinstance(r, str) else r.decode())
    homes = data.get("result", {}).get("homelist", [])
    return homes[0]["id"] if homes else 0

def main():
    parser = argparse.ArgumentParser(description="Mi Home Cloud Controller")
    sub = parser.add_subparsers(dest="command")
    
    p_login = sub.add_parser("login", help="Login to Xiaomi Cloud")
    p_login.add_argument("--username", "-u")
    p_login.add_argument("--password", "-p")
    
    sub.add_parser("devices", help="List all devices")
    
    p_get = sub.add_parser("get", help="Get device property")
    p_get.add_argument("did")
    p_get.add_argument("siid")
    p_get.add_argument("piid")
    
    p_set = sub.add_parser("set", help="Set device property")
    p_set.add_argument("did")
    p_set.add_argument("siid")
    p_set.add_argument("piid")
    p_set.add_argument("value")
    
    p_status = sub.add_parser("status", help="Read device status")
    p_status.add_argument("did")
    
    p_batch = sub.add_parser("batch", help="Batch set from JSON")
    p_batch.add_argument("json_file")
    
    p_sc = sub.add_parser("scene_create", help="Create automation scene")
    p_sc.add_argument("json_file")
    
    p_tts = sub.add_parser("tts", help="Make speaker say text")
    p_tts.add_argument("did")
    p_tts.add_argument("text")
    p_tts.add_argument("--volume", type=int, default=50)
    p_tts.add_argument("--no-remute", dest="remute", action="store_false", default=True)

    sub.add_parser("scene_list", help="List scenes")
    
    args = parser.parse_args()
    
    commands = {
        "login": cmd_login, "devices": cmd_devices, "get": cmd_get,
        "set": cmd_set, "status": cmd_status, "batch": cmd_batch, "tts": cmd_tts,
        "scene_create": cmd_scene_create, "scene_list": cmd_scene_list
    }
    
    if args.command in commands:
        commands[args.command](args)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()

---
name: feishu-send-image
description: |
  Send images directly in Feishu chat as native image messages (not file attachments).
  Use when: need to send a generated image, chart, screenshot, or any local image file to a Feishu user or group chat.
  Triggers: "发图", "send image", "发送图片", sending charts/graphs/screenshots via Feishu.
  Note: OpenClaw's built-in message tool (filePath/media/buffer) does NOT work for Feishu image sending — it only produces file attachments. This skill bypasses that limitation by calling Feishu Bot API directly.
---

# Feishu Send Image

Send local image files as native Feishu image messages via the Feishu Bot API.

## Why This Skill Exists

OpenClaw's `message` tool with `filePath`, `media`, or `buffer` parameters only sends file attachments in Feishu, not inline images. This skill calls the Feishu API directly to send proper image messages.

## Quick Usage

Run the script:

```bash
bash scripts/feishu_send_image.sh <image_path> <receive_id> <app_id> <app_secret> [receive_id_type]
```

### Arguments

| Arg | Description |
|-----|-------------|
| `image_path` | Local path to image (png/jpg/gif/webp) |
| `receive_id` | Feishu `open_id` (user) or `chat_id` (group) |
| `app_id` | Feishu app ID from `~/.openclaw/openclaw.json` → `feishu.accounts.default.appId` |
| `app_secret` | Feishu app secret from `~/.openclaw/openclaw.json` → `feishu.accounts.default.appSecret` |
| `receive_id_type` | `open_id` (default) or `chat_id` |

### Example

```bash
bash scripts/feishu_send_image.sh /tmp/chart.png \
  ou_38470740452f6083ce189b7ddec722f8 \
  cli_a92c368412f9dcb1 \
  7uM7aLqeqYqm0Fsy0IP5QhOyTBSwxlfT
```

## Getting Credentials

Read `~/.openclaw/openclaw.json` and extract:
- `channels.feishu.accounts.default.appId`
- `channels.feishu.accounts.default.appSecret`

The receiver's `open_id` comes from inbound message metadata (`sender_id`).

## How It Works

1. **Get token** — `POST /auth/v3/tenant_access_token/internal` with appId/appSecret
2. **Upload image** — `POST /im/v1/images` with `image_type=message`, returns `image_key`
3. **Send message** — `POST /im/v1/messages` with `msg_type=image` and the `image_key`

## Output

On success: `OK: image_key=<key> message_id=<id>`
On failure: prints error and exits with code 1.

FILE:README.md
# feishu-send-image

在飞书聊天中发送原生图片消息的 OpenClaw 技能。

## 解决什么问题

OpenClaw 内置的 `message` 工具在飞书中发送图片时，无论使用 `filePath`、`media` 还是 `buffer` 参数，都只会生成文件附件，而非可直接预览的图片消息。

本技能通过直接调用飞书 Bot API，实现真正的图片消息发送。

## 使用场景

- 发送 AI 生成的图表（股票走势、数据分析图等）
- 发送截图、处理后的图片
- 任何需要在飞书对话中直接展示图片的场景

## 工作原理

三步完成图片发送：
1. 获取飞书 `tenant_access_token`
2. 上传图片获取 `image_key`
3. 发送 `image` 类型消息

## 使用方法

```bash
bash scripts/feishu_send_image.sh <图片路径> <接收者ID> <app_id> <app_secret> [id类型]
```

### 参数说明

| 参数 | 说明 |
|------|------|
| `图片路径` | 本地图片文件路径，支持 png/jpg/gif/webp |
| `接收者ID` | 飞书用户的 `open_id` 或群聊的 `chat_id` |
| `app_id` | 飞书应用 App ID |
| `app_secret` | 飞书应用 App Secret |
| `id类型` | 可选，`open_id`（默认）或 `chat_id` |

### 示例

```bash
# 发送图片给用户
bash scripts/feishu_send_image.sh /tmp/chart.png ou_xxxxx cli_xxxxx secret_xxxxx

# 发送图片到群聊
bash scripts/feishu_send_image.sh /tmp/chart.png oc_xxxxx cli_xxxxx secret_xxxxx chat_id
```

## 凭证获取

App ID 和 App Secret 可从 OpenClaw 配置文件中获取：

```
~/.openclaw/openclaw.json → channels.feishu.accounts.default.appId / appSecret
```

接收者 `open_id` 来自飞书消息的 `sender_id` 字段。

## 前置条件

- 飞书自建应用已创建并启用机器人能力
- 应用已获取 `im:message:send_as_bot`（发送消息）和 `im:resource`（上传图片）权限
- 系统已安装 `curl` 和 `python3`

## 安装

```bash
openclaw skill install feishu-send-image.skill
```

或手动将 `feishu-send-image/` 目录放入 `~/.openclaw/workspace/skills/` 下。

FILE:scripts/feishu_send_image.sh
#!/bin/bash
# Send an image to a Feishu user/chat via Feishu Bot API
# Usage: feishu_send_image.sh <image_path> <receive_id> <app_id> <app_secret> [receive_id_type]
#
# Arguments:
#   image_path       - Local path to the image file (png/jpg/gif/webp)
#   receive_id       - Feishu open_id (user) or chat_id (group)
#   app_id           - Feishu app ID
#   app_secret       - Feishu app secret
#   receive_id_type  - "open_id" (default) or "chat_id"

set -euo pipefail

IMAGE_PATH="?Usage: feishu_send_image.sh <image_path> <receive_id> <app_id> <app_secret> [receive_id_type]"
RECEIVE_ID="?Missing receive_id"
APP_ID="?Missing app_id"
APP_SECRET="?Missing app_secret"
RECEIVE_ID_TYPE="-open_id"

# Step 1: Get tenant_access_token
TOKEN_RESP=$(curl -s -X POST 'https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal' \
  -H 'Content-Type: application/json' \
  -d "{\"app_id\":\"$APP_ID\",\"app_secret\":\"$APP_SECRET\"}")

TOKEN=$(echo "$TOKEN_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['tenant_access_token'])" 2>/dev/null)
if [ -z "$TOKEN" ]; then
  echo "ERROR: Failed to get access token" >&2
  echo "$TOKEN_RESP" >&2
  exit 1
fi

# Step 2: Upload image to get image_key
UPLOAD_RESP=$(curl -s -X POST 'https://open.feishu.cn/open-apis/im/v1/images' \
  -H "Authorization: Bearer $TOKEN" \
  -F 'image_type=message' \
  -F "image=@$IMAGE_PATH")

IMAGE_KEY=$(echo "$UPLOAD_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['image_key'])" 2>/dev/null)
if [ -z "$IMAGE_KEY" ]; then
  echo "ERROR: Failed to upload image" >&2
  echo "$UPLOAD_RESP" >&2
  exit 1
fi

# Step 3: Send image message
SEND_RESP=$(curl -s -X POST "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=$RECEIVE_ID_TYPE" \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d "{\"receive_id\":\"$RECEIVE_ID\",\"msg_type\":\"image\",\"content\":\"{\\\"image_key\\\":\\\"$IMAGE_KEY\\\"}\"}")

CODE=$(echo "$SEND_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin).get('code',1))" 2>/dev/null)
if [ "$CODE" = "0" ]; then
  MSG_ID=$(echo "$SEND_RESP" | python3 -c "import sys,json; print(json.load(sys.stdin)['data']['message_id'])" 2>/dev/null)
  echo "OK: image_key=$IMAGE_KEY message_id=$MSG_ID"
else
  echo "ERROR: Failed to send message" >&2
  echo "$SEND_RESP" >&2
  exit 1
fi