尧图网页设计 
简介本资源是一个面向数据分析初学者与Web开发学习者的综合性实战项目聚焦Steam游戏市场趋势与用户行为挖掘提供从数据爬取、存储、处理到可视化展示的完整技术闭环。项目基于Flask构建轻量级Web平台融合Python数据处理含12个核心py脚本、SQL数据库操作3个sql文件及前端交互式图表依赖84个js、25个css及Bootstrap/FullCalendar/Summernote等前端库支持多维度统计分析与动态响应。压缩包共206个文件总大小27.87MB涵盖HTML页面、字体资源、静态图片、配置文件yml/cnf、数据样本csv/json及说明文档pdf/md结构清晰、模块解耦便于分阶段学习与二次开发。目前已有91人下载学习适合希望掌握大数据分析全流程、提升前后端协同开发能力的学习者可直接部署运行并基于现有代码拓展爬虫策略或可视化维度。1. 这不是又一个“爬 Steam 再画个折线图”的玩具项目它是一套能跑在单机上的轻量级大数据分析闭环——从requests抓取真实游戏页、用pandas做多维聚合、靠Flask暴露 REST 接口前端交互图表、最后用Redis缓存高频查询结果全程不依赖 Hadoop/Spark 集群但保留了大数据处理的典型分层逻辑采集→存储→清洗→计算→服务→展示适合课程设计、毕设、技术栈验证或中小团队快速验证 Steam 市场假设。如果你正卡在「想分析 Steam 游戏销量/评价/价格变化却找不到结构化数据源想用 Python 做全栈但总在 Flask 路由和 ECharts 数据对接上反复翻车或者被导师问『你这个可视化背后的数据 pipeline 是什么』答不出」——那这个项目就是为你写的实操锚点。它不吹“实时流处理”但把离线批处理的每个环节都踩实不堆炫酷大屏但每张图表背后都有可追溯的 SQL 或 DataFrame 操作链。2. 用 Flask 搭建最小可行服务骨架路由设计、静态资源托管与前后端数据桥接2.1 初始化 Flask 应用并配置关键参数先创建项目根目录steam_analytics/进入后执行mkdir -p app/{templates,static/{css,js,images}} pip install flask pandas numpy redis requests beautifulsoup4 plotly新建app/__init__.pyfrom flask import Flask import redis def create_app(): app Flask(__name__) # 关键配置禁用模板自动转义Plotly 图表需 raw HTML app.jinja_env.autoescape False # Redis 连接池本地开发用默认端口生产环境建议加密码和连接超时 app.redis_client redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue, # 返回字符串而非字节 socket_connect_timeout2, socket_timeout2 ) # 注册蓝图后续按功能拆分 from app.routes import main_bp app.register_blueprint(main_bp) return app提示decode_responsesTrue是血泪经验——若不设从 Redis 读出的所有值都是bxxx字节串后续 JSON 解析会直接报TypeError: the JSON object must be str, bytes or bytearray。这不是 Flask 的锅是 Redis-Py 默认行为。2.2 定义核心路由让前端能拿到结构化数据而非 HTML 页面在app/routes.py中定义两个关键接口from flask import Blueprint, jsonify, render_template from app import create_app import json main_bp Blueprint(main, __name__) main_bp.route(/) def index(): return render_template(index.html) # 前端主页面 main_bp.route(/api/games/top_sales) def top_sales(): 返回销量 Top 10 游戏的 JSON 数据模拟实际从 Redis 或 DB 读 # 实际项目中这里应调用 service 层方法如 get_top_sales_from_cache() mock_data [ {name: Cyberpunk 2077, sales_millions: 25.5, year: 2020}, {name: Elden Ring, sales_millions: 22.3, year: 2022}, {name: Stardew Valley, sales_millions: 15.8, year: 2016}, {name: Terraria, sales_millions: 14.2, year: 2011}, {name: Hades, sales_millions: 10.1, year: 2020} ] return jsonify({data: mock_data, total: len(mock_data)}) main_bp.route(/api/games/price_trend/game_id) def price_trend(game_id): 按 game_id 返回价格历史示例返回近 30 天均价波动 # 真实场景中 game_id 对应数据库主键或 Steam AppID trend_data [ {date: 2024-04-01, price_cny: 198.0}, {date: 2024-04-02, price_cny: 198.0}, {date: 2024-04-03, price_cny: 188.0}, # 打折日 {date: 2024-04-04, price_cny: 188.0}, {date: 2024-04-05, price_cny: 198.0} ] return jsonify({game_id: game_id, trend: trend_data})参数说明/api/games/top_sales是无参 GET 接口返回固定结构 JSON供前端 ECharts 初始化调用/api/games/price_trend/game_id使用 Flask URL 变量捕获game_id会被自动解析为字符串传入函数避免手动解析 query string所有接口统一用jsonify()包装确保 Content-Type 自动设为application/json省去手动设置 header。2.3 前端页面与图表初始化用 ECharts 加载动态数据在app/templates/index.html中!DOCTYPE html html head meta charsetutf-8 titleSteam 游戏数据分析平台/title script srchttps://cdn.jsdelivr.net/npm/echarts5.4.3/dist/echarts.min.js/script link relstylesheet href{{ url_for(static, filenamecss/style.css) }} /head body div idtop-sales-chart stylewidth: 800px; height: 400px;/div div idprice-trend-chart stylewidth: 800px; height: 400px;/div script // 初始化销量 Top10 柱状图 const salesChart echarts.init(document.getElementById(top-sales-chart)); fetch(/api/games/top_sales) .then(res res.json()) .then(data { const names data.data.map(d d.name); const sales data.data.map(d d.sales_millions); salesChart.setOption({ title: { text: Steam 销量 Top 5 游戏百万份 }, tooltip: {}, xAxis: { data: names }, yAxis: {}, series: [{ type: bar, data: sales, label: { show: true } }] }); }); // 初始化价格趋势折线图默认加载 Cyberpunk 2077 const trendChart echarts.init(document.getElementById(price-trend-chart)); fetch(/api/games/price_trend/238460) // Cyberpunk 2077 的 Steam AppID .then(res res.json()) .then(data { const dates data.trend.map(d d.date); const prices data.trend.map(d d.price_cny); trendChart.setOption({ title: { text: Cyberpunk 2077 近期价格趋势¥ }, tooltip: { trigger: axis }, xAxis: { type: category, data: dates }, yAxis: { type: value }, series: [{ type: line, data: prices, smooth: true, areaStyle: {} }] }); }); /script /body /html逻辑说明前端不写死数据所有图表数据通过fetch()从 Flask API 异步拉取url_for(static, ...)是 Flask 内置函数生成正确静态资源路径避免硬编码/static/...导致部署时 404ECharts 初始化必须等 DOM 元素渲染完成此处直接写在script标签内因页面简单复杂项目建议用DOMContentLoaded事件封装。3. Steam 数据采集绕过反爬、解析 HTML、提取关键字段的稳定方案3.1 为什么不用 Steam 官方 API——现实约束下的选型依据Steam 官方提供 Web API 需申请 Key但其公开接口极度受限GetAppList仅返回 AppID 和名称无价格、评价、发行日期GetGlobalAchievementPercentagesForApp等接口需绑定具体游戏且调用频次极低所有用户行为数据如愿望单数、评测数、在线人数均未开放更致命的是Steam 官网 HTML 结构稳定、字段语义清晰、无 JS 渲染依赖——这意味着requests BeautifulSoup足够可靠且比逆向 WebSocket 或模拟登录更轻量、更易维护。常见误用对比❌ 用 Selenium 模拟浏览器启动慢、内存占用高、Headless Chrome 易被识别❌ 直接抓取store.steampowered.com/app/xxx的原始 HTML部分字段如当前价格由 JS 动态注入requests拿不到✅ 正确做法抓取store.steampowered.com/app/xxx/?snr1_7_7_151_150_1带snr参数的 URL该链接返回完整 HTML价格、评测、发行日期等全部静态存在。3.2 构建稳健爬虫User-Agent 轮换、请求间隔、错误重试新建crawler/steam_crawler.pyimport requests from bs4 import BeautifulSoup import time import random from urllib.parse import urljoin class SteamCrawler: def __init__(self): self.session requests.Session() # 设置通用 headers模拟真实浏览器 self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en-US;q0.8,en;q0.7, Accept-Encoding: gzip, deflate, Connection: keep-alive, }) # 设置请求间隔避免触发风控 self.delay_range (1.5, 3.0) def get_game_page(self, app_id): 获取单个游戏详情页 HTML url fhttps://store.steampowered.com/app/{app_id}/?snr1_7_7_151_150_1 try: resp self.session.get(url, timeout10) resp.raise_for_status() if Sorry, we couldnt find that page. in resp.text: return None # 游戏下架或 ID 无效 return resp.text except requests.exceptions.RequestException as e: print(f[ERROR] Failed to fetch {url}: {e}) return None def parse_game_info(self, html, app_id): 解析 HTML 提取关键字段 soup BeautifulSoup(html, html.parser) # 游戏名称必填 name_tag soup.find(div, class_apphub_AppName) name name_tag.get_text(stripTrue) if name_tag else fUnknown Game ({app_id}) # 当前价格含折扣逻辑 price_tag soup.find(div, class_game_purchase_price) if not price_tag: price_tag soup.find(div, class_discount_original_price) # 折扣价原价 current_price price_tag.get_text(stripTrue) if price_tag else Free # 发行日期 release_tag soup.find(div, class_date) release_date release_tag.get_text(stripTrue) if release_tag else Unknown # 用户评测好评率 总数 review_summary soup.find(span, class_game_review_summary) review_text review_summary.get_text(stripTrue) if review_summary else No reviews # 愿望单数需定位到特定 div wishlist_tag soup.find(div, {data-tooltip-text: True}) wishlist_text if wishlist_tag and wishlisted in wishlist_tag.get(data-tooltip-text, ): wishlist_text wishlist_tag.get(data-tooltip-text).split()[0].replace(,, ) return { app_id: app_id, name: name, current_price: current_price, release_date: release_date, review_summary: review_text, wishlist_count: int(wishlist_text) if wishlist_text.isdigit() else 0 } def crawl_single_game(self, app_id): 完整爬取单个游戏流程 html self.get_game_page(app_id) if not html: return None time.sleep(random.uniform(*self.delay_range)) # 随机延迟 return self.parse_game_info(html, app_id) # 使用示例 if __name__ __main__: crawler SteamCrawler() result crawler.crawl_single_game(238460) # Cyberpunk 2077 print(result)参数说明snr1_7_7_151_150_1是 Steam 商店 URL 的标准参数确保返回完整页面session.headers.update()统一设置 UA 和 Accept 头避免单次请求缺失导致 403time.sleep(random.uniform(1.5, 3.0))是防封关键固定 2 秒易被模式识别浮动区间更自然soup.find(div, class_game_purchase_price)是精准定位比模糊搜索price文本更可靠。3.3 批量爬取与失败重试机制用队列管理 AppID 并记录日志新建crawler/batch_crawl.pyimport json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from steam_crawler import SteamCrawler # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(crawler.log, encodingutf-8), logging.StreamHandler() ] ) def crawl_task(app_id, crawler): 单任务封装便于线程池调度 try: result crawler.crawl_single_game(app_id) if result: logging.info(f✅ Success: {result[name]} (AppID {app_id})) return result else: logging.warning(f⚠️ Skipped: AppID {app_id} (page not found or parsing failed)) return None except Exception as e: logging.error(f❌ Error on AppID {app_id}: {e}) return None def batch_crawl(app_ids, max_workers5): 批量爬取入口 crawler SteamCrawler() results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_appid {executor.submit(crawl_task, aid, crawler): aid for aid in app_ids} # 收集结果 for future in as_completed(future_to_appid): result future.result() if result: results.append(result) # 保存结果 with open(data/raw_games.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) logging.info(f Crawled {len(results)}/{len(app_ids)} games. Saved to data/raw_games.json) if __name__ __main__: # 示例爬取 10 个热门游戏实际项目中从 SteamDB 或社区榜单获取 sample_appids [238460, 1245620, 271590, 1086940, 730, 570, 304930, 413150, 252490, 1172380] batch_crawl(sample_appids)逻辑说明ThreadPoolExecutor控制并发数max_workers5既保证效率又避免 IP 被限as_completed()按完成顺序收集结果无需等待全部结束日志同时输出到文件和控制台crawler.log可用于事后审计失败请求输出raw_games.json是原始采集层绝不在此阶段做清洗或计算——保持数据血缘可追溯。4. 数据清洗与多维统计用 Pandas 构建 Steam 分析模型的核心计算层4.1 从原始 JSON 到结构化 DataFrame处理缺失值与类型转换新建analysis/data_processor.pyimport pandas as pd import re from datetime import datetime def load_raw_data(filepathdata/raw_games.json): 加载原始爬取数据 with open(filepath, r, encodingutf-8) as f: data json.load(f) return pd.DataFrame(data) def clean_price(price_str): 清洗价格字符串提取数字处理 Free 和 ¥ 符号 if not isinstance(price_str, str): return 0.0 if Free in price_str: return 0.0 # 匹配 ¥ 后的数字支持千分位逗号 match re.search(r¥\s*([\d,]\.?\d*), price_str) if match: # 去除逗号并转浮点 return float(match.group(1).replace(,, )) return 0.0 def parse_release_date(date_str): 标准化发行日期统一为 YYYY-MM-DD if not isinstance(date_str, str): return pd.NaT try: # Steam 日期格式如 Nov 18, 2020 或 Feb 21, 2023 return pd.to_datetime(date_str, errorscoerce) except: return pd.NaT def build_analysis_df(): 构建用于分析的 DataFrame df load_raw_data() # 清洗价格列 df[price_cny] df[current_price].apply(clean_price) # 解析发行日期 df[release_date] df[release_date].apply(parse_release_date) df[release_year] df[release_date].dt.year # 提取好评率如 Overwhelmingly Positive → 95% # 实际项目中需更复杂规则此处简化 df[review_score] df[review_summary].map({ Overwhelmingly Positive: 0.95, Very Positive: 0.85, Mostly Positive: 0.75, Positive: 0.65, Mixed: 0.5, Negative: 0.3, Mostly Negative: 0.2, Very Negative: 0.1, Overwhelmingly Negative: 0.05 }).fillna(0.5) # 愿望单数转为数值 df[wishlist_count] pd.to_numeric(df[wishlist_count], errorscoerce).fillna(0).astype(int) return df # 使用示例 if __name__ __main__: df build_analysis_df() print(df[[name, price_cny, release_year, review_score, wishlist_count]].head())参数说明clean_price()用正则匹配¥后数字兼容¥198和¥ 198.00parse_release_date()用pd.to_datetime(..., errorscoerce)将无法解析的日期转为NaTNot a Time避免中断review_score映射表基于 Steam 评测等级官方定义非主观打分errorscoerce在pd.to_numeric()中至关重要——若某行wishlist_count是12,345带逗号不加此参数会直接报错。4.2 多维度统计实战按年份、价格区间、好评率分组聚合继续在data_processor.py中添加def generate_reports(df): 生成多维度统计报告 reports {} # 1. 按发行年份统计游戏数量 平均价格 yearly_stats df.groupby(release_year).agg( game_count(name, count), avg_price(price_cny, mean), median_price(price_cny, median), total_wishlist(wishlist_count, sum) ).round(2).reset_index() reports[yearly_stats] yearly_stats # 2. 按价格区间分桶0元、1-100元、101-200元、200元 bins [-1, 0, 100, 200, float(inf)] labels [Free, 1-100¥, 101-200¥, 200¥] df[price_range] pd.cut(df[price_cny], binsbins, labelslabels) price_range_stats df.groupby(price_range).agg( game_count(name, count), avg_review_score(review_score, mean), avg_wishlist_per_game(wishlist_count, mean) ).round(3).reset_index() reports[price_range_stats] price_range_stats # 3. 高愿望单游戏 Top 10排除免费游戏 top_wishlist df[df[price_cny] 0].nlargest(10, wishlist_count)[ [name, price_cny, release_year, review_score, wishlist_count] ].copy() top_wishlist[wishlist_count_formatted] top_wishlist[wishlist_count].apply( lambda x: f{x:,} # 添加千分位逗号 ) reports[top_wishlist] top_wishlist return reports # 保存报告为 CSV供 BI 工具或下游使用 def save_reports(reports): for name, df_report in reports.items(): df_report.to_csv(fdata/reports/{name}.csv, indexFalse, encodingutf-8-sig) if __name__ __main__: df build_analysis_df() reports generate_reports(df) save_reports(reports) print(✅ Reports generated:) for name, df_r in reports.items(): print(f- {name}: {len(df_r)} rows)逻辑说明pd.cut()实现价格分桶比if-elif更向量化、更易维护nlargest(10, wishlist_count)比sort_values().head(10)更高效尤其大数据集encodingutf-8-sig是 Windows Excel 打开中文 CSV 不乱码的关键——utf-8本身没问题但 Excel 默认用 ANSI加 BOM 才能识别所有统计结果保存为 CSV不直接存入数据库——CSV 是最通用、最易审计的中间产物。4.3 避坑Steam 数据清洗的 4 个高频翻车点现象 1price_cny列全是0.0但网页明明显示价格原因Steam 页面中价格可能以不同 class 名出现如game_purchase_price,discount_final_price,bundle_discount_amount而爬虫只查了第一个或价格被包裹在span内get_text()未递归提取。解决在parse_game_info()中扩展价格查找逻辑# 替换原 price_tag 查找 price_selectors [ div.game_purchase_price, div.discount_final_price, div.bundle_discount_amount ] for selector in price_selectors: price_tag soup.select_one(selector) if price_tag: current_price price_tag.get_text(stripTrue) break else: current_price Free现象 2release_date解析失败整列变NaT原因Steam 日期格式不统一除Nov 18, 2020外还有Coming soon、TBD、Q4 2024等非标准值。解决增强parse_release_date()def parse_release_date(date_str): if not isinstance(date_str, str): return pd.NaT # 过滤明显无效值 invalid_patterns [Coming soon, TBD, Q, Est.] if any(p in date_str for p in invalid_patterns): return pd.NaT try: return pd.to_datetime(date_str, errorscoerce) except: return pd.NaT现象 3wishlist_count为12,345字符串to_numeric()后变NaN原因pd.to_numeric()默认不处理千分位逗号需先清洗。解决在build_analysis_df()中预处理df[wishlist_count] df[wishlist_count].astype(str).str.replace(,, ).astype(float)现象 4review_score映射后大量NaN导致分组统计失真原因review_summary字段值如Very Positive (12,345)而映射表只匹配Very Positive。解决用正则提取关键词def extract_review_level(text): if not isinstance(text, str): return Mixed match re.search(r(Overwhelmingly|Very|Mostly|) (Positive|Negative), text) if match: return match.group(0) return Mixed df[review_level] df[review_summary].apply(extract_review_level) df[review_score] df[review_level].map({...}) # 映射表同上5. 数据持久化与缓存策略SQLite 存原始数据Redis 缓存聚合结果5.1 用 SQLite 建模轻量、零运维、ACID 可靠新建storage/db_manager.pyimport sqlite3 import pandas as pd from pathlib import Path DB_PATH Path(data/steam_analysis.db) def init_db(): 初始化 SQLite 数据库表结构 conn sqlite3.connect(DB_PATH) cursor conn.cursor() # 游戏主表 cursor.execute( CREATE TABLE IF NOT EXISTS games ( app_id TEXT PRIMARY KEY, name TEXT NOT NULL, current_price TEXT, release_date TEXT, review_summary TEXT, wishlist_count INTEGER DEFAULT 0, crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 统计快照表每日/每周聚合结果 cursor.execute( CREATE TABLE IF NOT EXISTS daily_stats ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, report_type TEXT NOT NULL, -- yearly, price_range, top_wishlist data_json TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def save_games_to_db(df): 将清洗后的 DataFrame 存入 SQLite conn sqlite3.connect(DB_PATH) # replace 模式每次全量覆盖适合小规模分析 df.to_sql(games, conn, if_existsreplace, indexFalse) conn.close() print(f✅ Saved {len(df)} games to SQLite) def save_daily_stats(report_name, data_df): 保存统计报告到 daily_stats 表 conn sqlite3.connect(DB_PATH) data_json data_df.to_json(orientrecords, force_asciiFalse) cursor conn.cursor() cursor.execute( INSERT INTO daily_stats (date, report_type, data_json) VALUES (?, ?, ?), (pd.Timestamp.now().strftime(%Y-%m-%d), report_name, data_json) ) conn.commit() conn.close() # 使用示例 if __name__ __main__: init_db() df pd.read_csv(data/reports/yearly_stats.csv) save_daily_stats(yearly_stats, df)参数说明if_existsreplace是关键避免append模式下重复插入导致主键冲突data_json TEXT存储 JSON 字符串而非建宽表——灵活应对报告结构变化crawled_at和created_at时间戳为数据溯源提供基础。5.2 Redis 缓存设计为高频 API 接口加速回到app/__init__.py增强 Redis 使用# 在 create_app() 中追加 def init_redis_cache(app): 初始化 Redis 缓存策略 app.before_first_request def setup_cache(): # 预热将常用统计结果存入 Redis try: # 从 SQLite 读 yearly_stats 并缓存 conn sqlite3.connect(data/steam_analysis.db) yearly_df pd.read_sql_query(SELECT * FROM daily_stats WHERE report_typeyearly_stats ORDER BY id DESC LIMIT 1, conn) conn.close() if not yearly_df.empty: latest_data json.loads(yearly_df.iloc[0][data_json]) app.redis_client.setex(yearly_stats, 3600, json.dumps(latest_data, ensure_asciiFalse)) print(✅ Cached yearly_stats to Redis (TTL1h)) except Exception as e: print(f[WARN] Failed to preload cache: {e}) # 在 create_app() 末尾调用 init_redis_cache(app) return app然后修改app/routes.py中的top_sales接口main_bp.route(/api/games/top_sales) def top_sales(): 从 Redis 缓存读取未命中则回退到计算 cached app.redis_client.get(yearly_stats) if cached: return jsonify({data: json.loads(cached), source: redis}) # 回退逻辑从 SQLite 读取并重新计算此处简化为 mock mock_data [{name: Cyberpunk 2077, sales_millions: 25.5, year: 2020}] return jsonify({data: mock_data, source: fallback})逻辑说明setex(key, 3600, value)设置键值对并指定 1 小时过期避免脏数据长期滞留before_first_request确保应用启动时预热缓存首次请求不经历冷启动接口返回source: redis字段便于前端或监控系统判断数据新鲜度。5.3 避坑SQLite 与 Redis 协同使用的 3 个边界问题现象 1SQLite 写入后Redis 缓存未更新前端看到旧数据原因数据写入 SQLite 和刷新 Redis 是两个独立操作缺乏事务一致性。解决在save_daily_stats()后主动刷新 Redisdef save_daily_stats(report_name, data_df): conn sqlite3.connect(DB_PATH) data_json data_df.to_json(orientrecords, force_asciiFalse) cursor conn.cursor() cursor.execute(INSERT INTO daily_stats ..., (...)) conn.commit() conn.close() # 同步刷新 Redis from app import create_app app create_app() app.redis_client.setex(f{report_name}_latest, 3600, data_json)现象 2Redis 内存爆满OOM command not allowed when used memory maxmemory原因未设置 Redis 最大内存或淘汰策略。解决启动 Redis 时指定配置redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru或在redis.conf中设置maxmemory 512mb maxmemory-policy allkeys-lru现象 3SQLite 数据库被多个进程同时写入报database is locked原因Flask 默认多线程sqlite3.connect()在多线程下未加锁。解决使用check_same_threadFalse并启用 WAL 模式conn sqlite3.connect(DB_PATH, check_same_threadFalse) conn.execute(PRAGMA journal_modeWAL) # 启用 WAL提升并发写性能6. 交互式可视化进阶用 Plotly Dash 嵌入 Flask实现筛选联动与导出功能6.1 Dash 与 Flask 集成复用 Flask 的路由和 Redis 连接新建visualization/dash_app.pyimport dash from dash import dcc, html, Input, Output, callback import plotly.express as px import pandas as pd import json from app import create_app # 复用主 Flask 应用的 Redis 实例 # 创建 Dash 实例挂载到 Flask 上 server create_app() app dash.Dash( __name__, serverserver, url_base_pathname/dashboard/, # 注意斜杠结尾 external_stylesheets[/static/css/style.css] ) # 从 Redis 读取数据复用 p a hrefhttps://download.csdn.net/download/weixin_43880734/91933863 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p Steam游戏数据分析实战：Python全栈闭环方案
简介本资源是一个面向数据分析初学者与Web开发学习者的综合性实战项目聚焦Steam游戏市场趋势与用户行为挖掘提供从数据爬取、存储、处理到可视化展示的完整技术闭环。项目基于Flask构建轻量级Web平台融合Python数据处理含12个核心py脚本、SQL数据库操作3个sql文件及前端交互式图表依赖84个js、25个css及Bootstrap/FullCalendar/Summernote等前端库支持多维度统计分析与动态响应。压缩包共206个文件总大小27.87MB涵盖HTML页面、字体资源、静态图片、配置文件yml/cnf、数据样本csv/json及说明文档pdf/md结构清晰、模块解耦便于分阶段学习与二次开发。目前已有91人下载学习适合希望掌握大数据分析全流程、提升前后端协同开发能力的学习者可直接部署运行并基于现有代码拓展爬虫策略或可视化维度。1. 这不是又一个“爬 Steam 再画个折线图”的玩具项目它是一套能跑在单机上的轻量级大数据分析闭环——从requests抓取真实游戏页、用pandas做多维聚合、靠Flask暴露 REST 接口前端交互图表、最后用Redis缓存高频查询结果全程不依赖 Hadoop/Spark 集群但保留了大数据处理的典型分层逻辑采集→存储→清洗→计算→服务→展示适合课程设计、毕设、技术栈验证或中小团队快速验证 Steam 市场假设。如果你正卡在「想分析 Steam 游戏销量/评价/价格变化却找不到结构化数据源想用 Python 做全栈但总在 Flask 路由和 ECharts 数据对接上反复翻车或者被导师问『你这个可视化背后的数据 pipeline 是什么』答不出」——那这个项目就是为你写的实操锚点。它不吹“实时流处理”但把离线批处理的每个环节都踩实不堆炫酷大屏但每张图表背后都有可追溯的 SQL 或 DataFrame 操作链。2. 用 Flask 搭建最小可行服务骨架路由设计、静态资源托管与前后端数据桥接2.1 初始化 Flask 应用并配置关键参数先创建项目根目录steam_analytics/进入后执行mkdir -p app/{templates,static/{css,js,images}} pip install flask pandas numpy redis requests beautifulsoup4 plotly新建app/__init__.pyfrom flask import Flask import redis def create_app(): app Flask(__name__) # 关键配置禁用模板自动转义Plotly 图表需 raw HTML app.jinja_env.autoescape False # Redis 连接池本地开发用默认端口生产环境建议加密码和连接超时 app.redis_client redis.Redis( hostlocalhost, port6379, db0, decode_responsesTrue, # 返回字符串而非字节 socket_connect_timeout2, socket_timeout2 ) # 注册蓝图后续按功能拆分 from app.routes import main_bp app.register_blueprint(main_bp) return app提示decode_responsesTrue是血泪经验——若不设从 Redis 读出的所有值都是bxxx字节串后续 JSON 解析会直接报TypeError: the JSON object must be str, bytes or bytearray。这不是 Flask 的锅是 Redis-Py 默认行为。2.2 定义核心路由让前端能拿到结构化数据而非 HTML 页面在app/routes.py中定义两个关键接口from flask import Blueprint, jsonify, render_template from app import create_app import json main_bp Blueprint(main, __name__) main_bp.route(/) def index(): return render_template(index.html) # 前端主页面 main_bp.route(/api/games/top_sales) def top_sales(): 返回销量 Top 10 游戏的 JSON 数据模拟实际从 Redis 或 DB 读 # 实际项目中这里应调用 service 层方法如 get_top_sales_from_cache() mock_data [ {name: Cyberpunk 2077, sales_millions: 25.5, year: 2020}, {name: Elden Ring, sales_millions: 22.3, year: 2022}, {name: Stardew Valley, sales_millions: 15.8, year: 2016}, {name: Terraria, sales_millions: 14.2, year: 2011}, {name: Hades, sales_millions: 10.1, year: 2020} ] return jsonify({data: mock_data, total: len(mock_data)}) main_bp.route(/api/games/price_trend/game_id) def price_trend(game_id): 按 game_id 返回价格历史示例返回近 30 天均价波动 # 真实场景中 game_id 对应数据库主键或 Steam AppID trend_data [ {date: 2024-04-01, price_cny: 198.0}, {date: 2024-04-02, price_cny: 198.0}, {date: 2024-04-03, price_cny: 188.0}, # 打折日 {date: 2024-04-04, price_cny: 188.0}, {date: 2024-04-05, price_cny: 198.0} ] return jsonify({game_id: game_id, trend: trend_data})参数说明/api/games/top_sales是无参 GET 接口返回固定结构 JSON供前端 ECharts 初始化调用/api/games/price_trend/game_id使用 Flask URL 变量捕获game_id会被自动解析为字符串传入函数避免手动解析 query string所有接口统一用jsonify()包装确保 Content-Type 自动设为application/json省去手动设置 header。2.3 前端页面与图表初始化用 ECharts 加载动态数据在app/templates/index.html中!DOCTYPE html html head meta charsetutf-8 titleSteam 游戏数据分析平台/title script srchttps://cdn.jsdelivr.net/npm/echarts5.4.3/dist/echarts.min.js/script link relstylesheet href{{ url_for(static, filenamecss/style.css) }} /head body div idtop-sales-chart stylewidth: 800px; height: 400px;/div div idprice-trend-chart stylewidth: 800px; height: 400px;/div script // 初始化销量 Top10 柱状图 const salesChart echarts.init(document.getElementById(top-sales-chart)); fetch(/api/games/top_sales) .then(res res.json()) .then(data { const names data.data.map(d d.name); const sales data.data.map(d d.sales_millions); salesChart.setOption({ title: { text: Steam 销量 Top 5 游戏百万份 }, tooltip: {}, xAxis: { data: names }, yAxis: {}, series: [{ type: bar, data: sales, label: { show: true } }] }); }); // 初始化价格趋势折线图默认加载 Cyberpunk 2077 const trendChart echarts.init(document.getElementById(price-trend-chart)); fetch(/api/games/price_trend/238460) // Cyberpunk 2077 的 Steam AppID .then(res res.json()) .then(data { const dates data.trend.map(d d.date); const prices data.trend.map(d d.price_cny); trendChart.setOption({ title: { text: Cyberpunk 2077 近期价格趋势¥ }, tooltip: { trigger: axis }, xAxis: { type: category, data: dates }, yAxis: { type: value }, series: [{ type: line, data: prices, smooth: true, areaStyle: {} }] }); }); /script /body /html逻辑说明前端不写死数据所有图表数据通过fetch()从 Flask API 异步拉取url_for(static, ...)是 Flask 内置函数生成正确静态资源路径避免硬编码/static/...导致部署时 404ECharts 初始化必须等 DOM 元素渲染完成此处直接写在script标签内因页面简单复杂项目建议用DOMContentLoaded事件封装。3. Steam 数据采集绕过反爬、解析 HTML、提取关键字段的稳定方案3.1 为什么不用 Steam 官方 API——现实约束下的选型依据Steam 官方提供 Web API 需申请 Key但其公开接口极度受限GetAppList仅返回 AppID 和名称无价格、评价、发行日期GetGlobalAchievementPercentagesForApp等接口需绑定具体游戏且调用频次极低所有用户行为数据如愿望单数、评测数、在线人数均未开放更致命的是Steam 官网 HTML 结构稳定、字段语义清晰、无 JS 渲染依赖——这意味着requests BeautifulSoup足够可靠且比逆向 WebSocket 或模拟登录更轻量、更易维护。常见误用对比❌ 用 Selenium 模拟浏览器启动慢、内存占用高、Headless Chrome 易被识别❌ 直接抓取store.steampowered.com/app/xxx的原始 HTML部分字段如当前价格由 JS 动态注入requests拿不到✅ 正确做法抓取store.steampowered.com/app/xxx/?snr1_7_7_151_150_1带snr参数的 URL该链接返回完整 HTML价格、评测、发行日期等全部静态存在。3.2 构建稳健爬虫User-Agent 轮换、请求间隔、错误重试新建crawler/steam_crawler.pyimport requests from bs4 import BeautifulSoup import time import random from urllib.parse import urljoin class SteamCrawler: def __init__(self): self.session requests.Session() # 设置通用 headers模拟真实浏览器 self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en-US;q0.8,en;q0.7, Accept-Encoding: gzip, deflate, Connection: keep-alive, }) # 设置请求间隔避免触发风控 self.delay_range (1.5, 3.0) def get_game_page(self, app_id): 获取单个游戏详情页 HTML url fhttps://store.steampowered.com/app/{app_id}/?snr1_7_7_151_150_1 try: resp self.session.get(url, timeout10) resp.raise_for_status() if Sorry, we couldnt find that page. in resp.text: return None # 游戏下架或 ID 无效 return resp.text except requests.exceptions.RequestException as e: print(f[ERROR] Failed to fetch {url}: {e}) return None def parse_game_info(self, html, app_id): 解析 HTML 提取关键字段 soup BeautifulSoup(html, html.parser) # 游戏名称必填 name_tag soup.find(div, class_apphub_AppName) name name_tag.get_text(stripTrue) if name_tag else fUnknown Game ({app_id}) # 当前价格含折扣逻辑 price_tag soup.find(div, class_game_purchase_price) if not price_tag: price_tag soup.find(div, class_discount_original_price) # 折扣价原价 current_price price_tag.get_text(stripTrue) if price_tag else Free # 发行日期 release_tag soup.find(div, class_date) release_date release_tag.get_text(stripTrue) if release_tag else Unknown # 用户评测好评率 总数 review_summary soup.find(span, class_game_review_summary) review_text review_summary.get_text(stripTrue) if review_summary else No reviews # 愿望单数需定位到特定 div wishlist_tag soup.find(div, {data-tooltip-text: True}) wishlist_text if wishlist_tag and wishlisted in wishlist_tag.get(data-tooltip-text, ): wishlist_text wishlist_tag.get(data-tooltip-text).split()[0].replace(,, ) return { app_id: app_id, name: name, current_price: current_price, release_date: release_date, review_summary: review_text, wishlist_count: int(wishlist_text) if wishlist_text.isdigit() else 0 } def crawl_single_game(self, app_id): 完整爬取单个游戏流程 html self.get_game_page(app_id) if not html: return None time.sleep(random.uniform(*self.delay_range)) # 随机延迟 return self.parse_game_info(html, app_id) # 使用示例 if __name__ __main__: crawler SteamCrawler() result crawler.crawl_single_game(238460) # Cyberpunk 2077 print(result)参数说明snr1_7_7_151_150_1是 Steam 商店 URL 的标准参数确保返回完整页面session.headers.update()统一设置 UA 和 Accept 头避免单次请求缺失导致 403time.sleep(random.uniform(1.5, 3.0))是防封关键固定 2 秒易被模式识别浮动区间更自然soup.find(div, class_game_purchase_price)是精准定位比模糊搜索price文本更可靠。3.3 批量爬取与失败重试机制用队列管理 AppID 并记录日志新建crawler/batch_crawl.pyimport json import logging from concurrent.futures import ThreadPoolExecutor, as_completed from steam_crawler import SteamCrawler # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(crawler.log, encodingutf-8), logging.StreamHandler() ] ) def crawl_task(app_id, crawler): 单任务封装便于线程池调度 try: result crawler.crawl_single_game(app_id) if result: logging.info(f✅ Success: {result[name]} (AppID {app_id})) return result else: logging.warning(f⚠️ Skipped: AppID {app_id} (page not found or parsing failed)) return None except Exception as e: logging.error(f❌ Error on AppID {app_id}: {e}) return None def batch_crawl(app_ids, max_workers5): 批量爬取入口 crawler SteamCrawler() results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务 future_to_appid {executor.submit(crawl_task, aid, crawler): aid for aid in app_ids} # 收集结果 for future in as_completed(future_to_appid): result future.result() if result: results.append(result) # 保存结果 with open(data/raw_games.json, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) logging.info(f Crawled {len(results)}/{len(app_ids)} games. Saved to data/raw_games.json) if __name__ __main__: # 示例爬取 10 个热门游戏实际项目中从 SteamDB 或社区榜单获取 sample_appids [238460, 1245620, 271590, 1086940, 730, 570, 304930, 413150, 252490, 1172380] batch_crawl(sample_appids)逻辑说明ThreadPoolExecutor控制并发数max_workers5既保证效率又避免 IP 被限as_completed()按完成顺序收集结果无需等待全部结束日志同时输出到文件和控制台crawler.log可用于事后审计失败请求输出raw_games.json是原始采集层绝不在此阶段做清洗或计算——保持数据血缘可追溯。4. 数据清洗与多维统计用 Pandas 构建 Steam 分析模型的核心计算层4.1 从原始 JSON 到结构化 DataFrame处理缺失值与类型转换新建analysis/data_processor.pyimport pandas as pd import re from datetime import datetime def load_raw_data(filepathdata/raw_games.json): 加载原始爬取数据 with open(filepath, r, encodingutf-8) as f: data json.load(f) return pd.DataFrame(data) def clean_price(price_str): 清洗价格字符串提取数字处理 Free 和 ¥ 符号 if not isinstance(price_str, str): return 0.0 if Free in price_str: return 0.0 # 匹配 ¥ 后的数字支持千分位逗号 match re.search(r¥\s*([\d,]\.?\d*), price_str) if match: # 去除逗号并转浮点 return float(match.group(1).replace(,, )) return 0.0 def parse_release_date(date_str): 标准化发行日期统一为 YYYY-MM-DD if not isinstance(date_str, str): return pd.NaT try: # Steam 日期格式如 Nov 18, 2020 或 Feb 21, 2023 return pd.to_datetime(date_str, errorscoerce) except: return pd.NaT def build_analysis_df(): 构建用于分析的 DataFrame df load_raw_data() # 清洗价格列 df[price_cny] df[current_price].apply(clean_price) # 解析发行日期 df[release_date] df[release_date].apply(parse_release_date) df[release_year] df[release_date].dt.year # 提取好评率如 Overwhelmingly Positive → 95% # 实际项目中需更复杂规则此处简化 df[review_score] df[review_summary].map({ Overwhelmingly Positive: 0.95, Very Positive: 0.85, Mostly Positive: 0.75, Positive: 0.65, Mixed: 0.5, Negative: 0.3, Mostly Negative: 0.2, Very Negative: 0.1, Overwhelmingly Negative: 0.05 }).fillna(0.5) # 愿望单数转为数值 df[wishlist_count] pd.to_numeric(df[wishlist_count], errorscoerce).fillna(0).astype(int) return df # 使用示例 if __name__ __main__: df build_analysis_df() print(df[[name, price_cny, release_year, review_score, wishlist_count]].head())参数说明clean_price()用正则匹配¥后数字兼容¥198和¥ 198.00parse_release_date()用pd.to_datetime(..., errorscoerce)将无法解析的日期转为NaTNot a Time避免中断review_score映射表基于 Steam 评测等级官方定义非主观打分errorscoerce在pd.to_numeric()中至关重要——若某行wishlist_count是12,345带逗号不加此参数会直接报错。4.2 多维度统计实战按年份、价格区间、好评率分组聚合继续在data_processor.py中添加def generate_reports(df): 生成多维度统计报告 reports {} # 1. 按发行年份统计游戏数量 平均价格 yearly_stats df.groupby(release_year).agg( game_count(name, count), avg_price(price_cny, mean), median_price(price_cny, median), total_wishlist(wishlist_count, sum) ).round(2).reset_index() reports[yearly_stats] yearly_stats # 2. 按价格区间分桶0元、1-100元、101-200元、200元 bins [-1, 0, 100, 200, float(inf)] labels [Free, 1-100¥, 101-200¥, 200¥] df[price_range] pd.cut(df[price_cny], binsbins, labelslabels) price_range_stats df.groupby(price_range).agg( game_count(name, count), avg_review_score(review_score, mean), avg_wishlist_per_game(wishlist_count, mean) ).round(3).reset_index() reports[price_range_stats] price_range_stats # 3. 高愿望单游戏 Top 10排除免费游戏 top_wishlist df[df[price_cny] 0].nlargest(10, wishlist_count)[ [name, price_cny, release_year, review_score, wishlist_count] ].copy() top_wishlist[wishlist_count_formatted] top_wishlist[wishlist_count].apply( lambda x: f{x:,} # 添加千分位逗号 ) reports[top_wishlist] top_wishlist return reports # 保存报告为 CSV供 BI 工具或下游使用 def save_reports(reports): for name, df_report in reports.items(): df_report.to_csv(fdata/reports/{name}.csv, indexFalse, encodingutf-8-sig) if __name__ __main__: df build_analysis_df() reports generate_reports(df) save_reports(reports) print(✅ Reports generated:) for name, df_r in reports.items(): print(f- {name}: {len(df_r)} rows)逻辑说明pd.cut()实现价格分桶比if-elif更向量化、更易维护nlargest(10, wishlist_count)比sort_values().head(10)更高效尤其大数据集encodingutf-8-sig是 Windows Excel 打开中文 CSV 不乱码的关键——utf-8本身没问题但 Excel 默认用 ANSI加 BOM 才能识别所有统计结果保存为 CSV不直接存入数据库——CSV 是最通用、最易审计的中间产物。4.3 避坑Steam 数据清洗的 4 个高频翻车点现象 1price_cny列全是0.0但网页明明显示价格原因Steam 页面中价格可能以不同 class 名出现如game_purchase_price,discount_final_price,bundle_discount_amount而爬虫只查了第一个或价格被包裹在span内get_text()未递归提取。解决在parse_game_info()中扩展价格查找逻辑# 替换原 price_tag 查找 price_selectors [ div.game_purchase_price, div.discount_final_price, div.bundle_discount_amount ] for selector in price_selectors: price_tag soup.select_one(selector) if price_tag: current_price price_tag.get_text(stripTrue) break else: current_price Free现象 2release_date解析失败整列变NaT原因Steam 日期格式不统一除Nov 18, 2020外还有Coming soon、TBD、Q4 2024等非标准值。解决增强parse_release_date()def parse_release_date(date_str): if not isinstance(date_str, str): return pd.NaT # 过滤明显无效值 invalid_patterns [Coming soon, TBD, Q, Est.] if any(p in date_str for p in invalid_patterns): return pd.NaT try: return pd.to_datetime(date_str, errorscoerce) except: return pd.NaT现象 3wishlist_count为12,345字符串to_numeric()后变NaN原因pd.to_numeric()默认不处理千分位逗号需先清洗。解决在build_analysis_df()中预处理df[wishlist_count] df[wishlist_count].astype(str).str.replace(,, ).astype(float)现象 4review_score映射后大量NaN导致分组统计失真原因review_summary字段值如Very Positive (12,345)而映射表只匹配Very Positive。解决用正则提取关键词def extract_review_level(text): if not isinstance(text, str): return Mixed match re.search(r(Overwhelmingly|Very|Mostly|) (Positive|Negative), text) if match: return match.group(0) return Mixed df[review_level] df[review_summary].apply(extract_review_level) df[review_score] df[review_level].map({...}) # 映射表同上5. 数据持久化与缓存策略SQLite 存原始数据Redis 缓存聚合结果5.1 用 SQLite 建模轻量、零运维、ACID 可靠新建storage/db_manager.pyimport sqlite3 import pandas as pd from pathlib import Path DB_PATH Path(data/steam_analysis.db) def init_db(): 初始化 SQLite 数据库表结构 conn sqlite3.connect(DB_PATH) cursor conn.cursor() # 游戏主表 cursor.execute( CREATE TABLE IF NOT EXISTS games ( app_id TEXT PRIMARY KEY, name TEXT NOT NULL, current_price TEXT, release_date TEXT, review_summary TEXT, wishlist_count INTEGER DEFAULT 0, crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) # 统计快照表每日/每周聚合结果 cursor.execute( CREATE TABLE IF NOT EXISTS daily_stats ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, report_type TEXT NOT NULL, -- yearly, price_range, top_wishlist data_json TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() conn.close() def save_games_to_db(df): 将清洗后的 DataFrame 存入 SQLite conn sqlite3.connect(DB_PATH) # replace 模式每次全量覆盖适合小规模分析 df.to_sql(games, conn, if_existsreplace, indexFalse) conn.close() print(f✅ Saved {len(df)} games to SQLite) def save_daily_stats(report_name, data_df): 保存统计报告到 daily_stats 表 conn sqlite3.connect(DB_PATH) data_json data_df.to_json(orientrecords, force_asciiFalse) cursor conn.cursor() cursor.execute( INSERT INTO daily_stats (date, report_type, data_json) VALUES (?, ?, ?), (pd.Timestamp.now().strftime(%Y-%m-%d), report_name, data_json) ) conn.commit() conn.close() # 使用示例 if __name__ __main__: init_db() df pd.read_csv(data/reports/yearly_stats.csv) save_daily_stats(yearly_stats, df)参数说明if_existsreplace是关键避免append模式下重复插入导致主键冲突data_json TEXT存储 JSON 字符串而非建宽表——灵活应对报告结构变化crawled_at和created_at时间戳为数据溯源提供基础。5.2 Redis 缓存设计为高频 API 接口加速回到app/__init__.py增强 Redis 使用# 在 create_app() 中追加 def init_redis_cache(app): 初始化 Redis 缓存策略 app.before_first_request def setup_cache(): # 预热将常用统计结果存入 Redis try: # 从 SQLite 读 yearly_stats 并缓存 conn sqlite3.connect(data/steam_analysis.db) yearly_df pd.read_sql_query(SELECT * FROM daily_stats WHERE report_typeyearly_stats ORDER BY id DESC LIMIT 1, conn) conn.close() if not yearly_df.empty: latest_data json.loads(yearly_df.iloc[0][data_json]) app.redis_client.setex(yearly_stats, 3600, json.dumps(latest_data, ensure_asciiFalse)) print(✅ Cached yearly_stats to Redis (TTL1h)) except Exception as e: print(f[WARN] Failed to preload cache: {e}) # 在 create_app() 末尾调用 init_redis_cache(app) return app然后修改app/routes.py中的top_sales接口main_bp.route(/api/games/top_sales) def top_sales(): 从 Redis 缓存读取未命中则回退到计算 cached app.redis_client.get(yearly_stats) if cached: return jsonify({data: json.loads(cached), source: redis}) # 回退逻辑从 SQLite 读取并重新计算此处简化为 mock mock_data [{name: Cyberpunk 2077, sales_millions: 25.5, year: 2020}] return jsonify({data: mock_data, source: fallback})逻辑说明setex(key, 3600, value)设置键值对并指定 1 小时过期避免脏数据长期滞留before_first_request确保应用启动时预热缓存首次请求不经历冷启动接口返回source: redis字段便于前端或监控系统判断数据新鲜度。5.3 避坑SQLite 与 Redis 协同使用的 3 个边界问题现象 1SQLite 写入后Redis 缓存未更新前端看到旧数据原因数据写入 SQLite 和刷新 Redis 是两个独立操作缺乏事务一致性。解决在save_daily_stats()后主动刷新 Redisdef save_daily_stats(report_name, data_df): conn sqlite3.connect(DB_PATH) data_json data_df.to_json(orientrecords, force_asciiFalse) cursor conn.cursor() cursor.execute(INSERT INTO daily_stats ..., (...)) conn.commit() conn.close() # 同步刷新 Redis from app import create_app app create_app() app.redis_client.setex(f{report_name}_latest, 3600, data_json)现象 2Redis 内存爆满OOM command not allowed when used memory maxmemory原因未设置 Redis 最大内存或淘汰策略。解决启动 Redis 时指定配置redis-server --maxmemory 512mb --maxmemory-policy allkeys-lru或在redis.conf中设置maxmemory 512mb maxmemory-policy allkeys-lru现象 3SQLite 数据库被多个进程同时写入报database is locked原因Flask 默认多线程sqlite3.connect()在多线程下未加锁。解决使用check_same_threadFalse并启用 WAL 模式conn sqlite3.connect(DB_PATH, check_same_threadFalse) conn.execute(PRAGMA journal_modeWAL) # 启用 WAL提升并发写性能6. 交互式可视化进阶用 Plotly Dash 嵌入 Flask实现筛选联动与导出功能6.1 Dash 与 Flask 集成复用 Flask 的路由和 Redis 连接新建visualization/dash_app.pyimport dash from dash import dcc, html, Input, Output, callback import plotly.express as px import pandas as pd import json from app import create_app # 复用主 Flask 应用的 Redis 实例 # 创建 Dash 实例挂载到 Flask 上 server create_app() app dash.Dash( __name__, serverserver, url_base_pathname/dashboard/, # 注意斜杠结尾 external_stylesheets[/static/css/style.css] ) # 从 Redis 读取数据复用 p a hrefhttps://download.csdn.net/download/weixin_43880734/91933863 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p