px普贤教程PuXian
普贤教程 › 进阶教程

综合实例

三个完整例子，覆盖三种最常见的使用场景：命令行工具、Web 接口、定时任务。都能直接跑，或按注释改两行就用。

实例一：词频统计（命令行工具）

读一个文件，统计每个词出现的次数，按字母序输出。

px
def pad(s, n):
    var out = s
    while len(out) < n:
        out += " "
    return out

def word_freq(text):
    var freq = {}
    for w in split(text, " "):
        var k = to_lower(trim(w))
        if len(k) == 0:
            continue
        if freq.has(k):
            freq[k] = freq[k] + 1
        else:
            freq[k] = 1
    return freq

def main():
    var a = args()
    if len(a) < 2:
        print("用法：px run wc.px <文件路径>")
        return
    var path = a[1]
    var r = read_file_opt(path)
    if r.is_err():
        print("读不到文件：" + path)
        return

    var freq = word_freq(r.ok())
    for k in sorted(freq.keys()):
        print(pad(k, 8) + str(freq[k]))

准备一个文件 sample.txt：

text
The quick brown fox the THE end
sh
px run wc.px sample.txt
text
brown   1
end     1
fox     1
quick   1
the     3

关键点：

  • args() 返回列表，第 0 项是脚本名，参数从第 1 项开始；
  • read_file_opt 让"文件不存在"变成可处理的 Err，而不是崩溃；
  • to_lower 让 THE / the 合并计数。

实例二：留言接口（Web）

一个 Ma 托管的接口：POST 写留言、GET 读全部。数据存 JSON 文件，够用且没有依赖。

px
let DATA = "/data/www/api_example/messages.json"

def get_env(k, dflt = ""):
    var v = env(k)
    if v == null:
        return dflt
    return str(v)

def reason(st):
    if st == 200:
        return "OK"
    if st == 400:
        return "Bad Request"
    if st == 405:
        return "Method Not Allowed"
    if st == 500:
        return "Internal Server Error"
    return "OK"

def json_out(status, obj):
    print("Status: " + str(status) + " " + reason(status))
    print("Content-Type: application/json; charset=utf-8")
    print("")
    print(json_stringify(obj))

def load():
    var r = read_file_opt(DATA)
    if r.is_err():
        return []
    var p = json_parse_opt(r.ok())
    if p.is_err():
        return []
    return p.ok()

def main():
    var method = get_env("REQUEST_METHOD")
    var rows = load()

    if method == "GET":
        json_out(200, {"ok": true, "total": len(rows), "rows": rows})
        return

    if method != "POST":
        json_out(405, {"ok": false, "error": "只支持 GET / POST"})
        return

    var raw = ""
    var b = read_file_opt("/dev/stdin")
    if b.is_ok():
        raw = b.ok()
    var p = json_parse_opt(raw)
    if p.is_err():
        json_out(400, {"ok": false, "error": "请求体不是合法 JSON"})
        return

    var body = p.ok()
    if not body.has("text"):
        json_out(400, {"ok": false, "error": "缺少 text"})
        return

    var text = trim(body["text"])
    if len(text) == 0:
        json_out(400, {"ok": false, "error": "text 不能为空"})
        return
    if len(text) > 200:
        json_out(400, {"ok": false, "error": "text 最长 200 字"})
        return

    var item = {"text": text, "ts": now_sec()}
    rows.append(item)
    write_file(DATA, json_stringify(rows))
    json_out(200, {"ok": true, "saved": item})

Ma 配置：

json
{"path": "/api/messages", "exact": true, "script": "messages.px",
 "methods": ["GET", "POST"], "mode": "fast", "timeout_ms": 3000}

本地测：

sh
mkdir -p /tmp/api_example && DATA=/tmp/api_example/messages.json
printf '%s' '{"text":"第一条"}' | REQUEST_METHOD=POST px run messages.px
printf '' | REQUEST_METHOD=GET px run messages.px
这个写法有并发问题两个请求同时读到同一份 JSON、各自 append、再写回，会丢一条。生产上要换成数据库 + 事务或行锁。文件存储只适合单用户、低并发的工具。

实例三：幂等定时任务

每天只做一次的事，用标记文件保证幂等——cron 触发多次也不会重复执行。

px
let MARK = "/tmp/px_job_done"
let LOG = "/tmp/px_job.log"

def log(msg):
    append_file(LOG, now() + " " + msg + "\n")

def run_once():
    if exists(MARK):
        log("已执行过，跳过")
        return 0

    # 真正要做的活
    var total = 0
    var i = 0
    while i < 1000:
        total += i
        i += 1

    write_file(MARK, str(now_sec()))
    log("执行完成，结果 " + str(total))
    return 1

def main():
    var n = run_once()
    print(str(n))

crontab 里这么挂：

sh
0 3 * * * cd /opt/app && ./bin/job >> /var/log/job.out 2>&1

第二次运行会打印 0 并写一行「已执行过」，不会重复干活。

把「做过什么」写进日志定时任务出问题时，唯一能看的就是日志。每次执行至少记三件事：开始时间、结果、耗时。

三个例子的共同点

共性为什么
外部输入一律先判 Err*_opt 系列把失败变成值，能优雅返回
字符串拼接前先 str()语言不隐式转换
输出走统一函数改格式时只改一处
没有全局可变状态请求/进程之间不共享，避免竞态