快速开始

ambot-script 的词法语法分析遵循了 python 风格，支持与 Python 语法相同的变量类型、变量声明、函数声明、函数调用等功能。

在已经安装了最新版本的 ambot-service 下，可以使用 apy 程序执行 ambot-script 脚本，或直接将一个 .apy 文件赋予可执行权限后执行。

在低版本的 ambot-service 下，不提供 apy 程序，因此请注意保证使用最新版本的 ambot-service（v1.3.8 或更高版本，v1.4.0-rc.16 或更高版本）

Hello World!

那么，来一个最简单的脚本吧！执行以下脚本

print("Hello World!")

image-20220228152642224

输出是最基本的功能了，与 python 类似，ambot-script 的 print 函数也提供了对于 end 和 sep 参数的支持

另外，为了方便，ambot-script 的 print 函数额外支持传递 color 参数，目前支持 RED YELLOW GREEN BLUE BOLD UNDERLINE

image-20220228153946366

执行程序

如果想要执行程序，那么可以使用 exec 或 run 命令，二者的区别在于 exec 会将结果直接输出到屏幕上并在非 0 退出时报出异常退出，而 run 则会将退出码、输出等返回让你自己来决定做什么。

正常的结果

image-20220228155232749

  • exec 和 run 也可以接收一个 string list，但这种语法下不支持 bash 的高级功能（重定向、管道等）
  • exec 在执行失败时会直接抛出异常（ambot-script 的实现没有 try-execpt 的功能，因此异常意味着直接终止程序运行）
  • exec 在执行失败（exit code 不为 0）时也会正常返回 Result 对象，对这一对象进行 assert 会抛出异常

在使用 run 时，应始终配合 assert 以防止意外出现，assert(result) 等同于 assert(result.code==0, result.error == None)

工作目录切换

程序默认的工作目录等同于 run 的目录，在绝大多数命令中如果使用相对路径都是使用工作目录作为主目录的。

ambot-script 提供了 cd 和 pwd内置函数用于设定和获取工作目录，同时提供了 root 预定义变量用于获取程序启动时最初的工作目录

image-20220301102442210

词法与语法

ambot-script 是一种 Python 方言，或者说可以认为它是 Python 的子集。

与 Python 类似，ambot-script 是一种运行时确定类型的动态强类型语言。

与 Python 不同，ambot-script 设计并非为持久性运行的应用，其不支持 Python 的异常、类、上下文、反射等高级功能

数据类型

数据类型说明
NoneType只有固定值 None 是这一类型
bool只有固定值 True, False 是这一类型
int所有的整数类型，与 Python 类似，不区分 signed/unsigned，在脚本引擎内是无上限下限的，但在进行和引擎提供的指令进行交互时上下限被限定为 int64 范围内
float浮点数，双精度
str字符串
bytes二进制序列
list列表
typle元组
dict字典
set集合
function函数

同时，ambot-script 提供一个了 type 指令 type(value: any) → str 用以根据 value 获得其类型名称

变量

ambot-script 中的变量赋值等行为与 Python 保持一致

函数

ambot-script 中的函数调用语句语法与 Python 一致，但某些预定义的函数对于参数的传递有所限制

一个典型的函数定义可能为 a(arg1: string, arg2: int?, arg3:int?=6) → bool

  • 函数（指令）名称为 a
  • 函数接收 3 个参数：第 1 个 string 类型的参数和第 2, 3 个 int 类型的参数
  • 参数 arg2 是可选的
  • 参数 arg3 是可选的，如果省略会默认传递 6
  • 有一个返回值，返回值为 bool

实际的函数还有可变个数参数、动态类型参数等，将在具体用到时在文档中进行说明。

另外，如果声明中没有指定返回值则为返回 None

ambot-script engine 所提供的一些预定义函数为了未来的扩展性等可能禁止了位置参数和命名参数的混用，如果禁止了则会使用文档 https://docs.python.org/3/tutorial/controlflow.html#special-parameters 中的表示进行函数声明的撰写或是明确标注某一参数为「命名参数」或「位置参数」。

如果文档中明确标注一个参数为「位置参数」则在调用函数时不可利用名称指示来更换参数位置，而如果明确标注一个参数为「命名参数」则在使用时不可省略名称调用。

输出

输出用于打印一些内容供脚本使用者查看

输出语句所输出的内容将始终在 stdout，目前暂不提供在 stderr 输出调试信息的能力

预定义变量 RED, YELLOW, GREEN, BLUE, BOLD, UNDERLINE

YELLOW, BOLD, UNDERLINE 在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本引入

用于传递给 print 来控制输出颜色

print(..., /, *, sep=' ', end='\n', color?)

用于输出若干个变量，变量可以传递多个，使用 sep 分隔（默认为一个空格），在所有输出完毕后会输出 end（默认为一个换行）

要输出的信息必须为位置参数，指定的 分隔符/结束符/颜色 必须为命名参数。

颜色信息目前可选值请参考上述预定义变量，如需更多的颜色可以联系开发团队添加

section(section: string, comment: string?)

以一个特定方式打印标题数据，可以快速打印信息，具体效果如图

image-20220301152234023

assert

assert 的目的是对于一些可能异常的值进行判断，如果传递给 assert 的参数不为 True 则会中断整个程序的执行（并可打印错误原因）

assert(..., /, *, info:string?)

assert 函数可以传递任意多个位置参数，所有的参数都必须为真值。

assert 通常用于判断一个表达式是否符合预期或一个错误信息是否为空，如果不符合预期会直接退出当前脚本执行（并汇报一个错误）

可以传递一个 info 命名参数，如果失败时会打印出来，便于展示或定位错误原因

工作目录

工作目录提供了一个脚本内的程序的执行目录，在 run、exec 等绝大多数情况下如果使用相对路径都是使用工作目录作为主目录的

预定义变量

root 脚本启动时的工作目录（类型为 str）

root 保证是一个目录的绝对路径，并且不以 / 结尾

函数

cwd() → str

返回当前运行目录，默认为 root

cd(str)

切换当前的运行目录，这一切换行为会在整个脚本内有效

如果只是想临时切换，可选择先记录再切换回，或是直接在脚本中使用 shell 的 cd

exec("cd /path/to/some && touch something")

命令执行

命令执行提供了在本机执行相关命令的功能（如需在集群中执行命令请参考 命令执行（集群中））

函数原型

和命令执行相关的两个函数为 run 和 exec，二者所接收的参数和返回值都是一样的，仅在行为有所不同。

  • exec 会将命令的执行结果输出到终端，并在出现异常（例如退出码非 0）时中断脚本执行
  • run 不会将命令的执行结果输出到终端，同时在出现异常时也不会中断脚本执行（可利用返回的 result.error 和 result.code 查看）

第一个参数：要执行的命令

run 接收的第一个参数为要执行的命令，这一命令支持传递一个 str list 或是一个字符串。如果传递的是 str list 的话会进行参数 join

命令的行为根据 参数 bash 的行为而有所不同，具体请参考 要执行的命令

参数 bash（默认为 True）

是否使用 bash (login shell)，默认是打开的，在某些异常情况下可以关闭

具体效果请参考 将执行的命令

参数 timeout（默认为 60s）

需要特别注意，所有命令的 timeout 默认为 60s，因此对于一些会长期运行的程序请手动指定一个 timeout

timeout 指定的是超时时间，接收的为一个数字（代表秒数）或一个 time.Duration 对象

如果设定为 0 或负数可以关闭超时功能，但任何情况均不推荐该做法，一个更好的方式是将 timeout 设定为一个预计不可能达到的时间

timeout 参数支持传递 float 或 Duration 对象在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本引入

低版本仅支持传递整数作为超时秒数

参数 env（默认为空）

设定额外使用的环境变量，支持传递 List[str] 或 Dict[str, str] 类型

参数 merge（默认为 True）

是否将 stdout 和 stderr 合并至 stdout，默认打开（打开后 result.stdout 直接包含了 stderr 的内容，而 stderr 不会有返回值）

返回结果

exec 的返回值仅在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本才会返回

返回的类型为 subprocess.Result，其包括

error 错误 str/None

为进程的错误信息，一般只有在要调用的程序不存在时才会出现。在 exec 中只有 None 会被返回（有错误的会导致脚本直接被中断执行）

code 退出码 int

为进程的退出码，0 代表成功。在 exec 中只有 0 会被返回（非 0 退出码会导致脚本直接被中断执行）

stdout 标准输出内容 bytes

当调用参数 merge=False 时为进程的 stdout

当调用参数 merge=True 时（默认）为进程的 stdout+stderr

stderr 标准错误内容 bytes

当调用参数 merge=False 时为进程的 stderr

当调用参数 merge=True 时（默认）为空

高级功能

assert

对于 result 进行 assert 时等价于

assert(result.code==0, result.error == None)

将执行的命令

要执行的命令会受到 bash 参数的影响

  • 对于 bash=False 时
    • 传递的字符串会被 shlex.split 后执行
    • 传递的列表会被直接执行，列表的第一个参数为要执行的程序（如果不为路径则从 PATH 环境变量中搜索执行）
  • 对于 bash=True （默认）时
    • 传递的字符串会作为 ["bash", "-c", "<something>" ] 中的第三个参数
    • 传递的列表会被 shlex.join 后作为上述的第三个参数执行

功能模块

模块 os：系统功能

os 提供了和当前操作系统有关的一些功能，目前只支持环境变量

该模块在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本引入

环境变量

函数

os.getenv(name: str) → <str/None>

获取当前系统中所有的环境变量

os.setenv(name: str, value: str)

设定环境变量

os.listenv() -> [str]

获取当前系统中所有的环境变量

示例

# 假定已经设定了 AA=1, BB=2, CC=3, DD= 几个环境变量
assert(os.getenv("AA") == "1")
assert(os.getenv("BB") == "2")
assert(os.getenv("CC") == "3")
assert(os.getenv("DD") == "")
assert(os.getenv("EE") == None)

os.putenv("BB", "22")
assert(os.getenv("BB") == "22")

def to_dict(l):
    d = {}
    for e in l:
        i = e.find("=")
        if i != -1:
            d[e[:i]] = e[i+1:]
    return d

all = to_dict(os.listenv())
assert(all["AA"] == "1")
assert(all["BB"] == "22")
assert(all["CC"] == "3")
assert(all["DD"] == "")
assert(all.get("EE") == None)

result = run("env | grep ^BB=")
assert(result)
assert(str(result.stdout).strip() == "BB=22")

os.path

os.path.join(...) -> str

将传递的参数进行 path join，返回 join 后的结果

assert(os.path.join("a") == "a")
assert(os.path.join("a", "b") == "a/b")
assert(os.path.join("a", "b/", "c") == "a/b/c")
assert(os.path.join("a", "/b", "c") == "/b/c")

文件功能

os.readfile(filename: str) -> bytes

当前 filename 仅支持绝对路径

从系统中读取文件内容，返回文件内容的 bytes 对象

os.writefile(filename: str, data: bytes, perm: int = 0644)

当前 filename 仅支持绝对路径

将 data 写入文件，如果文件不存在，利用 perm 创建新文件（默认权限 0644）

模块 env：ambot-script engine 环境

env 暴露了 ambot-script engine 的版本信息，未来对于外部传递参数的支持也将在这个模块中实现

预定义变量

env.version: str

返回当前 ambot-script engine 的版本号（没有 v 前缀）

Deprecated

在 ambot-script engine v1.3 系中 v1.3.7 或更低版本、v1.4 系中 v1.4.0-rc.15 或更低版本时，env 模块被用作环境变量的获取与控制，但其实现存在着一些问题，当前已经在 os 模块中重新实现，同时 env.get env.set env.list 方法也被标记为弃用，将在未来版本的 ambot-script engine 中被移除

env.get, env.set, env.list

这三个用于操作环境变量的函数已经被 os 模块 的 环境变量 部分取代，并将在未来的版本中删除，请使用的脚本尽快切换到新的实现

另外，受限于实现，env.set 所做的环境变量操作不会在 run/exec 中生效，而 os.putenv 则无这两个问题

模块 time：时间操作

time 模块提供了时间日期的有关操作，主要为 time.Datetime 与 time.Duration

time.Datetime

提供了日期与时间的相关能力

属性 year, month, day, month, minute, second, nanosecond (int)

返回底层的年月日时分秒和纳秒

属性 timestamp, nanotimestamp (int)

返回 Unix 时间戳

  • timetamp 属性返回 int 类型的秒级时间戳
  • nanotimestamp 属性为 int 类型的纳秒级时间戳

如需获取毫秒级时间戳可利用 .nanotimestamp // 1E6 获得

如需获取小数的秒级时间戳可利用 .nanotimestamp / 1E9 获得

d = time.datetime(2021, 3, 6, 14, 7, 21)
assert(d.year == 2021)
assert(d.month == 3)
assert(d.day == 6)
assert(d.hour == 14)
assert(d.minute == 7)
assert(d.second == 21)
assert(d.timestamp == 1615010841)
assert(d.nanotimestamp == 1615010841000000000)

d += 123 * time.MILLISECOND + 456 * time.MICROSECOND + 789 * time.NANOSECOND
assert(d.timestamp == 1615010841)
assert(d.nanotimestamp == 1615010841123456789)
assert(d.nanotimestamp // 1E6 == 1615010841123)
assert(d.nanotimestamp / 1E9 > 1615010841)
assert(d.nanotimestamp / 1E9 < 1615010842)

format(format: str) → str

进行格式化，format 为 Go 的 datetime 格式化字符串

  • 6 06 2006 代表年
  • 1 01 Jan 代表月
  • 2 02 代表日
  • 3 03 15 代表时，可增加 AM PM 代表上午/下午
  • 4 04 代表分
  • 5 05 代表秒

常见的组合有：

const (
    ANSIC       = "Mon Jan _2 15:04:05 2006"
    UnixDate    = "Mon Jan _2 15:04:05 MST 2006"
    RubyDate    = "Mon Jan 02 15:04:05 -0700 2006"
    RFC822      = "02 Jan 06 15:04 MST"
    RFC822Z     = "02 Jan 06 15:04 -0700" // RFC822 with numeric zone
    RFC850      = "Monday, 02-Jan-06 15:04:05 MST"
    RFC1123     = "Mon, 02 Jan 2006 15:04:05 MST"
    RFC1123Z    = "Mon, 02 Jan 2006 15:04:05 -0700" // RFC1123 with numeric zone
    RFC3339     = "2006-01-02T15:04:05Z07:00"
    RFC3339Nano = "2006-01-02T15:04:05.999999999Z07:00"
    Kitchen     = "3:04PM"
    // Handy time stamps.
    Stamp      = "Jan _2 15:04:05"
    StampMilli = "Jan _2 15:04:05.000"
    StampMicro = "Jan _2 15:04:05.000000"
    StampNano  = "Jan _2 15:04:05.000000000"
)

比较运算（相等、不等、大于、小于、不大于、不小于）

两个 Datetime 之间支持全部六种比较

数值运算（加减）

Datetime 可以和 Duration 进行加减（Datetime 必须为左操作数、Duration 必须为右操作数），结果依然是 Datetime

两个 Datetime 之间可以进行相减，结果为 Duration

base = time.datetime(2001, 1, 1, 10, 0)
cmp = base + 3 * time.DAY - 2 * time.HOUR + 8 * time.MINUTE
assert(cmp > base)
assert(cmp  == time.datetime(2001, 1, 4, 8, 8))
assert(cmp - base == 3 * time.DAY - 2 * time.HOUR + 8 * time.MINUTE)

time.Duration

提供了时长的相关能力

属性 seconds, nanoseconds (int)

均为 int 类型，分别代表了底层有多少秒/纳秒

base = 5 * time.MINUTE + 2 * time.MILLISECOND
assert(base.seconds == 300)
assert(base.seconds == base // time.SECOND)
assert(base.seconds < base / time.SECOND)
assert(base.nanoseconds == base / time.NANOSECOND)
assert(base.nanoseconds == 300002000000)

比较运算（相等、不等、大于、小于、不大于、不小于）

两个 Duration 之间支持全部六种比较

数值运算（取相反数，加减乘除，整除）

Duration 可以取相反数，取反结果依然是 Duration

两个 Duration 之间可以加减，加减结果依然是 Duration

Duration 和 int/float 可以相乘，结果依然是 Duration

Duration（除数）和 int/float（被除数）可以相除，结果依然是 Duration

两个 Duration 之间可以相除，结果是 float

两个 Duration 之间可以相整除，结果是 int

base = 10 * time.SECOND
assert(base == 10 * time.SECOND)
assert(-base == 10 * -time.SECOND)
assert(base * 2 == 20 * time.SECOND)
assert(2 * base == 20 * time.SECOND)
assert(2 * base == 20 * time.SECOND)
assert(base + base == 20 * time.SECOND)
assert(base - 3 * time.SECOND == 7 * time.SECOND)
assert(base - base == 0 * time.SECOND)
assert(base / 2 == 5 * time.SECOND)
assert(base / 3 > 3 * time.SECOND)
assert(base / (3 * time.SECOND) > 3)
assert(base // (3 * time.SECOND) == 3)

预定义变量：NANOSECOND, MICROSECOND, MILLISECOND, SECOND, MINUTE, HOUR, DAY (Duration)

  • NANOSECOND 纳秒
  • MICROSECOND 微秒
  • MILLISECOND 毫秒
  • SECOND 秒
  • MINUTE 分钟
  • HOUR 小时
  • DAY 天

内置函数

time.now() → Datetime

返回当前时间

time.datetime(year, month, day, hour=0, minute=0, second=0) → Datetime

返回指定时间，所使用的时区为东八区

time.fromtimestamp(timestamp: int) → Datetime

根据秒级时间戳获得时间

time.fromnanotimestamp(timestamp: int) → Datetime

根据纳秒级时间戳获得时间

time.strptime(date_string: string, format: string) → Datetime

解析字符串时间，format 请参考上面的 Datetime.format 中的内容

d = time.strptime("2021-03-06 14:07:21", "2006-01-02 15:04:05")
assert(d.year == 2021)
assert(d.month == 3)
assert(d.day == 6)
assert(d.hour == 14)
assert(d.minute == 7)
assert(d.second == 21)
assert(d.timestamp == 1615010841)
assert(d.nanotimestamp == 1615010841000000000)

模块 shlex

模块 shlex 提供了 shell 转义相关的功能

shlex.quote(s: str) -> str

返回 s 经过 shell 转义的字符串 。

返回的字符串可以安全地用作 shell 命令行中的词法单元。

filename = 'somefile; rm -rf ~'

command = 'ls -l {}'.format(shlex.quote(filename))
assert(command == "ls -l 'somefile; rm -rf ~'")

remote_command = 'ssh home {}'.format(shlex.quote(command))
assert(remote_command == "ssh home 'ls -l '\"'\"'somefile; rm -rf ~'\"'\"''")

shlex.split(s: str) -> [str]

用 shell 的语法拆分字符串 s。

assert(shlex.split("echo -n 'Multiple words'") == ['echo', '-n', 'Multiple words'])

shlex.join(split_command: [str]) -> str

将列表 split_command 中的词法单元用 shell 语法串联起来，返回一个字符串。本函数是 .split 的逆运算。

assert(shlex.join(['echo', '-n', 'Multiple words']), "echo -n 'Multiple words'")

模块 json

json 模块提供了 JSON 字符串的数据结构与 ambot-script 内部数据结构之间进行转换的功能

json.dumps(obj: any, /, *, indent=<str/int>) → str

将 obj 对象序列化为字符串

json.loads(s: <str/bytes>) → any

将 s 反序列化为相关数据结构

s 参数支持 bytes 在 v1.3.11 及更高版本引入

模块 math

math 模块暂时不在 Tier 1 支持的范围，如确认需要使用请提前联系开发团队说明，否则未来可能随时有破坏性更新

math 模块支持若干数学函数，包括

模块 http

http 模块还在开发中，如果有没有的功能请告知开发团队并先使用 curl 替代

函数原型

目前提供了 http.get http.post http.put http.delete http.patch 五个方法，这几个可以接收的参数和返回值是完全相同的。

例如对于 get 而言，函数原型是 get(url: str, *, params: {[query_name: str]: query_value: str} = {}, headers: {[header_name: str]: header_value: <str / [str]>} = {}, body: str?, json: any?, form: {[query_name: str]: query_value: str} = {}, auth: (username: str, password: str)?) -> Response

函数参数

参数 url（必填）

要请求的 url，str 类型，例如 https://www.baidu.com

https 请求的证书目前是完全忽略的状态，即类似于 curl 中的 -k 始终被开启

参数 headers

要传递的请求 headers，类型为 dict，dict 的 key 为 request header name，类型为 str；value 的类型可以是 str 或 [str]，即传递 [str] 时会对生成多个同名的 request header

headers 参数为选填

因为类型是 dict，因此顺序是不被保证的，确实有需要按一定的顺序传递 headers 的需求请暂时使用 curl 并告知 ambot-script 开发团队

参数 params

要传递的 url query，类型为 Dict[str, str]，会合并入传入的 url 参数的 query string 部分

因为类型是 dict，因此顺序是不被保证的，确实有需要按一定的顺序传递 params 的需求请暂时使用 curl 并告知 ambot-script 开发团队

参数 auth

如果需要使用 basic 认证，可以传递 auth 参数。auth 参数为一个 Tuple[str, str]，元组的第一个元素为用户名、第二个元素为密码。

auth 参数用于生成 Basic Auth 的 Authorization headers，即例如传递 auth=("admin", "123456")，事实上等同于添加 Request Headers Authorization: Basic YWRtaW46MTIzNDU2

当前 auth 专为 basic auth 实现，因此如果有其他的认证需求请自行通过添加 headers 实现，对于比较常用的认证方式也可以联系 ambot-script 开发团队申请添加入 ambot-script http 标准库中

参数 body

用于传递请求 body，类型为 str，默认不会发送任何 body

body 参数的类型为 str，如果确实需要发送 bytes 数据可以直接考虑能否转换为 str，如果不能请暂时使用 curl 实现并告知 ambot-script 开发团队该场景的信息

参数 json

会对传递的数据进行 JSON encode 后作为 body 传送，理论上支持所有内置的数据类型。同时使用该项会设定 Content-Type 为 application/json

json 参数理论上支持所有内置的数据类型，如果出现问题请暂时手动转换为 dict 并告知 ambot-script 开发团队出现异常的类型信息

参数 form

会对传递的数据进行 percent encode，接收的参数类型是 Dict[str, str]，同时使用该项会设定 Content-Type 为 application/x-www-form-urlencoded

如需利用 form 上传文件，请使用 curl 实现；对于 multiparty/form-data 的实现在计划列表中

返回 响应对象 Response

http 的相关函数返回的都是 http.Response 对象

  • url: str 原始请求 URL
  • status_code: int 返回值
  • headers 响应头
  • body() -> str body 信息
  • json() -> any body 信息，并进行 JSON decode

.url: str

等于 request 时的 url + params 的值

.status_code: int 响应返回值

一般我们认为返回值在 [200, 400) 的请求是成功的

在编写脚本的过程中，调用 request 方法后的第一件事应该就是判断 status_code，可以手动判断也可以利用 assert

.headers: {[header_name: str]: header_values: [str]}

headers 为服务器的响应头，key 为 header name，value 为 header value，注意 value 的类型是一个 list（因为可能存在同一个 headers 有多个返回值的情况）

.body() -> str

body 方法用于获得响应 body，返回值是 str 类型

.json() -> any

body 方法用于获得响应 body，并进行 JSON decode，返回 decode 以后的类型

assert

对于 response 进行 assert 等同于 assert(response.status_code >= 200, esponse.status_code < 400)

虚拟 hostname

虚拟 hostname 的功能的支持在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本添加

如果请求时的 host 域名以 .ambot 结尾，则会进行 ambot 虚拟 hostname 解析，具体规则请参考 core.resolve_domain

模块 re

re 模块提供了正则表达式支持，包括利用正则进行匹配、分隔和替换的功能

re 模块还在开发中，随时可能有破坏性更新，如果需要使用请提前联系 ambot-script 开发团队

re.compile(pattern: str) -> Regex

编译出一个 Regex 对象，可以在后面对该对象调用 search、match、split、findall、sub 方法（参数与下面的五个函数相同）

  • .search(string)
  • .match(string)
  • .split(string, maxsplit=0)
  • .findall(string)
  • .sub(repl, string, count=0)

正则表达式的语法请参考 Golang regexp/syntax 文档

一个可能需要的场景是：想要 Case Insensitive 的匹配，与其他的语言额外指定 flag 不同，你需要在正则表达式最前面写上 (?i) 来开启 flag

re.search(pattern: str, string: str) -> Option[(int, int)]

根据指定的 pattern 在 string 中搜索出现的位置，如果搜索到则返回搜索到的结果（二元组，第一个元素为匹配到的左边界、第二个元素为匹配到的右边界+1），如果未搜索到返回 None

re.search('x', "hello, world!") # 返回 None
re.search('l', "hello, world!") # 返回 (2, 3)

re.match(pattern: str, string: str) -> bool

在 string 的起始处进行搜索 pattern，如果存在返回 True，否则返回 False

这一函数未来可能会改为如果存在返回 Match 对象、不存在返回 None 因此当前情况下建议使用 bool(re.match(...)) 进行调用该函数以避免未来改动造成的脚本异常

re.split(pattern: str, string: str, maxsplit: int = 0) -> Tuple[str]

利用正则做字符串切分，maxsplit 为最大切分次数（0 代表不限制）

re.findall(pattern: str, string: str) -> Tuple[str]

查找所有可能的结果，返回匹配到的元素列表（不会重叠）

re.findall('l+', "hello, world!") # 返回 ("ll", "l")

re.sub(pattern: str, repl: str, string: str, count: int = 0)

利用正则做字符串文本替换，count 为最大替换次数（默认的 0 代表替换全部）

模块 base64

encode(data: str) -> str

将字符串使用 base64 编码

decode(data: str) -> str

解码 base64 字符串

集群功能

集群环境

core.is_master() -> bool

返回当前机器是否有 master 角色，例如简单升级脚本的第一行通常为

assert (core.is_master())

core.resolve

Core Resolve 依赖于 ambot-core 进行虚拟 hostname 的解析，core.resolve 会在调用时向 ambot-core 发起 RPC 调用来获取解析结果。

core.resolve 的函数原型为 core.resolve(targets: <str / [str]>, mode: str = "normal") -> [str] （返回值为解析完成的 hosts 列表）

在 v1.3.11 或更高版本中，core.resolve 可以接受一个 info=True 参数来返回 Server 的详细信息，请参看下面的参数说明

参数 mode：解析模式（默认为 standard）

mode 参数在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本引入 旧版本将自动使用 standard 模式

当前 ambot-core 支持四种解析模式

  • 标准模式 standard
  • 增强模式 enhanced
  • 模糊模式 fuzzy
  • 严格模式 strict
模式描述strict 严格模式standard 标准模式enhanced 增强模式fuzzy 模糊模式
self本机✓✓✓✓
all集群中所有机器✓✓✓✓
{customName}外部传递想要使用的值✓✓✓✓
+service集群中所有包含指定服务机器✓✓✓✓
.role集群中所有包含指定角色机器✓✓✓✓
:pkg-id集群中所有包含指定 package（利用 id 查询）机器✓✓✓✓
:pkg-name集群中所有包含指定 package（利用 name 查询）机器✓✓✓✓
<ip-in-cluster>集群中存在指定 IP 的机器，返回✓✓✓
<hostname-in-cluster>集群中存在指定 hostname 的机器，返回✓✓✓
service集群中所有包含指定角色机器✓✓
role集群中所有包含指定角色机器✓✓
pkg-name集群中所有包含指定 package（利用 name 查询）机器✓✓
pkg-id集群中所有包含指定 package（利用 id 查询）机器✓✓
IP 后缀匹配集群中所有 IP 以指定 '.后缀' 结尾的机器✓
机器 hostname 前缀匹配✓
机器 hostname 前缀匹配✓
纯数字修改当前机器 IP 中 D 段✓
ip返回相应 IP✓✓✓
hostname返回相应 hostname✓✓✓

IP 后缀匹配在 ambot-script engine v1.3.22 引入

在 ambot-script engine v1.3.21 版本下纯数字优先级在机器 hostname 前缀匹配之前；
在 ambot-script engine v1.3.22 版本起纯数字优先级在机器 hostname 前缀匹配之后

第一个参数：目标机器 <str / List[str]>

用于指定想要解析的 hostname / ip / 虚拟hostname。返回值将返回分别解析的结果合并去重后的结果

扩展模式

对于以上的所有模式，单个虚拟 hostname 后可以加 #1 来代表只取第一个返回结果。因为 core.resolve 的返回值始终是排序的，因此对于相同的输入， #1 的结果是稳定的

{customName} 格式

当前仅供升级系统使用，利用 Dynamic Script 可以设定自定义的 key -> hosts 映射

allow_not_exist=True

默认情况下，如果引用了内部资源（例如查询一个 package），内部资源不存在会报错导致整个程序的中断。可以使用 allow_not_exist 来让程序返回空而不是中断运行。

allow_not_exist=True 的参数在 ambot-script-engine v1.3.11 及更高版本可用

info=True

默认情况下，core.resolve 返回的是解析完成的 hostname 列表

如果传递了 info=True，则返回的是 ServerInfo 列表

info=True 的参数在 ambot-script-engine v1.3.11 及更高版本可用

core.resolve_domain

core.resolve_domain 仅用于解析 Ambot 虚拟 Domain，不能用来进行标准的 DNS 解析。对于其他域名（任何不以 .ambot 结尾的域名）均会直接导致脚本异常结束

ambot virtual domain 采用 {参数}.{命名空间}.ambot 的格式

函数签名：core.resolve_domain(domain: str) -> List[str]，即接收一个虚拟域名，返回域名对应的 IP 列表（可能为零个、一个或多个）

命名空间

命名空间是用于确认域名归属的，目前支持以下命名空间

  • enhanced：使用 Core.Resolve 的增强模式进行解析
  • fuzzy：使用 Core.Resolve 的模糊模式进行解析
  • service：解析某一特定 service
  • role：解析某一特定角色
  • pkg 或 package：解析某一特定 Package（支持使用 package id 或 package name）

命名空间支持「修饰符」，主要是针对多个返回值的情况的，修饰符的格式为将 {命名空间} 替换为 {修饰符}-{命名空间}。目前支持两个命名空间

  • first 用于只取第一个结果（这一结果是稳定的，排序等同于 Core Resolve）
  • any 随机返回一个结果（任意两次调用结果可能相同可能不同）

命名空间参数

不同的命名空间有着不同的参数

enhanced, fuzzy 可接受的参数

去除掉命名空间和 .ambot 后缀的全部内容均是 enhanced 和 fuzzy 命名空间可接受的参数。但需要注意的是，因为域名的特殊性，+service :pkg-id .role 这种用法是无法实现的（因此并未支持标准模式），如需使用请使用其他命名空间。

几个例子

self.enhanced.ambot           # 本机
master.enhanced.ambot         # master 机器
sensor.first-enhanced.ambot     # 第一台 Sensor
kafka.any-enhanced.ambot      # 随机一台安装了kafka的机器

service 可接受的参数

Service 名称

即目前 Service 用法为 {Service 名称}.service.ambot

例如

nta-light-api.service.ambot   # 安装了 nta-light-api service 的机器
agent.service.ambot           # 安装了 agent service 的全部

role 可接受的参数

角色名称

即目前 Role 用法为 {角色名称}.role.ambot

另外，对于角色的场景，是可以省略 role 的，即直接使用 {角色名称}.ambot 也是可以的

例如

master.ambot
sensor.ambot
sensor.role.ambot  
master.role.ambot

pkg/package 可接受的参数

Package ID 或 Package Name

例如

matrix.pkg.ambot
matrix.package.ambot
com.topahnt.matrix.pkg.ambot 
com.topahnt.matrix.package.ambot

命令执行（集群中）

命令执行提供了在集群执行相关命令的功能（如需在当前执行命令请参考 命令执行，或命令中的 on 参数传递 self）

函数原型

和集群命令执行相关的两个函数为 run 和 exec，二者所接收的参数和返回值都是一样的，仅在行为有所不同。

可以发现，这个名称和非集群中的 命令执行 是一致的，事实上，二者的区别主要在于是否传递了 on 参数

  • exec 会将命令的执行结果输出到终端，并在出现异常（例如退出码非 0）时中断脚本执行
  • run 不会将命令的执行结果输出到终端，同时在出现异常时也不会中断脚本执行（可利用返回的 result.error 和 result.code 查看）

参数 on：要执行的目标机器（必填）

on 参数可以传递一个 str 或 List[str] 类型的参数，用于指定在若干个机器上同时执行。指定列表时将合并列表中的机器列表并去重后执行。

on 指定的机器列表请参考 core.resolve，这里使用的模式是标准模式（standard）

第一个参数：要执行的命令

run 接收的第一个参数为要执行的命令，这一命令支持传递一个 str list 或是一个字符串。如果传递的是 str list 的话会进行参数 join

命令的行为根据 参数 bash 的行为而有所不同，具体请参考 要执行的命令

参数 bash（默认为 True）

是否使用 bash (login shell)，默认是打开的，在某些异常情况下可以关闭

具体效果请参考 将执行的命令

参数 env（默认为空）

设定额外使用的环境变量，支持传递 List[str] 或 Dict[str, str] 类型

参数 realtime（默认为 False）

该参数仅在 exec 中有效

因 ambot-run 的 -o 参数为 beta 版本，因此本参数也可能在未来被取消，请尽可能避免依赖该参数的逻辑

是否开启运行结果的实时输出，等价于执行 ambot-run 时传递了 -o / --output 参数

返回结果相关的数据类型

exec 的返回值仅在 ambot-script engine v1.3 系中 v1.3.8 或更高版本、v1.4 系中 v1.4.0-rc.16 或更高版本才会返回

rexec.Results

Results 是一个 rexec.Server → rexec.Result 的 dict

迭代 iterate

可以对 Results 进行 for 循环，循环时取到的是 server 列表（该循环无序，且不保证稳定）

.items() →[(Server, Result)]

返回一个 (Server, Result) 的二元组，用于在 for 循环中同时得到 server 和 result 使用

下标取值 [<str/Server>] → <Result/None>

利用下标取值时可以利用 Server 对象（通常在 for 循环中使用）也可以利用 hostname / ip 字符串取值。

如果存在返回 Result 对象，如果不存在返回 None

assert

等同于对所有的 Result 进行 assert

服务器信息 rexec.Server

命令执行时所在的目标机器

属性数据类型说明
hoststr执行目标的 hostname
ipstr执行目标的 ip 地址
isLocalbool执行目标是否为本机

.one(*, allow_no=False, allow_multiple=False) -> Option[Result]

**如果 allow_no 和 allow_multiple 均为 False ** （默认）

如果这次 run/exec 调用只在一台机器上执行了，返回这个执行结果 如果不在任何一台机器上执行或是在多于一台的机器上执行了，报错退出

如果 allow_no=True

如果不在任何一台机器上执行，返回 None

如果 allow_multiple=True

如果在多于一台的机器上执行了，随机返回一个执行结果

返回结果 rexec.Result

属性数据类型说明
codeint退出码
outputbytes输出内容（包括 stdout 和 stderr，目前只支持合并输出）
error<str/None>错误信息（非 0 退出不认为是错误，通常的错误可能为命令不存在或超时等）

assert

当无错误且返回值为 0 时成功

等同于 assert(result.error == None, result.code == 0)

.server -> rexec.Server

返回产生这一结果的 Server

.outputString -> str

等同于 .output，但返回的数据类型是 str

示例

results = run("echo 123; echo 123", on=['prs-master', 'prs-sensor'])

for i, x in enumerate(results, 1):
    print("[Server {}] host={}, ip={}, isLocal={}".format(i, x.host, x.ip, x.isLocal))
    # print("    -> Exit with", results[x].code)

for server, result in results.items():
    print("[{}] (Exit with {})\n{}".format(server.host, result.code, string(result.output)))

assert(results)

高级功能

将执行的命令

要执行的命令会受到 bash 参数的影响

  • 对于 bash=False 时
    • 传递的字符串会被 shlex.split 后执行
    • 传递的列表会被直接执行，列表的第一个参数为要执行的程序（如果不为路径则从 PATH 环境变量中搜索执行）
  • 对于 bash=True （默认）时
    • 传递的字符串会作为 ["bash", "-c", "<something>" ] 中的第三个参数
    • 传递的列表会被 shlex.join 后作为上述的第三个参数执行

上传文件

ambot-script 提供 upload 命令来上传文件到远程机器

upload(local: str, *remote: str, allow_error: bool = false) -> UploadResult

参数

local 本地路径

*remote 目标机器 + 目标路径

remote 为可变参数，支持传递多个以在一台机器上传多份或同时在多台机器上传

remote 的格式为 目标机器:目标路径，目标机器支持的语法与 ambot-core 相同；目标路径仅支持绝对路径

具体使用请查看示例

allow_error 允许错误（默认为 False）

默认情况下当有任何一台机器上传失败时将会终止脚本执行，可以传递这一参数以自行决定是否要退出

返回值 UploadResult

upload 的结果，为一个 Dict[str, bool]

支持 for in 或是 result['xxx'], 循环时的 key 为 host:path 字符串，value 为结果布尔值

这里的 host 是被解析后的结果，而不是传入函数参数的

示例

简单上传到某一远程机器

upload("/etc/hosts", ".sensor:/tmp/a")

传到多台机器

upload("/etc/hosts", ".master:/tmp/a", ".sensor:/tmp/b")

上传到多台机器

upload("/etc/hosts", "(.master,.sensor):/tmp/a")

允许错误

result = upload("/etc/hosts", "self:/tmp/a", "self:/tmp/not-exist-dir/a", allow_error=True)
for desc, success in result:
    if success:
        print("Upload to", desc, "success")
    else:
        print("Upload to", desc, "fail")

assert(result) # 如果任一执行失败，assert 会终止程序执行

allow_error 对 host 解析失败无效

upload("/etc/hosts", ".not-exist:/tmp/a", allow_error=True)
print("Unreachable")

core.info

core.info 提供了一些内部的信息

core.info 模块从 v1.3.11 起开始支持

servers() -> List[ServerInfo] 返回当前集群中的机器列表

ServerInfo 包括如下字段：

  • host: str 机器的 hostname
  • ip: str 机器的 IP 地址
  • isLocal, is_local: bool 该机器是否为本机

core.info.all

core.all 是 UnDocumented 的 API，不保证长久可用性、无稳定性保证，请仅用于临时脚本测试。相关可获取到的内容请参考 help。

如果发现某一信息只能通过 all 获取，请发稳定版本的 Feature Request

注：因历史原因，使用 'pkg-id' in core.info.all().packageMap 判断 package 是否存在将保证长期可用

core.vars

core.vars 提供了和 vars 有关的接口

WIP：至少需要 v1.3.11 以上版本才可使用，仍在开发中

render(template_file: str, use: [str], save_to: str?, extra: dict = {}) -> str

利用 vars 渲染模板，返回渲染结果

参数：template_file

模板文件名，支持绝对路径或相对于运行目录的相对路径

参数：use

使用 vars 的限定，必须存在，如不使用任何 vars（即完全使用 extra）请传递 []

参数：extra

除了 vars 之外使用的其他变量，会遵循 vars 导入原则进行覆盖原有的 vars

vars 导入原则 的文档还在撰写中，简单说就是 例如存在着已有的 vars {"k1": {"a": 1, "b": 2}}

如果 extra = {"k1": {"a": 3}} ，则结果为 {"k1": {"a": 3}}，即完全覆盖 如果 extra = {"k1.a": 3} ，则结果为 {"k1": {"a": 3, "b": 2}}，即部分设定

参数：save_to

除了返回渲染结果，还将渲染结果保存到文件中，传递文件名绝对路径或相对于运行目录的相对路径

core.systemd

core.systemd 提供了和 systemd 有关的接口

all(raw=False) -> Dict[key = str, value = SystemdServiceInfo]

core.systemd.all() 会返回当前系统中运行的（由 ambot-service 管理的）全部服务信息。输出是一个 dict，字典的 key 为 service name，value 为这个 service 的详细信息（SystemdServiceInfo 类型）

all 可以接受一个 raw 参数（默认为 False），如果 raw=True 将返回在 systemd 中和这一 service 有关的所有 property，如果 raw=False（默认）则返回的 SystemdServiceInfo 对象中的 properties 字段会为空字典

SystemdServiceInfo

name: str

服务名称

is_enable: bool

服务状态是否 enable（受限于 systemd 的实现，这个值并不一定准，请不要依赖这个属性来做核心逻辑）

state: ENUM{"unknown" | "running" | "stopped" | "stopping" | "starting" | "failed"}

  • starting 正在启动
  • running 正在运行
  • stopping 正在停止
  • stopped 已经停止
  • failed 失败
  • unknown 未知

is_running: bool

服务是否在运行中，即上面的 state 是否为 running

is_started: bool

服务是否已经启动，即上面的 stagte 是否为 starting 或 running

active_time: time.Datetime

服务最近一次的启动时间

inactive_time: time.Datetime

服务最近一次的停止时间

need_daemon_reload: bool

在服务已经启动后是否修改了服务的配置文件且并未执行过 ambot-service daemon-reload

properties: Dict[str, any]

服务的 properties 信息，key 为 property 名称，value 类型与 key 相关（如使用请自行测试）

真实用例

实例 1

# 在与当前脚本同级的目录下有 com.tophant.prs.nta-light-api.ambotup 文件

resources_dir = '/data/.ambot/4.8-upgrade/code-upgrade/'
exec('mkdir -p ' + resources_dir, on='+nta-light-api')
upload("com.tophant.prs.nta-light-api.ambotup", "+nta-light-api:" + resources_dir)
exec("chmod +x " + resources_dir + "/com.tophant.prs.nta-light-api.ambotup", on="+nta-light-api")
exec(resources_dir + "/com.tophant.prs.nta-light-api.ambotup -y --author 'Jenkins <ci@tophant.com>' --unsafe-allow-dirty --start nta-light-api", on="+nta-light-api")

这个脚本做了以下几件事情

  • 声明 resources_dir 变量，为存储的资源路径
  • 在存在「nta-light-api」服务的机器上创建了这个资源路径
  • 将 up 升级包传到了存在「nta-light-api」服务的机器上的资源路径下
  • 给这个升级包赋予了可执行权限
  • 执行这个升级包

在这个例子中，ambot-script 提供了以下能力

  • 上传文件到符合条件的机器上
  • 在符合条件的机器上执行命令

建议改写

resources_dir = '/data/.ambot/4.8-upgrade/code-upgrade/'
pkg_id = 'com.tophant.prs.nta-light-api'
services = ['nta-light-api']

# ---------- #

assert(core.is_master())

target = ":" + pkg_id
pkg_file = pkg_id + ".ambotup"
pkg_upload_to = os.path.join(resources_dir, pkg_file)

exec(['mkdir', '-p', resources_dir], on=target)
upload(pkg_file, "{}:{}".format(target, resources_dir))
exec(['chmod', '+x', pkg_upload_to], on=target)
exec([pkg_upload_to, '-y', '--author', 'Jenkins <ci@tophant.com>', '--start', ','.join(services)], on=target)
  1. 在升级类脚本的第一行加上 assert(core.is_master()) 是一个好习惯
  2. exec 执行的命令在不使用 bash 高级功能时优先使用 list 传递而非拼接字符串
  3. 路径的拼接有问题（resources_dir 已经以 / 结尾了）
  4. 自动化脚本中禁止使用任何 --unsafe 的功能
  5. 将公有参数提出来作为变量，方便脚本复用

进一步的优化空间？

（下一场培训）

实例 2

allServers=core.resolve('all')
centos8Servers=core.resolve('.sensor')
centos7Servers = []

for s in allServers:
    if s not in centos8Servers:
        centos7Servers.append(s)

run("mkdir -p /data/update-polkit/", on="all")

upload("polkit-el7", "(%s):/data/update-polkit/" % ",".join(centos7Servers))
upload("polkit-el8", "(%s):/data/update-polkit/" % ",".join(centos8Servers))

run("rpm -Uvh /data/update-polkit/polkit-el7/*.rpm", on=centos7Servers)
run("rpm -Uvh /data/update-polkit/polkit-el8/*.rpm", on=centos8Servers)

ret = run("ambot-run -- rpm -qa polkit")

print("")
print("========== after update", color=GREEN)
print(ret.stdout)

这个脚本用到了：

  1. core.resolve
  2. upload 支持传递多个目标机器

这个脚本的问题

  1. run 命令需要手动判断返回值，请换成 exec
  2. upload 命令和 run 命令所传递的目标参数有可能为空
  3. 不要套娃 ambot-run

改写

allServers = core.resolve('all')
centos8Servers = core.resolve('.sensor')
centos7Servers = [x for x in allServers if x not in centos8Servers]

assert(run("mkdir -p /data/update-polkit/", on="all"))

if centos8Servers:
	upload("polkit-el8", "(%s):/data/update-polkit/" % ",".join(centos8Servers))
	assert(run("rpm -Uvh /data/update-polkit/polkit-el8/*.rpm", on=centos8Servers))

if centos7Servers:
	upload("polkit-el7", "(%s):/data/update-polkit/" % ",".join(centos7Servers))
	assert(run("rpm -Uvh /data/update-polkit/polkit-el7/*.rpm", on=centos7Servers))


print()
print("========== after update", color=GREEN)

exec("rpm -qa polkit", on='all')

问题反馈与建议

敬请移步（内网访问）