Dify 中级实验（08）：代码节点进阶——如何用标准库处理文件与数据？

variables:

- label: CSV数据（含表头）

  max_length: 10000          # ⚠️ 长文本粘贴必须显式放大，否则报 48 字符限制

  required: true

  type: paragraph

  variable: csv_text

- label: 分析类型

  max_length: 48

  options: [summary, trend, anomaly]

  required: true

  type: select

  variable: analysis_type

def main(csv_text: str, analysis_type: str) -> dict:

    import csv

    import io

    rows = []

    try:

        reader = csv.DictReader(io.StringIO(csv_text or ""))

        rows = list(reader)

    except Exception:

        rows = []

    if not rows:

        return {"record_count": 0, "result_text": "空数据，请粘贴 CSV 文本"}

    atype = (analysis_type or "summary").strip()

    cols = list(rows[0].keys())

    # 识别数值列

    numeric_cols = []

    for col in cols:

        vals = []

        for r in rows:

            v = (r.get(col) or "").strip()

            if v:

                try:

                    float(v)

                    vals.append(float(v))

                except Exception:

                    pass

        if vals:

            numeric_cols.append(col)

    lines = ["共 {} 行，列：{}".format(len(rows), ", ".join(cols))]

    if atype == "trend" and numeric_cols:

        col = numeric_cols[0]

        lines.append("{} 趋势（按行序）：".format(col))

        for i, r in enumerate(rows):

            lines.append("  第{}行 {} = {}".format(i + 1, r.get(cols[0], ""), r.get(col, "")))

    elif atype == "anomaly" and numeric_cols:

        col = numeric_cols[0]

        vals = [float(r[col]) for r in rows if (r.get(col) or "").strip()]

        if len(vals) > 3:

            mean = sum(vals) / len(vals)

            std = (sum((x - mean) ** 2 for x in vals) / len(vals)) ** 0.5

            lines.append("{} 异常检测（均值 {:.2f}，2σ={:.2f}）：".format(col, mean, 2 * std))

            for i, r in enumerate(rows):

                v = (r.get(col) or "").strip()

                if v and abs(float(v) - mean) > 2 * std:

                    lines.append("  第{}行 {}={} 超出正常范围".format(i + 2, col, v))

        else:

            lines.append("数据不足 4 行，无法做异常检测")

    else:

        for col in numeric_cols:

            vals = [float(r[col]) for r in rows if (r.get(col) or "").strip()]

            if vals:

                lines.append("{}: min={} max={} avg={:.2f} sum={:.2f}".format(

                    col, min(vals), max(vals), sum(vals) / len(vals), sum(vals)))

    return {"record_count": len(rows), "result_text": "\n".join(lines)}

def main(csv_text: str) -> dict:

    import csv

    import io

    from datetime import datetime

    formats = [

        ("%Y-%m-%d", "YYYY-MM-DD"),

        ("%Y/%m/%d", "YYYY/MM/DD"),

        ("%m/%d/%Y", "MM/DD/YYYY"),

        ("%Y-%m-%d %H:%M:%S", "YYYY-MM-DD HH:MM:SS"),

        ("%Y年%m月%d日", "中文格式"),

    ]

    # ... 解析 CSV 后找到第一个日期列，逐行识别并标准化

    return {"normalized_dates": normalized, "date_summary": "...每行 原始值 -> 标准值 (格式名)..."}