简介这是一个面向 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 选择器原生只有 XPathPyYAML读规则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-skumany: 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再stripfloat(¥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(encodingutf-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(retries3, timeout15): s requests.Session() adapter HTTPAdapter(max_retriesretries, pool_connections10, pool_maxsize10) 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, timeoutgetattr(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, encodingunicode) 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_asciiFalse) \n) def write_csv(rows, fp): if not rows: return writer csv.DictWriter(fp, fieldnameslist(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原样存盘用浏览器打开看。十次里有七八次打开一眼就明白了。别在没看原文的情况下改选择器那是纯猜。现象二中文全是ä¸æ–‡这种乱码英文正常。原因响应头没写 charsetrequests按 RFC 默认给ISO-8859-1resp.text就按单字节解了。解决就是第 4 章那三行判断encoding为空或者是iso-8859-1时切到apparent_encoding。更稳的做法是先resp.content拿到 bytes正则从meta charset...里读读不到再猜——多写五行能省掉一整类玄学问题。现象三前几十个页面很顺跑到一两百条开始大面积超时、429甚至直接连不上。原因并发开太大、没有间隔、每次请求新建连接。三项叠加目标站点的风控和你的本机端口都会受不了。解决一个任务复用一个Session连接池复用能省掉大量握手并发用信号量卡在 35别贪请求之间加 0.31 秒的随机间隔遇到 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(fCREATE TABLE IF NOT EXISTS {self.table} ({ddl})) sql fINSERT INTO {self.table} VALUES ({, .join(? * len(cols))}) self.conn.executemany(sql, [ [json.dumps(r.get(c), ensure_asciiFalse) 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确认字段命中率和改动前一致才允许它进生产任务。这套动作多花二十分钟换的是不用半夜爬起来查「为什么昨天还有数据今天全空」。希望帮到你。本文还有配套的精品资源点击获取