import copy import json import os import subprocess import sys import tempfile import unittest from pathlib import Path from PIL import Image from pptx import Presentation sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from engine import dispatch from exports import export_ppt class ApprovedPPTTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name) self.image = self.root / 'image.png' Image.new('RGB', (600, 400), 'white').save(self.image) self.record = dict(id='2', projectName='项目甲', team='摄影组', name='张三', date='2026-09-13', type='道具耗材', amountText='123.45', note='采购材料', materials=[]) def tearDown(self): self.temporary.cleanup() def material(self, kind, path=None): return dict(kind=kind, path=str(path or self.image), name='测试材料') def export(self, records): destination = self.root / 'approved.pptx' dispatch(dict(operation='approved-ppt', state=dict(approvedRecords=records), destination=str(destination))) return Presentation(destination) def assert_matches_local(self, deck, matches): destination = self.root / 'local.pptx' export_ppt(dict(matches=matches), destination, True) local = Presentation(destination) self.assertEqual(deck.slide_width, local.slide_width) self.assertEqual(deck.slide_height, local.slide_height) self.assertEqual(len(deck.slides), len(local.slides)) for actual, expected in zip(deck.slides, local.slides): self.assertEqual(actual._element.xml, expected._element.xml) self.assertEqual([shape.image.blob for shape in actual.shapes if shape.shape_type == 13], [shape.image.blob for shape in expected.shapes if shape.shape_type == 13]) def local_match(self, invoices=1, payments=1, category='道具耗材'): item = dict(path=str(self.image), ocr=dict(rawText='', dates=['2026-09-13'])) return dict(invoices=[item] * invoices, payments=[item] * payments, category=category) def test_all_materials_and_payment_pagination(self): self.record['materials'] = [self.material('invoice')] + [self.material('payment')] * 5 + [self.material('receipt'), self.material('custom')] deck = self.export([self.record]) self.assert_matches_local(deck, [self.local_match(payments=5)]) pictures = [shape for page in deck.slides for shape in page.shapes if shape.shape_type == 13] self.assertEqual(len(pictures), 6) self.assertTrue(any('请手动粘贴' in shape.text for page in deck.slides for shape in page.shapes if shape.has_text_frame)) def test_multipage_pdf_uses_first_page_like_local(self): pdf = self.root / 'invoice.pdf' first = Image.new('RGB', (300, 200), 'white') second = Image.new('RGB', (300, 200), 'blue') first.save(pdf, save_all=True, append_images=[second]) self.record['materials'] = [self.material('invoice', pdf)] self.assertEqual(len(self.export([self.record]).slides), 1) def test_preserves_selected_order_without_merging_people(self): other = copy.deepcopy(self.record) other.update(id='1', name='李四', date='2026-09-01') blue = self.root / 'blue.png' Image.new('RGB', (600, 400), 'blue').save(blue) self.record['materials'] = [self.material('payment')] other['materials'] = [self.material('payment', blue)] deck = self.export([self.record, other]) self.assertEqual(len(deck.slides), 4) self.assertEqual(deck.slides[0].shapes[0].image.blob, self.image.read_bytes()) self.assertEqual(deck.slides[2].shapes[0].image.blob, blue.read_bytes()) def test_no_added_metadata_and_same_simple_batch_layout(self): self.record['note'] = '采购备注' * 250 self.record['materials'] = [self.material('invoice'), self.material('payment')] deck = self.export([self.record, copy.deepcopy(self.record)]) self.assert_matches_local(deck, [self.local_match(), self.local_match()]) text = '\n'.join(shape.text for page in deck.slides for shape in page.shapes if shape.has_text_frame) for value in ('采购备注', '张三', '项目甲', '已批准', '#2', '2026-09-13', '123.45'): self.assertNotIn(value, text) def test_travel_and_accommodation_use_local_categories(self): self.record['materials'] = [self.material('invoice'), self.material('payment')] for remote, local in [('交通差旅', '交通'), ('住宿费用', '住宿'), ('交通', '交通')]: self.record['type'] = remote self.assert_matches_local(self.export([self.record]), [self.local_match(category=local)]) def test_multiple_invoices_use_local_layout(self): self.record['materials'] = [self.material('invoice')] * 3 + [self.material('payment')] * 2 self.assert_matches_local(self.export([self.record]), [self.local_match(invoices=3, payments=2)]) def test_empty_or_only_extra_materials_fail_without_extra_slides(self): for materials in ([], [self.material('receipt')], [self.material('custom')]): self.record['materials'] = materials with self.assertRaisesRegex(ValueError, '没有发票或付款截图'): self.export([self.record]) def test_same_name_different_accounts_do_not_share_payment_page(self): self.record.update(groupId=1, userId='1', materials=[self.material('invoice'), self.material('payment')]) other = copy.deepcopy(self.record) other.update(userId='2') self.assertEqual(len(self.export([self.record, other]).slides), 6) def test_failure_keeps_existing_output(self): destination = self.root / 'original.pptx' destination.write_bytes(b'original') self.record['materials'] = [self.material('invoice', self.root / 'missing.pdf')] with self.assertRaisesRegex(ValueError, '申请 #2'): dispatch(dict(operation='approved-ppt', state=dict(approvedRecords=[self.record]), destination=str(destination))) self.assertEqual(destination.read_bytes(), b'original') self.assertEqual(list(self.root.glob('.*.pptx')), []) def test_rejects_empty_and_excessive_batches(self): for records in ([], [self.record] * 101): with self.assertRaises(ValueError): export_ppt(dict(approvedRecords=records), self.root / 'invalid.pptx', False) def test_engine_process_protocol(self): binary = os.environ.get('RECEIPT_ENGINE_BINARY') command = [binary] if binary else [sys.executable, str(Path(__file__).resolve().parents[1] / 'engine.py')] self.record['materials'] = [self.material('invoice'), self.material('payment')] destination = self.root / 'process.pptx' request = dict(operation='approved-ppt', state=dict(approvedRecords=[self.record]), destination=str(destination)) process = subprocess.run(command, input=json.dumps(request) + '\n', capture_output=True, text=True, timeout=60) self.assertEqual(process.returncode, 0, process.stdout + process.stderr) events = [json.loads(line) for line in process.stdout.splitlines()] self.assertEqual(events[-1]['event'], 'result') self.assertEqual(events[-1]['result']['destination'], str(destination)) self.assertEqual(len(Presentation(destination).slides), 3) if __name__ == '__main__': unittest.main()