434 lines
16 KiB
Python
434 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
KOCR v4 — 金蝶 K3 凭证 OCR 填表工具
|
||
小唯 A06 出品 | 自动调参版
|
||
|
||
用法:
|
||
python3 kocr.py # 批量处理所有新照片
|
||
python3 kocr.py --new # 只处理未识别过的照片
|
||
python3 kocr.py --status # 查看处理进度
|
||
python3 kocr.py IMG_001.jpg # 单张处理
|
||
python3 kocr.py --merge # 合并输出排序好的汇总 Excel
|
||
python3 kocr.py --config my.yaml # 指定配置文件
|
||
|
||
特性:
|
||
- 自动图像分析 + 自动调参（按照片亮度/阴影/模糊度选参数）
|
||
- 大小写金额交叉验证（大写转数字比对）
|
||
- PPStructure TableMaster 表格识别（失败回退 DB+SVTR）
|
||
- 交错匹配（dr1, cr1, dr2, cr2...）
|
||
- 多页凭证自动合并
|
||
- 外置配置文件（部署时只需调整 config.yaml）
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import argparse
|
||
import yaml
|
||
import cv2
|
||
|
||
# 项目根目录
|
||
PROJECT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, PROJECT_DIR)
|
||
|
||
from core.recognizer import run_ocr
|
||
from core.parser import K3VoucherParser
|
||
from core.postprocessor import validate_voucher, validate_merged_vouchers, generate_review_file
|
||
from core.excel_writer import fill_template, merge_output, merge_voucher_data
|
||
|
||
|
||
# ======== 路径常量 ========
|
||
DEFAULT_CONFIG = os.path.join(PROJECT_DIR, 'config.yaml')
|
||
BATCH_INPUT_DIR = os.path.join(PROJECT_DIR, 'input')
|
||
BATCH_INPUT_DONE = os.path.join(PROJECT_DIR, 'input/done')
|
||
BATCH_OUTPUT_DIR = os.path.join(PROJECT_DIR, 'output')
|
||
INDEX_FILE = os.path.join(PROJECT_DIR, '.kocr_index.json')
|
||
|
||
|
||
# ======== 配置加载 ========
|
||
def load_config(config_path=None):
|
||
"""加载 YAML 配置，支持命令行覆盖"""
|
||
path = config_path or DEFAULT_CONFIG
|
||
if not os.path.exists(path):
|
||
print(f"⚠️ 配置文件不存在: {path}，使用默认配置")
|
||
return _default_config()
|
||
|
||
with open(path, 'r', encoding='utf-8') as f:
|
||
config = yaml.safe_load(f)
|
||
return config
|
||
|
||
|
||
def _default_config():
|
||
"""默认配置（当 config.yaml 不存在时使用）"""
|
||
return {
|
||
'preprocessing': {
|
||
'max_side': 1600,
|
||
'denoise': True,
|
||
'deskew': True,
|
||
'remove_shadow': True,
|
||
'sharpen': True,
|
||
},
|
||
'ocr': {
|
||
'use_gpu': False,
|
||
'lang': 'ch',
|
||
'show_log': False,
|
||
'use_angle_cls': True,
|
||
'det_db_thresh': 0.3,
|
||
'det_db_box_thresh': 0.5,
|
||
'det_db_unclip_ratio': 2.2,
|
||
'max_side_len': 1600,
|
||
'det_db_score_mode': 'slow',
|
||
'min_size': 8,
|
||
'rec_drop_score': 0.5,
|
||
},
|
||
'postprocessing': {
|
||
'amount_balance_tolerance': 0.01,
|
||
},
|
||
'paths': {
|
||
'input_dir': 'input',
|
||
'output_dir': 'output',
|
||
'done_dir': 'input/done',
|
||
'template': '~/.hermes/cache/documents/doc_704eac18442d_凭证模板.xls',
|
||
},
|
||
'photo_analysis': {
|
||
'brightness_thresholds': {'flash': 200, 'dark': 60},
|
||
'blur_threshold': 100,
|
||
'shadow_threshold': 0.3,
|
||
},
|
||
}
|
||
|
||
|
||
# ======== 索引管理 ========
|
||
def load_index():
|
||
if os.path.exists(INDEX_FILE):
|
||
try:
|
||
with open(INDEX_FILE, 'r', encoding='utf-8') as f:
|
||
return yaml.safe_load(f) or {}
|
||
except:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def save_index(idx):
|
||
os.makedirs(os.path.dirname(INDEX_FILE), exist_ok=True)
|
||
with open(INDEX_FILE, 'w', encoding='utf-8') as f:
|
||
yaml.dump(idx, f, allow_unicode=True, default_flow_style=False)
|
||
|
||
|
||
def mark_processed(filename, voucher_info, output_path):
|
||
idx = load_index()
|
||
idx[filename] = {
|
||
'voucher_word': voucher_info.get('voucher_word', '记'),
|
||
'voucher_no': voucher_info.get('voucher_no'),
|
||
'date': voucher_info.get('date').strftime('%Y-%m-%d') if voucher_info.get('date') else '',
|
||
'entries': len(voucher_info.get('entries', [])),
|
||
'total_debit': round(sum(e.get('debit', 0) or 0 for e in voucher_info.get('entries', [])), 2),
|
||
'total_credit': round(sum(e.get('credit', 0) or 0 for e in voucher_info.get('entries', [])), 2),
|
||
'output': os.path.basename(output_path) if output_path else '',
|
||
'processed_at': __import__('datetime').datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
}
|
||
save_index(idx)
|
||
|
||
|
||
def is_processed(filename):
|
||
idx = load_index()
|
||
return filename in idx
|
||
|
||
|
||
# ======== 单张凭证处理 ========
|
||
def process_single(image_path, config, output_path=None, verbose=False, mark_done=True):
|
||
"""处理单张凭证照片
|
||
|
||
流程: 原图 → PaddleOCR → 解析 → 填Excel → 验证
|
||
|
||
Returns:
|
||
(voucher_data, quality)
|
||
"""
|
||
filename = os.path.basename(image_path)
|
||
|
||
if verbose:
|
||
print(f"\n📷 {filename}")
|
||
|
||
# 1. 读取图像
|
||
img = cv2.imread(image_path)
|
||
if img is None:
|
||
raise ValueError(f"无法读取图片: {image_path}")
|
||
|
||
# 2. 原图直出 OCR（无预处理）
|
||
ocr_config = config.get('ocr', {})
|
||
|
||
if verbose:
|
||
print(f" 📋 OCR 识别中...")
|
||
lines = run_ocr(img, ocr_config)
|
||
|
||
if verbose:
|
||
print(f" OCR: {len(lines)} 行文字")
|
||
|
||
# 3. 解析
|
||
parser = K3VoucherParser(lines)
|
||
voucher = parser.parse()
|
||
|
||
if verbose:
|
||
print(f" 公司: {voucher['company']}")
|
||
print(f" 日期: {voucher['date'].strftime('%Y-%m-%d') if voucher['date'] else '未识别'}")
|
||
print(f" 凭证: {voucher['voucher_word']}-{voucher['voucher_no']}")
|
||
print(f" 分录: {len(voucher['entries'])} 行")
|
||
for e in voucher['entries']:
|
||
dr = f"借 {e['debit']:.2f}" if e['debit'] else ""
|
||
cr = f"贷 {e['credit']:.2f}" if e['credit'] else ""
|
||
print(f" {e['account_code']} {e['account_name']:20s} {dr:>12s} {cr:>12s}")
|
||
|
||
# 4. 大小写金额交叉验证
|
||
if voucher['total_str'] and voucher['total_amount'] > 0:
|
||
from core.postprocessor import amount_check
|
||
total_dr = sum(e.get('debit', 0) or 0 for e in voucher['entries'])
|
||
total_cr = sum(e.get('credit', 0) or 0 for e in voucher['entries'])
|
||
total_max = max(total_dr, total_cr)
|
||
if total_max > 0:
|
||
cn_part = __import__('re').sub(r'[\\d,.\\s￥¥]', '', voucher['total_str'])
|
||
if cn_part:
|
||
ok = amount_check(total_max, cn_part)
|
||
if verbose:
|
||
status = '✅' if ok else '❌'
|
||
print(f" 大小写校验: {status} 小写={total_max:.2f} 大写={cn_part}")
|
||
|
||
# 5. 填 Excel
|
||
if output_path is None:
|
||
basename = os.path.splitext(filename)[0]
|
||
output_path = os.path.join(BATCH_OUTPUT_DIR, f'{basename}_K3.xls')
|
||
|
||
paths_config = config.get('paths', {})
|
||
template = os.path.expanduser(paths_config.get('template', '~/.hermes/cache/documents/doc_704eac18442d_凭证模板.xls'))
|
||
|
||
fill_template(voucher, template, output_path, verbose=verbose)
|
||
|
||
# 6. 标记已处理
|
||
if mark_done:
|
||
mark_processed(filename, voucher, output_path)
|
||
|
||
return voucher, None
|
||
|
||
|
||
# ======== 状态报告 ========
|
||
def show_status(config):
|
||
paths_config = config.get('paths', {})
|
||
input_dir = os.path.expanduser(paths_config.get('input_dir', 'input'))
|
||
done_dir = os.path.expanduser(paths_config.get('done_dir', 'input/done'))
|
||
|
||
idx = load_index()
|
||
|
||
exts = ('.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif', '.webp')
|
||
pending_files = sorted([f for f in os.listdir(input_dir) if f.lower().endswith(exts)]) if os.path.exists(input_dir) else []
|
||
done_files = sorted([f for f in os.listdir(done_dir) if f.lower().endswith(exts)]) if os.path.exists(done_dir) else []
|
||
|
||
processed = set(idx.keys())
|
||
|
||
print("📊 KOCR v4 处理状态")
|
||
print(f" {'='*40}")
|
||
print(f" 索引记录: {len(processed)} 张已处理")
|
||
print(f" input/ 待处理: {len(pending_files)} 张")
|
||
print(f" done/ 已归档: {len(done_files)} 张")
|
||
print()
|
||
|
||
if idx:
|
||
print(" 最近处理的:")
|
||
sorted_items = sorted(
|
||
idx.items(),
|
||
key=lambda x: (x[1].get('date', ''), x[1].get('voucher_no', 0)),
|
||
reverse=True
|
||
)[:10]
|
||
for fname, info in sorted_items:
|
||
v = f"{info.get('voucher_word', '记')}-{info.get('voucher_no', '?')}"
|
||
d = info.get('date', '')
|
||
e = info.get('entries', 0)
|
||
t = info.get('processed_at', '')
|
||
print(f" {fname:30s} → {v:>8s} {d} {e}笔分录 {t}")
|
||
|
||
if pending_files:
|
||
new_pending = [f for f in pending_files if f not in processed]
|
||
if new_pending:
|
||
print(f"\n 📋 未处理的新照片 ({len(new_pending)} 张):")
|
||
for f in new_pending[:10]:
|
||
print(f" {f}")
|
||
if len(new_pending) > 10:
|
||
print(f" ... 还有 {len(new_pending)-10} 张")
|
||
|
||
print()
|
||
|
||
|
||
# ======== 主入口 ========
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='KOCR v4 — 金蝶 K3 凭证 OCR 填表工具')
|
||
parser.add_argument('input', nargs='?', default=None,
|
||
help='凭证照片路径（留空则批量处理 input 目录）')
|
||
parser.add_argument('--output', '-o', default=None, help='输出 Excel 路径')
|
||
parser.add_argument('--config', '-c', default=DEFAULT_CONFIG, help='配置文件路径')
|
||
parser.add_argument('--verbose', '-v', action='store_true', help='显示详细结果')
|
||
parser.add_argument('--status', '-s', action='store_true', help='显示处理进度')
|
||
parser.add_argument('--new', '-n', action='store_true', help='只处理未识别过的新照片')
|
||
parser.add_argument('--archive', '-a', action='store_true', help='归档已处理的照片到 done/')
|
||
parser.add_argument('--review', '-r', action='store_true', help='生成数据校验报告')
|
||
parser.add_argument('--merge', '-m', action='store_true', help='合并输出汇总 Excel')
|
||
parser.add_argument('--force', '-f', action='store_true', help='强制重新处理所有图片')
|
||
args = parser.parse_args()
|
||
|
||
# 加载配置
|
||
config = load_config(args.config)
|
||
|
||
# 路径
|
||
paths_config = config.get('paths', {})
|
||
input_dir = os.path.expanduser(paths_config.get('input_dir', BATCH_INPUT_DIR))
|
||
output_dir = os.path.expanduser(paths_config.get('output_dir', BATCH_OUTPUT_DIR))
|
||
done_dir = os.path.expanduser(paths_config.get('done_dir', BATCH_INPUT_DONE))
|
||
template_path = os.path.expanduser(paths_config.get('template', '~/.hermes/cache/documents/doc_704eac18442d_凭证模板.xls'))
|
||
|
||
# 确保目录存在
|
||
for d in [input_dir, done_dir, output_dir]:
|
||
os.makedirs(d, exist_ok=True)
|
||
|
||
if not os.path.exists(template_path):
|
||
print(f"❌ 找不到模板: {template_path}")
|
||
sys.exit(1)
|
||
|
||
# ---- 状态查看 ----
|
||
if args.status:
|
||
show_status(config)
|
||
return
|
||
|
||
# ---- 归档 ----
|
||
if args.archive:
|
||
idx = load_index()
|
||
if not os.path.exists(input_dir):
|
||
print("📭 input 目录不存在")
|
||
return
|
||
moved = 0
|
||
for fname in os.listdir(input_dir):
|
||
if fname in idx:
|
||
src = os.path.join(input_dir, fname)
|
||
dst = os.path.join(done_dir, fname)
|
||
os.rename(src, dst)
|
||
moved += 1
|
||
print(f"📦 已归档 {moved} 张照片到 input/done/")
|
||
return
|
||
|
||
# ---- 确定文件列表 ----
|
||
input_path = args.input or input_dir
|
||
|
||
if os.path.isdir(input_path):
|
||
# 批量模式
|
||
exts = ('.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.tif', '.webp')
|
||
files = sorted([f for f in os.listdir(input_path) if f.lower().endswith(exts)])
|
||
|
||
if not files:
|
||
print(f"📭 {input_path} 中没有图片文件")
|
||
sys.exit(0)
|
||
|
||
# --new 模式过滤
|
||
idx = load_index()
|
||
if args.new or not args.force:
|
||
new_files = [f for f in files if f not in idx]
|
||
skipped = len(files) - len(new_files)
|
||
files = new_files
|
||
if skipped > 0:
|
||
print(f"⏭️ 跳过 {skipped} 张已处理过的照片")
|
||
if not files:
|
||
print("✅ 全部照片都已处理过了，无需重复识别")
|
||
return
|
||
|
||
print(f"📂 KOCR v4 — 批量处理: {len(files)} 张凭证")
|
||
print(f" 输入: {input_path}")
|
||
print(f" 输出: {output_dir}")
|
||
print(f" {'='*40}")
|
||
|
||
results = []
|
||
for i, fname in enumerate(files):
|
||
fpath = os.path.join(input_path, fname)
|
||
basename = os.path.splitext(fname)[0]
|
||
out_path = os.path.join(output_dir, f'{basename}_K3.xls')
|
||
|
||
print(f" [{i+1}/{len(files)}] {fname}")
|
||
try:
|
||
voucher, quality = process_single(fpath, config, out_path, verbose=args.verbose)
|
||
vno = voucher.get('voucher_no', '?')
|
||
vdate = voucher['date'].strftime('%Y-%m-%d') if voucher.get('date') else '?'
|
||
results.append((fname, voucher, out_path, quality))
|
||
print(f" ✅ ['raw'] {vdate} 记-{vno} → {os.path.basename(out_path)}")
|
||
except Exception as e:
|
||
print(f" ❌ 失败: {e}")
|
||
print()
|
||
|
||
# ---- 合并凭证数据（多页合并）+ 统一校验 ----
|
||
merged_vouchers = merge_voucher_data(results)
|
||
validation_results = validate_merged_vouchers(
|
||
merged_vouchers,
|
||
config.get('postprocessing', {}).get('amount_balance_tolerance', 0.01)
|
||
)
|
||
|
||
clean_count = sum(1 for _, is_clean, _, _ in validation_results if is_clean)
|
||
multi_count = sum(1 for v in merged_vouchers if v.get('is_multi_page'))
|
||
|
||
print(f" {'='*40}")
|
||
print(f" 📋 合并后校验结果（{len(merged_vouchers)} 笔凭证，含 {multi_count} 笔多页）:")
|
||
for voucher, is_clean, issues, summary in validation_results:
|
||
date_str = voucher['date'].strftime('%m-%d') if voucher.get('date') else '??'
|
||
vno = voucher.get('voucher_no', '?')
|
||
multi = '📑' if voucher.get('is_multi_page') else ' '
|
||
entries = len(voucher['entries'])
|
||
status = '✅' if is_clean else '⚠️'
|
||
print(f" {status} {multi} {date_str} 记-{vno} ({entries}条分录)")
|
||
if not is_clean:
|
||
for iss in issues[:3]:
|
||
print(f" {iss}")
|
||
print(f" {'='*40}")
|
||
print(f" ✅ 校验通过: {clean_count} 笔")
|
||
print(f" ⚠️ 需人工复核: {len(validation_results) - clean_count} 笔")
|
||
|
||
# ---- 合并输出 ----
|
||
if args.merge and len(results) > 1:
|
||
merge_path = merge_output(results, template_path, output_dir)
|
||
print(f"\n 📊 已生成: {merge_path}")
|
||
|
||
# ---- 校验报告（基于合并后的数据） ----
|
||
if args.review and results:
|
||
review_results = []
|
||
for voucher, is_clean, issues, summary in validation_results:
|
||
fname = ' + '.join(voucher.get('pages', ['unknown']))
|
||
review_results.append((fname, voucher, '', None))
|
||
review_path = generate_review_file(review_results, output_dir)
|
||
print(f" 📋 校验报告: {os.path.basename(review_path)}")
|
||
|
||
# ---- 自动归档 ----
|
||
archived = 0
|
||
for fname, voucher, out_path, quality in results:
|
||
src = os.path.join(input_path, fname)
|
||
dst = os.path.join(done_dir, fname)
|
||
if os.path.exists(src):
|
||
os.rename(src, dst)
|
||
archived += 1
|
||
if archived > 0:
|
||
print(f" 📦 已归档 {archived} 张照片到 done/")
|
||
|
||
print(f"🎉 完成! {len(results)}/{len(files)} 成功")
|
||
|
||
else:
|
||
# 单文件模式
|
||
if not os.path.exists(input_path):
|
||
print(f"❌ 找不到图片: {input_path}")
|
||
sys.exit(1)
|
||
|
||
if args.output:
|
||
out_path = args.output
|
||
else:
|
||
basename = os.path.splitext(os.path.basename(input_path))[0]
|
||
out_path = os.path.join(output_dir, f'{basename}_K3.xls')
|
||
|
||
voucher, quality = process_single(input_path, config, out_path, verbose=args.verbose)
|
||
results = [(os.path.basename(input_path), voucher, out_path, quality)]
|
||
|
||
if args.review:
|
||
review_path = generate_review_file(results, output_dir)
|
||
print(f" 📋 校验报告: {os.path.basename(review_path)}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|