291 lines
13 KiB
Python
291 lines
13 KiB
Python
import pandas as pd
|
||
import os
|
||
|
||
|
||
def process_contracts(file_path):
|
||
print(f"正在读取并处理文件: {file_path} ...")
|
||
|
||
# --- 1. 读取CSV文件 (容错处理) ---
|
||
df = None
|
||
encodings = ['utf-8', 'gbk', 'gb18030']
|
||
for enc in encodings:
|
||
try:
|
||
df = pd.read_csv(file_path, encoding=enc)
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
|
||
if df is None:
|
||
try:
|
||
print("注意: 标准编码读取失败，正在尝试忽略乱码强制读取...")
|
||
df = pd.read_csv(file_path, encoding='gb18030', encoding_errors='replace')
|
||
except Exception as e:
|
||
print(f"读取文件彻底失败: {e}")
|
||
return
|
||
|
||
# --- 2. 确认厂家列名 ---
|
||
col_factory_general = '厂家'
|
||
col_factory_detail = '厂家.1'
|
||
|
||
if col_factory_detail not in df.columns:
|
||
print("警告: 未检测到第二个'厂家'列，明细表将被迫使用第一个'厂家'列。")
|
||
col_factory_detail = '厂家'
|
||
else:
|
||
print(f"厂家列识别成功：总表使用 '{col_factory_general}'，明细表使用 '{col_factory_detail}'")
|
||
|
||
# --- 3. 定义表头 ---
|
||
|
||
# 3.1 外贸/内贸 总表表头
|
||
columns_general = [
|
||
"合同编号", "签署公司", "外贸合同号", "收款情况", "合同签订日期",
|
||
"销售员", "最终用户单位", "最终用户信息联系人、电话、邮箱", "最终用户所在地",
|
||
"厂家", "型号/货号", "合同标的", "数量", "单位", "币种", "折扣率",
|
||
"合同", "总合同额", "外购", "已收款", "未收款", "收款日期",
|
||
"最晚发货期", "付款方式", "发货港", "目的港", "发货日期",
|
||
"买方单位", "买方信息联系人、电话、邮箱", "收货人信息"
|
||
]
|
||
columns_domestic_general = [c if c != "外贸合同号" else "内贸合同号" for c in columns_general]
|
||
|
||
# 3.2 明细表表头
|
||
columns_detail = [
|
||
"合同编号", "销售员", "厂家", "合同标的", "货号", "产品描述", "数量", "单位",
|
||
"币种", "报价单价", "报价总价", "销售单价", "销售总价", "折扣率",
|
||
"外购", "合同币种/美元", "外购转美元", "报价总价美元", "净合同额美元"
|
||
]
|
||
|
||
# 3.3 其他表表头
|
||
columns_other = [
|
||
"合同编号", "签署公司", "内贸合同号", "收款情况", "签订日期",
|
||
"销售员", "最终用户单位", "最终用户信息联系人、电话、邮箱", "最终用户所在地",
|
||
"买方单位", "买方信息联系人、电话、邮箱", "合同标的", "合同总额",
|
||
"已收款", "未收款", "收款日期"
|
||
]
|
||
|
||
# --- 4. 辅助函数：安全转数字 ---
|
||
def safe_float(val):
|
||
try:
|
||
if isinstance(val, str):
|
||
val = val.replace(',', '').strip()
|
||
if val == '': return 0.0
|
||
return float(val)
|
||
except (ValueError, TypeError):
|
||
return 0.0
|
||
|
||
# --- 5. 数据转换逻辑 ---
|
||
|
||
# 5.1 外贸/内贸 总表转换逻辑
|
||
def transform_general_row(row, trade_type):
|
||
target_cols = columns_general if trade_type == '外贸' else columns_domestic_general
|
||
new_row = {col: "" for col in target_cols}
|
||
|
||
# 拆分合同号
|
||
order_no_raw = str(row.get('合同订单编号', ''))
|
||
parts_no = order_no_raw.split(' ')
|
||
new_row['合同编号'] = parts_no[0] if len(parts_no) > 0 else order_no_raw
|
||
contract_no_col = '外贸合同号' if trade_type == '外贸' else '内贸合同号'
|
||
new_row[contract_no_col] = parts_no[1] if len(parts_no) > 1 else ""
|
||
|
||
# 拆分合同标的 (不再从这里取总价)
|
||
target_raw = str(row.get('合同标的（品名/型号/数量/单价/总价）', ''))
|
||
parts_target = target_raw.split('/')
|
||
if len(parts_target) >= 1: new_row['合同标的'] = parts_target[0]
|
||
if len(parts_target) >= 2: new_row['型号/货号'] = parts_target[1]
|
||
if len(parts_target) >= 3: new_row['数量'] = parts_target[2]
|
||
if len(parts_target) >= 4: new_row['合同'] = parts_target[3] # 单价
|
||
|
||
# 【修改点】总合同额：直接读取 CSV 中的“合同总额”列
|
||
new_row['总合同额'] = row.get('合同总额', '')
|
||
|
||
# 映射其他字段
|
||
new_row['签署公司'] = row.get('收款账户', '')
|
||
new_row['收款情况'] = row.get('收款状态', '')
|
||
new_row['合同签订日期'] = row.get('签约日期', '')
|
||
new_row['销售员'] = row.get('负责人', '')
|
||
new_row['最终用户单位'] = row.get('客户名称', '')
|
||
new_row['最终用户信息联系人、电话、邮箱'] = row.get('联系人姓名', '')
|
||
new_row['厂家'] = row.get(col_factory_general, '')
|
||
new_row['币种'] = row.get('货币(选完产品再改)', '')
|
||
new_row['外购'] = row.get('外购产品金额', '')
|
||
new_row['收款日期'] = row.get('最新收款日期', '')
|
||
new_row['最晚发货期'] = row.get('最晚发货期', '')
|
||
new_row['付款方式'] = row.get('付款比例及期限', '')
|
||
new_row['发货港'] = row.get('发货地', '')
|
||
new_row['目的港'] = row.get('目的港', '')
|
||
new_row['买方单位'] = row.get('合同买方（名称/联系人/电话/邮箱）', '')
|
||
|
||
return pd.Series(new_row)
|
||
|
||
# 5.2 明细表转换逻辑
|
||
def transform_detail_row(row):
|
||
new_row = {col: "" for col in columns_detail}
|
||
|
||
detail_manuf_val = str(row.get(col_factory_detail, ''))
|
||
order_no_raw = str(row.get('合同订单编号', ''))
|
||
new_row['合同编号'] = order_no_raw.split(' ')[0] if order_no_raw else ""
|
||
new_row['销售员'] = row.get('负责人', '')
|
||
new_row['厂家'] = detail_manuf_val
|
||
new_row['货号'] = row.get('产品编码', '')
|
||
new_row['数量'] = row.get('数量', '')
|
||
new_row['单位'] = ""
|
||
new_row['币种'] = row.get('原币种', '')
|
||
new_row['折扣率'] = ""
|
||
|
||
target_raw = str(row.get('合同标的（品名/型号/数量/单价/总价）', ''))
|
||
parts_target = target_raw.split('/')
|
||
new_row['合同标的'] = parts_target[0] if len(parts_target) >= 1 else ""
|
||
|
||
val_outsourcing_raw = safe_float(row.get('外购产品金额', 0))
|
||
val_rate = safe_float(row.get('汇率', 1))
|
||
if val_rate == 0: val_rate = 1
|
||
|
||
raw_price_unit = row.get('美元报价', '')
|
||
raw_price_total = row.get('产品小计', '')
|
||
|
||
if '外购' in detail_manuf_val:
|
||
new_row['外购'] = val_outsourcing_raw
|
||
new_row['产品描述'] = row.get('备注', '')
|
||
new_row['报价单价'] = ""
|
||
new_row['报价总价'] = ""
|
||
new_row['销售单价'] = ""
|
||
new_row['销售总价'] = ""
|
||
current_outsourcing_cost = val_outsourcing_raw
|
||
else:
|
||
new_row['外购'] = ""
|
||
new_row['产品描述'] = row.get('产品名称', '')
|
||
new_row['报价单价'] = raw_price_unit
|
||
new_row['报价总价'] = raw_price_total
|
||
new_row['销售单价'] = ""
|
||
new_row['销售总价'] = ""
|
||
current_outsourcing_cost = 0
|
||
|
||
new_row['合同币种/美元'] = ""
|
||
|
||
if current_outsourcing_cost > 0:
|
||
new_row['外购转美元'] = round(current_outsourcing_cost / val_rate, 2)
|
||
else:
|
||
new_row['外购转美元'] = ""
|
||
|
||
new_row['报价总价美元'] = ""
|
||
new_row['净合同额美元'] = ""
|
||
|
||
return pd.Series(new_row)
|
||
|
||
# 5.3 其他表转换逻辑
|
||
def transform_other_row(row):
|
||
new_row = {col: "" for col in columns_other}
|
||
|
||
# 拆分合同号
|
||
order_no_raw = str(row.get('合同订单编号', ''))
|
||
parts_no = order_no_raw.split(' ')
|
||
new_row['合同编号'] = parts_no[0] if len(parts_no) > 0 else order_no_raw
|
||
new_row['内贸合同号'] = parts_no[1] if len(parts_no) > 1 else ""
|
||
|
||
# 合同标的 (取第一部分)
|
||
target_raw = str(row.get('合同标的（品名/型号/数量/单价/总价）', ''))
|
||
parts_target = target_raw.split('/')
|
||
if len(parts_target) >= 1:
|
||
new_row['合同标的'] = parts_target[0]
|
||
|
||
# 【修改点】合同总额：直接读取源CSV的“合同总额”列
|
||
new_row['合同总额'] = row.get('合同总额', '')
|
||
|
||
# 映射其他字段
|
||
new_row['签署公司'] = row.get('收款账户', '')
|
||
new_row['收款情况'] = row.get('收款状态', '')
|
||
new_row['签订日期'] = row.get('签约日期', '')
|
||
new_row['销售员'] = row.get('负责人', '')
|
||
new_row['最终用户单位'] = row.get('客户名称', '')
|
||
new_row['最终用户信息联系人、电话、邮箱'] = row.get('联系人姓名', '')
|
||
new_row['买方单位'] = row.get('合同买方（名称/联系人/电话/邮箱）', '')
|
||
new_row['收款日期'] = row.get('最新收款日期', '')
|
||
|
||
return pd.Series(new_row)
|
||
|
||
# --- 6. 主处理流程 ---
|
||
|
||
df[col_factory_general] = df[col_factory_general].fillna('').astype(str)
|
||
df['合同类型'] = df['合同类型'].fillna('').astype(str)
|
||
|
||
# 文件拆分逻辑
|
||
df_asd = df[df[col_factory_general].str.contains('ASD', case=False, na=False)]
|
||
df_non_asd = df[~df[col_factory_general].str.contains('ASD', case=False, na=False)]
|
||
|
||
def create_excel(dataframe, filename):
|
||
raw_foreign = dataframe[dataframe['合同类型'] == '外贸'].copy()
|
||
raw_domestic = dataframe[dataframe['合同类型'] == '内贸'].copy()
|
||
raw_other = dataframe[~dataframe['合同类型'].isin(['外贸', '内贸'])].copy()
|
||
|
||
# === 1. 生成外贸数据 ===
|
||
if not raw_foreign.empty:
|
||
df_gen = raw_foreign.apply(lambda row: transform_general_row(row, '外贸'), axis=1)
|
||
df_gen = df_gen[columns_general]
|
||
df_gen_unique = df_gen.drop_duplicates(subset=['合同编号'], keep='first')
|
||
df_gen_unique = df_gen_unique.sort_values(by='合同编号', ascending=True)
|
||
|
||
df_det = raw_foreign.apply(lambda row: transform_detail_row(row), axis=1)
|
||
df_det = df_det[columns_detail]
|
||
df_det = df_det.sort_values(by='合同编号', ascending=True)
|
||
mask_duplicates = df_det.duplicated(subset=['合同编号'], keep='first')
|
||
df_det.loc[mask_duplicates, '合同标的'] = ""
|
||
|
||
else:
|
||
df_gen_unique = pd.DataFrame(columns=columns_general)
|
||
df_det = pd.DataFrame(columns=columns_detail)
|
||
|
||
# === 2. 生成内贸数据 ===
|
||
if not raw_domestic.empty:
|
||
df_dom_gen = raw_domestic.apply(lambda row: transform_general_row(row, '内贸'), axis=1)
|
||
df_dom_gen = df_dom_gen[columns_domestic_general]
|
||
df_dom_gen_unique = df_dom_gen.drop_duplicates(subset=['合同编号'], keep='first')
|
||
df_dom_gen_unique = df_dom_gen_unique.sort_values(by='合同编号', ascending=True)
|
||
|
||
df_dom_det = raw_domestic.apply(lambda row: transform_detail_row(row), axis=1)
|
||
df_dom_det = df_dom_det[columns_detail]
|
||
df_dom_det = df_dom_det.sort_values(by='合同编号', ascending=True)
|
||
mask_duplicates_dom = df_dom_det.duplicated(subset=['合同编号'], keep='first')
|
||
df_dom_det.loc[mask_duplicates_dom, '合同标的'] = ""
|
||
|
||
else:
|
||
df_dom_gen_unique = pd.DataFrame(columns=columns_domestic_general)
|
||
df_dom_det = pd.DataFrame(columns=columns_detail)
|
||
|
||
# === 3. 生成其他数据 ===
|
||
if not raw_other.empty:
|
||
df_other = raw_other.apply(lambda row: transform_other_row(row), axis=1)
|
||
df_other = df_other[columns_other]
|
||
# 去重
|
||
df_other_unique = df_other.drop_duplicates(subset=['合同编号'], keep='first')
|
||
# 排序
|
||
df_other_unique = df_other_unique.sort_values(by='合同编号', ascending=True)
|
||
else:
|
||
df_other_unique = pd.DataFrame(columns=columns_other)
|
||
|
||
# === 4. 写入 Excel ===
|
||
try:
|
||
print(f"[{filename}] 正在写入Excel...")
|
||
with pd.ExcelWriter(filename, engine='openpyxl') as writer:
|
||
df_gen_unique.to_excel(writer, sheet_name='外贸总表', index=False)
|
||
df_det.to_excel(writer, sheet_name='外贸明细', index=False)
|
||
df_dom_gen_unique.to_excel(writer, sheet_name='内贸总表', index=False)
|
||
df_dom_det.to_excel(writer, sheet_name='内贸明细', index=False)
|
||
df_other_unique.to_excel(writer, sheet_name='其他', index=False)
|
||
print(f"成功生成文件: {filename}")
|
||
except Exception as e:
|
||
print(f"生成 {filename} 时发生错误: {e}")
|
||
|
||
# 执行生成
|
||
print("-" * 40)
|
||
create_excel(df_asd, 'ASD.xlsx')
|
||
print("-" * 40)
|
||
create_excel(df_non_asd, '非ASD.xlsx')
|
||
print("-" * 40)
|
||
print("全部处理完成！")
|
||
|
||
|
||
# --- 运行入口 ---
|
||
if __name__ == "__main__":
|
||
csv_file = 'test.csv'
|
||
if os.path.exists(csv_file):
|
||
process_contracts(csv_file)
|
||
else:
|
||
print(f"找不到文件: {csv_file}，请检查路径。") |