Initial commit: macOS receipt workspace

2026-09-11 12:58:09 +08:00
commit bbacd1d6f3
40 changed files with 2790 additions and 0 deletions
@@ -0,0 +1,63 @@
+import argparse
+import json
+import subprocess
+import sys
+import shutil
+from pathlib import Path
+from PIL import Image
+
+
+def invoke(command, request):
+    process = subprocess.run(command, input=json.dumps(request, ensure_ascii=False) + '\n', capture_output=True, text=True, timeout=180)
+    events = [json.loads(line) for line in process.stdout.splitlines() if line.strip()]
+    errors = [event for event in events if event['event'] == 'error']
+    if process.returncode or errors:
+        raise RuntimeError(str(errors) + '\n' + process.stderr)
+    results = [event['result'] for event in events if event['event'] == 'result']
+    if len(results) != 1:
+        raise AssertionError('处理引擎没有返回唯一结果')
+    return results[0]
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--binary')
+    parser.add_argument('--output', required=True)
+    arguments = parser.parse_args()
+    root = Path(__file__).resolve().parents[3]
+    output = Path(arguments.output).resolve()
+    output.mkdir(parents=True, exist_ok=True)
+    engine = Path(__file__).resolve().parents[1] / 'engine.py'
+    command = [arguments.binary] if arguments.binary else [sys.executable, str(engine)]
+    state = invoke(command, dict(operation='scan', path=str(root / 'fapiao/demo-materials'), workspacePath=str(output / 'materials'), folderName='演示报账材料'))
+    assert len(state['invoices']) == 2
+    assert len(state['payments']) == 2
+    assert len(state['photos']) == 1
+    assert len(state['matches']) == 2
+    assert {match['category'] for match in state['matches']} == {'餐饮', '住宿'}
+    assert state['directoryPaymentTotal'] == 596.5
+    assert sum(match['expenseAmount'] for match in state['matches']) == 596.5
+    assert not state['warnings']
+    (output / 'state.json').write_text(json.dumps(state, ensure_ascii=False, indent=2))
+    for operation, extension in [('ppt', 'pptx'), ('expense', 'xlsx')]:
+        destination = output / ('演示报账材料.' + extension)
+        invoke(command, dict(operation=operation, state=state, destination=str(destination), classified=True, purposes={match['id']: match['category'] + '费用' for match in state['matches']}, signatures=['经办人', '财务']))
+        assert destination.stat().st_size > 1000
+    pdf_input = output / 'pdf-input'
+    (pdf_input / '发票').mkdir(parents=True, exist_ok=True)
+    (pdf_input / '付款截图').mkdir(parents=True, exist_ok=True)
+    with Image.open(root / 'fapiao/demo-materials/发票/住宿_差旅发票_A001.png') as original:
+        original.convert('RGB').save(pdf_input / '发票/两页发票.pdf', save_all=True, append_images=[Image.new('RGB', (800, 500), 'white')])
+    (pdf_input / '发票/损坏文件.pdf').write_bytes(b'not a pdf')
+    shutil.copyfile(root / 'fapiao/demo-materials/付款截图/酒店支付记录_B009.png', pdf_input / '付款截图/付款.png')
+    pdf_state = invoke(command, dict(operation='scan', path=str(pdf_input), workspacePath=str(output / 'pdf-materials')))
+    assert len(pdf_state['invoices']) == 2
+    assert len(pdf_state['matches']) == 1
+    assert pdf_state['matches'][0]['expenseAmount'] == 468
+    assert sum(item['ocr']['status'] == 'failed' for item in pdf_state['invoices']) == 1
+    invoke(command, dict(operation='ppt', state=pdf_state, destination=str(output / 'PDF首页贴票.pptx')))
+    print('完整链路通过：5 份材料、2 组自动核对、总额 596.50；PPT、个人报销单、PDF 首页与损坏文件降级均通过。')
+
+
+if __name__ == '__main__':
+    main()
@@ -0,0 +1,109 @@
+import copy
+import sys
+import unittest
+from decimal import Decimal
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from domain import auto_match, category, combinations, enrich, extract, invoice_amount, payment_total
+
+
+def material(identifier, kind, amount, content='', dates=None, merchants=None):
+    ocr = extract(content.splitlines())
+    ocr.update(amounts=[amount] if isinstance(amount, str) else amount, dates=dates or [], merchants=merchants or [])
+    return dict(id=identifier, name=identifier + '.png', path=identifier, previewPath='', type=kind,
+                size=1, matched=False, ocr=ocr)
+
+
+def workspace(invoices, payments):
+    return dict(rootPath='测试', invoices=invoices, payments=payments, photos=[], matches=[], warnings=[])
+
+
+def train(identifier, traveler):
+    return material(identifier, 'invoice', '1058.00', f'铁路电子客票\n乘车人：{traveler}\n北京南 → 上海虹桥\n2026年06月24日 15:00\nG21次', ['2026-06-24'], ['中国铁路'])
+
+
+class ExtractionTests(unittest.TestCase):
+    def test_train(self):
+        data = extract(['铁路电子客票', '乘车人：张三', '北京南 → 上海虹桥', '2026年08月12日 09:30', 'G101次', '票价 ￥553.00'])
+        self.assertEqual(data['travel'], dict(type='train', travelerName='张三', departure='北京南', destination='上海虹桥', departureTime='2026-08-12 09:30', transportNumber='G101'))
+
+    def test_flight(self):
+        travel = extract(['航空电子客票行程单', '旅客姓名：李四', '出发地：广州', '目的地：成都', '2026-08-15 14:20', '航班号 CZ3401'])['travel']
+        self.assertEqual((travel['type'], travel['travelerName'], travel['departure'], travel['destination'], travel['transportNumber']), ('flight', '李四', '广州', '成都', 'CZ3401'))
+
+    def test_invoice_date_is_not_departure(self):
+        travel = extract(['电子发票（铁路电子客票） 北京市税务局', '发票号码：26119110010005751560 开票日期：2026年06月26日', '北京南站 G21 上海虹桥站', '2026年06月24日 15:00开 17车04C号', '2107021990****0234 王天行', '票价：￥1058.00'])['travel']
+        self.assertEqual(travel['travelerName'], '王天行')
+        self.assertEqual(travel['departureTime'], '2026-06-24 15:00')
+
+    def test_invalid_date_does_not_abort_material(self):
+        self.assertEqual(extract(['2026年02月31日 金额 20'])['dates'], [])
+
+    def test_amount_order(self):
+        self.assertEqual(extract(['票价 ￥1058.00', '支付总额 2116'])['amounts'][0], '2116.00')
+
+
+class MatchingTests(unittest.TestCase):
+    def test_original_many_to_many_fixture(self):
+        state = workspace([train('i1', '王天行'), train('i2', '王超文')], [
+            material('p1', 'payment', ['1058.00', '2116.00'], '12306订单 中国铁路 支付总额 2116', ['2026-06-24']),
+            material('p2', 'payment', '2116.00', '12306消费 中国铁路', ['2026-06-22'])])
+        auto_match(state)
+        self.assertEqual(len(state['matches']), 1)
+        match = state['matches'][0]
+        self.assertEqual((len(match['invoices']), len(match['payments']), match['score'], match['category']), (2, 2, 100, '交通'))
+
+    def test_shared_travel_payment_evidence(self):
+        state = workspace([train('i1', '张三'), train('i2', '李四')], [material('p1', 'payment', '2116.00', '12306 支付成功', ['2026-06-24'])])
+        auto_match(state)
+        self.assertEqual(state['matches'][0]['score'], 95)
+
+    def test_unique_amount(self):
+        state = workspace([material('i', 'invoice', '100.00')], [material('p', 'payment', '100.00')])
+        auto_match(state)
+        self.assertEqual(state['matches'][0]['score'], 75)
+
+    def test_ambiguous_stays_unmatched(self):
+        state = workspace([material('i', 'invoice', '100.00', dates=['2026-08-01'])], [material('p1', 'payment', '100.00', dates=['2026-08-01']), material('p2', 'payment', '100.00', dates=['2026-08-01'])])
+        self.assertFalse(auto_match(state)['matches'])
+
+    def test_high_score_keeps_original_tie_rule(self):
+        state = workspace([material('i', 'invoice', '100.00', dates=['2026-08-01'], merchants=['测试公司'])], [material('p1', 'payment', '100.00', dates=['2026-08-01'], merchants=['测试公司']), material('p2', 'payment', '100.00', dates=['2026-08-01'], merchants=['测试公司'])])
+        self.assertEqual(len(auto_match(state)['matches']), 1)
+
+    def test_split_payment(self):
+        state = workspace([material('i', 'invoice', '100.00')], [material('p1', 'payment', '30.00'), material('p2', 'payment', '70.00')])
+        self.assertEqual(len(auto_match(state)['matches'][0]['payments']), 2)
+
+    def test_ambiguous_split_payment(self):
+        state = workspace([material('i', 'invoice', '100.00')], [material('p1', 'payment', '30.00'), material('p2', 'payment', '70.00'), material('p3', 'payment', '40.00'), material('p4', 'payment', '60.00')])
+        self.assertFalse(auto_match(state)['matches'])
+
+    def test_filename_does_not_match(self):
+        state = workspace([material('same', 'invoice', '100.00')], [material('same', 'payment', '90.00')])
+        self.assertFalse(auto_match(state)['matches'])
+
+    def test_decimal_combinations(self):
+        options = combinations([('first', Decimal('.10')), ('second', Decimal('.20'))], Decimal('.30'))
+        self.assertEqual(options, [['first', 'second']])
+
+    def test_original_categories(self):
+        for content, expected in [('海棠酒店 住宿 房费 客房', '住宿'), ('餐厅 美团外卖 咖啡 饮品', '餐饮'), ('办公用品 打印纸 墨盒 文具', '办公用品'), ('京东商城 采购家具设备', '采购'), ('客户招待 商务宴请', '招待')]:
+            self.assertEqual(category([material('i', 'invoice', '100.00', content)]), expected)
+
+    def test_summary_duplicate(self):
+        self.assertEqual(payment_total([material('order', 'payment', '100.00', '订单详情 支付总额 100.00'), material('proof', 'payment', '100.00', '支付成功 实付金额 100.00')]), Decimal('100.00'))
+
+    def test_invoice_total_labeled(self):
+        self.assertEqual(invoice_amount(material('i', 'invoice', ['200.00', '100.00'], '发票金额 100.00')), Decimal('100.00'))
+
+    def test_enrich_preserves_group_identity(self):
+        state = workspace([material('i', 'invoice', '100.00')], [material('p', 'payment', '100.00')])
+        enrich(auto_match(state))
+        self.assertTrue(state['matches'][0]['invoices'][0]['matched'])
+        self.assertEqual(state['matches'][0]['expenseAmount'], 100)
+
+
+if __name__ == '__main__':
+    unittest.main()
@@ -0,0 +1,95 @@
+import copy
+import tempfile
+import unittest
+import zipfile
+from pathlib import Path
+
+from test_domain import material, workspace
+from domain import auto_match, enrich
+from exports import export_expense, export_ppt, export_travel
+from PIL import Image
+from openpyxl import load_workbook
+from pptx import Presentation
+
+
+class ExportTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        image = self.root / 'receipt.png'
+        Image.new('RGB', (400, 200), 'white').save(image)
+        self.state = enrich(auto_match(workspace([material('i', 'invoice', '100.00')], [material('p', 'payment', '100.00')])))
+        for item in self.state['invoices'] + self.state['payments']:
+            item['previewPath'] = str(image)
+        self.template = Path(__file__).parents[1] / 'personal-expense-template.xlsx'
+
+    def tearDown(self):
+        self.temporary.cleanup()
+
+    def test_simple_ppt_with_photo_placeholder(self):
+        destination = self.root / 'test.pptx'
+        export_ppt(self.state, destination, True)
+        deck = Presentation(destination)
+        self.assertEqual(len(deck.slides), 3)
+        self.assertEqual(deck.slide_width, 1280 * 12700)
+        self.assertTrue(any('实物照片' in shape.text for shape in deck.slides[2].shapes if shape.has_text_frame))
+
+    def test_travel_category_excludes_photo(self):
+        self.state['matches'][0]['category'] = '交通'
+        destination = self.root / 'test.pptx'
+        export_ppt(self.state, destination, False)
+        self.assertEqual(len(Presentation(destination).slides), 2)
+
+    def test_complex_payment_pagination(self):
+        self.state['matches'][0]['payments'] *= 5
+        destination = self.root / 'test.pptx'
+        export_ppt(self.state, destination, True)
+        self.assertEqual(len(Presentation(destination).slides), 5)
+
+    def test_expense_template_values_and_formulas(self):
+        destination = self.root / 'expense.xlsx'
+        match = self.state['matches'][0]
+        export_expense(self.state, destination, self.template, {match['id']: '=不是公式'}, ['经办人'])
+        workbook = load_workbook(destination)
+        self.assertEqual(workbook.sheetnames, ['个人报销单'])
+        sheet = workbook.active
+        self.assertEqual(sheet['C9'].value, '=不是公式')
+        self.assertEqual(sheet['C9'].data_type, 's')
+        self.assertEqual(sheet['H9'].value, 100)
+        self.assertEqual(sheet['G9'].value, 1)
+        self.assertEqual(sheet['A30'].value, '经办人：')
+        self.assertEqual(sheet['D30'].value, '')
+        original = load_workbook(self.template)['1个人报销单']
+        formulas = {cell.coordinate: cell.value for row in original for cell in row if cell.data_type == 'f'}
+        self.assertTrue(formulas)
+        for coordinate, value in formulas.items():
+            self.assertEqual(sheet[coordinate].value, value)
+        self.assertEqual(str(sheet.print_area).split('!')[-1], str(original.print_area).split('!')[-1])
+
+    def test_expense_pagination(self):
+        self.state['matches'] = [copy.deepcopy(self.state['matches'][0]) for _ in range(14)]
+        destination = self.root / 'expense.xlsx'
+        export_expense(self.state, destination, self.template, {}, [])
+        workbook = load_workbook(destination)
+        self.assertEqual(workbook.sheetnames, ['个人报销单-1', '个人报销单-2'])
+        self.assertEqual(workbook.worksheets[1]['H9'].value, 100)
+        self.assertEqual(workbook.worksheets[1]['H10'].value, '')
+        self.assertEqual(workbook.worksheets[1]['A30'].value, '部门长：')
+
+    def test_travel_only_verified_invoices(self):
+        travel = self.state['invoices'][0]['ocr']['travel']
+        travel.update(type='train', travelerName='张三', departure='北京南', destination='上海虹桥', departureTime='2026-06-24 15:00', transportNumber='G21')
+        destination = self.root / 'travel.xlsx'
+        export_travel(self.state, destination)
+        sheet = load_workbook(destination).active
+        self.assertEqual(sheet.max_row, 2)
+        self.assertEqual(sheet['C2'].value, '张三')
+        self.assertEqual(sheet['G2'].value, 'G21')
+
+    def test_no_travel_raises(self):
+        with self.assertRaises(ValueError):
+            export_travel(self.state, self.root / 'travel.xlsx')
+
+
+if __name__ == '__main__':
+    unittest.main()