结构化数据清洗与报表生成实战

# 危险：字段里含逗号时，cut 会把一条记录切成多段
cut -d, -f2 data.csv

# 危险：字段里含换行时，wc -l 统计的不是记录数
wc -l data.csv

# 安全：用专门的解析器，例如 python 的 csv 模块
python3 -c "
import csv, sys
for row in csv.reader(open('data.csv', newline='', encoding='utf-8')):
    print(row[1])
"

file data.csv                 # 编码与行尾提示
head -c 3 data.csv | xxd      # 检查是否 UTF-8 BOM (efbbbf)
head -1 data.csv              # 看表头
awk -F, 'NR<=3 {print NF}' data.csv   # 每行字段数是否一致

# 表头字段数
n=$(head -1 data.csv | awk -F, '{print NF}')
# 找出字段数异常的物理行
awk -F, -v n="$n" 'NF != n {print NR": "NF" 字段"}' data.csv

# 选择列（按名字，不怕列序变化）
csvcut -c name,email data.csv

# 过滤行（正则匹配）
csvgrep -c status -m 'active' data.csv

# 转成 JSON，便于下游程序消费
csvjson data.csv | jq '.[0]'

# 统计
csvstat data.csv

# 选列、过滤、排序、聚合
mlr --icsv --opprint \
  cut -f name,amount,region then \
  filter '$amount > 100' then \
  sort -f region then \
  stats1 -a sum,count -f amount -g region \
  data.csv

# 格式转换：CSV 进 JSON 出
mlr --icsv --ojson cat data.csv | jq '.[0]'

# 零依赖：只用 Python 标准库做选列
python3 - <<'PY'
import csv
with open('data.csv', newline='', encoding='utf-8') as f:
    r = csv.DictReader(f)
    for row in r:
        print(f"{row['name']}\t{row['amount']}")
PY

# 简单 CSV：按逗号切分求和
awk -F, 'NR>1 {sum += $3} END {print sum}' data.csv

# 输出带表头的报表
awk -F, 'BEGIN {OFS="\t"; print "区域","金额"} NR>1 {print $2,$3}' data.csv

# FPAT：字段要么是引号包裹的任意内容，要么是不含逗号的内容
gawk -v FPAT='([^,]*)|("[^"]*")' '
  NR>1 { gsub(/^"|"$/, "", $2); print $2 }
' data.csv

# 校验金额是数字，日期符合格式，不合规的行写入 reject.csv
awk -F, -v OFS=',' '
  NR==1 { print > "clean.csv"; next }
  $3 !~ /^[0-9]+(\.[0-9]+)?$/ { print > "reject.csv"; next }
  $4 !~ /^[0-9]{4}-[0-9]{2}-[0-9]{2}$/ { print > "reject.csv"; next }
  { print > "clean.csv" }
' data.csv

# 去掉 UTF-8 BOM（否则第一个字段名会带不可见前缀）
sed -i '1s/^\xEF\xBB\xBF//' data.csv

# 转成 UTF-8（源为 GBK）
iconv -f GBK -t UTF-8 data.csv > data.utf8.csv

# 行尾从 CRLF 归一为 LF
dos2unix data.csv

# 探测编码
file -I data.csv
# 转换时用 -c 丢弃无法转换的字符，或保留以便排查
iconv -f GBK -t UTF-8 -c data.csv > data.utf8.csv
# 统计丢弃了多少字符
iconv -f GBK -t UTF-8 data.csv 2>/dev/null | wc -c

# 用 Python 正确输出 CSV
python3 - <<'PY'
import csv, sys
w = csv.writer(sys.stdout, quoting=csv.QUOTE_MINIMAL)
w.writerow(['name', 'note'])
w.writerow(['张三', '含,逗号与"引号"的备注'])
PY

# 反向：把 TSV 转成合法 CSV
mlr --itsv --ocsv cat data.tsv

# 按区域分组求和（awk 关联数组）
awk -F, 'NR>1 {sum[$2] += $3} END {for (k in sum) printf "%s\t%.2f\n", k, sum[k]}' \
  data.csv | sort

# 两表连接（csvkit）
csvjoin -c id users.csv orders.csv | csvcut -c name,amount

# CSV 直接导入内存数据库查询
sqlite3 :memory: <<'SQL'
.mode csv
.import data.csv sales
SELECT region, SUM(amount) AS total
FROM sales GROUP BY region ORDER BY total DESC;
SQL

# 长表转宽表
mlr --icsv --opprint reshape -s metric,value data.csv

# awk 方式：按 (行,列) 累积到二维数组
awk -F, 'NR>1 {a[$1][$2]=$3} END {
  for (r in a) { printf "%s", r; for (c in a[r]) printf "\t%s", a[r][c]; print "" }
}' data.csv

# 生成固定格式的报表
{
  printf '区域\t订单数\t金额\n'
  awk -F, 'NR>1 {n[$2]++; s[$2]+=$3}
           END {for (k in s) printf "%s\t%d\t%.2f\n", k, n[k], s[k]}' data.csv \
    | sort
} > report.tsv

src_total=$(awk -F, 'NR>1 {s+=$3} END {print s}' data.csv)
rep_total=$(awk -F'\t' 'NR>1 {s+=$3} END {print s}' report.tsv)
awk -v a="$src_total" -v b="$rep_total" 'BEGIN {
  if (a - b > 0.01 || b - a > 0.01) { print "对账不平: " a " vs " b > "/dev/stderr"; exit 1 }
  print "对账通过: " b
}'

# JSON 输出
mlr --icsv --ojson cat report.tsv | jq .

# 对齐文本输出（column 或 mlr 的 pprint）
mlr --itsv --opprint cat report.tsv

# HTML 表格（csvkit）
csvjson report.tsv | jq -r '["区域","订单数","金额"], (.[]|[.[0],.[1],.[2]]) | "<tr>" + (map("<td>" + (.|tostring) + "</td>")|join("")) + "</tr>"'

#!/usr/bin/env bash
set -euo pipefail

SRC="${1:?用法: pipeline.sh <源 CSV>}"
WORK=$(mktemp -d); trap 'rm -rf "$WORK"' EXIT

# 第一步：编码归一，去掉 BOM 与 CRLF
iconv -f "$(file -I "$SRC" | sed 's/.*charset=//')" -t UTF-8 -c "$SRC" \
  | sed '1s/^\xEF\xBB\xBF//' | tr -d '\r' > "$WORK/norm.csv"

# 第二步：校验并分流
awk -F, -v OFS=',' '
  NR==1 { print > "'"$WORK"'/clean.csv"; next }
  NF != 4 { print > "'"$WORK"'/reject.csv"; next }
  $3 !~ /^[0-9]+(\.[0-9]+)?$/ { print > "'"$WORK"'/reject.csv"; next }
  { print > "'"$WORK"'/clean.csv" }
' "$WORK/norm.csv"

# 分组聚合
awk -F, 'NR>1 {n[$2]++; s[$2]+=$3}
         END {for (k in s) printf "%s\t%d\t%.2f\n", k, n[k], s[k]}' \
  "$WORK/clean.csv" | sort > "$WORK/report.tsv"

# 输出交付格式
{ printf '区域\t订单数\t金额\n'; cat "$WORK/report.tsv"; } > report.tsv
mlr --itsv --ojson cat report.tsv > report.json

# 统计拒收
printf '清洗完成：干净 %s 行，拒收 %s 行\n' \
  "$(( $(wc -l < "$WORK/clean.csv") - 1 ))" \
  "$(( $(wc -l < "$WORK/reject.csv" 2>/dev/null || echo 1) - 1 ))"

src=$(awk -F, 'NR>1 {s+=$3} END {print s+0}' "$WORK/norm.csv")
out=$(awk -F'\t' '{s+=$3} END {print s+0}' "$WORK/report.tsv")
awk -v a="$src" -v b="$out" 'BEGIN {
  if ((a-b > 0.01) || (b-a > 0.01)) { print "对账失败: " a " vs " b > "/dev/stderr"; exit 1 }
}'
printf '对账通过，报表已生成: report.tsv / report.json\n'