简介：这是一个面向 Go 开发者的轻量级内存数据集过滤开源项目，适合需要在服务端快速实现条件筛选、学习过滤表达式与算子设计的开发者参考。压缩包共 40 个文件，其中 34 个 Go 源码文件承载核心逻辑与单元测试，另含 YAML 持续集成配置、Markdown 说明、JSON 算子示例及 LICENSE，整体约 29KB，体量虽小但目录与模块划分完整。已有 193 人学习或下载。项目围绕内存数据集过滤展开，提供等于、不等于、大于等于、小于、包含、正则匹配、空值判断等常用算子，并通过较完整的测试用例展示条件组合、真假值判断与边界处理方式。读者可据此理解算子抽象、查询对象构建和测试组织思路，也可将相关实现裁剪后用于自己的 Go 服务，适合作为数据过滤组件、源码阅读与代码练习的参考。
1. 规则和代码分家：distil 这份开源项目源码包到底解决什么问题
上周帮朋友处理一个采集任务：十几个结构类似的商品页，字段就六七个，他写了两千多行 Python，选择器散落在几十个函数里，站点一改版就得全文件搜cssselect。这类活儿的痛点很具体：代码里混着两种语言，页面结构用 CSS 选择器写，字段逻辑用 Python 写，改一次结构就得动一次代码、跑一次回归。
distil 这个开源项目的思路是把两件事分开：页面结构用一份 YAML/JSON 声明描述，执行器按声明去跑。改版时你只改规则文件，不动 Python。源码包不大，解压就能读，适合两类人：手上维护着几十个站点采集规则的脚本作者，以及正在搭数据管道、想让采集逻辑变成配置而不是代码的后端。下面按我自己拆包的顺序，从目录、规则、执行链路一路讲到怎么加自己的转换器。
2. 解压之后先看哪里：distil 的三层目录结构与最小可跑通命令
拿到一个陌生的源码包，我一般不看 README 的第一段，直接看目录。目录决定了这个项目是「一个脚本」还是「一个有分层的工具」，也决定了你二次开发时该往哪个文件里塞东西。
2.1 目录分层：声明层、执行层、输出层各管什么
这个包不大，按职责看能切成三层，解压后大致是这么个形状：
distil-src/ ├── distil/ │ ├── cli.py # 命令行入口，run / validate / dump 几个子命令 │ ├── loader.py # 读规则文件，转成强类型对象 │ ├── schema.py # Spec / Field 这些数据类，规则的「形状」 │ ├── fetch.py # 请求层：会话复用、重试、编码兜底 │ ├── extract.py # 提取层：把 CSS 选择器应用到 DOM 上 │ ├── transform.py # 清洗层：类型转换、去符号、拼 URL │ ├── pipeline.py # 把 fetch → extract → transform 串起来 │ └── output.py # 写出 JSON / JSONL / CSV ├── examples/ │ └── books.yaml # 一份可以直接跑的规则样例 ├── tests/ ├── setup.py └── README.md我按模块职责记成上面这样，文件名可能有一两个出入，位置不会差太多。关键是分层意图很清楚：schema.py定义规则能写什么，loader.py负责把用户写的 YAML 变成Spec对象，extract.py只认Spec不认 YAML。这个边界很有用——你想加一种新的字段类型，改schema.py和transform.py就够了，extract.py不用碰。
examples/那一份 YAML 是最值钱的东西。很多人拆包先读cli.py，读半天不知道规则该长什么样；反过来，先照抄一份能跑的规则，再回头找它是谁解析的，路径最短。
2.2 依赖与安装：从源码包到第一条 distil run 命令
这类工具站在几个成熟轮子上，依赖不会重：requests管请求，lxml管解析，cssselect让 lxml 能直接用 CSS 选择器（原生只有 XPath），PyYAML读规则，CLI 层常见做法是argparse，也有的包用click。
# 1. 解压（<源码包> 换成你下载的那个 zip 文件名） unzip <源码包>.zip -d distil-src cd distil-src # 2. 建虚拟环境，别装到系统 Python 里去 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 3. 用 -e 装成可编辑模式，改了源码立刻生效，方便二次开发 pip install -e . # 4. 确认入口通了 distil --helppip install -e .这一步别省成python setup.py install。可编辑模式下distil/目录里改一行、命令行立刻是新行为；装成正式包之后你每次改都要重装，调试节奏会断。
跑第一条规则之前，先用validate子命令把规则文件本身过一遍，比直接run报错好定位：
# 只校验规则，不发请求，几秒钟出结果 distil validate examples/books.yaml # 校验通过再真跑 distil run examples/books.yaml --url https://demo.example/list提示：
--url传的地址只用一份 demo 页面就行，第一遍跑通的目标是「输出结构对」，不是「数据全」。别一上来就对着生产站点跑，报错和风控混在一起，你分不清是规则写错还是被拦了。
2.3 一份最小规则文件：先跑通再谈其它
规则文件长这样，我把它当成模板记，写新站点时只改item和各字段的selector：
# examples/books.yaml name: book_list # 规则名，输出时会带上，方便区分任务 item: "div.book-item" # 重复块的选择器，一个块 = 输出里的一行 fields: title: selector: "h3.title" attr: text # text 取文本，href/src 取属性，html 取整段 HTML price: selector: "span.price" attr: text type: float strip: "¥" detail_url: selector: "a.detail" attr: href url_join: "https://demo.example" # 相对路径拼成绝对地址 tags: selector: "span.tag" many: true # 一组值，输出成数组而不是单个值三个键要理解到位：item决定「一行数据」的边界，它写在顶层不在fields里；selector永远相对于item节点往下找，不是整页找；attr是取值方式，text、href、>fields: title: selector: "h3.title" attr: text images: selector: "ul.thumbs img" many: true attr: src variants: selector: "ul.variants li" many: true fields: # 嵌套：每个 li 再按下面这套规则取一遍 name: selector: "span.name" attr: text sku: selector: "span.sku" attr: "data-sku"
many: true加fields是嵌套数组，输出是[{name, sku}, ...]；many: true不加fields是标量数组，输出是["红色", "蓝色"]。这两种写法在loader.py里走的是不同分支，混着写会把结构搞乱——比如给标量字段误加了fields，加载期就该报错，别等跑完发现 JSON 里躺着一堆{"": null}。
嵌套层数不要超过两层。三层以上规则文件会变得比 Python 还难读，而且调试时你根本不知道是第几层选择器没命中。真需要三层结构，我的做法是在规则里把中间层拍平，输出后再用一段脚本重组——配置文件的表达能力是有上限的，别硬撑。
3.3 清洗与派生：货币、日期、相对 URL 怎么归一
采集出来的原始字符串几乎不能直接入库：价格带符号和千分位、日期是「3 天前」、链接是/item/123。清洗层用一串transforms表达，顺序很重要：
fields: price: selector: "span.price" attr: text transforms: - strip: "¥" # 先去掉货币符号 - replace: [",", ""] # 再去千分位，否则 1,299 转 float 会炸 - type: float # 最后才转类型 default: 0 published_at: selector: "time" attr: datetime # 优先取 datetime 属性，比文本规范 type: date format: "%Y-%m-%d" detail_url: selector: "a.detail" attr: href url_join: "https://demo.example"顺序写反的典型翻车：先type: float再strip，float("¥99")直接抛ValueError，而这个字段又没配default，结果是整行数据的这个字段为null——不会有异常冒到顶层，你只会看到一堆空价格。清洗链的设计原则是「字符串操作在前，类型转换在后，兜底值最后」。
url_join走的是urljoin语义，能正确处理/a/b、../c、//cdn.x/y三种相对形式。不要自己写base + href字符串拼接，协议相对地址拼出来就是https://demo.example//cdn.x/y，多一个斜杠有些服务端会 404。
3.4 加载期校验：把错误拦在跑之前
规则写错的成本很高——你得先发请求、拿到页面、跑提取，才发现某个字段少写了selector。所以加载阶段要做一次结构校验：
# distil/loader.py（节选，按职责还原） import yaml from pathlib import Path from .schema import Spec, SpecError def load_spec(path): raw = yaml.safe_load(Path(path).read_text(encoding="utf-8")) if not isinstance(raw, dict): raise SpecError("规则文件顶层必须是 mapping") spec = Spec.from_dict(raw) # dict → 强类型对象，缺 item 这里就炸 for name, field in spec.walk_fields(): if not field.selector: raise SpecError(f"字段 {name} 缺 selector") if field.many and field.fields is not None and not isinstance(field.fields, dict): raise SpecError(f"字段 {name} 的嵌套必须写成 mapping") if field.type == "date" and not field.format: raise SpecError(f"字段 {name} 声明了 date 但没给 format") return specwalk_fields()要做递归，把嵌套字段也遍历到，只校验顶层会漏掉第二层的问题。Spec.from_dict用dataclass或pydantic都行，前者零依赖，后者校验规则更省事——我一般选dataclass加手写校验，因为错误信息可以写得比框架默认的更贴近规则文件（直接报「字段 price 缺 selector」而不是一串 pydantic 的 loc 路径）。
这段校验的价值在第 5 章会体现：它能挡掉大约一半的「跑完结果不对」，剩下的一半才是真·页面问题。
4. 执行链路：一条 URL 从进来到变成 JSONL 的完整过程
规则只是输入，真正干活的是pipeline.py那几十行。这一章按执行顺序拆开看，每一层都有它的失败模式。
4.1 会话与抓取层：重试、超时和编码兜底
抓取层最容易被写薄。一个裸requests.get(url)在真实站点上活不过十分钟，缺的是会话复用、重试和编码处理：
# distil/fetch.py（节选） import requests from requests.adapters import HTTPAdapter def make_session(retries=3, timeout=15): s = requests.Session() adapter = HTTPAdapter(max_retries=retries, pool_connections=10, pool_maxsize=10) s.mount("http://", adapter) s.mount("https://", adapter) s.headers.update({"User-Agent": "distil (+rules)"}) # 默认 UA 很多站点直接 403 s.request_timeout = timeout return s def fetch(session, url): resp = session.get(url, timeout=getattr(session, "request_timeout", 15)) resp.raise_for_status() # 编码兜底：headers 里写 iso-8859-1 但页面其实是 UTF-8 的情况太常见 if not resp.encoding or resp.encoding.lower() == "iso-8859-1": resp.encoding = resp.apparent_encoding return resp.text三个点值得说。pool_maxsize决定连接池大小，配得太小在批量跑 URL 列表时会频繁建连，速度掉一半；HTTPAdapter(max_retries=...)重试的是连接层错误和 5xx，不会重试 4xx——429 需要你自己读Retry-After再决定睡多久。编码那三行是纯经验：requests在响应头没写 charset 时会按 RFC 默认给ISO-8859-1，中文页面直接变乱码，apparent_encoding靠字符分布猜，虽然不是百分百准，但比默认值强得多。
session.request_timeout这种挂属性的写法不太优雅，好处是超时值能一路传到fetch不用改函数签名。二次开发时如果你要加代理池或者签名参数，改make_session一个地方就够。
4.2 提取核心：把规则应用到 DOM 上的那段代码
提取层是全项目最短也最关键的一段。它的输入是一个 lxml 节点和一个Field，输出是字符串或None：
# distil/extract.py（节选） from lxml import html def parse(text): return html.fromstring(text) def pick(node, field): try: nodes = node.cssselect(field.selector) except Exception: return [] if field.many else None if not nodes: return [] if field.many else None if field.many: return [read_one(n, field) for n in nodes] return read_one(nodes[0], field) def read_one(node, field): attr = field.attr or "text" if attr == "text": value = node.text_content() elif attr == "html": value = html.tostring(node, encoding="unicode") else: value = node.get(attr) # href / src /># distil/pipeline.py（节选） from .extract import parse, pick from .transform import apply_transforms from .fetch import fetch def extract_page(spec, text): page = parse(text) rows = [] for node in page.cssselect(spec.item): row, hit = {}, 0 for name, field in spec.fields.items(): raw = pick(node, field) value = apply_transforms(raw, field) if value not in (None, [], ""): hit += 1 row[name] = value if hit: # 全空的块直接丢，多半是表头或广告位 rows.append(row) return rows def run(spec, url, session): return extract_page(spec, fetch(session, url))hit这个计数器是这段代码里最容易被忽略、但最实用的一处：item选择器写宽了会把表头行、推荐位、空占位一起选进来，输出一堆全null的记录。用「至少命中一个字段」作为过滤条件，能自动滤掉大部分噪声，还不用你回头改选择器。
另一个设计点是逐字段容错。任何一个字段的转换抛异常都被apply_transforms内部吞掉、回落到default，不会中断整页。代价是错误被静默了——所以第 5 章会讲怎么在调试模式下把这些被吞掉的异常打出来。
4.4 输出层：JSON、JSONL 与 CSV 的取舍
# distil/output.py（节选） import json def write_jsonl(rows, fp, spec_name): for row in rows: row["_rule"] = spec_name fp.write(json.dumps(row, ensure_ascii=False) + "\n") def write_csv(rows, fp): if not rows: return writer = csv.DictWriter(fp, fieldnames=list(rows[0].keys())) writer.writeheader() writer.writerows(rows)三种格式各有它的位置：JSON 适合单页调试，人能直接看；JSONL 适合落盘和后续流式处理，一行一条、追加写不会破坏已有文件，中断了也能续；CSV 适合丢给非技术的同事看，但嵌套字段会被序列化成字符串，variants那一列会变成一坨 JSON 文本。我的习惯是默认 JSONL，需要交付时再转 CSV。
_rule这个字段是我自己加的，用来标记这行数据来自哪份规则。多规则混跑、输出到同一个文件时，没这个字段你根本分不清哪行是哪来的——加一个字段的成本远低于事后回溯。
5. 避坑与排查：五个把采集任务跑挂的真实原因
这一章按「现象 → 原因 → 解决」写，都是我或者身边的人真踩过的。排查顺序建议固定：先 dump 原始 HTML，再看规则，最后才怀疑代码。
5.1 抓取阶段：拿到手的就是一张空壳页面
现象一：规则 validate 通过了，run 也不报错，但输出是空数组或者一堆 null。原因：fetch拿回来的根本不是你以为的那个页面。常见三种——站点给爬虫返回了精简版/无 JS 的落地页；请求被重定向到验证页，raise_for_status()因为状态码是 200 不会拦；UA 被识别后返回了内容为空的壳。解决：在fetch里加一个--dump-html开关，把resp.text原样存盘，用浏览器打开看。十次里有七八次，打开一眼就明白了。别在没看原文的情况下改选择器，那是纯猜。
现象二：中文全是ä¸æ–‡这种乱码，英文正常。原因：响应头没写 charset，requests按 RFC 默认给ISO-8859-1，resp.text就按单字节解了。解决：就是第 4 章那三行，判断encoding为空或者是iso-8859-1时切到apparent_encoding。更稳的做法是先resp.content拿到 bytes，正则从<meta charset=...>里读，读不到再猜——多写五行，能省掉一整类玄学问题。
现象三：前几十个页面很顺，跑到一两百条开始大面积超时、429，甚至直接连不上。原因：并发开太大、没有间隔、每次请求新建连接。三项叠加，目标站点的风控和你的本机端口都会受不了。解决：一个任务复用一个Session（连接池复用能省掉大量握手）；并发用信号量卡在 3～5，别贪；请求之间加 0.3～1 秒的随机间隔；遇到 429 读Retry-After头，没有就指数退避。这块没有银弹，慢就是快。
5.2 提取阶段：选择器对、结果还是错的两种情况
现象四：明明页面上有 20 个商品，结果只抓到 1 条，或者抓出来的每条都一样。原因：item选择器写错了层级。写成div.list会只命中一个外层容器，里面再怎么循环都只有一条；写成div这种过宽的会命中一堆无关块。还有一种隐蔽情况：fields里的selector写成了整页范围的绝对路径（.list .item .title），在item节点内部再匹配就永远匹配不到，全变 null。解决：item只写到「重复块」这一层，fields里的选择器一律相对写。验证方法是把item单独抽出来跑一次len(page.cssselect(spec.item))，数字对了再谈字段。
现象五：价格字段大面积ValueError，或者1,299转出来是1.0。原因：清洗链顺序错、清洗项漏。float("1,299")抛异常；float("1.299")会安静地给你1.299，如果你的意图是 1299 那就错得很难发现；还有「暂无报价」「面议」这类非数值文本。解决：strip→replace→type的固定顺序，replace要把千分位逗号和全角逗号都列上；所有数值字段一律配default；调试时打开--strict，让被吞掉的转换异常打出来，否则你只会看到一堆null而不知道原因。这类问题最典型的特征就是「有的行对有的行错」，别用抽样三行就判定规则没问题。
6. 二次开发：加一个自定义 transform 和一种输出后端
规则层的表达能力是有限的，迟早会碰到「这个字段的换算逻辑用 YAML 写不出来」的情况。这时候别去改extract.py，扩展点有两个：transform.py的转换器注册表，和output.py的写出后端。
先说转换器。清洗层本质上是一张名字 → 函数的表，加一个进去就行：
# distil/transform.py 里追加 def to_cents(value, **_): """把 '¥1,299.00' 这类文本换成整数分，避免浮点误差""" text = str(value).replace("¥", "").replace(",", "").strip() if not text or not text[0].isdigit(): return None return int(round(float(text) * 100)) TRANSFORMS = { "cents": to_cents, # ... 原有的 strip / replace / type 等比这里多 }函数签名统一成(value, **kwargs)，多出来的参数由规则文件里的写法传进来，这样以后想加「按汇率换算」之类的可配置转换器不用改签名。挂上去之后规则里直接写- cents，规则和代码就此解耦——这是我最看重的一点：以后再有人要改换算逻辑，改的是transform.py一个函数，所有引用它的规则文件同时生效。
再说输出后端。output.py里的每个写出函数形状都该是(rows, fp)，要加一个落库的 sink，照着这个形状写就行：
# distil/output.py 追加一个 SQLite sink import sqlite3, json class SqliteSink: def __init__(self, path, table): self.conn = sqlite3.connect(path) self.table = table def write(self, rows): if not rows: return cols = list(rows[0].keys()) ddl = ", ".join(f'"{c}" TEXT' for c in cols) self.conn.execute(f'CREATE TABLE IF NOT EXISTS "{self.table}" ({ddl})') sql = f'INSERT INTO "{self.table}" VALUES ({", ".join("?" * len(cols))})' self.conn.executemany(sql, [ [json.dumps(r.get(c), ensure_ascii=False) if isinstance(r.get(c), (list, dict)) else r.get(c) for c in cols] for r in rows ]) self.conn.commit()两个细节值得注意：嵌套字段（variants那种）在关系表里没有原生类型，统一json.dumps成 TEXT 是成本最低的做法，别急着建关联表；CREATE TABLE IF NOT EXISTS用列名直接拼 DDL 意味着规则加字段不会自动改表结构，字段一变你就得手动ALTER，这是这种「无 schema」写法的必然代价，接受它或者上迁移脚本。
改完怎么验证？我一般按三步走：先拿examples/里那份已知结果的规则跑一遍，确认老功能没被带崩；再挑三个结构差异最大的目标页面，逐字段对比输出条数和字段命中率（写个小脚本统计每个字段的null比例，超过 30% 就该看规则了）；最后把这三份页面的 HTML 存成本地 fixture，加进tests/，以后改extract.py一律先跑这组用例。第三步最枯燥也最值钱，我吃过没做的亏——一次「顺手优化」选择器匹配逻辑，线上跑了三天才发现嵌套数组全变成了一维。
从那以后我每加一个自定义转换器或者 sink，都会先把三个页面的 HTML dump 下来存成 fixture，跑一遍--strict模式的 dry-run，确认字段命中率和改动前一致，才允许它进生产任务。这套动作多花二十分钟，换的是不用半夜爬起来查「为什么昨天还有数据今天全空」。希望帮到你。
本文还有配套的精品资源，点击获取
