Python全栈数据采集分析与可视化平台：爬虫、Flask与手写识别实战

简介：基于Python的全栈式数据采集分析与可视化平台是一套面向教学演示与实战开发的完整资源包，整合网络爬虫抓取、多源异构数据存储、交互式前端展示与深度学习手写数字识别模型，适合高校学生、教师及数据应用开发者用于课程设计、毕设参考或快速搭建原型。压缩包共含2002个文件、约89.64MB，其中1323个Markdown文档承载说明与实验笔记，550个JavaScript文件支撑前端交互，15个Python脚本实现爬虫与模型推理，另有JSON配置、CSS样式和SQL脚本协同完成数据层与展示层搭建。已有73人学习，资料内附赠配套文档与前端样式库，可直接对照目录结构理解从数据采集、存储到可视化、模型识别的完整链路。借助这套资源，读者既能掌握爬虫、数据库、前端可视化与深度学习建模等关键技能，也能节省从零搭建环境的时间，聚焦业务逻辑与教学演示。

1. 全栈式数据采集分析与可视化平台：包里不只是爬虫和手写数字模型

如果你同时带项目或者自己在补全栈技能，手头最缺的往往不是一个算法，而是一条能从数据源头一路跑到页面展示的完整链路。这个基于Python的全栈式数据采集分析与可视化平台，核心价值恰恰在这里：网络爬虫数据抓取、多源异构数据存储管理、交互式前端展示界面、深度学习手写数字识别模型被装进同一个工程，技术栈以Scrapy为主线，从爬虫产出数据、数据落库、接口封装，到前端图表展示和手写数字识别，每一步都能在本地跑通。它不是面向生产环境的高并发系统，而是为教学演示与实战开发准备的一套可复现骨架，适合刚开始接触全栈数据项目的学生，也适合需要快速搭建演示Demo的从业者。

2. 爬虫层：Scrapy工程结构、数据字段与下载策略

下载完压缩包，先别急着跑scrapy crawl，把工程目录和Item结构看明白，后面能省掉大量返工。这份资源里的爬虫部分遵循Scrapy标准工程布局，对熟悉框架的人来说可直接扩展，对刚入门的人来说也是很好的阅读样本。

2.1 工程骨架与Item定义：先想清楚要爬什么字段

打开工程先看items.py。爬虫爬到的每条记录本质是一个字典，Item就是给这个字典规定好字段名和类型的「合同」。字段没定义清楚，后面Pipeline、入库、前端接口全都会跟着乱。

import scrapy class DemoItem(scrapy.Item): title = scrapy.Field() # 标题 url = scrapy.Field() # 原始链接，入库时做唯一约束 author = scrapy.Field() # 作者 publish_time = scrapy.Field() # 发布时间，入库前转成datetime content = scrapy.Field() # 正文或摘要 source_site = scrapy.Field() # 来源站点标识，多源数据靠它区分

这个Item定义了六类信息。title和url是核心字段，url在后续建表时要加UNIQUE约束，用来去重；source_site是故意留出来的，因为平台强调「多源异构」，等数据来自三四个站点时，没有这个字段就无法回答「哪条数据是哪个站点来的」。publish_time不直接用字符串入库，后面Pipeline里会统一解析成datetime，这是异构数据治理的第一步。

再看爬虫本体，压缩包里一般有一个或多个Spider文件，核心逻辑集中在parse()方法里。用CSS选择器抽取页面元素是Scrapy里最直观的写法：

import scrapy from demo_spider.items import DemoItem class DemoSiteSpider(scrapy.Spider): name = "demo_site" # 启动时用 scrapy crawl demo_site start_urls = ["https://example-site.local/list"] def parse(self, response): # 每张卡片是一条记录 for card in response.css("div.card"): item = DemoItem() item["title"] = card.css("h2 a::text").get() item["url"] = response.urljoin(card.css("h2 a::attr(href)").get()) item["author"] = card.css(".author::text").get() item["publish_time"] = card.css(".time::text").get() item["content"] = card.css(".content::text").get() item["source_site"] = "demo_site" yield item

这里的核心逻辑分三步：先定位每条记录的容器节点，再通过CSS选择器从容器里抽出具体字段，最后用yield把Item交给Pipeline。response.urljoin()很关键，它能把页面里的相对链接（比如/detail/123）拼成完整URL，避免入库后拿去访问时报错。.get()只取第一个匹配结果，.getall()返回所有匹配结果的列表，抓列表页时用前者居多。

参数说明：name是爬虫唯一标识，命令行启动时要用；start_urls只负责第一批请求，后续翻页需要自己在parse()里yield scrapy.Request(next_url)手动跟进。如果目标站点改了页面结构，优先调整选择器而不是重写整个Spider。

2.2 settings与中间件：下载延迟、请求头、去重和日志

Scrapy的settings.py是整个爬虫运行策略的集中控制点。这套平台面向教学演示，配置走的是「稳」而不是「快」：宁可爬得慢一点，也别把对方站点压出问题。下面这些配置一般直接能用：

BOT_NAME = "demo_spider" DOWNLOAD_DELAY = 2.0 # 同一域名下两个请求之间的间隔，单位秒 CONCURRENT_REQUESTS = 4 # 全局最大并发请求数 USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" ROBOTSTXT_OBEY = True # 遵守robots协议，演示项目建议开着 ITEM_PIPELINES = { "demo_spider.pipelines.CleanPipeline": 300, "demo_spider.pipelines.SQLitePipeline": 400, } LOG_LEVEL = "INFO"

DOWNLOAD_DELAY和CONCURRENT_REQUESTS是调整爬取压力最常用的两个旋钮。真实数据采集场景里，2秒延迟、4个并发是比较保守的组合；如果目标是本地测试站点，可以把延迟调到0.5、并发调到8，速度会明显提升。ROBOTSTXT_OBEY在演示阶段保持True，一方面培养合规采集习惯，另一方面避免学生拿外网站点测试时造成不必要的访问压力。ITEM_PIPELINES里字典的值是执行顺序，数字小的先跑，所以CleanPipeline负责清洗，SQLitePipeline负责入库，先后关系很明确。

UA池也一并放在这里。很多站点会拦掉默认UA，所以这份资源里一般会配一个随机UA中间件：

# middlewares.py import random USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36", ] class RandomUserAgentMiddleware: def process_request(self, request, spider): request.headers["User-Agent"] = random.choice(USER_AGENTS) return None

同时在settings.py里通过DOWNLOADER_MIDDLEWARES注册这个中间件。逻辑说明：process_request在每个请求发出前执行，random.choice从列表里随机挑一个UA覆写请求头，避免所有请求都带同一个标识。注意中间件和Pipeline的区别——中间件作用于请求/响应链路，Pipeline作用于Item落地链路，两者职责别混。

调试阶段常见做法是用scrapy shell "https://example-site.local/list"进交互环境，随手敲response.css(...)验证选择器是否正确。这比反复跑完整爬虫看日志高效得多，尤其是页面结构变化时，几乎成了排查问题的首选手段。

3. 存储层：多源异构数据的入库流程与去重治理

爬虫产出的是Raw Data，能不能变成平台可用的数据资产，取决于存储设计。标题里「多源异构」四个字不是摆设，这一层的核心矛盾是：数据来源五花八门，格式千差万别，如何统一收口。

3.1 存储选型：SQLite为主、文件目录为辅的策略

教学演示项目不适合一上来就部署MySQL集群，但要体现「多源异构」的工程设计思路。这套平台采用的方案是分层存储：结构化字段进SQLite，原始文件进文件系统，中间JSON保留在导出目录里。

SQLite在这类教学项目里几乎是默认答案：不需要额外安装数据库服务，一个文件就是一个库，学生拷走整个工程目录就能跑。但代价是并发写能力弱，这个坑后面专门讲。图片和正文快照就不建议塞进SQLite了，一方面数据库文件会迅速膨胀，另一方面这类数据很少需要按字段查询，保留文件路径反而更灵活。

工程目录里通常会预留这样的数据组织方式：

data/ demo.db # 结构化数据：文章表、来源表 exports/ raw/ # 爬虫抓回的原始JSON json/ # 清洗后导出的中间JSON files/ images/ # 图片按source_site分目录存放

exports/json是给「不会SQL的人」准备的出口。教学演示时经常需要把数据导出给其他同学分析，统一JSON格式比直接丢一个SQLite文件更友好。多源数据在写入SQLite之前，先在这个目录里过一次手，也方便排查清洗逻辑的问题。

3.2 清洗与入库：类型转换、去重和事务控制

Pipeline是数据入库前的最后一站。我在查看这套资源的Pipeline实现时，最关心的就是三件事：脏数据怎么拦、重复数据怎么去、批量写怎么不锁库。一个合格的SQLitePipeline大概长这样：

import sqlite3 from datetime import datetime from scrapy.exceptions import DropItem def parse_time(value): # 兼容 "2024-01-05 12:30:00" 和 "2024/01/05" 两种常见格式 for fmt in ("%Y-%m-%d %H:%M:%S", "%Y/%m/%d", "%Y-%m-%d"): try: return datetime.strptime(value.strip(), fmt) except (ValueError, AttributeError): continue return None class SQLitePipeline: def open_spider(self, spider): self.conn = sqlite3.connect("data/demo.db") self.conn.execute(""" CREATE TABLE IF NOT EXISTS articles ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, url TEXT UNIQUE, author TEXT, publish_time DATETIME, content TEXT, source_site TEXT ) """) def process_item(self, item, spider): # title为空直接丢弃，不污染数据表 if not item.get("title"): raise DropItem(f"Missing title: {item}") self.conn.execute( "INSERT OR IGNORE INTO articles " "(title, url, author, publish_time, content, source_site) " "VALUES (?, ?, ?, ?, ?, ?)", (item["title"], item["url"], item.get("author"), parse_time(item.get("publish_time")), item.get("content"), item["source_site"]) ) self.conn.commit() return item def close_spider(self, spider): self.conn.close()

这段代码里有几个设计点值得抄进自己的项目。url TEXT UNIQUE是数据库层面的去重约束，配合INSERT OR IGNORE，重复URL直接跳过，不需要在Python里做一次if url in seen的判断。parse_time做多格式兼容，不同站点的发布时间格式几乎不可能统一，解析失败返回None，保留NULL而不是写入脏字符串。DropItem的调用位置在插入前，title为空就丢弃，避免前端拿到空标题的尴尬。

提交策略需要注意：这里每条Item都commit()，在低数据量教学场景中没问题，但爬虫跑几千条时SQLite会越来越慢。更优做法是攒够一定条数或每隔几秒批量提交一次，既能提速又能减少锁冲突。压缩包里一般保留简单写法便于阅读，生产环境要自己改成批量提交。

对于手工数据和API数据，这类平台通常还会附带一个导入脚本，把JSON/CSV里已有的数据合并进同一个SQLite库：

import json, sqlite3 def import_json_export(path): conn = sqlite3.connect("data/demo.db") with open(path, "r", encoding="utf-8") as f: records = json.load(f) for r in records: conn.execute( "INSERT OR IGNORE INTO articles (title, url, author, publish_time, content, source_site) " "VALUES (?, ?, ?, ?, ?, ?)", (r["title"], r["url"], r.get("author"), parse_time(r.get("publish_time")), r.get("content"), r.get("source_site", "manual")) ) conn.commit() conn.close() print(f"imported {len(records)} records")

参数说明：path指向前一步导出目录里的JSON文件，source_site缺省为manual，这样手工导入的数据和爬虫数据能在前端看板里区分开。这里体现的多源异构核心思路是：不同来源在入口处清洗，在表结构上统一，在查询时通过source_site字段回溯来源。

4. 展示层与手写数字识别：接口封装、前端看板和CNN模型

这一章是平台最有「演示效果」的部分。爬虫抓回来的数据通过Flask暴露成接口，前端用图表呈现；手写数字模型同样封装成HTTP接口，前端画板调用后返回识别结果。整个模块的逻辑链是：数据表 → Flask接口 → 前端图表；模型文件 → 推理函数 → Flask接口 → 前端画板。

4.1 Flask数据接口：把数据库和模型统一变成API

Flask在这类工程里承担的是轻量后端角色。它的优势是简单，一个文件就能把数据查询和模型推理都包进来。平台中典型的app.py核心路由有两个：

from flask import Flask, jsonify, request import sqlite3 import torch from model import Net app = Flask(__name__) model = Net() model.load_state_dict(torch.load("model/model.pt", map_location="cpu")) model.eval() @app.route("/api/articles") def articles(): conn = sqlite3.connect("data/demo.db") rows = conn.execute( "SELECT id, title, author, publish_time, source_site " "FROM articles ORDER BY id DESC LIMIT 50" ).fetchall() cols = ["id", "title", "author", "publish_time", "source_site"] return jsonify([dict(zip(cols, row)) for row in rows]) @app.route("/api/predict", methods=["POST"]) def predict(): payload = request.get_json() pixels = payload["pixels"] # 前端传回的28x28灰度列表 tensor = torch.tensor(pixels, dtype=torch.float32).view(1, 1, 28, 28) with torch.no_grad(): logits = model(tensor) pred = logits.argmax(dim=1).item() prob = torch.softmax(logits, dim=1).max().item() return jsonify({"digit": pred, "confidence": round(prob, 4)}) if __name__ == "__main__": app.run(host="127.0.0.1", port=5000, debug=True)

路由设计的要点在pixels的数据形状转换。前端canvas是280×280的绘制区域，传给后端前要先缩放到28×28并归一化，后端拿到后view(1, 1, 28, 28)变成模型要求的(batch, channel, height, width)。模型文件在服务启动时加载一次，放在路由外面，避免每次请求都重新读盘——这是推理服务最基本的性能习惯。map_location="cpu"让模型在无GPU的机器上也能跑，教学环境绝大多数是CPU。

如果你要对接其他前端页面，跨域是个绕不开的问题。Flask默认不允许跨域请求，前端页面如果跑在不同端口，浏览器会拦截响应。常见做法是装flask-cors，然后初始化时加一句CORS(app)。这类平台的前后端通常都在本地演示，不配置CORS的话，页面经常是「接口有数据、浏览器全报错」的诡异状态。

4.2 交互式前端：图表轮询与手写画板

前端文件一般是一个静态HTML加ECharts、原生JavaScript，不引入前端脚手架，目的是降低阅读门槛。核心功能有两个：定时刷新数据图表、canvas手写数字识别。

<canvas id="board" width="280" height="280"></canvas> <button id="predictBtn">识别</button> <div id="result">识别结果：</div> <script> // 每5秒拉一次最新文章数据，更新来源分布柱状图 async function refreshChart() { const res = await fetch("/api/articles"); const list = await res.json(); const stats = {}; list.forEach(item => { stats[item.source_site] = (stats[item.source_site] || 0) + 1; }); // 用stats更新ECharts的series.data } // 把canvas内容缩小成28x28的灰度数组 function getPixels() { const canvas = document.getElementById("board"); const ctx = canvas.getContext("2d"); const imgData = ctx.getImageData(0, 0, 280, 280); const resized = new Array(28 * 28); // 遍历280x280像素，按区块平均降采样到28x28 for (let i = 0; i < 28; i++) { for (let j = 0; j < 28; j++) { let sum = 0; for (let dy = 0; dy < 10; dy++) { for (let dx = 0; dx < 10; dx++) { const idx = ((i * 10 + dy) * 280 + (j * 10 + dx)) * 4; sum += imgData.data[idx]; // 取R通道，灰度下RGB相等 } } resized[i * 28 + j] = sum / 100 / 255; // 归一化到0~1 } } return resized; } </script>

refreshChart的轮询逻辑是教学场景里常见做法：每5秒调用一次接口，数据变了图表自己跟着变，免去手动刷新页面。相比WebSocket，轮询牺牲实时性换来了实现简单，对演示平台来说足够。getPixels是整个手写识别前端里最容易出错的地方——它把280×280画板按10×10区块平均降采样到28×28，然后除以255做归一化。前端画板若是黑底白字，这里取到的R通道值分布和白底黑字完全相反，模型预测结果会天差地别，这一点在避坑章节还会重点讲。

4.3 CNN手写数字模型：网络结构、训练配置与落盘

手写数字识别模型基于PyTorch，用的是一组经典CNN结构。这个网络在MNIST这类数据集上是「基本功」级别的存在，参数量不大，CPU上几十秒就能完成一轮训练，非常适合教学演示。

import torch.nn as nn import torch.nn.functional as F class Net(nn.Module): def __init__(self): super().__init__() self.conv1 = nn.Conv2d(1, 32, 3, padding=1) # 输入1通道，输出32通道 self.conv2 = nn.Conv2d(32, 64, 3, padding=1) # 32 -> 64 self.pool = nn.MaxPool2d(2) # 2x2最大池化 self.fc1 = nn.Linear(64 * 7 * 7, 128) self.fc2 = nn.Linear(128, 10) # 10个数字类别 def forward(self, x): x = self.pool(F.relu(self.conv1(x))) # 28 -> 14 x = self.pool(F.relu(self.conv2(x))) # 14 -> 7 x = x.view(x.size(0), -1) # 展平，64*7*7 x = F.relu(self.fc1(x)) return self.fc2(x)

网络结构解析：两个卷积层各接一个ReLU和2×2最大池化，28×28输入经过两次池化变成7×7，通道数从1涨到64，展平后经过两个全连接层输出10类得分。padding=1保证卷积不改变特征图尺寸，所以尺寸变化完全由池化层控制。训练部分一般配套一份train.py：

from torch import optim import torch model = Net() optimizer = optim.Adam(model.parameters(), lr=0.001) loss_fn = nn.CrossEntropyLoss() # 训练5个epoch，batch_size=64 for epoch in range(5): for images, labels in train_loader: optimizer.zero_grad() outputs = model(images) loss = loss_fn(outputs, labels) loss.backward() optimizer.step() print(f"epoch {epoch+1} finished") torch.save(model.state_dict(), "model/model.pt")

训练参数有三个值得留意：lr=0.001是Adam在MNIST这种简单任务上的惯例起点，改到0.01容易震荡、改到0.0001收敛过慢；5个epoch在CPU上也就几分钟，验证集准确率一般在98%以上，再增大epoch收益变低，教学场景不需要刷点；batch_size=64在显存或内存有限的机器上比较稳妥，调大能略微加速但会明显吃内存。model.state_dict()只保存权重不保存网络结构，所以推理时Net()和load_state_dict必须成对出现。

5. 避坑与常见问题：三条链路各自的翻车现场

这类多模块项目，跑通一次不难，难的是换了数据源、换了机器、换了Python版本之后还能稳定跑。下面是我拆解这套平台时整理的踩坑记录，每一条都是真实项目里出现过的现象，按「现象 → 原因 → 解决」列出。

5.1 爬虫与存储侧问题

现象一：启动scrapy crawl demo_site后一个请求都不发，日志停留在抓取robots.txt，最后整个爬虫直接退出。原因：ROBOTSTXT_OBEY = True时，Scrapy会先请求目标站点的robots.txt。测试站点如果没配置robots或者对非浏览器UA返回403，爬虫会认为「不允许抓取」而拒绝发送后续请求。解决：本地演示环境把ROBOTSTXT_OBEY改成False，或者给测试站点补一个允许抓取的robots.txt。正式采集时再根据目标站点策略决定是否开启，不要用同一套配置通吃所有场景。

现象二：爬虫跑到一半，Pipeline报sqlite3.OperationalError: database is locked。原因：SQLite同一时刻只允许一个写者。多个爬虫进程并发跑，或者Pipeline里频繁 commit 导致写锁冲突。解决：先检查是否有多个爬虫进程在同时写同一个数据库文件，教学场景强烈建议一次只跑一个爬虫；再把Pipeline改成批量提交，攒满50条或每隔3秒提交一次，减少锁竞争。如果数据量真的上来了，就该换MySQL或PostgreSQL。

现象三：入库后打开数据库，中文全部是乱码，或者报UnicodeEncodeError。原因：目标站点响应编码不是UTF-8（常见于GBK老站点），Scrapy默认按UTF-8解码，解码失败就产生乱码。解决：在Spider里对response显式指定编码，常见做法是先打印response.encoding确认，再设置response.encoding = "gbk"，然后重新抽取字段。这个位置属于「设一次就再也不会碰到」的坑，但第一次碰上确实恼火。

现象四：爬虫正常跑完，日志显示item_scraped_count: 0，数据表里干干净净一行都没有。原因：页面结构变了，CSS选择器没匹配到任何元素。目标站点改版、加了登录墙、接口改成动态渲染，都会触发这个问题。解决：用scrapy shell把页面拉下来确认当前结构，重新写选择器。另外可以看一眼DOWNLOADER_MIDDLEWARES里是否加了处理JS渲染的中间件——没有的话，动态加载内容自然抓不到。

5.2 前端与模型侧问题

现象五：接口直接访问有正常JSON，浏览器前端页面里图表一直空白，控制台报CORS错误。原因：Flask服务跑在5000端口，前端页面跑在另一个端口或直接用file协议打开，跨域请求被浏览器拦截。即使两个服务都在本机，不同端口之间也属于跨域。解决：安装flask-cors，在app里加CORS(app, resources={r"/api/*": {"origins": "*"}})；或者让Flask直接托管静态页面，把前端HTML放进templates或static，浏览器和接口同源，彻底绕开跨域。

现象六：模型训练时验证集准确率98%，但前端画板里写数字，识别结果一塌糊涂，甚至稳定预测成同一个数。原因：几乎都是预处理不一致。训练时MNIST数据是白底黑字、28×28、数值归一化到0-1；前端canvas传过来的可能是黑底白字，或者没有归一化，或者图像缩放方式跟训练集不一样。模型的输入分布变了，输出自然崩。解决：把前端预处理统一成训练时的标准——灰度图、反色（如果训练集是白底黑字）、缩放28×28、逐像素除以255。这个坑最隐蔽的地方在于缩放算法：280降采样到28，简单跳跃取点和区块平均取点的结果不同，推荐用和MNIST转换一致的均值降采样，也就是前面getPixels里的写法。

6. 进阶：把四条链路串起来做端到端验证

多模块项目最怕「单模块都正常、一连起来就完蛋」。拿到这个压缩包后，我强烈建议你按下面的顺序跑一遍最小闭环，任何一步卡住，问题范围立刻缩小到具体模块，不用满工程排查。

第一步验证爬虫链路。在工程根目录执行：

scrapy crawl demo_site

跑完后用SQL查一下数据落库情况：

sqlite3 data/demo.db "SELECT source_site, count(*) FROM articles GROUP BY source_site;"

看到每个来源站点都有记录，说明爬虫、Pipeline、SQLite这一侧通了。如果source_site字段全是空，回去查Item里有没有给每个Spider赋值——这是新手最容易漏的一步。

第二步验证接口链路。启动Flask服务，另开终端：

python app.py curl http://127.0.0.1:5000/api/articles

返回JSON数组就说明数据库到接口通了。再验证模型接口：

curl -X POST http://127.0.0.1:5000/api/predict \ -H "Content-Type: application/json" \ -d '{"pixels": [0, 0, ...]}'

这里pixels数组我用省略号略写了，实际传的时候要传满28×28共784个数。

第三步验证前端展示。浏览器打开看板页，图表有数据、画板能识别，整个闭环就通了。我一般会用一张手写的数字5来做基准测试——如果前端画板把5断成3或9，优先怀疑预处理不一致，而不是模型训练不够。

验证完成后，后续扩展方向也很清晰：爬虫层可以加Cron定时调度做增量抓取，存储层可以把SQLite平滑替换成MongoDB体现真正的多源异构，模型层还能换ResNet变体或者加入数据增强提升鲁棒性，前端可以引入WebSocket替代5秒轮询。这套工程的模块边界比较干净，每个部分都是可替换的，这正是它适合实战开发的原因。

回头看我自己第一次拆这种多模块工程时，图省事跳过了端到端验证，直接闷头改模型结构，结果改完发现前端图表一直是空的——折腾了一下午，最后定位到是CORS没配。从那以后，我每次拿到这类平台包，都强制先跑一遍从爬虫到前端的最小闭环，再往里加功能。顺序对了，排错成本能降一个数量级。希望这个习惯也能帮到你少走这一步弯路。

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