导航收藏不迷路—制造业数据与AI践行者老蒋的技术博客全系列文章汇总持续更新专栏WorkBuddy工作场景应用实践 ·#01 源码篇系列认知破冰 · 源码首发https://gitee.com/javy21/javy21_case01_csdn_data_analysis.git发布日期2026-07-29标签WorkBuddyCSDN数据分析Python源码离线H5IT老兵开源关联主文制造业IT老兵用WorkBuddy扒了自己CSDN后台3个反直觉的数据真相WorkBuddy实战CSDN后台数据分析系统完整代码一、开篇兄弟们上回说到我用WorkBuddy 30分钟搭了一套CSDN后台数据分析系统-1。文章发了之后有兄弟私下问代码什么时候开源。说实话看到这些留言我挺欣慰的——说明大家是真感兴趣不是随便点个赞就划走了。所以我决定把整套源码整理出来首次公开。代码仓库https://gitee.com/javy21/javy21_case01_csdn_data_analysis.git需要急用的朋友在评论区留言我单独发你。后续整理完毕后会统一开源。二、这套系统是干什么的一句话说清楚把你从CSDN后台下载的数据变成可交互的离线分析看板。功能模块说明数据导入把CSDN导出的Excel丢进指定目录一键导入增量合并新数据自动合并不会重复不会丢历史离线看板双击HTML直接打开ECharts图表全交互全程本地不联网、不上传、数据不出你的电脑技术栈极简Python 3.10数据处理 导入脚本Pandas openpyxl xlrdExcel读写ECharts 5.x SheetJS前端图表已本地化三、项目结构拿到源码后目录结构是这样的csdn_analytics/ ├── data/ │ ├── inbox/ # ① 把CSDN后台下载的Excel丢进这里 │ ├── master/ # ② 持久化主库 csdn_master.xlsx3个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(parentsTrue, exist_okTrue) log_file LOG_DIR / import.log handlers [ logging.StreamHandler(), logging.FileHandler(log_file, encodingutf-8), ] logging.basicConfig( levelgetattr(logging, LOG_LEVEL, logging.INFO), format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlershandlers, ) def ensure_dirs() - None: for directory in (INBOX_DIR, MASTER_DIR, ARCHIVE_DIR, DASHBOARD_DIR): directory.mkdir(parentsTrue, exist_okTrue) 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_name0, nrows3) 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], errorscoerce).fillna(0).astype(int) df df.dropna(subset[日期]).drop_duplicates(subsetKEY_DAILY, keeplast) 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], errorscoerce).fillna(0).astype(int) df df.dropna(subset[文章标题]).drop_duplicates(subsetKEY_ARTICLE, keeplast) 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_namesheet) 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_indexTrue) combined combined.drop_duplicates(subsetkeys, keeplast).reset_index(dropTrue) 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(orientrecords) if daily_sorted is not None else [], article: article.to_dict(orientrecords) if article is not None else [], } js // 由 import_data.py 自动生成请勿手动修改\n js window.CSDN_DATA json.dumps(payload, ensure_asciiFalse, indent2) ;\n DASH_DATA_JS.write_text(js, encodingutf-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_indexTrue) with pd.ExcelWriter(MASTER_XLSX, engineopenpyxl) as writer: (daily if daily is not None else pd.DataFrame(columnsDAILY_COLS)).to_excel( writer, sheet_nameSH_DAILY, indexFalse ) (article if article is not None else pd.DataFrame(columnsARTICLE_COLS)).to_excel( writer, sheet_nameSH_ARTICLE, indexFalse ) log_df.to_excel(writer, sheet_nameSH_LOG, indexFalse) 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_name0) except Exception as exc: logger.error(f读取失败: {path.name}, {exc}, exc_infoTrue) 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(dropTrue) if master_article is not None and len(master_article): master_article master_article.sort_values(创建日期, ascendingFalse).reset_index(dropTrue) 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_infoTrue) 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(parentsTrue, exist_okTrue) log_file LOG_DIR / dashboard.log handlers [ logging.StreamHandler(), logging.FileHandler(log_file, encodingutf-8), ] logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlershandlers, ) def start_dashboard(port: int DEFAULT_PORT) - None: os.chdir(DASHBOARD_DIR) handler http.server.SimpleHTTPRequestHandler url fhttp://127.0.0.1:{port}/index.html logger.info(f启动本地看板服务 (仅本机, 不联网): {url}) logger.info(按 CtrlC 停止服务) 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页面主要功能模块包括模块功能KPI概览总阅读量、总评论、总收藏、峰值粉丝、文章篇数、篇均阅读阅读量趋势每日阅读走势 7日移动平均支持指标切换互动指标趋势评论与收藏的每日变化粉丝增长分析累计粉丝与每日净增文章表现排名按阅读/展现/收藏/评论排序曝光→阅读转化散点图展示转化效率单篇雷达画像选中文章多维度对比数据明细表全字段搜索 表头排序技术实现ECharts 5.x 本地化dashboard/lib/echarts.min.jsSheetJS 本地化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 方式Bpython scripts/start_dashboard.py如果浏览器拦截file://5.3 日常更新新数据下载 → 丢进 data/inbox/ → python scripts/import_data.py → 刷新浏览器看板六、数据存储结构系统自动维护三张SheetSheet主键说明daily_trend日期每日阅读/评论/粉丝/收藏article_stats标题创建日期单篇文章展现/阅读/评论/收藏/关注import_log自增导入审计日志时间/文件/类型/新增/更新/状态/归档两表通过日期关联文章的创建日期可以与当天站点趋势对照分析。七、几个设计亮点亮点说明日期格式自适应同时支持2026年06月25日和2026-07-21两种格式-1增量合并幂等按主键去重新值覆盖旧值重复执行不产生重复数据原始文件归档每次导入带时间戳备份到data/archive/可追溯数据注入防跨域用script注入window.CSDN_DATA规避file://协议限制双数据源降级优先读data.js支持手动导入master.xlsx作为后备八、实在人总结怕你忘了我再啰嗦一遍这套源码是WorkBuddy生成的我只改了配置路径。不是我自己写的但每一行我都看过、跑过、验证过。源码今天开放。工程源码包后续整理完毕后会统一开源。环境配置只给提示不事无巨细。兄弟们都是搞技术的pip install 不用我教。真遇到问题评论区留言我看到就回。数据安全第一。所有代码纯本地运行不上传任何数据到云端。你的CSDN数据只留在你电脑上。九、评论区互动兄弟们源码拿到了记得跑一下试试。跑通了的评论区扣个1让我看看有多少人真的动手了。跑不通的评论区贴报错截图我帮你看看。跑完发现有改进想法的直接说下一版我加上。 系列导航系列制造业数据与AI落地实战AI赋能数据开发工程手册上一篇#01 制造业IT老兵用WorkBuddy扒了自己CSDN后台下一篇#02 《28张截图一键归档WorkBuddy帮我写了个批量重命名工具》本文所有代码均为WorkBuddy生成经博主实测验证。源码开放地址Gitee仓库地址csdn_data_analysis