177 lines
7.2 KiB
Python
177 lines
7.2 KiB
Python
import pandas as pd
|
||
import os
|
||
from openpyxl import load_workbook
|
||
from openpyxl.styles import Alignment
|
||
import xlsxwriter
|
||
import re
|
||
import argparse
|
||
|
||
def normalize_product_name(product):
|
||
"""彻底规范化产品名：去除所有空白/不可见字符，统一全角半角"""
|
||
if product is None:
|
||
return ''
|
||
name = str(product)
|
||
name = re.sub(r'[\s\u3000]+', '', name) # 去掉所有空白和全角空格
|
||
name = name.replace('\n', '').replace('\r', '')
|
||
name = name.replace('（', '(').replace('）', ')') # 全角括号转半角
|
||
return name.strip()
|
||
|
||
def load_product_reviewer_mapping(mapping_file):
|
||
mapping = {}
|
||
try:
|
||
with open(mapping_file, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line and ':' in line:
|
||
product, reviewer = line.split(':', 1)
|
||
norm_product = normalize_product_name(product)
|
||
mapping[norm_product] = reviewer.strip()
|
||
except FileNotFoundError:
|
||
print(f"警告: 未找到映射文件 {mapping_file}")
|
||
return mapping
|
||
|
||
def prepare_sheet(df, sheet_name, reviewer_mapping):
|
||
"""准备工作表数据，并填充审核人"""
|
||
# 按源码包全名和关联产品排序
|
||
df['源码包全名'] = df['源码包全名'].astype(str).str.strip()
|
||
df['关联产品'] = df['关联产品'].astype(str).str.strip()
|
||
df.sort_values(by=['源码包全名', '关联产品'], inplace=True)
|
||
|
||
# 添加新列（如果不存在）
|
||
new_cols = ['审核人', '审核意见（可发布、不可发布）', '备注：审核人+备注信息（不可发布的关键信息可适当加红）', '技术经理审核意见（可发布、不可发布）']
|
||
for col in new_cols:
|
||
if col not in df.columns:
|
||
df[col] = ''
|
||
|
||
# 用规范化产品名填充审核人
|
||
df['关联产品_规范化'] = df['关联产品'].apply(normalize_product_name)
|
||
df['审核人'] = df['关联产品_规范化'].map(reviewer_mapping).fillna('')
|
||
# print(df[df['审核人']=='']['关联产品'])
|
||
|
||
# 根据工作表名称设置不同的列顺序
|
||
if sheet_name == 'bug':
|
||
columns_order = [
|
||
'源码包全名', '关联产品', '审核人', '审核意见（可发布、不可发布）', '备注：审核人+备注信息（不可发布的关键信息可适当加红）',
|
||
'技术经理审核意见（可发布、不可发布）', '涉及架构', '责任人', '任务类型', '禅道编号',
|
||
'紧急程度', '修改描述信息-对外', '复现方法', '修复方法', '测试方法',
|
||
'测试结果', '登记日期', '更新时间', '二进制包名'
|
||
]
|
||
elif sheet_name == 'cve':
|
||
columns_order = [
|
||
'源码包全名', '关联产品', '审核人', '审核意见（可发布、不可发布）', '备注：审核人+备注信息（不可发布的关键信息可适当加红）',
|
||
'技术经理审核意见（可发布、不可发布）', '涉及架构', '责任人', '任务类型', '风险等级',
|
||
'漏洞编号', '登记日期', '更新时间', '验证情况', '修改内容',
|
||
'二进制包名'
|
||
]
|
||
else:
|
||
columns_order = df.columns.tolist()
|
||
|
||
existing_columns = [col for col in columns_order if col in df.columns]
|
||
extra_columns = [col for col in df.columns if col not in existing_columns and col != '关联产品_规范化']
|
||
final_columns = existing_columns + extra_columns
|
||
df = df[final_columns]
|
||
|
||
return df
|
||
|
||
def process_sheet(ws):
|
||
"""视觉合并单元格，不处理审核人"""
|
||
# 设置列宽
|
||
col_widths = {
|
||
'A': 300, 'B': 300, 'C': 300, 'D': 50, 'E': 200,
|
||
'F': 50, 'G': 15, 'H': 15, 'I': 10, 'J': 15,
|
||
'K': 10, 'L': 40, 'M': 40, 'N': 40, 'O': 40,
|
||
'P': 20, 'Q': 20, 'R': 20, 'S': 60
|
||
}
|
||
for col, width in col_widths.items():
|
||
if col in ws.column_dimensions:
|
||
ws.column_dimensions[col].width = width
|
||
|
||
# 文本自动换行 顶部对齐
|
||
for row in ws.iter_rows():
|
||
for cell in row:
|
||
cell.alignment = Alignment(wrap_text=False, vertical='top')
|
||
|
||
# 合并逻辑
|
||
current_package = None
|
||
start_row = 2
|
||
package_merge_ranges = {}
|
||
|
||
# 合并A列
|
||
for row in range(2, ws.max_row + 1):
|
||
package = ws[f'A{row}'].value
|
||
if package != current_package:
|
||
if current_package is not None:
|
||
end_row = row - 1
|
||
if start_row < end_row:
|
||
ws.merge_cells(f'A{start_row}:A{end_row}')
|
||
package_merge_ranges[current_package] = (start_row, end_row)
|
||
current_package = package
|
||
start_row = row
|
||
if current_package is not None and start_row < ws.max_row:
|
||
ws.merge_cells(f'A{start_row}:A{ws.max_row}')
|
||
package_merge_ranges[current_package] = (start_row, ws.max_row)
|
||
|
||
# 合并B、C、D、E、F列
|
||
for package, (start, end) in package_merge_ranges.items():
|
||
current_product = None
|
||
product_start = start
|
||
for row in range(start, end + 1):
|
||
product = ws[f'B{row}'].value
|
||
if product != current_product:
|
||
if current_product is not None:
|
||
product_end = row - 1
|
||
if product_start < product_end:
|
||
for col in ['B', 'C', 'D', 'E', 'F']:
|
||
ws.merge_cells(f'{col}{product_start}:{col}{product_end}')
|
||
current_product = product
|
||
product_start = row
|
||
if current_product is not None and product_start <= end:
|
||
if product_start < end:
|
||
for col in ['B', 'C', 'D', 'E', 'F']:
|
||
ws.merge_cells(f'{col}{product_start}:{col}{end}')
|
||
|
||
return ws
|
||
|
||
def process_integration_sheet(input_file, output_file, sheet_names, mapping_file):
|
||
reviewer_mapping = load_product_reviewer_mapping(mapping_file)
|
||
xl = pd.ExcelFile(input_file)
|
||
temp_file = "temp_output.xlsx"
|
||
writer = pd.ExcelWriter(temp_file, engine='xlsxwriter')
|
||
|
||
for sheet_name in xl.sheet_names:
|
||
df = xl.parse(sheet_name)
|
||
if sheet_name in sheet_names:
|
||
df = prepare_sheet(df, sheet_name, reviewer_mapping)
|
||
df.to_excel(writer, sheet_name=sheet_name, index=False)
|
||
writer.close()
|
||
|
||
wb = load_workbook(temp_file)
|
||
for sheet_name in wb.sheetnames:
|
||
if sheet_name in sheet_names:
|
||
ws = wb[sheet_name]
|
||
process_sheet(ws)
|
||
wb.save(output_file)
|
||
print(f"处理完成，结果已保存到: {output_file}")
|
||
try:
|
||
os.remove(temp_file)
|
||
except OSError:
|
||
pass
|
||
|
||
if __name__ == "__main__":
|
||
|
||
|
||
parser = argparse.ArgumentParser(description="Excel集成单处理程序")
|
||
parser.add_argument('input_file', help='输入文件路径')
|
||
parser.add_argument('output_file', help='输出文件路径')
|
||
parser.add_argument('--mapping_file', default='product_responsible_mapping.txt', help='产品审核人映射文件路径')
|
||
parser.add_argument('--sheets', nargs='+', default=['bug', 'cve'], help='要处理的工作表名称(可多个)')
|
||
|
||
args = parser.parse_args()
|
||
|
||
process_integration_sheet(
|
||
input_file=args.input_file,
|
||
output_file=args.output_file,
|
||
sheet_names=args.sheets,
|
||
mapping_file=args.mapping_file
|
||
)
|