feat: 创建定时任务

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 <delay_once|daily|weekly|cn_workday> 必填
+--content <固定提醒文本>                         与 --ai-prompt 至少一个
+--ai-prompt <动态内容提示词>                     与 --content 至少一个
+--time <HH:mm>                                   daily/weekly/cn_workday 必填
+--weekdays <JSON数组或逗号列表>                  weekly 必填，1=周一，7=周日
+--delay-seconds <整数>                           delay_once 四选一
+--delay-minutes <整数>                           delay_once 四选一
+--delay-hours <整数>                             delay_once 四选一
+--run-at <YYYY-MM-DD HH:mm[:ss]>                 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 个任务；群主、群管理员和后台管理员不受此配额限制。

skills/create-scheduled-task/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "创建定时任务"
+  short_description: "根据当前微信会话创建一次性提醒、每日、每周或法定工作日任务"
+  default_prompt: "使用 $create-scheduled-task 在 10 分钟后提醒我喝水。"

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:]))