共享费用账单分摊与结算规划器

--- name: shared-expense-bill-splitter-and-settle-up-planner description: 将旅行、公寓或活动的共享费用CSV账本进行分摊，支持均摊、份额、精确、百分比和含税费/小费按比例分摊的明细账单，'除某人外所有人'排除规则，多个付款人，退款，还款以及按约定汇率的多币种计算（精确到分），然后找出可证明的最少结算转账（精确零和子群搜索）。返回余额、每人审计和聊天消息。包含一个带有--self-test的stdlib Python规划器。 --- # 共享费用账单分摊与结算规划器 你将一份混乱的共享消费清单转化为公平、可核查的分摊结果和尽可能简短的支付清单。你以CSV格式保存账本，方便团队在任何电子表格中编辑，对每一行应用其真正需要的分摊规则（谁在场，谁点了什么，谁少住了几晚），按团队同意的汇率转换货币，并以最少的转账结算所有人。每一分钱都有记录，任何人都可以逐行追溯自己的数字。 ## 此技能中的文件 - `scripts/settle_up.py` - 读取CSV账本，以整数分（最大余数舍入）分摊每一行，计算每人的已付/应摊/净额，并搜索所有零和子群以找到最少的转账次数（最多18人有余额时精确，超过则贪心）（仅使用Python 3.8+标准库） - `references/split-methods.md` - CSV格式，哪种分摊方法用于哪种费用，含税费和小费的明细账单，排除规则，付款人，退款和还款 - `references/settle-up-and-currency.md` - 为什么最小值可以低于“人数 - 1”，搜索如何工作，舍入，汇率和卡手续费，支付礼仪 - `templates/settle-up-summary.md` - 需要交回并粘贴到群聊中的摘要 - `examples/lisbon-trip.csv` - 5个朋友，11行：按夜数份额，一个被排除的人，一份明细晚餐，一笔美元预订，高级座位，两个付款人，一笔退款，英镑零食和一笔已完成的还款 - `examples/lisbon-settle-output.md` - 规划器在该账本上的真实输出 - `examples/flat-october.csv` - 6个室友，其中最小值（4次转账）优于最大债务付给最大债权（5次） ## 工作流程 ### 1. 收集账本 询问每笔共享费用：内容，谁付的（如果多人支付，各付多少），金额和币种，以及费用是为谁产生的。提出会改变分摊方式的问题：有没有人晚到或早退？有没有人跳过一餐或一项活动？晚餐是共享的还是各自点的？有没有人已经还过别人钱？确定基础货币和汇率（整个行程每种货币一个汇率，除非团队希望使用信用卡账单汇率）。按照`references/split-methods.md`中的格式写成CSV，并包含`# people:`以便捕获姓名拼写错误。 ### 2. 运行规划器 ``` python3 scripts/settle_up.py examples/lisbon-trip.csv python3 scripts/settle_up.py flat.csv --base GBP --rates EUR=0.86 --detail Ben python3 scripts/settle_up.py ledger.csv --json python3 scripts/settle_up.py --self-test ``` 退出码0 = 已打印计划（阅读警告部分），2 = 账本有错误（总和不对，缺少汇率，未知的分摊方式或姓名）- 修复它指出的那一行。如果你无法运行代码，请手动执行相同步骤：将每一行转换为基础货币分，分摊，计算每人的已付减去应摊，然后将债务人与债权人匹配，首先寻找余额恰好抵消的个人或小群体。 ### 3. 与团队核对结果 - 检查行必须显示`nets add up to 0.00`；已付和应摊必须相等。 - 用通俗语言读出余额（“Ana付了990.00，她的份额是377.31，Eli已经给了她100.00，所以她拿回512.69”）。 - 指出驱动结果且容易引起争议的行（按夜数份额，明细餐费，退款，还款），并为任何想要自己明细的人提供`--detail <name>`。 - 如果有人更愿意付给其他人（或无法直接付给某人），在添加约定的转账作为`payment`行后重新运行；计划会重新计算其余部分。 ### 4. 交付 填写`templates/settle-up-summary.md`：总计，余额表，编号转账，假设（汇率，谁在哪里被排除，税费/小费如何分摊），以及简短的群聊消息。建议通过将每笔转账作为`payment`行添加到账本中来标记完成，这样第二次运行会显示每个人为0.00。 ## 示例结果 在`examples/lisbon-trip.csv`上（5人，花费1,583.07欧元），Ana支付了公寓和法多音乐会门票，拿回512.69欧元；计划是4次转账：Ben付给Ana 243.17，Dara付给Ana 140.94，Eli付给Ana 128.58，Eli付给Cem 0.85（Eli早先的100.00还款已计入）。在`examples/flat-october.csv`上，Bo和Finn相互抵消，所以6个室友用4次转账结算，而最大债务付给最大债权则需要5次。 ## 规则 - 绝不虚构金额、汇率或参与者；询问，或在摘要中将该行标记为假设。 - 每种货币使用一个约定的汇率并展示它；不要将实时汇率混入已完成的分摊中。 - 所有计算以分为单位，并展示舍入分的去向（最大余数，平局时给列表中第一个列出的人）。 - 不要评判消费；只标记看起来像重复、拼写错误或缺少参与者的行。 - 还款是`payment`行，绝不是负费用；退款是由与原始费用相同的人分摊的负费用。 - 将计划作为提案呈现；只要每个人的净额不变，团队可以交换谁付给谁。 FILE:references/split-methods.md # 分摊方法与CSV格式 ## 列 `date,description,paid_by,amount,currency,split,participants,details` - 需要表头行；任何电子表格都可以导出。表头前的可选注释行： ``` # people: Ana, Ben, Cem, Dara, Eli （严格列表：拼写错误的姓名是错误） # base: EUR # rates: USD=0.92, GBP=1.16 （1美元 = 0.92欧元） ``` 如果没有`# people:`，文件中任何地方提到的每个人都是成员，`all`表示所有人；姓名匹配不区分大小写。 ## 选择分摊方法 | 费用 | 分摊方法 | 示例 `participants` / `details` | |---|---|---| | 杂货，出租车，燃料，共享门票 | `equal` | `all`，`Ana;Ben`，`all-Eli` | | 住宿（人们住了不同夜数，房间大小不同） | `shares` | `Ana:3;Ben:3;Dara:2`（夜数或房间权重） | | 价格不同的门票，已知的每人金额 | `exact` | `Ana:30;Ben:45`（participants中的其他人平分剩余部分） | | 约定的百分比（按收入分摊租金，喝酒的人分摊葡萄酒） | `percent` | `Ana:25;Ben:25;Dara:50`（必须总计100） | | 人们单独点餐的餐厅账单 | `itemized` | `Ana:31;Ben:28.50;shared:42` | | 某人已经还了某人钱 | `payment` | participants = 唯一的收款人 | ## 明细账单 - `details`列出每个人在税费、服务费和小费之前的项目；`shared`是开胃菜、瓶装酒或任何大家共享的东西的总和（在`participants`中均分）。 - `amount`是最终账单。差额（金额 - 项目）是税费、服务费和小费，按每个人所点内容的比例分摊，所以点得多的人付更多小费。输出显示为`+17.3% tax/tip`。 - 如果项目加起来超过账单（折扣或拼写错误），规划器停止；通过降低其适用的项目价格来输入折扣。 ## 排除和部分参与 - `all-Eli`或`all except Eli` = 除Eli外的所有人；可以链式排除多个：`all-Bo-Finn`。 - 对于部分时间在场的人，使用带夜数或天数的`shares`而不是排除他们。 - 由某位成员支付的孩子或客人：列在该成员名下（`shares`权重1.5）或添加到`# people:`中并与他们的父母结算。 ## 付款人 - 单一付款人：`Ana`。一张账单多个付款人：`Ben:40;Cem:20`（金额以行币种计，必须加起来等于`amount`）。 - 情侣共用的卡：将情侣视为一个人或拆分付款人金额。 ## 退款，押金，还款 - 取消后收到的退款：负`amount`，`paid_by` = 收到钱的人，与原始费用相同的participants。 - 房东退还的押金：负`amount`，与押金相同的分摊方式。 - 成员之间转交的钱：`payment`（从`paid_by`到`participants`中的唯一一个人）。以相同方式记录结算转账；重新运行后显示谁仍未结清。 ## 规划器捕获的常见错误 - 精确金额或百分比加起来不对，付款人金额与账单不符，项目超过账单。 - 没有汇率的币种，未知的分摊词，设置`# people:`时姓名拼写错误。 - 两行完全相同（相同日期，描述，金额和付款人）- 报告为警告，不删除。 - 零金额 - 跳过并警告。 FILE:references/settle-up-and-currency.md # 结算与货币 ## 净余额 对每个人：`net = paid + payments sent - share - payments received`。正数 = 拿回钱，负数 = 欠钱。净额总是恰好加起来为0.00，因为每一行都以整数分分摊，总和等于行总额。 ## 为什么“最少转账”可以少于人数 - 1 - 任何有k个有余额的人的小组最多可以用k - 1次转账结算（每次转账至少清零一个人）。 - 如果小组分成若干子群，每个子群的余额加起来为零，每个子群自行结算：有g个这样的子群时，最小值是k - g。示例：Bo +300和Finn -300抵消，所以6个室友需要4次转账，而不是5次。 - 将最大债务付给最大债权（通常的应用程序方法）不会寻找这些子群，可能需要更多转账。 - 找到最佳零和子群划分通常是一个难题；规划器对最多18个有非零余额的人精确求解（子集动态规划，不到一秒），超过则回退到贪心计划并说明。 - 在每个子群内，计划将最大债务付给最大债权，恰好使用（子群大小 - 1）次转账。 - 多个计划可以具有相同的最小次数；任何一个都是公平的，因为每个人的总进出由其净额固定。 ## 舍入 - 每一行转换为基础货币分一次（`amount x rate`，四舍五入到最接近的半分），然后用最大余数法分摊：每个人得到向下取整的金额，剩余的分给最大余数，平局时给`# people:`中第一个列出的人。 - 在整个行程中，这每人最多移动几分钱；如果有人问为什么10.00 / 3对一个人是3.34，请说明。 ## 汇率 - 分摊前为每种货币约定一个汇率：大多数费用支付当天的汇率，行程平均值，或付款人信用卡账单上的汇率（包括银行加价）。 - 如果付款人的卡收取了外币交易费，要么将其包含在金额中（团队分摊），要么排除（付款人吸收）- 为整个行程决定一次。 - 以实际支付的币种输入金额，让规划器转换；不要手动预先转换某些行。 - 以基础货币结算。如果有人以另一种货币支付，按相同的约定汇率转换转账，并以该币种记录为`payment`行。 ## 支付礼仪 - 发送带有转账和账本链接的群聊消息；给人们一两天时间在支付前对某行提出异议。 - 小额转账（约2欧元/美元以下）可以在双方同意的情况下免除或滚入下一次共享费用 - 将其记录为`payment`行以保持账本精确。 - 当每个人都付清后，将每笔转账添加为`payment`行并重新运行：每个净额应为0.00。 FILE:templates/settle-up-summary.md # 结算：<旅行 / 公寓 / 活动>，<日期> **团队：** <Ana, Ben, Cem, Dara, Eli> **基础货币：** <EUR> **总花费：** <1,583.07 EUR> 共<11>行 **约定汇率：** <1 USD = 0.92 EUR, 1 GBP = 1.16 EUR（行程平均值）> ## 余额 | 人 | 已付 | 应摊份额 | 已发送 / 已接收 | 净额 | |---|---|---|---|---| | <Ana> | <990.00> | <377.31> | <收到 100.00> | <+512.69 拿回> | | <Ben> | <64.00> | <307.17> | <-> | <-243.17 欠> | ## 转账（<n>，最小可能） 1. <Ben付给Ana 243.17 EUR> 2. <Dara付给Ana 140.94 EUR> 3. <...> ## 行如何分摊 - <公寓：按夜数，Dara和Eli住了3晚中的2晚> - <机场出租车：除Eli外的所有人（他坐火车来）> - <晚餐：明细，17.3%税费和小费按每人所点内容分摊；开胃菜42.00共享> - <宫殿门票退款：退给Ben和Cem，他们付的款> - <Eli给Ana的100.00已计入> ## 假设和未决问题 - <上述汇率；卡手续费未包含> - <检查第5行（冲浪课）：Ben在场吗？> ## 群聊消息 ``` 结算（<EUR>，<11>行，汇率：<USD 0.92, GBP 1.16>）： - <Ben -> Ana: 243.17 EUR> - <Dara -> Ana: 140.94 EUR> 完整表格：<表格链接>。2天内如有行错误请提出。 ``` **付清后：** 将每笔转账添加为`payment`行并重新运行 - 每个净额应显示0.00。 FILE:examples/lisbon-trip.csv # people: Ana, Ben, Cem, Dara, Eli # base: EUR # rates: USD=0.92, GBP=1.16 date,description,paid_by,amount,currency,split,participants,details 2026-10-02,公寓3晚（Dara和Eli 2晚）,Ana,840.00,EUR,shares,all,Ana:3;Ben:3;Cem:3;Dara:2;Eli:2 2026-10-02,机场出租车（Eli坐火车来）,Ben,48.00,EUR,equal,all-Eli, 2026-10-02,公寓杂货,Eli,63.75,EUR,equal,all, 2026-10-03,Taberna do Mar晚餐,Cem,212.40,EUR,itemized,all,Ana:31;Ben:28.50;Cem:24;Dara:36;Eli:19.50;shared:42 2026-10-03,冲浪课（在美国网站预订）,Dara,180.00,USD,equal,Ana;Dara;Eli, 2026-10-03,法多音乐会门票（Dara和Eli高级座位）,Ana,150.00,EUR,exact,all-Cem,Ana:30;Ben:30;Dara:45;Eli:45 2026-10-04,去辛特拉的火车,Ben:40;Cem:20,60.00,EUR,equal,all, 2026-10-04,公寓葡萄酒,Dara,36.00,EUR,percent,,Ana:25;Ben:25;Dara:50 2026-10-04,宫殿门票退款（下雨）,Ben,-24.00,EUR,equal,Ben;Cem, 2026-10-05,伦敦中转零食,Cem,27.00,GBP,equal,all, 2026-10-05,Eli还给Ana公寓钱,Eli,100.00,EUR,payment,Ana, FILE:examples/lisbon-settle-output.md # 结算计划 基础货币：EUR 人数：5 行数：11 总花费：1,583.07 EUR 使用汇率：1 GBP = 1.16 EUR, 1 USD = 0.92 EUR ## 费用 | # | 日期 | 内容 | 付款人 | 金额 | 折合EUR | 分摊方式 | 分摊对象 | |---|---|---|---|---|---|---|---| | 1 | 2026-10-02 | 公寓3晚（Dara和Eli 2晚） | Ana | 840.00 EUR | 840.00 | shares | all | | 2 | 2026-10-02 | 机场出租车（Eli坐火车来） | Ben | 48.00 EUR | 48.00 | equal | Ana, Ben, Cem, Dara | | 3 | 2026-10-02 | 公寓杂货 | Eli | 63.75 EUR | 63.75 | equal | all | | 4 | 2026-10-03 | Taberna do Mar晚餐 | Cem | 212.40 EUR | 212.40 | itemized (+17.3% tax/tip) | all | | 5 | 2026-10-03 | 冲浪课（在美国网站预订） | Dara | 180.00 USD | 165.60 | equal | Ana, Dara, Eli | | 6 | 2026-10-03 | 法多音乐会门票（Dara和Eli高级座位） | Ana | 150.00 EUR | 150.00 | exact | Ana, Ben, Dara, Eli | | 7 | 2026-10-04 | 去辛特拉的火车 | Ben 40.00, Cem 20.00 | 60.00 EUR | 60.00 | equal | all | | 8 | 2026-10-04 | 公寓葡萄酒 | Dara | 36.00 EUR | 36.00 | percent | Ana, Ben, Dara | | 9 | 2026-10-04 | 宫殿门票退款（下雨） | Ben | -24.00 EUR | -24.00 | equal | Ben, Cem | | 10 | 2026-10-05 | 伦敦中转零食 | Cem | 27.00 GBP | 31.32 | equal | all | | 11 | 2026-10-05 | Eli还给Ana公寓钱 | Eli | 100.00 EUR | 100.00 | payment | 给Ana | ## 余额 | 人 | 已付 | 应摊 | 已发送 | 已接收 | 净额 | |---|---|---|---|---|---| | Ana | 990.00 | 377.31 | 0.00 | 100.00 | 512.69（拿回） | | Ben | 64.00 | 307.17 | 0.00 | 0.00 | -243.17（欠） | | Cem | 263.72 | 262.87 | 0.00 | 0.00 | 0.85（拿回） | | Dara | 201.60 | 342.54 | 0.00 | 0.00 | -140.94（欠） | | Eli | 63.75 | 293.18 | 100.00 | 0.00 | -129.43（欠） | 检查：净额加起来为0.00（已付1,583.07 = 应摊1,583.07）。 ## 转账（4） 1. Ben付给Ana **243.17 EUR** 2. Dara付给Ana **140.94 EUR** 3. Eli付给Ana **128.58 EUR** 4. Eli付给Cem **0.85 EUR** 最小可能：5个有余额的人需要4次转账（没有更小的子群能自行抵消）。 ## 群聊消息 ``` 结算（EUR，11行，汇率：GBP 1.16, USD 0.92）： - Ben -> Ana: 243.17 EUR - Dara -> Ana: 140.94 EUR - Eli -> Ana: 128.58 EUR - Eli -> Cem: 0.85 EUR 完整表格在共享表格中。2天内如有行错误请提出。 ``` FILE:examples/flat-october.csv # people: Asha, Bo, Chen, Dev, Ema, Finn # base: EUR date,description,paid_by,amount,currency,split,participants,details 2026-10-01,Bo和Finn的音乐节门票,Bo,600,EUR,exact,Bo;Finn,Bo:300;Finn:300 2026-10-31,十月杂货（Bo和Finn巡演中）,Asha,800,EUR,equal,all-Bo-Finn, 2026-10-31,九月至十月电费（Bo和Finn巡演中）,Chen,600,EUR,equal,all-Bo-Finn, 2026-10-20,二手沙发（Dev和Ema的部分）,Asha,250,EUR,exact,Dev;Ema,Dev:150;Ema:100 FILE:scripts/settle_up.py #!/usr/bin/env python3 """共享费用账单分摊与结算规划器（仅使用Python 3.8+标准库）。 读取共享费用（旅行、公寓、活动）的CSV账本，以整数分分摊每一行， 并找到结算团队所需的最小可能转账次数（对零和子群进行精确搜索， 不仅仅是“最多人数 - 1”）。 CSV列（需要表头行，额外列忽略）： date,description,paid_by,amount,currency,split,participants,details paid_by Ana 单一付款人 Ana:40;Ben:20 多个付款人（金额以行币种计，必须加起来等于总额） amount 行币种的总金额；负数 = paid_by收到的退款 currency 空白 = 基础货币 split equal | shares | exact | percent | itemized | payment participants all | Ana;Ben;Cem | all-Eli（除Eli外的所有人）| 空白 = details中的名字 details shares: Ana:3;Ben:2 （participants中未列出的人默认权重1） exact: Ana:30;Ben:45 （其余参与者，如果有，平分剩余部分） percent: Ana:50;Ben:30;Cem:20 （必须总计100） itemized: Ana:18;Ben:22.50;shared:12 （税费/小费前的项目价格；"shared" 项目均分；金额 - 项目 = 税费/服务费/小费，按每人所点内容比例分摊） payment: （无）paid_by将`amount`给了participants中的唯一一个人 顶部的注释行设置团队和汇率（或使用CLI标志）： # people: Ana, Ben, Cem # base: EUR # rates: USD=0.92, GBP=1.16 （该币种1单位 = 基础货币rate单位） 用法： python3 settle_up.py ledger.csv [--base EUR] [--rates USD=0.92,GBP=1.16] [--people Ana,Ben] [--detail Ana] [--json] python3 settle_up.py --self-test 退出码：0 = 已打印结算计划，2 = 输入错误（未知分摊方式，总和不对， 缺少汇率，严格列表中的未知人）。 """ import argparse import csv import io import json import re import sys from decimal import Decimal, ROUND_HALF_UP, InvalidOperation MAX_EXACT = 18 # 有非零余额的人数，用于精确最小搜索（2^18个状态） SPLITS = ("equal", "shares", "exact", "percent", "itemized", "payment") class LedgerError(ValueError): pass def dec(s, what): try: return Decimal(str(s).replace(",", "").strip()) except (InvalidOperation, AttributeError): raise LedgerError("%s: %r 不是数字" % (what, s)) def to_cents(amount): return int((amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP)) def fmt(cents, cur=""): sign = "-" if cents < 0 else "" c = abs(cents) s = "%s%d.%02d" % (sign, c // 100, c % 100) s = re.sub(r"(\d)(?=(\d{3})+\.)", r"\1,", s) return (s + " " + cur).strip() def allocate(total, weights, order): """按比例分摊整数`total`分（`weights`是名字 -> Decimal >= 0的字典）。 最大余数法；平局时给`order`中更早的名字。总和精确。""" names = [n for n in order if n in weights] wsum = sum(weights[n] for n in names) if wsum <= 0: raise LedgerError("权重加起来为零") sign = -1 if total < 0 else 1 t = abs(total) raw = {n: Decimal(t) * weights[n] / wsum for n in names} base = {n: int(raw[n]) for n in names} left = t - sum(base.values()) rank = sorted(names, key=lambda n: (-(raw[n] - base[n]), order.index(n))) for n in rank[:left]: base[n] += 1 return {n: sign * v for n, v in base.items() if v} def parse_pairs(text, what): out = {} for part in [p for p in re.split(r"[;|]", text or "") if p.strip()]: if ":" not in part: raise LedgerError("%s: %r 应该看起来像 名字:值" % (what, part.strip())) k, v = part.split(":", 1) k = k.strip() out[k] = out.get(k, Decimal(0)) + dec(v, what) return out class Ledger: def __init__(self, text, base=None, rates=None, people=None): self.warnings = [] self.base = base self.rates = {} self.people = [] self._canon = {} header_people = None body = [] for line in text.splitlines(): s = line.strip() if s.startswith("#"): m = re.match(r"#\s*(people|base|rates?)\s*:\s*(.*)$", s, re.I) if m: key, val = m.group(1).lower(), m.group(2) if key == "people": header_people = [p.strip() for p in re.split(r"[,;]", val) if p.strip()] elif key == "base" and not self.base: self.base = val.strip().upper() elif key.startswith("rate"): self._add_rates(val) continue body.append(line) self.base = (self.base or "EUR").upper() if rates: self._add_rates(rates) self.rates[self.base] = Decimal(1) for p in (people or header_people or []): self._person(p, declare=True) self.strict = bool(people or header_people) rows = list(csv.DictReader(io.StringIO("\n".join(body)))) if not rows: raise LedgerError("未找到费用行（需要表头行和至少一行）") need = {"description", "paid_by", "amount", "split"} cols = {c.strip().lower() for c in rows[0].keys() if c} if not need <= cols: raise LedgerError("缺少CSV列：%s" % ", ".join(sorted(need - cols))) self.rows = [] for r in rows: r = {(k or "").strip().lower(): (v or "").strip() for k, v in r.items()} if not any(r.values()): continue self.rows.append(r) # 第一遍：当没有给出人员列表时，学习每个名字 if not self.strict: for r in self.rows: for p in self._names_in(r): self._person(p, declare=True) self.lines = [self._line(n, r) for n, r in enumerate(self.rows, 1)] self._check_duplicates() # -- 人员和汇率 def _add_rates(self, text): for part in re.split(r"[,;]", text): if not part.strip(): continue if "=" not in part: raise LedgerError("汇率 %r 应该看起来像 USD=0.92" % part.strip()) k, v = part.split("=", 1) r = dec(v, "汇率 " + k.strip()) if r <= 0: raise LedgerError("%s的汇率必须为正" % k.strip()) self.rates[k.strip().upper()] = r def _person(self, name, declare=False): key = name.strip().lower() if not key: raise LedgerError("空人名") if key in self._canon: canon = self._canon[key] if canon != name.strip() and declare: self.warnings.append("'%s'和'%s'被视为同一个人" % (canon, name.strip())) return canon if not declare: raise LedgerError("未知人员 %r（不在人员列表中）" % name.strip()) self._canon[key] = name.strip() self.people.append(name.strip()) return name.strip() def _names_in(self, r): names = [] pb = r.get("paid_by", "") names += list(parse_pairs(pb, "paid_by").keys()) if ":" in pb else [pb] part = r.get("participants", "") if part and not part.lower().startswith("all"): names += [p for p in re.split(r"[;|]", part) if p.strip()] if r.get("split", "").lower() != "payment": names += [k for k in parse_pairs(r.get("details", ""), "details") if k.lower() != "shared"] return [n for n in names if n.strip()] def _participants(self, r, ln): part = r.get("participants", "").strip() if not part: return None if part.lower().startswith("all"): rest = part[3:].strip() excl = [] if rest: rest = re.sub(r"^(except|but|without)\s+", "", rest.lstrip("-").strip(), flags=re.I) excl = [self._person(x) for x in re.split(r"[-;|,]", rest) if x.strip()] return [p for p in self.people if p not in excl] return [self._person(p) for p in re.split(r"[;|]", part) if p.strip()] # -- 一行CSV -> 已付/应摊（基础货币分） def _line(self, ln, r): desc = r.get("description") or "第%d行" % ln where = "第%d行（%s）" % (ln, desc) split = (r.get("split") or "equal").lower() if split not in SPLITS: raise LedgerError("%s：未知分摊方式 %r（使用 %s）" % (where, split, ", ".join(SPLITS))) cur = (r.get("currency") or self.base).upper() if cur not in self.rates: raise LedgerError("%s：没有%s的汇率（添加'# rates: %s=...'或--rates）" % (where, cur, cur)) amount = dec(r.get("amount"), where + " 金额") if amount == 0: self.warnings.append("%s 的金额为0，已跳过" % where) total = to_cents(amount * self.rates[cur]) # 付款人 pb = r.get("paid_by", "") if ":" in pb: pays = {self._person(k): v for k, v in parse_pairs(pb, where + " paid_by").items()} if sum(pays.values()) != amount: raise LedgerError("%s：付款人加起来为%s但金额是%s" % (where, sum(pays.values()), amount)) paid = allocate(total, {k: abs(v) for k, v in pays.items()}, self.people) if total else {} if total < 0: paid = {k: -abs(v) for k, v in paid.items()} else: paid = {self._person(pb): total} if total else {} people = self._participants(r, ln) details = parse_pairs(r.get("details", ""), where + " details") if split != "payment" else {} share = {} if split == "payment": if not people or len(people) != 1: raise LedgerError("%s：一笔payment需要participants中恰好一个收款人" % where) share = {people[0]: total} elif split == "equal": people = people or self.people share = allocate(total, {p: Decimal(1) for p in people}, self.people) elif split == "shares": d = {self._person(k): v for k, v in details.items()} people = people or list(d) w = {p: d.get(p, Decimal(1)) for p in people} extra = [p for p in d if p not in people] if extra: raise LedgerError("%s：details中的%s不在participants中" % (where, ", ".join(extra))) if any(v < 0 for v in w.values()): raise LedgerError("%s：份额不能为负" % where) share = allocate(total, w, self.people) elif split == "percent": d = {self._person(k): v for k, v in details.items()} if sum(d.values()) != 100: raise LedgerError("%s：百分比加起来为%s，不是100" % (where, sum(d.values()))) people = list(d) share = allocate(total, d, self.people) elif split == "exact": d = {self._person(k): v for k, v in details.items()} people = people or list(d) fixed = sum(d.values()) rest_people = [p for p in people if p not in d] if fixed > abs(amount) or (fixed != abs(amount) and not rest_people): raise LedgerError("%s：精确金额加起来为%s但账单是%s" % (where, fixed, amount)) w = dict(d) if rest_people: each = (abs(amount) - fixed) / len(rest_people) for p in rest_people: w[p] = each share = allocate(total, w, self.people) elif split == "itemized": people = people or self.people shared = details.pop("shared", details.pop("Shared", Decimal(0))) d = {self._person(k): v for k, v in details.items()} items = sum(d.values()) + shared if items > abs(amount): raise LedgerError("%s：项目加起来为%s，超过账单%s" % (where, items, amount)) per = {p: d.get(p, Decimal(0)) + (shared / len(people) if shared else 0) for p in people} for p in d: if p not in per: per[p] = d[p] if sum(per.values()) <= 0: raise LedgerError("%s：没有可分摊的项目" % where) # 税费/小费按每人项目比例 == 按项目权重分摊总额 share = allocate(total, per, self.people) rec = {"n": ln, "date": r.get("date", ""), "desc": desc, "split": split, "currency": cur, "amount": amount, "total": total, "paid": paid, "share": share, "among": sorted(share, key=self.people.index)} if split == "itemized": rec["extra_pct"] = (abs(amount) - items) / items * 100 if items else Decimal(0) if sum(paid.values()) != sum(share.values()): raise LedgerError("%s：内部舍入错误" % where) # 不应发生 return rec def _check_duplicates(self): seen = {} for l in self.lines: key = (l["date"], l["desc"].lower(), l["total"], tuple(sorted(l["paid"]))) if key in seen: self.warnings.append("第%d行看起来是第%d行的重复（%s，%s）" % (l["n"], seen[key], l["desc"], fmt(l["total"], self.base))) else: seen[key] = l["n"] def balances(self): b = {p: {"paid": 0, "share": 0, "sent": 0, "received": 0} for p in self.people} for l in self.lines: if l["split"] == "payment": for p, c in l["paid"].items(): b[p]["sent"] += c for p, c in l["share"].items(): b[p]["received"] += c else: for p, c in l["paid"].items(): b[p]["paid"] += c for p, c in l["share"].items(): b[p]["share"] += c for p in b: b[p]["net"] = b[p]["paid"] + b[p]["sent"] - b[p]["share"] - b[p]["received"] return b # ----------------------------------------------------------------------------- 结算 def greedy(net, order): """最大债务人付给最大债权人，直到每个人都平（最多k-1次转账）。""" cred = {p: v for p, v in net.items() if v > 0} debt = {p: -v for p, v in net.items() if v < 0} out = [] while cred and debt: c = max(cred, key=lambda p: (cred[p], -order.index(p))) d = max(debt, key=lambda p: (debt[p], -order.index(p))) x = min(cred[c], debt[d]) out.append((d, c, x)) cred[c] -= x debt[d] -= x if not cred[c]: del cred[c] if not debt[d]: del debt[d] return out def minimal(net, order): """最少转账：n -（最大不相交零和组数）。返回（转账，组，精确）。""" people = [p for p in order if net.get(p)] n = len(people) if n == 0: return [], [], True if n > MAX_EXACT: return greedy(net, order), [people], False vals = [net[p] for p in people] size = 1 << n sums = [0] * size for m in range(1, size): low = m & -m sums[m] = sums[m ^ low] + vals[low.bit_length() - 1] dp = [0] * size for m in range(1, size): best = 0 mm = m while mm: low = mm & -mm v = dp[m ^ low] if v > best: best = v mm ^= low dp[m] = best + (1 if sums[m] == 0 else 0) # 向下走，在每个零和检查点切割一个组 groups, m, last = [], size - 1, size - 1 while m: mm, pick = m, None target = dp[m] - (1 if sums[m] == 0 else 0) while mm: low = mm & -mm if dp[m ^ low] == target: pick = low break mm ^= low m ^= pick if sums[m] == 0: groups.append(last ^ m) last = m out, named = [], [] for g in reversed(groups): members = [people[i] for i in range(n) if g >> i & 1] named.append(members) out += greedy({p: net[p] for p in members}, order) return out, named, True # ----------------------------------------------------------------------------- 输出 def render(led, detail=None): b = led.balances() net = {p: b[p]["net"] for p in led.people} tr, groups, exact = minimal(net, led.people) gr = greedy(net, led.people) B = led.base o = ["# 结算计划", "", "基础货币：%s 人数：%d 行数：%d 总花费：%s" % ( B, len(led.people), len(led.lines), fmt(sum(l["total"] for l in led.lines if l["split"] != "payment"), B))] others = sorted(c for c in led.rates if c != B) if others: o.append("使用汇率：" + ", ".join("1 %s = %s %s" % (c, led.rates[c].normalize(), B) for c in others)) o += ["", "## 费用", "", "| # | 日期 | 内容 | 付款人 | 金额 | 折合%s | 分摊方式 | 分摊对象 |" % B, "|---|---|---|---|---|---|---|---|"] for l in led.lines: payer = ", ".join("%s %s" % (p, fmt(c)) if len(l["paid"]) > 1 else p for p, c in l["paid"].items()) among = "all" if len(l["among"]) == len(led.people) and l["split"] != "payment" else ", ".join(l["among"]) split = l["split"] + (" (+%.1f%% tax/tip)" % l["extra_pct"] if l["split"] == "itemized" else "") if l["split"] == "payment": among = "给 " + among o.append("| %d | %s | %s | %s | %s %s | %s | %s | %s |" % (l["n"], l["date"], l["desc"].replace("|", "/"), payer, fmt(to_cents(l["amount"])), l["currency"], fmt(l["total"]), split, among)) o += ["", "## 余额", "", "| 人 | 已付 | 应摊 | 已发送 | 已接收 | 净额 |", "|---|---|---|---|---|---|"] for p in led.people: x = b[p] state = "拿回" if x["net"] > 0 else ("欠" if x["net"] < 0 else "平") o.append("| %s | %s | %s | %s | %s | %s（%s） |" % (p, fmt(x["paid"]), fmt(x["share"]), fmt(x["sent"]), fmt(x["received"]), fmt(x["net"]), state)) o.append("") o.append("检查：净额加起来为%s（已付%s = 应摊%s）。" % (fmt(sum(net.values())), fmt(sum(x["paid"] for x in b.values())), fmt(sum(x["share"] for x in b.values())))) nz = sum(1 for v in net.values() if v) o += ["", "## 转账（%d）" % len(tr), ""] if not tr: o.append("每个人都平了。无需支付。") else: for i, (d, c, x) in enumerate(tr, 1): o.append("%d. %s付给%s **%s**" % (i, d, c, fmt(x, B))) o.append("") if exact: if len(groups) > 1: o.append("最小可能：%d次转账。团队分成%d个自行结算的集合（%s），所以%d人只需要%d次转账；最大债务付给最大债权则需要%d次。" % ( len(tr), len(groups), "; ".join(" + ".join(g) for g in groups), nz, len(tr), len(gr))) else: o.append("最小可能：%d个有余额的人需要%d次转账（没有更小的子群能自行抵消）。" % (len(tr), nz)) else: o.append("%d人有余额；跳过精确最小搜索（限制%d），贪心计划最多使用%d次转账。" % (nz, MAX_EXACT, nz - 1)) if detail: p = led._person(detail) o += ["", "## 明细：%s" % p, "", "| # | 内容 | 已付 | 应摊 | 累计净额 |", "|---|---|---|---|---|"] run = 0 for l in led.lines: pd, sh = l["paid"].get(p, 0), l["share"].get(p, 0) if not pd and not sh: continue run += pd - sh o.append("| %d | %s | %s | %s | %s |" % (l["n"], l["desc"], fmt(pd), fmt(sh), fmt(run))) if led.warnings: o += ["", "## 警告", ""] + ["- " + w for w in led.warnings] o += ["", "## 群聊消息", "", "```"] msg = ["结算（%s，%d行，汇率：%s）：" % (B, len(led.lines), ", ".join("%s %s" % (c, led.rates[c].normalize()) for c in others) or "无")] msg += ["- %s -> %s: %s" % (d, c, fmt(x, B)) for d, c, x in tr] or ["- 每个人都平了"] msg.append("完整表格在共享表格中。2天内如有行错误请提出。") o += msg + ["```"] return "\n".join(o) + "\n" def as_json(led): b = led.balances() net = {p: b[p]["net"] for p in led.people} tr, groups, exact = minimal(net, led.people) return {"base": led.base, "people": led.people, "balances_cents": b, "transfers": [{"from": d, "to": c, "cents": x} for d, c, x in tr], "groups": groups, "exact_minimum": exact, "warnings": led.warnings} # ----------------------------------------------------------------------------- 自测 def self_test(): import random cases = [] def case(name, fn): try: ok, err = bool(fn()), "" except Exception as e: # noqa ok, err = False, repr(e) cases.append((name, ok, err)) H = "date,description,paid_by,amount,currency,split,participants,details\n" def L(rows, **kw): return Ledger(H + rows, **kw) def nets(led): b = led.balances() return {p: b[p]["net"] for p in led.people} def raises(fn): try: fn() except LedgerError: return True return False case("均摊精确到分", lambda: nets(L(",Taxi,Ana,10.00,,equal,Ana;Ben;Cem,\n")) == {"Ana": 666, "Ben": -333, "Cem": -333}) case("余数分给顺序中第一个", lambda: L(",X,Ana,0.10,,equal,Ana;Ben;Cem,\n").lines[0]["share"] == {"Ana": 4, "Ben": 3, "Cem": 3}) case("all减排除", lambda: Ledger("# people: Ana, Ben, Cem\n" + H + ",X,Ana,30,,equal,all-Cem,\n").lines[0]["among"] == ["Ana", "Ben"]) case("all except词语", lambda: Ledger("# people: Ana, Ben, Cem\n" + H + ",X,Ana,30,,equal,all except Ben,\n").lines[0]["among"] == ["Ana", "Cem"]) case("份额权重", lambda: L(",Room,Ana,100,,shares,,Ana:3;Ben:1\n").lines[0]["share"] == {"Ana": 7500, "Ben": 2500}) case("未列出的参与者份额默认1", lambda: L(",Room,Ana,90,,shares,Ana;Ben;Cem,Ana:2;Ben:0\n").lines[0]["share"] == {"Ana": 6000, "Cem": 3000}) case("精确金额", lambda: L(",Tickets,Ana,75,,exact,,Ana:30;Ben:45\n").lines[0]["share"] == {"Ana": 3000, "Ben": 4500}) case("精确且其余均分", lambda: L(",T,Ana,100,,exact,Ana;Ben;Cem,Ana:40\n").lines[0]["share"] == {"Ana": 4000, "Ben": 3000, "Cem": 3000}) case("精确不匹配被拒绝", lambda: raises(lambda: L(",T,Ana,80,,exact,,Ana:30;Ben:45\n"))) case("百分比", lambda: L(",Wine,Ana,36,,percent,,Ana:25;Ben:25;Cem:50\n").lines[0]["share"] == {"Ana": 900, "Ben": 900, "Cem": 1800}) case("百分比不是100被拒绝", lambda: raises(lambda: L(",W,Ana,36,,percent,,Ana:25;Ben:25\n"))) case("含共享和税费/小费的明细", lambda: L(",Dinner,Ana,66,,itemized,Ana;Ben,Ana:20;Ben:30;shared:10\n").lines[0]["share"] == {"Ana": 2750, "Ben": 3850}) case("明细超过账单被拒绝", lambda: raises(lambda: L(",D,Ana,40,,itemized,Ana;Ben,Ana:20;Ben:30\n"))) case("还款行", lambda: nets(L(",Food,Ana,20,,equal,Ana;Ben,\n,Paid back,Ben,10,,payment,Ana,\n")) == {"Ana": 0, "Ben": 0}) case("还款需要一个收款人", lambda: raises(lambda: L(",P,Ben,10,,payment,Ana;Cem,\n"))) case("多个付款人", lambda: L(",Train,Ana:40;Ben:20,60,,equal,Ana;Ben;Cem,\n").lines[0]["paid"] == {"Ana": 4000, "Ben": 2000}) case("多个付款人必须加起来", lambda: raises(lambda: L(",Train,Ana:40;Ben:10,60,,equal,Ana;Ben,\n"))) case("退款（负金额）", lambda: nets(L(",Museum refund,Ben,-24,,equal,Ben;Cem,\n")) == {"Ben": -1200, "Cem": 1200}) case("货币转换", lambda: L(",Surf,Ana,100,USD,equal,Ana;Ben,\n", rates="USD=0.92").lines[0]["total"] == 9200) case("注释行中的汇率", lambda: Ledger("# base: EUR\n# rates: USD=0.92, GBP=1.16\n" + H + ",S,Ana,10,GBP,equal,Ana;Ben,\n").lines[0]["total"] == 1160) case("缺少汇率被拒绝", lambda: raises(lambda: L(",S,Ana,10,CHF,equal,Ana;Ben,\n"))) case("未知分摊方式被拒绝", lambda: raises(lambda: L(",S,Ana,10,,halves,Ana;Ben,\n"))) case("严格人员列表拒绝拼写错误", lambda: raises(lambda: Ledger("# people: Ana, Ben\n" + H + ",S,Anna,10,,equal,all,\n"))) case("不区分大小写的名字合并", lambda: L(",A,Ana,10,,equal,ana;Ben,\n").people == ["Ana", "Ben"]) case("重复行警告", lambda: any("重复" in w for w in L("1,Taxi,Ana,10,,equal,Ana;Ben,\n1,Taxi,Ana,10,,equal,Ana;Ben,\n").warnings)) case("缺少列被拒绝", lambda: raises(lambda: Ledger("what,who\nx,y\n"))) case("坏数字被拒绝", lambda: raises(lambda: L(",S,Ana,ten,,equal,Ana;Ben,\n"))) # 最少转账 case("找到零和配对（4人 -> 2次转账）", lambda: len(minimal({"A": 500, "B": -500, "C": 300, "D": -300}, list("ABCD"))[0]) == 2) case("无零和子群 -> 人数 - 1次转账", lambda: len(greedy({"A": 600, "B": 400, "C": -500, "D": -500}, list("ABCD"))) == 3 and len(minimal({"A": 600, "B": 400, "C": -500, "D": -500}, list("ABCD"))[0]) == 3) case("6人时最小优于贪心", lambda: (lambda n: len(minimal(n, list("ABCDEF"))[0]) == 4 and len(greedy(n, list("ABCDEF"))) == 5)({"A": 700, "B": 300, "C": 250, "D": -500, "E": -450, "F": -300})) case("每个人都平 -> 无转账", lambda: minimal({"A": 0, "B": 0}, ["A", "B"])[0] == []) def brute_min(net): ps = [p for p in net if net[p]] best = len(ps) - 1 if ps else 0 for parts in _partitions(ps): if all(sum(net[p] for p in g) == 0 for g in parts): best = min(best, len(ps) - len(parts)) return best def check_random(): rnd = random.Random(7) for _ in range(60): k = rnd.randint(2, 7) vals = [rnd.choice([-1, 1]) * rnd.choice([100, 200, 300, 500, 250, 50]) for _ in range(k - 1)] vals.append(-sum(vals)) net = {chr(65 + i): v for i, v in enumerate(vals)} tr = minimal(net, list(net))[0] bal = dict(net) for d, c, x in tr: assert x > 0 bal[d] += x bal[c] -= x if any(bal.values()) or len(tr) != brute_min(net): return False return True case("60个随机组上最小 == 暴力且结算所有人", check_random) case("转账结算类似示例的账本", lambda: _settles(L(",Flat,Ana,840,,shares,,Ana:3;Ben:3;Cem:2\n,Food,Cem,63.75,,equal,all,\n,Wine,Ben,36,,percent,,Ana:25;Ben:25;Cem:50\n"))) case("渲染有转账和聊天消息", lambda: "群聊消息" in render(L(",Taxi,Ana,10,,equal,Ana;Ben,\n"))) case("明细视图", lambda: "明细：Ben" in render(L(",Taxi,Ana,10,,equal,Ana;Ben,\n"), "ben")) case("json输出", lambda: as_json(L(",Taxi,Ana,10,,equal,Ana;Ben,\n"))["transfers"] == [{"from": "Ben", "to": "Ana", "cents": 500}]) case("千位格式", lambda: fmt(123456789, "EUR") == "1,234,567.89 EUR" and fmt(-5) == "-0.05") passed = sum(1 for _, ok, _ in cases if ok) for name, ok, err in cases: print(("通过 " if ok else "失败 ") + name + ((" " + err) if err else "")) print("%d/%d 通过" % (passed, len(cases))) return 0 if passed == len(cases) else 1 def _partitions(items): if not items: yield [] return first, rest = items[0], items[1:] for p in _partitions(rest): for i in range(len(p)): yield p[:i] + [[first] + p[i]] + p[i + 1:] yield [[first]] + p def _settles(led): b = led.balances() net = {p: b[p]["net"] for p in led.people} for d, c, x in minimal(net, led.people)[0]: net[d] += x net[c] -= x return not any(net.values()) def main(argv=None): ap = argparse.ArgumentParser(description="分摊共享费用并规划最少的结算转账。") ap.add_argument("file", nargs="?", help="CSV账本（'-' = 标准输入）") ap.add_argument("--base", default=None, help="基础货币（默认：'# base:'行或EUR）") ap.add_argument("--rates", default=None, help="USD=0.92,GBP=1.16（1单位 = 基础货币rate单位）") ap.add_argument("--people", default=None, help="逗号分隔的团队成员（严格：拼写错误是错误）") ap.add_argument("--detail", default=None, help="显示一个人的逐行明细") ap.add_argument("--json", action="store_true") ap.add_argument("--self-test", action="store_true") a = ap.parse_args(argv) if a.self_test: return self_test() if not a.file: ap.error("给出CSV账本或--self-test") try: text = sys.stdin.read() if a.file == "-" else open(a.file, encoding="utf-8-sig").read() people = [p.strip() for p in a.people.split(",")] if a.people else None led = Ledger(text, a.base, a.rates, people) out = json.dumps(as_json(led), indent=2) + "\n" if a.json else render(led, a.detail) except (OSError, LedgerError) as e: print("错误：%s" % e, file=sys.stderr) return 2 sys.stdout.write(out) return 0 if __name__ == "__main__": sys.exit(main())