做开发的同学应该都有过这种尴尬时刻产品上线前想给项目找一个合适的域名翻遍注册商也找不到心仪的或者安全研究时需要快速检索一批历史域名但数据散落在多个 CSV、Excel 和数据库里根本没法统一搜索。这个时候一个能自己控制的“域名搜索引擎”就会变得非常有用。本文要拆解的就是一个非常典型的 maker 级后端实战项目用一个周末的时间以约 10 美元的成本搭建一个面向 50 万条域名数据的小型搜索引擎。整个项目不依赖重型中间件用 Python SQLite FTS5 一个轻量前端就能完成既能跑在本地学习也能部署到低配云服务器对外提供查询服务。1. 背景与核心概念1.1 什么是域名搜索引擎这里的“域名搜索引擎”不是指 Google、Bing 那种搜网页的通用搜索引擎而是指“以域名数据为索引对象”的专用检索系统。你可以把它理解成一个“域名卡片库”每条记录是一条域名例如example.com字段包括域名、顶级域、排名、长度、注册商信息等用户可以输入关键词快速查到这个域名的基本属性或者找到符合某种命名模式的域名。常见的应用场景包括产品初期做品牌命名根据输入词联想可参考域名对站点做资产盘点把一定范围内的子域名或业务域名统一索引安全研究场景下分析钓鱼域名、仿冒域名通过相似性搜索做初步研判数据运维人员维护内部业务域名清单快速定位某条域名是否注册、是否在库。本文最终构建出来的东西就是一个“域名数据的垂直搜索引擎”数据规模在 50 万级别单机即可胜任。1.2 为什么 50 万域名不需要上 Elasticsearch很多同学听到“搜索引擎”四个字第一反应就是上 Elasticsearch。但 Elasticsearch 本身是面向大规模全文检索的分布式系统部署、内存、运维成本都不小。如果只是 50 万条短文本数据用 SQLite 自带的 FTS5 全文索引已经绰绰有余。用几个数字来说明量级50 万条域名平均每个域名 20 个字符左右域名文本总量大约 10MB即使不加任何索引SQLite 全表扫描一次也只要几十毫秒到百毫秒级别加上 FTS5 全文索引后普通关键词查询通常可以在 10ms 左右返回。所以这个小场景完全没必要引入一个重型搜索引擎中间件。用 SQLite FTS5成本低、代码简单、备份方便。1.3 核心概念FTS5 与倒排索引SQLite FTS5 是 SQLite 扩展内置的全文搜索引擎模块它基于倒排索引实现把每个域名的内容拆成一个个 token然后记录每个 token 出现在哪些文档即哪一行数据里。当我们搜索一个关键词时FTS5 不需要逐行扫描而是直接去倒排索引里找包含该 token 的行号再回表取出完整数据所以速度很快。关于“token”不同分词规则会直接影响搜索结果。域名my-example.com如果按默认 unicode61 分词器切分会被拆成my、example、com三个 token。这样我们搜索example或example*时都能命中。这正好符合域名搜索的常见习惯。2. 环境准备与版本说明2.1 开发环境本项目面向低成本部署所以开发和运行环境都很轻量。推荐环境操作系统LinuxUbuntu 22.04 或 Debian 11/12Python3.10 及以上SQLite3.35 及以上自带 FTS5无需额外安装Web 框架Flask 或 FastAPI本文以 FastAPI 为例前端原生 HTML JavaScript不引入大型前端框架。如果你只是本地学习Windows/macOS 也能运行相同代码。SQLite 是文件型数据库跨平台兼容性很好。2.2 示例项目结构一个最小可运行项目可以这样组织domain-search/ ├── data/ │ └── top-500k.csv ├── scripts/ │ ├── download_data.sh │ ├── clean_domains.py │ ├── import_sqlite.py │ └── server.py ├── static/ │ ├── index.html │ ├── app.js │ └── style.css └── requirements.txt其中data/存放原始域名数据 CSVscripts/download_data.sh下载公开域名榜单scripts/clean_domains.py清洗域名scripts/import_sqlite.py导入 SQLite 并构建 FTS5 索引scripts/server.py提供查询 APIstatic/存放前端页面。后续每个文件都会给出可参考的完整代码。3. 数据获取与清洗3.1 域名数据从哪来构建域名搜索引擎的第一步是要有一批可用的域名数据。对于 50 万量级的个人项目通常不需要自己去全网爬域名优先使用公开的数据集公共域名排行榜 例如 Tranco、Majestic Million 等会公布每日排名靠前的域名列表。这类数据通常是 CSV 格式包含“排名、域名”两列下载后就能用。证书透明度日志 像 crt.sh 这类服务会公开 SSL/TLS 证书的签发记录里面包含大量域名和子域名。用它的 API 可以按关键字查询近期证书中的域名适合做子域名收集。Common Crawl 派生数据 Common Crawl 是公开的网页爬虫数据集里面保存了数十亿网页 URL。通过解析 URL 中的 Host 字段可以提取出大量域名。不过这种方式数据量大处理复杂适合已经具备一定数据工程能力的同学。这里必须强调无论使用哪个数据源都要先确认数据的许可协议和使用限制。个人研究和学习通常没问题但如果要商业化部署必须阅读数据源授权条款避免版权或服务条款风险。3.2 下载 Top 50 万域名下面先给出一段示例下载脚本。由于不同数据源 URL 会变化这里使用占位符 URL实际使用时请替换成你所用数据源的最新地址。#!/usr/bin/env bash # 文件scripts/download_data.sh # 说明下载公开域名榜单并保留前 50 万条 mkdir -p data # 请替换为实际的数据源下载链接 curl -L -o data/domain-list.csv https://example.invalid/top-1m.csv # 保留前 50 万行并输出统计信息 head -n 500000 data/domain-list.csv data/top-500k.csv echo 下载完成数据条数 wc -l data/top-500k.csv真实场景中下载后的 CSV 可能是rank,domain两列结构例如1,example.com 2,google.com 3,facebook.com也有一些数据源会包含更多列比如“机构名称、国家、分类”等。我们后续导入时会按列位置解析所以只要前两列符合“排名、域名”的顺序即可。3.3 域名清洗规则原始域名数据不等于可直接入库的数据必须做清洗。常见的清洗逻辑有转小写 域名本身不区分大小写统一转小写避免重复。去重 排行榜数据里可能因为跳转、子域名的原因出现重复记录。去掉协议头和路径 部分数据源可能把 URL 写成了https://www.example.com/path需要解析出纯域名。去掉末尾点 有些 DNS 场景会写成example.com.末尾点不是域名的一部分。过滤明显非域名数据 比如localhost、空行、IP 地址、包含空格的脏数据。下面是一个简单的清洗函数示例# 文件scripts/clean_domains.py from urllib.parse import urlparse def clean_domain(raw: str) - str | None: 清洗原始域名返回标准化域名无法处理的返回 None。 if not raw: return None raw raw.strip().lower() # 去掉协议头 if raw.startswith((http://, https://)): raw urlparse(raw).netloc # 去掉端口号 if : in raw: raw raw.split(:)[0] # 去掉末尾点 raw raw.rstrip(.) # 简单校验必须包含一个点并且不能是 IP if . not in raw: return None parts raw.split(.) if all(p.isdigit() for p in parts): return None return raw这里的重点是容错真实数据集里会有大量脏数据清洗函数宁可返回None也不能把错误数据写入数据库。4. 数据库设计与索引构建4.1 SQLite 表结构清洗后的域名我们可以设计两张表一对是主表domains保存域名的完整信息CREATE TABLE IF NOT EXISTS domains ( id INTEGER PRIMARY KEY AUTOINCREMENT, rank INTEGER, domain TEXT UNIQUE, name TEXT, tld TEXT, length INTEGER ); CREATE INDEX IF NOT EXISTS idx_domains_tld ON domains(tld);字段说明rank域名在榜单中的排名搜索排序时可以使用domain清洗后的完整域名name二级域名部分例如google.com的 name 是googletld顶级域例如com、org、cnlength域名长度方便做筛选。另一张是 FTS5 全文索引表CREATE VIRTUAL TABLE IF NOT EXISTS domain_fts USING fts5( domain, tld, content );content表示这是一张无内容表只保存 token 索引和 rowid。这样搜索时需要在查询语句中 JOIN 回主表。4.2 批量导入数据导入脚本需要完成两件事把清洗后的域名写入domains表把域名写入domain_fts表。两个表之间通过 rowid 建立对应关系domain_fts.rowid等于domains.id。下面给出一段完整导入代码# 文件scripts/import_sqlite.py import csv import sqlite3 import sys from clean_domains import clean_domain def split_domain(domain: str): parts domain.split(.) tld parts[-1] if len(parts) 1 else name ..join(parts[:-1]) if len(parts) 1 else domain return name, tld def import_domains(csv_path: str, db_path: str): conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA synchronousNORMAL;) conn.execute(BEGIN;) try: with open(csv_path, r, encodingutf-8, errorsignore) as f: reader csv.reader(f) # 跳过标题行 header next(reader, None) if header and header[0].lower() rank: pass for row in reader: if len(row) 2: continue rank row[0].strip() raw_domain row[1].strip() domain clean_domain(raw_domain) if not domain: continue name, tld split_domain(domain) # 插入主表 cur conn.execute( INSERT INTO domains(rank, domain, name, tld, length) VALUES (?, ?, ?, ?, ?) ON CONFLICT(domain) DO UPDATE SET rankexcluded.rank , (rank, domain, name, tld, len(domain)) ) # 插入 FTS5 索引表rowid 与 domains.id 对应 domain_id cur.lastrowid conn.execute( INSERT INTO domain_fts(rowid, domain, tld) VALUES (?, ?, ?), (domain_id, domain, tld) ) conn.commit() except Exception: conn.rollback() raise finally: conn.close() if __name__ __main__: import_domains(sys.argv[1], sys.argv[2]) print(导入完成)需要注意这里使用了ON CONFLICT(domain) DO UPDATE重复域名会更新排名而不会重复插入BEGIN和COMMIT包裹了批量事务避免每条数据单独提交带来的性能损耗PRAGMA journal_modeWAL可以大幅减少读写阻塞。4.3 验证索引数据导入完成后可以在命令行里快速验证sqlite3 data/domains.db然后执行SELECT COUNT(*) FROM domains; SELECT COUNT(*) FROM domain_fts;两个表的数量通常应该一致因为每条域名都对应一条 FTS 记录。再测试一下搜索SELECT d.domain FROM domain_fts f JOIN domains d ON d.id f.rowid WHERE domain_fts MATCH google* LIMIT 10;如果 FTS5 默认分词器把google.com拆成google、com那么搜google*就能命中。5. 搜索 API 与前端实战5.1 编写 FastAPI 搜索接口清洗数据和分析完表结构后下一步就是提供对外查询接口。这里以 FastAPI 为例因为它的代码结构清晰自带交互式文档适合快速搭建 API。先写一个依赖文件# 文件requirements.txt fastapi uvicorn再编写服务端代码# 文件scripts/server.py import sqlite3 import time from fastapi import FastAPI, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles DB_PATH data/domains.db app FastAPI(titleDomain Search API) # 允许本地前端跨域调用 app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA query_onlyON;) return conn app.get(/api/search) def search( q: str Query(..., min_length1, max_length100), limit: int Query(20, ge1, le100), ): if not q: return {error: empty keyword} start time.time() conn get_conn() # 转义 FTS5 特殊字符 safe_q q.strip().replace(, ) # 优先使用前缀匹配搜索速度更快 sql SELECT d.rank, d.domain, d.tld, d.name, d.length FROM domain_fts f JOIN domains d ON d.id f.rowid WHERE domain_fts MATCH ? ORDER BY d.rank ASC LIMIT ? # 这里构造 FTS 查询词 google* 表示前缀搜索 fts_query f{safe_q}* rows conn.execute(sql, (fts_query, limit)).fetchall() conn.close() results [dict(row) for row in rows] cost_ms int((time.time() - start) * 1000) return { query: q, total: len(results), time_ms: cost_ms, results: results, } # 挂载静态文件用于展示前端页面 app.mount(/, StaticFiles(directorystatic, htmlTrue), namestatic)需要留意几个细节FTS5 的MATCH查询语法和我们平时用的 SQL 不太一样字段名不能直接拼字符串否则会语法报错这里将用户输入强制构造成前缀搜索关键词*效率和可用性比较平衡生产环境不能无条件开放allow_origins[*]只适合本地开发和临时演示。5.2 编写前端页面前端只需要一个输入框、一个结果列表和一小段 JavaScript。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title域名搜索引擎/title style body { font-family: Arial, sans-serif; max-width: 800px; margin: 40px auto; padding: 0 16px; } input { width: 100%; padding: 10px; font-size: 16px; box-sizing: border-box; } ul { list-style: none; padding: 0; } li { padding: 12px; border-bottom: 1px solid #eee; } .domain { font-weight: bold; } .meta { color: #888; font-size: 14px; } /style /head body h1域名搜索引擎/h1 input typetext idsearchInput placeholder输入关键词例如google ul idresultList/ul script const input document.getElementById(searchInput); const list document.getElementById(resultList); let timer null; input.addEventListener(input, function () { clearTimeout(timer); timer setTimeout(doSearch, 200); }); async function doSearch() { const q input.value.trim(); list.innerHTML ; if (!q) return; const resp await fetch(/api/search?q${encodeURIComponent(q)}limit20); const data await resp.json(); if (!data.results) return; for (const item of data.results) { const li document.createElement(li); li.innerHTML div classdomain${item.domain}/div div classmeta排名 ${item.rank} · ${item.tld} · 长度 ${item.length}/div ; list.appendChild(li); } } /script /body /html前端逻辑很简单输入关键词后延迟 200ms 再请求避免每敲一个字符就触发一次查询使用fetch请求本地/api/search接口遍历结果列表把域名和元信息渲染到页面上。5.3 启动与测试在项目根目录下执行pip install -r requirements.txt python scripts/import_sqlite.py data/top-500k.csv data/domains.db uvicorn scripts.server:app --host 0.0.0.0 --port 8000启动后打开浏览器访问http://localhost:8000输入google即可看到结果。也可以用 curl 测试 APIcurl http://localhost:8000/api/search?qgooglelimit5预期返回类似{ query: google, total: 1, time_ms: 12, results: [ { rank: 2, domain: google.com, tld: com, name: google, length: 10 } ] }至此一个最小可用的域名搜索引擎已经跑通了。6. 常见问题与排查思路6.1 请求域名查询接口时提示{code:1004,error:domain forbidden}这个报错在一些公开 API 中经常出现。它并不是 SQLite 报错而是后端服务对请求方的一种“拒绝访问”响应。可能原因请求频率过高触发了限流策略当前服务器 IP 不在数据源允许的调用范围内缺少必要的请求头或 Token被查询的域名本身被数据源标记为禁止访问。排查步骤查看 API 文档确认是否需要携带 API Key检查请求的频率是否过高增加间隔时间尝试换一个合法、未被标记的测试域名在日志中记录请求头和响应体确认是业务层拒绝还是网络层阻断。从工程角度来说批量查询域名时一定要对自己抓到的错误做分级。domain forbidden这类错误不应该直接让整个导入任务崩溃而是记录日志、跳过、稍后重试。6.2 DNS 查询出现non-existent domain如果我们的域名数据集中包含一些内网、测试或已失效域名在后续做批量解析时可能看到这样的错误*** dc4.wsytot.cn 找不到 _ldap._tcp.dc._msdsc.wsytot.cn: non-existent domain这通常表示该域名没有对应的 AD DC SRV 记录或者域名已经不再解析到任何 DNS 服务器或者查询的是内网域公网 DNS 根本无法解析。解决方案import dns.resolver def query_srv_with_timeout(domain: str, timeout: float 2.0): try: answers dns.resolver.resolve(f_ldap._tcp.dc._msdsc.{domain}, SRV, lifetimetimeout) return [str(r) for r in answers] except dns.resolver.NXDOMAIN: # 域名不存在 SRV 记录属于正常情况 return [] except dns.resolver.NoAnswer: return [] except dns.resolver.LifetimeTimeout: # 超时可能是 DNS 服务不可达需要重试 return None这里的关键思路是NXDOMAIN和NoAnswer都应该被当作“没有该记录”处理而不是异常。超时则需要重试不能无限等待。6.3 SQLite 报错database is locked出现这个问题的场景往往是在写库的同时有查询请求。解决思路打开 WAL 模式PRAGMA journal_modeWAL;设置合理的busy_timeoutconn.execute(PRAGMA busy_timeout5000;)查询用只读连接conn.execute(PRAGMA query_onlyON;)如果写库频率很高可以单独把导入过程放到一台离线环境完成导出数据库文件后再上线替换避免读写竞争。6.4 FTS5 搜索结果不完整很多域名搜不到FTS5 默认按 token 匹配不是“任意子串匹配”。举例输入google可以命中google.com输入oogle无法命中google.com。如果希望支持任意子串搜索可以增加一个LIKE接口作为备选SELECT * FROM domains WHERE domain LIKE %oogle% ORDER BY rank LIMIT 20;但要注意LIKE %keyword%会触发全表扫描50 万行数据量下性能并不稳定。实际项目中更合理的做法是短关键词默认走 FTS5 前缀搜索只有明确需要模糊搜索时才走 LIKE并且限制limit。6.5 内存不够导致服务崩溃低配 VPS 内存通常只有 1-2GB。如果导数据时使用 pandas 一次性读入 50 万行内存可能瞬间飙升。更稳妥的做法是使用流式读取 CSV逐行写入 SQLite避免把所有数据加载到内存中。本文导入脚本用的就是逐行读取方案。另外FastAPI 的开发模式--reload会额外消耗内存生产部署时建议关闭。6.6 前端页面 404 或者跨域报错如果前端直接双击 HTML 打开fetch(/api/search)会因为文件路径和 API 不在同一个源而报错。解决办法把index.html放到服务端static目录使用uvicorn启动后访问或者给 FastAPI 配置 CORS 中间件不要在生产环境长期使用allow_origins[*]。7. 最佳实践与工程建议7.1 数据更新策略域名数据不是一成不变的排行榜每天都会变化。建议建立定期更新机制。更新流程可以这样设计每天凌晨下载最新榜单只保留原有 50 万条中的新增或热度变化明显的记录清除已经被删除的域名记录重建 FTS5 索引。对于简单项目重建整个数据库更加可靠。数据量只有 50 万重建耗时通常几秒到几十秒完全可接受mv data/domains.db data/domains.db.bak python scripts/import_sqlite.py data/top-500k.csv data/domains.db更新前备份旧库更新失败时可以快速回滚。7.2 安全与合规使用域名数据时有几个底线需要注意只使用已获得授权或允许公开使用的数据源不要采集 WHOIS 中的个人注册者隐私字段做展示如果需要对外提供 API必须加访问限制或限流避免被恶意抓取涉及安全研究的域名收集只允许在合法授权的范围内测试不要在日志里记录用户的搜索关键词如果必须记录要做好脱敏和访问控制。7.3 搜索性能优化50 万条数据是基础量级但如果后续扩展到 500 万条性能优化点会非常明显尽量使用前缀搜索MATCH google*比MATCH google命中范围更可控但索引查询速度都很快给domains(rank)加索引 排序字段单独建索引可以减少 ORDER BY 的临时排序开销对常用查询加缓存 可以用 Redis 或者简单的进程内 LRU 缓存搜索热词直接走缓存如果数据量继续增长再迁移到 Meilisearch、Typesense 或 Elasticsearch。7.4 部署与监控生产环境部署建议用 systemd 托管服务并带日志输出# /etc/systemd/system/domain-search.service [Unit] DescriptionDomain Search API Afternetwork.target [Service] Userwww-data WorkingDirectory/opt/domain-search ExecStart/usr/local/bin/uvicorn scripts.server:app --host 0.0.0.0 --port 8000 --workers 2 Restartalways [Install] WantedBymulti-user.target然后执行sudo systemctl daemon-reload sudo systemctl enable domain-search sudo systemctl start domain-search启动后务必观察两个指标查询接口的平均响应时间内存占用是否持续增长。如果内存持续上涨优先检查是不是 FastAPI 的全局变量缓存、日志收集或者连接池没有被正确释放。7.5 进一步学习路线完成这个项目后可以沿着下面的方向继续深入换成 Meilisearch 或 Typesense做一个真正的即搜即得搜索框体验全文搜索引擎的配置细节引入更多字段比如网站标题、描述、ICP 备案号做更丰富的“域名信息搜索”加入自动更新任务把每日榜单变化做成增量同步增加导出接口方便批量导出搜索结果用于其他系统如果对排序有要求可以引入归一化分数综合域名长度、排名、关键词命中位置做自定义排序。动手实现一个 50 万级别的域名搜索引擎门槛并不高。真正有价值的部分是后面的数据清洗、索引设计、搜索策略调整和低成本部署方案。这些经验放到其他小型搜索项目里也完全通用。如果你正想找一个练手项目不妨从今天开始先把公开榜单下载下来跑通导入脚本再写一个简单的搜索页面。很快你就会发现很多看似需要复杂搜索引擎的场景用 SQLite FTS5 已经可以解决大半。 用SQLite FTS5构建10美元成本的50万级域名搜索引擎
做开发的同学应该都有过这种尴尬时刻产品上线前想给项目找一个合适的域名翻遍注册商也找不到心仪的或者安全研究时需要快速检索一批历史域名但数据散落在多个 CSV、Excel 和数据库里根本没法统一搜索。这个时候一个能自己控制的“域名搜索引擎”就会变得非常有用。本文要拆解的就是一个非常典型的 maker 级后端实战项目用一个周末的时间以约 10 美元的成本搭建一个面向 50 万条域名数据的小型搜索引擎。整个项目不依赖重型中间件用 Python SQLite FTS5 一个轻量前端就能完成既能跑在本地学习也能部署到低配云服务器对外提供查询服务。1. 背景与核心概念1.1 什么是域名搜索引擎这里的“域名搜索引擎”不是指 Google、Bing 那种搜网页的通用搜索引擎而是指“以域名数据为索引对象”的专用检索系统。你可以把它理解成一个“域名卡片库”每条记录是一条域名例如example.com字段包括域名、顶级域、排名、长度、注册商信息等用户可以输入关键词快速查到这个域名的基本属性或者找到符合某种命名模式的域名。常见的应用场景包括产品初期做品牌命名根据输入词联想可参考域名对站点做资产盘点把一定范围内的子域名或业务域名统一索引安全研究场景下分析钓鱼域名、仿冒域名通过相似性搜索做初步研判数据运维人员维护内部业务域名清单快速定位某条域名是否注册、是否在库。本文最终构建出来的东西就是一个“域名数据的垂直搜索引擎”数据规模在 50 万级别单机即可胜任。1.2 为什么 50 万域名不需要上 Elasticsearch很多同学听到“搜索引擎”四个字第一反应就是上 Elasticsearch。但 Elasticsearch 本身是面向大规模全文检索的分布式系统部署、内存、运维成本都不小。如果只是 50 万条短文本数据用 SQLite 自带的 FTS5 全文索引已经绰绰有余。用几个数字来说明量级50 万条域名平均每个域名 20 个字符左右域名文本总量大约 10MB即使不加任何索引SQLite 全表扫描一次也只要几十毫秒到百毫秒级别加上 FTS5 全文索引后普通关键词查询通常可以在 10ms 左右返回。所以这个小场景完全没必要引入一个重型搜索引擎中间件。用 SQLite FTS5成本低、代码简单、备份方便。1.3 核心概念FTS5 与倒排索引SQLite FTS5 是 SQLite 扩展内置的全文搜索引擎模块它基于倒排索引实现把每个域名的内容拆成一个个 token然后记录每个 token 出现在哪些文档即哪一行数据里。当我们搜索一个关键词时FTS5 不需要逐行扫描而是直接去倒排索引里找包含该 token 的行号再回表取出完整数据所以速度很快。关于“token”不同分词规则会直接影响搜索结果。域名my-example.com如果按默认 unicode61 分词器切分会被拆成my、example、com三个 token。这样我们搜索example或example*时都能命中。这正好符合域名搜索的常见习惯。2. 环境准备与版本说明2.1 开发环境本项目面向低成本部署所以开发和运行环境都很轻量。推荐环境操作系统LinuxUbuntu 22.04 或 Debian 11/12Python3.10 及以上SQLite3.35 及以上自带 FTS5无需额外安装Web 框架Flask 或 FastAPI本文以 FastAPI 为例前端原生 HTML JavaScript不引入大型前端框架。如果你只是本地学习Windows/macOS 也能运行相同代码。SQLite 是文件型数据库跨平台兼容性很好。2.2 示例项目结构一个最小可运行项目可以这样组织domain-search/ ├── data/ │ └── top-500k.csv ├── scripts/ │ ├── download_data.sh │ ├── clean_domains.py │ ├── import_sqlite.py │ └── server.py ├── static/ │ ├── index.html │ ├── app.js │ └── style.css └── requirements.txt其中data/存放原始域名数据 CSVscripts/download_data.sh下载公开域名榜单scripts/clean_domains.py清洗域名scripts/import_sqlite.py导入 SQLite 并构建 FTS5 索引scripts/server.py提供查询 APIstatic/存放前端页面。后续每个文件都会给出可参考的完整代码。3. 数据获取与清洗3.1 域名数据从哪来构建域名搜索引擎的第一步是要有一批可用的域名数据。对于 50 万量级的个人项目通常不需要自己去全网爬域名优先使用公开的数据集公共域名排行榜 例如 Tranco、Majestic Million 等会公布每日排名靠前的域名列表。这类数据通常是 CSV 格式包含“排名、域名”两列下载后就能用。证书透明度日志 像 crt.sh 这类服务会公开 SSL/TLS 证书的签发记录里面包含大量域名和子域名。用它的 API 可以按关键字查询近期证书中的域名适合做子域名收集。Common Crawl 派生数据 Common Crawl 是公开的网页爬虫数据集里面保存了数十亿网页 URL。通过解析 URL 中的 Host 字段可以提取出大量域名。不过这种方式数据量大处理复杂适合已经具备一定数据工程能力的同学。这里必须强调无论使用哪个数据源都要先确认数据的许可协议和使用限制。个人研究和学习通常没问题但如果要商业化部署必须阅读数据源授权条款避免版权或服务条款风险。3.2 下载 Top 50 万域名下面先给出一段示例下载脚本。由于不同数据源 URL 会变化这里使用占位符 URL实际使用时请替换成你所用数据源的最新地址。#!/usr/bin/env bash # 文件scripts/download_data.sh # 说明下载公开域名榜单并保留前 50 万条 mkdir -p data # 请替换为实际的数据源下载链接 curl -L -o data/domain-list.csv https://example.invalid/top-1m.csv # 保留前 50 万行并输出统计信息 head -n 500000 data/domain-list.csv data/top-500k.csv echo 下载完成数据条数 wc -l data/top-500k.csv真实场景中下载后的 CSV 可能是rank,domain两列结构例如1,example.com 2,google.com 3,facebook.com也有一些数据源会包含更多列比如“机构名称、国家、分类”等。我们后续导入时会按列位置解析所以只要前两列符合“排名、域名”的顺序即可。3.3 域名清洗规则原始域名数据不等于可直接入库的数据必须做清洗。常见的清洗逻辑有转小写 域名本身不区分大小写统一转小写避免重复。去重 排行榜数据里可能因为跳转、子域名的原因出现重复记录。去掉协议头和路径 部分数据源可能把 URL 写成了https://www.example.com/path需要解析出纯域名。去掉末尾点 有些 DNS 场景会写成example.com.末尾点不是域名的一部分。过滤明显非域名数据 比如localhost、空行、IP 地址、包含空格的脏数据。下面是一个简单的清洗函数示例# 文件scripts/clean_domains.py from urllib.parse import urlparse def clean_domain(raw: str) - str | None: 清洗原始域名返回标准化域名无法处理的返回 None。 if not raw: return None raw raw.strip().lower() # 去掉协议头 if raw.startswith((http://, https://)): raw urlparse(raw).netloc # 去掉端口号 if : in raw: raw raw.split(:)[0] # 去掉末尾点 raw raw.rstrip(.) # 简单校验必须包含一个点并且不能是 IP if . not in raw: return None parts raw.split(.) if all(p.isdigit() for p in parts): return None return raw这里的重点是容错真实数据集里会有大量脏数据清洗函数宁可返回None也不能把错误数据写入数据库。4. 数据库设计与索引构建4.1 SQLite 表结构清洗后的域名我们可以设计两张表一对是主表domains保存域名的完整信息CREATE TABLE IF NOT EXISTS domains ( id INTEGER PRIMARY KEY AUTOINCREMENT, rank INTEGER, domain TEXT UNIQUE, name TEXT, tld TEXT, length INTEGER ); CREATE INDEX IF NOT EXISTS idx_domains_tld ON domains(tld);字段说明rank域名在榜单中的排名搜索排序时可以使用domain清洗后的完整域名name二级域名部分例如google.com的 name 是googletld顶级域例如com、org、cnlength域名长度方便做筛选。另一张是 FTS5 全文索引表CREATE VIRTUAL TABLE IF NOT EXISTS domain_fts USING fts5( domain, tld, content );content表示这是一张无内容表只保存 token 索引和 rowid。这样搜索时需要在查询语句中 JOIN 回主表。4.2 批量导入数据导入脚本需要完成两件事把清洗后的域名写入domains表把域名写入domain_fts表。两个表之间通过 rowid 建立对应关系domain_fts.rowid等于domains.id。下面给出一段完整导入代码# 文件scripts/import_sqlite.py import csv import sqlite3 import sys from clean_domains import clean_domain def split_domain(domain: str): parts domain.split(.) tld parts[-1] if len(parts) 1 else name ..join(parts[:-1]) if len(parts) 1 else domain return name, tld def import_domains(csv_path: str, db_path: str): conn sqlite3.connect(db_path) conn.execute(PRAGMA journal_modeWAL;) conn.execute(PRAGMA synchronousNORMAL;) conn.execute(BEGIN;) try: with open(csv_path, r, encodingutf-8, errorsignore) as f: reader csv.reader(f) # 跳过标题行 header next(reader, None) if header and header[0].lower() rank: pass for row in reader: if len(row) 2: continue rank row[0].strip() raw_domain row[1].strip() domain clean_domain(raw_domain) if not domain: continue name, tld split_domain(domain) # 插入主表 cur conn.execute( INSERT INTO domains(rank, domain, name, tld, length) VALUES (?, ?, ?, ?, ?) ON CONFLICT(domain) DO UPDATE SET rankexcluded.rank , (rank, domain, name, tld, len(domain)) ) # 插入 FTS5 索引表rowid 与 domains.id 对应 domain_id cur.lastrowid conn.execute( INSERT INTO domain_fts(rowid, domain, tld) VALUES (?, ?, ?), (domain_id, domain, tld) ) conn.commit() except Exception: conn.rollback() raise finally: conn.close() if __name__ __main__: import_domains(sys.argv[1], sys.argv[2]) print(导入完成)需要注意这里使用了ON CONFLICT(domain) DO UPDATE重复域名会更新排名而不会重复插入BEGIN和COMMIT包裹了批量事务避免每条数据单独提交带来的性能损耗PRAGMA journal_modeWAL可以大幅减少读写阻塞。4.3 验证索引数据导入完成后可以在命令行里快速验证sqlite3 data/domains.db然后执行SELECT COUNT(*) FROM domains; SELECT COUNT(*) FROM domain_fts;两个表的数量通常应该一致因为每条域名都对应一条 FTS 记录。再测试一下搜索SELECT d.domain FROM domain_fts f JOIN domains d ON d.id f.rowid WHERE domain_fts MATCH google* LIMIT 10;如果 FTS5 默认分词器把google.com拆成google、com那么搜google*就能命中。5. 搜索 API 与前端实战5.1 编写 FastAPI 搜索接口清洗数据和分析完表结构后下一步就是提供对外查询接口。这里以 FastAPI 为例因为它的代码结构清晰自带交互式文档适合快速搭建 API。先写一个依赖文件# 文件requirements.txt fastapi uvicorn再编写服务端代码# 文件scripts/server.py import sqlite3 import time from fastapi import FastAPI, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles DB_PATH data/domains.db app FastAPI(titleDomain Search API) # 允许本地前端跨域调用 app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) def get_conn(): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row conn.execute(PRAGMA query_onlyON;) return conn app.get(/api/search) def search( q: str Query(..., min_length1, max_length100), limit: int Query(20, ge1, le100), ): if not q: return {error: empty keyword} start time.time() conn get_conn() # 转义 FTS5 特殊字符 safe_q q.strip().replace(, ) # 优先使用前缀匹配搜索速度更快 sql SELECT d.rank, d.domain, d.tld, d.name, d.length FROM domain_fts f JOIN domains d ON d.id f.rowid WHERE domain_fts MATCH ? ORDER BY d.rank ASC LIMIT ? # 这里构造 FTS 查询词 google* 表示前缀搜索 fts_query f{safe_q}* rows conn.execute(sql, (fts_query, limit)).fetchall() conn.close() results [dict(row) for row in rows] cost_ms int((time.time() - start) * 1000) return { query: q, total: len(results), time_ms: cost_ms, results: results, } # 挂载静态文件用于展示前端页面 app.mount(/, StaticFiles(directorystatic, htmlTrue), namestatic)需要留意几个细节FTS5 的MATCH查询语法和我们平时用的 SQL 不太一样字段名不能直接拼字符串否则会语法报错这里将用户输入强制构造成前缀搜索关键词*效率和可用性比较平衡生产环境不能无条件开放allow_origins[*]只适合本地开发和临时演示。5.2 编写前端页面前端只需要一个输入框、一个结果列表和一小段 JavaScript。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title域名搜索引擎/title style body { font-family: Arial, sans-serif; max-width: 800px; margin: 40px auto; padding: 0 16px; } input { width: 100%; padding: 10px; font-size: 16px; box-sizing: border-box; } ul { list-style: none; padding: 0; } li { padding: 12px; border-bottom: 1px solid #eee; } .domain { font-weight: bold; } .meta { color: #888; font-size: 14px; } /style /head body h1域名搜索引擎/h1 input typetext idsearchInput placeholder输入关键词例如google ul idresultList/ul script const input document.getElementById(searchInput); const list document.getElementById(resultList); let timer null; input.addEventListener(input, function () { clearTimeout(timer); timer setTimeout(doSearch, 200); }); async function doSearch() { const q input.value.trim(); list.innerHTML ; if (!q) return; const resp await fetch(/api/search?q${encodeURIComponent(q)}limit20); const data await resp.json(); if (!data.results) return; for (const item of data.results) { const li document.createElement(li); li.innerHTML div classdomain${item.domain}/div div classmeta排名 ${item.rank} · ${item.tld} · 长度 ${item.length}/div ; list.appendChild(li); } } /script /body /html前端逻辑很简单输入关键词后延迟 200ms 再请求避免每敲一个字符就触发一次查询使用fetch请求本地/api/search接口遍历结果列表把域名和元信息渲染到页面上。5.3 启动与测试在项目根目录下执行pip install -r requirements.txt python scripts/import_sqlite.py data/top-500k.csv data/domains.db uvicorn scripts.server:app --host 0.0.0.0 --port 8000启动后打开浏览器访问http://localhost:8000输入google即可看到结果。也可以用 curl 测试 APIcurl http://localhost:8000/api/search?qgooglelimit5预期返回类似{ query: google, total: 1, time_ms: 12, results: [ { rank: 2, domain: google.com, tld: com, name: google, length: 10 } ] }至此一个最小可用的域名搜索引擎已经跑通了。6. 常见问题与排查思路6.1 请求域名查询接口时提示{code:1004,error:domain forbidden}这个报错在一些公开 API 中经常出现。它并不是 SQLite 报错而是后端服务对请求方的一种“拒绝访问”响应。可能原因请求频率过高触发了限流策略当前服务器 IP 不在数据源允许的调用范围内缺少必要的请求头或 Token被查询的域名本身被数据源标记为禁止访问。排查步骤查看 API 文档确认是否需要携带 API Key检查请求的频率是否过高增加间隔时间尝试换一个合法、未被标记的测试域名在日志中记录请求头和响应体确认是业务层拒绝还是网络层阻断。从工程角度来说批量查询域名时一定要对自己抓到的错误做分级。domain forbidden这类错误不应该直接让整个导入任务崩溃而是记录日志、跳过、稍后重试。6.2 DNS 查询出现non-existent domain如果我们的域名数据集中包含一些内网、测试或已失效域名在后续做批量解析时可能看到这样的错误*** dc4.wsytot.cn 找不到 _ldap._tcp.dc._msdsc.wsytot.cn: non-existent domain这通常表示该域名没有对应的 AD DC SRV 记录或者域名已经不再解析到任何 DNS 服务器或者查询的是内网域公网 DNS 根本无法解析。解决方案import dns.resolver def query_srv_with_timeout(domain: str, timeout: float 2.0): try: answers dns.resolver.resolve(f_ldap._tcp.dc._msdsc.{domain}, SRV, lifetimetimeout) return [str(r) for r in answers] except dns.resolver.NXDOMAIN: # 域名不存在 SRV 记录属于正常情况 return [] except dns.resolver.NoAnswer: return [] except dns.resolver.LifetimeTimeout: # 超时可能是 DNS 服务不可达需要重试 return None这里的关键思路是NXDOMAIN和NoAnswer都应该被当作“没有该记录”处理而不是异常。超时则需要重试不能无限等待。6.3 SQLite 报错database is locked出现这个问题的场景往往是在写库的同时有查询请求。解决思路打开 WAL 模式PRAGMA journal_modeWAL;设置合理的busy_timeoutconn.execute(PRAGMA busy_timeout5000;)查询用只读连接conn.execute(PRAGMA query_onlyON;)如果写库频率很高可以单独把导入过程放到一台离线环境完成导出数据库文件后再上线替换避免读写竞争。6.4 FTS5 搜索结果不完整很多域名搜不到FTS5 默认按 token 匹配不是“任意子串匹配”。举例输入google可以命中google.com输入oogle无法命中google.com。如果希望支持任意子串搜索可以增加一个LIKE接口作为备选SELECT * FROM domains WHERE domain LIKE %oogle% ORDER BY rank LIMIT 20;但要注意LIKE %keyword%会触发全表扫描50 万行数据量下性能并不稳定。实际项目中更合理的做法是短关键词默认走 FTS5 前缀搜索只有明确需要模糊搜索时才走 LIKE并且限制limit。6.5 内存不够导致服务崩溃低配 VPS 内存通常只有 1-2GB。如果导数据时使用 pandas 一次性读入 50 万行内存可能瞬间飙升。更稳妥的做法是使用流式读取 CSV逐行写入 SQLite避免把所有数据加载到内存中。本文导入脚本用的就是逐行读取方案。另外FastAPI 的开发模式--reload会额外消耗内存生产部署时建议关闭。6.6 前端页面 404 或者跨域报错如果前端直接双击 HTML 打开fetch(/api/search)会因为文件路径和 API 不在同一个源而报错。解决办法把index.html放到服务端static目录使用uvicorn启动后访问或者给 FastAPI 配置 CORS 中间件不要在生产环境长期使用allow_origins[*]。7. 最佳实践与工程建议7.1 数据更新策略域名数据不是一成不变的排行榜每天都会变化。建议建立定期更新机制。更新流程可以这样设计每天凌晨下载最新榜单只保留原有 50 万条中的新增或热度变化明显的记录清除已经被删除的域名记录重建 FTS5 索引。对于简单项目重建整个数据库更加可靠。数据量只有 50 万重建耗时通常几秒到几十秒完全可接受mv data/domains.db data/domains.db.bak python scripts/import_sqlite.py data/top-500k.csv data/domains.db更新前备份旧库更新失败时可以快速回滚。7.2 安全与合规使用域名数据时有几个底线需要注意只使用已获得授权或允许公开使用的数据源不要采集 WHOIS 中的个人注册者隐私字段做展示如果需要对外提供 API必须加访问限制或限流避免被恶意抓取涉及安全研究的域名收集只允许在合法授权的范围内测试不要在日志里记录用户的搜索关键词如果必须记录要做好脱敏和访问控制。7.3 搜索性能优化50 万条数据是基础量级但如果后续扩展到 500 万条性能优化点会非常明显尽量使用前缀搜索MATCH google*比MATCH google命中范围更可控但索引查询速度都很快给domains(rank)加索引 排序字段单独建索引可以减少 ORDER BY 的临时排序开销对常用查询加缓存 可以用 Redis 或者简单的进程内 LRU 缓存搜索热词直接走缓存如果数据量继续增长再迁移到 Meilisearch、Typesense 或 Elasticsearch。7.4 部署与监控生产环境部署建议用 systemd 托管服务并带日志输出# /etc/systemd/system/domain-search.service [Unit] DescriptionDomain Search API Afternetwork.target [Service] Userwww-data WorkingDirectory/opt/domain-search ExecStart/usr/local/bin/uvicorn scripts.server:app --host 0.0.0.0 --port 8000 --workers 2 Restartalways [Install] WantedBymulti-user.target然后执行sudo systemctl daemon-reload sudo systemctl enable domain-search sudo systemctl start domain-search启动后务必观察两个指标查询接口的平均响应时间内存占用是否持续增长。如果内存持续上涨优先检查是不是 FastAPI 的全局变量缓存、日志收集或者连接池没有被正确释放。7.5 进一步学习路线完成这个项目后可以沿着下面的方向继续深入换成 Meilisearch 或 Typesense做一个真正的即搜即得搜索框体验全文搜索引擎的配置细节引入更多字段比如网站标题、描述、ICP 备案号做更丰富的“域名信息搜索”加入自动更新任务把每日榜单变化做成增量同步增加导出接口方便批量导出搜索结果用于其他系统如果对排序有要求可以引入归一化分数综合域名长度、排名、关键词命中位置做自定义排序。动手实现一个 50 万级别的域名搜索引擎门槛并不高。真正有价值的部分是后面的数据清洗、索引设计、搜索策略调整和低成本部署方案。这些经验放到其他小型搜索项目里也完全通用。如果你正想找一个练手项目不妨从今天开始先把公开榜单下载下来跑通导入脚本再写一个简单的搜索页面。很快你就会发现很多看似需要复杂搜索引擎的场景用 SQLite FTS5 已经可以解决大半。