474 lines
18 KiB
Python
474 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
家庭收支表合并脚本
|
||
读取两张招行储蓄卡流水CSV，过滤内部转账、理财调拨等，生成家庭收支明细和月度汇总。
|
||
"""
|
||
|
||
import csv
|
||
import os
|
||
import sys
|
||
from collections import defaultdict
|
||
from datetime import datetime, timedelta
|
||
|
||
# 修复Windows控制台编码
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
DATA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
|
||
|
||
FILE_1207 = os.path.join(DATA_DIR, "CMB_6214--------1207_20250726_20260726.csv")
|
||
FILE_1617 = os.path.join(DATA_DIR, "CMB_6214--------1617_20250726_20260726.csv")
|
||
|
||
OUT_DETAIL = os.path.join(DATA_DIR, "家庭收支明细.csv")
|
||
OUT_SUMMARY = os.path.join(DATA_DIR, "家庭月度汇总.csv")
|
||
OUT_LOG = os.path.join(DATA_DIR, "处理日志.txt")
|
||
|
||
|
||
def parse_cmb_csv(filepath, account):
|
||
"""解析招行CSV，返回交易列表。"""
|
||
transactions = []
|
||
with open(filepath, "r", encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
|
||
# 跳过前7行注释，从第8行(index=7)开始是表头+数据
|
||
reader = csv.reader(lines[7:])
|
||
header = next(reader) # 表头行
|
||
|
||
for fields in reader:
|
||
# 跳过空行或字段不足的行
|
||
if not fields or len(fields) < 7:
|
||
continue
|
||
|
||
date_str = fields[0].strip()
|
||
time_str = fields[1].strip()
|
||
income = fields[2].strip()
|
||
expense = fields[3].strip()
|
||
balance = fields[4].strip()
|
||
tx_type = fields[5].strip()
|
||
remark = fields[6].strip()
|
||
|
||
# 跳过空行（只有空白tab）
|
||
if not date_str:
|
||
continue
|
||
|
||
# 解析金额
|
||
amount = 0.0
|
||
direction = ""
|
||
if income and float(income) > 0:
|
||
amount = float(income)
|
||
direction = "收入"
|
||
elif expense and float(expense) > 0:
|
||
amount = float(expense)
|
||
direction = "支出"
|
||
else:
|
||
continue
|
||
|
||
# 解析日期
|
||
try:
|
||
tx_date = datetime.strptime(date_str, "%Y%m%d")
|
||
except ValueError:
|
||
continue
|
||
|
||
tx = {
|
||
"date": date_str,
|
||
"date_obj": tx_date,
|
||
"time": time_str,
|
||
"account": account,
|
||
"direction": direction,
|
||
"amount": amount,
|
||
"balance": balance,
|
||
"tx_type": tx_type,
|
||
"remark": remark,
|
||
"category": "",
|
||
"excluded": False,
|
||
"exclude_reason": "",
|
||
}
|
||
transactions.append(tx)
|
||
|
||
return transactions
|
||
|
||
|
||
def classify_transaction(tx):
|
||
"""
|
||
对单笔交易进行分类，返回 (category, excluded, exclude_reason)。
|
||
分类按优先级匹配。
|
||
"""
|
||
tx_type = tx["tx_type"]
|
||
remark = tx["remark"]
|
||
account = tx["account"]
|
||
direction = tx["direction"]
|
||
|
||
# === 1. 理财投资类（排除） ===
|
||
investment_keywords = ["理财申购", "理财赎回", "理财还本", "朝朝宝", "朝朝盈", "快速赎回", "银证转账", "银转证", "证转银", "基金赎回", "基金申购", "基金还本", "个人养老金缴存"]
|
||
for kw in investment_keywords:
|
||
if kw in tx_type:
|
||
return "理财投资", True, f"理财调拨({tx_type})"
|
||
|
||
# 银联代付-上海天天基金 = 基金赎回款，排除
|
||
if tx_type == "银联代付" and "天天基金" in remark:
|
||
return "理财投资", True, f"基金赎回款({remark})"
|
||
|
||
# 朝朝宝赎回在1617卡非常多，也需要排除
|
||
if "赎回" in tx_type and ("朝朝" in tx_type or "理财" in tx_type):
|
||
return "理财投资", True, f"理财调拨({tx_type})"
|
||
|
||
# === 2. 夫妻间内部转账（排除） ===
|
||
# 1207 → 1617 的各种转账
|
||
if account == "丈夫-1207":
|
||
if "张青" in remark:
|
||
return "内部转账", True, f"夫妻间转账({tx_type}: {remark})"
|
||
# 1207卡的微信转账（全部是给媳妇儿的）
|
||
if "微信转账" in remark and direction == "支出":
|
||
return "内部转账", True, f"夫妻间微信转账({remark})"
|
||
# 1207卡的微信转账退款
|
||
if "微信转账" in remark and direction == "收入" and tx_type == "网联退款":
|
||
return "内部转账", True, f"夫妻间微信转账退款({remark})"
|
||
|
||
# 1617卡：刘鑫(妻子)微信零钱提现回卡，属内部转账
|
||
if account == "妻子-1617" and direction == "收入" and "刘鑫" in remark:
|
||
return "内部转账", True, f"微信零钱提现({remark})"
|
||
|
||
# === 3. 信用卡还款（保留，但单独分类） ===
|
||
if "信用卡还款" in remark or "信用卡自动还款" in tx_type or "信用卡自扣" in tx_type:
|
||
return "信用卡还款", False, ""
|
||
|
||
# === 4. 退款（待配对） ===
|
||
# 携程部分退款——用户确认排除
|
||
if "退款" in tx_type and "携程" in remark:
|
||
return "退款", True, f"携程部分退款排除({remark})"
|
||
if "退款" in tx_type:
|
||
return "退款", False, "" # 暂时保留，后续配对处理
|
||
|
||
# === 5. 对外转账（1617卡微信转出，保留） ===
|
||
if account == "妻子-1617" and "微信转账" in remark and direction == "支出":
|
||
return "对外转账", False, ""
|
||
|
||
# === 6. 收入分类 ===
|
||
if direction == "收入":
|
||
# 汇入汇款-芒果光线：用户确认为之前支付宝消费退款，排除
|
||
if "汇入汇款" in tx_type and "芒果光线" in remark:
|
||
return "退款", True, f"支付宝退款-原消费退回({remark})"
|
||
# 银联代付（非理财类）单独分类
|
||
if tx_type == "银联代付":
|
||
return "银联代付收入", False, ""
|
||
# 股权结汇收益单独分类
|
||
if "结汇" in tx_type or "结汇" in remark or "股权" in remark:
|
||
return "股权结汇收益", False, ""
|
||
if tx_type == "工资" or "代发工资" in tx_type or "薪金" in remark or "EBPR" in remark or "工资" in remark or "代发" in remark:
|
||
return "工资收入", False, ""
|
||
if "保险" in remark or "理赔" in remark or "平安" in remark:
|
||
return "保险理赔", False, ""
|
||
if "公积金" in tx_type or "公积金" in remark:
|
||
return "公积金提取", False, ""
|
||
return "其他收入", False, ""
|
||
|
||
# === 7. 孩子零花钱 ===
|
||
if "支付宝-特约商户" in remark and tx["amount"] == 1.00:
|
||
return "孩子零花钱", False, ""
|
||
|
||
# === 8. 微信红包 ===
|
||
if "微信红包" in remark:
|
||
return "微信红包", False, ""
|
||
|
||
# === 9. 默认：日常消费 ===
|
||
return "日常消费", False, ""
|
||
|
||
|
||
def match_refunds(transactions):
|
||
"""
|
||
退款配对：对于退款收入，在同账户或跨账户的消费中查找同金额、相近日期(±3天)、同商户的交易。
|
||
配对成功则双双排除。
|
||
返回排除的交易集合。
|
||
"""
|
||
refunds = [tx for tx in transactions if tx["category"] == "退款" and not tx["excluded"]]
|
||
consumptions = [tx for tx in transactions if tx["direction"] == "支出" and not tx["excluded"]]
|
||
|
||
# 按金额建立消费索引
|
||
amount_index = defaultdict(list)
|
||
for i, tx in enumerate(consumptions):
|
||
amount_index[round(tx["amount"], 2)].append(i)
|
||
|
||
paired_consumption_ids = set()
|
||
paired_refund_ids = set()
|
||
|
||
for refund in refunds:
|
||
r_amount = round(refund["amount"], 2)
|
||
r_date = refund["date_obj"]
|
||
# 从备注中提取商户名
|
||
r_merchant = refund["remark"]
|
||
|
||
candidates = amount_index.get(r_amount, [])
|
||
found = False
|
||
for ci in candidates:
|
||
tx = consumptions[ci]
|
||
if id(tx) in paired_consumption_ids:
|
||
continue
|
||
# 日期在±3天内
|
||
if abs((tx["date_obj"] - r_date).days) <= 3:
|
||
# 商户名匹配（备注完全相同或包含关系）
|
||
if tx["remark"] == r_merchant or r_merchant in tx["remark"] or tx["remark"] in r_merchant:
|
||
paired_consumption_ids.add(id(tx))
|
||
paired_refund_ids.add(id(refund))
|
||
tx["excluded"] = True
|
||
tx["exclude_reason"] = f"退款配对(退款日期{refund['date']})"
|
||
refund["excluded"] = True
|
||
refund["exclude_reason"] = f"退款配对(消费日期{tx['date']})"
|
||
found = True
|
||
break
|
||
|
||
# 如果精确商户没匹配到，尝试同金额同日期范围（不同商户）
|
||
if not found:
|
||
for ci in candidates:
|
||
tx = consumptions[ci]
|
||
if id(tx) in paired_consumption_ids:
|
||
continue
|
||
if abs((tx["date_obj"] - r_date).days) <= 1:
|
||
paired_consumption_ids.add(id(tx))
|
||
paired_refund_ids.add(id(refund))
|
||
tx["excluded"] = True
|
||
tx["exclude_reason"] = f"退款配对-近似(退款日期{refund['date']})"
|
||
refund["excluded"] = True
|
||
refund["exclude_reason"] = f"退款配对-近似(消费日期{tx['date']})"
|
||
break
|
||
|
||
return len(paired_refund_ids)
|
||
|
||
|
||
def generate_detail_csv(transactions, output_path):
|
||
"""生成明细CSV。"""
|
||
kept = [tx for tx in transactions if not tx["excluded"]]
|
||
kept.sort(key=lambda x: (x["date_obj"], x["time"]))
|
||
|
||
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
|
||
writer = csv.writer(f)
|
||
writer.writerow(["日期", "时间", "账户", "收支方向", "金额", "交易类型", "交易备注", "分类"])
|
||
for tx in kept:
|
||
writer.writerow([
|
||
tx["date"],
|
||
tx["time"],
|
||
tx["account"],
|
||
tx["direction"],
|
||
f"{tx['amount']:.2f}",
|
||
tx["tx_type"],
|
||
tx["remark"],
|
||
tx["category"],
|
||
])
|
||
|
||
return len(kept)
|
||
|
||
|
||
def generate_summary_csv(transactions, output_path):
|
||
"""生成月度汇总CSV。"""
|
||
kept = [tx for tx in transactions if not tx["excluded"]]
|
||
|
||
# 按月统计
|
||
monthly = defaultdict(lambda: {
|
||
"income": 0.0, "expense": 0.0,
|
||
"categories": defaultdict(lambda: {"income": 0.0, "expense": 0.0})
|
||
})
|
||
|
||
for tx in kept:
|
||
month = tx["date"][:6] # YYYYMM
|
||
m = f"{month[:4]}-{month[4:6]}"
|
||
if tx["direction"] == "收入":
|
||
monthly[m]["income"] += tx["amount"]
|
||
monthly[m]["categories"][tx["category"]]["income"] += tx["amount"]
|
||
else:
|
||
monthly[m]["expense"] += tx["amount"]
|
||
monthly[m]["categories"][tx["category"]]["expense"] += tx["amount"]
|
||
|
||
# 收集所有出现过的分类
|
||
all_income_cats = set()
|
||
all_expense_cats = set()
|
||
for m_data in monthly.values():
|
||
for cat, vals in m_data["categories"].items():
|
||
if vals["income"] > 0:
|
||
all_income_cats.add(cat)
|
||
if vals["expense"] > 0:
|
||
all_expense_cats.add(cat)
|
||
|
||
all_income_cats = sorted(all_income_cats)
|
||
all_expense_cats = sorted(all_expense_cats)
|
||
|
||
with open(output_path, "w", encoding="utf-8-sig", newline="") as f:
|
||
writer = csv.writer(f)
|
||
|
||
# 表头
|
||
header = ["月份", "总收入", "总支出"]
|
||
for cat in all_income_cats:
|
||
header.append(f"收入-{cat}")
|
||
for cat in all_expense_cats:
|
||
header.append(f"支出-{cat}")
|
||
header.append("净结余")
|
||
writer.writerow(header)
|
||
|
||
for m in sorted(monthly.keys()):
|
||
data = monthly[m]
|
||
row = [m, f"{data['income']:.2f}", f"{data['expense']:.2f}"]
|
||
for cat in all_income_cats:
|
||
row.append(f"{data['categories'][cat]['income']:.2f}")
|
||
for cat in all_expense_cats:
|
||
row.append(f"{data['categories'][cat]['expense']:.2f}")
|
||
row.append(f"{data['income'] - data['expense']:.2f}")
|
||
writer.writerow(row)
|
||
|
||
# 合计行
|
||
total_income = sum(d["income"] for d in monthly.values())
|
||
total_expense = sum(d["expense"] for d in monthly.values())
|
||
total_row = ["合计", f"{total_income:.2f}", f"{total_expense:.2f}"]
|
||
for cat in all_income_cats:
|
||
total_row.append(f"{sum(d['categories'][cat]['income'] for d in monthly.values()):.2f}")
|
||
for cat in all_expense_cats:
|
||
total_row.append(f"{sum(d['categories'][cat]['expense'] for d in monthly.values()):.2f}")
|
||
total_row.append(f"{total_income - total_expense:.2f}")
|
||
writer.writerow(total_row)
|
||
|
||
return len(monthly)
|
||
|
||
|
||
def generate_log(transactions, excluded_stats, output_path):
|
||
"""生成处理日志。"""
|
||
lines = []
|
||
lines.append("=" * 70)
|
||
lines.append("家庭收支表处理日志")
|
||
lines.append("=" * 70)
|
||
lines.append("")
|
||
|
||
# 概况
|
||
total_1207 = sum(1 for tx in transactions if tx["account"] == "丈夫-1207")
|
||
total_1617 = sum(1 for tx in transactions if tx["account"] == "妻子-1617")
|
||
lines.append(f"原始交易总数：{len(transactions)} 笔")
|
||
lines.append(f" 丈夫-1207卡：{total_1207} 笔")
|
||
lines.append(f" 妻子-1617卡：{total_1617} 笔")
|
||
lines.append("")
|
||
|
||
# 排除统计
|
||
excluded = [tx for tx in transactions if tx["excluded"]]
|
||
lines.append(f"排除交易总数：{len(excluded)} 笔")
|
||
lines.append("-" * 50)
|
||
|
||
for cat, stats in sorted(excluded_stats.items()):
|
||
lines.append(f" {cat}: {stats['count']} 笔, 金额 ¥{stats['amount']:.2f}")
|
||
|
||
lines.append("")
|
||
|
||
# 保留统计
|
||
kept = [tx for tx in transactions if not tx["excluded"]]
|
||
lines.append(f"保留交易总数：{len(kept)} 笔")
|
||
lines.append("-" * 50)
|
||
|
||
kept_stats = defaultdict(lambda: {"count": 0, "income": 0.0, "expense": 0.0})
|
||
for tx in kept:
|
||
kept_stats[tx["category"]]["count"] += 1
|
||
if tx["direction"] == "收入":
|
||
kept_stats[tx["category"]]["income"] += tx["amount"]
|
||
else:
|
||
kept_stats[tx["category"]]["expense"] += tx["amount"]
|
||
|
||
total_income = sum(s["income"] for s in kept_stats.values())
|
||
total_expense = sum(s["expense"] for s in kept_stats.values())
|
||
for cat, stats in sorted(kept_stats.items()):
|
||
lines.append(f" {cat}: {stats['count']} 笔, 收入 ¥{stats['income']:.2f}, 支出 ¥{stats['expense']:.2f}")
|
||
|
||
lines.append("")
|
||
lines.append(f" 家庭总收入：¥{total_income:.2f}")
|
||
lines.append(f" 家庭总支出：¥{total_expense:.2f}")
|
||
lines.append(f" 家庭净结余：¥{total_income - total_expense:.2f}")
|
||
lines.append("")
|
||
|
||
# 列出排除的明细（按类别分组）
|
||
lines.append("=" * 70)
|
||
lines.append("排除交易明细")
|
||
lines.append("=" * 70)
|
||
|
||
by_cat = defaultdict(list)
|
||
for tx in excluded:
|
||
by_cat[tx["category"]].append(tx)
|
||
|
||
for cat in sorted(by_cat.keys()):
|
||
lines.append(f"\n【{cat}】({len(by_cat[cat])} 笔)")
|
||
for tx in sorted(by_cat[cat], key=lambda x: x["date_obj"]):
|
||
lines.append(f" {tx['date']} {tx['time']} | {tx['account']} | {tx['direction']} ¥{tx['amount']:.2f} | {tx['tx_type']} | {tx['remark']} | 原因: {tx['exclude_reason']}")
|
||
|
||
# 列出保留的大额交易（>5000元），供用户核查
|
||
lines.append("")
|
||
lines.append("=" * 70)
|
||
lines.append("保留的大额交易（>5000元）供核查")
|
||
lines.append("=" * 70)
|
||
|
||
big_txs = sorted([tx for tx in kept if tx["amount"] > 5000], key=lambda x: -x["amount"])
|
||
for tx in big_txs:
|
||
lines.append(f" {tx['date']} {tx['time']} | {tx['account']} | {tx['direction']} ¥{tx['amount']:.2f} | {tx['tx_type']} | {tx['remark']} | 分类: {tx['category']}")
|
||
|
||
with open(output_path, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines))
|
||
|
||
|
||
def main():
|
||
print("=" * 60)
|
||
print("家庭收支表合并脚本")
|
||
print("=" * 60)
|
||
|
||
# 1. 解析CSV
|
||
print("\n[1] 解析CSV文件...")
|
||
txs_1207 = parse_cmb_csv(FILE_1207, "丈夫-1207")
|
||
txs_1617 = parse_cmb_csv(FILE_1617, "妻子-1617")
|
||
all_txs = txs_1207 + txs_1617
|
||
print(f" 1207卡: {len(txs_1207)} 笔")
|
||
print(f" 1617卡: {len(txs_1617)} 笔")
|
||
print(f" 合计: {len(all_txs)} 笔")
|
||
|
||
# 2. 分类
|
||
print("\n[2] 交易分类...")
|
||
for tx in all_txs:
|
||
cat, excluded, reason = classify_transaction(tx)
|
||
tx["category"] = cat
|
||
tx["excluded"] = excluded
|
||
tx["exclude_reason"] = reason
|
||
|
||
# 3. 退款配对
|
||
print("\n[3] 退款配对...")
|
||
refund_count_before = sum(1 for tx in all_txs if tx["category"] == "退款" and not tx["excluded"])
|
||
paired = match_refunds(all_txs)
|
||
print(f" 待配对退款: {refund_count_before} 笔")
|
||
print(f" 成功配对: {paired} 笔")
|
||
print(f" 未配对退款(保留为收入): {refund_count_before - paired} 笔")
|
||
|
||
# 未配对退款重新分类
|
||
for tx in all_txs:
|
||
if tx["category"] == "退款" and not tx["excluded"]:
|
||
tx["category"] = "其他收入"
|
||
|
||
# 4. 统计排除项
|
||
print("\n[4] 统计...")
|
||
excluded_stats = defaultdict(lambda: {"count": 0, "amount": 0.0})
|
||
for tx in all_txs:
|
||
if tx["excluded"]:
|
||
excluded_stats[tx["category"]]["count"] += 1
|
||
excluded_stats[tx["category"]]["amount"] += tx["amount"]
|
||
|
||
for cat, stats in sorted(excluded_stats.items()):
|
||
print(f" 排除-{cat}: {stats['count']} 笔, ¥{stats['amount']:.2f}")
|
||
|
||
kept_count = sum(1 for tx in all_txs if not tx["excluded"])
|
||
print(f" 保留: {kept_count} 笔")
|
||
|
||
# 5. 生成输出
|
||
print("\n[5] 生成输出文件...")
|
||
detail_count = generate_detail_csv(all_txs, OUT_DETAIL)
|
||
print(f" 家庭收支明细.csv: {detail_count} 笔")
|
||
|
||
summary_count = generate_summary_csv(all_txs, OUT_SUMMARY)
|
||
print(f" 家庭月度汇总.csv: {summary_count} 个月")
|
||
|
||
generate_log(all_txs, excluded_stats, OUT_LOG)
|
||
print(f" 处理日志.txt")
|
||
|
||
print("\n" + "=" * 60)
|
||
print("完成！输出文件位于 data/ 目录下。")
|
||
print("=" * 60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|