From d10444d01f3fff6d9536a8dded7074875bb59b76 Mon Sep 17 00:00:00 2001 From: hp0912 <809211365@qq.com> Date: Sun, 19 Jul 2026 01:06:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=88=9B=E5=BB=BA=E5=AE=9A=E6=97=B6?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- skills/create-scheduled-task/SKILL.md | 102 +++++ .../create-scheduled-task/agents/openai.yaml | 4 + .../scripts/create_scheduled_task.py | 349 ++++++++++++++++++ 3 files changed, 455 insertions(+) create mode 100644 skills/create-scheduled-task/SKILL.md create mode 100644 skills/create-scheduled-task/agents/openai.yaml create mode 100644 skills/create-scheduled-task/scripts/create_scheduled_task.py diff --git a/skills/create-scheduled-task/SKILL.md b/skills/create-scheduled-task/SKILL.md new file mode 100644 index 0000000..acad206 --- /dev/null +++ b/skills/create-scheduled-task/SKILL.md @@ -0,0 +1,102 @@ +--- +name: create-scheduled-task +description: "创建当前微信会话中的提醒或定时任务。当用户说“X 分钟后提醒我 Y”“X 分钟提醒我 Y”“过一会提醒我”“每天/每周/工作日几点提醒我”“帮我设个提醒”，或“创建一个……定时任务”时使用。支持一次性延时、每日、每周和中国法定工作日任务，也支持按时生成 AI 内容。" +--- + +# 创建定时任务 + +通过 `scripts/create_scheduled_task.py` 调用机器人客户端的定时任务接口。自动从当前会话环境变量推导创建人和发送目标，不要求用户提供微信 ID。 + +## 执行流程 + +1. 从用户原话中提取任务名称、执行规则和发送内容。 +2. 缺少执行时间或发送内容时，只追问缺失项，不运行脚本。 +3. 按下表选择调度类型和参数。 +4. 在本 Skill 目录执行 `scripts/create_scheduled_task.py`，且不要在真实请求中传 `--dry-run`。 +5. 仅在脚本返回 `ok: true` 后确认创建成功。回复任务名称、执行规则、下一次执行时间、提醒内容和发送位置；群聊任务同时说明触发时会自动 @ 创建人。 +6. 脚本失败时立即向用户反馈原始错误含义，不得声称任务已创建。 + +## 调度规则 + +| 用户表达 | `--schedule-type` | 必需参数 | +| --- | --- | --- | +| “10 分钟后提醒我喝水”“10 分钟提醒我喝水” | `delay_once` | `--delay-minutes 10` | +| “2 小时后提醒我开会” | `delay_once` | `--delay-hours 2` | +| “明天 08:00 提醒我”且距离现在不超过 24 小时 | `delay_once` | `--run-at 'YYYY-MM-DD HH:mm'` | +| “每天 08:00 提醒我” | `daily` | `--time 08:00` | +| “每周一、周三 09:30 提醒我” | `weekly` | `--time 09:30 --weekdays '[1,3]'` | +| “每个中国法定工作日 09:00 提醒我” | `cn_workday` | `--time 09:00` | + +遵守以下语义： + +- 将星期一到星期日映射为 `1` 到 `7`。 +- 将“工作日”解释为中国法定工作日，包括法定调休补班；用户明确说“周一到周五”时改用 `weekly` 和 `[1,2,3,4,5]`。 +- 将缺少“后”但结构为“X 分钟提醒我 Y”的表达解释为 X 分钟后。 +- 使用 Asia/Shanghai 时区。每日、每周和法定工作日时间必须是零补齐的 `HH:mm`。 +- 中国法定工作日可用年份取决于客户端内置日历。当前配套客户端只包含 2026 年数据；创建 `cn_workday` 任务时提醒用户它在 2027 年前需要随客户端补充日历数据，否则跨年后无法继续计算下一次执行时间。 +- 一次性延时必须在 1 秒到 24 小时之间。超过 24 小时的绝对一次性提醒不受后端支持。 +- 后端不支持每月、每年、原始 Cron 表达式或“每隔 X 分钟”这类间隔循环。遇到这些请求时说明限制，并请用户改成支持的规则，不要伪造近似任务。 +- “创建一个定时任务”但没有给出明确规则或内容时，先追问，不要猜测。 + +## 内容规则 + +- 对普通提醒使用 `--content`。默认把“提醒我喝水”整理为面向用户的 `提醒：喝水`，但用户给出精确文案时保持原文。 +- 对“每天生成一份早报”这类动态内容使用 `--ai-prompt`，不要把普通固定提醒升级为 AI 任务。 +- 可以同时传 `--content` 和 `--ai-prompt`；执行时先发送固定文本，再发送 AI 结果。 +- 至少传 `--content` 或 `--ai-prompt` 之一。 +- 任务名应简短、可识别，不超过 100 个字符；固定文本不超过 500 个字符。 + +## 会话目标 + +脚本自动读取以下环境变量： + +- `ROBOT_FROM_WX_ID`：当前私聊好友 ID 或当前群聊 ID。 +- `ROBOT_SENDER_WX_ID`：当前消息发送人 ID。 +- `ROBOT_WECHAT_CLIENT_PORT`：机器人客户端端口。 + +私聊中把当前好友作为创建人和发送目标。群聊中把当前发言人作为创建人、当前群聊作为发送目标。因此，群聊里的“提醒我”会在当前群聊发送，并在任务触发后发送的第一条文本中自动 @ 创建人，不会私聊群成员。成功回复时明确说明这一点。不要让用户提供或猜测这些 ID。 + +## 脚本参数 + +```text +--name <任务名> 必填 +--schedule-type 必填 +--content <固定提醒文本> 与 --ai-prompt 至少一个 +--ai-prompt <动态内容提示词> 与 --content 至少一个 +--time daily/weekly/cn_workday 必填 +--weekdays weekly 必填，1=周一，7=周日 +--delay-seconds <整数> delay_once 四选一 +--delay-minutes <整数> delay_once 四选一 +--delay-hours <整数> delay_once 四选一 +--run-at delay_once 四选一，须在未来 24 小时内 +--dry-run 仅供开发校验，真实创建时禁止使用 +``` + +## 调用示例 + +10 分钟后提醒： + +```bash +python3 scripts/create_scheduled_task.py --name '喝水提醒' --schedule-type delay_once --delay-minutes 10 --content '提醒：喝水' +``` + +每周一和周三提醒： + +```bash +python3 scripts/create_scheduled_task.py --name '周会提醒' --schedule-type weekly --time 09:30 --weekdays '[1,3]' --content '提醒：参加周会' +``` + +法定工作日生成动态内容： + +```bash +python3 scripts/create_scheduled_task.py --name '工作日早报' --schedule-type cn_workday --time 08:30 --ai-prompt '生成今天的简短早报，直接给出可发送给用户的正文。' +``` + +调用 `execute_skill_script` 时，将以上命令中脚本路径之后的部分放入 `args`。 + +## 成功与失败处理 + +- 成功 JSON 包含 `task.id`、`task.schedule_summary`、`task.next_run_time` 和 `task.target_type`。使用这些实际返回值回复，不要只复述用户输入。 +- 接口业务错误会以“创建定时任务失败：……”返回。说明可操作的原因，例如发送目标不存在、任务数量达到上限、AI 配置缺失或调度器未初始化。 +- 创建接口不是幂等接口。发生超时、连接中断、无效响应，或错误中含“任务已保存，但刷新调度器失败”时，任务可能已经入库；明确告知用户先到定时任务列表核对，禁止自动重试，以免创建重复任务。 +- 普通好友和普通群成员最多创建 5 个任务；群主、群管理员和后台管理员不受此配额限制。 diff --git a/skills/create-scheduled-task/agents/openai.yaml b/skills/create-scheduled-task/agents/openai.yaml new file mode 100644 index 0000000..b911dad --- /dev/null +++ b/skills/create-scheduled-task/agents/openai.yaml @@ -0,0 +1,4 @@ +interface: + display_name: "创建定时任务" + short_description: "根据当前微信会话创建一次性提醒、每日、每周或法定工作日任务" + default_prompt: "使用 $create-scheduled-task 在 10 分钟后提醒我喝水。" diff --git a/skills/create-scheduled-task/scripts/create_scheduled_task.py b/skills/create-scheduled-task/scripts/create_scheduled_task.py new file mode 100644 index 0000000..5a2fa32 --- /dev/null +++ b/skills/create-scheduled-task/scripts/create_scheduled_task.py @@ -0,0 +1,349 @@ +#!/usr/bin/env python3 + +from __future__ import annotations + +import argparse +import json +import math +import os +import re +import socket +import sys +import urllib.error +import urllib.request +from datetime import datetime, timedelta, timezone +from typing import Any, NoReturn + +try: + from zoneinfo import ZoneInfo +except ImportError: # pragma: no cover - Python 3.8 fallback + ZoneInfo = None # type: ignore[assignment,misc] + +sys.stderr = sys.stdout + +SCHEDULE_TYPES = ("delay_once", "daily", "weekly", "cn_workday") +MAX_DELAY_SECONDS = 24 * 60 * 60 + + +class AmbiguousCreateError(RuntimeError): + """The POST may have reached the server, so retrying could create a duplicate.""" + + +class SkillArgumentParser(argparse.ArgumentParser): + def error(self, message: str) -> NoReturn: + raise ValueError(f"参数错误：{message}") + + +def _shanghai_timezone(): + if ZoneInfo is not None: + try: + return ZoneInfo("Asia/Shanghai") + except Exception: + pass + return timezone(timedelta(hours=8), name="Asia/Shanghai") + + +SHANGHAI_TZ = _shanghai_timezone() + + +def _parse_args(argv: list[str]) -> argparse.Namespace: + parser = SkillArgumentParser(description="创建当前微信会话的定时任务") + parser.add_argument("--name", required=True) + parser.add_argument( + "--schedule-type", + "--schedule_type", + dest="schedule_type", + choices=SCHEDULE_TYPES, + required=True, + ) + parser.add_argument("--content", default="") + parser.add_argument("--ai-prompt", "--ai_prompt", dest="ai_prompt", default="") + parser.add_argument("--time", default="") + parser.add_argument("--weekday", action="append", default=[]) + parser.add_argument("--weekdays", action="append", default=[]) + + delay = parser.add_mutually_exclusive_group() + delay.add_argument("--delay-seconds", "--delay_seconds", dest="delay_seconds", type=int) + delay.add_argument("--delay-minutes", "--delay_minutes", dest="delay_minutes", type=int) + delay.add_argument("--delay-hours", "--delay_hours", dest="delay_hours", type=int) + delay.add_argument("--run-at", "--run_at", dest="run_at") + + parser.add_argument("--dry-run", action="store_true") + return parser.parse_args(argv) + + +def _require_env(name: str) -> str: + value = os.environ.get(name, "").strip() + if not value: + raise ValueError(f"环境变量 {name} 未配置") + return value + + +def _conversation_identity() -> tuple[dict[str, str], dict[str, str], str]: + from_wechat_id = _require_env("ROBOT_FROM_WX_ID") + sender_wechat_id = os.environ.get("ROBOT_SENDER_WX_ID", "").strip() + + if from_wechat_id.endswith("@chatroom"): + if not sender_wechat_id: + raise ValueError("当前群聊缺少消息发送人微信 ID") + target = {"wechat_id": from_wechat_id, "type": "chat_room"} + creator = { + "type": "chat_room", + "wechat_id": sender_wechat_id, + "chat_room_id": from_wechat_id, + } + return target, creator, "当前群聊" + + creator_wechat_id = sender_wechat_id or from_wechat_id + target = {"wechat_id": from_wechat_id, "type": "friend"} + creator = {"type": "friend", "wechat_id": creator_wechat_id, "chat_room_id": ""} + return target, creator, "当前私聊" + + +def _validate_clock(value: str) -> str: + if not re.fullmatch(r"(?:[01]\d|2[0-3]):[0-5]\d", value): + raise ValueError("执行时间必须使用 HH:mm 格式，例如 08:30") + return value + + +def _parse_weekdays(values: list[str]) -> list[int]: + parsed: list[int] = [] + for raw in values: + text = raw.strip() + if not text: + continue + if text.startswith("["): + try: + items = json.loads(text) + except json.JSONDecodeError as exc: + raise ValueError(f"weekdays JSON 格式错误: {exc.msg}") from exc + if not isinstance(items, list): + raise ValueError("weekdays 必须是 JSON 数组或逗号分隔列表") + else: + items = [item.strip() for item in text.split(",") if item.strip()] + + for item in items: + if isinstance(item, bool): + raise ValueError("星期必须是 1 到 7 的整数") + try: + weekday = int(item) + except (TypeError, ValueError) as exc: + raise ValueError("星期必须是 1 到 7 的整数") from exc + if weekday < 1 or weekday > 7: + raise ValueError("星期必须在 1 到 7 之间，1 代表周一，7 代表周日") + parsed.append(weekday) + + result = sorted(set(parsed)) + if not result: + raise ValueError("每周任务至少需要一个星期") + return result + + +def _parse_run_at(value: str, now: datetime) -> int: + text = value.strip() + parsed: datetime | None = None + for pattern in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M", "%Y-%m-%dT%H:%M:%S"): + try: + parsed = datetime.strptime(text, pattern).replace(tzinfo=SHANGHAI_TZ) + break + except ValueError: + continue + if parsed is None: + raise ValueError("run-at 必须使用 YYYY-MM-DD HH:mm 或 YYYY-MM-DD HH:mm:ss 格式") + + delay_seconds = math.ceil((parsed - now).total_seconds()) + if delay_seconds < 1: + raise ValueError("一次性任务的执行时间必须晚于当前时间") + if delay_seconds > MAX_DELAY_SECONDS: + raise ValueError("一次性任务最多只能设置到未来 24 小时内") + return delay_seconds + + +def _delay_seconds(args: argparse.Namespace, now: datetime) -> int: + if args.delay_seconds is not None: + seconds = args.delay_seconds + elif args.delay_minutes is not None: + seconds = args.delay_minutes * 60 + elif args.delay_hours is not None: + seconds = args.delay_hours * 60 * 60 + elif args.run_at is not None: + seconds = _parse_run_at(args.run_at, now) + else: + raise ValueError( + "一次性任务必须提供 delay-seconds、delay-minutes、delay-hours 或 run-at" + ) + + if seconds < 1 or seconds > MAX_DELAY_SECONDS: + raise ValueError("一次性任务延时必须在 1 秒到 24 小时之间") + return seconds + + +def _build_schedule_config(args: argparse.Namespace, now: datetime) -> dict[str, Any]: + has_delay = any( + value is not None + for value in (args.delay_seconds, args.delay_minutes, args.delay_hours, args.run_at) + ) + weekday_values = [*args.weekday, *args.weekdays] + + if args.schedule_type == "delay_once": + if args.time or weekday_values: + raise ValueError("一次性任务不能同时设置 time 或 weekdays") + return {"delay_seconds": _delay_seconds(args, now)} + + if has_delay: + raise ValueError(f"{args.schedule_type} 任务不能设置延时参数") + clock = _validate_clock(args.time.strip()) + + if args.schedule_type == "weekly": + return {"time": clock, "weekdays": _parse_weekdays(weekday_values)} + if weekday_values: + raise ValueError(f"{args.schedule_type} 任务不能设置 weekdays") + return {"time": clock} + + +def _build_payload(args: argparse.Namespace) -> tuple[dict[str, Any], str]: + name = args.name.strip() + content = args.content.strip() + ai_prompt = args.ai_prompt.strip() + + if not name: + raise ValueError("任务名称不能为空") + if len(name) > 100: + raise ValueError("任务名称不能超过 100 个字符") + if len(content) > 500: + raise ValueError("固定提醒文本不能超过 500 个字符") + if not content and not ai_prompt: + raise ValueError("content 和 ai-prompt 至少需要提供一项") + + target, creator, target_label = _conversation_identity() + schedule_config = _build_schedule_config(args, datetime.now(SHANGHAI_TZ)) + payload: dict[str, Any] = { + "name": name, + "enabled": True, + "schedule_type": args.schedule_type, + "schedule_config": schedule_config, + "targets": [target], + "fixed_text": content, + "images": [], + "ai_prompt": ai_prompt, + "creator": creator, + } + return payload, target_label + + +def _post_json(url: str, payload: dict[str, Any], timeout: int = 30) -> dict[str, Any]: + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + request = urllib.request.Request( + url, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + response_text = response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as exc: + error_body = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"接口返回 HTTP {exc.code}: {error_body}") from exc + except (urllib.error.URLError, TimeoutError, socket.timeout, ConnectionError) as exc: + raise AmbiguousCreateError( + f"请求定时任务接口失败，结果可能未知：{exc}。请先在定时任务列表核对，勿直接重试" + ) from exc + + if not response_text.strip(): + raise AmbiguousCreateError( + "定时任务接口返回空响应，任务可能已创建。请先在定时任务列表核对，勿直接重试" + ) + try: + result = json.loads(response_text) + except json.JSONDecodeError as exc: + raise AmbiguousCreateError( + "定时任务接口返回了无效 JSON，任务可能已创建。请先在定时任务列表核对，勿直接重试" + ) from exc + if not isinstance(result, dict): + raise AmbiguousCreateError( + "定时任务接口响应不是 JSON 对象，任务可能已创建。请先在定时任务列表核对，勿直接重试" + ) + return result + + +def _unwrap_api_response(response: dict[str, Any]) -> dict[str, Any]: + if response.get("code") != 200: + message = str(response.get("message") or "接口返回未知业务错误") + if "任务已保存，但刷新调度器失败" in message: + raise AmbiguousCreateError(f"{message}。任务可能已经入库，请先核对任务列表，勿直接重试") + raise RuntimeError(message) + + data = response.get("data") + if not isinstance(data, dict): + raise AmbiguousCreateError( + "接口成功响应中缺少任务数据，任务可能已创建。请先在定时任务列表核对，勿直接重试" + ) + return data + + +def _format_next_run(timestamp: Any) -> str | None: + if timestamp is None: + return None + try: + value = int(timestamp) + except (TypeError, ValueError): + return None + return datetime.fromtimestamp(value, SHANGHAI_TZ).strftime("%Y-%m-%d %H:%M:%S Asia/Shanghai") + + +def _success_output(data: dict[str, Any], target_label: str) -> dict[str, Any]: + targets = data.get("targets") + first_target = targets[0] if isinstance(targets, list) and targets else {} + if not isinstance(first_target, dict): + first_target = {} + return { + "ok": True, + "message": "定时任务创建成功", + "task": { + "id": data.get("id"), + "name": data.get("name"), + "schedule_type": data.get("schedule_type"), + "schedule_summary": data.get("schedule_summary"), + "next_run_at": data.get("next_run_at"), + "next_run_time": _format_next_run(data.get("next_run_at")), + "fixed_text": data.get("fixed_text"), + "uses_ai_prompt": bool(data.get("ai_prompt")), + "target_label": target_label, + "target_type": first_target.get("type"), + }, + } + + +def main(argv: list[str]) -> int: + try: + args = _parse_args(argv) + payload, target_label = _build_payload(args) + + if args.dry_run: + print( + json.dumps( + {"ok": True, "dry_run": True, "target_label": target_label, "payload": payload}, + ensure_ascii=False, + indent=2, + ) + ) + return 0 + + client_port = _require_env("ROBOT_WECHAT_CLIENT_PORT") + url = f"http://127.0.0.1:{client_port}/api/v1/robot/scheduled-tasks" + response = _post_json(url, payload) + data = _unwrap_api_response(response) + print(json.dumps(_success_output(data, target_label), ensure_ascii=False, indent=2)) + return 0 + except (ValueError, RuntimeError) as exc: + print(f"创建定时任务失败：{exc}") + return 1 + except Exception as exc: # Keep script failures visible to the agent without hiding them. + print(f"创建定时任务失败：{type(exc).__name__}: {exc}") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:]))