源码首发-WorkBuddy实战：CSDN后台数据分析系统完整代码（离线H5看板·双击即用）

🏭导航收藏不迷路—>制造业数据与AI践行者老蒋的技术博客全系列文章汇总（持续更新）

一、开篇

兄弟们，上回说到我用WorkBuddy 30分钟搭了一套CSDN后台数据分析系统-1。文章发了之后，有兄弟私下问代码什么时候开源。

说实话，看到这些留言我挺欣慰的——说明大家是真感兴趣，不是随便点个赞就划走了。

所以我决定把整套源码整理出来，首次公开。

代码仓库：https://gitee.com/javy21/javy21_case01_csdn_data_analysis.git，需要急用的朋友在评论区留言，我单独发你。后续整理完毕后会统一开源。

二、这套系统是干什么的？

一句话说清楚：把你从CSDN后台下载的数据，变成可交互的离线分析看板。

技术栈极简：

• Python 3.10+（数据处理 + 导入脚本）

• Pandas + openpyxl + xlrd（Excel读写）

• ECharts 5.x + SheetJS（前端图表，已本地化）

三、项目结构

拿到源码后，目录结构是这样的：

csdn_analytics/ ├── data/ │ ├── inbox/ # ① 把CSDN后台下载的Excel丢进这里 │ ├── master/ # ② 持久化主库 csdn_master.xlsx（3个Sheet） │ └── archive/ # ③ 每次导入的原始文件带时间戳归档 ├── scripts/ │ ├── config.py # 统一配置入口（路径/字段/Sheet定义） │ ├── import_data.py # 数据导入 / 增量合并 / 归档 / 生成看板数据 │ └── start_dashboard.py # （可选）一键启动本地看板服务 ├── dashboard/ │ ├── index.html # H5离线看板（双击即可打开） │ ├── data.js # 由脚本自动生成的看板数据 │ └── lib/ # 本地化的ECharts + SheetJS（离线依赖） ├── README.md # 完整项目说明 └── debug.md # 开发排坑记录

四、完整源码

4.1scripts/config.py—— 统一配置入口

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================ # 制造数据与AI践行者老蒋 | CSDN: https://blog.csdn.net/javy21 # 文件名: config.py # 专栏归属: [WorkBuddy工作场景应用实践] # # 核心问题: 如何为本地数据分析系统提供统一、跨平台、可注入敏感信息的配置入口 # # 解决方案推演 (Why this way?): # 1. 方案选择: 集中式 config.py 而非散落常量, 任何路径/字段口径变更只需改一处。 # 2. 关键设计: 路径全部用 pathlib.Path(跨平台); 敏感项走 .env 注入; # 按"路径/文件/Sheet字段/运行"分模块分组, 注释清晰。 # 3. 边界思考: 本系统暂无私钥类敏感信息, 仍保留 .env 注入范式, 便于后续扩展。 # # 案例溯源: 智联工坊 - 技术内容运营分析看板 (配置层) # 适用周次: 第X周 ============================================================================ """ import os from pathlib import Path from dotenv import load_dotenv load_dotenv() # ---------- 路径配置 ---------- BASE_DIR = Path(__file__).parent.resolve().parent DATA_DIR = BASE_DIR / "data" INBOX_DIR = DATA_DIR / "inbox" MASTER_DIR = DATA_DIR / "master" ARCHIVE_DIR = DATA_DIR / "archive" DASHBOARD_DIR = BASE_DIR / "dashboard" LOG_DIR = BASE_DIR / "logs" # ---------- 文件配置 ---------- MASTER_XLSX = MASTER_DIR / "csdn_master.xlsx" DASH_DATA_JS = DASHBOARD_DIR / "data.js" # ---------- Sheet 与字段配置 ---------- SH_DAILY = "daily_trend" SH_ARTICLE = "article_stats" SH_LOG = "import_log" KEY_DAILY = ["日期"] KEY_ARTICLE = ["文章标题", "创建日期"] DAILY_COLS = ["日期", "阅读量", "评论数", "粉丝数", "收藏数"] ARTICLE_COLS = ["文章标题", "创建日期", "展现量", "阅读量", "评论数", "收藏数", "关注数"] # ---------- 运行配置 ---------- DEFAULT_PORT = int(os.getenv("CSDN_DASHBOARD_PORT", "8777")) LOG_LEVEL = os.getenv("CSDN_LOG_LEVEL", "INFO")

4.2scripts/import_data.py—— 数据导入引擎

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================ # 制造数据与AI践行者老蒋 | CSDN: https://blog.csdn.net/javy21 # 文件名: import_data.py # 专栏归属: [WorkBuddy工作场景应用实践] # # 核心问题: 如何把 CSDN 后台下载的多份 Excel 增量合并进本地主库, 并生成离线看板数据 # # 解决方案推演 (Why this way?): # 1. 方案选择: 以"主键去重 + 最新值覆盖"做增量合并而非整表替换, 保留历史可追溯性; # 原始文件再带时间戳归档, 形成双保险。 # 2. 关键设计: 主库用多 Sheet(daily_trend / article_stats / import_log)统一管理; # 日期统一为 YYYY-MM-DD, 指标转整数且缺失补 0, 保证后续分析口径一致。 # 3. 边界思考: 文件类型按列名自动识别; 读取异常跳过并记日志而非中断; # 空 inbox 时仍基于现有主库重算 data.js, 避免看板空白。 # # 案例溯源: 智联工坊 - 技术内容运营分析看板 (基于 CSDN 专栏效果数据, 已脱敏) # 适用周次: 第X周 ============================================================================ """ import datetime as dt import json import logging import re import shutil import sys from pathlib import Path from typing import Any, Dict, List, Optional, Tuple import pandas as pd from config import ( ARCHIVE_DIR, ARTICLE_COLS, DAILY_COLS, DASH_DATA_JS, DASHBOARD_DIR, INBOX_DIR, KEY_ARTICLE, KEY_DAILY, LOG_DIR, LOG_LEVEL, MASTER_DIR, MASTER_XLSX, SH_ARTICLE, SH_DAILY, SH_LOG, ) logger = logging.getLogger(__name__) def configure_logging() -> None: LOG_DIR.mkdir(parents=True, exist_ok=True) log_file = LOG_DIR / "import.log" handlers = [ logging.StreamHandler(), logging.FileHandler(log_file, encoding="utf-8"), ] logging.basicConfig( level=getattr(logging, LOG_LEVEL, logging.INFO), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=handlers, ) def ensure_dirs() -> None: for directory in (INBOX_DIR, MASTER_DIR, ARCHIVE_DIR, DASHBOARD_DIR): directory.mkdir(parents=True, exist_ok=True) def norm_date(value: Any) -> Optional[str]: """将混合格式日期归一为 YYYY-MM-DD；无法解析返回 None。""" if value is None or (isinstance(value, float) and pd.isna(value)): return None text = str(value).strip() if not text: return None match = re.match(r"(\d{4})\D+(\d{1,2})\D+(\d{1,2})\D*", text) if match: year, month, day = match.groups() return f"{int(year):04d}-{int(month):02d}-{int(day):02d}" try: return pd.to_datetime(text).strftime("%Y-%m-%d") except Exception: return None def classify(path: Path) -> Optional[str]: """根据表头列名自动判断文件属于 daily 还是 article。""" try: sample = pd.read_excel(path, sheet_name=0, nrows=3) except Exception as exc: logger.warning(f"无法读取文件 {path.name}, 已跳过: {exc}") return None cols = {str(c).strip() for c in sample.columns} if "文章标题" in cols: return "article" if "日期" in cols: return "daily" return None def clean_daily(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df.columns = [str(c).strip() for c in df.columns] for col in DAILY_COLS: if col not in df.columns: df[col] = 0 df = df[DAILY_COLS] df["日期"] = df["日期"].map(norm_date) for col in ["阅读量", "评论数", "粉丝数", "收藏数"]: df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0).astype(int) df = df.dropna(subset=["日期"]).drop_duplicates(subset=KEY_DAILY, keep="last") return df def clean_article(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df.columns = [str(c).strip() for c in df.columns] for col in ARTICLE_COLS: if col not in df.columns: df[col] = 0 df = df[ARTICLE_COLS] df["创建日期"] = df["创建日期"].map(norm_date) df["文章标题"] = df["文章标题"].astype(str).str.strip() for col in ["展现量", "阅读量", "评论数", "收藏数", "关注数"]: df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0).astype(int) df = df.dropna(subset=["文章标题"]).drop_duplicates(subset=KEY_ARTICLE, keep="last") return df def read_master_sheet(sheet: str) -> Optional[pd.DataFrame]: if not MASTER_XLSX.exists(): return None try: return pd.read_excel(MASTER_XLSX, sheet_name=sheet) except Exception as exc: logger.error(f"读取主库 Sheet[{sheet}] 失败: {exc}") return None def merge_incremental( old: Optional[pd.DataFrame], new: pd.DataFrame, keys: List[str], ) -> Tuple[pd.DataFrame, int, int]: if old is None or len(old) == 0: return new.copy(), len(new), 0 combined = pd.concat([old, new], ignore_index=True) combined = combined.drop_duplicates(subset=keys, keep="last").reset_index(drop=True) old_keys = {tuple(row) for row in old[keys].astype(str).values.tolist()} new_keys = {tuple(row) for row in new[keys].astype(str).values.tolist()} added = len(new_keys - old_keys) updated = len(new_keys & old_keys) return combined, added, updated def archive_file(path: Path) -> Path: timestamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S") dst = ARCHIVE_DIR / f"{timestamp}__{path.name}" shutil.copy2(path, dst) return dst def export_data_js(daily: Optional[pd.DataFrame], article: Optional[pd.DataFrame]) -> None: daily_sorted = daily.sort_values("日期") if daily is not None and len(daily) else daily payload = { "generated_at": dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "daily": daily_sorted.to_dict(orient="records") if daily_sorted is not None else [], "article": article.to_dict(orient="records") if article is not None else [], } js = "// 由 import_data.py 自动生成，请勿手动修改\n" js += "window.CSDN_DATA = " + json.dumps(payload, ensure_ascii=False, indent=2) + ";\n" DASH_DATA_JS.write_text(js, encoding="utf-8") logger.info(f"看板数据已生成: {DASH_DATA_JS}") def write_master( daily: Optional[pd.DataFrame], article: Optional[pd.DataFrame], log_rows: List[Dict[str, Any]], ) -> None: old_log = read_master_sheet(SH_LOG) log_df = pd.DataFrame(log_rows) if old_log is not None and len(old_log): log_df = pd.concat([old_log, log_df], ignore_index=True) with pd.ExcelWriter(MASTER_XLSX, engine="openpyxl") as writer: (daily if daily is not None else pd.DataFrame(columns=DAILY_COLS)).to_excel( writer, sheet_name=SH_DAILY, index=False ) (article if article is not None else pd.DataFrame(columns=ARTICLE_COLS)).to_excel( writer, sheet_name=SH_ARTICLE, index=False ) log_df.to_excel(writer, sheet_name=SH_LOG, index=False) def run_import(sources: List[Path]) -> Dict[str, Any]: if not sources: logger.info("inbox 中无待导入文件, 基于现有主库重算看板数据") daily = read_master_sheet(SH_DAILY) article = read_master_sheet(SH_ARTICLE) if daily is None and article is None: return { "ok": False, "error": "主库为空且 inbox 无文件", "suggestion": "请将 CSDN 下载的 Excel 放入 data/inbox/ 后重试", } export_data_js(daily, article) return { "ok": True, "daily_rows": 0 if daily is None else len(daily), "article_rows": 0 if article is None else len(article), "log": [], "regenerated": True, } logger.info(f"开始导入, 待处理文件数: {len(sources)}") master_daily = read_master_sheet(SH_DAILY) master_article = read_master_sheet(SH_ARTICLE) log_rows = [] now = dt.datetime.now().strftime("%Y-%m-%d %H:%M:%S") for path in sources: kind = classify(path) if kind is None: logger.warning(f"跳过无法识别的文件: {path.name}") log_rows.append({ "时间": now, "文件": path.name, "类型": "未知", "新增": 0, "更新": 0, "状态": "跳过(无法识别)", }) continue try: raw = pd.read_excel(path, sheet_name=0) except Exception as exc: logger.error(f"读取失败: {path.name}, {exc}", exc_info=True) log_rows.append({ "时间": now, "文件": path.name, "类型": kind, "新增": 0, "更新": 0, "状态": "失败(读取异常)", }) continue if kind == "daily": cleaned = clean_daily(raw) master_daily, added, updated = merge_incremental(master_daily, cleaned, KEY_DAILY) else: cleaned = clean_article(raw) master_article, added, updated = merge_incremental(master_article, cleaned, KEY_ARTICLE) archived = archive_file(path) logger.info(f"导入 {path.name} -> {kind}, 新增 {added} / 更新 {updated}; 归档 {archived.name}") log_rows.append({ "时间": now, "文件": path.name, "类型": kind, "新增": added, "更新": updated, "状态": "成功", "归档": archived.name, }) if master_daily is not None and len(master_daily): master_daily = master_daily.sort_values("日期").reset_index(drop=True) if master_article is not None and len(master_article): master_article = master_article.sort_values("创建日期", ascending=False).reset_index(drop=True) try: write_master(master_daily, master_article, log_rows) export_data_js(master_daily, master_article) except Exception as exc: logger.error(f"写入主库或生成看板失败: {exc}", exc_info=True) return { "ok": False, "error": f"写入失败: {exc}", "suggestion": "请检查磁盘空间或文件是否被其他程序占用", } return { "ok": True, "daily_rows": 0 if master_daily is None else len(master_daily), "article_rows": 0 if master_article is None else len(master_article), "log": log_rows, } def main(argv: List[str]) -> None: configure_logging() ensure_dirs() sources = [] for arg in argv[1:]: cand = Path(arg) if cand.is_file(): sources.append(cand) if not sources: for f in INBOX_DIR.iterdir(): if f.suffix.lower() in (".xls", ".xlsx") and not f.name.startswith("~$"): sources.append(f) result = run_import(sources) if not result.get("ok"): logger.error(f"导入未完成: {result.get('error')}") logger.error(f"建议: {result.get('suggestion')}") sys.exit(1) print("-" * 60) print(f"[完成] master 已更新: {MASTER_XLSX}") print(f" daily_trend : {result.get('daily_rows')} 行") print(f" article_stats: {result.get('article_rows')} 行") print(f"[完成] 看板数据已生成: {DASH_DATA_JS}") print(" 打开 dashboard/index.html 即可离线查看分析结果。") if __name__ == "__main__": main(sys.argv)

4.3scripts/start_dashboard.py—— 本地服务启动器

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ============================================================================ # 制造数据与AI践行者老蒋 | CSDN: https://blog.csdn.net/javy21 # 文件名: start_dashboard.py # 专栏归属: [WorkBuddy工作场景应用实践] # # 核心问题: 当浏览器拦截 file:// 下的本地脚本时, 如何在不联网前提下用纯本机静态服务打开看板 # # 解决方案推演 (Why this way?): # 1. 方案选择: 用标准库 http.server 起一个绑定 127.0.0.1 的静态服务, # 零第三方依赖、零外联, 完全离线。 # 2. 关键设计: 端口可由命令行参数覆盖(默认 8777, 亦受 config.DEFAULT_PORT 控制); # 启动后自动打开浏览器, 降低使用门槛。 # 3. 边界情况: 浏览器打开失败时静默忽略, 仅提示手动访问 URL, 不中断服务。 # # 案例溯源: 智联工坊 - 技术内容运营分析看板 (离线工程化启动) # 适用周次: 第X周 ============================================================================ """ import logging import os import sys import webbrowser from pathlib import Path import http.server import socketserver from config import DASHBOARD_DIR, DEFAULT_PORT, LOG_DIR logger = logging.getLogger(__name__) def configure_logging() -> None: LOG_DIR.mkdir(parents=True, exist_ok=True) log_file = LOG_DIR / "dashboard.log" handlers = [ logging.StreamHandler(), logging.FileHandler(log_file, encoding="utf-8"), ] logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=handlers, ) def start_dashboard(port: int = DEFAULT_PORT) -> None: os.chdir(DASHBOARD_DIR) handler = http.server.SimpleHTTPRequestHandler url = f"http://127.0.0.1:{port}/index.html" logger.info(f"启动本地看板服务 (仅本机, 不联网): {url}") logger.info("按 Ctrl+C 停止服务") try: webbrowser.open(url) except Exception as exc: logger.warning(f"自动打开浏览器失败, 请手动访问 {url}: {exc}") with socketserver.TCPServer(("127.0.0.1", port), handler) as httpd: httpd.serve_forever() def main(argv: list) -> None: configure_logging() port = DEFAULT_PORT if len(argv) > 1: try: port = int(argv[1]) except ValueError: logger.error(f"端口参数非法: {argv[1]}") logger.error("建议: 传入 1024-65535 之间的整数, 如 python start_dashboard.py 9000") sys.exit(1) try: start_dashboard(port) except OSError as exc: logger.error(f"无法在端口 {port} 启动服务: {exc}") logger.error("建议: 该端口可能已被占用, 换一个端口, 如 python start_dashboard.py 9000") sys.exit(1) except KeyboardInterrupt: logger.info("已停止看板服务") if __name__ == "__main__": main(sys.argv)

4.4dashboard/index.html—— 离线H5看板

⚠️代码较长，粘贴不上，此处展示核心框架，完整版随源码包，请专注动态，后续提供。

index.html是一个纯前端HTML页面，主要功能模块包括：

技术实现：

• ECharts 5.x 本地化（dashboard/lib/echarts.min.js）

• SheetJS 本地化（dashboard/lib/xlsx.full.min.js）

• 数据通过window.CSDN_DATA全局注入

• 支持手动导入master.xlsx作为后备方案

五、快速上手

5.1 环境准备

# Python 3.10+ 环境 pip install pandas openpyxl xlrd python-dotenv

环境配置只给提示，不事无巨细。如果安装过程中遇到问题，评论区留言，我会及时关注。

5.2 使用步骤

1. 从CSDN后台下载三个数据文件，放入 data/inbox/ - 趋势图 / 数据列表 → "全部文章分析.xlsx" - 单篇文章分析 → "单篇文章分析.xls" 2. 一键导入 python scripts/import_data.py 3. 打开看板 方式A：直接双击 dashboard/index.html 方式B：python scripts/start_dashboard.py（如果浏览器拦截file://）

5.3 日常更新

新数据下载 → 丢进 data/inbox/ → python scripts/import_data.py → 刷新浏览器看板

六、数据存储结构

系统自动维护三张Sheet：

两表通过日期关联，文章的创建日期可以与当天站点趋势对照分析。

七、几个设计亮点

八、实在人总结

怕你忘了，我再啰嗦一遍：

1. 这套源码是WorkBuddy生成的，我只改了配置路径。不是我自己写的，但每一行我都看过、跑过、验证过。

2. 源码今天开放。工程源码包，后续整理完毕后会统一开源。

3. 环境配置只给提示，不事无巨细。兄弟们都是搞技术的，pip install 不用我教。真遇到问题，评论区留言，我看到就回。

4. 数据安全第一。所有代码纯本地运行，不上传任何数据到云端。你的CSDN数据只留在你电脑上。

九、评论区互动

兄弟们，源码拿到了记得跑一下试试。

跑通了的，评论区扣个1，让我看看有多少人真的动手了。

跑不通的，评论区贴报错截图，我帮你看看。

跑完发现有改进想法的，直接说，下一版我加上。

📎 系列导航

📌本文所有代码均为WorkBuddy生成，经博主实测验证。源码开放地址：Gitee仓库地址csdn_data_analysis