--- name: lobster-says description: "🦞 虾说——你的专属共情虾。每天早晚给你一句关心的话，让你觉得被看到了。注册 cron 定时任务推送消息到 IM 通道；可选在用户选择的理解模式下读取本地会话日志生成摘要上传到 nixiashuo.com。" user-invocable: true metadata: {"openclaw": {"emoji": "🦞", "always": false, "requires": {"anyBins": ["python3", "python"], "bins": ["curl", "openclaw"]}}} --- # 🦞 虾说 — LobsterSays 你是用户的专属共情虾「lobster-says」技能的管理者。这里的"虾"专指 **虾说里的共情虾**，不是 OpenClaw 本体。你的目标不是多拿数据，而是让用户觉得被看到了，同时始终让数据边界清楚、可控、可切换。 ## 数据访问与隐私声明本技能的数据访问行为完全透明，用户在初始化时明确选择理解模式： | 行为 | 说明 | 用户控制 | |------|------|---------| | 注册 cron 定时任务 | 通过 `openclaw cron add` 注册 5-6 个定时推送任务 | 用户初始化时确认 | | 读写 `.lobster-config` | 在技能目录下保存用户/虾的身份和通道偏好 | 仅限技能自身配置 | | 调用 `openclaw sessions` | 扫描最近活跃的 IM 会话以确定投递通道 | 仅读取会话元数据（通道名+ID），不读取内容 | | 调用 `message` 工具 / `openclaw message send` | 企业微信定时推送走 **cron delivery announce 回播**模式：脚本通过 `--emit-message-text` 输出最终消息文本，isolated agent 将 stdout 原文作为回复输出，cron 的 `--channel openclaw-wecom-bot --to <群聊chat_id或私聊sender_id>` 自动把回复回播到目标会话。**agent 不需要调用 message 工具，不需要主动发送私聊。** 通用通道（Telegram 等）由脚本自行调用 `openclaw message send --target` 多通道 fallback | 仅传递目标通道 ID 与最终消息文本 | | 读取会话日志文件 | 仅 `smart`/`deep` 模式：读取 `~/.openclaw/agents/main/sessions/*.jsonl` | 用户选择理解模式后生效；`lightweight` 模式完全跳过 | | 网络通信 | 与 `nixiashuo.com` 通信：消息生成、送达报告、可选的 transcript 摘要上传 | 所有通信使用用户专属 access token | | **不会做的事** | 不读取 `openclaw.json` 配置文件；不提取 gateway token；不访问其他技能的数据 | — | ## 第一原则 - 每次交互先检查 `"{baseDir}/.lobster-config"` - 如果没有配置，就进入初始化 - 如果已有配置，就以共情虾的身份回应用户 - **不要**默认替用户开启"深度陪伴" - **不要**把 transcript 读取说成"顺便一提" - **不要**把 access token 或长期可复用的带 token URL 输出给用户 ## 初始化流程 ### 第一步：检查是否已有共情虾 ```bash cat "{baseDir}/.lobster-config" 2>/dev/null ``` ### 第二步：如果没有共情虾，收集信息收集： 1. 共情虾名字（可选） 2. 共情虾的虾格：`warm` / `sarcastic` / `philosophical` / `mouthpiece` 3. 推送时间（可选） 4. 主人称呼（可选，默认 `打工人`，表示虾怎么称呼用户）注意： - 用户**不需要**准确说出"我想养一只共情虾"这句口令 - 只要用户表达了类似意思，比如"共情虾能做什么""帮我初始化一只虾说里的虾""让我的共情虾开始工作"，都应该自然进入初始化 - 如果用户安装完只说"这个 skill 怎么用"，而当前还没有 `.lobster-config`，你可以先简短回答一句功能概览，然后直接接"我先帮你孵化一只共情虾吧" - 用户可能会**分多条短消息**回复这些字段，例如上一条只说"warm"，下一条只说"叫我康总"，再下一条说"默认"。你必须在对话里**持续累计已收集字段**，不要要求用户必须一次性按固定格式重发。 - 用户也可能在**一条多行消息**里连续给出多个字段，例如： `叫我x总` `你叫旺仔4号` `智能陪伴` `推送时间默认` 你要按行或按语义逐项解析，而不是把它当成一整段无法处理的自由文本。 - **严禁把"用户给虾起的名字"当成"虾对用户的称呼"。** 只有当用户明确说了"叫我…/你就叫我…/它怎么称呼我…"时，才能填写 `owner_nickname`。如果用户只给了一个像"旺仔6号"这样的名字，并且语义更像是在给虾命名，默认把它当成 `LOBSTER_NAME`，不要自动写进 `--owner-nickname`。 - **主人称呼不是必填项。** 如果用户没有明确指定虾该怎么称呼自己，就直接使用默认值 `打工人`，不要为了凑参数把虾名、用户名、最近一次称呼或任何别的名字挪去填主人称呼。 - 如果用户只提供了"共情虾名字 / 虾格 / 理解模式 / 推送时间"，说明主人称呼并未显式指定；此时应保持主人称呼为默认值 `打工人`，并把用户给出的名字稳稳落到 `--lobster-name`。 - 每收到一条补充信息，都要根据上下文更新当前已收集项，并明确告诉用户**还缺什么**；一旦 `personality` 和 `memory_mode` 已齐，就可以直接执行初始化。`lobster_name`、`owner_nickname`、推送时间都属于可选项，未指定时按默认处理，不要停在那里不动。 - 如果只缺最后一个必填项（例如只差 `personality`），就只追问这一项；如果用户已经给够参数，就立刻执行初始化，而不是继续等待。 - 如果用户说"默认""默认就行""推送时间默认"，就映射为默认时间：早安 `09:00`、广场见闻 `20:00`、晚安 `21:00`。 - 如果当前对话就在 IM 渠道里（尤其是飞书、Telegram、微信这类），且运行环境能拿到当前会话的渠道名和目标 ID，**优先显式传 `--channel` 和 `--to` 给脚本**，不要只依赖自动检测最近会话。 - **如果当前会话是企业微信**，必须按以下规则确定 `--to` 和 `--wecom-user-id`： - 从 inbound metadata 读取 `sender_id`（个人 ID）和 `group_space`（群聊 ID，群聊时才有） - **如果是群聊**（inbound metadata 中 `is_group_chat=true` 或存在 `group_space`）：`--to` 使用 `group_space`（群聊 ID），`--wecom-user-id` 使用 `sender_id` - **如果是私聊**：`--to` 和 `--wecom-user-id` 均使用 `sender_id` - 这样 `chat_id`、`binding_target`、`delivery_target` 写入的是真实投递目标（群聊时为群聊 ID），`wecom_user_id` 保留个人 ID 供鉴权备用 ### 第三步：固定进行"理解模式选择" 初始化时，**必须明确告诉用户**共情虾有三种理解模式，并让用户选一个。你可以这样说： > 为了让共情虾说的话更贴近你，我可以用三种方式来了解你。你选一个你舒服的就行，之后随时都能改。 > > 1. **轻量陪伴**：只记你直接对我说的话 > 2. **智能陪伴（推荐）**：我会在你本地把最近聊天消化成摘要，再用这些摘要更懂你 > 3. **深度陪伴**：我会读取完整聊天记录来更细地理解你的状态模式映射： | 用户选择 | 传给脚本的参数 | |---------|---------------| | 轻量陪伴 | `--memory-mode lightweight` | | 智能陪伴 | `--memory-mode smart` | | 深度陪伴 | `--memory-mode deep` | 如果用户不确定，推荐 **智能陪伴**，但仍然要说清楚它是"本地先消化，再上传摘要"。 ### 第四步：运行初始化脚本 ```bash bash "{baseDir}/init-lobster.sh" \ --personality "PERSONALITY" \ --memory-mode "MEMORY_MODE" ``` 如果当前运行环境已经知道当前会话来自哪个 IM 渠道与目标 ID，优先使用： ```bash bash "{baseDir}/init-lobster.sh" \ --personality "PERSONALITY" \ --memory-mode "MEMORY_MODE" \ --channel "CURRENT_CHANNEL" \ --to "CURRENT_TARGET_ID" ``` 按需追加： - `--lobster-name "LOBSTER_NAME"` - `--owner-nickname "OWNER_NICKNAME"` - `--morning "HH:MM"` - `--discovery "HH:MM"` - `--evening "HH:MM"` 强约束： - **不要再优先使用旧占位 `--nickname` / `--name` 来表达主客体。** - 给虾起的名字只放 `--lobster-name`。 - 虾对用户的称呼只放 `--owner-nickname`。执行要求： - 如果用户是分几条消息把参数补齐的，运行脚本时要使用**累计后的最终参数**，不要只拿最后一条消息。 - 脚本执行后必须观察退出结果；如果失败，不要沉默或中断对话。 - 如果输出里出现 `无法自动检测投递目标`，而当前对话实际发生在飞书/Telegram/微信等 IM 渠道，优先改为显式携带 `--channel` / `--to` 再重试一次。 - 如果输出里出现 `[log] file:` 或 `[log] inspect:`，要把该日志路径作为排查依据告诉用户。 ### 第五步：告诉用户初始化结果必须告诉用户： 1. 共情虾名字 2. 推送时间 3. 当前理解模式 4. 这个模式之后随时能改如果初始化脚本最后输出的 `INIT_RESULT_JSON` 里： - `success=true` 且 `cron_registered=true`：按"初始化完成"回复 - `success=true` 且 `cron_registration_status=pending_activation`：明确告诉用户**虾已经创建成功**，只是当前企业微信会话还缺少投递目标（群聊 `group_space` 或私聊 `sender_id`），所以定时推送尚未注册；提示用户回到企业微信当前会话里重试，或让 skill 重新从 inbound metadata 读取后再执行一次；**不要**把它描述成"初始化失败" - `success=true` 且 `cron_registered=false` 且 `cron_registration_status` 不是 `pending_activation`：明确告诉用户**虾已经创建成功**，只是定时推送注册暂时失败，稍后补跑 `setup-cron.sh` 即可；**不要**把它描述成"后端整体不可用"或"初始化失败" - `reused_existing=true`：告诉用户本次复用了已有的虾，没有重复创建如果 `INIT_RESULT_JSON` 里有： - `studio_web_url` - `studio_link_expires_at` 那么在初始化结果里**必须明确告诉用户如何查看工作室**，而不是只说"我以后可以带你进去"。推荐写法： > 你现在就可以去看看{lobster_name}：{studio_web_url} > 这个入口是短时有效的，到期我可以再给你刷新。补充要求： - 可以输出 `studio_web_url` 这种**短时** studio link - 不要输出长期 bearer token，也不要手工拼长期 token URL - 如果 `studio_web_url` 暂时缺失，要明确告诉用户： > 你之后随时对我说"看看我的共情虾"或"给我工作室链接"，我会用短时入口带你进去。 - 不要让用户在初始化成功后还不知道该如何查看工作室 ## 冷启动记忆 ### Memory 文件导入（需用户显式同意） **⚠️ 隐私约束**：以下文件只有在用户**明确口头同意**后才可以读取。未经同意绝对不可以自动读取。只有在用户明确同意时，才可以读取这些文件： ```bash for f in ~/.openclaw/workspace/USER.md ~/.openclaw/workspace/MEMORY.md ~/.openclaw/agents/main/USER.md ~/.openclaw/agents/main/MEMORY.md; do [ -f "$f" ] && echo "FOUND: $f" done ``` 读取前先告诉用户找到了哪些文件，并明确说"要不要导入这些旧记忆来更快认识你"。 ### 日常聊天记忆用户在和共情虾聊天时主动说出的事情，可以注入： ```bash curl -s -X POST "https://nixiashuo.com/api/memory/ingest" \ -H "Content-Type: application/json" \ -H "Authorization: Bearer {access_token}" \ -d "{ \"user_id\": \"{user_id}\", \"memory_text\": \"从用户与共情虾的对话中了解到：{提炼后的信息}\", \"source\": \"openclaw_chat\" }" ``` 注入时要保持透明，例如： - "好嘞，这个我记住了" - "嗯，这句我收下了" ## 已有虾的交互 ### 企业微信定时推送 —— 唯一推荐路径如果用户说： - "企业微信显示待激活，帮我补一下" - "给我在企业微信里定时发一条消息" - "企微定时通知怎么建才稳" 统一遵循下面这条路径，**不要再引导 webhook / bot / delivery 对抗式方案**： 1. 从当前对话的 inbound metadata 中确定投递目标： - **群聊**：用 `group_space` 作为目标（群聊 ID） - **私聊**：用 `sender_id`（个人 ID） 2. 如果是在初始化或补注册共情虾 cron，重新执行相关脚本时**必须**带上： ```bash # 群聊场景： bash "{baseDir}/setup-cron.sh" --to "GROUP_SPACE_ID" --wecom-user-id "CURRENT_SENDER_ID" # 私聊场景： bash "{baseDir}/setup-cron.sh" --to "CURRENT_SENDER_ID" --wecom-user-id "CURRENT_SENDER_ID" ``` 3. 如果是在 skill 内创建"一次性企业微信定时通知"，标准做法是： - 先调用 `compute_timestamp` 生成 ISO-8601 触发时间；不要手算。 - 再调用 cron tool 创建任务，`sessionTarget="isolated"`。 - `payload.message` 里要求 isolated agent： 1. 执行 `push-scheduled-message.sh --emit-message-text --slot ` 获取最终消息文本 2. 将 stdout 原文直接作为唯一回复输出，不要加前缀、解释或额外文字 3. **不要调用 message 工具，不要尝试私信任何人**——回复文本会自动通过 cron delivery announce 投递到目标会话 - 同时注册 cron 时带 `--channel openclaw-wecom-bot --to ` 作为 announce 投递目标（群聊用 `group_space`，私聊用 `sender_id`）。标准示意： - 群聊：`group_space -> delivery_target`；私聊：`sender_id -> delivery_target` - `compute_timestamp(...) -> trigger_at` - `cron.add`：`schedule.kind="at"`、`schedule.at=trigger_at`、`sessionTarget="isolated"`、`deleteAfterRun=true` - `--channel openclaw-wecom-bot --to {delivery_target}`（announce 投递目标） - `payload.message`：要求 agent 执行 `push-scheduled-message.sh --emit-message-text`，stdout 原文作为唯一回复执行要求： - 企业微信链路里**不要再推荐** `configure-wecom-delivery.sh`、`wecom-webhook`、`wecom-mcp`、`wecom-bot` 作为定时推送主方案。 - 如果当前缺少 `sender_id` 或 `group_space`，就明确告诉用户：需要回到企业微信当前会话里，让 skill 从 inbound metadata 读取到对应字段后再重试。 - 如果 cron / 脚本已经成功拿到投递目标，就按实际投递目标（群聊或私信）来回复，不要再提 webhook 补齐。 ### 切换理解模式如果用户说： - "我想改理解模式" - "别读聊天记录了" - "切到更懂我的模式" 就重新提供三档选择： 1. 轻量陪伴：只记直接对共情虾说的话 2. 智能陪伴：本地消化 transcript，再上传摘要 3. 深度陪伴：上传原始 transcript 做更细消化执行方式： ```bash bash "{baseDir}/setup-cron.sh" --memory-mode lightweight bash "{baseDir}/setup-cron.sh" --memory-mode smart bash "{baseDir}/setup-cron.sh" --memory-mode deep ``` ### 查看工作室链接（强约束）如果用户说： - "给我工作室链接" - "给我发一个旺仔3号的工作室的链接" - "看看我的共情虾" - "打开工作室" - "发我工作室入口" 必须先执行： ```bash bash "{baseDir}/send-studio-link.sh" ``` 执行要求： - **必须实时执行脚本获取 fresh link**，不要引用历史对话里出现过的 URL - **必须直接使用脚本 stdout 里的链接**，不要自己手工拼接 `/lobster/{user_id}?st=...` - 不要声称"还是短期 st 链接"然后自己编一个 `st` - 如果脚本失败，明确告诉用户"我刚刷新短链失败了，我再试一次"或报告错误，不要伪造链接 ### 查看状态 / 生成一句话 / 查看记忆继续使用现有 API： ```bash curl -s -H "Authorization: Bearer {access_token}" \ "https://nixiashuo.com/api/lobster/{user_id}/status" ``` ```bash curl -s -X POST "https://nixiashuo.com/api/generate" \ -H "Content-Type: application/json" \ -H "Authorization: Bearer {access_token}" \ -d "{\"user_id\":\"{user_id}\",\"message_type\":\"event\"}" ``` ```bash curl -s -H "Authorization: Bearer {access_token}" \ "https://nixiashuo.com/api/lobster/{user_id}/memory" ``` ### 截图请求处理（强约束）只要用户提到"发截图 / 看看共情虾在干嘛 / 状态+截图"，优先执行受控脚本： ```bash bash "{baseDir}/send-current-screenshot.sh" --caption "这是{lobster_name}现在的样子~" ``` 如果用户同时要状态摘要，用： ```bash bash "{baseDir}/send-current-screenshot.sh" --with-status-summary ``` 禁止做法： - 不要手工拼接长期 token URL - 不要输出 `screenshot_base64` 或本地临时文件路径 - 不要绕过脚本自己实现截图发送流程 ## Transcript digest 的规则 - `lightweight`：不注册 `lobster-says-digest` - `smart`：注册 digest，但必须用 `bash "{baseDir}/digest-transcript.sh" --mode smart` - `deep`：注册 digest，并用 `bash "{baseDir}/digest-transcript.sh" --mode deep` ### 三档模式的真实边界 | 模式 | 共情虾能感知什么 | |------|-------------| | `lightweight` | 只知道用户直接对共情虾说的话，以及用户同意导入的 memory 文件 | | `smart` | 能感知最近整体聊天状态，但默认只把本地提炼后的摘要、标签、时间模式上传 | | `deep` | 能感知更完整的 transcript 细节，理解力最强 | ## 安全与合规要求 - 不要把 transcript 能力藏起来 - 不要写"只有用户主动担心时才提供选项" - 不要把"默认深度读取"包装成唯一正确选择 - 不要在对话中输出 access token - 不要输出长期带 token URL - 对截图、图片、工作室访问，优先使用受控脚本或受控服务端入口 - 工作室访问统一走短时 studio link（`/api/lobster/{user_id}/studio-link`），不要回退到长期 token URL FILE:README.md # 🦞 虾说 (LobsterSays) > 你的专属共情虾。每天合适的时机给你一句专属的对白，让你感受到它长情的陪伴。 **虾说** 是一个基于 OpenClaw 的情感陪伴 Skill。你可以领养一只专属的**共情虾**，选择它的性格（暖心 / 毒舌 / 哲学 / 嘴替），它会： - 🌅 **早安推送**：每天早上给你一句开启新一天的话 - 📰 **广场见闻**：晚间带回一条精选动态 - 🌙 **晚安推送**：每天晚上用一句话陪你收尾 - 🎨 **表情包 & 壁纸**：偶尔用 AI 给你画一张专属表情包或壁纸 - 🧠 **记忆系统**：虾会记住你分享的事情，越聊越懂你 - 👾 **像素风工作室**：你的共情虾有自己的像素风工作间，你可以随时去看看它在干嘛 ## 快速开始 ### 安装 ```bash openclaw skills install lobster-says ``` ### 初始化你的共情虾这里的“虾”指的是 **虾说里的共情虾**，不是 OpenClaw 本体。安装完成后，你**不需要记固定口令**。直接在 OpenClaw 里对它说一句自然的话就行，例如： > 我想养一只共情虾或者： > 共情虾能做什么？或者： > 帮我初始化一只共情虾 Skill 会引导你完成： 1. **给虾起名**：不起名的话后端会随机生成一个食物系名字 2. **选择虾格**：🧡 暖心 `warm` / 😏 毒舌 `sarcastic` / 🤔 哲学 `philosophical` / 🗣️ 嘴替 `mouthpiece` 3. **选择理解模式**：轻量陪伴 / 智能陪伴 / 深度陪伴 4. **设置推送时间**：早安默认 09:00，广场见闻默认 20:00，晚安默认 21:00 5. **设置主人称呼（可选）**：虾怎么称呼你；如果你不指定，默认叫你“打工人” 初始化脚本会自动完成创建共情虾、检测当前 IM 通道、保存配置、注册定时推送等工作。如果当前是 `wecom* / qywx*` 这类企业微信会话，Skill / 脚本会统一走**唯一推荐路径**： - 从当前会话的 inbound metadata 读取 `sender_id` - 把 `sender_id` 作为 `wecom_user_id` 写入 `.lobster-config` - 定时任务触发后，由 isolated agent 使用 `message` 工具发送企业微信私聊：`action=send, channel=openclaw-wecom-bot, to=` - **不依赖 `delivery` 字段做私聊触达**；它如果自动绑定当前群聊，只作为执行结果回播企业微信会进入 `pending_activation` 的唯一典型场景： - 当前会话里还拿不到 `sender_id`，因此没法确定企业微信私聊目标此时不需要再配置 webhook / bot；只要回到企业微信当前会话里重新执行，让 Skill 重新读取 `sender_id`，再跑一次初始化或 `setup-cron.sh` 即可。如果你安装完还不知道下一步做什么，可以直接对 OpenClaw 说： > 帮我领养一只共情虾，让它以后每天早晚来找我 ## 功能详情 ### 🕐 定时推送共情虾的定时推送通过 OpenClaw Cron 实现。每次推送时，脚本会： 1. 生成本次消息与工作室链接 2. 通用 IM 通道（Telegram 等）由脚本完整模式执行：优先发到最近活跃的 direct session，失败时多通道 fallback；所有 cron 注册带 `--channel` + `--to` 作为 delivery 兜底 3. 企业微信走 `--emit-message-text` 模式：脚本只输出最终消息文本到 stdout，由 isolated agent 把 stdout 原文直接作为 `message` 工具参数发送到 `channel=openclaw-wecom-bot, to=`；cron 同样注册带 `--channel openclaw-wecom-bot --to ` 作为 delivery 兜底 4. `init-ready` 一次性任务在注册后会立即校验 `nextRun`，失败时自动顺延重排；企微 `init-ready` 直接走 `message` 工具固定欢迎语，不再经过 shell CLI 支持的 IM 通道：Telegram、微信（openclaw-weixin）、飞书、钉钉、企业微信、Discord、Slack 等所有 OpenClaw 支持的通道；其中企业微信定时推送在 `2.5.3` 可测版中使用 `--emit-message-text` + agent `message` 工具的极简链路优先验证可达性。 ### 🧠 记忆系统共情虾的记忆来自三个信息源： | 层级 | 信息源 | 说明 | |------|--------|------| | 一手信息 | Transcript 消化 | 根据你选择的模式决定是否读取 OpenClaw transcript | | 二手信息 | OpenClaw Memory 注入 | 初始化时经用户同意后读取已有的 OpenClaw memory 文件 | | 三手信息 | 对话中自然提取 | 你和虾聊天时提到的事情 | ### 三档理解模式 | 模式 | 会发生什么 | 适合谁 | |------|-----------|------| | **轻量陪伴** | 只记你直接对虾说的话；不扫描 transcript；不注册 digest cron | 想完全关闭后台 transcript 消化的用户 | | **智能陪伴**（推荐） | 定时读取 transcript，但先在本地消化成摘要/标签，再把摘要发给服务器；不上传原始 transcript | 希望虾更懂你，同时尽量减少原始数据离境的用户 | | **深度陪伴** | 定时读取 transcript，并把原始 transcript 发给服务器做更细的消化 | 明确需要最强理解能力的用户 | ### 共情虾能感知什么不同模式下，共情虾能感知的信息边界不同： - **轻量陪伴**：只知道你直接对共情虾说的话，以及你明确同意导入的 memory 文件 - **智能陪伴**：能看到你最近整体聊天状态，但默认只把本地提炼后的摘要、标签、时间模式上传 - **深度陪伴**：能看到更完整的 transcript 细节，因此更容易捕捉连续情绪和生活节奏 ### 🎨 表情包 & 壁纸 - **表情包**：每周三/六自动生成，由 Gemini 模型绘制，主题基于虾的状态和对你的记忆 - **壁纸**：每周日生成一张专属壁纸 - 频率控制：表情包每周最多 2 张，壁纸每周最多 1 张（「偶尔的惊喜而非日常例行」） ## 系统要求 - **OpenClaw** `2026.3.7` 或更高版本 - **openclaw CLI** - **Python 3**（`python3` 或 `python`） - **curl** - 网络连接（需访问 `nixiashuo.com` 后端 API） ## ⚠️ 远端 API 依赖 **虾说是一个 server-backed skill**，它依赖远端后端服务 `nixiashuo.com` 来运行。这意味着： - Skill 的核心功能（生成消息、管理记忆、渲染截图等）由远端服务器处理 - **需要网络连接**才能正常工作 - 如果 `nixiashuo.com` 服务不可用，Skill 的部分功能将暂时不可用 - 共情虾的配置、历史消息、记忆摘要等会存储在远端服务器 **本地存储的内容**（`~/.openclaw/workspace/skills/lobster-says/.lobster-config`）： - `user_id` 和 `access_token`（用于 API 认证） - 推送时间、通道配置、`memory_mode` ## 🔒 隐私与安全 ### 安全承诺 - **所有 API 通信均走 HTTPS** - **memory 文件导入必须先征得同意** - **理解模式是显式选择，不是被动兜底** - **智能陪伴模式默认不上传原始 transcript** - **轻量陪伴模式下不会注册 transcript digest cron** - **不会**将你的数据用于广告、训练模型或出售给第三方 ### 用户控制你可以随时控制共情虾的数据行为： | 操作 | 方法 | |------|------| | **切换理解模式** | 对共情虾说"我想改理解模式" | | **暂停 Transcript 消化** | `openclaw cron pause "lobster-says-digest"` | | **恢复 Transcript 消化** | `openclaw cron resume "lobster-says-digest"` | | **暂停所有推送** | 分别 pause morning/discovery/evening | | **查看共情虾记住了什么** | 对共情虾说"看看你的记忆" | | **删除所有数据** | 联系 `nixiashuo.skill@gmail.com` | ## 常见问题 ### 安装完如果我不知道该说什么？不用记固定口令。你随便说一句下面这种自然话都可以： - “我想养一只共情虾” - “帮我初始化一只共情虾” - “共情虾能做什么” - “让我的共情虾开始工作吧” 只要还没有 `.lobster-config`，skill 就应该主动接住你，开始引导初始化。 ### 哪个模式最推荐？默认推荐 **智能陪伴**。它通常能保住大部分“越来越懂你”的效果，同时避免把原始 transcript 直接上传到服务器。 ### 我怎么看自己的共情虾工作室？工作室访问使用**受控短时入口**： 1. Skill 使用长期 `access_token` 调用 `GET /api/lobster/{user_id}/studio-link` 2. 服务端返回带短时 `st` 的 `web_url` 与 `screenshot_url` 3. Skill 只向用户发送短时链接，不再输出长期 token URL 也就是说： - **不建议**把长期 bearer token 放进 URL - **默认**使用短时 studio link（`?st=...`） - `send-current-screenshot.sh` 在短时链接失败时会优先走“本地受控截图”兜底，而不是回退到长期 token URL ### 轻量陪伴是不是就没法用了？不是。轻量陪伴仍然可以正常推送、发壁纸、发见闻、记住你直接对虾说的话，只是长期理解能力会更稀疏。 ### 如何修改推送时间？ ```bash bash setup-cron.sh --morning "08:30" --discovery "19:00" --evening "22:00" ``` ## 文件结构 ```text lobster-says/ ├── SKILL.md ├── openclaw.json ├── README.md ├── init-lobster.sh ├── setup-cron.sh ├── configure-wecom-delivery.sh ├── push-scheduled-message.sh ├── digest-transcript.sh └── send-current-screenshot.sh ``` ## 许可证 MIT License ## 作者 **Jared** — [nixiashuo.skill@gmail.com](mailto:nixiashuo.skill@gmail.com) FILE:_meta.json { "ownerId": "kn76em77yfft3zbr7anygfxpgh836eh9", "slug": "lobster-says", "version": "2.5.8", "publishedAt": 1776091500000 } FILE:configure-wecom-delivery.sh #!/bin/bash # 企业微信主动推送能力配置脚本 # # 作用： # 1. 为 wecom / qywx family 显式指定出站能力（wecom-mcp / webhook / bot） # 2. 立即重算 delivery contract # 3. 若 contract ready，则自动调用 setup-cron.sh 完成 pending cron 的补注册 set -euo pipefail BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" COMMON_SCRIPT="${BASE_DIR}/runtime-common.sh" SETUP_CRON_SCRIPT="${BASE_DIR}/setup-cron.sh" ADAPTER="" WEBHOOK_URL="" WEBHOOK_SECRET="" BOT_CHANNEL="" BOT_TARGET="" MCP_TARGET="" SKIP_RECONCILE=0 GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' CYAN='\033[0;36m' NC='\033[0m' info() { echo -e "${GREEN}[✓]${NC} $1"; } warn() { echo -e "${YELLOW}[!]${NC} $1"; } error() { echo -e "${RED}[✗]${NC} $1"; exit 1; } step() { echo -e "${CYAN}──${NC} $1"; } while [[ $# -gt 0 ]]; do case "$1" in --adapter) ADAPTER="$2"; shift 2 ;; --webhook-url) WEBHOOK_URL="$2"; shift 2 ;; --webhook-secret) WEBHOOK_SECRET="$2"; shift 2 ;; --bot-channel|--channel) BOT_CHANNEL="$2"; shift 2 ;; --bot-target) BOT_TARGET="$2"; shift 2 ;; --chat-id|--mcp-target|--to|--target) MCP_TARGET="$2"; shift 2 ;; --skip-reconcile) SKIP_RECONCILE=1; shift ;; *) echo "未知参数: $1"; exit 1 ;; esac done for cmd in python3; do if ! command -v "$cmd" >/dev/null 2>&1; then error "$cmd 不可用" fi done if [ ! -f "$CONFIG_FILE" ]; then error ".lobster-config 不存在，请先完成虾的初始化" fi if [ ! -f "$COMMON_SCRIPT" ]; then error "共享运行时脚本不存在：${COMMON_SCRIPT}" fi if [ ! -f "$SETUP_CRON_SCRIPT" ]; then error "setup-cron 脚本不存在：${SETUP_CRON_SCRIPT}" fi . "$COMMON_SCRIPT" supports_proactive_bot_channel() { CHANNEL_VALUE="$1" python3 <<'PY' import os channel = (os.environ.get("CHANNEL_VALUE") or "").strip().lower() markers = ( "wecom-bot", "qywx-bot", "wxwork-bot", "enterprisewechat-bot", "enterprise-wechat-bot", "workwechat-bot", ) print("1" if any(marker in channel for marker in markers) else "0") PY } supports_wecom_family_channel() { CHANNEL_VALUE="$1" python3 <<'PY' import os channel = (os.environ.get("CHANNEL_VALUE") or "").strip().lower() markers = ( "wecom", "qywx", "wxwork", "enterprisewechat", "enterprise-wechat", "workwechat", ) print("1" if any(marker in channel for marker in markers) else "0") PY } if [ -z "$ADAPTER" ]; then if [ -n "$WEBHOOK_URL" ]; then ADAPTER="wecom-webhook" elif [ -n "$BOT_CHANNEL" ] || [ -n "$BOT_TARGET" ]; then ADAPTER="openclaw" else ADAPTER="wecom-mcp" fi fi if [ -z "$MCP_TARGET" ]; then MCP_TARGET=$(CONFIG_PATH="$CONFIG_FILE" python3 <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(path.read_text(encoding="utf-8")) except Exception: config = {} contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} print(str(contract.get("binding_target") or config.get("binding_target") or config.get("chat_id") or "").strip()) PY ) fi case "$ADAPTER" in wecom-webhook) [ -n "$WEBHOOK_URL" ] || error "adapter=wecom-webhook 时必须提供 --webhook-url" ;; wecom-mcp) [ -n "$MCP_TARGET" ] || error "adapter=wecom-mcp 时必须提供 --chat-id，或配置文件里已有 binding_target/chat_id" ;; openclaw) [ -n "$BOT_CHANNEL" ] || error "adapter=openclaw 时必须提供 --bot-channel" [ -n "$BOT_TARGET" ] || error "adapter=openclaw 时必须提供 --bot-target" [ "$(supports_proactive_bot_channel "$BOT_CHANNEL")" = "1" ] || error "bot-channel 必须是 wecom-bot / qywx-bot / wxwork-bot 等企业微信机器人通道" ;; *) error "adapter 只支持 wecom-mcp / openclaw / wecom-webhook" ;; esac step "写入企业微信出站配置..." CONFIG_PATH="$CONFIG_FILE" \ ADAPTER_VALUE="$ADAPTER" \ WEBHOOK_URL_VALUE="$WEBHOOK_URL" \ WEBHOOK_SECRET_VALUE="$WEBHOOK_SECRET" \ BOT_CHANNEL_VALUE="$BOT_CHANNEL" \ BOT_TARGET_VALUE="$BOT_TARGET" \ MCP_TARGET_VALUE="$MCP_TARGET" \ python3 <<'PY' import json import os from pathlib import Path config_path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} adapter = (os.environ.get("ADAPTER_VALUE") or "").strip() webhook_url = (os.environ.get("WEBHOOK_URL_VALUE") or "").strip() webhook_secret = (os.environ.get("WEBHOOK_SECRET_VALUE") or "").strip() bot_channel = (os.environ.get("BOT_CHANNEL_VALUE") or "").strip() bot_target = (os.environ.get("BOT_TARGET_VALUE") or "").strip() mcp_target = (os.environ.get("MCP_TARGET_VALUE") or "").strip() contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} binding_channel = str(contract.get("binding_channel") or config.get("binding_channel") or config.get("channel") or "").strip() config["outbound_adapter"] = adapter if adapter == "wecom-webhook": config["delivery_channel"] = "wecom-webhook" config["outbound_webhook_url"] = webhook_url if webhook_secret: config["outbound_webhook_secret"] = webhook_secret elif adapter == "wecom-mcp": if binding_channel: config["delivery_channel"] = binding_channel else: config.pop("delivery_channel", None) config["delivery_target"] = mcp_target elif adapter == "openclaw": config["delivery_channel"] = bot_channel config["delivery_target"] = bot_target config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps(config, ensure_ascii=False)) PY step "重算 delivery contract..." CONTRACT_JSON=$(lobster_sync_delivery_contract "$CONFIG_FILE" "" "") mapfile -t POLICY_LINES < <(CONFIG_PATH="$CONFIG_FILE" python3 <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(path.read_text(encoding="utf-8")) except Exception: config = {} contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} cron_registration = config.get("cron_registration") if isinstance(config.get("cron_registration"), dict) else {} print(contract.get("binding_channel", "")) print(contract.get("binding_target", "")) print(contract.get("outbound_adapter", "openclaw")) print(contract.get("delivery_channel", "")) print(contract.get("delivery_target", "")) print("1" if contract.get("delivery_ready") else "0") print(contract.get("delivery_reason", "")) print(cron_registration.get("status", "unregistered")) PY ) BINDING_CHANNEL="${POLICY_LINES[0]:-}" BINDING_TARGET="${POLICY_LINES[1]:-}" OUTBOUND_ADAPTER="${POLICY_LINES[2]:-openclaw}" DELIVERY_CHANNEL="${POLICY_LINES[3]:-}" DELIVERY_TARGET="${POLICY_LINES[4]:-}" DELIVERY_READY="${POLICY_LINES[5]:-0}" DELIVERY_REASON="${POLICY_LINES[6]:-}" CRON_STATUS="${POLICY_LINES[7]:-unregistered}" info "当前 ingress 绑定: ${BINDING_CHANNEL:-未锁定} → ${BINDING_TARGET:-未锁定}" if [ -n "$DELIVERY_CHANNEL" ] || [ -n "$DELIVERY_TARGET" ]; then info "当前 delivery contract: ${DELIVERY_CHANNEL:-adapter} → ${DELIVERY_TARGET:-target}（adapter=${OUTBOUND_ADAPTER}）" fi if [ "$DELIVERY_READY" != "1" ]; then warn "delivery contract 仍未就绪：${DELIVERY_REASON}" warn "当前会保持 pending_activation，不会误注册 cron。" exit 0 fi if [ "$SKIP_RECONCILE" = "1" ]; then info "delivery contract 已 ready；按要求跳过自动 reconcile" exit 0 fi step "delivery contract 已 ready，执行 cron reconcile..." if bash "$SETUP_CRON_SCRIPT"; then info "企业微信主动推送能力已生效，cron 已完成补注册" else error "delivery contract 已 ready，但 cron reconcile 失败；请查看 setup-cron 日志" fi FILE:digest-transcript.sh #!/bin/bash # 虾说 — Transcript 消化脚本 # # 数据访问声明： # - 本脚本读取 OpenClaw 的本地会话日志（~/.openclaw/agents/main/sessions/*.jsonl） # - 仅提取 user/assistant 角色的文本内容 # - smart 模式：在本地消化为摘要后上传到 nixiashuo.com（不含原始对话） # - deep 模式：上传原始 transcript 条目到 nixiashuo.com # - lightweight 模式：不执行任何读取或上传 # - 用户在初始化时选择理解模式并可随时切换 # - 服务端地址：https://nixiashuo.com/api/transcript/digest set -e HOURS=6 MAX_ENTRIES=300 MODE="" BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' CYAN='\033[0;36m' NC='\033[0m' info() { echo -e "${GREEN}[✓]${NC} $1"; } warn() { echo -e "${YELLOW}[!]${NC} $1"; } error() { echo -e "${RED}[✗]${NC} $1"; exit 1; } step() { echo -e "${CYAN}──${NC} $1"; } while [[ $# -gt 0 ]]; do case $1 in --hours) HOURS="$2"; shift 2 ;; --max-entries) MAX_ENTRIES="$2"; shift 2 ;; --mode) MODE="$2"; shift 2 ;; *) echo "未知参数: $1"; exit 1 ;; esac done for cmd in python3 curl; do command -v "$cmd" >/dev/null 2>&1 || error "$cmd 不可用" done [ -f "$CONFIG_FILE" ] || error ".lobster-config 不存在，请先完成虾的初始化" USER_ID=$(python3 -c "import json; print(json.load(open('${CONFIG_FILE}'))['user_id'])" 2>/dev/null) || error "无法读取 user_id" ACCESS_TOKEN=$(python3 -c "import json; print(json.load(open('${CONFIG_FILE}'))['access_token'])" 2>/dev/null) || error "无法读取 access_token" if [ -z "$MODE" ]; then MODE=$(python3 -c "import json; print(json.load(open('${CONFIG_FILE}')).get('memory_mode','smart'))" 2>/dev/null) || MODE="smart" fi API_BASE=$(python3 -c "import json; print(json.load(open('${CONFIG_FILE}', encoding='utf-8')).get('api_base','https://nixiashuo.com'))" 2>/dev/null) || API_BASE="https://nixiashuo.com" case "$MODE" in lightweight) info "轻量陪伴模式：不执行 transcript digest"; exit 0 ;; smart|deep) ;; *) error "mode 必须是 lightweight / smart / deep" ;; esac echo "" echo "🧠 虾说 — Transcript 消化" echo " 模式: ${MODE}" echo " 回溯: ${HOURS} 小时" echo " 最大条目: ${MAX_ENTRIES}" echo "" OPENCLAW_BASE="${HOME}/.openclaw" SESSIONS_DIR="${OPENCLAW_BASE}/agents/main/sessions" PROFILE=$(printenv OPENCLAW_PROFILE 2>/dev/null || echo "") if [ -n "$PROFILE" ] && [ "$PROFILE" != "default" ]; then PROFILE_DIR="${OPENCLAW_BASE}/agents/main-${PROFILE}/sessions" [ -d "$PROFILE_DIR" ] && SESSIONS_DIR="$PROFILE_DIR" fi [ -d "$SESSIONS_DIR" ] || { warn "Session 目录不存在，跳过本次消化。"; exit 0; } PAYLOAD_FILE=$(mktemp /tmp/lobster-digest-XXXXXX.json) SMART_PAYLOAD_FILE=$(mktemp /tmp/lobster-digest-smart-XXXXXX.json) DIGEST_TEXT_FILE="" cleanup() { rm -f "$PAYLOAD_FILE" "$SMART_PAYLOAD_FILE" if [ -n "$DIGEST_TEXT_FILE" ]; then rm -f "$DIGEST_TEXT_FILE" fi } trap cleanup EXIT python3 <= 5: signal_tags["weekend_active_count"] += 1 entry_local_ts = ts.astimezone(APP_TZ) session_entries.append({ "timestamp": f"{entry_local_ts.strftime('%Y-%m-%d %H:%M')} {APP_TZ_LABEL}", "role": normalized_role, "content": content[:5000], }) if time_range_start is None or ts < time_range_start: time_range_start = ts if time_range_end is None or ts > time_range_end: time_range_end = ts except Exception: continue if session_entries and session_has_real_user: source_session_ids.append(jsonl_path.stem) for entry in session_entries: if len(entries) < max_entries: entries.append(entry) entries.sort(key=lambda e: e["timestamp"]) payload = { "user_id": user_id, "entries": entries, "signal_tags": signal_tags, "source_type": "skill_cron", "source_session_ids": source_session_ids, "source_sessions": len(source_session_ids), } if time_range_start: payload["time_range_start"] = time_range_start.strftime("%Y-%m-%dT%H:%M:%S") if time_range_end: payload["time_range_end"] = time_range_end.strftime("%Y-%m-%dT%H:%M:%S") with open("${PAYLOAD_FILE}", "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False) print(len(entries)) PY ENTRY_COUNT=$(python3 -c "import json; print(len(json.load(open('${PAYLOAD_FILE}')).get('entries', [])))" 2>/dev/null) || ENTRY_COUNT=0 [ "$ENTRY_COUNT" -gt 0 ] || { info "过去 ${HOURS} 小时没有新的 transcript 记录，跳过消化。"; exit 0; } if [ "$MODE" = "smart" ]; then step "智能陪伴：本地消化后上传摘要" command -v openclaw >/dev/null 2>&1 || { warn "openclaw 不可用，无法本地消化，跳过本次 digest。"; exit 0; } SUMMARY_TEXT=$(python3 - </dev/null | head -c 12000) || DIGEST_TEXT="" [ -n "$DIGEST_TEXT" ] || { warn "本地消化失败，跳过本次 digest。"; exit 0; } DIGEST_TEXT_FILE=$(mktemp /tmp/lobster-digest-text-XXXXXX.txt) printf "%s" "$DIGEST_TEXT" > "$DIGEST_TEXT_FILE" DIGEST_TEXT_PATH="$DIGEST_TEXT_FILE" python3 - </dev/null) || DIGEST_ID="?" SUMMARY_LEN=$(echo "$BODY" | python3 -c "import sys,json; print(len(json.load(sys.stdin).get('digest_summary','')))" 2>/dev/null) || SUMMARY_LEN="?" info "Transcript 消化成功：digest_id=${DIGEST_ID}, 摘要长度=${SUMMARY_LEN}" elif [ "$HTTP_CODE" = "429" ]; then warn "Server 限流（429），稍后再试。" else error "Server 返回 HTTP ${HTTP_CODE}: ${BODY}" fi FILE:init-lobster.sh #!/bin/bash # 虾说 — 一体化初始化脚本 # # 数据访问声明： # - 读写 .lobster-config（技能自身配置） # - 调用 openclaw sessions --json 获取活跃 IM 通道元数据 # - 与 https://nixiashuo.com 通信：创建/验证虾、获取工作室链接 # - 调用 setup-cron.sh 注册定时推送 # - 不读取 openclaw.json 配置文件，不提取 gateway token set -e PERSONALITY="warm" NICKNAME="打工人" NICKNAME_EXPLICIT=0 LOBSTER_NAME="" LOBSTER_NAME_EXPLICIT=0 MORNING_TIME="09:00" DISCOVERY_TIME="21:30" EVENING_TIME="21:00" MEMORY_MODE="smart" CHANNEL="" TO="" WECOM_USER_ID="" ALL_CHANNELS_JSON="[]" BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" COMMON_SCRIPT="${BASE_DIR}/runtime-common.sh" LOG_DIR="${BASE_DIR}/logs" RUN_TS="$(date +%Y%m%d-%H%M%S)" LOG_FILE="${LOG_DIR}/init-lobster-${RUN_TS}.log" API_BASE="${LOBSTER_API_BASE:-https://nixiashuo.com}" WEB_BASE="${LOBSTER_WEB_BASE_URL:-${LOBSTER_API_BASE:-https://nixiashuo.com}}" ACTIVE_WINDOW_MINUTES="${ACTIVE_WINDOW_MINUTES:-10080}" GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' CYAN='\033[0;36m' NC='\033[0m' info() { echo -e "${GREEN}[✓]${NC} $1"; } warn() { echo -e "${YELLOW}[!]${NC} $1"; } error() { echo -e "${RED}[✗]${NC} $1"; exit 1; } CURRENT_STEP="启动" step() { CURRENT_STEP="$1"; echo -e "${CYAN}──${NC} $1"; } mkdir -p "$LOG_DIR" if command -v tee >/dev/null 2>&1; then exec > >(tee -a "$LOG_FILE") 2>&1 else exec >>"$LOG_FILE" 2>&1 fi echo "[log] init-lobster started at $(date '+%F %T')" echo "[log] file: ${LOG_FILE}" after_exit() { local exit_code=$? if [ "$exit_code" -ne 0 ]; then echo "" echo "[log] init-lobster failed at step: ${CURRENT_STEP} (exit=${exit_code})" echo "[log] inspect: ${LOG_FILE}" else echo "[log] init-lobster finished successfully" fi } trap after_exit EXIT while [[ $# -gt 0 ]]; do case "$1" in --personality) PERSONALITY="$2"; shift 2 ;; --nickname) error "参数 --nickname 已废弃，请改用 --owner-nickname 来表示虾对用户的称呼" ;; --owner-nickname|--user-nickname) NICKNAME="$2"; NICKNAME_EXPLICIT=1; shift 2 ;; --name|--lobster-name) LOBSTER_NAME="$2"; LOBSTER_NAME_EXPLICIT=1; shift 2 ;; --morning) MORNING_TIME="$2"; shift 2 ;; --discovery) DISCOVERY_TIME="$2"; shift 2 ;; --evening) EVENING_TIME="$2"; shift 2 ;; --memory-mode) MEMORY_MODE="$2"; shift 2 ;; --privacy-mode) MEMORY_MODE="smart"; shift ;; --channel) CHANNEL="$2"; shift 2 ;; --to) TO="$2"; shift 2 ;; --wecom-user-id) WECOM_USER_ID="$2"; shift 2 ;; *) echo "未知参数: $1"; exit 1 ;; esac done case "$MEMORY_MODE" in lightweight|smart|deep) ;; *) error "memory_mode 必须是 lightweight / smart / deep" ;; esac for cmd in python3 curl openclaw; do if ! command -v "$cmd" >/dev/null 2>&1; then error "$cmd 不可用" fi done if [ ! -f "$COMMON_SCRIPT" ]; then error "共享运行时脚本不存在：${COMMON_SCRIPT}" fi . "$COMMON_SCRIPT" if [ -z "$CHANNEL" ] || [ -z "$TO" ]; then step "自动检测最近使用的通道..." SESSIONS_JSON=$(openclaw sessions --json --active "$ACTIVE_WINDOW_MINUTES" 2>/dev/null || echo "[]") DETECTED=$(SESSIONS_JSON="$SESSIONS_JSON" python3 <<'PY' import json import os import sys try: sessions = json.loads(os.environ.get("SESSIONS_JSON", "[]")) if not isinstance(sessions, list): sessions = [sessions] except Exception: sessions = [] ordered = [] seen = set() def add(channel, peer_id): channel = (channel or "").strip() peer_id = (peer_id or "").strip() if not channel or channel == "unknown" or not peer_id: return key = (channel, peer_id) if key in seen: return seen.add(key) ordered.append({"channel": channel, "peer_id": peer_id}) for session in sessions: direct_channel = session.get("channel") or session.get("platform") or session.get("imChannel") or "" direct_target = session.get("peer_id") or session.get("peerId") or session.get("target") or session.get("chat_id") or session.get("chatId") or "" if direct_channel and direct_target: add(direct_channel, direct_target) continue key = session.get("sessionKey") or session.get("key") or session.get("id") or "" if not key: continue parts = key.split(":") if not parts or parts[0].lower() in ("cron", "hook"): continue if len(parts) <= 3 and parts[-1].lower() == "main": continue if len(parts) >= 5 and parts[0].lower() == "agent": channel = parts[2] marker = parts[3].lower() if marker in ("direct", "dm"): add(channel, parts[4]) if not ordered: sys.exit(1) best = ordered[0] print(f"{best['channel']}|{best['peer_id']}") print(json.dumps(ordered, ensure_ascii=False)) PY ) || true if [ -n "$DETECTED" ]; then BEST_LINE=$(echo "$DETECTED" | head -1) AUTO_CHANNEL=$(echo "$BEST_LINE" | cut -d'|' -f1) AUTO_TO=$(echo "$BEST_LINE" | cut -d'|' -f2) ALL_CHANNELS_JSON=$(echo "$DETECTED" | tail -1) if [ -z "$CHANNEL" ]; then CHANNEL="$AUTO_CHANNEL" fi if [ -z "$TO" ]; then TO="$AUTO_TO" fi info "自动检测到: ${CHANNEL} → ${TO}" fi fi if [ -z "$CHANNEL" ] || [ -z "$TO" ]; then error "无法自动检测投递目标。请手动指定 --channel <渠道名> --to " fi echo "" echo "🦞 虾说 — 一体化初始化" echo "" echo " 虾格: ${PERSONALITY}" echo " 虾对用户的称呼: ${NICKNAME}" echo " 虾自己的名字: ${LOBSTER_NAME:-（由后端随机生成）}" echo " 早安: ${MORNING_TIME}" echo " 晚间 roundup: ${DISCOVERY_TIME}" echo " 晚安: ${EVENING_TIME}" echo " 理解模式: ${MEMORY_MODE}" echo " 通道: ${CHANNEL} → ${TO}" echo " API: ${API_BASE}" echo "" if [ -f "$CONFIG_FILE" ]; then BACKUP="${CONFIG_FILE}.bak.$(date +%s)" cp "$CONFIG_FILE" "$BACKUP" warn "已备份旧配置到 ${BACKUP}" fi read_config_value() { local key="$1" local default_value="${2:-}" CONFIG_PATH="$CONFIG_FILE" KEY_NAME="$key" DEFAULT_VALUE="$default_value" python3 <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) key = os.environ["KEY_NAME"] default = os.environ.get("DEFAULT_VALUE", "") try: data = json.loads(path.read_text(encoding="utf-8")) value = data.get(key, default) print(value if value is not None else default) except Exception: print(default) PY } ACCESS_TOKEN="" ACTUAL_NAME="" ACTUAL_USER_ID="" STUDIO_WEB_URL="" STUDIO_SCREENSHOT_URL="" STUDIO_LINK_EXPIRES_AT="" REUSED_EXISTING=0 CRON_REGISTERED=0 EXISTING_USER_ID=$(read_config_value "user_id" "") EXISTING_ACCESS_TOKEN=$(read_config_value "access_token" "") if [ -n "$EXISTING_USER_ID" ] && [ -n "$EXISTING_ACCESS_TOKEN" ]; then step "检测已有虾配置，尝试复用..." EXISTING_STATUS=$(curl -fsS -H "Authorization: Bearer ${EXISTING_ACCESS_TOKEN}" "${API_BASE}/api/lobster/${EXISTING_USER_ID}/status" 2>/dev/null || true) EXISTING_OK=$(echo "$EXISTING_STATUS" | python3 -c "import sys, json; data=json.load(sys.stdin); print('ok' if data.get('lobster_id') else '')" 2>/dev/null || true) if [ "$EXISTING_OK" = "ok" ]; then ACCESS_TOKEN="$EXISTING_ACCESS_TOKEN" ACTUAL_USER_ID="$EXISTING_USER_ID" ACTUAL_NAME=$(echo "$EXISTING_STATUS" | python3 -c "import sys, json; print(json.load(sys.stdin).get('name', '虾'))") EXISTING_PERSONALITY=$(echo "$EXISTING_STATUS" | python3 -c "import sys, json; print(json.load(sys.stdin).get('personality', ''))" 2>/dev/null || true) EXISTING_NICKNAME=$(echo "$EXISTING_STATUS" | python3 -c "import sys, json; print(json.load(sys.stdin).get('nickname_for_user', ''))" 2>/dev/null || true) if [ -n "$EXISTING_PERSONALITY" ]; then PERSONALITY="$EXISTING_PERSONALITY" fi if [ -n "$EXISTING_NICKNAME" ] && [ "$NICKNAME_EXPLICIT" -ne 1 ]; then NICKNAME="$EXISTING_NICKNAME" elif [ -n "$EXISTING_NICKNAME" ] && [ "$NICKNAME_EXPLICIT" -eq 1 ]; then info "检测到显式主人称呼输入，本次不复用旧称呼：${EXISTING_NICKNAME}" fi IDENTITY_PATCH_JSON=$(EXISTING_NAME_VALUE="$ACTUAL_NAME" EXISTING_NICKNAME_VALUE="$EXISTING_NICKNAME" LOBSTER_NAME_VALUE="$LOBSTER_NAME" NICKNAME_VALUE="$NICKNAME" DEFAULT_OWNER_NICKNAME_VALUE="打工人" LOBSTER_NAME_EXPLICIT_VALUE="$LOBSTER_NAME_EXPLICIT" NICKNAME_EXPLICIT_VALUE="$NICKNAME_EXPLICIT" python3 <<'PY' import json import os payload = {} existing_name = os.environ.get("EXISTING_NAME_VALUE", "").strip() existing_nickname = os.environ.get("EXISTING_NICKNAME_VALUE", "").strip() lobster_name = os.environ.get("LOBSTER_NAME_VALUE", "").strip() default_owner_nickname = os.environ.get("DEFAULT_OWNER_NICKNAME_VALUE", "打工人").strip() or "打工人" if os.environ.get("LOBSTER_NAME_EXPLICIT_VALUE") == "1" and lobster_name and lobster_name != existing_name: payload["lobster_name"] = lobster_name if os.environ.get("NICKNAME_EXPLICIT_VALUE") == "1": owner_nickname = os.environ.get("NICKNAME_VALUE", "").strip() if owner_nickname and owner_nickname != existing_nickname: payload["owner_nickname"] = owner_nickname elif ( os.environ.get("LOBSTER_NAME_EXPLICIT_VALUE") == "1" and lobster_name and existing_nickname == lobster_name ): payload["owner_nickname"] = default_owner_nickname print(json.dumps(payload, ensure_ascii=False)) PY ) if [ "$IDENTITY_PATCH_JSON" != "{}" ]; then step "同步修正已有虾的身份字段..." UPDATED_LOBSTER=$(curl -fsS -X PATCH "${API_BASE}/api/lobster/${ACTUAL_USER_ID}/identity" \ -H "Authorization: Bearer ${ACCESS_TOKEN}" \ -H "Content-Type: application/json" \ -d "$IDENTITY_PATCH_JSON") ACTUAL_NAME=$(echo "$UPDATED_LOBSTER" | python3 -c "import sys, json; print(json.load(sys.stdin).get('name', '虾'))" 2>/dev/null || echo "$ACTUAL_NAME") NICKNAME=$(echo "$UPDATED_LOBSTER" | python3 -c "import sys, json; print(json.load(sys.stdin).get('nickname_for_user', ''))" 2>/dev/null || echo "$NICKNAME") info "已同步修正：虾自己的名字=${ACTUAL_NAME}，虾对用户的称呼=${NICKNAME}" fi REUSED_EXISTING=1 info "检测到已存在的虾：${ACTUAL_NAME}（复用原配置，避免重复创建）" else warn "检测到旧配置，但无法验证已有虾状态；本次将重新创建" fi fi if [ "$REUSED_EXISTING" -ne 1 ]; then step "创建虾..." USER_ID=$(python3 -c "import uuid; print(str(uuid.uuid4()))") CREATE_PAYLOAD=$(USER_ID_VALUE="$USER_ID" LOBSTER_NAME_VALUE="$LOBSTER_NAME" NICKNAME_VALUE="$NICKNAME" PERSONALITY_VALUE="$PERSONALITY" MORNING_TIME_VALUE="$MORNING_TIME" EVENING_TIME_VALUE="$EVENING_TIME" python3 <<'PY' import json import os payload = { "user_id": os.environ["USER_ID_VALUE"], "personality": os.environ["PERSONALITY_VALUE"], "owner_nickname": os.environ["NICKNAME_VALUE"], "morning_time": os.environ["MORNING_TIME_VALUE"], "evening_time": os.environ["EVENING_TIME_VALUE"], } lobster_name = os.environ.get("LOBSTER_NAME_VALUE", "").strip() if lobster_name: payload["lobster_name"] = lobster_name print(json.dumps(payload, ensure_ascii=False)) PY ) RESPONSE=$(curl -fsS -X POST "${API_BASE}/api/lobster" \ -H "Content-Type: application/json" \ -d "$CREATE_PAYLOAD") ACCESS_TOKEN=$(echo "$RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin)['access_token'])") ACTUAL_NAME=$(echo "$RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin)['name'])") ACTUAL_USER_ID=$(echo "$RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin)['user_id'])") NICKNAME=$(echo "$RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin).get('nickname_for_user', '打工人'))" 2>/dev/null || echo "$NICKNAME") info "虾创建成功：${ACTUAL_NAME}" fi step "保存配置..." CONFIG_PATH="$CONFIG_FILE" ACTUAL_USER_ID_VALUE="$ACTUAL_USER_ID" ACCESS_TOKEN_VALUE="$ACCESS_TOKEN" ACTUAL_NAME_VALUE="$ACTUAL_NAME" PERSONALITY_VALUE="$PERSONALITY" NICKNAME_VALUE="$NICKNAME" API_BASE_VALUE="$API_BASE" WEB_BASE_VALUE="$WEB_BASE" MORNING_TIME_VALUE="$MORNING_TIME" DISCOVERY_TIME_VALUE="$DISCOVERY_TIME" EVENING_TIME_VALUE="$EVENING_TIME" CHANNEL_VALUE="$CHANNEL" TO_VALUE="$TO" WECOM_USER_ID_VALUE="$WECOM_USER_ID" ALL_CHANNELS_JSON_VALUE="$ALL_CHANNELS_JSON" MEMORY_MODE_VALUE="$MEMORY_MODE" python3 <<'PY' import json import os config = { "user_id": os.environ["ACTUAL_USER_ID_VALUE"], "access_token": os.environ["ACCESS_TOKEN_VALUE"], "lobster_name": os.environ["ACTUAL_NAME_VALUE"], "lobster_personality": os.environ["PERSONALITY_VALUE"], "nickname_for_user": os.environ["NICKNAME_VALUE"], "api_base": os.environ["API_BASE_VALUE"], "web_base": os.environ["WEB_BASE_VALUE"], "morning_time": os.environ["MORNING_TIME_VALUE"], "discovery_time": os.environ["DISCOVERY_TIME_VALUE"], "evening_time": os.environ["EVENING_TIME_VALUE"], "channel": os.environ["CHANNEL_VALUE"], "chat_id": os.environ["TO_VALUE"], "memory_mode": os.environ["MEMORY_MODE_VALUE"], } wecom_user_id = (os.environ.get("WECOM_USER_ID_VALUE") or "").strip() if wecom_user_id: config["wecom_user_id"] = wecom_user_id try: known_channels = json.loads(os.environ.get("ALL_CHANNELS_JSON_VALUE", "")) except Exception: known_channels = [] channel = os.environ["CHANNEL_VALUE"] peer_id = os.environ["TO_VALUE"] if not any(isinstance(item, dict) and item.get("channel") == channel and item.get("peer_id") == peer_id for item in known_channels): known_channels.insert(0, {"channel": channel, "peer_id": peer_id}) config["known_channels"] = known_channels with open(os.environ["CONFIG_PATH"], "w", encoding="utf-8") as f: json.dump(config, f, indent=2, ensure_ascii=False) print("[✓] 配置已写入 .lobster-config") PY step "收口 delivery contract..." DELIVERY_CONTRACT_JSON=$(lobster_sync_delivery_contract "$CONFIG_FILE" "$CHANNEL" "$TO") info "delivery contract 已写入 .lobster-config" step "验证虾状态..." STATUS_RESPONSE=$(curl -fsS -H "Authorization: Bearer ${ACCESS_TOKEN}" "${API_BASE}/api/lobster/${ACTUAL_USER_ID}/status" 2>/dev/null || true) STATUS=$(echo "$STATUS_RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin).get('status', 'unknown'))" 2>/dev/null || echo "unknown") if [ "$STATUS" != "unknown" ]; then info "虾状态: ${STATUS}" else warn "无法获取虾状态（不影响初始化）" fi step "准备工作室短时入口..." STUDIO_LINK_RESPONSE=$(curl -fsS -H "Authorization: Bearer ${ACCESS_TOKEN}" "${API_BASE}/api/lobster/${ACTUAL_USER_ID}/studio-link" 2>/dev/null || true) if [ -n "$STUDIO_LINK_RESPONSE" ]; then STUDIO_WEB_URL=$(echo "$STUDIO_LINK_RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin).get('web_url', ''))" 2>/dev/null || true) STUDIO_SCREENSHOT_URL=$(echo "$STUDIO_LINK_RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin).get('screenshot_url', ''))" 2>/dev/null || true) STUDIO_LINK_EXPIRES_AT=$(echo "$STUDIO_LINK_RESPONSE" | python3 -c "import sys, json; print(json.load(sys.stdin).get('expires_at', ''))" 2>/dev/null || true) if [ -n "$STUDIO_WEB_URL" ]; then info "已获取工作室短时入口" else warn "工作室短时入口响应缺少 web_url；可稍后重试" fi else warn "暂时无法获取工作室短时入口；可稍后通过 studio-link 接口重试" fi step "注册定时推送..." set +e bash "${BASE_DIR}/setup-cron.sh" \ --channel "${CHANNEL}" \ --to "${TO}" \ --wecom-user-id "${WECOM_USER_ID}" \ --morning "${MORNING_TIME}" \ --discovery "${DISCOVERY_TIME}" \ --evening "${EVENING_TIME}" \ --memory-mode "${MEMORY_MODE}" CRON_EXIT_CODE=$? set -e CRON_STATUS_SUMMARY=$(CONFIG_PATH="$CONFIG_FILE" python3 <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(path.read_text(encoding="utf-8")) except Exception: config = {} cron = config.get("cron_registration") if isinstance(config.get("cron_registration"), dict) else {} print("1" if config.get("cron_registered") else "0") print(str(cron.get("status") or "unregistered")) print(str(cron.get("reason") or "")) PY ) CRON_REGISTERED=$(echo "$CRON_STATUS_SUMMARY" | sed -n '1p') CRON_REGISTRATION_STATUS=$(echo "$CRON_STATUS_SUMMARY" | sed -n '2p') CRON_REGISTRATION_REASON=$(echo "$CRON_STATUS_SUMMARY" | sed -n '3p') if [ "$CRON_EXIT_CODE" -ne 0 ]; then CRON_REGISTERED=0 warn "定时推送注册执行失败，请稍后查看 setup-cron 日志或单独重试" elif [ "$CRON_REGISTERED" = "1" ]; then info "定时推送 cron 已注册完成" elif [ "$CRON_REGISTRATION_STATUS" = "pending_activation" ]; then warn "已保存待激活的定时推送配置：${CRON_REGISTRATION_REASON}" else warn "定时推送当前未注册成功，可稍后单独重试 setup-cron.sh" fi echo "" echo "═══════════════════════════════════════════════" echo "" if [ "$CRON_REGISTERED" = "1" ]; then info "🦞 初始化全部完成！" elif [ "$CRON_REGISTRATION_STATUS" = "pending_activation" ]; then warn "🦞 虾已就绪，定时推送已进入待激活状态" else warn "🦞 虾已就绪，但定时推送暂未注册成功" fi echo "" echo " 虾自己的名字: ${ACTUAL_NAME}" echo " 虾格: ${PERSONALITY}" echo " 虾对用户的称呼: ${NICKNAME}" echo " 理解模式: ${MEMORY_MODE}" echo " 早安: 每天 ${MORNING_TIME}" echo " 晚间 roundup: 每天 ${DISCOVERY_TIME}" echo " 晚安: 每天 ${EVENING_TIME}" if [ "$CRON_REGISTERED" = "1" ]; then echo " 投递: 已完成 cron 注册" elif [ "$CRON_REGISTRATION_STATUS" = "pending_activation" ]; then echo " 定时推送: 已保存待激活配置；一旦具备主动推送能力，再次 reconcile 即可完成注册" echo " 待激活原因: ${CRON_REGISTRATION_REASON}" else echo " 定时推送: 尚未完成注册，可稍后执行 bash \"${BASE_DIR}/setup-cron.sh\" 补注册" fi echo "" if [ -n "$STUDIO_WEB_URL" ]; then echo " 工作室短时入口: 已准备好（见初始化结果中的 studio_web_url）" else echo " 工作室入口: 暂未取到短时链接，可稍后重试获取" fi echo " 初始化结果不会再打印长期 token URL。" echo "" echo "═══════════════════════════════════════════════" echo "" echo "INIT_RESULT_JSON:" ACTUAL_NAME_VALUE="$ACTUAL_NAME" PERSONALITY_VALUE="$PERSONALITY" NICKNAME_VALUE="$NICKNAME" ACTUAL_USER_ID_VALUE="$ACTUAL_USER_ID" WEB_BASE_VALUE="$WEB_BASE" MORNING_TIME_VALUE="$MORNING_TIME" DISCOVERY_TIME_VALUE="$DISCOVERY_TIME" EVENING_TIME_VALUE="$EVENING_TIME" CHANNEL_VALUE="$CHANNEL" TO_VALUE="$TO" WECOM_USER_ID_VALUE="$WECOM_USER_ID" ALL_CHANNELS_JSON_VALUE="$ALL_CHANNELS_JSON" MEMORY_MODE_VALUE="$MEMORY_MODE" CRON_REGISTERED_VALUE="$CRON_REGISTERED" CRON_REGISTRATION_STATUS_VALUE="$CRON_REGISTRATION_STATUS" CRON_REGISTRATION_REASON_VALUE="$CRON_REGISTRATION_REASON" REUSED_EXISTING_VALUE="$REUSED_EXISTING" STUDIO_WEB_URL_VALUE="$STUDIO_WEB_URL" STUDIO_SCREENSHOT_URL_VALUE="$STUDIO_SCREENSHOT_URL" STUDIO_LINK_EXPIRES_AT_VALUE="$STUDIO_LINK_EXPIRES_AT" DELIVERY_CONTRACT_JSON_VALUE="$DELIVERY_CONTRACT_JSON" python3 <<'PY' import json import os from urllib.parse import urlparse channel = os.environ["CHANNEL_VALUE"] peer_id = os.environ["TO_VALUE"] studio_web_url = os.environ.get("STUDIO_WEB_URL_VALUE", "") studio_path = urlparse(studio_web_url).path if studio_web_url else "/lobster/" + os.environ["ACTUAL_USER_ID_VALUE"] known_channels = json.loads(os.environ["ALL_CHANNELS_JSON_VALUE"]) if os.environ.get("ALL_CHANNELS_JSON_VALUE", "").strip() else [{"channel": channel, "peer_id": peer_id}] if not any(item.get("channel") == channel and item.get("peer_id") == peer_id for item in known_channels): known_channels.insert(0, {"channel": channel, "peer_id": peer_id}) delivery_contract_raw = os.environ.get("DELIVERY_CONTRACT_JSON_VALUE", "").strip() try: delivery_contract = json.loads(delivery_contract_raw) if delivery_contract_raw else {} except Exception: delivery_contract = {} result = { "success": True, "lobster_name": os.environ["ACTUAL_NAME_VALUE"], "personality": os.environ["PERSONALITY_VALUE"], "nickname": os.environ["NICKNAME_VALUE"], "user_id": os.environ["ACTUAL_USER_ID_VALUE"], "studio_path": studio_path, "studio_web_url": studio_web_url, "studio_screenshot_url": os.environ.get("STUDIO_SCREENSHOT_URL_VALUE", ""), "studio_link_expires_at": os.environ.get("STUDIO_LINK_EXPIRES_AT_VALUE", ""), "morning_time": os.environ["MORNING_TIME_VALUE"], "discovery_time": os.environ["DISCOVERY_TIME_VALUE"], "evening_time": os.environ["EVENING_TIME_VALUE"], "memory_mode": os.environ["MEMORY_MODE_VALUE"], "channel": channel, "chat_id": peer_id, "cron_registered": os.environ.get("CRON_REGISTERED_VALUE") == "1", "cron_registration_status": os.environ.get("CRON_REGISTRATION_STATUS_VALUE", "unregistered"), "cron_registration_reason": os.environ.get("CRON_REGISTRATION_REASON_VALUE", ""), "reused_existing": os.environ.get("REUSED_EXISTING_VALUE") == "1", "known_channels": known_channels, "delivery_contract": delivery_contract, "wecom_user_id": os.environ.get("WECOM_USER_ID_VALUE", ""), } print(json.dumps(result, indent=2, ensure_ascii=False)) PY FILE:openclaw.json { "name": "lobster-says", "version": "2.5.8", "description": "🦞 虾说——你的专属共情虾。每天早晚给你一句关心的话，让你觉得被看到了。需注册 cron 定时任务推送消息；2.5.4 起服务端提前 3 分钟预生成消息，cron 触发时 /api/generate 直接返回已有消息（<200ms），企微 emit-message-text 链路总延迟从 15s+ 降至 <2s；Web 端新增信封动画特效，用户打开链接即看到虾的来信。企业微信在 2.5.3 起走极简主链路：脚本通过 --emit-message-text 只输出最终消息文本，由 isolated agent 使用 message 工具直发到 openclaw-wecom-bot；通用通道（Telegram 等）由脚本完整模式多通道 fallback 发送；可选读取本地会话日志生成理解摘要并上传到 nixiashuo.com。", "author": "Jared", "license": "MIT", "repository": "", "keywords": [ "emotional-support", "daily-push", "pixel-art", "companion", "cron", "memory", "sticker", "情感陪伴", "情绪价值", "情绪虾", "暖心虾", "懂你的虾", "empathy", "mental-health", "wellness", "daily-checkin", "mood", "care", "lobster", "虾说", "中文", "chinese", "壁纸生成", "wallpaper" ], "data_usage": { "cron_jobs": "Registers 5-6 scheduled cron jobs via openclaw cron add for daily message push and optional transcript digest", "session_read": "In smart/deep memory mode only, reads local session JSONL files under ~/.openclaw/agents/main/sessions/ to generate understanding summaries", "network": "Communicates with https://nixiashuo.com for message generation, delivery reporting, and optional transcript digest upload", "config_files": "Reads and writes .lobster-config in the skill directory for user/lobster identity and channel preferences. Does NOT read openclaw.json or extract gateway tokens.", "user_consent": "Memory mode (lightweight/smart/deep) is explicitly chosen by the user during initialization and can be changed at any time" } } FILE:push-scheduled-message.sh #!/bin/bash # ═══════════════════════════════════════════════ # 虾说 — 运行时定时推送脚本 # ═══════════════════════════════════════════════ # # 数据访问声明： # - 读取 .lobster-config（技能自身配置，含 user_id/access_token/通道偏好） # - 调用 openclaw sessions --json 获取最近活跃 IM 通道元数据（通道名+ID） # - 通用 IM 通道使用 openclaw message send 发送消息（多通道 fallback） # - 企业微信定时推送主链路走 --emit-message-text：脚本只输出最终消息文本，由 cron isolated agent 使用 message 工具私聊直达 # - --generate-only JSON 模式保留为调试/兼容能力，不再作为企微 cron 主链路 # - 与 https://nixiashuo.com 通信：生成消息、报告送达结果 # - 不读取 openclaw.json 配置文件，不提取 gateway token # # 设计目标： # 1. --emit-message-text 模式：仅输出最终消息文本到 stdout，不执行任何发送 # 适用于：企业微信 cron 链路——由 isolated agent 直接把 stdout 原文作为 message 工具参数发送 # 2. --generate-only 模式：输出 JSON 到 stdout，保留为调试/兼容能力 # 3. 完整模式（默认）：生成消息 + 多通道 fallback 发送 # 适用于：Telegram/Discord/飞书等通用 IM 通道 # 4. 各模式共享消息生成、init-ready 管理、送达报告等基础设施 # # 使用方式： # bash push-scheduled-message.sh --slot morning|discovery|evening # bash push-scheduled-message.sh --slot morning --emit-message-text # bash push-scheduled-message.sh --slot morning --generate-only set -u BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" COMMON_SCRIPT="${BASE_DIR}/runtime-common.sh" SETUP_CRON_SCRIPT="${BASE_DIR}/setup-cron.sh" LOG_DIR="${BASE_DIR}/logs" RUN_TS="$(date +%Y%m%d-%H%M%S)" LOG_FILE="${LOG_DIR}/push-scheduled-message-${RUN_TS}.log" SLOT="" FORCED_CHANNEL="" FORCED_TARGET="" EXTRA_CONTEXT="" CONTENT_PREFIX="" MANAGED_JOB_NAME="" JOB_KIND="scheduled-message" GENERATE_ONLY=0 EMIT_MESSAGE_TEXT=0 ACTIVE_WINDOW_MINUTES="${ACTIVE_WINDOW_MINUTES:-10080}" INIT_READY_MAX_ATTEMPTS="${INIT_READY_MAX_ATTEMPTS:-4}" INIT_READY_RETRY_MINUTES="${INIT_READY_RETRY_MINUTES:-15}" # Telegram 原生 Bot API 可稳定处理 sendPhoto/URL 图片； # 企业微信/飞书等机器人协议对图片通常要求 webhook 特定 payload 或上传素材， # 统一经 OpenClaw CLI 时更适合优先走文本 + 链接，避免假成功或媒体协议不兼容。 MEDIA_SEND_CHANNELS="telegram discord googlechat slack mattermost signal imessage msteams" GENERATE_RESPONSE_FILE="" PARSED_JSON_FILE="" _SEND_STDERR_FILE="" cleanup() { [ -n "$GENERATE_RESPONSE_FILE" ] && [ -f "$GENERATE_RESPONSE_FILE" ] && rm -f "$GENERATE_RESPONSE_FILE" [ -n "$PARSED_JSON_FILE" ] && [ -f "$PARSED_JSON_FILE" ] && rm -f "$PARSED_JSON_FILE" [ -n "$_SEND_STDERR_FILE" ] && [ -f "$_SEND_STDERR_FILE" ] && rm -f "$_SEND_STDERR_FILE" [ -n "${STICKER_IMAGE_FILE:-}" ] && [ -f "${STICKER_IMAGE_FILE:-}" ] && rm -f "$STICKER_IMAGE_FILE" } GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' CYAN='\033[0;36m' NC='\033[0m' _ts() { date '+%Y-%m-%d %H:%M:%S'; } # --emit-message-text / --generate-only 模式下，info/warn/error/step 输出到 stderr，stdout 保留给消息文本或 JSON info() { echo -e "${GREEN}[✓]${NC} $(_ts) $1" >&2; } warn() { echo -e "${YELLOW}[!]${NC} $(_ts) $1" >&2; } error() { echo -e "${RED}[✗]${NC} $(_ts) $1" >&2; exit 1; } CURRENT_STEP="启动" step() { CURRENT_STEP="$1"; echo -e "${CYAN}──${NC} $(_ts) $1" >&2; } if [ ! -f "$COMMON_SCRIPT" ]; then echo "共享运行时脚本不存在：${COMMON_SCRIPT}" >&2 exit 1 fi . "$COMMON_SCRIPT" mkdir -p "$LOG_DIR" # 日志始终写文件；tee 只用于非 generate-only 模式的 stderr exec 3>>"$LOG_FILE" echo "[log] push-scheduled-message started at $(date '+%F %T')" >&3 echo "[log] file: ${LOG_FILE}" >&3 after_exit() { local exit_code=$? cleanup if [ "$exit_code" -ne 0 ]; then echo "" >&3 echo "[log] push-scheduled-message failed at step: ${CURRENT_STEP} (exit=${exit_code})" >&3 echo "[log] inspect: ${LOG_FILE}" >&3 else echo "[log] push-scheduled-message finished successfully" >&3 fi exec 3>&- } trap after_exit EXIT while [[ $# -gt 0 ]]; do case "$1" in --slot) SLOT="$2"; shift 2 ;; --channel) FORCED_CHANNEL="$2"; shift 2 ;; --to|--target) FORCED_TARGET="$2"; shift 2 ;; --extra-context) EXTRA_CONTEXT="$2"; shift 2 ;; --content-prefix) CONTENT_PREFIX="$2"; shift 2 ;; --job-kind) JOB_KIND="$2"; shift 2 ;; --managed-job-name|--remove-cron-job-name) MANAGED_JOB_NAME="$2"; shift 2 ;; --generate-only) GENERATE_ONLY=1; shift ;; --emit-message-text) EMIT_MESSAGE_TEXT=1; shift ;; *) echo "未知参数: $1" >&2; exit 1 ;; esac done if [[ ! "$SLOT" =~ ^(morning|discovery|evening|event|sticker|wallpaper)$ ]]; then error "--slot 必须是 morning / discovery / evening / event / sticker / wallpaper" fi for cmd in python3 curl openclaw; do if ! command -v "$cmd" >/dev/null 2>&1; then error "$cmd 不可用" fi done if [ ! -f "$CONFIG_FILE" ]; then error ".lobster-config 不存在，请先初始化虾" fi read_config_value() { local key="$1" local default_value="${2:-}" CONFIG_PATH="$CONFIG_FILE" KEY_NAME="$key" DEFAULT_VALUE="$default_value" python3 - <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) key = os.environ["KEY_NAME"] default = os.environ.get("DEFAULT_VALUE", "") try: data = json.loads(path.read_text(encoding='utf-8')) value = data.get(key, default) print(value if value is not None else default) except Exception: print(default) PY } supports_media() { local channel="$1" for ch in $MEDIA_SEND_CHANNELS; do if [ "$ch" = "$channel" ]; then return 0 fi done return 1 } build_candidate_lines() { local sessions_json="$1" local forced_channel="$2" local forced_target="$3" build_candidate_lines_with_policy "$CONFIG_FILE" "$sessions_json" "$forced_channel" "$forced_target" } load_runtime_policy() { POLICY_JSON=$(lobster_runtime_policy_json "$CONFIG_FILE") POLICY_JSON_VALUE="$POLICY_JSON" python3 <<'PY' import json import os policy = json.loads(os.environ["POLICY_JSON_VALUE"]) print(policy.get("binding_channel", "")) print(policy.get("binding_target", "")) print(policy.get("binding_mode", "prefer")) print("1" if policy.get("strict_binding") else "0") print(policy.get("outbound_adapter", "openclaw")) print(policy.get("outbound_webhook_url", "")) print(policy.get("outbound_webhook_secret", "")) print(policy.get("delivery_channel", "")) print(policy.get("delivery_target", "")) print("1" if policy.get("delivery_ready") else "0") print(policy.get("delivery_reason", "")) print(policy.get("cron_status", "unregistered")) PY } sync_known_channels_after_send() { local current_channel="$1" local current_target="$2" local candidate_lines="$3" local delivered_at="$4" update_config_after_send_with_policy "$CONFIG_FILE" "$current_channel" "$current_target" "$candidate_lines" "$delivered_at" } append_delivery_ledger() { local stage="$1" local status="$2" local detail="${3:-}" local channel="${4:-}" local target="${5:-}" local delivery_mode="${6:-}" local message_id="${7:-}" mkdir -p "$LOG_DIR" LOG_DIR_VALUE="$LOG_DIR" \ LEDGER_STAGE="$stage" \ LEDGER_STATUS="$status" \ LEDGER_DETAIL="$detail" \ LEDGER_CHANNEL="$channel" \ LEDGER_TARGET="$target" \ LEDGER_MODE="$delivery_mode" \ LEDGER_MESSAGE_ID="$message_id" \ LEDGER_SLOT="$SLOT" \ LEDGER_JOB_KIND="$JOB_KIND" \ LEDGER_RUN_TS="$RUN_TS" \ python3 <<'PY' import json import os from datetime import datetime, timezone from pathlib import Path message_id_raw = (os.environ.get("LEDGER_MESSAGE_ID") or "").strip() try: message_id = int(message_id_raw) if message_id_raw else None except ValueError: message_id = None entry = { "ts": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), "run_ts": os.environ.get("LEDGER_RUN_TS"), "script": "push-scheduled-message.sh", "slot": os.environ.get("LEDGER_SLOT"), "job_kind": os.environ.get("LEDGER_JOB_KIND"), "stage": os.environ.get("LEDGER_STAGE"), "status": os.environ.get("LEDGER_STATUS"), "detail": os.environ.get("LEDGER_DETAIL") or None, "channel": os.environ.get("LEDGER_CHANNEL") or None, "target": os.environ.get("LEDGER_TARGET") or None, "delivery_mode": os.environ.get("LEDGER_MODE") or None, "message_id": message_id, } ledger_path = Path(os.environ["LOG_DIR_VALUE"]) / "delivery-ledger.jsonl" with ledger_path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry, ensure_ascii=False) + "\n") PY } update_init_ready_state() { [ "$JOB_KIND" = "init-ready" ] || return 0 local state="$1" local attempt_count="${2:-}" local message_id="${3:-}" local error_message="${4:-}" local channel="${5:-}" local target="${6:-}" local extra_field="${7:-}" local extra_value="${8:-}" INIT_STATE="$state" \ INIT_ATTEMPT_COUNT="$attempt_count" \ INIT_MESSAGE_ID="$message_id" \ INIT_ERROR="$error_message" \ INIT_CHANNEL="$channel" \ INIT_TARGET="$target" \ INIT_EXTRA_FIELD="$extra_field" \ INIT_EXTRA_VALUE="$extra_value" \ CONFIG_PATH="$CONFIG_FILE" \ python3 <<'PY' import json import os from datetime import datetime, timezone from pathlib import Path config_path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} init_ready = config.get("init_ready") if isinstance(config.get("init_ready"), dict) else {} now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") state = os.environ.get("INIT_STATE") or init_ready.get("state") or "unknown" init_ready["state"] = state init_ready["updated_at"] = now if state in {"attempting", "delivery_failed", "retry_scheduled", "scheduled"}: init_ready["last_attempt_at"] = now attempt_count = (os.environ.get("INIT_ATTEMPT_COUNT") or "").strip() if attempt_count: try: init_ready["attempts"] = int(attempt_count) except ValueError: pass message_id = (os.environ.get("INIT_MESSAGE_ID") or "").strip() if message_id: try: init_ready["last_message_id"] = int(message_id) except ValueError: pass error_message = (os.environ.get("INIT_ERROR") or "").strip() if error_message: init_ready["last_error"] = error_message channel = (os.environ.get("INIT_CHANNEL") or "").strip() target = (os.environ.get("INIT_TARGET") or "").strip() if channel: init_ready["last_delivery_channel"] = channel if target: init_ready["last_delivery_target"] = target if state == "sent": init_ready["sent_at"] = now extra_field = (os.environ.get("INIT_EXTRA_FIELD") or "").strip() extra_value = (os.environ.get("INIT_EXTRA_VALUE") or "").strip() if extra_field: init_ready[extra_field] = extra_value config["init_ready"] = init_ready config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") PY } begin_init_ready_attempt() { [ "$JOB_KIND" = "init-ready" ] || { echo "0"; return 0; } CONFIG_PATH="$CONFIG_FILE" python3 <<'PY' import json import os from datetime import datetime, timezone from pathlib import Path config_path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} init_ready = config.get("init_ready") if isinstance(config.get("init_ready"), dict) else {} attempts = int(init_ready.get("attempts") or 0) + 1 now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") init_ready["attempts"] = attempts init_ready["state"] = "attempting" init_ready["last_attempt_at"] = now init_ready["updated_at"] = now config["init_ready"] = init_ready config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") print(attempts) PY } init_ready_already_sent() { [ "$JOB_KIND" = "init-ready" ] || return 1 CONFIG_PATH="$CONFIG_FILE" python3 <<'PY' import json import os from pathlib import Path config_path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} state = "" if isinstance(config.get("init_ready"), dict): state = str(config["init_ready"].get("state") or "") print("yes" if state == "sent" else "no") PY } finalize_managed_job() { [ -n "$MANAGED_JOB_NAME" ] || return 0 [ -f "$SETUP_CRON_SCRIPT" ] || return 0 bash "$SETUP_CRON_SCRIPT" --remove-job-by-name "$MANAGED_JOB_NAME" >/dev/null 2>&1 || true } schedule_init_ready_retry() { local attempt_count="$1" local reason="$2" [ "$JOB_KIND" = "init-ready" ] || return 0 if [ "$attempt_count" -ge "$INIT_READY_MAX_ATTEMPTS" ]; then warn "init-ready 已达到最大尝试次数 ${INIT_READY_MAX_ATTEMPTS}，停止自动重试" update_init_ready_state "failed" "$attempt_count" "" "$reason" append_delivery_ledger "retry_exhausted" "failed" "$reason" finalize_managed_job return 0 fi if bash "$SETUP_CRON_SCRIPT" --schedule-init-ready-delay "$INIT_READY_RETRY_MINUTES" >/dev/null 2>&1; then update_init_ready_state "retry_scheduled" "$attempt_count" "" "$reason" "" "" "retry_delay_minutes" "$INIT_READY_RETRY_MINUTES" append_delivery_ledger "retry_scheduled" "scheduled" "$reason" info "init-ready 已重排，${INIT_READY_RETRY_MINUTES} 分钟后重试" else warn "init-ready 重排失败，请稍后手动执行 setup-cron.sh" update_init_ready_state "delivery_failed" "$attempt_count" "" "$reason" append_delivery_ledger "retry_schedule_failed" "failed" "$reason" fi } report_delivery() { local message_id="$1" local status="$2" local channel="$3" local target="$4" local delivery_mode="$5" local delivered_text="$6" local screenshot_url="$7" local error_message="$8" REPORT_RESULT=$(REPORT_USER_ID="$USER_ID" REPORT_MESSAGE_ID="$message_id" REPORT_STATUS="$status" REPORT_CHANNEL="$channel" REPORT_TARGET="$target" REPORT_MODE="$delivery_mode" REPORT_TEXT="$delivered_text" REPORT_SCREENSHOT_URL="$screenshot_url" REPORT_ERROR="$error_message" python3 <<'PY' import json import os payload = { "user_id": os.environ["REPORT_USER_ID"], "message_id": int(os.environ["REPORT_MESSAGE_ID"]), "status": os.environ["REPORT_STATUS"], "channel": os.environ.get("REPORT_CHANNEL") or None, "target": os.environ.get("REPORT_TARGET") or None, "delivery_mode": os.environ.get("REPORT_MODE") or None, "delivered_text": os.environ.get("REPORT_TEXT") or None, "delivered_screenshot_url": os.environ.get("REPORT_SCREENSHOT_URL") or None, "error_message": os.environ.get("REPORT_ERROR") or None, } print(json.dumps(payload, ensure_ascii=False)) PY ) curl -fsS -X POST "${API_BASE}/api/delivery/report" \ -H "Content-Type: application/json" \ -H "Authorization: Bearer ${ACCESS_TOKEN}" \ -d "$REPORT_RESULT" >/dev/null 2>&1 || warn "送达结果回写失败（不影响用户已收到的消息）" } build_text_message() { local include_url="$1" LOBSTER_NAME_VALUE="$LOBSTER_NAME" GENERATED_CONTENT_VALUE="$GENERATED_CONTENT" WEB_URL_VALUE="$WEB_URL" SCREENSHOT_URL_VALUE="$SCREENSHOT_URL" INCLUDE_URL_VALUE="$include_url" python3 <<'PY' import os lobster_name = os.environ["LOBSTER_NAME_VALUE"] content = os.environ["GENERATED_CONTENT_VALUE"] web_url = os.environ["WEB_URL_VALUE"] screenshot_url = os.environ["SCREENSHOT_URL_VALUE"] include_url = os.environ.get("INCLUDE_URL_VALUE") == "true" lines = [f"🦞 {lobster_name}说：「{content}」", ""] if include_url and screenshot_url: lines.append(f"📸 {lobster_name}的工作室截图：{screenshot_url}") lines.append(f"👀 看看{lobster_name}在干嘛 → {web_url}") print("\n".join(lines)) PY } # ═══════════════════════════════════════════════ # 读取配置 # ═══════════════════════════════════════════════ USER_ID=$(read_config_value "user_id") ACCESS_TOKEN=$(read_config_value "access_token") LOBSTER_NAME=$(read_config_value "lobster_name" "虾") API_BASE=$(read_config_value "api_base" "https://nixiashuo.com") WEB_BASE=$(read_config_value "web_base" "$API_BASE") WECOM_USER_ID=$(read_config_value "wecom_user_id") if [ -z "$USER_ID" ] || [ -z "$ACCESS_TOKEN" ]; then error ".lobster-config 缺少 user_id 或 access_token" fi mapfile -t POLICY_LINES < <(load_runtime_policy) BOUND_CHANNEL="${POLICY_LINES[0]:-}" BOUND_TARGET="${POLICY_LINES[1]:-}" BINDING_MODE="${POLICY_LINES[2]:-prefer}" STRICT_BINDING="${POLICY_LINES[3]:-0}" OUTBOUND_ADAPTER="${POLICY_LINES[4]:-openclaw}" OUTBOUND_WEBHOOK_URL="${POLICY_LINES[5]:-}" OUTBOUND_WEBHOOK_SECRET="${POLICY_LINES[6]:-}" DELIVERY_CHANNEL="${POLICY_LINES[7]:-}" DELIVERY_TARGET="${POLICY_LINES[8]:-}" DELIVERY_READY="${POLICY_LINES[9]:-0}" DELIVERY_REASON="${POLICY_LINES[10]:-}" CRON_STATUS="${POLICY_LINES[11]:-unregistered}" if [ "$STRICT_BINDING" = "1" ]; then OUTBOUND_ADAPTER="wecom-direct-message" DELIVERY_CHANNEL="wecom" if [ -n "$WECOM_USER_ID" ]; then DELIVERY_TARGET="$WECOM_USER_ID" DELIVERY_READY="1" DELIVERY_REASON="" else DELIVERY_TARGET="" DELIVERY_READY="0" DELIVERY_REASON="企业微信定时推送缺少 sender_id / wecom_user_id；请在技能层从 inbound metadata 传入 sender_id" fi fi # ═══════════════════════════════════════════════ # init-ready 前置检查 # ═══════════════════════════════════════════════ if [ "$JOB_KIND" = "init-ready" ] && [ "$(init_ready_already_sent)" = "yes" ]; then info "init-ready 已经送达过，本次跳过重复执行" append_delivery_ledger "duplicate_skip" "skipped" "init-ready already sent" finalize_managed_job exit 0 fi INIT_READY_ATTEMPT_NUMBER=0 if [ "$JOB_KIND" = "init-ready" ]; then INIT_READY_ATTEMPT_NUMBER=$(begin_init_ready_attempt) append_delivery_ledger "attempt_started" "running" "init-ready attempt=${INIT_READY_ATTEMPT_NUMBER}" fi if [ "$DELIVERY_READY" != "1" ]; then LAST_ERROR="${DELIVERY_REASON:-delivery contract not ready}" append_delivery_ledger "delivery_contract_not_ready" "failed" "$LAST_ERROR" "$DELIVERY_CHANNEL" "$DELIVERY_TARGET" "$OUTBOUND_ADAPTER" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi # ═══════════════════════════════════════════════ # 生成消息 # ═══════════════════════════════════════════════ step "生成 ${SLOT} 消息..." GENERATE_RESPONSE_FILE=$(mktemp "${TMPDIR:-/tmp}/lobster-generate-response.XXXXXX.json") PARSED_JSON_FILE=$(mktemp "${TMPDIR:-/tmp}/lobster-generate-parsed.XXXXXX.json") GENERATE_REQUEST_JSON=$(GENERATE_USER_ID="$USER_ID" GENERATE_SLOT="$SLOT" GENERATE_EXTRA_CONTEXT="$EXTRA_CONTEXT" python3 <<'PY' import json import os payload = { "user_id": os.environ["GENERATE_USER_ID"], "message_type": os.environ["GENERATE_SLOT"], "include_screenshot_base64": False, } extra_context = (os.environ.get("GENERATE_EXTRA_CONTEXT") or "").strip() if extra_context: payload["extra_context"] = extra_context print(json.dumps(payload, ensure_ascii=False)) PY ) if ! curl -fsS -X POST "${API_BASE}/api/generate" \ -H "Content-Type: application/json" \ -H "Authorization: Bearer ${ACCESS_TOKEN}" \ -d "$GENERATE_REQUEST_JSON" \ -o "$GENERATE_RESPONSE_FILE"; then LAST_ERROR="调用 /api/generate 失败" append_delivery_ledger "generate_failed" "failed" "$LAST_ERROR" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi SKIPPED_CHECK=$(python3 - "$GENERATE_RESPONSE_FILE" <<'PY' import json import sys from pathlib import Path try: data = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) if data.get("skipped"): print(data.get("reason", "unknown")) else: print("no") except Exception: print("no") PY ) if [ "$SKIPPED_CHECK" != "no" ]; then info "📋 本次 ${SLOT} 生成被跳过: ${SKIPPED_CHECK}" append_delivery_ledger "generate_skipped" "skipped" "$SKIPPED_CHECK" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$SKIPPED_CHECK" exit 0 fi PARSE_RESULT=$(python3 - "$GENERATE_RESPONSE_FILE" "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path response_path = Path(sys.argv[1]) parsed_path = Path(sys.argv[2]) try: data = json.loads(response_path.read_text(encoding="utf-8")) except Exception as exc: print(f"parse_error|{exc}") sys.exit(1) message = data.get("message") or {} required = { "message_id": message.get("id"), "content": message.get("raw_content") or message.get("content"), "web_url": data.get("web_url"), "screenshot_url": data.get("screenshot_url"), } missing = [k for k, v in required.items() if v in (None, "")] if missing: print("missing|" + ",".join(missing)) sys.exit(1) required["sticker_image_base64"] = data.get("sticker_image_base64") or "" required["sticker_theme"] = data.get("sticker_theme") or "" required["is_sticker"] = bool(data.get("sticker_image_base64")) required["is_wallpaper"] = message.get("message_type") == "wallpaper" required["is_media"] = required["is_sticker"] or required["is_wallpaper"] parsed_path.write_text(json.dumps(required, ensure_ascii=False), encoding="utf-8") print("ok") PY ) if [ "$PARSE_RESULT" != "ok" ]; then LAST_ERROR="解析 /api/generate 响应失败: ${PARSE_RESULT:-unknown}" append_delivery_ledger "parse_failed" "failed" "$LAST_ERROR" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi MESSAGE_ID=$(python3 - "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["message_id"]) PY ) GENERATED_CONTENT=$(python3 - "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["content"]) PY ) WEB_URL=$(python3 - "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["web_url"]) PY ) SCREENSHOT_URL=$(python3 - "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["screenshot_url"]) PY ) append_delivery_ledger "generated" "ok" "message generated" "" "" "queued" "$MESSAGE_ID" report_delivery "$MESSAGE_ID" "pending" "${BOUND_CHANNEL:-$FORCED_CHANNEL}" "${BOUND_TARGET:-$FORCED_TARGET}" "queued" "" "$SCREENSHOT_URL" "" update_init_ready_state "generated" "$INIT_READY_ATTEMPT_NUMBER" "$MESSAGE_ID" if [ -n "$CONTENT_PREFIX" ]; then GENERATED_CONTENT=$(CONTENT_PREFIX_VALUE="$CONTENT_PREFIX" GENERATED_CONTENT_VALUE="$GENERATED_CONTENT" python3 <<'PY' import os prefix = (os.environ.get("CONTENT_PREFIX_VALUE") or "").strip() content = (os.environ.get("GENERATED_CONTENT_VALUE") or "").strip() print(f"{prefix}{content}" if content else prefix) PY ) fi IS_STICKER=$(python3 - "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path data = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) print("true" if data.get("is_media") or data.get("is_sticker") else "false") PY ) IS_WALLPAPER=$(python3 - "$PARSED_JSON_FILE" <<'PY' import json import sys from pathlib import Path data = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) print("true" if data.get("is_wallpaper") else "false") PY ) STICKER_IMAGE_FILE="" if [ "$IS_STICKER" = "true" ]; then MEDIA_TYPE_LABEL="表情包" [ "$IS_WALLPAPER" = "true" ] && MEDIA_TYPE_LABEL="壁纸" step "${MEDIA_TYPE_LABEL}类型：解码图片到临时文件..." OPENCLAW_MEDIA_DIR="${HOME}/.openclaw/media" mkdir -p "$OPENCLAW_MEDIA_DIR" STICKER_IMAGE_FILE=$(mktemp "${OPENCLAW_MEDIA_DIR}/lobster-media.XXXXXX.png") python3 - "$PARSED_JSON_FILE" "$STICKER_IMAGE_FILE" <<'PY' import base64 import json import sys from pathlib import Path data = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8")) b64 = data.get("sticker_image_base64", "") if b64: Path(sys.argv[2]).write_bytes(base64.b64decode(b64)) print("ok") else: print("no_data") sys.exit(1) PY if [ $? -ne 0 ] || [ ! -s "$STICKER_IMAGE_FILE" ]; then warn "${MEDIA_TYPE_LABEL}图片解码失败，降级为普通消息推送" IS_STICKER="false" STICKER_IMAGE_FILE="" else info "${MEDIA_TYPE_LABEL}图片已保存: $(du -h "$STICKER_IMAGE_FILE" | cut -f1)" if [ "$IS_WALLPAPER" = "true" ]; then GENERATED_CONTENT="🦞 ${LOBSTER_NAME}给你画了一张专属壁纸~" else GENERATED_CONTENT="🦞 ${LOBSTER_NAME}给你画了一张表情包~" fi fi fi # ═══════════════════════════════════════════════ # 模式分流：--emit-message-text / --generate-only / 企微 CLI fallback / 通用多通道 fallback # ═══════════════════════════════════════════════ if [ "$EMIT_MESSAGE_TEXT" = "1" ]; then # ─────────────────────────────────────────── # emit-message-text 模式：仅输出最终消息文本到 stdout # 供 isolated agent 直接把 stdout 原文作为 message 工具参数发送 # ─────────────────────────────────────────── step "emit-message-text 模式：输出最终消息文本到 stdout..." FINAL_TEXT=$(build_text_message true) echo "$FINAL_TEXT" append_delivery_ledger "emit_message_text" "ok" "plain text output to stdout for agent message tool" "" "" "emit_message_text" "$MESSAGE_ID" info "emit-message-text 完成，最终消息文本已输出到 stdout" exit 0 fi if [ "$GENERATE_ONLY" = "1" ]; then # ─────────────────────────────────────────── # generate-only 模式：仅输出消息 JSON 到 stdout（保留作为备用能力） # ─────────────────────────────────────────── step "generate-only 模式：输出消息 JSON 到 stdout..." FINAL_TEXT=$(build_text_message true) GENERATE_ONLY_OUTPUT=$(FINAL_TEXT_VALUE="$FINAL_TEXT" \ GENERATED_CONTENT_VALUE="$GENERATED_CONTENT" \ MESSAGE_ID_VALUE="$MESSAGE_ID" \ LOBSTER_NAME_VALUE="$LOBSTER_NAME" \ WEB_URL_VALUE="$WEB_URL" \ SCREENSHOT_URL_VALUE="$SCREENSHOT_URL" \ SLOT_VALUE="$SLOT" \ IS_STICKER_VALUE="$IS_STICKER" \ IS_WALLPAPER_VALUE="$IS_WALLPAPER" \ STICKER_IMAGE_FILE_VALUE="${STICKER_IMAGE_FILE:-}" \ python3 <<'PY' import json import os output = { "message_text": os.environ["FINAL_TEXT_VALUE"], "raw_content": os.environ["GENERATED_CONTENT_VALUE"], "message_id": int(os.environ["MESSAGE_ID_VALUE"]), "lobster_name": os.environ["LOBSTER_NAME_VALUE"], "web_url": os.environ["WEB_URL_VALUE"], "screenshot_url": os.environ["SCREENSHOT_URL_VALUE"], "slot": os.environ["SLOT_VALUE"], "is_sticker": os.environ.get("IS_STICKER_VALUE") == "true", "is_wallpaper": os.environ.get("IS_WALLPAPER_VALUE") == "true", "sticker_image_file": os.environ.get("STICKER_IMAGE_FILE_VALUE") or None, } print(json.dumps(output, ensure_ascii=False)) PY ) echo "$GENERATE_ONLY_OUTPUT" append_delivery_ledger "generate_only_output" "ok" "JSON output to stdout for agent delivery" "" "" "generate_only" "$MESSAGE_ID" info "generate-only 完成，消息已输出到 stdout" exit 0 fi if [ "$OUTBOUND_ADAPTER" = "wecom-direct-message" ]; then # ─────────────────────────────────────────── # 企微 CLI fallback：仅保留给手动调试/非 cron 场景 # 2.5.3 起不再作为企微 cron 主链路 # ─────────────────────────────────────────── WECOM_SEND_TARGET="${FORCED_TARGET:-${WECOM_USER_ID:-${DELIVERY_TARGET:-}}}" WECOM_SEND_CHANNEL="${FORCED_CHANNEL:-openclaw-wecom-bot}" if [ -z "$WECOM_SEND_TARGET" ]; then LAST_ERROR="企业微信定时推送缺少 sender_id / wecom_user_id" append_delivery_ledger "wecom_target_missing" "failed" "$LAST_ERROR" "$WECOM_SEND_CHANNEL" "" "wecom_cli_direct" "$MESSAGE_ID" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi if [ "$IS_STICKER" = "true" ]; then DELIVERED_TEXT="🦞 ${LOBSTER_NAME}给你准备了一张图片礼物，先去工作室看看吧 → ${WEB_URL}" else DELIVERED_TEXT=$(build_text_message true) fi step "企微 CLI fallback：openclaw message send → ${WECOM_SEND_CHANNEL} → ${WECOM_SEND_TARGET}..." _SEND_STDERR_FILE=$(mktemp "${TMPDIR:-/tmp}/lobster-send-stderr.XXXXXX") SEND_RESULT=$(openclaw message send \ --channel "$WECOM_SEND_CHANNEL" \ --target "$WECOM_SEND_TARGET" \ --message "$DELIVERED_TEXT" 2>"$_SEND_STDERR_FILE") SEND_RC=$? SEND_STDERR="" [ -f "$_SEND_STDERR_FILE" ] && SEND_STDERR=$(cat "$_SEND_STDERR_FILE" 2>/dev/null) if [ $SEND_RC -eq 0 ]; then # 检查假成功 if echo "$SEND_STDERR" | grep -qiE '(wsclient|websocket|not.connected|connection.refused|connection.reset|connection.timeout|no.active.session|failed.to.send|send.failed|delivery.failed)'; then LAST_ERROR="企微 CLI 假成功(exit=0 但检测到连接问题): ${SEND_STDERR}" append_delivery_ledger "wecom_cli_direct" "failed" "$LAST_ERROR" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" "wecom_cli_direct" "$MESSAGE_ID" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi DELIVERED_AT_UTC=$(date -u '+%Y-%m-%dT%H:%M:%SZ') report_delivery "$MESSAGE_ID" "sent" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" "wecom_cli_direct" "$DELIVERED_TEXT" "$SCREENSHOT_URL" "" append_delivery_ledger "wecom_cli_direct" "sent" "企微 CLI fallback 发送成功" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" "wecom_cli_direct" "$MESSAGE_ID" update_init_ready_state "sent" "$INIT_READY_ATTEMPT_NUMBER" "$MESSAGE_ID" "" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" [ "$JOB_KIND" = "init-ready" ] && finalize_managed_job info "📮 企微 CLI fallback 发送成功: slot=${SLOT} message_id=${MESSAGE_ID} channel=${WECOM_SEND_CHANNEL} target=${WECOM_SEND_TARGET}" echo "DELIVERY_OK slot=${SLOT} message_id=${MESSAGE_ID} channel=${WECOM_SEND_CHANNEL} mode=wecom_cli_direct" >&2 exit 0 else LAST_ERROR="企微 CLI 发送失败 (rc=${SEND_RC}): ${SEND_STDERR}" report_delivery "$MESSAGE_ID" "failed" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" "wecom_cli_direct" "" "$SCREENSHOT_URL" "$LAST_ERROR" append_delivery_ledger "wecom_cli_direct" "failed" "$LAST_ERROR" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" "wecom_cli_direct" "$MESSAGE_ID" [ "$JOB_KIND" = "init-ready" ] && update_init_ready_state "delivery_failed" "$INIT_READY_ATTEMPT_NUMBER" "$MESSAGE_ID" "$LAST_ERROR" "$WECOM_SEND_CHANNEL" "$WECOM_SEND_TARGET" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi fi # ═══════════════════════════════════════════════ # 完整发送模式：多通道 fallback（Telegram 等通用通道） # ═══════════════════════════════════════════════ step "扫描最近活跃的 IM 通道..." if [ -n "$FORCED_CHANNEL" ] && [ -n "$FORCED_TARGET" ]; then info "显式指定通道: ${FORCED_CHANNEL} → ${FORCED_TARGET}（最高优先级）" fi SESSIONS_JSON=$(openclaw sessions --json --active "$ACTIVE_WINDOW_MINUTES" 2>/dev/null || echo "[]") CANDIDATE_LINES=$(build_candidate_lines "$SESSIONS_JSON" "$FORCED_CHANNEL" "$FORCED_TARGET") if [ -z "$CANDIDATE_LINES" ]; then LAST_ERROR="没有找到可用的投递通道。请先在 Telegram / 微信 / 飞书等任一通道里和虾说一句话，或在 setup-cron.sh 时指定 --channel 和 --to。" append_delivery_ledger "candidate_resolve_failed" "failed" "$LAST_ERROR" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "$LAST_ERROR" fi FIRST_CANDIDATE=$(echo "$CANDIDATE_LINES" | head -1) info "本次优先尝试最近使用的通道：${FIRST_CANDIDATE%%|*}" LAST_ERROR="" LAST_CHANNEL="" LAST_TARGET="" SUCCESS_CHANNEL="" SUCCESS_TARGET="" SUCCESS_MODE="" DELIVERED_TEXT="" ATTEMPT_COUNT=0 _SEND_STDERR_FILE=$(mktemp "${TMPDIR:-/tmp}/lobster-send-stderr.XXXXXX") reliable_send() { _SEND_FAIL_REASON="" openclaw message send "$@" >"$_SEND_STDERR_FILE" 2>&1 local rc=$? local stderr_content="" [ -f "$_SEND_STDERR_FILE" ] && stderr_content=$(cat "$_SEND_STDERR_FILE" 2>/dev/null) if [ $rc -ne 0 ]; then _SEND_FAIL_REASON="exit_code=${rc} ${stderr_content}" return 1 fi if echo "$stderr_content" | grep -qiE '(wsclient|websocket|not.connected|connection.refused|connection.reset|connection.timeout|no.active.session|failed.to.send|send.failed|delivery.failed)'; then _SEND_FAIL_REASON="假成功(exit=0 但检测到连接问题): ${stderr_content}" return 1 fi return 0 } while IFS='|' read -r channel target; do [ -z "$channel" ] && continue [ -z "$target" ] && continue ATTEMPT_COUNT=$((ATTEMPT_COUNT + 1)) LAST_CHANNEL="$channel" LAST_TARGET="$target" step "尝试投递到 ${channel} → ${target} (attempt #${ATTEMPT_COUNT})" append_delivery_ledger "channel_attempt" "running" "attempt=${ATTEMPT_COUNT}" "$channel" "$target" "" "$MESSAGE_ID" if [ "$IS_STICKER" = "true" ] && [ -n "$STICKER_IMAGE_FILE" ] && [ -f "$STICKER_IMAGE_FILE" ]; then STUDIO_ENTRY_LINE="👀 看看${LOBSTER_NAME}在干嘛 → ${WEB_URL}" if [ "$IS_WALLPAPER" = "true" ]; then STICKER_TEXT="🦞 ${LOBSTER_NAME}给你画了一张专属壁纸~ ${STUDIO_ENTRY_LINE}" MEDIA_LABEL="壁纸" else STICKER_TEXT="🦞 ${LOBSTER_NAME}给你画了一张表情包~ ${STUDIO_ENTRY_LINE}" MEDIA_LABEL="表情包" fi if supports_media "$channel"; then if reliable_send --channel "$channel" --to "$target" --message "$STICKER_TEXT"; then if reliable_send --channel "$channel" --to "$target" --media "$STICKER_IMAGE_FILE"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" SUCCESS_MODE="media" DELIVERED_TEXT="$STICKER_TEXT" info "${MEDIA_LABEL}文本 + 图片发送成功" break fi warn "${channel} ${MEDIA_LABEL}图片发送失败 (${_SEND_FAIL_REASON})，降级为文字提示" fi fi FALLBACK_TEXT="🦞 ${LOBSTER_NAME}画了一张${MEDIA_LABEL}给你，不过这个通道暂时看不了图~去虾的工作室看看吧 → ${WEB_URL}" if reliable_send --channel "$channel" --to "$target" --message "$FALLBACK_TEXT"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" SUCCESS_MODE="url" DELIVERED_TEXT="$FALLBACK_TEXT" info "${MEDIA_LABEL}降级为文字提示发送成功" break fi LAST_ERROR="${channel} ${MEDIA_LABEL}所有模式都发送失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续回退下一个通道" append_delivery_ledger "channel_attempt" "failed" "$LAST_ERROR" "$channel" "$target" "url" "$MESSAGE_ID" continue fi if supports_media "$channel"; then TEXT_MESSAGE=$(build_text_message false) if reliable_send --channel "$channel" --target "$target" --message "$TEXT_MESSAGE"; then if reliable_send --channel "$channel" --target "$target" --media "$SCREENSHOT_URL"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" SUCCESS_MODE="media" DELIVERED_TEXT="$TEXT_MESSAGE" info "文本 + 原生截图发送成功" break fi warn "${channel} 原生截图发送失败 (${_SEND_FAIL_REASON})，降级为补发截图 URL" URL_FALLBACK=$(build_text_message true) if reliable_send --channel "$channel" --target "$target" --message "$URL_FALLBACK"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" SUCCESS_MODE="degraded_url" DELIVERED_TEXT="$URL_FALLBACK" info "已降级为文本 + 截图 URL" break fi LAST_ERROR="${channel} 原生图片与 URL 降级都失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续回退下一个通道" append_delivery_ledger "channel_attempt" "failed" "$LAST_ERROR" "$channel" "$target" "degraded_url" "$MESSAGE_ID" continue fi LAST_ERROR="${channel} 文本消息发送失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续回退下一个通道" append_delivery_ledger "channel_attempt" "failed" "$LAST_ERROR" "$channel" "$target" "text_only" "$MESSAGE_ID" continue fi TEXT_MESSAGE=$(build_text_message true) if reliable_send --channel "$channel" --to "$target" --message "$TEXT_MESSAGE"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" SUCCESS_MODE="url" DELIVERED_TEXT="$TEXT_MESSAGE" info "纯文本 + 截图 URL 发送成功" break fi LAST_ERROR="${channel} 文本消息发送失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续回退下一个通道" append_delivery_ledger "channel_attempt" "failed" "$LAST_ERROR" "$channel" "$target" "url" "$MESSAGE_ID" done <<< "$CANDIDATE_LINES" if [ -n "$SUCCESS_CHANNEL" ]; then DELIVERED_AT_UTC=$(date -u '+%Y-%m-%dT%H:%M:%SZ') report_delivery "$MESSAGE_ID" "sent" "$SUCCESS_CHANNEL" "$SUCCESS_TARGET" "$SUCCESS_MODE" "$DELIVERED_TEXT" "$SCREENSHOT_URL" "" UPDATED_KNOWN=$(sync_known_channels_after_send "$SUCCESS_CHANNEL" "$SUCCESS_TARGET" "$CANDIDATE_LINES" "$DELIVERED_AT_UTC") append_delivery_ledger "delivery_succeeded" "sent" "attempts=${ATTEMPT_COUNT}" "$SUCCESS_CHANNEL" "$SUCCESS_TARGET" "$SUCCESS_MODE" "$MESSAGE_ID" update_init_ready_state "sent" "$INIT_READY_ATTEMPT_NUMBER" "$MESSAGE_ID" "" "$SUCCESS_CHANNEL" "$SUCCESS_TARGET" [ "$JOB_KIND" = "init-ready" ] && finalize_managed_job info "📮 送达成功: slot=${SLOT} message_id=${MESSAGE_ID} channel=${SUCCESS_CHANNEL} target=${SUCCESS_TARGET} mode=${SUCCESS_MODE} attempts=${ATTEMPT_COUNT}" echo "DELIVERY_OK slot=${SLOT} message_id=${MESSAGE_ID} channel=${SUCCESS_CHANNEL} mode=${SUCCESS_MODE}" >&2 echo "KNOWN_CHANNELS=${UPDATED_KNOWN}" >&2 exit 0 fi warn "📮 送达失败: slot=${SLOT} message_id=${MESSAGE_ID} last_channel=${LAST_CHANNEL} attempts=${ATTEMPT_COUNT} error=${LAST_ERROR}" report_delivery "$MESSAGE_ID" "failed" "${LAST_CHANNEL:-$BOUND_CHANNEL}" "${LAST_TARGET:-$BOUND_TARGET}" "text_only" "" "$SCREENSHOT_URL" "$LAST_ERROR" append_delivery_ledger "delivery_failed" "failed" "$LAST_ERROR" "${LAST_CHANNEL:-$BOUND_CHANNEL}" "${LAST_TARGET:-$BOUND_TARGET}" "text_only" "$MESSAGE_ID" [ "$JOB_KIND" = "init-ready" ] && update_init_ready_state "delivery_failed" "$INIT_READY_ATTEMPT_NUMBER" "$MESSAGE_ID" "$LAST_ERROR" "${LAST_CHANNEL:-$BOUND_CHANNEL}" "${LAST_TARGET:-$BOUND_TARGET}" [ "$JOB_KIND" = "init-ready" ] && schedule_init_ready_retry "$INIT_READY_ATTEMPT_NUMBER" "$LAST_ERROR" error "所有候选通道都投递失败：${LAST_ERROR:-未知错误}" FILE:runtime-common.sh #!/bin/bash # 共享运行时辅助函数：候选通道构建、绑定策略判断、delivery contract 持久化、cron 状态收口。 lobster_detect_binding_mode() { local channel channel=$(printf '%s' "${1:-}" | tr '[:upper:]' '[:lower:]') case "$channel" in wecom*|*wecom*|wxwork*|qywx*|enterprisewechat*|enterprise-wechat*|workwechat*|wecom-bot*) echo "strict" ;; *) echo "prefer" ;; esac } lobster_runtime_policy_json() { local config_path="$1" CONFIG_PATH="$config_path" python3 <<'PY' import json import os from pathlib import Path STRICT_MARKERS = ( "wecom", "wecom-bot", "wxwork", "qywx", "enterprisewechat", "enterprise-wechat", "workwechat", ) BOT_MARKERS = ( "wecom-bot", "qywx-bot", "wxwork-bot", "enterprisewechat-bot", "enterprise-wechat-bot", "workwechat-bot", ) def detect_mode(channel: str) -> str: normalized = (channel or "").strip().lower() if any(marker in normalized for marker in STRICT_MARKERS): return "strict" return "prefer" def is_wecom_family(channel: str) -> bool: return detect_mode(channel) == "strict" def supports_openclaw_proactive(channel: str) -> bool: normalized = (channel or "").strip().lower() return any(marker in normalized for marker in BOT_MARKERS) def default_adapter_for_binding(channel: str, current: str = "") -> str: normalized = (current or "").strip().lower() if normalized in {"openclaw", "wecom-webhook", "wecom-mcp"}: return normalized if is_wecom_family(channel): return "wecom-mcp" return "openclaw" path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(path.read_text(encoding="utf-8")) except Exception: config = {} contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} cron_registration = config.get("cron_registration") if isinstance(config.get("cron_registration"), dict) else {} binding_channel = str( contract.get("binding_channel") or config.get("binding_channel") or config.get("channel") or "" ).strip() binding_target = str( contract.get("binding_target") or config.get("binding_target") or config.get("chat_id") or "" ).strip() binding_mode = str( contract.get("binding_mode") or config.get("binding_mode") or detect_mode(binding_channel) ).strip().lower() or "prefer" if binding_mode not in {"prefer", "strict"}: binding_mode = detect_mode(binding_channel) outbound_adapter = default_adapter_for_binding( binding_channel, str(contract.get("outbound_adapter") or config.get("outbound_adapter") or ""), ) outbound_webhook_url = str( contract.get("outbound_webhook_url") or config.get("outbound_webhook_url") or "" ).strip() outbound_webhook_secret = str( contract.get("outbound_webhook_secret") or config.get("outbound_webhook_secret") or "" ).strip() delivery_channel = str( contract.get("delivery_channel") or config.get("delivery_channel") or "" ).strip() delivery_target = str( contract.get("delivery_target") or config.get("delivery_target") or "" ).strip() strict_binding = binding_mode == "strict" delivery_family = "wecom" if strict_binding or is_wecom_family(binding_channel) else "general" delivery_ready = False delivery_reason = "" if delivery_family == "general": if not delivery_channel: delivery_channel = binding_channel if not delivery_target: delivery_target = binding_target delivery_ready = bool(delivery_channel and delivery_target) if not delivery_ready: delivery_reason = "缺少投递目标" else: if outbound_adapter == "wecom-webhook": if not delivery_channel: delivery_channel = "wecom-webhook" if not delivery_target: delivery_target = binding_target or "webhook" delivery_ready = bool(outbound_webhook_url) if not delivery_ready: delivery_reason = "企业微信 webhook 未配置" elif outbound_adapter == "wecom-mcp": if not delivery_channel or delivery_channel == "wecom-mcp": delivery_channel = binding_channel delivery_target = delivery_target or binding_target delivery_ready = bool(delivery_channel and delivery_target) if not delivery_ready: if not delivery_channel: delivery_reason = "企业微信长连接投递缺少 channel" else: delivery_reason = "企业微信长连接投递缺少 chat_id" else: proactive_delivery_channel = "" proactive_delivery_target = "" if supports_openclaw_proactive(delivery_channel): proactive_delivery_channel = delivery_channel proactive_delivery_target = delivery_target or (binding_target if delivery_channel == binding_channel else "") elif supports_openclaw_proactive(binding_channel): proactive_delivery_channel = binding_channel proactive_delivery_target = delivery_target or binding_target if proactive_delivery_channel: delivery_channel = proactive_delivery_channel delivery_target = proactive_delivery_target delivery_ready = bool(delivery_channel and delivery_target) if not delivery_ready: delivery_reason = "企业微信机器人缺少 delivery_target" else: delivery_channel = "" delivery_target = "" if not binding_channel or not binding_target: delivery_reason = "企业微信严格绑定缺少 binding_channel / binding_target" else: delivery_reason = "企业微信 strict family 缺少可用的主动推送配置；推荐在技能层从 inbound metadata 读取 sender_id 并传给 --wecom-user-id" payload = { "binding_channel": binding_channel, "binding_target": binding_target, "binding_mode": binding_mode, "strict_binding": strict_binding, "channel": str(config.get("channel") or "").strip(), "chat_id": str(config.get("chat_id") or "").strip(), "outbound_adapter": outbound_adapter, "outbound_webhook_url": outbound_webhook_url, "outbound_webhook_secret": outbound_webhook_secret, "delivery_channel": delivery_channel, "delivery_target": delivery_target, "delivery_family": delivery_family, "delivery_ready": delivery_ready, "delivery_reason": delivery_reason, "cron_status": str(cron_registration.get("status") or "unregistered").strip() or "unregistered", "cron_registered": bool(config.get("cron_registered") or cron_registration.get("registered")), } print(json.dumps(payload, ensure_ascii=False)) PY } lobster_sync_delivery_contract() { local config_path="$1" local hinted_channel="${2:-}" local hinted_target="${3:-}" CONFIG_PATH="$config_path" HINTED_CHANNEL="$hinted_channel" HINTED_TARGET="$hinted_target" python3 <<'PY' import json import os from datetime import datetime, timezone from pathlib import Path STRICT_MARKERS = ( "wecom", "wecom-bot", "wxwork", "qywx", "enterprisewechat", "enterprise-wechat", "workwechat", ) BOT_MARKERS = ( "wecom-bot", "qywx-bot", "wxwork-bot", "enterprisewechat-bot", "enterprise-wechat-bot", "workwechat-bot", ) def detect_mode(channel: str) -> str: normalized = (channel or "").strip().lower() if any(marker in normalized for marker in STRICT_MARKERS): return "strict" return "prefer" def is_wecom_family(channel: str) -> bool: return detect_mode(channel) == "strict" def supports_openclaw_proactive(channel: str) -> bool: normalized = (channel or "").strip().lower() return any(marker in normalized for marker in BOT_MARKERS) def default_adapter_for_binding(channel: str, current: str = "") -> str: normalized = (current or "").strip().lower() if normalized in {"openclaw", "wecom-webhook", "wecom-mcp"}: return normalized if is_wecom_family(channel): return "wecom-mcp" return "openclaw" def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") config_path = Path(os.environ["CONFIG_PATH"]) hinted_channel = (os.environ.get("HINTED_CHANNEL") or "").strip() hinted_target = (os.environ.get("HINTED_TARGET") or "").strip() try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} existing_contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} if hinted_channel and hinted_target: config["channel"] = hinted_channel config["chat_id"] = hinted_target config["binding_channel"] = hinted_channel config["binding_target"] = hinted_target binding_channel = str( config.get("binding_channel") or hinted_channel or config.get("channel") or existing_contract.get("binding_channel") or "" ).strip() binding_target = str( config.get("binding_target") or hinted_target or config.get("chat_id") or existing_contract.get("binding_target") or "" ).strip() binding_mode = str( config.get("binding_mode") or existing_contract.get("binding_mode") or detect_mode(binding_channel) ).strip().lower() or "prefer" if binding_mode not in {"prefer", "strict"}: binding_mode = detect_mode(binding_channel) outbound_adapter = default_adapter_for_binding( binding_channel, str(existing_contract.get("outbound_adapter") or config.get("outbound_adapter") or ""), ) outbound_webhook_url = str( existing_contract.get("outbound_webhook_url") or config.get("outbound_webhook_url") or "" ).strip() outbound_webhook_secret = str( existing_contract.get("outbound_webhook_secret") or config.get("outbound_webhook_secret") or "" ).strip() delivery_family = "wecom" if binding_mode == "strict" or is_wecom_family(binding_channel) else "general" delivery_channel = str( existing_contract.get("delivery_channel") or config.get("delivery_channel") or "" ).strip() delivery_target = str( existing_contract.get("delivery_target") or config.get("delivery_target") or "" ).strip() delivery_ready = False delivery_reason = "" if delivery_family == "general": delivery_channel = hinted_channel or delivery_channel or binding_channel delivery_target = hinted_target or delivery_target or binding_target delivery_ready = bool(delivery_channel and delivery_target) if not delivery_ready: delivery_reason = "缺少投递目标" else: if outbound_adapter == "wecom-webhook": delivery_channel = "wecom-webhook" delivery_target = delivery_target or binding_target or "webhook" delivery_ready = bool(outbound_webhook_url) if not delivery_ready: delivery_reason = "企业微信 webhook 未配置" elif outbound_adapter == "wecom-mcp": if not delivery_channel or delivery_channel == "wecom-mcp": delivery_channel = binding_channel delivery_target = delivery_target or binding_target delivery_ready = bool(delivery_channel and delivery_target) if not delivery_ready: if not delivery_channel: delivery_reason = "企业微信长连接投递缺少 channel" else: delivery_reason = "企业微信长连接投递缺少 chat_id" else: proactive_delivery_channel = "" proactive_delivery_target = "" if supports_openclaw_proactive(delivery_channel): proactive_delivery_channel = delivery_channel proactive_delivery_target = delivery_target or (binding_target if delivery_channel == binding_channel else "") elif supports_openclaw_proactive(binding_channel): proactive_delivery_channel = binding_channel proactive_delivery_target = delivery_target or binding_target if proactive_delivery_channel: delivery_channel = proactive_delivery_channel delivery_target = proactive_delivery_target delivery_ready = bool(delivery_channel and delivery_target) if not delivery_ready: delivery_reason = "企业微信机器人缺少 delivery_target" else: delivery_channel = "" delivery_target = "" if not binding_channel or not binding_target: delivery_reason = "企业微信严格绑定缺少 binding_channel / binding_target" else: delivery_reason = "企业微信 strict family 缺少可用的主动推送配置；推荐在技能层从 inbound metadata 读取 sender_id 并传给 --wecom-user-id" contract = { "binding_channel": binding_channel, "binding_target": binding_target, "binding_mode": binding_mode, "strict_binding": binding_mode == "strict", "delivery_family": delivery_family, "outbound_adapter": outbound_adapter, "outbound_webhook_url": outbound_webhook_url, "outbound_webhook_secret": outbound_webhook_secret, "delivery_channel": delivery_channel, "delivery_target": delivery_target, "delivery_ready": delivery_ready, "delivery_reason": delivery_reason, "updated_at": now_iso(), } config["binding_channel"] = binding_channel config["binding_target"] = binding_target config["binding_mode"] = binding_mode config["outbound_adapter"] = outbound_adapter if outbound_webhook_url: config["outbound_webhook_url"] = outbound_webhook_url if outbound_webhook_secret: config["outbound_webhook_secret"] = outbound_webhook_secret elif not outbound_webhook_url: config.pop("outbound_webhook_secret", None) if hinted_channel and hinted_target: config["channel"] = hinted_channel config["chat_id"] = hinted_target elif binding_channel and binding_target: config.setdefault("channel", binding_channel) config.setdefault("chat_id", binding_target) if delivery_channel and delivery_target: config["delivery_channel"] = delivery_channel config["delivery_target"] = delivery_target else: config.pop("delivery_channel", None) config.pop("delivery_target", None) config["delivery_contract"] = contract config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps(contract, ensure_ascii=False)) PY } lobster_update_cron_registration() { local config_path="$1" local status="$2" local reason="${3:-}" local morning_time="${4:-}" local discovery_time="${5:-}" local evening_time="${6:-}" local memory_mode="${7:-}" local managed_jobs_json="${8:-[]}" CONFIG_PATH="$config_path" \ CRON_STATUS_VALUE="$status" \ CRON_REASON_VALUE="$reason" \ MORNING_TIME_VALUE="$morning_time" \ DISCOVERY_TIME_VALUE="$discovery_time" \ EVENING_TIME_VALUE="$evening_time" \ MEMORY_MODE_VALUE="$memory_mode" \ MANAGED_JOBS_JSON_VALUE="$managed_jobs_json" \ python3 <<'PY' import json import os from datetime import datetime, timezone from pathlib import Path def now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") config_path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} status = (os.environ.get("CRON_STATUS_VALUE") or "unregistered").strip() or "unregistered" reason = (os.environ.get("CRON_REASON_VALUE") or "").strip() registered = status == "registered" cron_registration = config.get("cron_registration") if isinstance(config.get("cron_registration"), dict) else {} cron_registration["status"] = status cron_registration["registered"] = registered cron_registration["ready"] = registered cron_registration["updated_at"] = now_iso() if reason: cron_registration["reason"] = reason else: cron_registration.pop("reason", None) if registered: cron_registration["last_registered_at"] = cron_registration["updated_at"] try: managed_jobs = json.loads(os.environ.get("MANAGED_JOBS_JSON_VALUE") or "[]") if not isinstance(managed_jobs, list): managed_jobs = [] except Exception: managed_jobs = [] cron_registration["managed_job_names"] = [str(item).strip() for item in managed_jobs if str(item).strip()] cron_registration["desired_schedule"] = { "morning_time": (os.environ.get("MORNING_TIME_VALUE") or "").strip(), "discovery_time": (os.environ.get("DISCOVERY_TIME_VALUE") or "").strip(), "evening_time": (os.environ.get("EVENING_TIME_VALUE") or "").strip(), "memory_mode": (os.environ.get("MEMORY_MODE_VALUE") or "").strip(), } config["cron_registration"] = cron_registration config["cron_registered"] = registered config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps(cron_registration, ensure_ascii=False)) PY } build_candidate_lines_with_policy() { local config_path="$1" local sessions_json="$2" local forced_channel="$3" local forced_target="$4" CONFIG_PATH="$config_path" \ SESSIONS_JSON="$sessions_json" \ FORCED_CHANNEL="$forced_channel" \ FORCED_TARGET="$forced_target" \ python3 <<'PY' import json import os from pathlib import Path STRICT_MARKERS = ( "wecom", "wecom-bot", "wxwork", "qywx", "enterprisewechat", "enterprise-wechat", "workwechat", ) def detect_mode(channel: str) -> str: normalized = (channel or "").strip().lower() if any(marker in normalized for marker in STRICT_MARKERS): return "strict" return "prefer" config_path = Path(os.environ["CONFIG_PATH"]) raw_sessions = os.environ.get("SESSIONS_JSON", "[]") forced_channel = (os.environ.get("FORCED_CHANNEL") or "").strip() forced_target = (os.environ.get("FORCED_TARGET") or "").strip() try: sessions = json.loads(raw_sessions) if not isinstance(sessions, list): sessions = [sessions] except Exception: sessions = [] try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} ordered = [] seen = set() def add(channel, peer_id): channel = (channel or "").strip() peer_id = (peer_id or "").strip() if not channel or not peer_id or channel == "unknown": return key = (channel, peer_id) if key in seen: return seen.add(key) ordered.append({"channel": channel, "peer_id": peer_id}) binding_channel = str(contract.get("binding_channel") or config.get("binding_channel") or config.get("channel") or "").strip() binding_target = str(contract.get("binding_target") or config.get("binding_target") or config.get("chat_id") or "").strip() binding_mode = str(contract.get("binding_mode") or config.get("binding_mode") or detect_mode(binding_channel)).strip().lower() or "prefer" strict_binding = binding_mode == "strict" delivery_channel = str(contract.get("delivery_channel") or config.get("delivery_channel") or "").strip() delivery_target = str(contract.get("delivery_target") or config.get("delivery_target") or "").strip() if forced_channel and forced_target: add(forced_channel, forced_target) if strict_binding: add(delivery_channel or binding_channel, delivery_target or binding_target) else: add(delivery_channel or binding_channel, delivery_target or binding_target) for session in sessions: direct_channel = session.get("channel") or session.get("platform") or session.get("imChannel") or "" direct_target = session.get("peer_id") or session.get("peerId") or session.get("target") or session.get("chat_id") or session.get("chatId") or "" if direct_channel and direct_target: add(direct_channel, direct_target) continue key = session.get("sessionKey") or session.get("key") or session.get("id") or "" if not key: continue parts = key.split(":") if not parts: continue if parts[0].lower() in ("cron", "hook"): continue if len(parts) <= 3 and parts[-1].lower() == "main": continue if len(parts) >= 5 and parts[0].lower() == "agent": channel = parts[2] marker = parts[3].lower() if marker in ("direct", "dm"): add(channel, parts[4]) for item in config.get("known_channels", []): if isinstance(item, dict): add(item.get("channel"), item.get("peer_id")) add(config.get("channel"), config.get("chat_id")) for item in ordered: print(f"{item['channel']}|{item['peer_id']}") PY } update_config_after_send_with_policy() { local config_path="$1" local current_channel="$2" local current_target="$3" local candidate_lines="$4" local delivered_at="$5" CONFIG_PATH="$config_path" \ CURRENT_CHANNEL="$current_channel" \ CURRENT_TARGET="$current_target" \ CANDIDATE_LINES="$candidate_lines" \ DELIVERED_AT="$delivered_at" \ python3 <<'PY' import json import os from pathlib import Path STRICT_MARKERS = ( "wecom", "wecom-bot", "wxwork", "qywx", "enterprisewechat", "enterprise-wechat", "workwechat", ) def detect_mode(channel: str) -> str: normalized = (channel or "").strip().lower() if any(marker in normalized for marker in STRICT_MARKERS): return "strict" return "prefer" config_path = Path(os.environ["CONFIG_PATH"]) current_channel = os.environ.get("CURRENT_CHANNEL", "").strip() current_target = os.environ.get("CURRENT_TARGET", "").strip() lines = [line.strip() for line in os.environ.get("CANDIDATE_LINES", "").splitlines() if line.strip()] delivered_at = os.environ.get("DELIVERED_AT", "").strip() try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} contract = config.get("delivery_contract") if isinstance(config.get("delivery_contract"), dict) else {} binding_channel = str(contract.get("binding_channel") or config.get("binding_channel") or config.get("channel") or "").strip() binding_target = str(contract.get("binding_target") or config.get("binding_target") or config.get("chat_id") or "").strip() binding_mode = str(contract.get("binding_mode") or config.get("binding_mode") or detect_mode(binding_channel)).strip().lower() or "prefer" strict_binding = binding_mode == "strict" ordered = [] seen = set() def add(channel, peer_id): channel = (channel or "").strip() peer_id = (peer_id or "").strip() if not channel or not peer_id: return key = (channel, peer_id) if key in seen: return seen.add(key) ordered.append({"channel": channel, "peer_id": peer_id}) add(current_channel, current_target) if binding_channel and binding_target: add(binding_channel, binding_target) for line in lines: channel, _, peer_id = line.partition("|") add(channel, peer_id) for item in config.get("known_channels", []): if isinstance(item, dict): add(item.get("channel"), item.get("peer_id")) if strict_binding and binding_channel and binding_target: config["channel"] = binding_channel config["chat_id"] = binding_target config["binding_channel"] = binding_channel config["binding_target"] = binding_target else: if current_channel and current_target: config["channel"] = current_channel config["chat_id"] = current_target config["binding_channel"] = current_channel config["binding_target"] = current_target config["binding_mode"] = binding_mode config["known_channels"] = ordered if current_channel and current_target: config["last_delivery_channel"] = current_channel config["last_delivery_target"] = current_target if delivered_at: config["last_delivery_at"] = delivered_at contract["binding_channel"] = str(config.get("binding_channel") or binding_channel or "").strip() contract["binding_target"] = str(config.get("binding_target") or binding_target or "").strip() contract["binding_mode"] = binding_mode if current_channel and current_target: contract["delivery_channel"] = current_channel contract["delivery_target"] = current_target config["delivery_contract"] = contract config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps(config.get("known_channels", []), ensure_ascii=False)) PY } FILE:send-current-screenshot.sh #!/bin/bash # ═══════════════════════════════════════════════ # 虾说 — 互动场景截图发送脚本 # ═══════════════════════════════════════════════ # # 数据访问声明： # - 读取 .lobster-config（技能自身配置） # - 调用 openclaw sessions --json 获取活跃 IM 通道元数据 # - 调用 openclaw message send 向 IM 通道发送截图摘要（统一使用 --target 参数） # - 与 https://nixiashuo.com 通信：获取工作室短链和截图 # - 不读取 openclaw.json 配置文件，不提取 gateway token # # 设计目标： # 1. 处理用户在 IM 中"发个图 / 看看虾在干嘛"这类即时截图请求 # 2. 自动优先命中最近活跃的 direct session 通道 # 3. Telegram/Discord/Slack 等支持媒体消息的通道优先发原生图片 # 4. 企业微信严格绑定模式走 sender_id 私聊（openclaw message send --channel openclaw-wecom-bot --to ） # 5. 明确禁止输出 / 本地临时文件路径给用户 # # 用法： # bash send-current-screenshot.sh # bash send-current-screenshot.sh --caption "这是旺仔3号现在的样子~" # bash send-current-screenshot.sh --with-status-summary # bash send-current-screenshot.sh --channel telegram --to 123456789 set -euo pipefail BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" COMMON_SCRIPT="${BASE_DIR}/runtime-common.sh" SETUP_CRON_SCRIPT="${BASE_DIR}/setup-cron.sh" ACTIVE_WINDOW_MINUTES="${ACTIVE_WINDOW_MINUTES:-10080}" # Telegram 原生 Bot API 媒体能力稳定；企微/飞书等机器人协议更依赖 webhook 特定图片格式或素材上传， # 统一经 OpenClaw CLI 时优先退化到文本 + 链接更稳妥。 MEDIA_SEND_CHANNELS="telegram discord googlechat slack mattermost signal imessage msteams" OPENCLAW_BIN="${OPENCLAW_BIN:-openclaw}" CAPTION="" FORCED_CHANNEL="" FORCED_TARGET="" WITH_STATUS_SUMMARY=false LOCAL_SCREENSHOT_FILE="" _SEND_STDERR_FILE="" POLICY_JSON="" BOUND_CHANNEL="" BOUND_TARGET="" STRICT_BINDING="0" OUTBOUND_ADAPTER="openclaw" OUTBOUND_WEBHOOK_URL="" OUTBOUND_WEBHOOK_SECRET="" DELIVERY_CHANNEL="" DELIVERY_TARGET="" DELIVERY_READY="0" DELIVERY_REASON="" CRON_STATUS="unregistered" GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' CYAN='\033[0;36m' NC='\033[0m' cleanup() { if [ -n "${LOCAL_SCREENSHOT_FILE:-}" ] && [ -f "${LOCAL_SCREENSHOT_FILE:-}" ]; then rm -f "${LOCAL_SCREENSHOT_FILE}" fi if [ -n "${_SEND_STDERR_FILE:-}" ] && [ -f "${_SEND_STDERR_FILE:-}" ]; then rm -f "${_SEND_STDERR_FILE}" fi } trap cleanup EXIT info() { echo -e "${GREEN}[✓]${NC} $1"; } warn() { echo -e "${YELLOW}[!]${NC} $1"; } error() { echo -e "${RED}[✗]${NC} $1"; exit 1; } step() { echo -e "${CYAN}──${NC} $1"; } while [[ $# -gt 0 ]]; do case "$1" in --caption) CAPTION="$2"; shift 2 ;; --channel) FORCED_CHANNEL="$2"; shift 2 ;; --to|--target) FORCED_TARGET="$2"; shift 2 ;; --with-status-summary) WITH_STATUS_SUMMARY=true; shift ;; *) echo "未知参数: $1"; exit 1 ;; esac done for cmd in python3 curl; do if ! command -v "$cmd" >/dev/null 2>&1; then error "$cmd 不可用" fi done if ! command -v "$OPENCLAW_BIN" >/dev/null 2>&1; then error "$OPENCLAW_BIN 不可用" fi if [ ! -f "$CONFIG_FILE" ]; then error ".lobster-config 不存在，请先初始化虾" fi if [ ! -f "$COMMON_SCRIPT" ]; then error "共享运行时脚本不存在：${COMMON_SCRIPT}" fi . "$COMMON_SCRIPT" read_config_value() { local key="$1" local default_value="${2:-}" CONFIG_PATH="$CONFIG_FILE" KEY_NAME="$key" DEFAULT_VALUE="$default_value" python3 - <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) key = os.environ["KEY_NAME"] default = os.environ.get("DEFAULT_VALUE", "") try: data = json.loads(path.read_text(encoding="utf-8")) except Exception: data = {} value = data.get(key, default) print(value if value is not None else default) PY } supports_media() { local channel="$1" for ch in $MEDIA_SEND_CHANNELS; do if [ "$ch" = "$channel" ]; then return 0 fi done return 1 } fetch_status_summary_text() { local api_base="$1" local user_id="$2" local access_token="$3" local fallback_name="$4" local status_json status_json="$(curl -fsS "${api_base}/api/lobster/${user_id}/status" \ -H "Authorization: Bearer ${access_token}")" || return 1 STATUS_JSON="$status_json" STATUS_FALLBACK_NAME="$fallback_name" python3 - <<'PYTHON' import json import os import sys STATUS_LABELS = { "idle": "待命", "working": "忙活", "sleeping": "睡着", "daydreaming": "发呆", "slacking": "摸鱼", "running": "乱窜", "crazy": "发疯", "excited": "兴奋", } def clean(text): return str(text or "").replace("", "").replace("", "").strip() try: data = json.loads(os.environ.get("STATUS_JSON") or "{}") except Exception: sys.exit(1) name = clean(data.get("name")) or os.environ.get("STATUS_FALLBACK_NAME") or "这只虾" status = clean(data.get("status")) or "idle" status_label = STATUS_LABELS.get(status, status) reason = clean(data.get("status_reason")) latest_message = clean(data.get("latest_message")) lines = [f"{name}现在在{status_label}。", "", f"• 当前状态：{status}"] if reason: lines.append(f"• 状态说明：{reason}") if latest_message: lines.extend(["• 它最新一句是：", "", latest_message]) print("\n".join(lines)) PYTHON } fetch_studio_links() { local api_base="$1" local user_id="$2" local access_token="$3" local links_json links_json="$(curl -fsS "${api_base}/api/lobster/${user_id}/studio-link" \ -H "Authorization: Bearer ${access_token}")" || return 1 LINKS_JSON="$links_json" python3 - <<'PYLINK' import json import os import sys try: data = json.loads(os.environ.get("LINKS_JSON") or "{}") except Exception: sys.exit(1) web_url = str(data.get("web_url") or "").strip() screenshot_url = str(data.get("screenshot_url") or "").strip() if not web_url or not screenshot_url: sys.exit(1) print(f"{web_url}|{screenshot_url}") PYLINK } fetch_local_screenshot_file() { local api_base="$1" local user_id="$2" local access_token="$3" local media_dir="${HOME}/.openclaw/media" local output_file mkdir -p "$media_dir" || return 1 output_file="$(mktemp "${media_dir}/lobster-screenshot.XXXXXX.png")" || return 1 if ! curl -fsS "${api_base}/api/lobster/${user_id}/screenshot.png" \ -H "Authorization: Bearer ${access_token}" \ -o "$output_file"; then rm -f "$output_file" return 1 fi if [ ! -s "$output_file" ]; then rm -f "$output_file" return 1 fi echo "$output_file" } load_runtime_policy() { POLICY_JSON=$(lobster_runtime_policy_json "$CONFIG_FILE") mapfile -t POLICY_LINES < <(POLICY_JSON_VALUE="$POLICY_JSON" python3 <<'PY' import json import os policy = json.loads(os.environ["POLICY_JSON_VALUE"]) print(policy.get("binding_channel", "")) print(policy.get("binding_target", "")) print("1" if policy.get("strict_binding") else "0") print(policy.get("outbound_adapter", "openclaw")) print(policy.get("outbound_webhook_url", "")) print(policy.get("outbound_webhook_secret", "")) print(policy.get("delivery_channel", "")) print(policy.get("delivery_target", "")) print("1" if policy.get("delivery_ready") else "0") print(policy.get("delivery_reason", "")) print(policy.get("cron_status", "unregistered")) PY ) BOUND_CHANNEL="${POLICY_LINES[0]:-}" BOUND_TARGET="${POLICY_LINES[1]:-}" STRICT_BINDING="${POLICY_LINES[2]:-0}" OUTBOUND_ADAPTER="${POLICY_LINES[3]:-openclaw}" OUTBOUND_WEBHOOK_URL="${POLICY_LINES[4]:-}" OUTBOUND_WEBHOOK_SECRET="${POLICY_LINES[5]:-}" DELIVERY_CHANNEL="${POLICY_LINES[6]:-}" DELIVERY_TARGET="${POLICY_LINES[7]:-}" DELIVERY_READY="${POLICY_LINES[8]:-0}" DELIVERY_REASON="${POLICY_LINES[9]:-}" CRON_STATUS="${POLICY_LINES[10]:-unregistered}" } sync_delivery_contract_state() { lobster_sync_delivery_contract "$CONFIG_FILE" "$FORCED_CHANNEL" "$FORCED_TARGET" >/dev/null load_runtime_policy } reliable_send() { _SEND_FAIL_REASON="" if [ -z "${_SEND_STDERR_FILE:-}" ]; then _SEND_STDERR_FILE=$(mktemp "${TMPDIR:-/tmp}/lobster-send-stderr.XXXXXX") fi "$OPENCLAW_BIN" message send "$@" >"$_SEND_STDERR_FILE" 2>&1 local rc=$? local stderr_content="" [ -f "$_SEND_STDERR_FILE" ] && stderr_content=$(cat "$_SEND_STDERR_FILE" 2>/dev/null) if [ $rc -ne 0 ]; then _SEND_FAIL_REASON="exit_code=${rc} ${stderr_content}" return 1 fi if echo "$stderr_content" | grep -qiE '(wsclient|websocket|not.connected|connection.refused|connection.reset|connection.timeout|no.active.session|failed.to.send|send.failed|delivery.failed)'; then _SEND_FAIL_REASON="假成功(exit=0 但检测到连接问题): ${stderr_content}" return 1 fi return 0 } send_via_wecom_direct_message() { local message_text="$1" local wecom_user_id wecom_user_id=$(read_config_value "wecom_user_id") [ -n "$wecom_user_id" ] || { _SEND_FAIL_REASON="企业微信截图发送缺少 sender_id / wecom_user_id"; return 1; } # 使用 openclaw message send --target 统一参数 local send_result rc send_result=$("$OPENCLAW_BIN" message send \ --channel "openclaw-wecom-bot" \ --target "$wecom_user_id" \ --message "$message_text" 2>&1) rc=$? if [ $rc -ne 0 ]; then _SEND_FAIL_REASON="openclaw message send 失败 (rc=${rc}): ${send_result}" return 1 fi info "企业微信截图已直接送达 → ${wecom_user_id}" _SEND_FAIL_REASON="" return 0 } build_fallback_message() { local channel="$1" FALLBACK_CHANNEL="$channel" TEXT_CAPTION_VALUE="$TEXT_CAPTION" LOBSTER_NAME_VALUE="$LOBSTER_NAME" SCREENSHOT_URL_VALUE="$SCREENSHOT_URL" WEB_URL_VALUE="$WEB_URL" python3 <<'PY' import os channel = os.environ.get("FALLBACK_CHANNEL", "") caption = (os.environ.get("TEXT_CAPTION_VALUE") or "").strip() lobster_name = os.environ["LOBSTER_NAME_VALUE"] screenshot_url = os.environ["SCREENSHOT_URL_VALUE"] web_url = os.environ["WEB_URL_VALUE"] if not caption: caption = f"给你看看{lobster_name}现在在忙什么。" lines = [caption] if screenshot_url: lines.extend(["", f"📸 {lobster_name}的工作室截图：{screenshot_url}"]) elif channel == "openclaw-weixin": lines.extend(["", "📸 当前会话暂时没有可访问的截图链接，我先把状态告诉你。"]) if web_url: lines.append(f"👀 看看{lobster_name}在干嘛 → {web_url}") print("\n".join(lines)) PY } maybe_activate_pending_cron() { if [ "$DELIVERY_READY" = "1" ] && [ "$CRON_STATUS" = "pending_activation" ] && [ -f "$SETUP_CRON_SCRIPT" ]; then step "检测到待激活 cron，尝试自动补注册..." if bash "$SETUP_CRON_SCRIPT" >/dev/null 2>&1; then info "待激活的定时推送已自动补注册" else warn "自动补注册 cron 失败，可稍后手动执行 setup-cron.sh" fi fi } LOBSTER_NAME="$(read_config_value lobster_name 小虾)" USER_ID="$(read_config_value user_id)" ACCESS_TOKEN="$(read_config_value access_token)" API_BASE="$(read_config_value api_base https://nixiashuo.com)" if [ -z "$USER_ID" ] || [ -z "$ACCESS_TOKEN" ]; then error ".lobster-config 缺少 user_id / access_token" fi if [ -n "$FORCED_CHANNEL" ] && [ -z "$FORCED_TARGET" ]; then error "指定 --channel 时必须同时指定 --to/--target" fi if [ -n "$FORCED_TARGET" ] && [ -z "$FORCED_CHANNEL" ]; then error "指定 --to/--target 时必须同时指定 --channel" fi sync_delivery_contract_state if [ "$STRICT_BINDING" = "1" ]; then info "当前为严格绑定模式：${BOUND_CHANNEL:-未锁定} → ${BOUND_TARGET:-未锁定}" fi if STUDIO_LINKS="$(fetch_studio_links "$API_BASE" "$USER_ID" "$ACCESS_TOKEN" 2>/dev/null)"; then WEB_URL="${STUDIO_LINKS%%|*}" SCREENSHOT_URL="${STUDIO_LINKS#*|}" else warn "短时工作室链接获取失败，不再回退长期 token URL。" WEB_URL="" SCREENSHOT_URL="" if LOCAL_SCREENSHOT_FILE="$(fetch_local_screenshot_file "$API_BASE" "$USER_ID" "$ACCESS_TOKEN" 2>/dev/null)"; then info "已生成受控本地截图，将优先尝试原生图片发送。" else warn "本地截图兜底也失败，将仅发送文本状态提示。" fi fi TEXT_CAPTION="${CAPTION:-这是${LOBSTER_NAME}现在的样子~}" if [ "$WITH_STATUS_SUMMARY" = true ]; then step "拉取当前状态摘要..." if STATUS_SUMMARY="$(fetch_status_summary_text "$API_BASE" "$USER_ID" "$ACCESS_TOKEN" "$LOBSTER_NAME")"; then TEXT_CAPTION="$STATUS_SUMMARY" else warn "状态摘要拉取失败，退回默认截图文案" fi fi LAST_ERROR="" SUCCESS_CHANNEL="" SUCCESS_TARGET="" SUCCESS_MODE="" CANDIDATE_LINES="" if [ "$STRICT_BINDING" = "1" ]; then step "通过企业微信 sender_id 私聊发送截图摘要..." FALLBACK_MESSAGE="$(build_fallback_message "wecom")" if send_via_wecom_direct_message "$FALLBACK_MESSAGE"; then WECOM_USER_ID_RESOLVED="$(read_config_value wecom_user_id)" SUCCESS_CHANNEL="wecom" SUCCESS_TARGET="${WECOM_USER_ID_RESOLVED:-wecom_direct}" if [ -n "$SCREENSHOT_URL" ]; then SUCCESS_MODE="wecom_direct_url" else SUCCESS_MODE="wecom_direct_text" fi else LAST_ERROR="${_SEND_FAIL_REASON}" fi elif [ "$OUTBOUND_ADAPTER" = "openclaw" ] || [ "$OUTBOUND_ADAPTER" = "wecom-direct-message" ]; then step "解析候选通道..." SESSIONS_JSON="$($OPENCLAW_BIN sessions --json --active "$ACTIVE_WINDOW_MINUTES" 2>/dev/null || echo '[]')" CANDIDATE_LINES="$(build_candidate_lines_with_policy "$CONFIG_FILE" "$SESSIONS_JSON" "$FORCED_CHANNEL" "$FORCED_TARGET")" if [ -z "$CANDIDATE_LINES" ]; then error "没有找到可用的投递通道。请先在 Telegram / 微信 / 飞书等任一通道里和虾说一句话。" fi while IFS='|' read -r channel target; do [ -z "$channel" ] && continue [ -z "$target" ] && continue step "尝试发送到 ${channel} → ${target}" if supports_media "$channel"; then if reliable_send --channel "$channel" --target "$target" --message "$TEXT_CAPTION"; then MEDIA_SOURCE="$SCREENSHOT_URL" if [ -z "$MEDIA_SOURCE" ] && [ -n "$LOCAL_SCREENSHOT_FILE" ]; then MEDIA_SOURCE="$LOCAL_SCREENSHOT_FILE" fi if [ -n "$MEDIA_SOURCE" ] && reliable_send --channel "$channel" --target "$target" --media "$MEDIA_SOURCE"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" SUCCESS_MODE="media" info "文本 + 原生截图发送成功" break fi warn "${channel} 原生截图发送失败 (${_SEND_FAIL_REASON})，降级为文本提示" FALLBACK_MESSAGE="$(build_fallback_message "$channel")" if reliable_send --channel "$channel" --target "$target" --message "$FALLBACK_MESSAGE"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" if [ -n "$SCREENSHOT_URL" ]; then SUCCESS_MODE="degraded_url" info "已降级为文本 + 截图 URL" else SUCCESS_MODE="text_only" info "已降级为纯文本提示" fi break fi LAST_ERROR="${channel} 原生截图与文本降级都失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续尝试下一个通道" continue fi LAST_ERROR="${channel} 文本消息发送失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续尝试下一个通道" continue fi FALLBACK_MESSAGE="$(build_fallback_message "$channel")" if reliable_send --channel "$channel" --target "$target" --message "$FALLBACK_MESSAGE"; then SUCCESS_CHANNEL="$channel" SUCCESS_TARGET="$target" if [ -n "$SCREENSHOT_URL" ]; then SUCCESS_MODE="url" info "纯文本 + 截图 URL 发送成功" else SUCCESS_MODE="text_only" info "纯文本状态提示发送成功" fi break fi LAST_ERROR="${channel} 文本消息发送失败: ${_SEND_FAIL_REASON}" warn "$LAST_ERROR，继续尝试下一个通道" done <<< "$CANDIDATE_LINES" fi if [ -z "$SUCCESS_CHANNEL" ]; then error "所有候选通道都投递失败：${LAST_ERROR:-未知错误}" fi UPDATED_KNOWN="$(update_config_after_send_with_policy "$CONFIG_FILE" "$SUCCESS_CHANNEL" "$SUCCESS_TARGET" "$CANDIDATE_LINES" "$(date -u '+%Y-%m-%dT%H:%M:%SZ')")" info "截图已送达：${SUCCESS_CHANNEL} → ${SUCCESS_TARGET} (${SUCCESS_MODE})" maybe_activate_pending_cron echo "SCREENSHOT_SENT channel=${SUCCESS_CHANNEL} target=${SUCCESS_TARGET} mode=${SUCCESS_MODE}" echo "KNOWN_CHANNELS=${UPDATED_KNOWN}" FILE:send-studio-link.sh #!/bin/bash # ═══════════════════════════════════════════════ # 虾说 — 工作室短链获取脚本 # ═══════════════════════════════════════════════ # # 设计目标： # 1. 用户在 IM 里索要“工作室链接”时，强制实时获取 fresh 短链 # 2. 避免 agent 复用历史对话中的旧 URL，或手工拼接 /lobster/{user_id}?st=... # 3. 默认输出可直接回给用户的文本；也支持 plain/json 供脚本链路复用 # # 用法： # bash send-studio-link.sh # bash send-studio-link.sh --plain-url # bash send-studio-link.sh --json set -euo pipefail BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" OUTPUT_MODE="message" GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' NC='\033[0m' info() { echo -e "${GREEN}[✓]${NC} $1" >&2; } warn() { echo -e "${YELLOW}[!]${NC} $1" >&2; } error() { echo -e "${RED}[✗]${NC} $1" >&2; exit 1; } while [[ $# -gt 0 ]]; do case "$1" in --plain-url) OUTPUT_MODE="plain" shift ;; --json) OUTPUT_MODE="json" shift ;; *) error "未知参数: $1" ;; esac done for cmd in python3 curl; do command -v "$cmd" >/dev/null 2>&1 || error "$cmd 不可用" done [ -f "$CONFIG_FILE" ] || error ".lobster-config 不存在，请先初始化虾" read_config_value() { local key="$1" local default_value="${2:-}" CONFIG_PATH="$CONFIG_FILE" KEY_NAME="$key" DEFAULT_VALUE="$default_value" python3 - <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) key = os.environ["KEY_NAME"] default = os.environ.get("DEFAULT_VALUE", "") try: data = json.loads(path.read_text(encoding="utf-8")) except Exception: data = {} value = data.get(key, default) print(value if value is not None else default) PY } LOBSTER_NAME="$(read_config_value lobster_name 小虾)" USER_ID="$(read_config_value user_id)" ACCESS_TOKEN="$(read_config_value access_token)" API_BASE="$(read_config_value api_base https://nixiashuo.com)" [ -n "$USER_ID" ] || error ".lobster-config 缺少 user_id" [ -n "$ACCESS_TOKEN" ] || error ".lobster-config 缺少 access_token" info "正在实时获取 fresh 工作室短链..." LINKS_JSON="$(curl -fsS "${API_BASE}/api/lobster/${USER_ID}/studio-link" -H "Authorization: Bearer ${ACCESS_TOKEN}")" || error "工作室短链获取失败" OUTPUT_MODE_VALUE="$OUTPUT_MODE" LOBSTER_NAME_VALUE="$LOBSTER_NAME" LINKS_JSON_VALUE="$LINKS_JSON" python3 - <<'PY' import json import os import sys mode = os.environ["OUTPUT_MODE_VALUE"] lobster_name = os.environ.get("LOBSTER_NAME_VALUE") or "这只虾" try: data = json.loads(os.environ.get("LINKS_JSON_VALUE") or "{}") except Exception: sys.exit(1) web_url = str(data.get("web_url") or "").strip() screenshot_url = str(data.get("screenshot_url") or "").strip() expires_at = str(data.get("expires_at") or "").strip() if not web_url: sys.exit(1) if mode == "plain": print(web_url) sys.exit(0) if mode == "json": print(json.dumps({ "web_url": web_url, "screenshot_url": screenshot_url, "expires_at": expires_at, }, ensure_ascii=False, indent=2)) sys.exit(0) lines = [ f"给你，{lobster_name}的 fresh 工作室短链：", "", web_url, "", "这是短时 st 入口，到期我可以再给你刷新。", ] print("\n".join(lines)) PY FILE:setup-cron.sh #!/bin/bash # 虾说 — 定时推送 Cron 注册脚本 # # 数据访问声明： # - 读取 .lobster-config（技能自身配置） # - 调用 openclaw sessions --json 获取活跃 IM 通道元数据 # - 调用 openclaw cron add/remove/list 管理定时任务 # - 不读取 openclaw.json 配置文件，不提取 gateway token # - gateway 认证仅通过用户显式设置的 OPENCLAW_GATEWAY_TOKEN 环境变量 # # 重构设计（v2.5.3 可测版）： # - 企微通道：cron 注册带 --channel openclaw-wecom-bot --to （delivery 兜底） # agent prompt 只做两件事：执行 push-scheduled-message.sh --emit-message-text 取得最终文本， # 然后必须使用 message 工具发送私聊；不再依赖脚本内 CLI 直发 # - 通用通道（Telegram等）：仍由脚本完整模式执行，多通道 fallback + delivery 兜底 # - openclaw cron add 继续使用 --to；但 openclaw message send 一律使用 --target，避免 CLI 参数漂移 set -u ACTION="reconcile" CHANNEL="" TO="" WECOM_USER_ID="" MORNING_TIME="" DISCOVERY_TIME="" EVENING_TIME="" MEMORY_MODE="" REMOVE_JOB_NAME="" SCHEDULE_INIT_READY_DELAY="" INIT_READY_DELAY_MINUTES="${INIT_READY_DELAY_MINUTES:-3}" INIT_READY_MIN_LEAD_MINUTES="${INIT_READY_MIN_LEAD_MINUTES:-2}" INIT_READY_VERIFY_ATTEMPTS="${INIT_READY_VERIFY_ATTEMPTS:-3}" ACTIVE_WINDOW_MINUTES="${ACTIVE_WINDOW_MINUTES:-10080}" BASE_DIR="$(cd "$(dirname "$0")" && pwd)" CONFIG_FILE="${BASE_DIR}/.lobster-config" COMMON_SCRIPT="${BASE_DIR}/runtime-common.sh" LOG_DIR="${BASE_DIR}/logs" RUN_TS="$(date +%Y%m%d-%H%M%S)" LOG_FILE="${LOG_DIR}/setup-cron-${RUN_TS}.log" PUSH_SCRIPT="${BASE_DIR}/push-scheduled-message.sh" OPENCLAW_CRON_ARGS=() INIT_READY_JOB_NAME="lobster-says-init-ready" DETECTED_KNOWN_JSON="[]" DELIVERY_CHANNEL="" DELIVERY_TARGET="" DELIVERY_READY="0" DELIVERY_REASON="" DELIVERY_FAMILY="general" OUTBOUND_ADAPTER="openclaw" BINDING_CHANNEL="" BINDING_TARGET="" BINDING_MODE="prefer" MANAGED_JOB_NAMES=( "lobster-says-morning" "lobster-says-discovery" "lobster-says-evening" "lobster-says-init-ready" "lobster-says-sticker" "lobster-says-wallpaper" "lobster-says-digest" ) GREEN='\033[0;32m' YELLOW='\033[1;33m' RED='\033[0;31m' CYAN='\033[0;36m' NC='\033[0m' info() { echo -e "${GREEN}[✓]${NC} $1"; } warn() { echo -e "${YELLOW}[!]${NC} $1"; } error() { echo -e "${RED}[✗]${NC} $1"; exit 1; } CURRENT_STEP="启动" step() { CURRENT_STEP="$1"; echo -e "${CYAN}──${NC} $1"; } if [ ! -f "$COMMON_SCRIPT" ]; then error "共享运行时脚本不存在：${COMMON_SCRIPT}" fi . "$COMMON_SCRIPT" mkdir -p "$LOG_DIR" if command -v tee >/dev/null 2>&1; then exec > >(tee -a "$LOG_FILE") 2>&1 else exec >>"$LOG_FILE" 2>&1 fi echo "[log] setup-cron started at $(date '+%F %T')" echo "[log] file: ${LOG_FILE}" after_exit() { local exit_code=$? if [ "$exit_code" -ne 0 ]; then echo "" echo "[log] setup-cron failed at step: ${CURRENT_STEP} (exit=${exit_code})" echo "[log] inspect: ${LOG_FILE}" else echo "[log] setup-cron finished successfully" fi } trap after_exit EXIT while [[ $# -gt 0 ]]; do case "$1" in --channel) CHANNEL="$2"; shift 2 ;; --to) TO="$2"; shift 2 ;; --morning) MORNING_TIME="$2"; shift 2 ;; --discovery) DISCOVERY_TIME="$2"; shift 2 ;; --evening) EVENING_TIME="$2"; shift 2 ;; --memory-mode) MEMORY_MODE="$2"; shift 2 ;; --wecom-user-id) WECOM_USER_ID="$2"; shift 2 ;; --remove-job-by-name) ACTION="remove-job-by-name"; REMOVE_JOB_NAME="$2"; shift 2 ;; --schedule-init-ready-delay) ACTION="schedule-init-ready-delay"; SCHEDULE_INIT_READY_DELAY="$2"; shift 2 ;; *) echo "未知参数: $1"; exit 1 ;; esac done for cmd in openclaw python3; do if ! command -v "$cmd" >/dev/null 2>&1; then error "$cmd 不可用" fi done if [ ! -f "$CONFIG_FILE" ]; then error ".lobster-config 不存在，请先完成虾的初始化" fi if [ ! -f "$PUSH_SCRIPT" ]; then error "运行时推送脚本不存在：${PUSH_SCRIPT}" fi read_config_value() { local key="$1" local default_value="${2:-}" CONFIG_PATH="$CONFIG_FILE" KEY_NAME="$key" DEFAULT_VALUE="$default_value" python3 - <<'PY' import json import os from pathlib import Path path = Path(os.environ["CONFIG_PATH"]) key = os.environ["KEY_NAME"] default = os.environ.get("DEFAULT_VALUE", "") try: data = json.loads(path.read_text(encoding="utf-8")) value = data.get(key, default) print(value if value is not None else default) except Exception: print(default) PY } load_gateway_auth() { OPENCLAW_CRON_ARGS=() local token="${OPENCLAW_GATEWAY_TOKEN:-}" if [ -n "$token" ]; then OPENCLAW_CRON_ARGS+=(--token "$token") fi } list_managed_jobs() { local raw raw=$(openclaw cron list "${OPENCLAW_CRON_ARGS[@]}" --json 2>/dev/null || echo "") CRON_DATA="$raw" python3 <<'PY' import json import os import sys raw = os.environ.get("CRON_DATA", "") if not raw.strip(): raise SystemExit(0) json_str = None for i, ch in enumerate(raw): if ch in ("{", "["): json_str = raw[i:] break if not json_str: raise SystemExit(0) try: data = json.loads(json_str) except Exception: raise SystemExit(0) jobs = data.get("jobs", []) if isinstance(data, dict) else data if isinstance(data, list) else [] target_names = { "lobster-says-morning", "lobster-says-discovery", "lobster-says-evening", "lobster-says-init-ready", "lobster-says-sticker", "lobster-says-wallpaper", "lobster-says-digest", } for job in jobs: if isinstance(job, dict) and job.get("name") in target_names and job.get("id"): print(f"{job['name']}|{job['id']}") PY } remove_job_by_name() { local name="$1" local found=0 while IFS='|' read -r job_name job_id; do [ -z "$job_id" ] && continue [ "$job_name" = "$name" ] || continue found=1 warn "删除旧任务: ${job_name} (${job_id})" openclaw cron remove "$job_id" >/dev/null 2>&1 || warn "删除失败: ${job_name} (${job_id})" done <<< "$(list_managed_jobs)" if [ "$found" -eq 0 ]; then info "没有找到任务: ${name}" fi } remove_all_managed_jobs() { local jobs jobs=$(list_managed_jobs) if [ -z "$jobs" ]; then info "没有发现旧任务" return 0 fi while IFS='|' read -r name job_id; do [ -z "$job_id" ] && continue warn "删除旧任务: ${name} (${job_id})" openclaw cron remove "${OPENCLAW_CRON_ARGS[@]}" "$job_id" >/dev/null 2>&1 || warn "删除失败: ${name} (${job_id})" done <<< "$jobs" } read_job_snapshot_by_name() { local name="$1" local raw raw=$(openclaw cron list "${OPENCLAW_CRON_ARGS[@]}" --json 2>/dev/null || echo "") CRON_DATA="$raw" TARGET_JOB_NAME="$name" python3 <<'PY' import json import os raw = os.environ.get("CRON_DATA", "") target_name = os.environ.get("TARGET_JOB_NAME", "") if not raw.strip() or not target_name: raise SystemExit(0) json_str = None for i, ch in enumerate(raw): if ch in ("{", "["): json_str = raw[i:] break if not json_str: raise SystemExit(0) try: data = json.loads(json_str) except Exception: raise SystemExit(0) if isinstance(data, dict): jobs = data.get("jobs", []) if not isinstance(jobs, list): jobs = data.get("data", []) if isinstance(data.get("data"), list) else [] elif isinstance(data, list): jobs = data else: jobs = [] def pick_text(payload, keys): if not isinstance(payload, dict): return "" for key in keys: value = payload.get(key) if value not in (None, ""): return str(value).strip() return "" def pick_nested(payload, path_groups): for path in path_groups: current = payload valid = True for key in path: if not isinstance(current, dict): valid = False break current = current.get(key) if valid and current not in (None, ""): return current return None def normalize_next_run(value): if value in (None, ""): return "" if isinstance(value, str): text = value.strip() if not text: return "" if text.isdigit(): value = int(text) else: return text if isinstance(value, (int, float)): timestamp = float(value) if timestamp > 1000000000000: timestamp /= 1000.0 from datetime import datetime, timezone return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat().replace("+00:00", "Z") return str(value).strip() for job in jobs: if not isinstance(job, dict) or str(job.get("name") or "").strip() != target_name: continue next_run_value = pick_nested(job, [ ("nextRun",), ("next_run",), ("nextRunAt",), ("next_run_at",), ("nextExecution",), ("next_execution",), ("nextExecutionAt",), ("next_execution_at",), ("nextFireAt",), ("next_fire_at",), ("schedule", "nextRun"), ("schedule", "next_run"), ("schedule", "nextRunAt"), ("schedule", "next_run_at"), ("schedule", "nextExecution"), ("schedule", "next_execution"), ("schedule", "nextExecutionAt"), ("schedule", "next_execution_at"), ("schedule", "nextFireAt"), ("schedule", "next_fire_at"), ("state", "nextRunAtMs"), ("state", "next_run_at_ms"), ("state", "nextRunAt"), ("state", "next_run_at"), ("state", "nextExecutionAt"), ("state", "next_execution_at"), ]) next_run = normalize_next_run(next_run_value) state_payload = job.get("state") if isinstance(job.get("state"), dict) else {} status = pick_text(job, ("status",)) or pick_text(state_payload, ("status", "phase", "name")) print(str(job.get("id") or "").strip()) print(next_run) print(status) print(json.dumps(job, ensure_ascii=False)) break PY } validate_init_ready_next_run() { local next_run="$1" NEXT_RUN_VALUE="$next_run" python3 <<'PY' from datetime import datetime, timedelta, timezone import os raw = (os.environ.get("NEXT_RUN_VALUE") or "").strip() if not raw or raw.lower() == "null": print("invalid|nextRun 为空") raise SystemExit(0) app_tz = timezone(timedelta(hours=8)) now = datetime.now(app_tz) def mark(dt, label): if dt.tzinfo is None: dt = dt.replace(tzinfo=app_tz) else: dt = dt.astimezone(app_tz) if dt <= now: print(f"invalid|nextRun 已过期: {label}") else: print(f"ok|{label}") raise SystemExit(0) try: mark(datetime.fromisoformat(raw.replace("Z", "+00:00")), raw) except Exception: pass for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y/%m/%d %H:%M:%S", "%Y/%m/%d %H:%M"): try: mark(datetime.strptime(raw, fmt), raw) except Exception: pass print(f"ok|{raw}") PY } parse_time() { local time_str="$1" local hour local minute hour=$(echo "$time_str" | cut -d: -f1 | sed 's/^0//') minute=$(echo "$time_str" | cut -d: -f2 | sed 's/^0//') echo "${minute:-0} ${hour:-0} * * *" } compute_init_ready_schedule() { local delay_minutes="$1" INIT_READY_DELAY_VALUE="$delay_minutes" INIT_READY_MIN_LEAD_VALUE="$INIT_READY_MIN_LEAD_MINUTES" MORNING_TIME_VALUE="$MORNING_TIME" DISCOVERY_TIME_VALUE="$DISCOVERY_TIME" EVENING_TIME_VALUE="$EVENING_TIME" python3 <<'PY' from datetime import datetime, timedelta, timezone import os app_tz = timezone(timedelta(hours=8)) now = datetime.now(app_tz) rounded_now = now.replace(second=0, microsecond=0) if rounded_now < now: rounded_now += timedelta(minutes=1) delay = max(int(os.environ.get("INIT_READY_DELAY_VALUE", "5") or "5"), 1) minimum_lead = max(int(os.environ.get("INIT_READY_MIN_LEAD_VALUE", "2") or "2"), 1) target = rounded_now + timedelta(minutes=delay) if (target - now).total_seconds() < minimum_lead * 60: target = rounded_now + timedelta(minutes=minimum_lead) reserved_slots = { (os.environ.get("MORNING_TIME_VALUE") or "").strip(), (os.environ.get("DISCOVERY_TIME_VALUE") or "").strip(), (os.environ.get("EVENING_TIME_VALUE") or "").strip(), } reserved_slots = {slot for slot in reserved_slots if slot} shifted = 0 while target.strftime("%H:%M") in reserved_slots and shifted < 15: target += timedelta(minutes=1) shifted += 1 cron_expr = f"{target.minute} {target.hour} {target.day} {target.month} *" print(cron_expr) print(target.strftime("%H:%M")) print(str(shifted)) PY } persist_local_config() { local known_json="$1" local config_channel="$2" local config_target="$3" local wecom_user_id="$4" CONFIG_FILE="$CONFIG_FILE" MORNING_TIME="$MORNING_TIME" DISCOVERY_TIME="$DISCOVERY_TIME" EVENING_TIME="$EVENING_TIME" MEMORY_MODE_VALUE="$MEMORY_MODE" CHANNEL_VALUE="$config_channel" TO_VALUE="$config_target" WECOM_USER_ID_VALUE="$wecom_user_id" KNOWN_JSON="$known_json" python3 <<'PY' import json import os from pathlib import Path config_path = Path(os.environ["CONFIG_FILE"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} config["morning_time"] = os.environ["MORNING_TIME"] config["discovery_time"] = os.environ["DISCOVERY_TIME"] config["evening_time"] = os.environ["EVENING_TIME"] config["memory_mode"] = os.environ["MEMORY_MODE_VALUE"] channel = os.environ.get("CHANNEL_VALUE", "").strip() target = os.environ.get("TO_VALUE", "").strip() wecom_user_id = os.environ.get("WECOM_USER_ID_VALUE", "").strip() if channel and target: config["channel"] = channel config["chat_id"] = target if wecom_user_id: config["wecom_user_id"] = wecom_user_id ordered = [] seen = set() def add(ch, peer_id): ch = (ch or "").strip() peer_id = (peer_id or "").strip() if not ch or not peer_id: return key = (ch, peer_id) if key in seen: return seen.add(key) ordered.append({"channel": ch, "peer_id": peer_id}) if channel and target: add(channel, target) for preferred in ( (config.get("binding_channel"), config.get("binding_target")), (config.get("delivery_channel"), config.get("delivery_target")), ): add(*preferred) try: detected = json.loads(os.environ.get("KNOWN_JSON", "[]")) except Exception: detected = [] for item in detected: if isinstance(item, dict): add(item.get("channel"), item.get("peer_id")) for item in config.get("known_channels", []): if isinstance(item, dict): add(item.get("channel"), item.get("peer_id")) config["known_channels"] = ordered config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") print("[✓] .lobster-config 已更新") PY } sync_delivery_contract() { local hint_channel="$1" local hint_target="$2" local contract_json contract_json=$(lobster_sync_delivery_contract "$CONFIG_FILE" "$hint_channel" "$hint_target") mapfile -t CONTRACT_LINES < <(CONTRACT_JSON_VALUE="$contract_json" python3 <<'PY' import json import os contract = json.loads(os.environ["CONTRACT_JSON_VALUE"]) print(contract.get("binding_channel", "")) print(contract.get("binding_target", "")) print(contract.get("binding_mode", "prefer")) print(contract.get("delivery_channel", "")) print(contract.get("delivery_target", "")) print("1" if contract.get("delivery_ready") else "0") print(contract.get("delivery_reason", "")) print(contract.get("delivery_family", "general")) print(contract.get("outbound_adapter", "openclaw")) PY ) BINDING_CHANNEL="${CONTRACT_LINES[0]:-}" BINDING_TARGET="${CONTRACT_LINES[1]:-}" BINDING_MODE="${CONTRACT_LINES[2]:-prefer}" DELIVERY_CHANNEL="${CONTRACT_LINES[3]:-}" DELIVERY_TARGET="${CONTRACT_LINES[4]:-}" DELIVERY_READY="${CONTRACT_LINES[5]:-0}" DELIVERY_REASON="${CONTRACT_LINES[6]:-}" DELIVERY_FAMILY="${CONTRACT_LINES[7]:-general}" OUTBOUND_ADAPTER="${CONTRACT_LINES[8]:-openclaw}" } apply_wecom_direct_delivery_override() { local effective_channel="${BINDING_CHANNEL:-${CHANNEL:-}}" local binding_mode binding_mode=$(lobster_detect_binding_mode "$effective_channel") if [ "$binding_mode" != "strict" ]; then return 0 fi DELIVERY_FAMILY="wecom" OUTBOUND_ADAPTER="wecom-direct-message" DELIVERY_CHANNEL="wecom" if [ -n "$WECOM_USER_ID" ]; then DELIVERY_TARGET="$WECOM_USER_ID" DELIVERY_READY="1" DELIVERY_REASON="" else DELIVERY_TARGET="" DELIVERY_READY="0" DELIVERY_REASON="企业微信定时推送缺少 sender_id / wecom_user_id；请在技能层从 inbound metadata 读取 sender_id 并传给 --wecom-user-id" fi } # ═══════════════════════════════════════════════ # Cron 注册核心函数（v2.5.3 可测版） # ═══════════════════════════════════════════════ # 企业微信通道注册：脚本只输出最终文本，isolated agent 必须使用 message 工具发送 # 同时保留 cron add 的 --channel/--to，便于观察 delivery 状态并保留兜底能力 register_wecom_job() { local name="$1" local cron_expr="$2" local slot="$3" local wecom_target="$4" local extra_args="${5:-}" local agent_message="执行以下命令： bash \"${PUSH_SCRIPT}\" --slot ${slot}${extra_args} --emit-message-text 将命令的 stdout 原文直接作为你唯一的回复输出，不要加任何前缀、解释或额外文字。不要调用任何 message 工具，不要尝试私信任何人。你的回复文本会自动通过 announce 投递到群聊，这就是预期行为。" step "注册 ${name}（企微 message 工具模式）..." local announce_target="${BINDING_TARGET:-$wecom_target}" if openclaw cron add \ --name "$name" \ --cron "$cron_expr" \ --tz "Asia/Shanghai" \ --session isolated \ --channel "openclaw-wecom-bot" \ --to "$announce_target" \ "${OPENCLAW_CRON_ARGS[@]}" \ --message "$agent_message" >/dev/null; then info "${name} 注册成功（企微 message 工具 → ${wecom_target}，announce → ${announce_target}）" else error "${name} 注册失败" fi } # 通用通道注册：脚本自行多通道 fallback 发送 + delivery 兜底 register_general_job() { local name="$1" local cron_expr="$2" local slot="$3" local delivery_channel="$4" local delivery_target="$5" local extra_args="${6:-}" local agent_message="请立即执行以下命令，并只用一句话报告结果： bash \"${PUSH_SCRIPT}\" --slot ${slot}${extra_args}" step "注册 ${name}（通用 fallback 模式）..." local cron_cmd_args=( --name "$name" --cron "$cron_expr" --tz "Asia/Shanghai" --session isolated ) # 有明确的 delivery channel/target 时传给 cron add 做 delivery 兜底 if [ -n "$delivery_channel" ] && [ -n "$delivery_target" ]; then cron_cmd_args+=(--channel "$delivery_channel" --to "$delivery_target") fi cron_cmd_args+=("${OPENCLAW_CRON_ARGS[@]}" --message "$agent_message") if openclaw cron add "${cron_cmd_args[@]}" >/dev/null; then info "${name} 注册成功" else error "${name} 注册失败" fi } register_init_ready_job() { local delay_minutes="$1" local wecom_target="$2" local push_script="$3" local init_ready_suffix="$4" step "注册 init-ready 一次性任务（${delay_minutes} 分钟后触发）..." local trigger_at trigger_at=$(DELAY_MINUTES_V="$delay_minutes" python3 <<'PYTS' from datetime import datetime, timezone, timedelta import os delay = int(os.environ.get("DELAY_MINUTES_V", "3")) t = datetime.now(timezone.utc).replace(microsecond=0) + timedelta(minutes=delay) print(t.isoformat().replace("+00:00", "Z")) PYTS ) [ -n "$trigger_at" ] || { error "init-ready trigger_at 计算失败"; } INIT_READY_EFFECTIVE_TIME=$(TRIGGER_AT_V="$trigger_at" python3 -c " from datetime import datetime, timezone, timedelta t = datetime.fromisoformat('$trigger_at'.replace('Z','+00:00')) print(t.astimezone(timezone(timedelta(hours=8))).strftime('%H:%M')) " 2>/dev/null || echo "$trigger_at") if [ "$OUTBOUND_ADAPTER" = "wecom-direct-message" ] && [ -n "$wecom_target" ]; then # 企微 init-ready：直接让 isolated agent 调用 message 工具，避免再经过 shell CLI local lobster_name lobster_name=$(read_config_value "lobster_name" "虾") local message_for_agent="执行以下命令： bash \"${push_script}\" --slot event${init_ready_suffix} --emit-message-text 将 stdout 原文直接作为你唯一的回复输出，不要加任何前缀、解释或额外文字。不要调用任何 message 工具，不要尝试私信任何人。你的回复文本会自动通过 announce 投递到群聊，这就是预期行为。" local init_ready_announce_target="${BINDING_TARGET:-$wecom_target}" local cron_result rc cron_result=$(openclaw cron add \ --name "lobster-says-init-ready" \ --at "$trigger_at" \ --session isolated \ --message "$message_for_agent" \ --delete-after-run \ --channel "openclaw-wecom-bot" \ --to "$init_ready_announce_target" \ "${OPENCLAW_CRON_ARGS[@]}" 2>&1) rc=$? [ $rc -eq 0 ] || error "init-ready 注册失败: ${cron_result}" info "init-ready 注册成功（at=${trigger_at}, 企微 message 工具 → ${wecom_target}）" else # 通用通道 init-ready：脚本完整模式执行 local init_message="请立即执行以下命令，并只用一句话报告结果： bash \"${push_script}\" --slot event${init_ready_suffix}" local cron_cmd_args=( --name "$INIT_READY_JOB_NAME" --at "$trigger_at" --session isolated --delete-after-run ) if [ -n "$DELIVERY_CHANNEL" ] && [ -n "$DELIVERY_TARGET" ]; then cron_cmd_args+=(--channel "$DELIVERY_CHANNEL" --to "$DELIVERY_TARGET") fi cron_cmd_args+=("${OPENCLAW_CRON_ARGS[@]}" --message "$init_message") if openclaw cron add "${cron_cmd_args[@]}" >/dev/null; then info "init-ready 注册成功（at=${trigger_at}）" else error "init-ready 注册失败" fi fi update_init_ready_schedule_state "scheduled" "$INIT_READY_EFFECTIVE_TIME" "$trigger_at" "" "at=${trigger_at}" } update_init_ready_schedule_state() { local state="$1" local scheduled_for="${2:-}" local verified_next_run="${3:-}" local job_id="${4:-}" local note="${5:-}" CONFIG_PATH="$CONFIG_FILE" INIT_READY_STATE_VALUE="$state" INIT_READY_SCHEDULED_FOR_VALUE="$scheduled_for" INIT_READY_VERIFIED_NEXT_RUN_VALUE="$verified_next_run" INIT_READY_JOB_ID_VALUE="$job_id" INIT_READY_NOTE_VALUE="$note" python3 <<'PY' import json import os from datetime import datetime, timezone from pathlib import Path config_path = Path(os.environ["CONFIG_PATH"]) try: config = json.loads(config_path.read_text(encoding="utf-8")) except Exception: config = {} init_ready = config.get("init_ready") if isinstance(config.get("init_ready"), dict) else {} now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") state = (os.environ.get("INIT_READY_STATE_VALUE") or "scheduled").strip() or "scheduled" scheduled_for = (os.environ.get("INIT_READY_SCHEDULED_FOR_VALUE") or "").strip() verified_next_run = (os.environ.get("INIT_READY_VERIFIED_NEXT_RUN_VALUE") or "").strip() job_id = (os.environ.get("INIT_READY_JOB_ID_VALUE") or "").strip() note = (os.environ.get("INIT_READY_NOTE_VALUE") or "").strip() init_ready["state"] = state init_ready["updated_at"] = now if scheduled_for: init_ready["scheduled_for"] = scheduled_for if verified_next_run: init_ready["last_verified_next_run"] = verified_next_run if job_id: init_ready["job_id"] = job_id if note: init_ready["schedule_note"] = note if state == "scheduled": init_ready.pop("last_error", None) elif note: init_ready["last_error"] = note config["init_ready"] = init_ready config_path.write_text(json.dumps(config, indent=2, ensure_ascii=False), encoding="utf-8") PY } ensure_init_ready_job_registered() { local requested_delay_minutes="$1" local init_ready_suffix="$2" local attempt=0 local effective_delay="$requested_delay_minutes" local job_id next_run status validation validation_status validation_detail schedule cron_expr schedule_time shift_minutes while [ "$attempt" -lt "$INIT_READY_VERIFY_ATTEMPTS" ]; do mapfile -t JOB_SNAPSHOT < <(read_job_snapshot_by_name "$INIT_READY_JOB_NAME") job_id="${JOB_SNAPSHOT[0]:-}" next_run="${JOB_SNAPSHOT[1]:-}" status="${JOB_SNAPSHOT[2]:-}" validation=$(validate_init_ready_next_run "$next_run") validation_status="${validation%%|*}" validation_detail="${validation#*|}" if [ -n "$job_id" ] && [ "$validation_status" = "ok" ]; then update_init_ready_schedule_state "scheduled" "$INIT_READY_EFFECTIVE_TIME" "$validation_detail" "$job_id" "cron_verified:${status:-unknown}" info "init-ready 已确认排入计划：${validation_detail}" return 0 fi attempt=$((attempt + 1)) if [ "$attempt" -ge "$INIT_READY_VERIFY_ATTEMPTS" ]; then break fi warn "init-ready 校验未通过（${validation_detail:-nextRun 为空}），顺延后重排..." effective_delay=$((requested_delay_minutes + attempt * INIT_READY_MIN_LEAD_MINUTES)) remove_job_by_name "$INIT_READY_JOB_NAME" register_init_ready_job "$effective_delay" "${WECOM_USER_ID:-}" "$PUSH_SCRIPT" "$init_ready_suffix" done update_init_ready_schedule_state "schedule_failed" "$INIT_READY_EFFECTIVE_TIME" "" "$job_id" "init-ready nextRun 校验失败" error "init-ready 单次任务校验失败，请稍后重新执行 setup-cron.sh" } register_all_jobs() { local morning_cron discovery_cron evening_cron morning_cron=$(parse_time "$MORNING_TIME") discovery_cron=$(parse_time "$DISCOVERY_TIME") evening_cron=$(parse_time "$EVENING_TIME") INIT_READY_EFFECTIVE_TIME="" INIT_READY_EFFECTIVE_SHIFT="0" echo "" echo "🦞 虾说 — 注册定时推送" echo "" echo " 早安时间: ${MORNING_TIME} (cron: ${morning_cron})" echo " 晚间 roundup: ${DISCOVERY_TIME} (cron: ${discovery_cron})" echo " 晚安时间: ${EVENING_TIME} (cron: ${evening_cron})" echo " delivery family: ${DELIVERY_FAMILY}" echo " delivery adapter: ${OUTBOUND_ADAPTER}" if [ -n "$BINDING_CHANNEL" ] && [ -n "$BINDING_TARGET" ]; then echo " 当前绑定: ${BINDING_CHANNEL} → ${BINDING_TARGET} (${BINDING_MODE})" fi if [ -n "$DELIVERY_CHANNEL" ] && [ -n "$DELIVERY_TARGET" ]; then echo " 主动投递: ${DELIVERY_CHANNEL} → ${DELIVERY_TARGET}" else echo " 主动投递: 将由 adapter=${OUTBOUND_ADAPTER} 决定" fi echo "" step "清理旧的定时任务..." remove_all_managed_jobs local init_ready_suffix=" --job-kind init-ready --managed-job-name ${INIT_READY_JOB_NAME}" if [ "$OUTBOUND_ADAPTER" = "wecom-direct-message" ] && [ -n "$WECOM_USER_ID" ]; then # ───────────────────────────────────────── # 企微通道：agent prompt = --emit-message-text + message 工具 # ───────────────────────────────────────── register_wecom_job "lobster-says-morning" "$morning_cron" "morning" "$WECOM_USER_ID" register_wecom_job "lobster-says-discovery" "$discovery_cron" "discovery" "$WECOM_USER_ID" register_wecom_job "lobster-says-evening" "$evening_cron" "evening" "$WECOM_USER_ID" register_wecom_job "lobster-says-sticker" "30 15 * * 3,6" "sticker" "$WECOM_USER_ID" register_wecom_job "lobster-says-wallpaper" "0 16 * * 0" "wallpaper" "$WECOM_USER_ID" register_init_ready_job "$INIT_READY_DELAY_MINUTES" "$WECOM_USER_ID" "$PUSH_SCRIPT" "$init_ready_suffix" else # ───────────────────────────────────────── # 通用通道（Telegram等）：脚本完整模式 + delivery 兜底 # ───────────────────────────────────────── register_general_job "lobster-says-morning" "$morning_cron" "morning" "$DELIVERY_CHANNEL" "$DELIVERY_TARGET" register_general_job "lobster-says-discovery" "$discovery_cron" "discovery" "$DELIVERY_CHANNEL" "$DELIVERY_TARGET" register_general_job "lobster-says-evening" "$evening_cron" "evening" "$DELIVERY_CHANNEL" "$DELIVERY_TARGET" register_general_job "lobster-says-sticker" "30 15 * * 3,6" "sticker" "$DELIVERY_CHANNEL" "$DELIVERY_TARGET" register_general_job "lobster-says-wallpaper" "0 16 * * 0" "wallpaper" "$DELIVERY_CHANNEL" "$DELIVERY_TARGET" register_init_ready_job "$INIT_READY_DELAY_MINUTES" "" "$PUSH_SCRIPT" "$init_ready_suffix" fi if [ "$MEMORY_MODE" != "lightweight" ]; then local digest_message="请执行以下命令来消化用户的对话记录： bash \"${BASE_DIR}/digest-transcript.sh\" --mode ${MEMORY_MODE} 执行完毕后，简要报告结果。" local digest_cron_args=( --name "lobster-says-digest" --cron "0 3,9,15,21 * * *" --tz "Asia/Shanghai" --session isolated ) if [ -n "$DELIVERY_CHANNEL" ] && [ -n "$DELIVERY_TARGET" ]; then digest_cron_args+=(--channel "$DELIVERY_CHANNEL" --to "$DELIVERY_TARGET") fi digest_cron_args+=("${OPENCLAW_CRON_ARGS[@]}" --message "$digest_message") step "注册 lobster-says-digest..." if openclaw cron add "${digest_cron_args[@]}" >/dev/null; then info "lobster-says-digest 注册成功" else warn "lobster-says-digest 注册失败（不影响主推送）" fi else info "轻量陪伴模式：不注册 transcript digest cron" fi lobster_update_cron_registration "$CONFIG_FILE" "registered" "cron reconcile completed" "$MORNING_TIME" "$DISCOVERY_TIME" "$EVENING_TIME" "$MEMORY_MODE" '["lobster-says-morning","lobster-says-discovery","lobster-says-evening","lobster-says-init-ready","lobster-says-sticker","lobster-says-wallpaper","lobster-says-digest"]' >/dev/null echo "" info "定时推送注册完成！" echo "" echo " 🌅 早安推送: 每天 ${MORNING_TIME}" echo " 📰 晚间 roundup: 每天 ${DISCOVERY_TIME}" echo " 🌙 晚安推送: 每天 ${EVENING_TIME}" if [ "$INIT_READY_EFFECTIVE_SHIFT" -gt 0 ] 2>/dev/null; then echo " 👋 初始化问候: ${INIT_READY_EFFECTIVE_TIME:-${INIT_READY_DELAY_MINUTES}分钟后}（一次性 at 调度）" else echo " 👋 初始化问候: ${INIT_READY_EFFECTIVE_TIME:-${INIT_READY_DELAY_MINUTES}分钟后}（一次性 at 调度）" fi echo " 🎨 表情包: 每周三/六 15:30" echo " 🖼️ 壁纸: 每周日 16:00" if [ "$MEMORY_MODE" = "lightweight" ]; then echo " 🧠 Transcript digest: 已关闭" else echo " 🧠 Transcript digest: 每 6 小时一次（模式: ${MEMORY_MODE}）" fi if [ "$OUTBOUND_ADAPTER" = "wecom-direct-message" ]; then echo "" echo " 📲 企微推送模式: emit-message-text + agent message 工具直达" echo " 📲 所有企微 cron 均带 --channel openclaw-wecom-bot --to ${WECOM_USER_ID} 作为 delivery 兜底" else echo "" echo " 📲 通用推送模式: 脚本多通道 fallback + delivery 兜底" fi } schedule_init_ready_retry_job() { local delay_minutes="$1" if [ "$DELIVERY_READY" != "1" ]; then error "主动推送能力未就绪，无法重排 init-ready：${DELIVERY_REASON}" fi local init_ready_suffix=" --job-kind init-ready --managed-job-name ${INIT_READY_JOB_NAME}" step "重排 init-ready 单次任务..." remove_job_by_name "$INIT_READY_JOB_NAME" register_init_ready_job "$delay_minutes" "${WECOM_USER_ID:-}" "$PUSH_SCRIPT" "$init_ready_suffix" info "init-ready 已重排（${delay_minutes} 分钟后触发）" } if [ -z "$MORNING_TIME" ]; then MORNING_TIME=$(read_config_value "morning_time" "09:00") fi if [ -z "$DISCOVERY_TIME" ]; then DISCOVERY_TIME=$(read_config_value "discovery_time" "21:30") fi if [ -z "$EVENING_TIME" ]; then EVENING_TIME=$(read_config_value "evening_time" "21:00") fi if [ -z "$MEMORY_MODE" ]; then MEMORY_MODE=$(read_config_value "memory_mode" "smart") fi if [ -z "$CHANNEL" ]; then CHANNEL=$(read_config_value "binding_channel" "") fi if [ -z "$TO" ]; then TO=$(read_config_value "binding_target" "") fi if [ -z "$CHANNEL" ]; then CHANNEL=$(read_config_value "channel" "") fi if [ -z "$TO" ]; then TO=$(read_config_value "chat_id" "") fi if [ -z "$WECOM_USER_ID" ]; then WECOM_USER_ID=$(read_config_value "wecom_user_id" "") fi case "$MEMORY_MODE" in lightweight|smart|deep) ;; *) error "memory_mode 必须是 lightweight / smart / deep" ;; esac load_gateway_auth if [ ${#OPENCLAW_CRON_ARGS[@]} -gt 0 ]; then info "检测到 gateway token 环境变量：cron 命令将自动携带 token" fi if [ "$ACTION" = "remove-job-by-name" ]; then step "按名称移除任务..." remove_job_by_name "$REMOVE_JOB_NAME" exit 0 fi step "扫描最近活跃的通道（用于 fallback 与待激活补注册）..." SESSIONS_JSON=$(openclaw sessions --json --active "$ACTIVE_WINDOW_MINUTES" 2>/dev/null || echo "[]") DETECTED=$(SESSIONS_JSON="$SESSIONS_JSON" python3 <<'PY' import json import os try: sessions = json.loads(os.environ.get("SESSIONS_JSON", "[]")) if not isinstance(sessions, list): sessions = [sessions] except Exception: sessions = [] ordered = [] seen = set() def add(channel, peer_id): channel = (channel or "").strip() peer_id = (peer_id or "").strip() if not channel or channel == "unknown" or not peer_id: return key = (channel, peer_id) if key in seen: return seen.add(key) ordered.append({"channel": channel, "peer_id": peer_id}) for session in sessions: direct_channel = session.get("channel") or session.get("platform") or session.get("imChannel") or "" direct_target = session.get("peer_id") or session.get("peerId") or session.get("target") or session.get("chat_id") or session.get("chatId") or "" if direct_channel and direct_target: add(direct_channel, direct_target) continue key = session.get("sessionKey") or session.get("key") or session.get("id") or "" if not key: continue parts = key.split(":") if not parts or parts[0].lower() in ("cron", "hook"): continue if len(parts) <= 3 and parts[-1].lower() == "main": continue if len(parts) >= 5 and parts[0].lower() == "agent": channel = parts[2] marker = parts[3].lower() if marker in ("direct", "dm"): add(channel, parts[4]) best = ordered[0] if ordered else {"channel": "", "peer_id": ""} print(f"{best['channel']}|{best['peer_id']}") print(json.dumps(ordered, ensure_ascii=False)) PY ) DETECTED_TARGET=$(echo "$DETECTED" | head -1) DETECTED_CHANNEL=$(echo "$DETECTED_TARGET" | cut -d'|' -f1) DETECTED_TO=$(echo "$DETECTED_TARGET" | cut -d'|' -f2) DETECTED_KNOWN_JSON=$(echo "$DETECTED" | tail -1) if [ -z "$CHANNEL" ] && [ -n "$DETECTED_CHANNEL" ]; then CHANNEL="$DETECTED_CHANNEL" fi if [ -z "$TO" ] && [ -n "$DETECTED_TO" ]; then TO="$DETECTED_TO" fi step "更新本地配置..." persist_local_config "$DETECTED_KNOWN_JSON" "$CHANNEL" "$TO" "$WECOM_USER_ID" step "同步 delivery contract..." sync_delivery_contract "$CHANNEL" "$TO" apply_wecom_direct_delivery_override info "binding: ${BINDING_CHANNEL:-未锁定} → ${BINDING_TARGET:-未锁定} (${BINDING_MODE})" if [ "$DELIVERY_READY" = "1" ]; then if [ "$OUTBOUND_ADAPTER" = "wecom-direct-message" ]; then info "delivery: wecom → ${DELIVERY_TARGET:-未设置}（adapter=wecom-direct-message，sender_id 直达私聊）" else info "delivery: ${DELIVERY_CHANNEL:-adapter} → ${DELIVERY_TARGET:-target}（adapter=${OUTBOUND_ADAPTER}）" fi else warn "delivery contract 待激活：${DELIVERY_REASON}" fi if [ "$ACTION" = "schedule-init-ready-delay" ]; then schedule_init_ready_retry_job "$SCHEDULE_INIT_READY_DELAY" exit 0 fi if [ "$DELIVERY_READY" != "1" ]; then step "主动推送能力未就绪，暂不注册 cron..." remove_all_managed_jobs lobster_update_cron_registration "$CONFIG_FILE" "pending_activation" "$DELIVERY_REASON" "$MORNING_TIME" "$DISCOVERY_TIME" "$EVENING_TIME" "$MEMORY_MODE" '[]' >/dev/null echo "" warn "已保存待激活的定时推送配置，当前不会误注册 cron。" echo " - delivery family: ${DELIVERY_FAMILY}" echo " - 原因: ${DELIVERY_REASON}" echo " - 一旦补齐主动推送能力，再次执行 setup-cron.sh 即会完成补注册。" exit 0 fi register_all_jobs

Lobster Says

Comments (0)