Python 自动化提取国家网络安全通报 PDF：恶意 IP 情报提取脚本实战

Python 自动化提取国家网络安全通报 PDF：恶意 IP 情报提取脚本实战

本文介绍一个用于解析国家网络安全通报中心《重点防范境外恶意网址和恶意IP》系列 PDF、自动提取结构化威胁情报（IOC）的 Python 脚本：从原始通报到可直接入库的恶意 IP 记录，全程脚本化完成。

一、背景与痛点

国家网络安全通报中心会定期发布《重点防范境外恶意网址和恶意IP》系列通报（如"续二十一"），每期包含 10 条左右的境外恶意地址情报，字段包括：

• 恶意地址（域名或 IP）
• 关联 IP 地址
• 归属地
• 威胁类型（僵尸网络 / 后门）
• 病毒家族（Mirai、AsyncRAT、CondiBot 等）
• 攻击描述

这类通报以 PDF 形式发布，内容结构化程度高、格式相对固定，非常适合自动化提取。如果靠人工逐条复制粘贴整理，不仅效率低，还容易漏行、串行、引入格式错误。于是就有了这个脚本：输入通报 PDF，一键输出整理好的结构化 TXT，方便后续入库恶意 IP 管理系统或做进一步分析。

二、脚本功能概览

一次运行输出示例：

IP地址: 31.77.227.102
关联域名: msupdate.us
归属地: 美国/犹他州/奥勒姆
威胁类型: 僵尸网络
病毒家族: Mirai
攻击描述: 这是一种Linux僵尸网络病毒，通过网络下载、漏洞利用、Telnet和SSH暴力破解等方式进行扩散…

三、核心实现解析

3.1 PDF 文本提取

脚本基于 pdfplumber 逐页提取文本，并按页拼接：

def extract_pdf_text(pdf_path: str) –> str:
texts = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
text = page.extract_text() or ""
texts.append(text)
return "\\n".join(texts)

page.extract_text() 会按 PDF 内部文本流顺序返回该页文字，是 pdfplumber 最常用的能力。

3.2 记录分块

通报正文中每条情报以"（一）恶意地址：…“、”（二）恶意地址：…"开头，直到"二、排查方法"结束。利用这个规律，先截断正文，再用正则按块切分：

raw_text = raw_text.split('二、排查方法')[0]

pattern = re.compile(
r'（[一二三四五六七八九十]+）恶意地址：.*?(?=(?:（[一二三四五六七八九十]+）恶意地址：|二、排查方法|三、处置建议|\\Z))',
re.S
)
blocks = pattern.findall(raw_text)

关键点：

• （[一二三四五六七八九十]+）恶意地址： 匹配序号 + 起始字段；
• .*? 非贪婪匹配到下一个序号块或章节标题为止；
• re.S 让 . 可以跨行匹配。

3.3 字段提取

每个块内按"字段名：值"的形式提取。为了避免误吞下一个字段，用正向预查（lookahead）在下一个已知字段名处截断：

def get_field(block: str, field: str) –> str:
pattern = rf'{re.escape(field)}[:：]\\s*(.*?)(?=\\s*(?:恶意地址|关联IP地址|归属地|威胁类型|病毒家族|描述|攻击描述)[:：]|\\Z)'
m = re.search(pattern, block, re.S)
return clean_text(m.group(1)) if m else ''

clean_text 负责清洗：去掉 \\r、合并连续空白、消除中文标点前的多余空格，保证输出整洁。

3.4 IP 与域名逻辑

部分记录（如"恶意地址"本身是 IP 的条目）没有"关联IP地址"字段，脚本用 is_ip() 判断并做兜底：

def is_ip(addr: str) –> bool:
return bool(re.fullmatch(r'(?:\\d{1,3}\\.){3}\\d{1,3}', addr.strip()))

if not ip and is_ip(addr):
ip = addr
domain = '无'
else:
domain = addr if addr else '无'
ip = ip if ip else '无'

即：恶意地址是域名时，输出"关联域名=域名 + 关联IP=IP"；恶意地址本身就是 IP 时，输出"关联域名=无 + 关联IP=该IP"。

3.5 输出保存

按固定字段顺序写出，记录间以空行分隔，方便人工阅读和后续程序解析：

def save_to_txt(records, output_path='整理结果.txt'):
with open(output_path, 'w', encoding='utf-8') as f:
for i, r in enumerate(records):
f.write(f"IP地址: {r['IP地址']}\\n")
f.write(f"关联域名: {r['关联域名']}\\n")
# … 其余字段
if i != len(records) – 1:
f.write("\\n")

四、应用场景

五、实战踩坑与优化方向

以下问题均来自真实运行环境，可作为后续迭代方向：

5.1 序号字符编码不一致导致漏条（重要）

通报 PDF 内嵌字体对部分汉字使用了康熙部首码点（U+2F00 区）与 CJK 部首补充区（U+2E80 区），例如"一、二、八、十"被编码为 ⼀(U+2F00)、⼆(U+2F06)、⼋(U+2F0B)、⼗(U+2F17)，而"三、四、五、六、七、九"是普通汉字（U+4E00 区）。正则 [一二三四五六七八九十] 无法匹配部首码点，导致一、二、八、十条目被整体跳过（10 条只提取出 6 条）。

优化：提取文本后做部首码点归一化，U+2F00 区用 unicodedata.normalize('NFKC')，U+2E80 补充区（如 ⻔→门、⻢→马）手动映射：

import unicodedata

KANGXI_SUPP = {'⺠': '民', '⻔': '门', '⻢': '马', '⻥': '鱼'}

def norm_kangxi(s: str) –> str:
out = []
for ch in s:
cp = ord(ch)
if 0x2F00 <= cp <= 0x2FD5:
out.append(unicodedata.normalize('NFKC', ch))
elif ch in KANGXI_SUPP:
out.append(KANGXI_SUPP[ch])
else:
out.append(ch)
return ''.join(out)

注意：不要对整个文本直接 NFKC，否则全角冒号、全角括号会被半角化，破坏原有正则。

5.2 页眉页脚混入正文

微信文章导出的 PDF 每页顶部有页眉（日期 + 标题），底部有页脚（公众号链接 + 页码）。它们会被 extract_text() 当正文提取，穿插在跨页记录的字段之间（例如"病毒家族"后面混入 https://mp.weixin.qq.com/… 1/3）。

优化：逐行过滤，命中页眉/页脚特征的行直接丢弃：

if re.fullmatch(r'https?://\\S+\\s+\\d+/\\d+', s): # 页脚：链接+页码
continue
if re.search(r'^\\d{4}/\\d{1,2}/\\d{1,2}\\s+\\d{1,2}:\\d{2}\\s+重点防范', s): # 页眉
continue

5.3 pdfminer 的 FontBBox 警告

运行时控制台可能出现：

Could not get FontBBox from font descriptor because None cannot be parsed as 4 floats

这是 pdfplumber 底层 pdfminer.six 解析字体元数据时的非致命警告（该 PDF 部分字体描述符缺少 FontBBox 字段），不影响提取结果。可在脚本开头压制日志：

import logging
logging.getLogger('pdfminer').setLevel(logging.ERROR)

5.4 入库去重与冲突处理

对接恶意 IP 管理系统时，建议按 IP 做去重合并；若 IP 已存在但字段内容不一致，保留旧记录并单独上报人工确认，避免误覆盖历史数据。

六、完整代码

import re
from pathlib import Path
import pdfplumber

def clean_text(s: str) –> str:
s = s.replace('\\r', '')
s = re.sub(r'\\s+', ' ', s)
s = re.sub(r'\\s+([、。；：,，.])', r'\\1', s)
return s.strip()

def get_field(block: str, field: str) –> str:
pattern = rf'{re.escape(field)}[:：]\\s*(.*?)(?=\\s*(?:恶意地址|关联IP地址|归属地|威胁类型|病毒家族|描述|攻击描述)[:：]|\\Z)'
m = re.search(pattern, block, re.S)
return clean_text(m.group(1)) if m else ''

def is_ip(addr: str) –> bool:
return bool(re.fullmatch(r'(?:\\d{1,3}\\.){3}\\d{1,3}', addr.strip()))

def extract_pdf_text(pdf_path: str) –> str:
texts = []
with pdfplumber.open(pdf_path) as pdf:
for page in pdf.pages:
text = page.extract_text() or ""
texts.append(text)
return "\\n".join(texts)

def parse_records(raw_text: str):
raw_text = raw_text.split('二、排查方法')[0]

pattern = re.compile(
r'（[一二三四五六七八九十]+）恶意地址：.*?(?=(?:（[一二三四五六七八九十]+）恶意地址：|二、排查方法|三、处置建议|\\Z))',
re.S
)
blocks = pattern.findall(raw_text)

results = []
for block in blocks:
block = clean_text(block)

addr = get_field(block, '恶意地址')
ip = get_field(block, '关联IP地址')
location = get_field(block, '归属地')
threat = get_field(block, '威胁类型')
family = get_field(block, '病毒家族')
desc = get_field(block, '描述') or get_field(block, '攻击描述')

if not ip and is_ip(addr):
ip = addr
domain = '无'
else:
domain = addr if addr else '无'
ip = ip if ip else '无'

results.append({
'IP地址': ip,
'关联域名': domain,
'归属地': location if location else '无',
'威胁类型': threat if threat else '无',
'病毒家族': family if family else '无',
'攻击描述': desc if desc else '无'
})

return results

def save_to_txt(records, output_path='整理结果.txt'):
with open(output_path, 'w', encoding='utf-8') as f:
for i, r in enumerate(records):
f.write(f"IP地址: {r['IP地址']}\\n")
f.write(f"关联域名: {r['关联域名']}\\n")
f.write(f"归属地: {r['归属地']}\\n")
f.write(f"威胁类型: {r['威胁类型']}\\n")
f.write(f"病毒家族: {r['病毒家族']}\\n")
f.write(f"攻击描述: {r['攻击描述']}\\n")
if i != len(records) – 1:
f.write("\\n")

if __name__ == "__main__":
pdf_file = "重点防范境外恶意网址和恶意IP（续二十一）.pdf"
text = extract_pdf_text(pdf_file)
records = parse_records(text)
save_to_txt(records)
print(f"完成，共提取 {len(records)} 条记录")

运行方式：

pip install pdfplumber
python pdf文档内容提取.py

七、扩展思路

本文代码与优化建议均来自实际运行环境，数据真实可复现。