quality_tools/tools/分数字段结果分析/result_export_ai.py

from datetime import datetime
import pandas as pd
import numpy as np
from pymongo import MongoClient
from openpyxl import load_workbook

#高质量字段以及错误原因
# 将这个函数定义放在你的脚本顶部或其他函数定义附近
def convert_numpy_int(obj):
    if isinstance(obj, np.int64):
        return int(obj)
    elif isinstance(obj, dict):
        return {key: convert_numpy_int(value) for key, value in obj.items()}
    elif isinstance(obj, list):
        return [convert_numpy_int(elem) for elem in obj]
    else:
        return obj


# MongoDB连接配置
host = '192.168.3.149'  # MongoDB主机地址
port = 27180  # MongoDB端口
dbname = 'data_quality'  # 数据库名称
collection_name = 'bidding_20241219_ai'  # 集合名称

# 创建MongoDB连接
client = MongoClient(host, port)
db = client[dbname]
collection = db[collection_name]
# 定义字段中英文映射
column_name_mapping = {
    "area_qa": "省份",
    "bidamount_qa": "中标金额",
    "budget_qa": "预算",
    "buyer_qa": "采购单位",
    "multipackage_qa": "分包",
    "projectcode_qa": "项目编号",
    "projectname_qa": "项目名称",
    "title_qa": "标题",
    "winner_qa": "中标单位",
    "score": "标讯总分数",
    "bidopentime_qa": "开标时间",
    "publishtime_qa": "发布时间",
    "toptype_qa": "信息一级分类",
    "subtype_qa": "信息二级分类"
}

# 从MongoDB读取数据
data = pd.DataFrame(list(collection.find({},{k:1 for k,v in column_name_mapping.items()})))


# 选择字段名以 '_qa' 结尾的列
qa_columns = [col for col in data.columns if col.endswith('_qa')]

# 仅保留 '_qa' 结尾的字段，并进行列名映射
data = data[qa_columns]
data.rename(columns=column_name_mapping, inplace=True)

# 输出当前的数据列名
print("当前的列名：")
print(data.columns)


# analyze_column 函数，处理 NaN 值
def analyze_column(dataframe, column_name):
    if column_name not in dataframe.columns:
        total = len(dataframe)
        correct = total
        error = 0
    else:
        total = len(dataframe[column_name])
        correct = dataframe[column_name].apply(lambda x: pd.isna(x) or x == {}).sum()
        error = total - correct

    accuracy = correct / total if total > 0 else 0
    error_rate = error / total if total > 0 else 0

    # 收集错误原因
    error_reasons = dataframe[column_name].apply(
        lambda x: x if x != {} and not pd.isna(x) else None).dropna().value_counts()

    return total, correct, error, accuracy, error_rate, error_reasons


# 重新格式化错误原因的数据结构
def reformat_error_reasons_safe(error_reasons_series):
    reformatted_reasons = {}

    for error_dict, count in error_reasons_series.items():
        if isinstance(error_dict, dict):  # 如果是字典类型的错误原因
            for error_code, reason in error_dict.items():
                if ',' in reason:
                    parts = reason.split(',')
                    formatted_reason = parts[1].strip()
                else:
                    formatted_reason = reason.strip()

                if formatted_reason:
                    key = (formatted_reason,)
                    if key not in reformatted_reasons:
                        reformatted_reasons[key] = count
                    else:
                        reformatted_reasons[key] += count
        elif isinstance(error_dict, list):  # 如果是列表类型的错误原因
            key = (tuple(error_dict),) if error_dict else None
            if key not in reformatted_reasons:
                reformatted_reasons[key] = count
            else:
                reformatted_reasons[key] += count
        else:  # 其他类型的错误原因
            key = (error_dict,) if error_dict else None
            if key not in reformatted_reasons:
                reformatted_reasons[key] = count
            else:
                reformatted_reasons[key] += count

    formatted_results = {
        str(key[0]): value for key, value in reformatted_reasons.items() if key and key[0] != ''
    }
    return formatted_results


# 对每个字段进行分析
fields_to_analyze = data.columns  # 直接使用已选定的 '_qa' 字段
expanded_analysis_results = []

for col in fields_to_analyze:
    total, correct, error, accuracy, error_rate, error_reasons = analyze_column(data, col)
    reformatted_error_reasons = reformat_error_reasons_safe(error_reasons)

    for reason, count in reformatted_error_reasons.items():
        reason = str(reason).replace('(', '').replace(',)', '').replace("'", '')
        if error > 0:
            single_reason_error_rate = count / error
        else:
            single_reason_error_rate = 0  # 防止除以零的情况

        expanded_analysis_results.append({
            '字段': col,
            '总量': total,
            '正确数量': correct,
            '错误数量': error,
            '正确率': f'{accuracy:.2%}',
            '错误率': f'{error_rate:.2%}',
            '错误原因': reason,
            '错误次数': count,
            '单个原因错误率': f'{single_reason_error_rate:.2%}'
        })

# 创建DataFrame并进行写入操作
expanded_analysis_results_df = pd.DataFrame(expanded_analysis_results)

# 使用 pd.ExcelWriter 进行写入操作
with pd.ExcelWriter('临时文件.xlsx', engine='openpyxl') as writer:
    # 将分析结果写入Excel
    expanded_analysis_results_df.to_excel(writer, sheet_name='字段分析结果', index=False)

# 假设您的分析结果已经保存在一个临时文件中
temp_analysis_file = '临时文件.xlsx'  # 临时文件的路径

# 加载您想要合并结果到的Excel文件
modified_file_path = 'pin.xlsx'  # 拼接文件路径
wb = load_workbook(modified_file_path)

# 加载包含分析结果的临时Excel文件
temp_wb = load_workbook(temp_analysis_file)

# 将临时文件中的工作表复制到修改过的文件中
for sheet_name in temp_wb.sheetnames:
    source = temp_wb[sheet_name]
    target = wb.create_sheet(sheet_name)

    for row in source.iter_rows(min_row=1, max_col=source.max_column, max_row=source.max_row, values_only=True):
        target.append(row)

# 保存最终的合并文件
final_merged_file_path = '质量分析报告.xlsx'  # 最终合并文件的路径
wb.save(final_merged_file_path)

# 关闭MongoDB连接
client.close()