AutoDataFlow

从读取、清洗、统计、分层、出图到打包输出的六步数据加工流水线。

喂给它一份 CSV / Excel / JSON / TXT，它帮你完成：去重清洗 → 分组统计 → 数值区间分层 / 时间分段 → 生成图表 → 打包成 Excel / CSV / JSON / JPG / ZIP 交付。

功能一览

| 模块 | 文件 | 功能 | | --- | --- | --- | | 01 读取 | 01_reader/data_reader.py | 读取 CSV / Excel / JSON / TXT，支持 Excel 多工作表 | | 02 清洗 | 02_cleaner/data_cleaner.py | 去重、缺失值、类型转换、按列独立清洗，返回清洗报告 | | 03 统计 | 03_stats/data_stats.py | 分组统计（count / mean / max / sum / 占比） | | 04 分层 | 04_binner/data_binner.py | 数值区间分类（年龄段、BMI 等）+ 时间分段（支持跨天时段） | | 05 出图 | 05_charter/data_charter.py | 柱状图 / 饼图 / 折线图，支持多色，中文自动适配 | | 06 打包 | 06_exporter/data_exporter.py | 输出 Excel / CSV / JSON / JPG 并打成 ZIP | | 工具 | utils.py | 时间列自动识别、人类可读文件大小等公共函数 |

安装

环境要求：Python 3.10+，依赖 pandas、matplotlib、openpyxl。

# 1. 拿到代码后进入项目目录
cd AutoDataFlow

# 2. （推荐）创建虚拟环境
python -m venv venv
# Windows
venv\Scripts\activate
# macOS / Linux
source venv/bin/activate

# 3. 安装依赖
pip install pandas matplotlib openpyxl

没有 requirements.txt 时，上面一行 pip 就是全部依赖。

快速开始

方式一：用自带总入口跑通示例

# Windows 建议先设编码，避免中文乱码
set PYTHONIOENCODING=utf-8

python run_pipeline.py                                  # 默认跑健康数据集
python run_pipeline.py --data-file test_data/patient_data.csv   # 换自己的数据

方式二：在自己的编排脚本里逐模块调用

新建 test_scripts/my_pipeline.py：

import os, sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
for sub in ("01_reader", "02_cleaner", "03_stats", "04_binner", "05_charter", "06_exporter"):
    sys.path.insert(0, os.path.join(ROOT, sub))
sys.path.insert(0, ROOT)          # 必须加：reader/exporter 要 import utils

from data_reader import read_data
from data_cleaner import clean_data
from data_stats import group_stats
from data_charter import generate_chart
from data_exporter import export_data

OUTPUT_DIR = os.path.join(ROOT, "output")
os.makedirs(OUTPUT_DIR, exist_ok=True)

# 1. 读取
df = read_data(os.path.join(ROOT, "test_data", "patient_data.csv"))

# 2. 清洗
df, report = clean_data(df, return_report=True)
print(report)

# 3. 分组统计：按年龄段统计住院天数
df["年龄区间"] = __import__("pandas").cut(
    df["Age"], bins=[0, 26, 36, 46, 56, 66, 200], right=False,
    labels=["<25", "26-35", "36-45", "46-55", "56-65", ">65"])
res = group_stats(df, "年龄区间", value_col="DaysInHospital",
                  metrics=["count", "mean", "max"])

# 4. 出图
chart = os.path.join(OUTPUT_DIR, "患者_各年龄段住院天数.jpg")
generate_chart(res, chart_type="bar", x_col="年龄区间",
               y_col="mean", output_path=chart,
               color=["#4E79A7", "#F28E2B", "#E15759", "#76B7B2", "#59A14F", "#B07AA1"])

# 5. 打包输出（ZIP 自动落在 output/ 下）
export_data({"各年龄段住院天数": res},
            output_dir=os.path.join(OUTPUT_DIR, "患者交付结果"),
            formats=["xlsx", "csv", "json"], image_paths=[chart], zip_output=True)

运行：

python test_scripts/my_pipeline.py

跑完后到 output/ 目录拿结果：

output/
├── 患者_各年龄段住院天数.jpg     # 图片直接放 output 根下
├── 患者交付结果/                 # Excel / CSV / JSON + 图片副本
└── 患者交付结果.zip              # ZIP 与交付文件夹同级

目录结构

AutoDataFlow/
├── run_pipeline.py            # 总入口
├── utils.py
├── 01_reader/ … 06_exporter/  # 六个功能模块
├── test_data/                 # 示例数据集（健康手环、患者数据、智能音箱、智能照明等）
├── test_scripts/              # 示例编排脚本（只放 .py，不放产物）
└── output/                    # 所有产物的统一交付目录

注意事项

所有产物只写 output/，不要往 test_scripts/ 里写任何数据或图表。
路径用 __file__ 推导，不要写死盘符，换机器才能直接跑。
Windows 中文乱码：跑脚本前 set PYTHONIOENCODING=utf-8。
多列独立上报的数据（IoT / 健康指标，大量 NaN）：清洗时用 dropna_cols=[要分析的列]、dropna_threshold=None、fillna_method=None、 outlier_method=None，否则指标列会被按缺失率误删。
group_stats 一次只能统计一个指标列，多列分别调用后 merge 合并。
整数分层用 right=False（左闭右开），分层后用 groupby(区间)[原列].agg(['min','max','count']) 自检边界。
算比例先造 0/1 标记列，group_stats(..., metrics=["count","sum","mean"]) 的 mean 就是比例。

已验证的测试数据集

| 数据集 | 说明 | | --- | --- | | 智能健康手环数据集.xlsx | 双工作表（步数统计记录 16800 行 / 功能查看记录 3836 行），时间分段 + 同步延迟统计 | | patient_data.csv | 1000 行患者数据，Age / BMI 分层 + 高风险比例统计 | | 智能健康监测系统数据集.xlsx | 多列独立上报典型场景，清洗零误删 | | 智能音箱数据集.xlsx | 默认清洗参数全流程 | | 智能照明系统数据集.xlsx | 双指标统计合并 + 多色柱状图 |

以上数据集均通过完整流水线实测，示例脚本见 test_scripts/。

License

MIT