Scan to join WeChat groupCT INV PKL 汇总合并 - 北江仓多板数据汇总
概述
将越南 北江仓 多个批次的 CT INV PKL Excel 文件汇总合并到目标文件中。
适用场景：跨境物流报关单据中，多个批次（1板、3板、6板等）的 Packing List 清单汇总。
触发词
- CT INV PKL 汇总
- 北江仓汇总
- 汇总 CT INV PKL
- 报关清单汇总
输入文件
数据源文件（多个批次）
MẪU ANNEX CT- INV-PKL-BS26-090701越南北江仓.xlsx- 1板MẪU ANNEX CT- INV-PKL-BS26-090702越南北江仓-1板.xlsx- 1板MẪU ANNEX CT- INV-PKL-BS26-090703越南北江仓-3板.xlsx- 3板MẪU ANNEX CT- INV-PKL-BS26-090704越南北江仓-6板.xlsx- 6板
目标文件
9月10日 桂FD8288 CT INV PK-YTD-BD宇飞11板.xlsx或类似命名
输出
- 汇总后的 Excel 文件（PL清单 sheet 含 357 行数据）
- Item 序号连续递增 (1, 2, 3, ...)
- BILL sheet 装运日期更新为今天，车牌清除
工作流程
步骤1: 分析数据源
import openpyxl
import json
import os
# 数据源文件列表
source_files = [
'MẪU ANNEX CT- INV-PKL-BS26-090701越南北江仓.xlsx',
'MẪU ANNEX CT- INV-PKL-BS26-090702越南北江仓-1板.xlsx',
'MẪU ANNEX CT- INV-PKL-BS26-090703越南北江仓-3板.xlsx',
'MẪU ANNEX CT- INV-PKL-BS26-090704越南北江仓-6板.xlsx'
]
# 目标文件
target_file = '9月10日 桂FD8288 CT INV PK-YTD-BD宇飞11板.xlsx'
# 读取所有数据源
all_data_rows = []
for src_file in source_files:
wb_src = openpyxl.load_workbook(src_file)
ws_src = wb_src['PL清单']
# 找到header行 (Item, Part No., ...)
header_row = 14
for i in range(10, 20):
cell = ws_src.cell(row=i, column=1)
if cell.value == 'Item':
header_row = i
break
# 读取数据行 (从 header_row + 1 开始)
row_num = header_row + 1
while True:
part_no = ws_src.cell(row=row_num, column=2).value
if part_no is None:
break
row_data = {
'item': ws_src.cell(row=row_num, column=1).value,
'part_no': part_no,
'commodity': ws_src.cell(row=row_num, column=3).value,
'commodity_cn': ws_src.cell(row=row_num, column=4).value,
'specifications': ws_src.cell(row=row_num, column=5).value,
'model': ws_src.cell(row=row_num, column=6).value,
'brand': ws_src.cell(row=row_num, column=7).value,
'carton_no': ws_src.cell(row=row_num, column=8).value,
'coo': ws_src.cell(row=row_num, column=9).value,
'unit': ws_src.cell(row=row_num, column=10).value,
'quantity': ws_src.cell(row=row_num, column=11).value,
'cartons': ws_src.cell(row=row_num, column=12).value,
'pallet': ws_src.cell(row=row_num, column=13).value,
'nw': ws_src.cell(row=row_num, column=14).value,
'gw': ws_src.cell(row=row_num, column=15).value,
'volume': ws_src.cell(row=row_num, column=16).value,
'source_file': src_file,
'source_row': row_num
}
all_data_rows.append(row_data)
row_num += 1
wb_src.close()
print(f'总计读取: {len(all_data_rows)} 行数据')
# 保存数据
with open('merge_data.json', 'w', encoding='utf-8') as f:
json.dump({
'data_rows': all_data_rows,
'target_file': target_file,
'start_row': 15
}, f, ensure_ascii=False, default=str)
步骤2: 写入目标文件
import openpyxl
from openpyxl.cell.cell import MergedCell
import json
# 读取数据
with open('merge_data.json', 'r', encoding='utf-8') as f:
data = json.load(f)
data_rows = data['data_rows']
target_file = data['target_file']
# 打开目标文件
wb = openpyxl.load_workbook(target_file)
ws = wb['PL清单']
# 找到数据结束行
last_data_row = 14
for i in range(15, ws.max_row + 1):
cell = ws.cell(row=i, column=2)
if cell.value is not None:
last_data_row = i
print(f'旧数据结束行: {last_data_row}')
# 清除旧数据
for row_num in range(15, last_data_row + 1):
for col_num in range(1, 15):
cell = ws.cell(row=row_num, column=col_num)
if isinstance(cell, MergedCell):
continue
cell.value = None
# 写入新数据
start_row = 15
target_col_map = {
'item': 1,
'part_no': 2,
'commodity': 3,
'specifications': 4,
'model': 5,
'brand': 6,
'coo': 7,
'unit': 8,
'quantity': 9,
'cartons': 10,
'pallet': 11,
'nw': 12,
'gw': 13,
'volume': 14
}
for idx, row_data in enumerate(data_rows):
row_num = start_row + idx
# Item 序号 (连续递增)
cell = ws.cell(row=row_num, column=1)
if not isinstance(cell, MergedCell):
cell.value = idx + 1
# 其他字段
for field_name, col_num in target_col_map.items():
if field_name in row_data and row_data[field_name] is not None:
cell = ws.cell(row=row_num, column=col_num)
if not isinstance(cell, MergedCell):
cell.value = row_data[field_name]
# 保存
output_file = target_file.replace('.xlsx', '_汇总后.xlsx')
wb.save(output_file)
wb.close()
步骤3: BILL 后处理
import openpyxl
from datetime import datetime
output_file = '9月10日 桂FD8288 CT INV PK-YTD-BD宇飞11板_汇总后.xlsx'
wb = openpyxl.load_workbook(output_file)
ws = wb['BILL']
# 查找并更新装运日期
for row in range(1, 20):
for col in range(1, 10):
cell = ws.cell(row=row, column=col)
if cell.value and 'Date of Loading' in str(cell.value):
next_cell = ws.cell(row=row+1, column=col)
next_cell.value = datetime.now().strftime('%Y-%m-%d')
break
# 查找并清除车牌
for row in range(1, 30):
for col in range(1, 15):
cell = ws.cell(row=row, column=col)
if cell.value and 'TRUCK NO' in str(cell.value).upper():
next_cell = ws.cell(row=row+1, column=col)
next_cell.value = None
break
wb.save(output_file)
wb.close()
注意事项
- 文件占用：如果目标文件被腾讯文档预览占用，需要保存到新文件（如
_v2.xlsx,_v3.xlsx） - 合并单元格：写入时需检查 MergedCell，避免 AttributeError
- Item 序号：必须连续递增，不能保留原数据源的序号
- 仅汇总模式：本 skill 执行"仅汇总不合并"，保留所有数据行
- GW/Volume：数据源中这两列数据较少，大部分行为空，这是正常的
步骤4: 生成统计报告（可选）
汇总完成后，自动生成一份统计报告，包含：
- 各 sheet 汇总的数据条数
- 汇总的文件数量及各文件对应的数据条数
- 汇总后各数值列的合计（Quantity、Cartons、NW、GW等）
import openpyxl
import json
from datetime import datetime
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
# 读取合并数据
with open('merge_data.json', 'r', encoding='utf-8') as f:
data = json.load(f)
data_rows = data['data_rows']
# 1. 按文件统计行数
file_stats = {}
for row in data_rows:
src = row['source_file']
if src not in file_stats:
file_stats[src] = 0
file_stats[src] += 1
# 2. 各列合计统计
qty_sum = 0
cartons_sum = 0
nw_sum = 0
gw_count = 0
gw_sum = 0
volume_count = 0
for row in data_rows:
# Quantity
qty = row.get('quantity')
if qty is not None and isinstance(qty, (int, float)):
qty_sum += qty
# Cartons
cartons = row.get('cartons')
if cartons is not None and isinstance(cartons, (int, float)):
cartons_sum += cartons
# NW
nw = row.get('nw')
if nw is not None and isinstance(nw, (int, float)):
nw_sum += nw
# GW
gw = row.get('gw')
if gw is not None and isinstance(gw, (int, float)):
gw_count += 1
gw_sum += gw
# Volume
vol = row.get('volume')
if vol is not None:
volume_count += 1
# 3. 生成统计报告 Excel
wb_report = Workbook()
ws = wb_report.active
ws.title = "汇总统计报告"
# 样式
header_font = Font(bold=True, size=12)
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
header_font_white = Font(bold=True, size=12, color="FFFFFF")
center_align = Alignment(horizontal='center', vertical='center')
# 标题
ws['A1'] = "CT INV PKL 汇总统计报告"
ws['A1'].font = Font(bold=True, size=14)
ws.merge_cells('A1:D1')
ws['A2'] = f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
ws['A2'].font = Font(size=10)
# 汇总概览
ws['A4'] = "汇总概览"
ws['A4'].font = header_font
ws.merge_cells('A4:B4')
ws['A5'] = "汇总文件数量"
ws['B5'] = len(file_stats)
ws['A6'] = "汇总数据总条数"
ws['B6'] = len(data_rows)
# 按文件统计
ws['A8'] = "按文件统计"
ws['A8'].font = header_font
ws.merge_cells('A8:B8')
ws['A9'] = "文件名"
ws['B9'] = "数据条数"
ws['A9'].font = header_font_white
ws['B9'].font = header_font_white
ws['A9'].fill = header_fill
ws['B9'].fill = header_fill
row_idx = 10
for filename, count in file_stats.items():
ws[f'A{row_idx}'] = filename
ws[f'B{row_idx}'] = count
row_idx += 1
# 各列合计
ws['A{row_idx+1}'] = "数值列合计"
ws['A{row_idx+1}'].font = header_font
ws.merge_cells(f'A{row_idx+1}:B{row_idx+1}')
ws['A{row_idx+2}'] = "Quantity 合计"
ws['B{row_idx+2}'] = qty_sum
ws['A{row_idx+3}'] = "Cartons 合计"
ws['B{row_idx+3}'] = cartons_sum
ws['A{row_idx+4}'] = "NW 合计 (kg)"
ws['B{row_idx+4}'] = round(nw_sum, 4)
ws['A{row_idx+5}'] = "GW 合计 (kg)"
ws['B{row_idx+5}'] = round(gw_sum, 2) if gw_sum else 0
ws['A{row_idx+6}'] = "GW 有数据行数"
ws['B{row_idx+6}'] = gw_count
ws['A{row_idx+7}'] = "Volume 有数据行数"
ws['B{row_idx+7}'] = volume_count
# 调整列宽
ws.column_dimensions['A'].width = 50
ws.column_dimensions['B'].width = 20
# 保存报告
report_file = target_file.replace('.xlsx', '_统计报告.xlsx')
wb_report.save(report_file)
print(f'统计报告已生成: {report_file}')
列映射参考
| 目标列 | 字段名 | 说明 | |--------|--------|------| | A | item | 序号（连续递增） | | B | part_no | 物料号 | | C | commodity | 商品名称 | | D | specifications | 规格 | | E | model | 型号 | | F | brand | 品牌 | | G | coo | 原产国 | | H | unit | 单位 | | I | quantity | 数量 | | J | cartons | 箱数 | | K | pallet | 托盘号 | | L | nw | 净重 | | M | gw | 毛重 | | N | volume | 体积 |
依赖
- Python 3.9+
- openpyxl