This commit is contained in:
csj
2026-09-17 11:12:57 +08:00
parent 45b6f693d6
commit ced5db5473
11 changed files with 550 additions and 3 deletions
+1
View File
@@ -19,6 +19,7 @@ BINARY="$ROOT/native-engine/dist/receipt-engine-helper.app/Contents/MacOS/receip
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_exports.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_ppt_portrait.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_approved_ppt.py -k engine_process -v
RECEIPT_ENGINE_BINARY="$BINARY" "$PYTHON" -m unittest discover -s tests -p test_match_explanation.py -k engine_process -v
if [ "$SOURCE_HASH" != "$(bash "$ROOT/native-engine/engine-fingerprint.sh")" ]; then
echo "error: 打包期间引擎源码发生变化,请重新构建。" >&2
exit 1
+3
View File
@@ -344,6 +344,8 @@ def payment_total(items):
def enrich(state):
from match_explanation import explain_match
for item in state['invoices'] + state['payments'] + state['photos']:
item['displayAmount'] = float(invoice_amount(item) if item['type'] == 'invoice' else payment_amount(item))
item['sortDate'] = material_date(item)
@@ -354,5 +356,6 @@ def enrich(state):
match['payments'] = [lookup[item['id']] for item in match['payments']]
match['paymentTotal'] = float(payment_total(match['payments']))
match['expenseAmount'] = float(sum(max((Decimal(value) for value in item['ocr']['amounts']), default=Decimal(0)) for item in match['invoices']))
match['explanation'] = explain_match(match)
state['directoryPaymentTotal'] = float(payment_total(state['payments']))
return state
+121
View File
@@ -0,0 +1,121 @@
from domain import common, date_distance, merchant_similarity
def check(title, points, maximum, detail, issue=''):
return dict(title=title, points=points, maximum=maximum, detail=detail,
issue=issue if points < maximum else '')
def values(items, field):
return ''.join(dict.fromkeys(value for item in items for value in item['ocr'][field])) or '未识别'
def missing_fields(invoices, payments, field):
missing = []
if not any(item['ocr'][field] for item in invoices):
missing.append('发票')
if not any(item['ocr'][field] for item in payments):
missing.append('付款截图')
return ''.join(missing)
def date_check(invoices, payments, maximum=15, tiered=True):
distance = date_distance(invoices, payments)
missing = missing_fields(invoices, payments, 'dates')
detail = f"发票日期:{values(invoices, 'dates')};付款日期:{values(payments, 'dates')}"
if missing:
return check('日期接近', 0, maximum, detail + f'{missing}未识别到日期,无法比较;请打开原图核对。', '日期未识别')
partial = 8 if maximum == 15 else 5
points = (maximum if distance <= 3 else partial if distance <= 7 else 0) if tiered else (maximum if distance <= 7 else 0)
rule = f'相差 03 天得 {maximum} 分,47 天得 {partial} 分,超过 7 天不得分。' if tiered else f'相差不超过 7 天得 {maximum} 分,否则不得分。'
return check('日期接近', points, maximum, detail + f'最近相差 {distance} 天。{rule}开票和付款时间可能不同,请核对实际业务日期。',
f'日期相差 {distance}')
def single_checks(match, hundred_point=False):
invoice, payment = match['invoices'][0], match['payments'][0]
invoices, payments = [invoice], [payment]
shared = common(invoice, payment, 'amounts')
amount_max, merchant_max, date_max, order_max, unique_max = (40, 20, 10, 15, 15) if hundred_point else (55, 25, 15, 20, 20)
checks = [check('共同金额', amount_max if shared else 0, amount_max,
'识别到共同金额:' + ''.join('¥' + amount for amount in shared) +
'。此项比较 OCR 提取的金额,不保证它就是价税合计或实付总额,请以原图为准。', '未识别到共同金额')]
similarity = merchant_similarity(invoice, payment)
merchant_points = int(merchant_max * similarity + .5) if similarity >= .35 else 0
missing = missing_fields(invoices, payments, 'merchants')
detail = f"发票商户:{values(invoices, 'merchants')};付款商户:{values(payments, 'merchants')}"
if missing:
detail += f'{missing}未识别到商户,无法比较。请核对原图中的销售方和收款方。'
issue = '商户未识别'
else:
detail += f'最高名称相似度 {int(similarity * 100 + .5)}%;达到 35% 才按相似度 × {merchant_max} 分四舍五入计分。名称简称、收款平台或 OCR 误识别可能影响结果,请核对原图。'
issue = '商户未充分印证'
checks.append(check('商户相似', merchant_points, merchant_max, detail, issue))
checks.append(date_check(invoices, payments, date_max))
orders = common(invoice, payment, 'orderNumbers')
missing = missing_fields(invoices, payments, 'orderNumbers')
detail = f"发票单号:{values(invoices, 'orderNumbers')};付款单号:{values(payments, 'orderNumbers')}"
if orders:
detail += '识别到相同交易单号。'
elif missing:
detail += f'{missing}未识别到交易单号,无法交叉验证;不表示单据错误。'
else:
detail += '未找到相同交易单号,可能是不同平台的编号,请人工核对。'
checks.append(check('交易单号', order_max if orders else 0, order_max, detail, '单号未识别' if missing else '单号不一致'))
unique_reason = next((reason for reason in match['reasons'] if '在本批材料中唯一' in reason), '')
checks.append(check('批次金额唯一', unique_max if unique_reason else 0, unique_max,
unique_reason + ',获得唯一性加分。' if unique_reason else
'自动匹配时,同一金额还出现在其他发票或付款材料中,未获得唯一性加分;请核对商户、日期或单号,避免同金额串单。',
'同金额存在其他材料'))
return checks
def explain_match(match):
if match.get('matchType') != 'auto':
return None
invoices, payments = match['invoices'], match['payments']
reasons = match.get('reasons', [])
title = '单张发票与付款截图'
single_pair = False
if any(reason.startswith('同一行程多人发票合计一致') for reason in reasons):
title = '同一行程多人发票'
checks = [
check('行程及合计金额', 95, 95, '同一行程的多张发票合计与铁路付款材料一致,存在付款凭证,最近日期相差不超过 14 天。此规则基础分为 95 分。'),
check('补充凭证', 5 if len(payments) >= 2 else 0, 5,
'已归入至少两份相关付款/订单材料。' if len(payments) >= 2 else
'目前仅有一份相关付款材料,缺少第二份订单/付款材料的交叉印证,因此未获得额外 5 分。', '缺少补充凭证')
]
elif any(reason.startswith('多张发票合计一致') for reason in reasons):
title = '多张发票合并付款'
same_route = '车次、路线及行程日期一致' in reasons
checks = [
check('唯一合计组合', 70, 70, '找到唯一的多张发票组合,其合计与付款金额一致,基础得 70 分。'),
check('补充凭证', 10 if len(payments) >= 2 else 0, 10,
'至少两份相关付款/订单材料相互印证。' if len(payments) >= 2 else
'只有一份付款材料,缺少其他相关订单/付款材料印证,未获得 10 分。', '缺少补充凭证'),
date_check(invoices, payments, 10, tiered=False),
check('同一行程', 10 if same_route else 0, 10,
'已识别为出行材料,提取的车次、路线及行程日期相同。' if same_route else
'未满足“所有发票均识别为出行材料且路线、时间、车次相同”的条件,未获得 10 分;非出行类报销不适用此加分项,并不代表配对错误。',
'未获得同一行程加分')
]
elif any(reason.startswith('多笔付款合计一致') for reason in reasons):
title = '一张发票分多笔付款'
checks = [
check('唯一分笔付款组合', 95, 95, '多笔付款合计与发票中提取的某一金额一致,且组合唯一;原匹配规则固定为 95 分。'),
check('规则保留分', 0, 5, '该规则没有逐笔验证商户、日期和单号,固定保留 5 分供人工复核;不是少匹配了 5% 的金额或材料。', '分笔付款需人工复核')
]
elif len(invoices) == 1 and len(payments) == 1 and any(reason.startswith('金额一致 ¥') for reason in reasons):
single_pair = True
checks = single_checks(match)
else:
return dict(version=2, title='历史匹配记录', rawScore=match['score'], checks=[],
note='这条记录没有足够的原始评分依据,无法准确还原各项得分;保留原评分,不推测扣分原因。可查看下方原匹配依据并人工核对。')
raw_score = sum(item['points'] for item in checks)
if min(100, raw_score) != match['score']:
return dict(version=2, title=title, rawScore=match['score'], checks=[],
note='当前识别信息与历史评分不一致,无法准确还原各项得分;保留原评分及配对结果,请结合下方原匹配依据核对。')
if single_pair:
checks = single_checks(match, hundred_point=True)
return dict(version=2, title=title, rawScore=sum(item['points'] for item in checks), checks=checks,
note='每项满分相加恰好为 100 分,实际得分相加就是下方百分比;各项未得分相加就是距满分的差额。这是证据评分,不是金额或材料的匹配比例,也不是匹配成功概率;即使满分也请核对原始单据。')
@@ -0,0 +1,186 @@
import copy
import json
import os
import subprocess
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from domain import auto_match, enrich
from match_explanation import explain_match, single_checks
from test_domain import material, train, workspace
class MatchExplanationTests(unittest.TestCase):
def pair(self, dates=None, merchants=None, content=''):
return workspace(
[material('invoice', 'invoice', '100.00', content, dates, merchants)],
[material('payment', 'payment', '100.00', content, dates, merchants)])
def explained(self, state):
return enrich(auto_match(state))['matches'][0]
def assert_scores(self, match, score, evidence_score=None):
explanation = match['explanation']
self.assertEqual(match['score'], score)
self.assertTrue(explanation['checks'])
self.assertEqual(sum(item['points'] for item in explanation['checks']), explanation['rawScore'])
self.assertEqual(explanation['version'], 2)
self.assertEqual(sum(item['maximum'] for item in explanation['checks']), 100)
self.assertEqual(explanation['rawScore'], score if evidence_score is None else evidence_score)
self.assertTrue(all(0 <= item['points'] <= item['maximum'] for item in explanation['checks']))
self.assertEqual(sum(item['maximum'] - item['points'] for item in explanation['checks']), 100 - explanation['rawScore'])
def test_90_missing_merchant_and_order_not_amount(self):
match = self.explained(self.pair(dates=['2026-09-01']))
self.assert_scores(match, 90, 65)
checks = match['explanation']['checks']
self.assertEqual([item['points'] for item in checks], [40, 0, 10, 0, 15])
self.assertEqual([item['maximum'] for item in checks], [40, 20, 10, 15, 15])
self.assertEqual(checks[1]['issue'], '商户未识别')
self.assertEqual(checks[3]['issue'], '单号未识别')
self.assertIn('不是金额', match['explanation']['note'])
def test_75_missing_dates_and_merchants(self):
match = self.explained(self.pair())
self.assert_scores(match, 75, 55)
self.assertEqual(match['explanation']['checks'][2]['issue'], '日期未识别')
self.assertIn('发票和付款截图', match['explanation']['checks'][2]['detail'])
def test_70_nonunique_with_rejected_alternative(self):
state = self.pair(dates=['2026-09-01'])
state['payments'].append(material('other', 'payment', '100.00'))
match = self.explained(state)
self.assert_scores(match, 70, 50)
self.assertEqual(match['explanation']['checks'][-1]['points'], 0)
self.assertIn('其他', match['explanation']['checks'][-1]['detail'])
def test_old_capped_100_is_now_75_when_date_and_order_missing(self):
match = self.explained(self.pair(merchants=['海棠餐厅']))
self.assert_scores(match, 100, 75)
self.assertEqual(match['explanation']['checks'][2]['points'], 0)
self.assertIn('每项', match['explanation']['note'])
def test_all_fields_total_exactly_100_without_cap(self):
match = self.explained(self.pair(['2026-09-01'], ['海棠餐厅'], '订单号 ORDER12345678'))
self.assert_scores(match, 100)
self.assertEqual(match['explanation']['rawScore'], 100)
def test_date_tiers_and_one_sided_missing(self):
for dates, points, score in [(['2026-09-04'], 10, 90), (['2026-09-05'], 5, 83),
(['2026-09-09'], 0, 75), ([], 0, 75)]:
with self.subTest(dates=dates):
state = self.pair(['2026-09-01'])
state['payments'][0]['ocr']['dates'] = dates
match = self.explained(state)
self.assert_scores(match, score, 55 + points)
self.assertEqual(match['explanation']['checks'][2]['points'], points)
if not dates:
self.assertIn('付款截图未识别', match['explanation']['checks'][2]['detail'])
def test_merchant_partial_and_below_threshold(self):
for merchant, old_points, points in [('海棠餐饮', 13, 10), ('其他商户', 0, 0)]:
state = self.pair(merchants=['海棠餐厅'])
state['payments'][0]['ocr']['merchants'] = [merchant]
match = self.explained(state)
self.assert_scores(match, 75 + old_points, 55 + points)
self.assertIn('35%', match['explanation']['checks'][1]['detail'])
self.assertIn(merchant, match['explanation']['checks'][1]['detail'])
def test_different_order_numbers(self):
state = self.pair(content='订单号 ORDER12345678')
state['payments'][0]['ocr']['orderNumbers'] = ['OTHER12345678']
match = self.explained(state)
self.assert_scores(match, 75, 55)
self.assertEqual(match['explanation']['checks'][3]['issue'], '单号不一致')
def test_shared_trip_95_and_100(self):
for count in [1, 2]:
state = workspace([train('first', '张三'), train('second', '李四')],
[material(f'payment-{index}', 'payment', '2116.00', '12306 支付成功', ['2026-06-24'])
for index in range(count)])
match = self.explained(state)
self.assert_scores(match, 95 if count == 1 else 100)
self.assertEqual(match['explanation']['checks'][1]['points'], 0 if count == 1 else 5)
def test_combined_invoices_90_nontravel(self):
state = workspace(
[material('first', 'invoice', '30.00', dates=['2026-09-01']),
material('second', 'invoice', '70.00', dates=['2026-09-01'])],
[material('pay', 'payment', '100.00', '订单号 ORDER12345678', ['2026-09-01']),
material('order', 'payment', '100.00', '订单号 ORDER12345678', ['2026-09-01'])])
match = self.explained(state)
self.assert_scores(match, 90)
self.assertEqual([item['points'] for item in match['explanation']['checks']], [70, 10, 10, 0])
self.assertIn('不适用', match['explanation']['checks'][-1]['detail'])
def test_split_payment_fixed_95_is_not_fabricated_deduction(self):
state = workspace([material('invoice', 'invoice', '100.00')],
[material('first', 'payment', '30.00'), material('second', 'payment', '70.00')])
match = self.explained(state)
self.assert_scores(match, 95)
self.assertEqual(match['explanation']['checks'][-1]['title'], '规则保留分')
self.assertIn('固定', match['explanation']['checks'][-1]['detail'])
def test_legacy_refresh_preserves_pairing_and_historical_uniqueness(self):
state = auto_match(self.pair(['2026-09-01']))
original = copy.deepcopy(state['matches'][0])
state['invoices'].append(material('later', 'invoice', '100.00'))
enriched = enrich(state)
match = enriched['matches'][0]
self.assert_scores(match, 90, 65)
for field in ['id', 'score', 'reasons', 'category']:
self.assertEqual(match[field], original[field])
self.assertEqual([item['id'] for item in match['invoices']], [item['id'] for item in original['invoices']])
self.assertEqual(enrich(enriched), enriched)
def test_previous_explanations_upgrade_without_rematching(self):
for dates, merchants, content, expected in [
([], [], '', 55),
(['2026-09-01'], ['海棠餐厅'], '订单号 ORDER12345678', 100)
]:
state = auto_match(self.pair(dates, merchants, content))
match = state['matches'][0]
checks = single_checks(match)
match['explanation'] = dict(title='旧版评分', rawScore=sum(item['points'] for item in checks),
checks=checks, note='旧版封顶说明')
previous_id, previous_score = match['id'], match['score']
enrich(state)
self.assert_scores(match, previous_score, expected)
self.assertEqual(match['id'], previous_id)
previous = copy.deepcopy(state)
self.assertEqual(enrich(state), previous)
def test_manual_and_unknown_records(self):
match = self.explained(self.pair())
match['matchType'] = 'manual'
self.assertIsNone(explain_match(match))
match['matchType'] = 'auto'
match['reasons'] = ['历史匹配']
self.assertEqual(explain_match(match)['checks'], [])
self.assertIn('不推测', explain_match(match)['note'])
def test_changed_ocr_does_not_invent_historical_scores(self):
match = self.explained(self.pair())
match['invoices'][0]['ocr']['dates'] = ['2026-09-01']
match['payments'][0]['ocr']['dates'] = ['2026-09-01']
self.assertEqual(explain_match(match)['checks'], [])
self.assertIn('不一致', explain_match(match)['note'])
def test_engine_process_backfills_existing_workspace(self):
state = auto_match(self.pair(['2026-09-01']))
engine = Path(__file__).resolve().parents[1] / 'engine.py'
command = [os.environ['RECEIPT_ENGINE_BINARY']] if os.environ.get('RECEIPT_ENGINE_BINARY') else [sys.executable, str(engine)]
result = subprocess.run(command, input=json.dumps(dict(operation='refresh', state=state)),
capture_output=True, text=True, timeout=60)
self.assertEqual(result.returncode, 0, result.stderr)
response = json.loads(result.stdout.splitlines()[-1])
self.assertEqual(response['event'], 'result')
match = response['result']['matches'][0]
self.assert_scores(match, 90, 65)
self.assertEqual(match['id'], state['matches'][0]['id'])
if __name__ == '__main__':
unittest.main()