HHY 参考

标准库函数索引

运行时 Registry 中全部 128 个 V1.7.1 核心 callable 的签名与用途。

如何阅读签名

本页以 V1.7.1 Runtime 的 Callable Contract Registry 为权威来源，共 128 个核心 callable；扩展动态注册的 callable 在各扩展文档中说明。T/U 表示泛型占位值，? 表示可选参数或可空结果，Map? 表示可选 options Map。所有函数都可普通调用；在管道中，左侧值会注入为第一个参数。

核心值、集合、环境与控制（22）

函数print#

print(Value...) -> Null

把值写到标准输出；传入 Stream 时逐项输出并消费它。

函数print_error#

print_error(Value...) -> Null

把值写到标准错误；适合诊断信息。

函数exit#

exit(Int?) -> Never

立即以给定状态码结束脚本，省略时使用 0，并触发资源清理。

函数length#

length(String | List | Map) -> Int

返回 String 的 code point 数或 List/Map 的元素数；Stream 应使用 count。

函数byte_length#

byte_length(String | BytesBuffer) -> Int

返回 String 的 UTF-8 字节数或 BytesBuffer 大小。

函数type#

type(Value) -> String

返回值的逻辑类型名。

函数is_type#

is_type(Value, String) -> Bool

判断值是否具有指定逻辑类型，返回 Bool。

函数to_int#

to_int(Int | Float | String) -> Int

把 Int/Float/String 显式转换为 Int，失败或溢出产生 ValueError。

函数to_float#

to_float(Int | Float | String) -> Float

把 Int/Float/String 显式转换为 Float，失败产生 ValueError。

函数get#

get(List | Map | Record, Int | String) -> Value | Null

安全读取 List 索引、Map 键或对象字段；缺失返回 null。

函数require#

require(Map, String) -> Value

读取必需 Map 键；键缺失产生 KeyError，存在且为 null 时返回 null。

函数pick#

pick(Map, List<String>) -> Map

返回只保留指定键的新 Map，并保留存在的 null 字段。

函数put#

put(Map, String, Value) -> Map

返回新增或替换一个键的新 Map，不修改原 Map。

函数remove_key#

remove_key(Map, String) -> Map

返回移除指定键的新 Map。

函数append#

append(List<T>, T) -> List<T>

返回末尾增加一个元素的新 List。

函数remove_at#

remove_at(List<T>, Int) -> List<T>

返回移除指定索引的新 List；越界产生 IndexError。

函数now#

now() -> DateTime

返回带时区的当前 DateTime。

函数datetime.parse#

datetime.parse(String, String, String) -> DateTime

按明确的格式和时区解析 DateTime，非法输入产生 ValueError。

函数require_env#

require_env(String) -> String

读取必需环境变量；不存在时产生 KeyError。

函数sleep#

sleep(Duration) -> Null

可取消地等待指定 Duration。

函数cancel#

cancel() -> Never

触发当前执行的根取消令牌并开始统一清理。

函数throw#

throw(Error) -> Never

抛出 Error，并沿调用栈或 Flow 传播。

Flow 与 Stream（25）

map/where/take 等转换保持惰性；collect、count、reduce 等终端操作消费 Stream；sort_by 与 group_by 会在资源上限内物化输入。parallel 使用有界隔离 worker 并保持输出顺序。

函数stream#

stream(List<T> | Map | Range) -> Stream<T>

把 List、Map entries 或 Range 转成惰性单次消费 Stream。

函数range#

range(Int, Int) -> Stream<Int>

创建从 start 到 end（不含 end）的 Int Stream。

函数map#

map(Stream<T>, Function(T -> U)) -> Stream<U>

惰性地对每项调用闭包，一项输入对应一项输出，不自动展开。

函数flat_map#

flat_map(Stream<T>, Function(T -> Stream<U>)) -> Stream<U>

对每项返回一个子 Stream，并惰性地把子流依次展开。

函数where#

where(Stream<T>, Function(T -> Bool)) -> Stream<T>

惰性保留闭包返回 true 的项目；闭包必须返回 Bool。

函数take#

take(Stream<T>, Int) -> Stream<T>

惰性保留前 n 项，达到数量后提前关闭上游。

函数skip#

skip(Stream<T>, Int) -> Stream<T>

惰性丢弃前 n 项，然后传递其余项目。

函数inspect#

inspect(Stream<T>, Function(T -> Value)) -> Stream<T>

为每项执行观察闭包，再原样传递项目。

函数distinct#

distinct(Stream<Hashable>) -> Stream<Hashable>

惰性去除重复的可 Hash 标量，并保存已见集合。

函数sort_by#

sort_by(Stream<T>, Map, Function(T -> Comparable)) -> Stream<T>

物化有限输入，按闭包 key 和 asc/desc 选项稳定排序。

函数group_by#

group_by(Stream<T>, Function(T -> Hashable)) -> Stream<Group<T>>

物化有限输入并按 Hash key 输出 Group；Group 含 key 与 values。

函数debounce#

debounce(Stream<T>, Duration) -> Stream<T>

在指定 Duration 内合并快速连续事件，常用于 watch。

函数on_error#

on_error(Stream<T>, Function(Error -> Stream<T>)) -> Stream<T>

当 Stream 失败时调用闭包，用返回的 Stream 恢复或替换后续输出。

函数parallel#

parallel(Stream<T>, Int, Function(T -> U)) -> Stream<U>

用最多 n 个隔离 worker 并发处理，保序、有界缓冲且 fail-fast。

函数collect#

collect(Stream<T>) -> List<T>

消费有限 Stream 并物化为 List。

函数count#

count(Stream<T>) -> Int

消费 Stream 并返回项目数。

函数first#

first(Stream<T>) -> T | Null

返回第一项或 null，并提前关闭上游。

函数last#

last(Stream<T>) -> T | Null

消费 Stream 并返回最后一项或 null。

函数min#

min(Stream<Number>) -> Number | Null

消费数值 Stream，返回最小值或空流的 null。

函数max#

max(Stream<Number>) -> Number | Null

消费数值 Stream，返回最大值或空流的 null。

函数sum#

sum(Stream<Number>) -> Number

消费数值 Stream 并求和，遵守 Int 溢出规则。

函数reduce#

reduce(Stream<T>, U, Function(State<T,U> -> U)) -> U

以 initial 累积 Stream；闭包接收含 acc/item/index 的 state。

函数any#

any(Stream<T>, Function(T -> Bool)) -> Bool

任一项目满足谓词即返回 true，并短路关闭上游。

函数all#

all(Stream<T>, Function(T -> Bool)) -> Bool

所有项目满足谓词才返回 true；首个 false 时短路。

函数for_each#

for_each(Stream<T>, Function(T -> Value)) -> Null

消费 Stream 并为每项执行闭包，返回 null。

文本、Regex、JSON 与 CSV（18）

函数contains#

contains(String | List, Value) -> Bool

判断 String 是否含子串，或 List 是否含相等值。

函数upper#

upper(String) -> String

返回 Unicode 大写转换后的新 String。

函数lower#

lower(String) -> String

返回 Unicode 小写转换后的新 String。

函数trim#

trim(String) -> String

移除 String 两端空白。

函数trim_start#

trim_start(String) -> String

移除 String 开头空白。

函数trim_end#

trim_end(String) -> String

移除 String 末尾空白。

函数starts_with#

starts_with(String, String) -> Bool

判断 String 是否以指定文本开头。

函数ends_with#

ends_with(String, String) -> Bool

判断 String 是否以指定文本结尾。

函数replace#

replace(String, String, String) -> String

返回把匹配文本替换后的新 String。

函数split#

split(String, String) -> List<String>

按分隔文本把 String 分割成 List<String>。

函数join#

join(List<String>, String) -> String

用分隔文本连接 List<String>。

函数regex_match#

regex_match(String, Regex) -> Bool

判断 PCRE2 Regex 是否匹配 String，受正则资源限制。

函数regex_captures#

regex_captures(String, Regex) -> Map | Null

返回完整匹配、字节位置、编号和命名捕获；不匹配返回 null。

函数url_resolve#

url_resolve(String, String?) -> Map

解析绝对或相对 HTTP(S) URL，移除 fragment、默认端口与点路径，并返回 host、path 和稳定指纹。

函数parse_json#

parse_json(String) -> JsonValue

严格解析 JSON String 为普通 HHY 值，错误包含行列。

函数encode_json#

encode_json(JsonValue, Map?) -> String

把可编码普通值转成 JSON；options 可启用 pretty。

函数parse_csv#

parse_csv(String | Stream<String>, Map?) -> Stream<Map>

把 String 或行 Stream 流式解析成 Stream<Map>。

函数encode_csv#

encode_csv(Stream<Map>, Map?) -> Stream<String>

把 Stream<Map> 流式编码为不含换行符的 CSV record Stream。

路径、文件与监听（15）

read_* 是读取操作；write_text、write_bytes、save_text 和 save_lines 使用同目录临时文件加原子替换；append_text 追加到目标文件。文件系统 action 会被 dry-run 拦截。

函数path#

path(String) -> Path

把 String 词法规范化为 Path，不访问文件系统。

函数path_join#

path_join(Path, String | Path) -> Path

组合 Path 与子路径并返回规范化的新 Path。

函数files#

files(Path, String, Map?) -> Stream<File | Directory>

按 glob 惰性遍历根目录，返回 File/Directory Stream。

函数read_text#

read_text(Path) -> String

完整读取 UTF-8 文件为 String。

函数read_lines#

read_lines(Path) -> Stream<String>

逐行惰性读取 UTF-8 文件并移除行终止符。

函数read_bytes#

read_bytes(Path) -> BytesBuffer

完整读取二进制文件为 BytesBuffer。

函数write_text#

write_text(Path, String, Map?) -> Path

以原子替换方式写 String，支持 overwrite/create_parents。

函数append_text#

append_text(Path, String) -> Path

把 String 追加到文件末尾。

函数write_bytes#

write_bytes(Path, BytesBuffer, Map?) -> Path

以原子替换方式写 BytesBuffer。

函数save_text#

save_text(String | Stream<String>, Path, Map?) -> Path

把 String 或文本 Stream 边拉取边原子保存。

函数save_lines#

save_lines(Stream<String>, Path, Map?) -> Path

把 String Stream 逐项写入并补 LF，最终原子替换。

函数copy#

copy(Path, Path, Map?) -> Path

复制文件，支持原子 no-replace；目标父目录须预先存在。

函数move#

move(Path, Path, Map?) -> Path

移动或重命名文件，遵守覆盖选项。

函数remove#

remove(Path) -> Path

删除明确 Path，并返回该 Path。

函数watch#

watch(Path, Map?) -> Stream<FileEvent>

返回无限 FileEvent Stream，支持 recursive 选项并响应取消。

进程、标准输入与定时（6）

run 直接传递 argv，不经过 Shell；只有 shell 明确采用 Shell 解析。进程启动会受 timeout、输出和进程数限制。

函数run#

run(List<String>, Map?) -> CommandResult

直接执行 argv，不经过 Shell；返回 CommandResult。

函数shell#

shell(String, Map?) -> CommandResult

显式用 Shell 执行 String；仅在需要重定向、管道等 Shell 语义时使用。

函数stdout_lines#

stdout_lines(CommandResult) -> Stream<String>

把 CommandResult.stdout 转为惰性行 Stream。

函数processes#

processes() -> Stream<Process>

获取当前进程快照的 Stream<Process>。

函数stdin_lines#

stdin_lines() -> Stream<String>

惰性读取标准输入行，直到 EOF 或取消。

函数every#

every(Duration) -> Stream<Int>

按指定 Duration 产生无限计时 tick Stream。

HTTP（10）

http.* 只构造不可变请求计划，timeout/retry 修改计划，只有 send 产生网络副作用。response_body 返回 UTF-8 文本，二进制响应使用 response_bytes。

函数http.get#

http.get(String, Map?) -> HttpRequest

构造 GET HttpRequest 计划，不发送网络请求。

函数http.post#

http.post(String, Map?) -> HttpRequest

构造 POST HttpRequest 计划，不发送网络请求。

函数http.put#

http.put(String, Map?) -> HttpRequest

构造 PUT HttpRequest 计划，不发送网络请求。

函数http.delete#

http.delete(String, Map?) -> HttpRequest

构造 DELETE HttpRequest 计划，不发送网络请求。

函数timeout#

timeout(HttpRequest, Duration) -> HttpRequest

返回设置请求超时的新 HttpRequest。

函数retry#

retry(HttpRequest, Map) -> HttpRequest

返回配置重试次数和退避的新 HttpRequest。

函数send#

send(HttpRequest) -> HttpResponse

执行 HttpRequest 网络副作用并返回 HttpResponse。

函数send_to#

send_to(HttpRequest, Path) -> HttpResponse

把 HTTP body 流式写入原子文件，返回包含 path 和 size 的 HttpResponse。

函数response_body#

response_body(HttpResponse) -> String

验证响应状态并把有界 body 解码为 UTF-8 String。

函数response_bytes#

response_bytes(HttpResponse) -> BytesBuffer

验证响应状态并返回有界二进制 BytesBuffer。

Web Runtime（25）

web.* 构建常驻应用、响应、中间件和生产端点；只有 web.listen 绑定网络端口。

函数web.app#

web.app() -> WebApp

创建 Web 应用。

函数web.get#

web.get(WebApp, String, Function) -> WebApp

函数web.post#

web.post(WebApp, String, Function) -> WebApp

函数web.put#

web.put(WebApp, String, Function) -> WebApp

函数web.patch#

web.patch(WebApp, String, Function) -> WebApp

函数web.delete#

web.delete(WebApp, String, Function) -> WebApp

函数web.use#

web.use(WebApp, Function) -> WebApp

追加可短路的 Middleware。

函数web.static#

web.static(WebApp, String, Path) -> WebApp

挂载安全静态目录。

函数web.cors#

web.cors(WebApp, Map?) -> WebApp

启用 CORS 与 OPTIONS 预检。

函数web.request_id#

web.request_id(WebApp) -> WebApp

启用请求 ID。

函数web.health#

web.health(WebApp, String?) -> WebApp

启用健康检查端点。

函数web.trust_proxy#

web.trust_proxy(WebApp) -> WebApp

信任代理提供的客户端地址。

函数web.text#

web.text(String, Map?) -> WebResponse

创建文本响应。

函数web.html#

web.html(String, Map?) -> WebResponse

创建 HTML 响应。

函数web.json#

web.json(JsonValue, Map?) -> WebResponse

创建 JSON 响应。

函数web.redirect#

web.redirect(String, Map?) -> WebResponse

创建重定向响应。

函数web.response#

web.response(Map) -> WebResponse

从选项 Map 创建响应。

函数web.multipart#

web.multipart(WebRequest, Path?) -> Map

解析受限 multipart 表单和上传。

函数web.stream#

web.stream(Stream, Map?) -> WebResponse

创建背压式流响应。

函数web.sse#

web.sse(Stream<String>) -> WebResponse

创建 Server-Sent Events 响应。

函数web.sign_cookie#

web.sign_cookie(String, String) -> String

用 HMAC 签名 Cookie 值。

函数web.verify_cookie#

web.verify_cookie(String, String) -> String | Null

验证并读取签名 Cookie。

函数web.metrics#

web.metrics(WebApp, String?) -> WebApp

启用 Prometheus 指标端点。

函数web.gzip#

web.gzip(WebApp) -> WebApp

按 Accept-Encoding 启用 gzip。

函数web.listen#

web.listen(WebApp, Map?) -> Null

启动有界 HTTP/1.1 Server。

受控并发（7，实验默认关闭）

函数atomic_state#

atomic_state(data)

创建 Runtime/PID 所有者绑定的状态句柄。（实验，默认关闭）

函数atomic_read#

atomic_read(state)

读取不可变快照。（实验，默认关闭）

函数atomic_update#

atomic_update(state, callback)

校验纯回调候选值并同步发布；提交前错误保留旧值。（实验，默认关闭）

函数atomic_close#

atomic_close(state)

关闭句柄；关闭后访问或再次关闭均失败。（实验，默认关闭）

函数atomic_file_update#

atomic_file_update(path, timeout, callback)

锁内读改写已有普通文件；timeout 仅限制等锁时间。（实验，默认关闭）

函数transaction_strict#

transaction_strict(config, options?, callback)

SQL 绑定当前 tx；捕获错误仍禁止提交；未知提交需对账。（实验，默认关闭）

函数task_map#

task_map(data, parallelism, result_limit, callback)

隔离进程执行，按输入顺序返回 List，限制直接子任务和接收结果。（实验，默认关闭）

阅读完整指南与示例包含保证矩阵、错误处理和实际验收证据