针对客户列表的批量 DR,作为 40 行 Python 作业
你有一个域名电子表格和一个需要每行一个数字的控制台。针对该任务的反向链接域名检查器主要是管道工作,而这个脚本就是管道。

脚本
它读取 domains.txt,每次请求发送 200 个域名,将每个原始响应保留在 SQLite 中,并打印运行成本。我们存储整个负载,因为字段名最好从真实响应中复制,而不是从博客文章中复制。
import os, sqlite3, time, json, requests
BASE = "https://seomasters.dev/api.php"
KEY = os.environ["SEO_API_KEY"]
BATCH = 200
db = sqlite3.connect("dr.db")
db.execute("create table if not exists raw (ts int, batch text, body text)")
domains = [d.strip() for d in open("domains.txt") if d.strip()]
spent = 0
for i in range(0, len(domains), BATCH):
chunk = domains[i:i + BATCH]
for attempt in range(4):
try:
r = requests.get(BASE, timeout=30,
params={"action": "bulk_dr", "domains": ",".join(chunk)},
headers={"Authorization": "Bearer " + KEY})
break
except requests.RequestException:
time.sleep(2 ** attempt)
else:
raise SystemExit(f"network gave up at batch {i}")
if r.status_code in (401, 402):
raise SystemExit(f"HTTP {r.status_code}: fix the key or top up, do not retry")
body = r.json()
if not body.get("ok"):
raise SystemExit(f"ok=false at batch {i}: {json.dumps(body)[:300]}")
db.execute("insert into raw values (?,?,?)", (int(time.time()), str(i), r.text))
db.commit()
print(body.get("_credits"))各部分的功能
action=bulk_dr
每次调用消耗 1 个信用额度。文档规定的上限为每次调用 2,000 个域名。
domains=
这是一个 GET 参数。该端点也接受 POST 正文,当列表变长时这是更好的选择。在完整运行之前,用两个域名测试分隔符。
BATCH = 200
我们的习惯,而非限制:200 个短域名使 URL 远低于常见代理限制。
_credits
出现在经过身份验证的响应中:调用的成本和剩余量。每批打印它是成本最低的审计跟踪。
原始 body 列
稍后解析,一次针对存储的样本。如果我们的架构发生变化,你只需运行查询,而无需重新运行整个列表。

假设的运行,包含算术
这是一个示例,而非客户故事。想象一家机构在 Google Sheet 中跟踪 1,800 个客户和潜在客户域名,并希望每周一刷新 DR。
每批 200 个域名,即 9 次请求。9 次请求,每次 1 个信用额度,共 9 个信用额度,即每次运行 $0.0009,在 52 个周一内总计 468 个信用额度,四舍五入后不到五美分。
成本不是限制因素。新鲜度才是。索引大约每月重建一次,因此每周运行在四周中的三周会显示相同的值,而每月运行捕获的变动量约为调用量的四分之一。
因此,每月运行一次,并保留每周的电子表格供人工编辑。分数停滞不前的原因详情见 DR 参考。
此脚本中可能出错的事项
| 症状 | 原因 | 修复 |
|---|---|---|
SystemExit: HTTP 402 | 余额为零 | 从账户页面充值,重新运行;已完成的批次已存储在 SQLite 中 |
SystemExit: HTTP 401 | 环境变量为空或密钥拼写错误 | 打印 len(KEY) 并检查前缀 sk_ |
第 3 批 ok=false | 列表中的值错误,例如完整 URL | 发送前剥离方案和路径,参见 错误表 |
| 与上周相同的 DR | 索引尚未重建 | 预期;按月比较 |
| 长 URL 被拒绝 | 批次对于 GET 来说太大 | 降低 BATCH 或移至 POST 正文 |
长列表的批次大小、GET 和 POST
上限为每次调用 2,000 个域名,而脚本发送其中的十分之一。每次调用成本相同的固定信用额度,因此批次大小是关于失败和负载的问题,而非价格。
为什么是 200 而不是 2,000
较小的批次失败成本更低。如果运行中途停止,需要重做的尾部长度为一个批次,而非整个列表。
较短的列表也使每个保存的正文足够小,以便在数字看起来错误时肉眼扫描,这比节省几次调用更重要。
如果后来增加批次大小,请一次性在一个常量中完成,并重新运行双域名测试。更大的数字不是跳过测试的理由。
根据你能容忍的运行长度来规划批次,而不是根据上限。必须在晨间报告之前完成的夜间作业可以承受更小、更安全的批次,而一次性回填则不然。
短列表使用 GET,长列表使用 POST
GET 形式将每个域名放入 URL。这在代理或日志行截断查询之前有效,这就是为什么上面的表说先降低批次。一旦列表变长,在 POST 正文中发送相同的 domains 值。
在完整运行之前,用两个域名测试分隔符和编码。一个多余的逗号会在第一批中出现,在那里看到它成本更低。
POST 正文还将列表从许多主机默认保留的访问日志中排除。这对于客户列表是一个小的隐私增益,并且在脚本中增加了一行额外的标头。
当你切换到 POST 时,保留 GET 版本的脚本作为短列表的备用。两个代码路径听起来很重,但一个根据长度选择其一的小函数易于阅读和测试。
在第一次请求之前修剪和去重
在数据离开你的机器之前,剥离空白,将主机小写并删除重复项。重复项会占用 2,000 个域名上限下的空间,并使你的电子表格中的计数与响应中的计数不一致。
选择一个 www. 规则并在一个函数中应用它,以便同一个站点永远不会以两种拼写查找。任何扩展到数百个名称的反向链接域名检查器都需要这种纪律。
运行日志应显示的内容
在每个批次后打印 _credits 值,并在其旁边写入批次号、域名数量和耗时。这三个数字几乎回答了财务读者对运行提出的所有问题。
直到结束才打印任何内容的运行让你无法知道何时停滞。每批一行很便宜,它将无声的挂起转化为你可以命名的批次号。
将日志保存在与 SQLite 文件相同的文件夹中,以便一次备份复制两者。位于其他位置的日志是在事件期间丢失的那个。

值得花一下午时间的三个扩展
在时间戳旁边存储批次号。它已经在上面的表中,它让你可以在运行中途 402 后只重新运行失败的批次。
在检查真实响应后,为原始表编写一个小视图:域名、分数、获取时间。仪表板读取视图,原始列保持作为数字受到质疑时的证据。
从 GitHub Actions 或小型 VPS 上的普通 cron 运行它。两者都可以。重要的是作业在任何失败时以非零退出,因为无声的失败在几周后看起来像是一个陈旧的图表。
对输出进行排序是前端的问题。按分数对投资组合进行排名、分桶,或将其与 dr_distribution(1 个信用额度)进行比较,以查看每个客户在全部 181,132,097 个域名中的位置。
字段漂移和原始表
当字段形状改变时,长期运行的作业会无声失败。原始列的存在使得失败保持明显且易于修复。
首先存储原始正文
原始列是安全网。如果字段被重命名或出现新字段,你可以对存储的响应运行查询,而不是再次为整个列表付费。稍后解析很便宜,而重新购买则不然。
每周保留一个保存的样本在自己的文件中。当解析中断时,你将本周的样本与上一个有效的样本进行差异比较。
按接收到的原样存储正文,不进行美化打印或修剪。你在存储之前对文本进行的任何更改都是你以后无法撤销的更改。
漂移在实际中的表现
假设仪表板突然显示一批空分数。原始行仍然存在。你打开它,将其键与上周的样本进行比较,并在几分钟内找到差异。
修复则是视图中的一行,而不是重新运行整个列表。这就是存储正文的全部意义。
在视图旁边保留一条简短说明,说明它读取哪个字段以及你上次何时将其与真实响应进行比较。写这条说明只需一分钟,但能节省以后的长时间搜索。
按数据顺序进行迁移
在视图之前创建原始表,在任何仪表板查询之前创建视图。将创建语句保存在一个文件中,按顺序运行它们,并为每一行记录批次号。
每个客户端的数据库漂移与共享数据库的漂移方式相同。规则保持不变:每次调用一个存储的正文,一个解析位置。
将解析器保留在测试旁边
将解析函数放在读取一个保存正文的测试的同一存储库中。测试是漂移字段首先出现的地方,远在图表出现之前。
在文件顶部附近的一个常量中命名解析器期望的字段。然后,审查者可以在单个屏幕中看到对响应形状的所有依赖项。
首次运行,按顺序
- 1
测试一个域名
用你知道的一个域名调用
bulk_dr,并阅读整个响应。在编写任何解析代码之前写下键。 - 2
运行两个批次
给脚本一个 400 个域名的列表,以便恰好运行两个批次。检查
_credits的下降是否与两次调用每次一个信用额度相符。 - 3
计算存储的行数
打开 SQLite 文件并将行数与批次计数进行比较。不匹配意味着批次被跳过或存储了两次,你可以在仪表板之前发现这一点。
- 4
安排完整列表
只有在测试运行干净后,才添加完整列表和调度器。保持非零退出,以便在第一个晚上失败时发出响亮警报。
在调度之前
- 首先对列表进行去重和小写处理
- 在一个函数中决定一次,
www.是保留还是去掉 - 将密钥保存在秘密存储中,而不是在 crontab 中
- 对非零退出代码发出警报
- 根据你的自身规模运行 信用额度预算 数字