工具概况

此文件比对工具主要是在数据采集作业从一个工具迁移到另一个工具后, 对两边数据进行差异比对, 避免迁移后数据出现重大差异, 导致下游出现错误的情况

工具为抽样比对, 抽样条数可以程序中调整指定, 主要比对内容如下:

数据文件大小是否一致

数据文件行数是否一致

数据文件编码是否一致

数据文件字段数是否一致

小数类型字段精度是否一致

日期类型字段格式是否一致

如果文件大小不一致, 抽样比对内容差异

其他说明:

如果数据文件大小一致, 行数一致, 基本可以认证迁移前后完全一致, 无任何差异

如果行数一致, 字段数一致, 大小不一致, 一般是以下几种情况:

trim去空格差异

小数精度差异

日期格式差异

程序说明

data_compare_tool.py

数据文件比对核心程序:

自动比对两文件的大小/行数/编码/字段数/小数精度/日期格式

小数: 自动判断字段是否是小数类型字段

日期: 自动判断字段是否是日期类型字段

内容: 大小不一致时抽样比对内容差异

比对结果输出为json类型

run_batch_compare.sh

嵌套 data_compare_tool.py主程序, 实现批量文件的比对, 并将比对结果写入到指定的Excel文件, 同时对存在差异的结果给对应单元格填充颜色

可根据需要自行自定义脚本

版本变更:
V4 - 指定比对文件清单(格式: old_file.txt|new_file.txt|old_file_delim|new_file_delim); 参数若只传入一个分隔符, 则新老文件均按此分隔符判断; 清单中分隔符优先级最高, 会覆盖参数传入的分隔符参数
v3 - 调整了最后输出的Excel的样式
v2 - 增加了特殊字符的处理, 若数据中存在特殊字符, 原样输出
v1 - 无对特殊字符的处理, 若数据中存在特殊字符, 在进程内容差异比对时会报错

使用方式

比对单个文件

分隔符以十六进制编码传入

1
2
3

# 示例
# python3 data_compare_tool.py 原文件 新文件 分隔符(十六进制表示, 如:7F5E)
python3 data_compare_tool.py /home/mds/file/input/add/HR/20260629/HR_PS_MDS_ZHICHEG_VW.txt /home/mds/file/input/add/HR/20260629/ZHAOTY_PS_MDS_ZHICHEG_VW.txt 7F5E

比对多个文件

V3版本

当前Shell脚本限定了原文件和新文件在同一目录下, 且新文件名为 SHARKDATA_原文件名格式

可根据实际情况及需求自行调整脚本

1
2
3

# 示例
# sh run_batch_compare.sh 文件目录 分隔符 结果Excel文件
sh run_batch_compare.sh /home/file/20260629 7F5E ./res.xlsx

V4版本

传入文件清单, 按照清单内容进行比对:

格式: old_file.txt|new_file.txt|old_file_delim|new_file_delim

old_file.txt: 老文件名（绝对路径）

new_file.txt: 新文件名（绝对路径）

old_file_delim: 老文件分隔符(若不传入以参数传入为准, 若传入则覆盖参数)

new_file_delim: 新文件分隔符(若不传入以参数传入为准, 若传入则覆盖参数)

1 2	`# 示例 sh run_batch_compare.sh /tmp/file.list 7C 7F5E ./res.xlsx`

代码

Shell入口, 调用Python

Python

import os
import sys
import re
import json
import logging
import chardet
from datetime import datetime

# 配置日志，输出到标准错误流，避免污染传给 Shell 的标准输出 JSON
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', stream=sys.stderr)
logger = logging.getLogger(__name__)

SAMPLE_SIZE = 10000

# 比对结果中 Excel 汇总依赖的字段名（务必保持契约稳定）
RESULT_KEYS = [
    "file_name", "old_file_path", "new_file_path", "status", "encoding_diff", "count_diff", "size_diff",
    "col_count_diff", "decimal_diff", "date_diff", "content_lookup_diff",
]


def detect_encoding(file_path):
    """检测文件编码"""
    try:
        with open(file_path, 'rb') as f:
            raw_data = f.read(60000)
        if not raw_data:
            return 'utf-8'
        result = chardet.detect(raw_data)
        encoding = (result or {}).get('encoding') or 'utf-8'
        if encoding in ['GB2312', 'GBK']:
            encoding = 'GB18030'
        return encoding
    except Exception as e:
        logger.warning(f"Encoding detection failed for {file_path}: {e}")
        return 'utf-8'


def read_lines_streaming(file_path, encoding, limit=None):
    """生成器：逐行读取文件，节省内存（仅去除行尾换行，保留行内内容）"""
    try:
        with open(file_path, 'r', encoding=encoding, errors='ignore') as f:
            count = 0
            for line in f:
                if limit and count >= limit:
                    break
                yield line.rstrip('\n').rstrip('\r')
                count += 1
    except Exception as e:
        logger.error(f"Error reading file {file_path}: {e}")


def parse_separator(hex_sep):
    """将十六进制字符串转换为实际的分隔符字符串（跨系统/跨编码兼容）

    支持单字节或多字节分隔符，例如：
      "7C"   -> "|"
      "09"   -> "\\t"
      "7F5E" -> "\\x7f^"
      "EFFBC8C" -> 中文全角逗号"，"
    """
    if not hex_sep:
        raise ValueError("分隔符十六进制为空")
    h = str(hex_sep).strip().lower()
    if h.startswith('0x'):
        h = h[2:]
    try:
        raw = bytes.fromhex(h)
    except ValueError:
        raise ValueError(f"非法的分隔符十六进制: {hex_sep!r}（应为偶数字符，如 7C / 7F5E）")
    if not raw:
        raise ValueError("分隔符十六进制解码为空字节")
    # 优先按 utf-8 解码（兼容中文等多字节分隔符），失败则回退 latin-1（字节保真）
    for enc in ('utf-8', 'latin-1'):
        try:
            return raw.decode(enc)
        except UnicodeDecodeError:
            continue
    return raw.decode('latin-1')


def is_valid_date_str(val):
    """严格判断一个字符串是否为合法的日期/时间格式"""
    if not val or not isinstance(val, str):
        return False
    val = val.strip()
    if len(val) < 8:
        return False
    if val.isdigit():
        if len(val) == 8:
            try:
                datetime.strptime(val, "%Y%m%d")
                return True
            except ValueError:
                return False
        return False
    date_patterns = [
        (r"\d{4}[-/.]\d{1,2}[-/.]\d{1,2}$", ["%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"]),
        (r"\d{4}[-/.]\d{1,2}[-/.]\d{1,2}\s+\d{1,2}:\d{1,2}:\d{1,2}$", ["%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S"]),
    ]
    for pattern, fmts in date_patterns:
        if re.fullmatch(pattern, val):
            for fmt in fmts:
                try:
                    dt_obj = datetime.strptime(val, fmt)
                    if 1900 <= dt_obj.year <= 2100:
                        return True
                except ValueError:
                    continue
    return False


def get_standard_format(val):
    """获取日期的标准格式字符串"""
    val = val.strip()
    formats_to_try = [
        "%Y-%m-%d", "%Y/%m/%d",
        "%Y-%m-%d %H:%M:%S", "%Y/%m/%d %H:%M:%S",
        "%Y%m%d", "%Y-%m-%d %H:%M:%S.%f"
    ]
    for fmt in formats_to_try:
        try:
            datetime.strptime(val, fmt)
            return fmt
        except ValueError:
            continue
    return None


def get_number_format_template(val):
    """ 获取数字的格式模板。
    返回: (is_decimal, int_len, dec_len)
    例如: "01.20" -> (True, 2, 2), "100" -> (False, 3, 0)
    """
    val = val.strip()
    if not val:
        return None
    is_negative = val.startswith('-')
    clean_val = val.lstrip('-')
    if '.' in clean_val:
        # 包含小数点，说明是小数
        parts = clean_val.split('.')
        int_part = parts[0]
        dec_part = parts[1]
        return (True, len(int_part), len(dec_part))
    else:
        # 纯整数
        return (False, len(clean_val), 0)


def analyze_columns(lines, sep):
    """分析列的类型，返回日期列和数值列的索引集合（按文件各自的分隔符）"""
    col_types = {}
    for line in lines:
        if not line.strip():
            continue
        parts = line.split(sep)
        for i, val in enumerate(parts):
            val = val.strip()
            if not val:
                continue
            stats = col_types.setdefault(i, {'date': 0, 'numeric': 0, 'total': 0})
            stats['total'] += 1
            if is_valid_date_str(val):
                stats['date'] += 1
            # 仅当值包含小数点才视为小数部分（纯整数列不进入小数精度比对，避免误判）
            try:
                float(val)
                if '.' in val:
                    stats['numeric'] += 1
            except ValueError:
                pass
    date_cols = set()
    numeric_cols = set()
    for idx, stats in col_types.items():
        if stats['total'] > 5 and (stats['date'] / stats['total']) > 0.8:
            date_cols.add(idx)
        elif stats['total'] > 5 and (stats['numeric'] / stats['total']) > 0.8:
            numeric_cols.add(idx)
    return date_cols, numeric_cols


def count_lines_fast(path):
    """快速统计文件行数（按字节统计 \\n，兼容 Windows CRLF）"""
    count = 0
    with open(path, 'rb') as f:
        for chunk in iter(lambda: f.read(8192 * 1024), b''):
            count += chunk.count(b'\n')
    return count


def first_non_empty(lines):
    for ln in lines:
        if ln.strip():
            return ln
    return None


def _error_result(old_file, new_file, status, err):
    """统一的错误结果结构，保证 Excel 汇总能读取到所有字段"""
    return {
        "file_name": os.path.basename(old_file),
        "old_file_path": old_file,
        "new_file_path": new_file,
        "status": status,
        "error": str(err),
        "encoding_diff": "ERROR",
        "count_diff": "ERROR",
        "size_diff": "ERROR",
        "col_count_diff": "ERROR",
        "decimal_diff": "ERROR",
        "date_diff": "ERROR",
        "content_lookup_diff": "ERROR",
    }


def _empty_result(old_file, new_file):
    """样本为空时的结果"""
    return {
        "file_name": os.path.basename(old_file),
        "old_file_path": old_file,
        "new_file_path": new_file,
        "status": "OK",
        "encoding_diff": "N/A",
        "count_diff": "N/A",
        "size_diff": "N/A",
        "col_count_diff": "N/A (Empty)",
        "decimal_diff": "N/A",
        "date_diff": "N/A",
        "content_lookup_diff": "N/A (Empty Sample)",
    }


def compare_files(old_file, new_file, old_hex_sep, new_hex_sep):
    # 1. 分别解析新老文件分隔符（任一非法都返回结构化错误，而非直接退出进程）
    try:
        old_sep = parse_separator(old_hex_sep)
        new_sep = parse_separator(new_hex_sep)
    except ValueError as e:
        return _error_result(old_file, new_file, "ERROR", e)

    old_size = os.path.getsize(old_file)
    new_size = os.path.getsize(new_file)
    old_enc = detect_encoding(old_file)
    new_enc = detect_encoding(new_file)

    old_lines_count = count_lines_fast(old_file)
    new_lines_count = count_lines_fast(new_file)

    results = {
        "file_name": os.path.basename(old_file),
        "old_file_path": old_file,
        "new_file_path": new_file,
        "status": "OK",
        "encoding_diff": "OK" if old_enc == new_enc else f"{old_enc} | {new_enc}",
        "count_diff": "OK" if old_lines_count == new_lines_count else f"{old_lines_count} | {new_lines_count}",
        "size_diff": "OK" if old_size == new_size else f"{old_size} | {new_size}",
        "col_count_diff": "OK",
        "decimal_diff": "OK",
        "date_diff": "OK",
        "content_lookup_diff": "OK",
    }

    logger.info(f"Sampling data from {old_file} ...")
    old_sample = list(read_lines_streaming(old_file, old_enc, SAMPLE_SIZE + 1))
    new_sample = list(read_lines_streaming(new_file, new_enc, SAMPLE_SIZE + 1))

    if not old_sample or not new_sample:
        return _empty_result(old_file, new_file)

    # 2. 列数比对（各自分隔符）
    old_first = first_non_empty(old_sample)
    new_first = first_non_empty(new_sample)
    old_col_count = len(old_first.split(old_sep)) if old_first else 0
    new_col_count = len(new_first.split(new_sep)) if new_first else 0
    results["col_count_diff"] = "OK" if old_col_count == new_col_count else f"{old_col_count} | {new_col_count}"

    # 3. 按文件各自的分隔符分析列类型（日期 / 数值）
    date_cols_old, numeric_cols_old = analyze_columns(old_sample, old_sep)
    date_cols_new, numeric_cols_new = analyze_columns(new_sample, new_sep)
    # 仅比对“新老双方都被判定为该类型”的列，避免分隔符差异造成的误判
    date_cols = sorted(date_cols_old & date_cols_new)
    numeric_cols = sorted(numeric_cols_old & numeric_cols_new)

    # 4. 行号对齐：日期格式 & 小数精度比对（用各自分隔符切分后按列索引对齐）
    date_diffs = []
    decimal_diffs = []
    min_rows = min(len(old_sample), len(new_sample))
    max_cols = max(old_col_count, new_col_count)

    for r_idx in range(min_rows):
        old_parts = old_sample[r_idx].split(old_sep)
        new_parts = new_sample[r_idx].split(new_sep)
        for c_idx in range(max_cols):
            ov = old_parts[c_idx].strip() if c_idx < len(old_parts) else ""
            nv = new_parts[c_idx].strip() if c_idx < len(new_parts) else ""

            if c_idx in date_cols and ov and nv:
                of = get_standard_format(ov)
                nf = get_standard_format(nv)
                if of and nf and of != nf and len(date_diffs) < 20:
                    date_diffs.append(f"Row{r_idx + 1}|Col{c_idx + 1}: Old[{ov}]({of}) vs New[{nv}]({nf})")

            if c_idx in numeric_cols and ov and nv:
                ot = get_number_format_template(ov)
                nt = get_number_format_template(nv)
                if ot and nt and ot != nt and len(decimal_diffs) < 20:
                    decimal_diffs.append(
                        f"Row{r_idx + 1}|Col{c_idx + 1}: Old[{ov}](int={ot[1]},dec={ot[2]}) "
                        f"vs New[{nv}](int={nt[1]},dec={nt[2]})"
                    )

    results["date_diff"] = "OK" if not date_diffs else "\n".join(date_diffs)
    results["decimal_diff"] = "OK" if not decimal_diffs else "\n".join(decimal_diffs)

    # 5. 内容查找比对（仅当文件大小不一致时执行）：以新文件构建前缀索引，按主键查找对应行后逐列比对
    content_lookup_diffs = []
    if old_size != new_size:
        logger.info(f"File size differs. Performing content lookup comparison for {os.path.basename(old_file)} ...")

        # 5.1 用“新文件分隔符”构建内存索引：逐层前缀 -> 行原始串（保证键唯一才认为是匹配行）
        new_file_index = {}
        for new_line in new_sample:
            new_parts = new_line.split(new_sep)
            clean_parts = [part.strip() for part in new_parts]
            for depth in range(1, len(clean_parts) + 1):
                prefix_key = tuple(clean_parts[:depth])
                new_file_index.setdefault(prefix_key, []).append(new_line)

        # 5.2 用“老文件分隔符”切分每行，逐步增加列数查找唯一匹配
        for old_line in old_sample:
            old_parts = old_line.split(old_sep)
            old_key_base = [part.strip() for part in old_parts]

            target_new_line = None
            for depth in range(1, len(old_key_base) + 1):
                search_key = tuple(old_key_base[:depth])
                matches = new_file_index.get(search_key)
                if matches and len(matches) == 1:
                    target_new_line = matches[0]
                    break

            if target_new_line:
                new_parts = target_new_line.split(new_sep)
                for c_idx in range(max(len(old_parts), len(new_parts))):
                    ov = old_parts[c_idx] if c_idx < len(old_parts) else ""
                    nv = new_parts[c_idx] if c_idx < len(new_parts) else ""
                    if ov != nv and len(content_lookup_diffs) < 5:
                        content_lookup_diffs.append(
                            f"Key:{old_key_base[0]}|Col{c_idx + 1}:Old[{ov}]->New[{nv}]"
                        )
        if content_lookup_diffs:
            results["content_lookup_diff"] = "\n".join(content_lookup_diffs)
        else:
            results["content_lookup_diff"] = "OK (Found but content matches)"
    else:
        results["content_lookup_diff"] = "OK"

    return results


def _write_result(res, output_path=None):
    """将结果 JSON 写入指定文件（优先）；无路径则回退 stdout（兼容手动调用）"""
    json_str = json.dumps(res, ensure_ascii=False)
    if output_path:
        try:
            out_dir = os.path.dirname(output_path)
            if out_dir:
                os.makedirs(out_dir, exist_ok=True)
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write(json_str)
                f.write('\n')
        except Exception as e:
            # 写文件失败时回退到 stdout，确保调用方总能拿到结果
            logger.warning(f"Failed to write result to {output_path}: {e}, falling back to stdout")
            print(json_str)
    else:
        print(json_str)


if __name__ == "__main__":
    if len(sys.argv) != 5:
        print("Usage: python data_compare_tool.py <old_file> <new_file> <old_hex_separator> <new_hex_separator>", file=sys.stderr)
        sys.exit(1)
    old_f = sys.argv[1]
    new_f = sys.argv[2]
    old_hex = sys.argv[3]
    new_hex = sys.argv[4]
    # 支持通过环境变量指定输出文件路径（Shell 批量模式传入），避免 stdout 重定向被污染
    output_json_path = os.environ.get('OUTPUT_JSON_PATH')
    try:
        res = compare_files(old_f, new_f, old_hex, new_hex)
        _write_result(res, output_json_path)
    except Exception as e:
        logger.error(f"Critical Error processing {old_f}: {e}", exc_info=True)
        _write_result(_error_result(old_f, "ERROR", e), output_json_path)

Shell

#!/bin/bash

# ================= 参数校验 =================
# 用法（向后兼容，老新同分隔符）：
#   $0 <file_list> <delimiter_hex> <output_excel>
# 用法（增强，老新不同分隔符）：
#   $0 <file_list> <old_delimiter_hex> <new_delimiter_hex> <output_excel>
#
# file_list 清单每行格式：
#   old_file|new_file
#   old_file|new_file|OLD_HEX|NEW_HEX      # 可选：按行覆盖分隔符（缺省使用全局参数）

if [ "$#" -eq 3 ]; then
    FILE_LIST="$1"
    DELIM_HEX_OLD="$2"
    DELIM_HEX_NEW="$2"   # 同分隔符：新老共用一个
    OUTPUT_EXCEL="$3"
elif [ "$#" -eq 4 ]; then
    FILE_LIST="$1"
    DELIM_HEX_OLD="$2"
    DELIM_HEX_NEW="$3"
    OUTPUT_EXCEL="$4"
else
    echo "Usage: $0 <file_list> <delimiter_hex> <output_excel>"
    echo "   or: $0 <file_list> <old_delimiter_hex> <new_delimiter_hex> <output_excel>"
    exit 1
fi

# ================= Python 解释器探测（兼容 Windows Git Bash / Linux / macOS） =================
if command -v python3 >/dev/null 2>&1; then
    PY=python3
elif command -v python >/dev/null 2>&1; then
    PY=python
else
    echo "[ERROR] Python interpreter not found (need python3 or python)."
    exit 1
fi

# Windows(Git Bash) 下将输出路径转为原生 Windows 路径，便于原生 python 写文件
if command -v cygpath >/dev/null 2>&1; then
    OUTPUT_EXCEL=$(cygpath -w "$OUTPUT_EXCEL" 2>/dev/null || echo "$OUTPUT_EXCEL")
fi
export RESULT_EXCEL="$OUTPUT_EXCEL"

# 检查输入清单是否存在
if [ ! -f "$FILE_LIST" ]; then
    echo "[ERROR] File list not found: $FILE_LIST"
    exit 1
fi

# 创建临时目录，用于存放各个文件的比对结果 JSON
# 注意：用 python 生成原生临时目录，避免 Windows(Git Bash) 下 mktemp 返回 POSIX 路径，
#       而原生 python 无法识别该路径，导致结果 JSON 丢失（多系统兼容性）
TEMP_DIR=$("$PY" -c "import tempfile; print(tempfile.mkdtemp(prefix='cmp_'))")
echo "[INFO] Starting batch comparison in $FILE_LIST ..."
echo "[INFO] Old Delimiter Hex: $DELIM_HEX_OLD | New Delimiter Hex: $DELIM_HEX_NEW"
echo "[INFO] Output Excel: $OUTPUT_EXCEL"
echo "[INFO] Python: $PY | Temp directory: ${TEMP_DIR}"
export TEMP_DIR="$TEMP_DIR"


# 退出时自动清理临时目录
cleanup() {
    echo "[INFO] Cleaning up temp files..."
    rm -rf "${TEMP_DIR}"
}
trap cleanup EXIT

# ================ 并行任务处理 =================
PIDS=()
MAX_PARALLEL=10 # 根据机器CPU核心数调整，避免内存溢出
LINE_NO=0 # 行号计数，用于同名文件(不同目录)临时 JSON 去重

# 逐行读取清单（兼容路径含空格、Windows CRLF 行尾）
while IFS= read -r tmp_info || [ -n "$tmp_info" ]; do
    LINE_NO=$((LINE_NO + 1))
    # 去除 Windows 行尾可能的 \r
    tmp_info="${tmp_info%$'\r'}"
    [ -z "$tmp_info" ] && continue

    old_file=$(echo "$tmp_info" | awk -F'|' '{print $1}')
    new_file=$(echo "$tmp_info" | awk -F'|' '{print $2}')

    # Windows(Git Bash) 下将清单中的路径转为原生 Windows 路径，便于原生 python 打开文件
    if command -v cygpath >/dev/null 2>&1; then
        old_file=$(cygpath -w "$old_file" 2>/dev/null || echo "$old_file")
        new_file=$(cygpath -w "$new_file" 2>/dev/null || echo "$new_file")
    fi
    # 按行覆盖的分隔符（可选）
    line_old_delim=$(echo "$tmp_info" | awk -F'|' '{print $3}')
    line_new_delim=$(echo "$tmp_info" | awk -F'|' '{print $4}')

    # 未指定则用全局分隔符
    if [ -n "$line_old_delim" ]; then
        file_old_delim="$line_old_delim"
    else
        file_old_delim="$DELIM_HEX_OLD"
    fi
    if [ -n "$line_new_delim" ]; then
        file_new_delim="$line_new_delim"
    else
        file_new_delim="$DELIM_HEX_NEW"
    fi

    base_name=$(basename "$new_file")
    # 用行号前缀保证同名文件(不同目录)的临时 JSON 不冲突
    result_file="${TEMP_DIR}/result_${LINE_NO}_${base_name}.json"

    # 检查新文件是否存在
    if [ ! -f "$new_file" ]; then
        echo "[WARN] New file not found, skipping: $new_file"
        "$PY" -c "
import json
data = {
    \"file_name\": \"$(basename "$new_file")\",
    \"old_file_path\": \"$old_file\",
    \"new_file_path\": \"$new_file\",
    \"status\": \"MISSING_NEW_FILE\",
    \"size_diff\": \"\",
    \"count_diff\": \"\",
    \"encoding_diff\": \"\",
    \"col_count_diff\": \"\",
    \"decimal_diff\": \"\",
    \"date_diff\": \"\",
    \"content_lookup_diff\": \"\"
}
with open('${result_file}', 'w') as f:
    json.dump(data, f)
        "
        continue
    fi

    # 检查老文件是否存在
    if [ ! -f "$old_file" ]; then
        echo "[WARN] Old file not found, skipping: $old_file"
        "$PY" -c "
import json
data = {
    \"file_name\": \"$(basename "$old_file")\",
    \"old_file_path\": \"$old_file\",
    \"new_file_path\": \"$new_file\",
    \"status\": \"MISSING_OLD_FILE\",
    \"size_diff\": \"\",
    \"count_diff\": \"\",
    \"encoding_diff\": \"\",
    \"col_count_diff\": \"\",
    \"decimal_diff\": \"\",
    \"date_diff\": \"\",
    \"content_lookup_diff\": \"\"
}
with open('${result_file}', 'w') as f:
    json.dump(data, f)
        "
        continue
    fi

    echo "[INFO] Comparing: $(basename "$old_file") <-> $(basename "$new_file") (sep old=$file_old_delim new=$file_new_delim)"

    # --- 后台执行 Python 比对（分别传入老/新分隔符十六进制） ---
    # 通过环境变量告知 Python 直接写结果文件，避免 stdout 重定向被污染（多系统兼容性）
    export OUTPUT_JSON_PATH="${result_file}"
    "$PY" data_compare_tool.py "$old_file" "$new_file" "$file_old_delim" "$file_new_delim" 2>/dev/null &

    PIDS+=($!)

    # --- 控制并行数量 ---
    if [ ${#PIDS[@]} -ge $MAX_PARALLEL ]; then
        wait -n # 等待任意一个后台任务完成
        for i in "${!PIDS[@]}"; do
            if ! kill -0 "${PIDS[i]}" 2>/dev/null; then
                unset 'PIDS[i]'
            fi
        done
    fi
done < "$FILE_LIST"

# 等待所有剩余的后台任务完成
echo "[INFO] Waiting for all parallel tasks to complete..."
for pid in "${PIDS[@]}"; do
    wait "$pid"
done

# ================ 结果合并与 Excel 生成 =================
echo "[INFO] All comparisons completed. Merging results..."

# 调用 Python 生成 Excel
"$PY" - <<'EOF'
import json
import os
import glob
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
import re

output_path = os.environ.get('RESULT_EXCEL')
TEMP_DIR = os.environ.get('TEMP_DIR')

# 定义非法字符的正则
ILLEGAL_CHARACTERS_RE=re.compile(r'[\x00-\x08\x0b\x0c\x0e-\x1f]')
def safe_text_for_excel(text):
    """
    将这些非法字符转换为可见的转义字符串
    这样既保留了信息, 又不会导致Excel报错
    """
    if not isinstance(text, str):
        return text
    def replace_char(match):
        return f"\\x{ord(match.group()):02x}"
    return ILLEGAL_CHARACTERS_RE.sub(replace_char, text)

# --- 读取所有 JSON 结果（容错：文件被污染时提取第一个合法 JSON 对象） ---
import re

# 括号平衡匹配：提取第一个完整的 { ... } JSON 对象（避免脏数据中多余的大括号干扰）
_JSON_OBJECT_RE = re.compile(r'(\{(?:[^{}]|(?:\{[^{}]*\}))*\})')


def _load_json_robust(filepath):
    """尝试标准 json.load；失败时用正则提取第一个完整 JSON 对象（容错脏数据）"""
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            return json.load(f)
    except json.JSONDecodeError:
        pass
    # 容错：提取第一个顶层 { ... } 段
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            raw = f.read()
        m = _JSON_OBJECT_RE.search(raw)
        if m:
            obj = json.loads(m.group(1))
            print(f"[WARN] {filepath} had extra data after JSON, parsed first object only")
            return obj
    except Exception:
        pass
    print(f"[ERROR] Failed to parse {filepath}")
    return None


results = []
json_files = glob.glob(os.path.join(TEMP_DIR, 'result_*.json'))
for jf in sorted(json_files):
    data = _load_json_robust(jf)
    if data is not None:
        results.append(data)

# --- 创建 Excel 报告 ---
wb = Workbook()
ws = wb.active
ws.title = "Comparison Report"

# 写入表头
headers = ["Old File Path", "New File Path", "Status", "File Size Diff", "Row Count Diff", "Encoding Diff", "Col Count Diff", "Decimal Precision", "Date Format", "Content Diff"]
ws.append(headers)

# 设置表头样式
header_font = Font(bold=True)
for cell in ws[1]:
    cell.font = header_font

# 填充数据
header_fill = PatternFill(start_color='0CC5A0', end_color='0CC5A0', fill_type='solid') # 青色
file_fill = PatternFill(start_color='2AC2D1', end_color='2AC2D1', fill_type='solid') # 蓝色
red_fill = PatternFill(start_color='FFC7CE', end_color='FFC7CE', fill_type='solid') # 红色
green_fill = PatternFill(start_color='C6EFCE', end_color='C6EFCE', fill_type='solid') # 绿色

# 定义细实线边框
thin_border = Border(
    left = Side(style='thin'),
    right = Side(style='thin'),
    top = Side(style='thin'),
    bottom = Side(style='thin'),
)

for item in results:
    row = [
        safe_text_for_excel(item.get("old_file_path", "Unknown")),
        safe_text_for_excel(item.get("new_file_path", "Unknown")),
        safe_text_for_excel(item.get("status", "OK")),
        safe_text_for_excel(item.get("size_diff", "OK")),
        safe_text_for_excel(item.get("count_diff", "OK")),
        safe_text_for_excel(item.get("encoding_diff", "OK")),
        safe_text_for_excel(item.get("col_count_diff", "OK")),
        safe_text_for_excel(item.get("decimal_diff", "OK")),
        safe_text_for_excel(item.get("date_diff", "OK")),
        safe_text_for_excel(item.get("content_lookup_diff", "OK"))
    ]
    ws.append(row)

    # --- 颜色标记逻辑 ---
    for row_idx, row_cells in enumerate(ws.iter_rows()):
        if row_idx == 0:
            for col_idx, cell in enumerate(row_cells):
                cell.fill = header_fill
                cell.border = thin_border
            continue
        for col_idx, cell in enumerate(row_cells):
            if col_idx in (0, 1):
                cell.fill = file_fill
                cell.border = thin_border
                continue
            val = str(cell.value)
            if val != 'OK' and val != 'N/A':
                cell.fill = red_fill
            elif val == 'OK':
                cell.fill = green_fill
            cell.border = thin_border

# --- 自动调整列宽 ---
for column_cells in ws.columns:
    max_length = 0
    col_letter = column_cells[0].column_letter
    for cell in column_cells:
        try:
            cell_len = len(str(cell.value)) if cell.value else 0
            if cell_len > max_length:
                max_length = cell_len
        except:
            pass
    adjusted_width = min(max_length + 2, 50)
    ws.column_dimensions[col_letter].width = adjusted_width

# --- 保存文件 ---
output_dir = os.path.dirname(output_path)
if output_dir and not os.path.exists(output_dir):
    os.makedirs(output_dir, exist_ok=True)

wb.save(output_path)
print(f"[SUCCESS] Final report generated: {output_path} ({len(results)} files processed)")
EOF

蚂蚁🐜再小也是肉🥩！

编程语言 > Python > Shell

#Linux #Python

数据文件比对工具

http://heibanbai.com.cn/posts/b83c204b/

作者

黑伴白

发布于

2026年8月14日

许可协议

“您的支持，我的动力！觉得不错的话，给点打赏吧 ୧(๑•̀⌄•́๑)૭”

微信支付

支付宝支付