1. 安全运营的每日情报聚合为什么总在重复劳动

如果你在安全运营（SOC）岗位待过一段时间，大概率经历过这样的早晨：打开十几个浏览器标签页，The Hacker News、FreeBuf、CISA KEV、NVD、GitHub Security Advisory 挨个刷一遍，把 CVE 编号、CVSS 评分、利用状态手动抄进 Excel，再去重、排序、写摘要。等日报发出去，两个小时没了，而且明天还要再来一遍。

这个场景的核心痛点不是"信息不够"，而是信息过载 + 格式不统一 + 重复劳动。每个情报源的字段结构都不一样：NVD 用cve.id和metrics.cvssMetricV31，CISA KEV 用cveID和dateAdded，GitHub 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，注册后在控制台创建 API Key。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，API Key 管理页面在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。API 的基础地址是 https://taotoken.net/api ，注意这个地址不带 UTM 参数，直接用于代码里的base_url。

拿到 Key 之后，建议先做一次连通性验证，确认 Key 有效、模型可调用。验证用的模型对话入口在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ，你可以先在网页上试一次对话，确认返回正常，再写进脚本。

环境依赖方面，只需要两个库：requests用于拉取情报源和调用 API，python-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?pubStartDate=2026-04-12T00:00:00.000&pubEndDate=2026-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"], timeout=30) 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", encoding="utf-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_ok=True) out = f"data/raw/{datetime.now().strftime('%Y-%m-%d')}.json" with open(out, "w", encoding="utf-8") as f: json.dump(all_records, f, ensure_ascii=False, indent=2) print(f"原始数据已落盘: {out}")

这个脚本跑完，你会得到一份原始记录列表，每条记录已经按sources.json里的字段映射做了初步归一化。但此时还没有去重，也没有生成摘要。下一步是调用 TaoToken 的 API 做摘要生成和去重判断。

调用 API 的部分，用 OpenAI 兼容的 SDK 最省事：

from openai import OpenAI client = OpenAI( api_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://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( model="claude-sonnet-4-5", messages=[{"role": "user", "content": prompt}], temperature=0.2 ) return resp.choices[0].message.content.strip()

这里model参数填你实际可用的模型 ID，具体可用列表可以在模型对话页面确认。base_url就是前面说的https://taotoken.net/api，不带任何查询参数。

字段映射表用表格对照更直观：

配置写到这里，采集和归一化的骨架就搭好了。下一节演示一次完整的从拉取到生成 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.py：

import 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", encoding="utf-8") as f: mapping = json.load(f) with open("data/raw/2026-04-19.json", encoding="utf-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", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) print(f"去重后: {len(result)} 条（原始 {len(records)} 条）")

跑一下：

python scripts/normalize.py

预期输出：

去重后: 412 条（原始 1633 条）

去重率通常在 70% 以上，因为 CISA KEV 和 NVD 有大量重叠。

第三步，生成 Markdown 报告scripts/report.py：

import json from datetime import datetime SEV_ICON = {"CRITICAL": "严重", "HIGH": "高危", "MEDIUM": "中危", "LOW": "低危", "UNKNOWN": "未知"} def render(records, date_str): records.sort(key=lambda 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", encoding="utf-8") as f: records = json.load(f) md = render(records, "2026-04-19") with open("output/daily-2026-04-19.md", "w", encoding="utf-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_key=os.environ["TAOTOKEN_API_KEY"], base_url="https://taotoken.net/api") with open("data/normalized-2026-04-19.json", encoding="utf-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( model="claude-sonnet-4-5", messages=[{"role": "user", "content": prompt}], temperature=0.2 ) r["summary"] = resp.choices[0].message.content.strip() with open("data/normalized-2026-04-19.json", "w", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False, indent=2) 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里加timeout=60和重试逻辑即可。

报错三：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 下用 cron，Windows 下用任务计划程序，核心就是把三个脚本串成一条命令：

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_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。

如果你打算把这套流程扩展到更多情报源，比如加入 GitHub Security Advisory 或某个厂商的公告页，只需要在sources.json里加一条配置，写好root_path和字段映射即可，采集脚本不用改。这就是统一中间结构的好处——新增源的成本很低。

对于需要长期跑、且调用量比较大的场景，可以了解一下 Coding Plan，它在持续调用场景下更划算，具体可以看 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。如果只是偶尔跑一次日报，按量调用就够了。

最后说一个实用技巧：把每天的归一化数据存成data/normalized-YYYY-MM-DD.json，积累一段时间后，你可以写个对比脚本，看哪些 CVE 连续多天出现在报告里但一直没修复——这些就是需要重点推动的"钉子户"漏洞。这个视角比单日日报更有价值，也是自动化流程真正省下来的时间该花的地方。