px普贤教程PuXian
普贤教程 › 生态包

数据访问与解析实战

SQL 构造、ORM 映射、迁移计划、DOM / XPath / 流式 XML、网页抓取、EXIF、路径运算、日期区间与相对时间。这一组的共同点是只产出数据，不碰 IO：SQL 是字符串，迁移是计划，XML 是树，路径运算是纯函数——所以能离线测、能断言。

一个脚本跑完 10 个包

px
import querybuilder
import orm
import migration
import domlite
import xmlstream
import xpath
import scraper
import fileinfo
import dateperiod
import strtotime

def main():
    var q = qb_select("users")
    q = qb_where(q, "age", ">=", 18)
    q = qb_where_in(q, "city", ["北京", "上海"])
    q = qb_order_by(q, "id", "desc")
    q = qb_limit(q, 20)
    print(qb_build(q))
    print(qb_build(qb_insert("users", {"name": "甲", "age": 20})))

    var m = orm_new("users", "id")
    m = orm_field(m, "name", "string")
    m = orm_field(m, "age", "integer")
    print(json_stringify(orm_to_row(m, {"id": 1, "name": "甲", "age": 20})))
    print(json_stringify(orm_from_row(m, {"id": "1", "name": "甲", "age": "20"})))
    print(json_stringify(orm_validate(m, {"name": "", "age": 20})))

    var mig = mig_new()
    mig_add(mig, 1, "create_users", "CREATE TABLE users(id INT)", "DROP TABLE users")
    mig_add(mig, 2, "add_email", "ALTER TABLE users ADD email TEXT", "ALTER TABLE users DROP email")
    print(str(mig_count(mig)) + " 条，待执行 " + json_stringify(mig_plan_up(mig, [1])))
    print("回滚到 1 " + json_stringify(mig_plan_rollback(mig, 1, [1, 2])))

    var ul = dom_new("ul", {"class": "list"}, [])
    dom_append(ul, dom_new("li", {}, [dom_text("甲")]))
    dom_append(ul, dom_new("li", {}, [dom_text("乙")]))
    print(dom_to_xml(ul, ""))
    print("li=" + str(xp_count(ul, "li")) + " 首项=" + xp_text(ul, "li"))

    print(str(xp_exists(ul, "ol")) + " " + str(xp_attr(ul, "ul", "class")))

    var w = xw_new()
    xw_start_document(w)
    xw_start_element(w, "root", {})
    xw_write_element(w, "name", "甲")
    xw_end_element(w, "root")
    xw_end_document(w)
    var xml = xw_get_output(w)
    print(xml)

    var r = xr_new("<a><b>1</b><c/></a>")
    var n = 0
    var ev = xr_read_next(r)
    while ev != null:
        n += 1
        ev = xr_read_next(r)
    print("事件 " + str(n))

    var html = "<html><head><title>示例页</title></head><body><a href=\"/a\">甲</a><a href=\"https://b.example\">乙</a><img src=\"/x.png\"></body></html>"
    print(json_stringify(sc_extract_links(html)))
    print(sc_extract_title(html) + " / " + json_stringify(sc_extract_images(html)))

    var fi = fi_new("/var/log/app/error.log")
    print(fi_get_filename(fi) + " " + fi_get_basename(fi) + " " + fi_get_extension(fi) + " " + fi_get_dirname(fi))
    print(fi_normalize(fi_join("/srv/app", "../etc/passwd")))

    var dp = dp_new(1700000000, 1700086400, 86400)
    print(str(dp_count(dp)) + " 个时间点 " + json_stringify(dp_format(dp, "%Y-%m-%d")) + " 含中期=" + str(dp_contains(dp, 1700043200)))

    var base = 1700000000
    print(str(st_parse_relative("tomorrow", base)) + " " + str(st_parse_relative("yesterday", base)) + " " + str(st_parse_relative("next monday", base)))
    print("+1 day = " + str(st_parse_relative("+1 day", base)) + "，2 days ago = " + str(st_parse_relative("2 days ago", base)))
text
SELECT * FROM `users` WHERE `age` >= 18 AND `city` IN ('北京', '上海') ORDER BY `id` desc LIMIT 20
INSERT INTO `users` (`name`, `age`) VALUES ('甲', 20)
{"name":"甲","age":20}
{"name":"甲","age":20}
["name is empty"]
2 条，待执行 [{"version":2,"name":"add_email","sql":"ALTER TABLE users ADD email TEXT"}]
回滚到 1 [{"version":2,"sql":"ALTER TABLE users DROP email"}]
<ul class="list"><li>甲</li><li>乙</li></ul>
li=2 首项=甲
false list
<?xml version="1.0" encoding="UTF-8"?>
<root><name>甲</name></root>

事件 6
["/a","https://b.example"]
示例页 / [{"src":"/x.png"}]
error.log error log /var/log/app
/srv/etc/passwd
2 个时间点 ["2023-11-14","2023-11-15"] 含中期=false
1700086400 1699913600 1700518400
+1 day = 1700086400，2 days ago = 1699827200

querybuilder：链式 SQL

px
var q = qb_select("users")
q = qb_where(q, "age", ">=", 18)
q = qb_where_in(q, "city", ["北京", "上海"])
q = qb_where_like(q, "name", "甲%")
q = qb_order_by(q, "id", "desc")
q = qb_limit(q, 20)
qb_build(q)              # 内联转义后的 SQL 字符串
qb_build_params(q)       # [SQL（值是 ?）, 参数列表]
qb_build_ok(q)           # Ok([sql, params]) / Err("表名不能为空")
qb_build 是内联转义，qb_build_params 才是参数化两个入口分工不同：qb_build 把值经 qb_escape 转义后直接拼进 SQL，转义只处理引号与反斜杠；接用户输入一律用 qb_build_params，产出 [sql, [7]] 这种形态，交给驱动的预处理绑定。要前置校验（空表名、空列）用 qb_build_ok，返回 Err 而不是拼出半条 SQL。

dbal：参数化查询构造

dbal 的函数全部是 db_* 前缀，与 querybuilder 的 qb_* 分开，两个包可以在同一个文件里 import：

px
import dbal
import querybuilder

var q = db_new("users")
q = db_select(q, ["id", "name"])
q = db_where(q, "age", ">", 18)
q = db_limit(q, 10)
print(db_to_sql(q))      # SELECT id, name FROM users WHERE age > ? LIMIT 10
print(db_params(q))      # [18]
print(db_build(q))       # [sql, params] 一起取
print(db_insert_sql("users", {"name": "甲"}))   # INSERT INTO users (name) VALUES (?)
两个包的函数前缀分开dbal 全部导出 db_*，querybuilder 导出 qb_*，同一文件里同时 import 不冲突。qb_select 是 querybuilder 的（只收表名），dbal 的是 db_select(qb, columns)。

二选一的口径：要安全用 dbal（参数化，适合接真实的 MySQL 驱动）；要方便用 querybuilder（链式、能拼复杂 join，取参数用 qb_build_params）。生产写库一律 dbal。

orm：字段映射与校验

px
var m = orm_new("users", "id")
m = orm_field(m, "name", "string")
m = orm_field(m, "age", "integer")
orm_to_row(m, {"id":1,"name":"甲","age":20})          # {"name":"甲","age":20}（pk 不进 row）
orm_from_row(m, {"id":"1","name":"甲","age":"20"})    # {"name":"甲","age":20}（integer 转成 int）
orm_validate(m, {"name":"", "age":20})               # ["name is empty"]
只处理声明过的字段orm_from_row 按 orm_field 逐个取出并按声明类型转换（声明 integer，从库里拿来 "20" 转成 20）；没声明的字段（例子里的 id）不会出现在结果里，主键要自己另外取。orm_validate 查三类问题：必填缺失、空串、类型不符。

migration：迁移计划

px
var mig = mig_new()
mig_add(mig, 1, "create_users", up_sql, down_sql)
mig_add(mig, 2, "add_email", up_sql, down_sql)
mig_plan_up(mig, applied_versions)        # 待执行的 [{version,name,sql}]
mig_plan_rollback(mig, target, applied)   # 要回滚的
mig_latest(mig) / mig_count(mig)

只生成计划，不执行 SQL、不记录版本。版本表要自己维护：跑完把 version 写进 schema_migrations，下次读出来传给 mig_plan_up。

domlite + xpath：建树与查询

px
var ul = dom_new("ul", {"class":"list"}, [])
dom_append(ul, dom_new("li", {}, [dom_text("甲")]))
dom_to_xml(ul, "")          # <ul class="list"><li>甲</li></ul>
dom_inner_text(ul)          # 甲
dom_find_by_tag(ul, "li")   # list

xp_count(ul, "li")          # 2
xp_text(ul, "li")           # 甲（第一个）
xp_select_all(ul, "li")     # 节点列表
xp_select_by_attr(ul, "ul", "class", "list")
xp_attr(ul, "ul", "class")  # list
dom_to_xml 的 indent 只收字符串传 0 会在 indent + "<" 处抛 R1002 无法相加: int + string。要缩进传 " "，不要缩进传 ""。 另外 xpath 不是真 XPath：只支持按标签名的直接子节点查询，没有路径表达式、谓词、轴。//div[@id="x"] 这种写法它不认。

xmlstream：流式读写

px
var w = xw_new()
xw_start_document(w)
xw_start_element(w, "root", {})
xw_write_element(w, "name", "甲")
xw_end_element(w, "root")
xw_end_document(w)
xw_get_output(w)

var r = xr_new(xml)
var ev = xr_read_next(r)     # {type:"start"/"end"/"pi"/"comment", name, attrs}
while ev != null:
    …
    ev = xr_read_next(r)
读循环看 xr_is_done，或直接取全量xr_read_next 读完返回 null 并推进位置，所以 while not xr_is_done(r) 与 while ev != null 两种写法都对。要一次拿完用 xr_read_all(r) 返回事件列表；重读同一份 XML 前先 xr_reset(r)。写侧 xw_write_element 会自动转义 & / < / >，不要把已转义的文本再喂进去。

scraper：抓页面里的链接与文本

px
sc_extract_links(html)     # ["/a", "https://b.example"]
sc_extract_images(html)    # [{"src":"/x.png"}]
sc_extract_title(html)     # 示例页
sc_extract_meta(html, "description")
sc_extract_text(html)      # 去标签后的正文
sc_strip_tags(html)

基于字符串匹配，不是 HTML 解析器：不处理实体、不补全错误嵌套、不管 <base>。相对链接不会自动补全，要自己拼上 https://host。对「拿一张静态页的几个字段」够用；复杂的页面换 htmlparse 包。

exif：JPEG 元数据

px
exif_has_marker(jpeg_bytes)
var seg = exif_find_segment(jpeg_bytes, 0xE1)     # APP1
exif_parse_tiff(jpeg_bytes, tiff_offset)          # {tag: value}
exif_tag_name(0x0112)                             # Orientation

只解析 EXIF 的 TIFF 结构（IFD0 / ExifIFD / GPS 的基本标签），不做缩略图、不解析 MakerNote。要拿拍摄时间与 GPS 够用；要做相机型号专有字段得自己按 tag 读原始值。

入参是 bytes，不是路径先 read_file_opt(path).ok() 读成字节再传。大图只读前 64KB 也能解出 IFD（EXIF 段在文件头附近），不必整文件加载。

fileinfo：纯路径运算

px
var fi = fi_new("/var/log/app/error.log")
fi_get_filename(fi)      # error.log
fi_get_basename(fi)      # error
fi_get_extension(fi)     # log
fi_get_dirname(fi)       # /var/log/app
fi_is_absolute(fi)       # true
fi_join("/srv/app", "../etc/passwd")    # /srv/app/../etc/passwd
fi_normalize("/srv/app/../etc/passwd")  # /srv/etc/passwd
fi_normalize 会真的把 .. 折上去上面这条从 /srv/app 一路折到了 /srv/etc/passwd。如果拿用户输入拼路径，必须先 normalize 再校验前缀，否则 ../ 直接穿越出沙箱。fi_join 本身不做任何限制。

dateperiod：日期区间

px
var dp = dp_new(1700000000, 1700086400, 86400)
dp_count(dp)          # 2（含端点）
dp_to_list(dp)        # [1700000000, 1700086400]
dp_contains(dp, ts)   # 是否落在步长上
dp_format(dp, "%Y-%m-%d")       # ["2023-11-14","2023-11-15"] ← 全区间
dp_format_one(dp, "%Y-%m-%d")   # "2023-11-14" ← 只取第一个
dp_format_at(dp, "%Y-%m-%d", 1) # "2023-11-15"
dp_days(s, e) / dp_hours(s, e) / dp_weeks(s, e) / dp_months(s, e)
dp_month_starts(s, 3)           # 之后 3 个日历月的起点
dp_format 返回整个列表，单值用 dp_format_onedp_format(dp, fmt) 返回区间内所有时间点的格式化结果；只要一个值用 dp_format_one，要第 n 个用 dp_format_at。interval_sec 必须为正，dp_new(s, e, 0) 这种会直接判无效（dp_valid → false），不会挂住。另外 dp_months 是 30 天近似，要日历月用 dp_month_starts（31 号的月会在短月落到月末）。

strtotime：英文相对时间

px
st_parse_relative("tomorrow", base)      # base + 86400
st_parse_relative("yesterday", base)     # base - 86400
st_parse_relative("next monday", base)   # 下一个周一
st_parse_relative("+1 day", base)        # base + 86400
st_parse_relative("2 days ago", base)    # base - 172800
st_parse_relative_ok("garbage", base)    # Err(无法识别的写法: garbage)
st_parse("now")                          # 以当前时间为基准
认不出的写法用 st_parse_relative_ok 显式报错st_parse_relative 遇到不认识的词会原样返回基准时间（和「本来就没变」分不清）；需要区分时用 _ok 版本，拿 Err 里的诊断文案。数字与单位之间空格可有可无（+1 day / +1day / in 2 hours 都认）。