多家银行流水提取自动化(附源码)

一键提取 多家银行流水并合并为统一格式 Excel 的 Python 工具。

📌 一、项目背景

财务每月需要从多家银行网银导出账户流水（.xls / .xlsx），但这些文件格式各不相同：有的账号藏在表头单元格，有的账号就在数据列里；有的用"贷方/借方"两列表示收支，有的只有"借贷标志+发生额"；中行甚至把人民币、美元两个账户塞进同一个文件。

人工打开一个一个复制、对齐列、补空值、转文本格式，既慢又容易错。

本项目把"读取 → 清洗 → 统一 → 合并"全流程自动化，输出一张标准 8 列流水总表，可直接用于会计核对、银企对账或导入财务系统。

🏦 二、涉及的银行与命名规则

工具按文件名关键词自动识别银行，无需配置。当前支持 8 家：

命名规则（重要）

• • 文件名只需包含对应关键词即可自动识别，可自由加城市前缀、账号后缀、序号：
• • ✅ XX中行XX.xls、XX招行XX.xlsx、XX建行XX.xls、XX江苏XX.xls
• • ✅ XX招行XX.xlsx、XX浦发XX.xls、XX工行XX.xlsx
• • 识别顺序：先判 中行 → 再判 江苏 → 最后遍历其余关键词。
• • 新增银行只需在 BANK_CONFIG 加一条配置，文件名带该关键词即可。

📋 三、可以提取的内容

所有银行统一输出 8 列，顺序固定：

统一业务规则

• • 对方名称为空 → 一般如结息、扣费等填入 当前公司名称
• • 对方账号为空 → 填入当前行银行账号
• • 江苏银行自动删除最后一行的 ----- 的分隔尾行
• • 中行一个文件含多个账户（人民币 / 美元），逐账户块提取，来账取付款人、往账取收款人为对方

🎨 四、设计实现与样式结果

架构：配置驱动 + 通用流水线

BANK_CONFIG（各银行配置）        │        ▼process_generic（通用清洗：读表→重命名→算金额→清日期→填空值→文本化）process_boc（中行专属：多账户块动态切片）        │        ▼sort_result（按 银行名称→银行账号→交易日期 排序）        │        ▼to_excel + openpyxl（强制账号列为文本格式 @）

输出样式结果

• • 合并为单个 extracted_bank_data.xlsx，无多余 sheet。
• • 排序结果：招商银行 → 江苏银行 → 中国银行 → 中信银行 → 建设银行 → 工商银行 → 兴业银行 → 浦发银行（改顺序只改 BANK_ORDER 一行常量）。
• • 账号列在 Excel 中显示为文本（左上角绿色小三角，不丢失前导 0、不变 3.2E+19）。
• • 日期列全部 2026/06/21 样式，可直接按时间排序筛选。

示例（账号等脱敏展示）：

🚀 五、如何使用

1. 准备环境

pip install pandas openpyxl xlrd

2. 放入文件

把所有银行流水 .xls / .xlsx 放进脚本同级的 银行流水/ 文件夹，文件名按上面的命名规则命名。

3. 调整

• • 改银行排序：编辑文件顶部 BANK_ORDER 列表顺序。
• • 加新银行：在 BANK_CONFIG 增加一条（含 skiprows、rename、account 等）。
• • 改公司名/空值填充规则：编辑顶部 COMPANY 常量或 fill_counterparty()。

4. 运行

python 银行流水.py

脚本会扫描文件夹、逐个识别并处理，最后打印 处理完成! 共 N 条记录，结果写入 银行流水/extracted_bank_data.xlsx。

💻 六、完整代码（银行流水.py）

"""银行流水提取脚本（配置驱动重构版）读取各银行流水 Excel，统一清洗为 8 列并合并输出。输出列顺序：银行账号 | 交易日期 | 交易金额 | 对方名称 | 摘要 | 对方账号 | 余额 | 银行名称规则：  - 交易日期统一 YYYY/MM/DD  - 银行账号、对方账号强制文本  - 对方名称为空 -> 当前公司名称  - 对方账号为空 -> 当前行银行账号  - 江苏银行删除最后一行分隔符(-----)  - 中行一个文件含多个账户块（人民币/美元），逐块提取"""import osimport reimport warningsimport pandas as pdimport numpy as npwarnings.filterwarnings("ignore")COMPANY = "当前公司名称"BASE = os.path.dirname(os.path.abspath(__file__))INPUT_FOLDER = os.path.join(BASE, "银行流水")OUTPUT_FILE = os.path.join(INPUT_FOLDER, "extracted_bank_data.xlsx")# 最终输出列OUT_COLS = ["银行账号", "交易日期", "交易金额", "对方名称", "摘要", "对方账号", "余额", "银行名称"]# 银行名称排序顺序（改顺序只改这里）BANK_ORDER = ["招商银行", "江苏银行", "中国银行", "中信银行","建设银行", "工商银行", "兴业银行", "浦发银行"]# -------------------------------------------------------------------# 通用工具# -------------------------------------------------------------------defclean_amount(x):"""清洗金额为 float（去逗号/空格/货币符号）。空返回 0。"""if pd.isna(x):return0.0    s = str(x).strip()if s in ("", "nan", "None", "-", "—"):return0.0    s = re.sub(r"[^\d.\-]", "", s)try:returnfloat(s) if s notin ("", "-") else0.0except ValueError:return0.0deffmt_date(x):"""统一返回 YYYY/MM/DD 字符串，无法解析返回空串。"""if pd.isna(x):return""    s = str(x).strip()if s in ("", "nan"):return""# YYYYMMDDif re.fullmatch(r"\d{8}", s):try:returnf"{s[:4]}/{s[4:6]}/{s[6:8]}"except Exception:return""# 去掉时间部分    s2 = s.split()[0] if" "in s else sfor fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d"):try:return pd.to_datetime(s2, format=fmt).strftime("%Y/%m/%d")except Exception:pass    d = pd.to_datetime(s2, errors="coerce")return d.strftime("%Y/%m/%d") ifnot pd.isna(d) else""def_norm_text(s):"""NaN/空/'nan' 统一为空串，并去除首尾空白。"""    s = s.fillna("").astype(str).str.strip()return s.mask(s.isin(["nan", "None", "NaT"]), "")deffill_counterparty(df):"""填充空值：对方名称->公司名；对方账号->银行账号。账号列转文本。"""if"对方名称"notin df.columns:        df["对方名称"] = ""if"对方账号"notin df.columns:        df["对方账号"] = ""    df["对方名称"] = _norm_text(df["对方名称"])    df["对方账号"] = _norm_text(df["对方账号"])# 对方名称空 -> 公司名    df.loc[df["对方名称"] == "", "对方名称"] = COMPANY# 对方账号空 -> 本行银行账号    empty = df["对方账号"] == ""    df.loc[empty, "对方账号"] = _norm_text(df.loc[empty, "银行账号"])# 账号列强制文本    df["银行账号"] = _norm_text(df["银行账号"])    df["对方账号"] = _norm_text(df["对方账号"])return df# -------------------------------------------------------------------# 银行配置表#   account: 'col'(用银行账号列) | 'fixed'(按文件名取) | 'cell'(读表头单元格) | 'col_or_cell'#   amount : 默认 贷方金额 - 借方金额；江苏用 sign_col 借贷标记# -------------------------------------------------------------------BANK_CONFIG = {"招行": dict(        skiprows=12, account="col",        rename={"交易日": "交易日期", "贷方金额": "贷方金额", "借方金额": "借方金额","摘要": "摘要", "收(付)方名称": "对方名称", "收(付)方账号": "对方账号","余额": "余额", "账号": "银行账号",        },        bank_name="招商银行",    ),"江苏银行": dict(        skiprows=2, account="cell", account_cell=(0, 1), sign_col="借贷标记",        rename={"交易日期": "交易日期", "借贷标记": "借贷标记", "交易金额": "交易金额","账户余额": "余额", "摘要代码": "摘要", "对方账号": "对方账号","对方户名": "对方名称",        },        bank_name="江苏银行", drop_sep=True,    ),"建行": dict(        skiprows=9, account="cell", account_cell=(3, 1),        rename={"交易时间": "交易日期", "借方发生额/元(支取)": "借方金额","贷方发生额/元(收入)": "贷方金额", "对方户名": "对方名称","对方账号": "对方账号", "余额": "余额", "摘要": "摘要",        },        bank_name="建设银行",    ),"浦发": dict(        skiprows=7, skiprows_map={"苏州浦发": 4}, account="col_or_cell", account_cell=(0, 1),        rename={"专户账号": "银行账号", "收款金额": "贷方金额", "付款金额": "借方金额","发生后余额": "余额", "交易对手户名": "对方名称", "交易对手账号": "对方账号","借方金额": "借方金额", "贷方金额": "贷方金额", "余额": "余额","对方账号": "对方账号", "对方户名": "对方名称", "摘要": "摘要","交易日期": "交易日期",        },        bank_name="浦发银行",    ),"工行": dict(        skiprows=1, account="col", sign_col="借贷标志",        rename={"交易时间": "交易日期", "发生额": "交易金额","余额": "余额", "本方账号": "银行账号", "对方单位": "对方名称","对方账号": "对方账号", "摘要": "摘要",        },        bank_name="工商银行",    ),"中信": dict(        skiprows=15, account="col",        rename={"交易日期": "交易日期", "借方发生额": "借方金额", "贷方发生额": "贷方金额","账户余额": "余额", "对方账号": "对方账号", "对方账户名称": "对方名称","交易账号": "银行账号", "摘要": "摘要",        },        bank_name="中信银行",    ),"兴业": dict(        skiprows=1, account="col",        rename={"记账日期": "交易日期", "借方金额(支出)": "借方金额", "贷方金额(收入)": "贷方金额","账户余额": "余额", "摘要": "摘要", "对方账号": "对方账号","对方户名": "对方名称", "账号": "银行账号",        },        bank_name="兴业银行",    ),}# -------------------------------------------------------------------# 通用处理# -------------------------------------------------------------------defget_account(df, cfg, file_path, sheet_name):if"银行账号"in df.columns and df["银行账号"].notna().any():return df["银行账号"].astype(str).str.strip()    mode = cfg.get("account")if mode == "fixed":for k, v in cfg["accounts"].items():if k in os.path.basename(file_path):return pd.Series([str(v)] * len(df))return pd.Series([str(list(cfg["accounts"].values())[0])] * len(df))if mode in ("cell", "col_or_cell"):        r, c = cfg["account_cell"]        raw = pd.read_excel(file_path, sheet_name=sheet_name, header=None,                            nrows=r + 1, dtype=str)        val = raw.iloc[r, c]return pd.Series([str(val).strip()] * len(df))return pd.Series([""] * len(df))defprocess_generic(file_path, sheet_name, cfg):try:        skiprows = cfg["skiprows"]if"skiprows_map"in cfg:for sub, sr in cfg["skiprows_map"].items():if sub in os.path.basename(file_path):                    skiprows = srbreak        df = pd.read_excel(file_path, sheet_name=sheet_name, skiprows=skiprows,                          dtype=str)if df.empty:returnNone# 去除表头首尾空白，避免 rename 匹配失败        df.columns = df.columns.astype(str).str.strip()# 删除分隔尾行（江苏银行）if cfg.get("drop_sep"):            first = df.iloc[:, 0].astype(str).str.strip()            mask = ~first.str.match(r"^[\-—_=]+$")            df = df[mask].copy()# 删除指定列（如工行空摘要列，避免与用途重名）for c in cfg.get("drop_cols", []):if c in df.columns:                df = df.drop(columns=[c])        df = df.rename(columns=cfg["rename"])# 银行账号        df["银行账号"] = get_account(df, cfg, file_path, sheet_name)# 金额if cfg.get("sign_col"):            sign = np.where(df[cfg["sign_col"]].astype(str).str.strip() == "借", -1, 1)            df["交易金额"] = df["交易金额"].apply(clean_amount) * signelse:            credit = df["贷方金额"].apply(clean_amount) if"贷方金额"in df else0.0            debit = df["借方金额"].apply(clean_amount) if"借方金额"in df else0.0            df["交易金额"] = credit - debitif"余额"in df.columns:            df["余额"] = df["余额"].apply(clean_amount)# 日期if"交易日期"in df.columns:            df["交易日期"] = df["交易日期"].apply(fmt_date)# 摘要兜底if"摘要"notin df.columns:            df["摘要"] = ""# 删除无效行（日期为空）        df = df[df["交易日期"].astype(str).str.strip() != ""]        df = fill_counterparty(df)        df["银行名称"] = cfg["bank_name"]        result = df[[c for c in OUT_COLS if c in df.columns]].copy()        result["银行名称"] = cfg["bank_name"]return resultexcept Exception as e:print(f"  [处理失败] {os.path.basename(file_path)}: {e}")returnNone# -------------------------------------------------------------------# 中行：一个文件多账户块# -------------------------------------------------------------------defprocess_boc(file_path, sheet_name):try:        df = pd.read_excel(file_path, sheet_name=0, header=None, dtype=str)        account_rows = df.index[df.iloc[:, 0].astype(str).str.contains("查询账号", na=False)].tolist()ifnot account_rows:returnNone        records = []for ai, arow inenumerate(account_rows):            account = str(df.iloc[arow, 1]).strip()            hidx = Nonefor i inrange(arow + 1, len(df)):if"交易类型"instr(df.iloc[i, 0]):                    hidx = ibreakif hidx isNone:continue            next_a = account_rows[ai + 1] if ai + 1 < len(account_rows) elselen(df)            header = [str(c) for c in df.iloc[hidx].tolist()]            col = lambda kw: next((j for j, h inenumerate(header) if kw in h), None)            c_date, c_amt, c_bal = col("交易日期"), col("交易金额"), col("交易后余额")            c_pn, c_pa = col("付款人名称"), col("付款人账号")            c_rn, c_ra = col("收款人名称"), col("收款人账号")            c_type, c_ref, c_use = col("交易类型"), col("摘要"), col("用途")for i inrange(hidx + 1, next_a):                row = df.iloc[i]                date_val = row[c_date] if c_date isnotNoneelseNoneif pd.isna(date_val) orstr(date_val).strip() in ("", "nan"):continue                direction = str(row[c_type]).strip() if c_type isnotNoneelse""# 来账：对手为付款人；往账：对手为收款人if"来"in direction:                    oname, oacct = row[c_pn], row[c_pa]else:                    oname, oacct = row[c_rn], row[c_ra]                oname = ""if pd.isna(oname) elsestr(oname).strip()                oacct = ""if pd.isna(oacct) elsestr(oacct).strip()                ref = ""if pd.isna(row[c_ref]) elsestr(row[c_ref]).strip()ifnot ref and c_use isnotNoneandnot pd.isna(row[c_use]):                    ref = str(row[c_use]).strip()                records.append({"银行账号": account,"交易日期": fmt_date(date_val),"交易金额": clean_amount(row[c_amt]) if c_amt isnotNoneelse0,"对方名称": oname,"摘要": ref,"对方账号": oacct,"余额": clean_amount(row[c_bal]) if c_bal isnotNoneelse0,"银行名称": "中国银行",                })ifnot records:returnNone        out = pd.DataFrame(records)        out = fill_counterparty(out)return out[OUT_COLS]except Exception as e:print(f"  [处理失败] {os.path.basename(file_path)}: {e}")returnNone# -------------------------------------------------------------------# 排序：银行名称(自定义顺序) -> 银行账号 -> 交易日期# 改顺序只需修改上方的 BANK_ORDER 常量# -------------------------------------------------------------------defsort_result(df):    order_map = {name: i for i, name inenumerate(BANK_ORDER)}    df = df.copy()    df["__bank_order__"] = df["银行名称"].map(lambda x: order_map.get(x, len(BANK_ORDER)))    df["__date__"] = pd.to_datetime(df["交易日期"], errors="coerce")    df = df.sort_values(        by=["__bank_order__", "银行账号", "__date__"],        kind="stable",    ).drop(columns=["__bank_order__", "__date__"]).reset_index(drop=True)return df# -------------------------------------------------------------------# 主流程# -------------------------------------------------------------------defresolve_handler(filename):if"中行"in filename:return"boc"if"江苏"in filename:return"江苏银行"for key in BANK_CONFIG:if key in filename:return keyreturnNonedefprocess_bank_files():    all_data = []for filename insorted(os.listdir(INPUT_FOLDER)):ifnot filename.lower().endswith((".xls", ".xlsx")):continueif filename.startswith("~$") or"extracted"in filename:continue        file_path = os.path.join(INPUT_FOLDER, filename)        handler = resolve_handler(filename)if handler isNone:print(f"  跳过未识别银行: {filename}")continueprint(f"  处理: {filename}")if handler == "boc":            df = process_boc(file_path, 0)else:            df = process_generic(file_path, 0, BANK_CONFIG[handler])if df isnotNoneandnot df.empty:            all_data.append(df)print(f"    -> {len(df)} 条")else:print("    -> 无有效数据")ifnot all_data:print("未找到有效数据")return    result = pd.concat(all_data, ignore_index=True)# 按 银行名称(自定义顺序) -> 银行账号 -> 交易日期 排序    result = sort_result(result)# 数值列兜底for c in ("交易金额", "余额"):        result[c] = pd.to_numeric(result[c], errors="coerce").fillna(0)# 写出并强制账号列为文本    result.to_excel(OUTPUT_FILE, index=False, engine="openpyxl")from openpyxl import load_workbook    wb = load_workbook(OUTPUT_FILE)    ws = wb.active    header = [cell.value for cell in ws[1]]for col_name in ("银行账号", "对方账号"):if col_name in header:            idx = header.index(col_name) + 1for cell in ws.iter_cols(min_col=idx, max_col=idx, min_row=2):for c in cell:                    c.number_format = "@"    wb.save(OUTPUT_FILE)print(f"\n处理完成! 共 {len(result)} 条记录 -> {OUTPUT_FILE}")if __name__ == "__main__":    process_bank_files()

💡 小结：把每月的流水文件丢进文件夹，跑一次脚本，30 分钟的工作量变成 10 秒钟。