269 lines
8.3 KiB
Python
269 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
||
"""KOCR v4 — 后处理模块
|
||
|
||
包含：字符纠错、大写金额转数字、大小写交叉校验、借贷平衡验证
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
|
||
|
||
# ======== 财务字符纠错字典 ========
|
||
FINANCE_CHAR_FIX = {
|
||
"染": "柒", "壶": "壹", "式": "贰", "参": "叁", "伍": "伍",
|
||
"O": "0", "o": "0", "l": "1", "I": "1", "Z": "2", "z": "2",
|
||
"S": "5", "s": "5", "B": "8", "b": "8", "〇": "零"
|
||
}
|
||
|
||
|
||
def char_correct(text: str) -> str:
|
||
"""财务字符纠错"""
|
||
for wrong, right in FINANCE_CHAR_FIX.items():
|
||
text = text.replace(wrong, right)
|
||
return text
|
||
|
||
|
||
# ======== 大写金额转数字 ========
|
||
|
||
CN_NUM = {
|
||
'零': 0, '壹': 1, '贰': 2, '叁': 3, '肆': 4,
|
||
'伍': 5, '陆': 6, '柒': 7, '捌': 8, '玖': 9,
|
||
'无': 0, '正': 0, # 易误读的常见替换
|
||
}
|
||
|
||
CN_UNIT = {
|
||
'拾': 10, '佰': 100, '仟': 1000,
|
||
'万': 10000, '亿': 100000000,
|
||
'元': 1, '角': 0.1, '分': 0.01,
|
||
'整': 0,
|
||
}
|
||
|
||
|
||
def chinese_amount_to_number(cn_str: str) -> float:
|
||
"""大写金额转阿拉伯数字
|
||
|
||
支持格式：
|
||
- 壹万贰仟叁佰肆拾伍元陆角柒分 → 12345.67
|
||
- 壹仟贰佰叁拾肆元整 → 1234.00
|
||
- 拾元整 → 10.00
|
||
- 壹佰贰拾叁元零角零分 → 123.00
|
||
|
||
Args:
|
||
cn_str: 中文大写金额字符串
|
||
|
||
Returns:
|
||
转换后的数字，失败返回 0.0
|
||
"""
|
||
if not cn_str:
|
||
return 0.0
|
||
|
||
# 清理字符串
|
||
cn_str = cn_str.strip().replace(' ', '').replace('（', '').replace('）', '')
|
||
|
||
# 去掉前缀（如 "合计："）
|
||
for prefix in ['合计', '合计大写', '人民币', 'RMB', '￥', '¥']:
|
||
if cn_str.startswith(prefix):
|
||
cn_str = cn_str[len(prefix):]
|
||
break
|
||
|
||
total = 0.0
|
||
temp = 0.0 # 当前节的值（万以下）
|
||
|
||
i = 0
|
||
while i < len(cn_str):
|
||
char = cn_str[i]
|
||
|
||
if char in CN_NUM:
|
||
temp = CN_NUM[char] # 暂存数字
|
||
elif char in CN_UNIT:
|
||
unit = CN_UNIT[char]
|
||
|
||
if char in ('万', '亿'):
|
||
# 万/亿：整节结算
|
||
total = (total + temp) * unit if temp > 0 else total * unit
|
||
temp = 0.0
|
||
elif char == '元':
|
||
# 元：加整数部分
|
||
total += temp if temp > 0 else 0
|
||
temp = 0.0
|
||
elif char in ('角', '分'):
|
||
# 角/分：加小数部分
|
||
total += temp * unit
|
||
temp = 0.0
|
||
else:
|
||
# 拾/佰/仟：节内进位
|
||
if temp == 0:
|
||
temp = 1 # "拾元" → 10元
|
||
total += temp * unit
|
||
temp = 0.0
|
||
elif char == '零':
|
||
# 零：跳过，继续
|
||
if i + 1 < len(cn_str) and cn_str[i + 1] in CN_UNIT:
|
||
# 零后面有单位 → 如"壹万零伍佰"
|
||
pass
|
||
temp = 0
|
||
elif char == '整':
|
||
# 整：结束
|
||
break
|
||
|
||
i += 1
|
||
|
||
# 处理末尾剩余
|
||
if temp > 0:
|
||
total += temp
|
||
|
||
return round(total, 2)
|
||
|
||
|
||
def amount_check(amount_num: float, amount_cn: str) -> bool:
|
||
"""大小写金额交叉验证
|
||
|
||
检查 OCR 识别的阿拉伯数字合计与中文大写合计是否匹配
|
||
|
||
Args:
|
||
amount_num: 阿拉伯数字合计（从 OCR 金额列解析）
|
||
amount_cn: 中文大写合计（从 OCR 文本栏解析）
|
||
|
||
Returns:
|
||
True if 差值 ≤ 0.01
|
||
"""
|
||
if not amount_cn or amount_num <= 0:
|
||
return True # 没有大写则跳过校验
|
||
|
||
cn_num = chinese_amount_to_number(amount_cn)
|
||
if cn_num <= 0:
|
||
return True # 大写解析失败，跳过
|
||
|
||
diff = abs(amount_num - cn_num)
|
||
return diff <= 0.01
|
||
|
||
|
||
# ======== 数据验证 ========
|
||
|
||
def validate_voucher(voucher_data: dict, balance_tolerance: float = 0.01) -> tuple:
|
||
"""验证 OCR 识别结果的完整性和正确性
|
||
|
||
Args:
|
||
voucher_data: K3VoucherParser.parse() 的输出
|
||
balance_tolerance: 借贷平衡允许误差（元）
|
||
|
||
Returns:
|
||
(is_clean, issues, summary_text)
|
||
"""
|
||
issues = []
|
||
warnings = []
|
||
entries = voucher_data.get('entries', [])
|
||
|
||
if not entries:
|
||
return False, ['❌ 未识别到任何分录'], '无分录'
|
||
|
||
# 1. 借贷平衡检查
|
||
total_dr = sum(e.get('debit', 0) or 0 for e in entries)
|
||
total_cr = sum(e.get('credit', 0) or 0 for e in entries)
|
||
balance_diff = abs(total_dr - total_cr)
|
||
|
||
if balance_diff > balance_tolerance:
|
||
issues.append(f'❌ 借贷不平衡: 借 {total_dr:.2f} ≠ 贷 {total_cr:.2f} (差 {balance_diff:.2f})')
|
||
|
||
# 2. 大小写金额交叉验证（如有可能）
|
||
total_str = voucher_data.get('total_str', '')
|
||
if total_str and total_dr > 0:
|
||
# 从合计字符串中提取大写金额
|
||
cn_part = re.sub(r'[\d,.\s￥¥]', '', total_str)
|
||
if cn_part:
|
||
cn_ok = amount_check(total_dr, cn_part)
|
||
if not cn_ok:
|
||
cn_val = chinese_amount_to_number(cn_part)
|
||
issues.append(f'❌ 大小写金额不匹配: 小写 {total_dr:.2f} ≠ 大写 {cn_part}(={cn_val:.2f})')
|
||
else:
|
||
warnings.append(f'✅ 大小写金额验证通过')
|
||
|
||
# 3. 金额合理性检查
|
||
for e in entries:
|
||
for side in ['debit', 'credit']:
|
||
val = e.get(side, 0) or 0
|
||
if val <= 0:
|
||
continue
|
||
if val > 99999999:
|
||
issues.append(f'⚠️ 金额异常过大: {e["account_code"]} {side}={val:.2f}')
|
||
if val < 0.01 and val > 0:
|
||
warnings.append(f' 金额异常小: {e["account_code"]} {side}={val:.2f}')
|
||
|
||
# 4. 科目代码检查
|
||
for e in entries:
|
||
code = e.get('account_code', '')
|
||
if not code or len(code) < 4:
|
||
warnings.append(f' 科目代码异常: {code} {e.get("account_name","")}')
|
||
|
||
# 构建摘要
|
||
is_clean = len(issues) == 0
|
||
summary_lines = []
|
||
if is_clean and not warnings:
|
||
summary_lines.append('✅ 验证通过（借贷平衡 + 金额合理）')
|
||
else:
|
||
summary_lines.extend(issues)
|
||
summary_lines.extend(warnings)
|
||
|
||
return is_clean, issues + warnings, '\n'.join(summary_lines)
|
||
|
||
|
||
def generate_review_file(results: list, output_dir: str) -> str:
|
||
"""生成批量处理的校验报告"""
|
||
from datetime import datetime
|
||
|
||
review_path = os.path.join(output_dir, '校验报告.txt')
|
||
lines = []
|
||
lines.append('=' * 50)
|
||
lines.append('KOCR v4 数据校验报告')
|
||
lines.append(f'生成时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
|
||
lines.append(f'处理张数: {len(results)}')
|
||
lines.append('=' * 50)
|
||
lines.append('')
|
||
|
||
all_clean = True
|
||
for fname, voucher, out_path, quality in results:
|
||
is_clean, issues, summary = validate_voucher(voucher)
|
||
if not is_clean:
|
||
all_clean = False
|
||
date_str = voucher['date'].strftime('%Y-%m-%d') if voucher.get('date') else '?'
|
||
vno = voucher.get('voucher_no', '?')
|
||
lines.append(f'📷 {fname}')
|
||
lines.append(f' 凭证: 记-{vno} 日期: {date_str}')
|
||
if quality:
|
||
lines.append(f' 类型: {quality.photo_type} 亮度:{quality.brightness} 模糊:{quality.blur_score}')
|
||
lines.append(f' {summary}')
|
||
lines.append('')
|
||
|
||
if all_clean:
|
||
lines.append('🎉 全部凭证验证通过')
|
||
else:
|
||
lines.append('⚠️ 存在需要人工复核的数据，请对照原始凭证核对')
|
||
|
||
lines.append('=' * 50)
|
||
|
||
with open(review_path, 'w', encoding='utf-8') as f:
|
||
f.write('\n'.join(lines))
|
||
|
||
return review_path
|
||
|
||
|
||
|
||
def validate_merged_vouchers(merged_list, balance_tolerance=0.01):
|
||
"""对合并后的凭证列表做完整校验
|
||
|
||
每笔凭证（多页已合并）统一做借贷平衡 + 大小写金额验证。
|
||
这才是最终的业务校验结果。
|
||
|
||
Args:
|
||
merged_list: merge_voucher_data() 的输出
|
||
balance_tolerance: 借贷平衡容差
|
||
|
||
Returns:
|
||
[(voucher, is_clean, issues, summary_text), ...]
|
||
"""
|
||
results = []
|
||
for voucher in merged_list:
|
||
is_clean, issues, summary = validate_voucher(voucher, balance_tolerance)
|
||
results.append((voucher, is_clean, issues, summary))
|
||
return results
|