企业官网建设流程全解析

简介：一份面向网络安全方向毕业设计的系统设计文档，主题为“基于Scrapy框架的威胁情报抓取以及处理系统的设计与实现”。文档针对开源威胁网站与博客等海量安全数据，提出以Scrapy爬虫框架进行数据提取、解析并存储为知识图谱底层数据，从而实现APT知识图谱数据需求的技术方案。内容覆盖爬虫模块、数据解析模块、数据展示模块的完整设计，详细讲解了Scrapy爬虫技术、知识图谱建模、Flask_admin+nginx后台管理，以及pyecharts可视化展示等关键技术，并包含课题背景、国内外研究现状、可行性分析、开发环境与系统详细设计等章节，结构清晰、层次分明。

资源为1个docx文档，压缩包大小2.58MB，便于直接阅读与二次编辑。文档中给出了系统模块划分、技术选型及实现要点，能够帮助毕业设计选题为威胁情报、爬虫系统或APT知识图谱方向的同学快速理解整体架构，也可作为系统设计与论文写作的参考模板。已有194人学习，适合网络安全相关专业的本科生和研究生参考使用。

1. 威胁情报抓取为什么绕不开 Scrapy+Redis 这套异步管道

威胁情报的核心是时效性。恶意 IP、域名、文件哈希这类 IOC（Indicator of Compromise）在报告发布几小时后就可能被攻击方换掉，如果靠人工翻安全博客再抄进表格，采集链路已经断在了最前面。MISP 事件库、安全厂商博客、公开的 APT 分析报告有共同特征：页面结构相对稳定、更新频率固定、字段散落在标题正文和 JSON 里。Scrapy 基于 Twisted 异步处理，相比 requests+手写线程池，能以更小的线程开销并发抓取大量页面，中间件、管道、调度器全部解耦；再配合 Redis 做请求队列和指纹去重，单机就能支撑每天定时跑完一轮全量增量，进程被 kill 后队列和去重状态也还在。这套方案适合安全数据分析、知识图谱方向的学生和工程师，既能拿到可完整复现的数据管道，也能看到生产环境里反爬、调度、展示的闭环。

2. Scrapy+Redis 爬虫架构、去重指纹与代理中间件的设计与实现

Scrapy 默认的去重机制是进程内 set 加磁盘上的 requests 队列，进程一断，已抓取 URL 的指纹全部丢失，重跑就会把已入库的数据再抓一遍。这就是本系统在单机场景下仍选用 scrapy-redis 的核心原因：请求队列、指纹集合、调度状态都在 Redis 里，进程重启后能接着队列续爬，而指纹去重用的 SHA1 哈希摘要会一直保留。

2.1 scrapy-redis 的组件与配置

启用 scrapy-redis 只需要在 settings.py 里替换调度器和去重类，并指定 Redis 连接参数：

# settings.py SCHEDULER = "scrapy_redis.scheduler.Scheduler" DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter" SCHEDULER_PERSIST = True REDIS_URL = "redis://127.0.0.1:6379/0"

注意redis://走的是 Redis 协议，和后面代理池的 HTTP 代理是两回事。测试阶段建议把 SCHEDULER_PERSIST 临时设为 False，避免调试时指纹堆积，影响下一次全量验证。

2.2 从报告列表页到详情页的爬取规则

爬虫的职责是拿到源站页面，不要把解析规则全塞进 parse。以 MISP 事件页面为例，列表页是事件摘要和详情链接，详情页里才是真正的威胁指示器。

import scrapy from urllib.parse import urljoin class MispEventSpider(scrapy.Spider): name = "misp_event" def start_requests(self): # 起始页从 settings 里读取，避免把 URL 写死在爬虫类中 for url in self.settings.get("START_URLS"): yield scrapy.Request(url, callback=self.parse_list, dont_filter=True) def parse_list(self, response): # 列表页中每条事件对应一个详情链接，逐个调度 for href in response.css("a.event-link ::attr(href)").getall(): yield scrapy.Request(urljoin(response.url, href), callback=self.parse_detail) # 处理翻页：这类列表页常用 page 参数 next_href = response.css("a.pagination-next ::attr(href)").get() if next_href: yield scrapy.Request(urljoin(response.url, next_href), callback=self.parse_list) def parse_detail(self, response): # 精确字段留给 item pipeline 处理，这里先返回原始内容 yield { "source": "misp", "title": response.css("h1.event-title ::text").get(), "detect_time": response.css("span.event-date ::text").get(), "content": response.text, }

start_requests 里的dont_filter=True是必须的：起始页如果被指纹过滤掉，整轮爬取会直接空转。详情页解析用的是 CSS 选择器，实际写的时候要先用开发者工具确认元素结构，不同版本的 MISP 页面差异很大，选择器写错的表现往往是标题正常但 content 为空。排查时先看 response.status 和 body 前 200 个字符，不要第一时间怀疑代理。

2.3 中间件：代理池随机抽取与失效回退

开源威胁站点的反爬强度不同，有些只检测 UA 频率，有些会封高频 IP。常见做法是把可用代理维护进 Redis 集合，每个请求随机取一个，失效后移入失败集合。

import redis r = redis.Redis.from_url("redis://127.0.0.1:6379/0") class RandomProxyMiddleware: """每个请求从 Redis 代理池随机取一个代理""" def process_request(self, request, spider): proxy = r.srandmember("proxy_pool") if proxy: request.meta["proxy"] = "http://" + proxy.decode() request.meta["proxy_raw"] = proxy def process_exception(self, request, exception, spider): # 连接失败时把代理从可用池挪到失败池，同时重试当前请求 proxy = request.meta.get("proxy_raw") if proxy: r.smove("proxy_pool", "proxy_pool_fail", proxy) return request

srandmember是 Redis 的随机取成员命令，适合代理池这种随时增删的场景；smove原子地把成员从集合 A 移到集合 B，不会出现两处都删掉或重复添加。注意 process_exception 返回 request 后，Scrapy 会按 RETRY_TIMES 重新调度，代理池太小会造成无限重试，所以 settings.py 里要把 RETRY_TIMES 限制在 2 或 3。UA 轮换用同样模式的中间件处理，随机从列表里取一个写进 headers。

2.4 管道写入与 MySQL 唯一键兜底

爬虫 yield 出去的 dict 最终由管道接收入库。到管道这步数据已经结构化，但仍要防住 parse 漏处理的重复记录，所以入库 SQL 用唯一键配合 ON DUPLICATE KEY UPDATE 做幂等兜底。

class IndicatorPipeline: def process_item(self, item, spider): sql = """ INSERT INTO indicator (source, type, value, detect_time, modified_time) VALUES (%(source)s, %(type)s, %(value)s, %(detect_time)s, NOW()) ON DUPLICATE KEY UPDATE modified_time = VALUES(modified_time) """ self.cursor.execute(sql, item) return item

这里的幂等写入很关键。如果 Redis 指纹因为误操作被清理，重跑全量爬虫时唯一键机制会把已存在的记录只更新时间戳，而不是插入重复行。

注意：Scrapy 默认只处理服务端渲染返回的 HTML。如果目标报告页改成前端异步加载，典型特征是页面源码里搜不到 IP，优先考虑配合 scrapy-playwright 渲染后再提取，这比直接逆向 XHR 接口成本低。本系统抓取的是服务端渲染为主的站点，不引入浏览器内核，资源和抓取速度都更可控。

3. 数据解析与入库：从报告文本到知识图谱底表

爬虫解决“把页面拿下来”，知识图谱真正需要的是实体和关系。这章解决三个问题：表怎么建、报告正文里的 IOC 怎么抽、不同来源的同一实体怎么对齐。

3.1 按知识图谱需求设计 MySQL 底表

系统爬取 MISP 事件、安全博客和公开 APT 报告，最终都要落到指示器、报告这类实体上。底表字段不是越全越好，而是保证每个字段都能被解析逻辑稳定填上。

-- 指示器表：IP、域名、URL、文件哈希统一存放 CREATE TABLE indicator ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, source VARCHAR(64) NOT NULL COMMENT '数据源标识，如 misp / talos', type VARCHAR(32) NOT NULL COMMENT 'ip / domain / url / md5 / sha256', value VARCHAR(255) NOT NULL, detect_time DATETIME DEFAULT NULL, modified_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_source_value (source, value) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; -- 报告表：保存文章标题、链接，作为指示器的溯源 CREATE TABLE report ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, source VARCHAR(64) NOT NULL, title VARCHAR(255) NOT NULL, url VARCHAR(512) NOT NULL, publish_time DATETIME DEFAULT NULL, content_hash CHAR(64) NOT NULL, UNIQUE KEY uk_url (url) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

UNIQUE KEY uk_source_value (source, value)就是上一节幂等兜底的根基。这里有一个容易踩的坑：type 字段必须存规范化的小写字符串，不要在管道里出现 "URL" 和 "Url" 混写，否则按类型统计时同一类数据会被拆成两列。

3.2 用正则从报告正文里抽取 IOC

威胁报告正文是半结构化文本。组织名、攻击链规律性不强，但 IP、域名、文件哈希格式固定，先抽确定性的指标项是最稳妥的一步。

import re PATTERNS = { "ip": re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), "domain": re.compile(r"\b(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}\b"), "md5": re.compile(r"\b[a-f0-9]{32}\b"), "sha256": re.compile(r"\b[a-f0-9]{64}\b"), } def extract_indicators(text): for ioc_type, pattern in PATTERNS.items(): for value in pattern.findall(text): value = value.lower().rstrip(".") if value in ("example.com", "localhost"): continue yield ioc_type, value

IP 正则只做格式匹配，999.999.1.1这种非法地址也会进来，所以批处理里要再对四段做 0-255 的范围校验。域名正则可能误匹配到1.2.3.4.traffic.com这种由 IP 前缀拼出的域名，实际项目里会先提取 IP，把匹配命中的 IP 从原文里抹掉再去提域名，减少交叉污染。

3.3 多源组织名的对齐策略

MISP 里记录的攻击组织和 ATTCK 网站上的命名经常不一致，直接按名字 join 是 join 不上的。常见做法是单独建一张别名表。

CREATE TABLE org_alias ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, org_name VARCHAR(64) NOT NULL COMMENT '攻击组织统一名', alias_name VARCHAR(128) NOT NULL COMMENT '源站使用的别名', source VARCHAR(64) NOT NULL, UNIQUE KEY uk_alias (alias_name, source) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

入库时先查 org_alias 做归一化，查不到就保持原样并写入记录，留待后续人工或模型补全。这套弱对齐方案的关键是不阻塞爬虫：对齐延迟处理，解析入库时只保证原始来源可追溯，后续脚本批量替换即可。

3.4 入库字段的清洗规则

无论数据来自哪个源，入库前统一走一遍清洗。越早统一规则，后面的查询和图表就越省事。

清洗放在 item pipeline 里而不是 spider 里，让 spider 保持对页面结构的单一职责。清洗后确定拒绝写入的空值，置成 None，避免 pymysql 抛 “'NoneType' object has no attribute 'format'” 这类低信息量错误。

4. Flask-Admin + pyecharts + Nginx 展示层的搭建与数据导出

数据入库后要解决“能查、能导、能看”。Flask-Admin 把 CRUD 视图、搜索、筛选、导出都做完了，注册模型类即可用；pyecharts 负责生成统计图表；Nginx 将 Flask 开发服务反代出去，形成完整查询链路。

4.1 用 ModelView 注册模型类

Flask-Admin 的常规用法是给每个模型定义一个 ModelView，注册到 admin 上。展示列、搜索、筛选都在类属性里声明。

from flask_admin.contrib.sqla import ModelView class IndicatorView(ModelView): column_list = ("id", "source", "type", "value", "detect_time", "modified_time") column_searchable_list = ("value",) column_filters = ("source", "type", "detect_time") can_export = True export_types = ("csv", "xlsx") page_size = 50 admin.add_view(IndicatorView(Indicator, db.session))

上面代码中，column_searchable_list生成搜索框，column_filters生成字段筛选器。detect_time 声明进 filters 后，界面会自动出现“最近 7 天 / 最近 30 天”这类时间范围，不需要手写查询条件。导出时要注意一个常见误区：Flask-Admin 导出的是当前筛选条件下的数据，不是全表。想导出“所有 ip 类型的指示器”，必须先设好 type 筛选器再点导出，否则拿到的只是当前页 50 条。

4.2 用 pyecharts 生成统计图页面

pyecharts 2.x 渲染结果是独立 HTML，Page 组件可以把多张图合成一个页面，适合放在后台的统计图下拉菜单。

from pyecharts.charts import Bar, Pie, Page from pyecharts import options as opts bar = Bar() bar.add_xaxis(days) bar.add_yaxis("新增指示器", counts) bar.set_global_opts(title_opts=opts.TitleOpts(title="近 30 天指示器增长趋势")) pie = Pie() pie.add("类型占比", type_pairs) pie.set_global_opts(title_opts=opts.TitleOpts(title="昨日指示器类型占比")) page = Page(layout=Page.SimplePageLayout) page.add(bar, pie) page.render("/root/python/flask-admin/examples/sqla/admin/static/trend.html")

render 用绝对路径是因为 Flask-Admin 的 example 工程里模板按相对位置解析，相对路径容易渲染到非预期目录。Page 的SimplePageLayout会把多张图纵向排布，适合后台直接嵌入 iframe。

4.3 Nginx 反向代理配置

Flask 自带开发服务器直接暴露公网，并发能力不足且路径头会泄露应用结构。生产环境用 Nginx 做反向代理：

server { listen 80; server_name _; location / { proxy_pass http://127.0.0.1:5000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } location /static/ { alias /root/python/flask-admin/examples/sqla/admin/static/; } }

proxy_set_header 的三行不能省：不传 X-Real-IP 时，Flask 侧拿到的 request.remote_addr 全是 127.0.0.1，访问日志基本失去分析价值。静态资源单独 alias 到目录，避免 static 请求也打进 Flask 进程，Nginx 直接返回文件的 IO 效率更高。

5. crontab 定时调度、断点续爬与 Redis 队列验证技巧

5.1 定时触发爬虫脚本

crontab 的五段依次是分、时、日、月、周。每天凌晨 2 点执行增量爬取：

0 2 * * * cd /root/python/ti-crawler && /usr/bin/python3 run_crawler.py >> /var/log/ti-crawler.log 2>&1

cd先切换工程目录再执行 python，保证脚本里相对路径的读取和输出位置正确。日志用>>追加而不是>覆盖，出现问题时才有历史数据可查。

5.2 验证 Redis 队列与指纹状态

进程被杀后，先确认队列和指纹是否还在：

redis-cli SCARD misp_event:dupefilter redis-cli LLEN misp_event:requests

SCARD 返回已去重的指纹数量，LLEN 返回待抓取请求的队列长度。如果这两个值在进程重启后仍然存在，说明持久化生效。如果 SCARD 一直增长但 LLEN 始终为 0，爬虫多半卡在解析环节，或者中间件把请求全部吞掉后没有 yield 任何 item。

5.3 用 SQL 快速验证入库效果

mysql -u root -p -e "SELECT source, type, COUNT(*) FROM threat_intel.indicator GROUP BY source, type ORDER BY COUNT(*) DESC LIMIT 10;"

这一行 SQL 是判断爬虫是否“跑废了”的最快手段。如果 domain 数量比 ip 多出几个数量级，IP 正则的合法性校验没生效；如果某个 source 一直是 0，回到对应 spider 的日志文件搜ERROR: Retrying，基本能看到连接超时或代理失效的报错。

真正的断点续爬验证方法是跑一半时kill掉进程，重启继续跑，观察记录数只在原有基础上增加而不是推倒重来。这一步同时验证 SCHEDULER_PERSIST、Redis 持久化配置和 MySQL 唯一键三个环节，建议系统刚搭完、还没接真实数据源时先做一次。

本文还有配套的精品资源，点击获取