1. 安全运营的每日情报聚合为什么总在重复劳动如果你在安全运营SOC岗位待过一段时间大概率经历过这样的早晨打开十几个浏览器标签页The Hacker News、FreeBuf、CISA KEV、NVD、GitHub Security Advisory 挨个刷一遍把 CVE 编号、CVSS 评分、利用状态手动抄进 Excel再去重、排序、写摘要。等日报发出去两个小时没了而且明天还要再来一遍。这个场景的核心痛点不是信息不够而是信息过载 格式不统一 重复劳动。每个情报源的字段结构都不一样NVD 用cve.id和metrics.cvssMetricV31CISA KEV 用cveID和dateAddedGitHub Advisory 用ghsa_id和severity。你想把它们归并成一张表光字段映射就能写半天。更麻烦的是归并去重这一步。同一个 CVE 可能同时出现在 NVD、CISA KEV 和某篇安全媒体的报道里如果不去重日报里就会出现三条一模一样的记录看起来很不专业。而去重规则又不能简单按 CVE 编号——有些情报源只给标题不给编号得靠标题相似度或漏洞描述关键词来兜底。我试过纯手工做这件事坚持了不到一周就放弃了。后来改成半自动写一个 Python 脚本拉取各源数据用统一 Key 走一个 API 通道做字段归一化和摘要生成最后输出 Markdown。这套流程跑下来每天早上从两小时手工压缩到十分钟检查 一键生成。这篇文章就把这套流程拆开讲清楚。适合谁看做安全运营、威胁情报、漏洞管理的同学或者任何需要每天汇总多源信息并生成结构化报告的人。你不需要是 Python 高手只要能看懂字典和列表操作就能跟着做下来。核心思路是采集脚本负责拉数据统一 Key/API 通道负责归并和摘要去重规则负责保证报告干净。下面从环境准备开始。2. TaoToken 统一 Key 与多源情报归并的前置准备在动手写采集脚本之前先把统一 Key/API 通道这件事说清楚。多源情报聚合的难点在于你从 NVD 拉到的是一条 JSON从 CISA KEV 拉到的是另一条 JSON字段名、时间格式、严重性分级全都不一样。如果每个源都单独写一套解析逻辑脚本会变得又长又难维护。更优雅的做法是把所有原始数据先归一化成统一的中间结构再交给一个模型通道做字段映射、摘要生成和去重判断。这个模型通道就是 TaoToken 发挥作用的地方——它提供统一的 API 入口你只需要一个 Key就能调用模型完成文本归一化和摘要任务不用为每个源单独对接不同的模型服务。具体来说TaoToken 在这个流程里承担三件事第一字段映射辅助。当你遇到一个没见过格式的情报源时可以把原始 JSON 片段丢给模型让它输出字段映射建议比如这个源的vuln_id对应统一结构的cve_id。第二摘要生成。原始漏洞描述往往很长NVD 的描述动辄几百字日报里需要压缩成一两句话。模型可以基于漏洞描述、CVSS 评分和利用状态生成一段适合放进日报的中文摘要。第三去重判断。当两个源的记录标题相似但编号不同时可以让模型判断它们是否指向同一个漏洞避免误删或漏删。前置准备需要三样东西一个 TaoToken 的 API Key、Python 3.9 环境、以及各情报源的访问方式大部分公开源不需要认证CISA KEV 和 NVD 都提供公开 JSON 接口。获取 Key 的路径是访问 TaoToken 官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 API Key。控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite API Key 管理页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。API 的基础地址是 https://taotoken.net/api 注意这个地址不带 UTM 参数直接用于代码里的base_url。拿到 Key 之后建议先做一次连通性验证确认 Key 有效、模型可调用。验证用的模型对话入口在 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 你可以先在网页上试一次对话确认返回正常再写进脚本。环境依赖方面只需要两个库requests用于拉取情报源和调用 APIpython-dateutil用于处理各源不一致的时间格式。安装命令pip install requests python-dateutil如果你打算把日报生成做成定时任务可以再加一个schedule库但这不是必须的用系统的 cron 或 Windows 任务计划也能实现。这里要提醒一个容易踩的坑不要把 API Key 硬编码在脚本里。建议用环境变量读取比如TAOTOKEN_API_KEY。这样脚本可以安全地提交到 Git 仓库也方便在不同机器上复用。下面进入具体的配置环节。3. 可复制的采集脚本配置与字段映射表这一节给出完整的可复制配置。先看目录结构建议这样组织threat-daily/ ├── config/ │ ├── sources.json # 情报源配置 │ └── field_mapping.json # 字段映射表 ├── scripts/ │ ├── fetch.py # 采集脚本 │ ├── normalize.py # 归一化 去重 │ └── report.py # 生成 Markdown ├── data/ │ └── raw/ # 原始数据落盘 └── output/ └── daily-2026-04-19.md先写config/sources.json定义每个情报源的拉取方式和字段路径{ sources: [ { name: CISA_KEV, url: https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json, type: json, root_path: vulnerabilities, fields: { cve_id: cveID, vendor: vendorProject, product: product, title: vulnerabilityName, date_added: dateAdded, due_date: dueDate, description: shortDescription } }, { name: NVD_RECENT, url: https://services.nvd.nist.gov/rest/json/cves/2.0?pubStartDate2026-04-12T00:00:00.000pubEndDate2026-04-19T00:00:00.000, type: json, root_path: vulnerabilities, fields: { cve_id: cve.id, title: cve.descriptions[0].value, cvss: cve.metrics.cvssMetricV31[0].cvssData.baseScore, severity: cve.metrics.cvssMetricV31[0].cvssData.baseSeverity, published: cve.published, description: cve.descriptions[0].value } } ] }注意 NVD 的字段路径用了点号和数组下标采集脚本里需要一个deep_get函数来解析这种路径。CISA KEV 的字段是扁平的直接取键即可。再看config/field_mapping.json这是统一中间结构的定义也是去重规则的依据{ unified_schema: { cve_id: string, title: string, severity: enum[CRITICAL,HIGH,MEDIUM,LOW,UNKNOWN], cvss: float, exploited: boolean, source: string, published: date, summary: string }, severity_normalize: { critical: CRITICAL, high: HIGH, medium: MEDIUM, low: LOW, 严重: CRITICAL, 高危: HIGH, 中危: MEDIUM }, dedup_rules: { primary_key: cve_id, fallback_key: title_normalized, merge_strategy: keep_highest_severity, source_priority: [CISA_KEV, NVD_RECENT, MEDIA] } }去重规则的核心是优先按cve_id去重如果没有 CVE 编号就用归一化后的标题去掉空格、转小写、去掉标点作为兜底键。合并策略是保留严重性最高的那条如果严重性相同按source_priority保留优先级高的源。接下来是采集脚本scripts/fetch.py的关键部分import json import os import requests from datetime import datetime def deep_get(obj, path): 支持 a.b[0].c 形式的路径取值 parts path.replace([, .).replace(], ).split(.) cur obj for p in parts: if p : continue if isinstance(cur, list): cur cur[int(p)] else: cur cur.get(p) if cur is None: return None return cur def fetch_source(src): resp requests.get(src[url], timeout30) resp.raise_for_status() data resp.json() root data.get(src[root_path], data) if src[root_path] else data records [] for item in root: rec {source: src[name]} for unified_key, raw_path in src[fields].items(): rec[unified_key] deep_get(item, raw_path) records.append(rec) return records if __name__ __main__: with open(config/sources.json, encodingutf-8) as f: cfg json.load(f) all_records [] for src in cfg[sources]: try: recs fetch_source(src) all_records.extend(recs) print(f[OK] {src[name]}: {len(recs)} 条) except Exception as e: print(f[FAIL] {src[name]}: {e}) os.makedirs(data/raw, exist_okTrue) out fdata/raw/{datetime.now().strftime(%Y-%m-%d)}.json with open(out, w, encodingutf-8) as f: json.dump(all_records, f, ensure_asciiFalse, indent2) print(f原始数据已落盘: {out})这个脚本跑完你会得到一份原始记录列表每条记录已经按sources.json里的字段映射做了初步归一化。但此时还没有去重也没有生成摘要。下一步是调用 TaoToken 的 API 做摘要生成和去重判断。调用 API 的部分用 OpenAI 兼容的 SDK 最省事from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlhttps://taotoken.net/api ) def summarize(record): prompt f你是安全情报分析助手。请基于以下漏洞信息生成一句不超过 60 字的中文摘要 突出漏洞类型、受影响组件和利用状态。只输出摘要不要解释。 CVE: {record.get(cve_id)} 标题: {record.get(title)} 严重性: {record.get(severity)} 描述: {record.get(description, )[:500]} resp client.chat.completions.create( modelclaude-sonnet-4-5, messages[{role: user, content: prompt}], temperature0.2 ) return resp.choices[0].message.content.strip()这里model参数填你实际可用的模型 ID具体可用列表可以在模型对话页面确认。base_url就是前面说的https://taotoken.net/api不带任何查询参数。字段映射表用表格对照更直观统一字段CISA KEV 来源NVD 来源说明cve_idcveIDcve.id主键去重依据titlevulnerabilityNamecve.descriptions[0].value标题兜底去重键severity无默认 HIGHcvssMetricV31.baseSeverity需归一化cvss无cvssMetricV31.baseScore浮点数exploited恒为 true需查 references 含 exploit布尔值publisheddateAddedcve.published统一转 ISO 日期配置写到这里采集和归一化的骨架就搭好了。下一节演示一次完整的从拉取到生成 Markdown 报告的验证动作。4. 从拉取到生成 Markdown 报告的完整验证这一节把前面的脚本串起来跑一次端到端的验证。假设你已经配好了sources.json、field_mapping.json并且设置了环境变量TAOTOKEN_API_KEY。第一步跑采集脚本cd threat-daily export TAOTOKEN_API_KEY你的Key python scripts/fetch.py预期输出类似[OK] CISA_KEV: 1247 条 [OK] NVD_RECENT: 386 条 原始数据已落盘: data/raw/2026-04-19.json如果某个源报[FAIL]先检查网络和 URL 是否可访问NVD 的接口偶尔会限流加个time.sleep(6)在请求之间会更稳。第二步写归一化和去重脚本scripts/normalize.pyimport json from datetime import datetime def normalize_severity(raw, mapping): if not raw: return UNKNOWN return mapping.get(str(raw).lower(), str(raw).upper()) def normalize_title(t): if not t: return return .join(ch for ch in t.lower() if ch.isalnum()) def dedup(records, rules, sev_map): seen {} for rec in records: rec[severity] normalize_severity(rec.get(severity), sev_map) key rec.get(cve_id) or normalize_title(rec.get(title)) if not key: continue if key not in seen: seen[key] rec else: old seen[key] old_rank rules[source_priority].index(old[source]) if old[source] in rules[source_priority] else 99 new_rank rules[source_priority].index(rec[source]) if rec[source] in rules[source_priority] else 99 sev_order [CRITICAL, HIGH, MEDIUM, LOW, UNKNOWN] if sev_order.index(rec[severity]) sev_order.index(old[severity]) or \ (rec[severity] old[severity] and new_rank old_rank): seen[key] rec return list(seen.values()) if __name__ __main__: with open(config/field_mapping.json, encodingutf-8) as f: mapping json.load(f) with open(data/raw/2026-04-19.json, encodingutf-8) as f: records json.load(f) result dedup(records, mapping[dedup_rules], mapping[severity_normalize]) with open(data/normalized-2026-04-19.json, w, encodingutf-8) as f: json.dump(result, f, ensure_asciiFalse, indent2) print(f去重后: {len(result)} 条原始 {len(records)} 条)跑一下python scripts/normalize.py预期输出去重后: 412 条原始 1633 条去重率通常在 70% 以上因为 CISA KEV 和 NVD 有大量重叠。第三步生成 Markdown 报告scripts/report.pyimport json from datetime import datetime SEV_ICON {CRITICAL: 严重, HIGH: 高危, MEDIUM: 中危, LOW: 低危, UNKNOWN: 未知} def render(records, date_str): records.sort(keylambda r: ([CRITICAL,HIGH,MEDIUM,LOW,UNKNOWN].index(r[severity]), r.get(cve_id) or )) lines [f# 每日安全情报报告 · {date_str}, ] lines.append(f生成时间{datetime.now().strftime(%Y-%m-%d %H:%M)} | 记录数{len(records)}) lines.append() lines.append(## 一、高危漏洞速报) lines.append() for r in records: if r[severity] not in (CRITICAL, HIGH): continue lines.append(f### {r.get(cve_id) or 无编号} — {r.get(title, )[:60]}) lines.append() lines.append(f- 严重性{SEV_ICON.get(r[severity], r[severity])}) lines.append(f- 来源{r[source]}) lines.append(f- 摘要{r.get(summary, 待生成)}) lines.append() return \n.join(lines) if __name__ __main__: with open(data/normalized-2026-04-19.json, encodingutf-8) as f: records json.load(f) md render(records, 2026-04-19) with open(output/daily-2026-04-19.md, w, encodingutf-8) as f: f.write(md) print(报告已生成: output/daily-2026-04-19.md)跑完打开output/daily-2026-04-19.md你会看到一份结构化的日报高危漏洞按严重性排序每条包含 CVE 编号、标题、来源和摘要。第四步把摘要生成接进去。在normalize.py之后加一步对去重后的记录批量调用 TaoToken 生成摘要from openai import OpenAI import os, json client OpenAI(api_keyos.environ[TAOTOKEN_API_KEY], base_urlhttps://taotoken.net/api) with open(data/normalized-2026-04-19.json, encodingutf-8) as f: records json.load(f) for r in records[:50]: # 先处理前 50 条避免一次调用太多 if r.get(summary): continue prompt f用一句不超过 60 字的中文概括这个漏洞突出类型、组件、利用状态{r.get(title)} | {r.get(description,)[:400]} resp client.chat.completions.create( modelclaude-sonnet-4-5, messages[{role: user, content: prompt}], temperature0.2 ) r[summary] resp.choices[0].message.content.strip() with open(data/normalized-2026-04-19.json, w, encodingutf-8) as f: json.dump(records, f, ensure_asciiFalse, indent2) print(摘要生成完成)验证成功的标志output/daily-2026-04-19.md里每条高危漏洞都有中文摘要且没有重复的 CVE 编号。到这里一次完整的拉取 → 归一化 → 去重 → 摘要 → 生成 Markdown流程就跑通了。5. 本篇常见报错排查实际跑这套流程时最容易卡在几个地方。下面按报错现象对照排查。报错一401 Unauthorized或invalid api key这是最常见的。原因通常是环境变量没设置或者 Key 复制时带了空格。检查方式echo $TAOTOKEN_API_KEY | head -c 10如果输出为空说明环境变量没生效。注意export只在当前终端会话有效换终端要重新设置。如果 Key 开头有空格用export TAOTOKEN_API_KEY$(echo $TAOTOKEN_API_KEY | tr -d )清理。另外确认base_url写的是https://taotoken.net/api不要多加/v1或斜杠。报错二local proxy failed或连接超时这个报错通常出现在请求情报源时而不是调用 API 时。NVD 和 CISA 的接口在国内访问偶尔不稳定。排查顺序先用curl -I https://services.nvd.nist.gov/rest/json/cves/2.0看是否能通如果不通检查本机网络配置。注意不要使用任何非合规的网络工具合规的网络环境本身就能访问这些公开接口。如果只是偶发超时在fetch.py里加timeout60和重试逻辑即可。报错三reading choices或KeyError: choices这个报错说明 API 返回的结构里没有choices字段通常是模型 ID 写错了或者请求体格式不对。检查model参数是否是你账号下可用的模型 ID可以在模型对话页面确认。另外确认messages是列表每条消息有role和content两个键。如果返回的是错误信息先打印resp的原始内容看看print(resp.model_dump() if hasattr(resp, model_dump) else resp)报错四OAuth相关报错或authentication failed如果你用的是某些需要 OAuth 流程的客户端比如 Claude Code 或 Codex 的 CLI报错可能出现在认证环节。这类客户端的配置通常需要三件套Base URL、API Key、Model ID。以 Claude Code 为例配置文件里需要写全{ base_url: https://taotoken.net/api, api_key: 你的Key, model: claude-sonnet-4-5 }三个字段缺一不可。如果只填了 Key 没填 Base URL客户端会默认走官方地址导致认证失败。Codex 的auth.json同理需要确认base_url指向https://taotoken.net/api。报错五去重后记录数异常少如果去重后只剩几十条可能是normalize_title把不同漏洞的标题归一化成了同一个键。检查方式打印被合并的记录看它们的标题是否真的指向同一个漏洞。如果误合并调整去重规则把fallback_key改成cve_id title的组合或者提高标题相似度阈值。报错六Markdown 报告里摘要为空说明摘要生成那一步没跑或者summary字段被覆盖了。检查normalize.py和摘要脚本的执行顺序——必须先归一化去重再生成摘要最后渲染报告。如果顺序反了摘要会被去重逻辑丢掉。排查完这些基本就能稳定跑通。建议把每天的原始数据和归一化数据都落盘保留方便回溯和对比。6. 把日报流程固化成可复用的通道跑通一次之后下一步是把它变成每天自动执行的任务。Linux 下用 cronWindows 下用任务计划程序核心就是把三个脚本串成一条命令cd /path/to/threat-daily \ python scripts/fetch.py \ python scripts/normalize.py \ python scripts/summarize.py \ python scripts/report.py建议在summarize.py里加一个并发控制比如用concurrent.futures限制同时 5 个请求避免触发限流。另外摘要生成可以只对 CRITICAL 和 HIGH 级别的记录做MEDIUM 以下的直接截取描述前 60 字这样能省不少调用量。关于 Key 的管理如果你有多个脚本或多个环境建议在 TaoToken 控制台创建多个 API Key按用途区分比如日报采集一个、临时测试一个。这样某个 Key 出问题时可以单独吊销不影响其他流程。API Key 管理入口在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。如果你打算把这套流程扩展到更多情报源比如加入 GitHub Security Advisory 或某个厂商的公告页只需要在sources.json里加一条配置写好root_path和字段映射即可采集脚本不用改。这就是统一中间结构的好处——新增源的成本很低。对于需要长期跑、且调用量比较大的场景可以了解一下 Coding Plan它在持续调用场景下更划算具体可以看 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。如果只是偶尔跑一次日报按量调用就够了。最后说一个实用技巧把每天的归一化数据存成data/normalized-YYYY-MM-DD.json积累一段时间后你可以写个对比脚本看哪些 CVE 连续多天出现在报告里但一直没修复——这些就是需要重点推动的钉子户漏洞。这个视角比单日日报更有价值也是自动化流程真正省下来的时间该花的地方。