Bot API 使用文档

https://im.ez7.life/bot-api/<BOT_TOKEN>/<METHOD>

python3 bot.py

curl --fail-with-body --request POST \
  "https://im.ez7.life/bot-api/${EZLIFE_BOT_TOKEN}/getMe" \
  --header 'Content-Type: application/json' --data '{}'

{
  "ok": true,
  "result": {
    "id": "1000000000",
    "is_bot": true,
    "display_name": "Example Bot",
    "username": "example_bot",
    "can_join_groups": true,
    "can_read_all_group_messages": true
  }
}

curl --fail-with-body --request POST \
  "https://im.ez7.life/bot-api/${EZLIFE_BOT_TOKEN}/sendMessage" \
  --header 'Content-Type: application/json' --data-binary @- <<'JSON'
{
  "chat_id": "USER_ID",
  "text": "订单已处理完成。"
}
JSON

{
  "chat_id": "GROUP_ID",
  "chat_type": "group",
  "text": "订单已处理完成。"
}

{
  "ok": true,
  "result": {
    "message_id": "EXAMPLE_MESSAGE_ID",
    "date": 1794297600,
    "chat": {
      "id": "USER_ID",
      "type": "private"
    },
    "text": "订单已处理完成。"
  }
}

{
  "chat_id": "USER_ID",
  "text": "示例系统登录验证码：003812，5分钟内有效。",
  "copy_text": "003812"
}

{
  "chat_id": "USER_ID",
  "text": "请选择需要的服务：",
  "reply_markup": {
    "keyboard": [
      [
        {
          "text": "语言设置"
        },
        {
          "text": "快递追踪"
        }
      ],
      [
        {
          "text": "通知设置"
        },
        {
          "text": "录入单号"
        }
      ]
    ],
    "resize_keyboard": true,
    "one_time_keyboard": false
  }
}

{
  "chat_id": "USER_ID",
  "text": "请选择需要的服务：",
  "reply_markup": {
    "keyboard": [
      [
        "语言设置",
        "快递追踪"
      ]
    ],
    "one_time_keyboard": true
  }
}

{
  "chat_id": "USER_ID",
  "text": "本次操作已结束。",
  "reply_markup": {
    "remove_keyboard": true
  }
}

curl --fail-with-body --request POST \
  "https://im.ez7.life/bot-api/${EZLIFE_BOT_TOKEN}/getUpdates" \
  --header 'Content-Type: application/json' \
  --data '{"offset": 0, "limit": 100}'

{
  "ok": true,
  "result": [
    {
      "update_id": 120,
      "message": {
        "message_id": "EXAMPLE_MESSAGE_ID",
        "date": 1794297600,
        "chat": {
          "id": "1000000001",
          "type": "private"
        },
        "from": {
          "id": "1000000001",
          "display_name": "Example User"
        },
        "text": "语言设置",
        "content_type": 101,
        "raw_content": "{\n  \"content\": \"语言设置\"\n}"
      }
    }
  ]
}

import json
import os
import time
import urllib.error
import urllib.request
from pathlib import Path

TOKEN = os.environ["EZLIFE_BOT_TOKEN"]
BASE = "https://im.ez7.life/bot-api/" + TOKEN
OFFSET_FILE = Path("bot-offset.json")
MENU = [
  [
    "语言设置",
    "快递追踪"
  ],
  [
    "通知设置",
    "录入单号"
  ]
]

class ApiFailure(Exception):
    pass

class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args):
        return None

HTTP = urllib.request.build_opener(NoRedirect())

def api(method, payload):
    req = urllib.request.Request(
        BASE + "/" + method,
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with HTTP.open(req, timeout=20) as response:
            body = json.load(response)
    except urllib.error.HTTPError as error:
        raise ApiFailure("HTTP " + str(error.code)) from None
    except (urllib.error.URLError, TimeoutError, OSError, ValueError):
        raise ApiFailure("Network or response failure") from None
    if not body.get("ok"):
        raise ApiFailure("API error " + str(body.get("error_code", "unknown")))
    return body["result"]

def handle(update):
    message = update.get("message", {})
    chat = message.get("chat", {})
    text = message.get("text", "")
    if not chat.get("id"):
        return
    payload = {"chat_id": str(chat["id"]),
               "chat_type": chat.get("type", "private")}
    if text == "/start" and payload["chat_type"] == "private":
        payload.update(text="请选择需要的服务：", reply_markup={
            "keyboard": MENU, "resize_keyboard": True,
            "one_time_keyboard": False,
        })
    elif text == "/ping":
        payload["text"] = "pong"
    elif text in [label for row in MENU for label in row]:
        payload["text"] = "本次操作已结束。"
    else:
        return
    api("sendMessage", payload)

def save_offset(value):
    temp = OFFSET_FILE.with_suffix(".tmp")
    fd = os.open(temp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as stream:
        json.dump(value, stream)
        stream.flush()
        os.fsync(stream.fileno())
    os.replace(temp, OFFSET_FILE)

def main():
    offset = json.loads(OFFSET_FILE.read_text()) if OFFSET_FILE.exists() else 0
    if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
        raise ValueError("Invalid offset file")
    delay = 1
    while True:
        try:
            updates = api("getUpdates", {"offset": offset, "limit": 100})
            for update in updates:
                handle(update)
                next_offset = int(update["update_id"]) + 1
                save_offset(next_offset)
                offset = next_offset
            delay = 1
            time.sleep(2)
        except (ApiFailure, OSError, ValueError):
            print("Processing failed; offset retained. Check configuration.", flush=True)
            time.sleep(delay)
            delay = min(delay * 2, 30)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass

{
  "ok": false,
  "error_code": 422,
  "description": "chat_id 与 1–5000 字的 text 为必填"
}