访问日志一天几十万条,逐条调用解析接口不现实。很多团队的第一反应是随机抽一千条,跑完却说不清结论:抽到的样本里到底有多少爬虫、多少异常,全凭感觉。
下面用数据鱼API提供的「UA 解析与爬虫识别」接口演示,调用的是「批量解析」分支(
POST /v1/api/ua/batch)。批量解析 UA,最多 10 条。
更稳的顺序是先用批量接口跑一个整体分布,再按判定结果去抽样细查。批量接口 POST /v1/api/ua/batch 一次能处理的数量有限,但足够回答“这批流量长什么样”。
uas 是必填的字符串数组。服务端会自动去重、丢弃空串,处理后的数量上限是 10 条,超出会直接报错。也就是说,你提交 30 条,其中 10 条重复、2 条为空,最终处理的数量会明显少于 30。
调用方最好在本地先做同样的清洗:去掉首尾空白和空串、按完整字符串去重、按 10 条切分。这样每一批都能装满有效样本,统计分母也可控。mode 可选 parse 或 detect,默认是 parse;ip 可选,用于给整批 UA 提供统一的 IP 上下文;headers 可选,用于无头浏览器检测。
返回里的 data.total 是总条数,data.returned 是本次实际返回条数。data.bot_count 和 data.human_count 给出爬虫与人类的粗略数量,data.verdict_dist 则是判定结果分布:键是本次批次实际出现的判定结果,值是对应数量。
verdict_dist 只会包含出现过的键。例如这一批全是人类流量,返回里可能只有 human 一个键,不能假设 human、suspicious、unknown、bot-friendly、bot-malicious 五个键永远都在。代码里用 get(key, 0) 取值,比直接索引安全。
data.max_limit 是允许返回的最大条数,data.mode 回显本次使用的模式,data.truncated 是保留字段,当前恒为 false。超过上限时会直接报错,不会截断,所以不要等 truncated 变成 true 才去分批。
parse 模式只做 UA 解析和基础识别,不需要 IP,适合大批量的日志归类。detect 模式会额外计算 IP 风险、PTR 和 ASN,但 ip 是整批共用的,不是每条一个。如果你的日志来自很多不同的客户端 IP,把它们混在一批里跑 detect,网络身份这一层就没有意义。
比较合理的做法是:先用 parse 跑全量分布,挑出 suspicious 和 bot-malicious 的样本,再按来源 IP 分组,用 detect 逐组复核。headers 在批量接口里要传对象,不是 JSON 字符串,这一点和单条解析接口正好相反。
分布里的数字来自去重后的结果,分母应该用 data.returned,而不是原始日志行数。用原始行数算爬虫占比,重复 UA 多的批次会被低估,空串多的批次会被高估。
抽样也不要只抽 human。风控真正需要看的是 suspicious 和 bot-malicious 这两类,以及 data.list[] 里 anomaly.score 较高的记录。把有限的人工复核时间放在这些样本上,比均匀随机更有效。
uas 超过 10 条会报错,不会自动截断。分批逻辑必须在调用方完成,并且要在去重之后切分,否则会出现“每批 10 条、实际只有 6 条有效”的浪费。
headers 的类型容易写错。单条接口用 JSON 字符串,批量接口用对象,封装时最好分别提供参数构造方法。
verdict_dist 的键是动态的。做图表时先合并所有批次出现过的键,再补零,不然不同批次拼起来的趋势图会缺线。
from collections import Counter
def ua_distribution(uas):
unique = list(dict.fromkeys(u for u in uas if u and u.strip()))
counter = Counter()
bots = humans = 0
for i in range(0, len(unique), 10):
resp = requests.post(
f"{API_BASE}/v1/api/ua/batch",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"uas": unique[i:i + 10], "mode": "parse"},
timeout=15,
)
payload = resp.json()["data"]
bots += payload.get("bot_count", 0)
humans += payload.get("human_count", 0)
counter.update(payload.get("verdict_dist", {}))
return {"returned": sum(counter.values()), "bots": bots,
"humans": humans, "dist": dict(counter)}批量接口解决的是“看全局”,单条接口解决的是“看个案”。先跑分布,再带着问题去抽样,复核才不会变成盲人摸象。
示例接口由数据鱼API提供;不同版本的字段可能微调,正式接入前建议再核对一次文档。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。