""" 支付宝账单清理模块 """ import csv from decimal import Decimal from .base import ( BaseCleaner, parse_amount, format_amount, is_in_date_range, create_arg_parser ) from category import reclassify_if_needed, get_platform_merchants class AlipayCleaner(BaseCleaner): """支付宝账单清理器""" def clean(self) -> None: """执行清理""" self.print_header() # 读取数据，跳过支付宝导出文件的头部信息 with open(self.input_file, "r", encoding="utf-8") as f: reader = csv.reader(f) header = None rows = [] for row in reader: # 跳过空行 if not row or not row[0].strip(): continue # 查找实际的CSV头部行（包含"交易时间"和"交易分类"） if header is None: if len(row) >= 2 and "交易时间" in row[0] and "交易分类" in row[1]: header = row continue # 跳过头部信息行 continue # 收集数据行 rows.append(row) # 确保找到了有效的头部 if header is None: raise ValueError("无法找到有效的支付宝账单表头（需包含'交易时间'和'交易分类'列）") self.stats["original_count"] = len(rows) print(f"原始数据行数: {len(rows)}") # 第一步：按日期范围筛选 rows_filtered = [ row for row in rows if row and is_in_date_range(row[0], self.start_date, self.end_date) ] self.stats["filtered_count"] = len(rows_filtered) date_desc = f"{self.start_date} ~ {self.end_date}" if self.start_date or self.end_date else "全部" print(f"筛选后数据行数: {len(rows_filtered)} ({date_desc})") # 第二步：分离退款和非退款条目 refund_rows = [] expense_rows = [] for row in rows_filtered: if len(row) > 1 and row[1] == "退款": refund_rows.append(row) else: expense_rows.append(row) print(f"退款条目数: {len(refund_rows)}") print(f"非退款条目数: {len(expense_rows)}") # 第三步：处理退款 order_refunds = self._aggregate_refunds(refund_rows) print(f"有退款的订单数: {len(order_refunds)}") # 第四步：处理每笔支出 final_rows = self._process_expenses(expense_rows, order_refunds) print(f"\n处理结果:") print(f" 全额退款删除: {self.stats['fully_refunded']} 条") print(f" 部分退款调整: {self.stats['partially_refunded']} 条") if self.stats.get("zero_amount", 0) > 0: print(f" 0元记录过滤: {self.stats['zero_amount']} 条") print(f" 最终保留行数: {len(final_rows)}") # 第五步：重新分类并添加"需复核"标注 final_rows = self.reclassify(final_rows, header) if self.stats["category_adjusted"] > 0: print(f" 分类调整: {self.stats['category_adjusted']} 条") self.stats["final_count"] = len(final_rows) # 写入文件 self.write_output(header, final_rows) print(f"\n清理后的数据已保存到: {self.output_file}") def _aggregate_refunds(self, refund_rows: list) -> dict: """聚合退款金额""" order_refunds = {} for row in refund_rows: if len(row) >= 11: refund_order_no = row[9].strip() refund_merchant_no = row[10].strip() refund_amount = parse_amount(row[6]) original_order = refund_order_no.split("_")[0] if "_" in refund_order_no else refund_order_no key = original_order if original_order else refund_merchant_no if key: if key not in order_refunds: order_refunds[key] = Decimal("0") order_refunds[key] += refund_amount print(f" 退款记录: {row[0]} | {row[2]} | {refund_amount}元") return order_refunds def _process_expenses(self, expense_rows: list, order_refunds: dict) -> list: """处理支出记录""" final_rows = [] for row in expense_rows: if len(row) >= 12: order_no = row[9].strip() merchant_no = row[10].strip() expense_amount = parse_amount(row[6]) # 查找对应的退款 refund_amount = Decimal("0") matched_key = None for key, amount in order_refunds.items(): if key and (order_no == key or merchant_no == key or order_no.startswith(key)): refund_amount = amount matched_key = key break if matched_key: if refund_amount >= expense_amount: # 全额退款，删除 self.stats["fully_refunded"] += 1 print(f" 全额退款删除: {row[0]} | {row[2]} | {row[4][:25]}... | 原{expense_amount}元") else: # 部分退款，保留差额 remaining = expense_amount - refund_amount new_row = row.copy() new_row[6] = format_amount(remaining) original_remark = new_row[11] if len(new_row) > 11 else "" new_row[11] = f"原金额{expense_amount}元,退款{refund_amount}元{';' + original_remark if original_remark else ''}" final_rows.append(new_row) self.stats["partially_refunded"] += 1 print(f" 部分退款: {row[0]} | {row[2]} | 原{expense_amount}元 -> {format_amount(remaining)}元") else: # 过滤掉金额为 0 的记录（预下单/加购物车等无效记录） if expense_amount > 0: final_rows.append(row) else: self.stats["zero_amount"] = self.stats.get("zero_amount", 0) + 1 else: final_rows.append(row) return final_rows def _is_platform_merchant(self, merchant: str) -> bool: """判断是否为平台型商家（从配置文件读取）""" platform_merchants = get_platform_merchants() return any(platform in merchant for platform in platform_merchants) def reclassify(self, rows: list, header: list) -> list: """ 重新分类支付宝账单，并添加"复核等级"标注字段 只对平台型商家（美团、京东、抖音等）进行分类调整， 其他商家直接信任支付宝原分类。 复核等级: 空 = 无需复核 低 = 分类被调整，需确认调整是否正确 高 = 完全无法判断，需人工分类 字段索引: 0: 交易时间 1: 交易分类 2: 交易对方 4: 商品说明 5: 收/支 """ # 添加"复核等级"字段到表头 if "复核等级" not in header: header.append("复核等级") review_low_count = 0 review_high_count = 0 for row in rows: if len(row) >= 6: original_category = row[1] merchant = row[2] product = row[4] income_expense = row[5] review_mark = "" # 只对平台型商家进行重新分类 if self._is_platform_merchant(merchant): new_category, changed, review_level = reclassify_if_needed( original_category, merchant, product, income_expense ) if changed: row[1] = new_category self.stats["category_adjusted"] += 1 print(f" 分类调整: {merchant[:15]}... | {original_category} -> {new_category}") # 添加复核等级标注 if review_level == 1: review_mark = "LOW" review_low_count += 1 elif review_level == 2: review_mark = "HIGH" review_high_count += 1 # 非平台商家：直接信任支付宝原分类，无需复核 # 确保行长度足够 while len(row) < len(header) - 1: row.append("") row.append(review_mark) if review_high_count > 0: print(f" 高优先级复核: {review_high_count} 条（无法判断）") if review_low_count > 0: print(f" 低优先级复核: {review_low_count} 条（分类已调整）") return rows def main(): """命令行入口""" parser = create_arg_parser("清理支付宝交易明细数据") args = parser.parse_args() from .base import get_output_file, compute_date_range cleaner = AlipayCleaner(args.input_file, args.output_file) start_date, end_date = compute_date_range(args) cleaner.set_date_range(start_date, end_date) cleaner.clean() if __name__ == "__main__": main()