脚本怎样汇总合并任务结果

wen 实用脚本 2026-06-27 32

本文目录导读：

脚本怎样汇总合并任务结果

Python脚本汇总合并（推荐）
Shell脚本批量合并
数据库结果汇总
按条件汇总合并
高级汇总：带进度条和错误处理
特定场景脚本
使用建议

我来为您介绍几种常见的脚本来汇总合并任务结果的方法。

Python脚本汇总合并（推荐）

合并CSV文件

import pandas as pd
import glob
import os
# 合并多个CSV文件
def merge_csv_files(input_pattern, output_file):
    # 获取所有匹配的文件
    csv_files = glob.glob(input_pattern)
    # 读取并合并所有CSV文件
    df_list = []
    for file in csv_files:
        df = pd.read_csv(file)
        df_list.append(df)
    # 合并所有数据框
    merged_df = pd.concat(df_list, ignore_index=True)
    # 保存合并后的文件
    merged_df.to_csv(output_file, index=False)
    print(f"合并完成！共合并 {len(csv_files)} 个文件")
    print(f"总行数: {len(merged_df)}")
# 使用示例
merge_csv_files("results/*.csv", "merged_results.csv")

合并JSON文件

import json
import glob
def merge_json_files(input_pattern, output_file):
    merged_data = []
    # 读取所有JSON文件
    for file in glob.glob(input_pattern):
        with open(file, 'r') as f:
            data = json.load(f)
            # 处理单个对象或列表
            if isinstance(data, list):
                merged_data.extend(data)
            else:
                merged_data.append(data)
    # 保存合并结果
    with open(output_file, 'w') as f:
        json.dump(merged_data, f, indent=2)
    print(f"合并完成！共 {len(merged_data)} 条记录")
# 使用示例
merge_json_files("results/*.json", "merged_results.json")

Shell脚本批量合并

合并文本文件

#!/bin/bash
# 合并所有txt文件
input_dir="results"
output_file="merged_results.txt"
# 清空输出文件
> $output_file
# 添加文件名标识并合并
for file in $input_dir/*.txt; do
    echo "=== $file ===" >> $output_file
    cat "$file" >> $output_file
    echo "" >> $output_file
done
echo "合并完成！共处理 $(ls $input_dir/*.txt | wc -l) 个文件"

合并CSV文件（保留表头）

#!/bin/bash
# 合并CSV文件，保留第一个文件的表头
first_file=true
for file in results/*.csv; do
    if [ "$first_file" = true ]; then
        cat "$file" > merged_results.csv
        first_file=false
    else
        # 跳过第一行（表头）
        tail -n +2 "$file" >> merged_results.csv
    fi
done
echo "CSV文件合并完成！"

数据库结果汇总

合并SQL查询结果

import sqlite3
import pandas as pd
def merge_database_results():
    # 连接数据库
    conn = sqlite3.connect('results.db')
    # 查询所有表
    tables = pd.read_sql("SELECT name FROM sqlite_master WHERE type='table'", conn)
    # 合并所有表
    all_data = []
    for table in tables['name']:
        df = pd.read_sql(f"SELECT * FROM {table}", conn)
        all_data.append(df)
    # 合并结果
    merged_df = pd.concat(all_data, ignore_index=True)
    # 保存到Excel
    merged_df.to_excel('merged_results.xlsx', index=False)
    print(f"合并完成！共 {len(merged_df)} 条记录")
# 使用示例
merge_database_results()

按条件汇总合并

import pandas as pd
import glob
def conditional_merge():
    # 读取所有结果文件
    all_files = glob.glob("results/task_*.csv")
    # 按任务类型分类合并
    task_results = {}
    for file in all_files:
        # 从文件名提取任务类型
        task_type = file.split('_')[1]  #  task_type_1.csv
        df = pd.read_csv(file)
        if task_type not in task_results:
            task_results[task_type] = []
        task_results[task_type].append(df)
    # 按类型分别合并
    for task_type, df_list in task_results.items():
        merged = pd.concat(df_list, ignore_index=True)
        merged.to_csv(f"merged_{task_type}.csv", index=False)
        # 统计信息
        print(f"{task_type}: {len(merged)} 条记录")
        print(f"  文件数: {len(df_list)}")
# 使用示例
conditional_merge()

高级汇总：带进度条和错误处理

import pandas as pd
import glob
from tqdm import tqdm
import logging
# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def robust_merge(input_pattern, output_file):
    """稳健的文件合并，带进度条和错误处理"""
    files = glob.glob(input_pattern)
    successful = 0
    failed = 0
    df_list = []
    # 使用进度条
    for file in tqdm(files, desc="合并进度"):
        try:
            df = pd.read_csv(file)
            df_list.append(df)
            successful += 1
        except Exception as e:
            logger.error(f"文件 {file} 读取失败: {e}")
            failed += 1
    if df_list:
        # 合并数据
        merged_df = pd.concat(df_list, ignore_index=True)
        # 数据清洗
        merged_df.drop_duplicates(inplace=True)
        merged_df.fillna('', inplace=True)
        # 保存结果
        merged_df.to_csv(output_file, index=False)
        # 生成汇总报告
        report = {
            '总文件数': len(files),
            '成功处理': successful,
            '失败数量': failed,
            '总记录数': len(merged_df),
            '输出文件': output_file
        }
        print("\n=== 合并报告 ===")
        for key, value in report.items():
            print(f"{key}: {value}")
        return merged_df
    else:
        logger.error("没有成功读取任何文件！")
        return None
# 使用示例
result = robust_merge("results/*.csv", "final_merged.csv")

特定场景脚本

合并日志文件

#!/bin/bash
# 按时间排序合并日志
for file in $(ls -tr logs/*.log); do
    echo "[$(date -r "$file" '+%Y-%m-%d %H:%M:%S')] $(basename $file)" >> merged_log.txt
    cat "$file" >> merged_log.txt
    echo "---" >> merged_log.txt
done

合并多线程运行结果

import concurrent.futures
import pandas as pd
def process_batch(files):
    """并行处理多个文件"""
    results = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        # 提交所有任务
        future_to_file = {executor.submit(pd.read_csv, file): file 
                         for file in files}
        # 收集结果
        for future in concurrent.futures.as_completed(future_to_file):
            file = future_to_file[future]
            try:
                data = future.result()
                results.append(data)
            except Exception as e:
                print(f"处理 {file} 时出错: {e}")
    # 合并所有结果
    return pd.concat(results, ignore_index=True)
# 使用示例
files = glob.glob("batch_results/*.csv")
merged_data = process_batch(files)
merged_data.to_csv("parallel_merged.csv", index=False)

使用建议

选择合适工具：
- 小文件用Shell脚本
- 复杂处理用Python
- 大数据用Spark或数据库
注意性能：
- 大文件分批处理
- 使用生成器减少内存占用
- 考虑并行处理
数据验证：
- 添加数据完整性检查
- 记录处理日志
- 实现错误恢复机制

根据您的具体需求选择适合的脚本，如果需要处理特定格式或特殊需求,请提供更多细节。