392 lines
15 KiB
Python
392 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""KOCR v5 — 金蝶 K3 凭证解析器（基于 y 坐标区间的垂直布局解析）"""
|
||
|
||
import re
|
||
from datetime import datetime
|
||
|
||
|
||
class K3VoucherParser:
|
||
"""基于 y 坐标区间的 K3 凭证解析器
|
||
|
||
核心思路：不依赖行级匹配，而是按 y 坐标区间划分功能区域：
|
||
- 头部区（y<1200）：公司名、日期、凭证号、摘要
|
||
- 科目区（y≈1200-2400）：科目代码和名称
|
||
- 借方金额区（借方标题 ~ 贷方标题之间）
|
||
- 贷方金额区（贷方标题 ~ 图片底部）
|
||
- 底部区：合计、制单等固定信息
|
||
"""
|
||
|
||
def __init__(self, lines):
|
||
"""lines: [(text, conf, cx, cy, box), ...]"""
|
||
self.lines = sorted(lines, key=lambda l: (l[3], l[2]))
|
||
|
||
def parse(self):
|
||
result = {
|
||
'company': '',
|
||
'date': None,
|
||
'year': None,
|
||
'period': None,
|
||
'voucher_word': '记',
|
||
'voucher_no': None,
|
||
'attachments': 0,
|
||
'entries': [],
|
||
'total_str': '',
|
||
'total_amount': 0,
|
||
'preparer': '',
|
||
'checker': '',
|
||
'cashier': '',
|
||
'handler': '',
|
||
'poster': '',
|
||
}
|
||
|
||
# 1. 找到关键 y 坐标边界
|
||
bounds = self._find_bounds()
|
||
|
||
# 2. 提取头部信息（日期、凭证号等）
|
||
self._extract_header(result, bounds)
|
||
|
||
# 3. 提取科目代码和名称
|
||
codes = self._extract_codes(bounds)
|
||
|
||
# 4. 提取金额（按借方/贷方分类）
|
||
debit_amts, credit_amts = self._extract_amounts(bounds)
|
||
|
||
# 5. 提取摘要
|
||
summary = self._extract_summary(bounds)
|
||
|
||
# 6. 汇总金额并去重
|
||
debit_amts = self._deduplicate_amounts(debit_amts)
|
||
credit_amts = self._deduplicate_amounts(credit_amts)
|
||
|
||
# 7. 按行数匹配金额到科目
|
||
self._match_entries(result, codes, debit_amts, credit_amts, summary, bounds)
|
||
|
||
return result
|
||
|
||
def _find_bounds(self):
|
||
"""找到凭证各区域的 y 坐标边界"""
|
||
bounds = {
|
||
'y_summary_header': None, # "摘要" 标题
|
||
'y_subject_header': None, # "科目" 标题
|
||
'y_debit_header': None, # "借方" 标题
|
||
'y_credit_header': None, # "贷方" 标题
|
||
'y_date': None, # 日期行
|
||
'y_total': None, # 合计行
|
||
}
|
||
|
||
for text, conf, cx, cy, box in self.lines:
|
||
t = text.strip()
|
||
if t == '摘要':
|
||
bounds['y_summary_header'] = cy
|
||
elif t == '科目':
|
||
bounds['y_subject_header'] = cy
|
||
elif t == '借方':
|
||
bounds['y_debit_header'] = cy
|
||
elif t == '贷方':
|
||
bounds['y_credit_header'] = cy
|
||
elif '合计' in t and ('壹' in t or '贰' in t or '参' in t or
|
||
'合计' in t and any(c in t for c in '零壹贰叁肆伍陆柒捌玖拾佰仟万亿')):
|
||
if bounds['y_total'] is None:
|
||
bounds['y_total'] = (cy, text)
|
||
elif cy > bounds['y_total'][0]:
|
||
bounds['y_total'] = (cy, text)
|
||
elif '日期' in t:
|
||
bounds['y_date'] = cy
|
||
|
||
return bounds
|
||
|
||
def _extract_header(self, result, bounds):
|
||
"""提取凭证头部信息：日期、凭证号、公司"""
|
||
for text, conf, cx, cy, box in self.lines:
|
||
t = text.strip()
|
||
|
||
if '物业' in t and '公司' in t and not result['company']:
|
||
result['company'] = t
|
||
elif '日期' in t:
|
||
m = re.search(r'(\d{4}[-年]\d{1,2}[-月]\d{1,2})', t)
|
||
if m:
|
||
ds = m.group(1).replace('年', '-').replace('月', '-').replace('日', '')
|
||
try:
|
||
result['date'] = datetime.strptime(ds, '%Y-%m-%d')
|
||
result['year'] = result['date'].year
|
||
result['period'] = result['date'].month
|
||
except:
|
||
pass
|
||
elif '凭证' in t:
|
||
m = re.search(r'[记凭证号][^\d]*(\d+)', t)
|
||
if m:
|
||
result['voucher_no'] = int(m.group(1))
|
||
elif '附件' in t:
|
||
m = re.search(r'附件数[：:]?(\d+)', t)
|
||
if m:
|
||
result['attachments'] = int(m.group(1))
|
||
elif '第' in t and '期' in t:
|
||
m = re.search(r'第(\d+)期', t)
|
||
if m:
|
||
result['period'] = int(m.group(1))
|
||
elif '合计' in t and any(c in t for c in '零壹贰叁肆伍陆柒捌玖拾佰仟万亿'):
|
||
result['total_str'] = t
|
||
m = re.search(r'([\d,]+\.\d{2})', t.replace(',', ''))
|
||
if m:
|
||
result['total_amount'] = float(m.group(1).replace(',', ''))
|
||
elif t.startswith('制单') or t.startswith('制单人'):
|
||
result['preparer'] = t.split('：')[-1].split(':')[-1].strip()
|
||
elif t.startswith('审核'):
|
||
result['checker'] = t.split('：')[-1].split(':')[-1].strip()
|
||
elif t.startswith('出纳'):
|
||
result['cashier'] = t.split('：')[-1].split(':')[-1].strip()
|
||
elif t.startswith('经办'):
|
||
result['handler'] = t.split('：')[-1].split(':')[-1].strip()
|
||
elif t.startswith('过账'):
|
||
result['poster'] = t.split('：')[-1].split(':')[-1].strip()
|
||
|
||
def _extract_codes(self, bounds):
|
||
"""提取科目代码和名称"""
|
||
codes = []
|
||
seen = set()
|
||
|
||
# 安全获取边界值
|
||
y_subj = bounds.get('y_subject_header')
|
||
y_debit = bounds.get('y_debit_header')
|
||
y_summary = bounds.get('y_summary_header')
|
||
|
||
# 默认边界
|
||
y_subj_val = y_subj if y_subj is not None else 2000
|
||
y_debit_val = y_debit if y_debit is not None else 99999
|
||
y_lower = y_summary if y_summary is not None else 800
|
||
|
||
for text, conf, cx, cy, box in self.lines:
|
||
if conf < 0.5:
|
||
continue
|
||
|
||
# 科目代码区域：摘要标题上方 ~ 借方标题之间
|
||
# 实际中代码可能在"科目"标题上方（y<y_subj），所以要放宽下界
|
||
if cy < y_lower - 100 or cy > y_debit_val:
|
||
continue
|
||
|
||
# x坐标：科目代码通常在表格中间区域
|
||
if cx < 1500 or cx > 2600:
|
||
continue
|
||
|
||
t = text.strip()
|
||
# 排除表头文字
|
||
if t in ('摘要', '科目', '借方', '贷方', '合计') or len(t) < 4:
|
||
continue
|
||
|
||
# 匹配科目代码模式：4位数字开头，后面跟 .数字
|
||
m = re.match(r'(\d{3,4}\.\d[\d.]*)', t)
|
||
if not m:
|
||
continue
|
||
code_part = m.group(1).rstrip('.')
|
||
|
||
# 验证是合法的科目代码（1000~9999）
|
||
first_num = int(code_part.split('.')[0])
|
||
if first_num < 1000 or first_num > 9999:
|
||
continue
|
||
if code_part in seen:
|
||
continue
|
||
seen.add(code_part)
|
||
|
||
# 提取名称
|
||
name = t[m.end():].lstrip('－-—= ')
|
||
# 清除尾随乱码（尾随的数字和特殊符号）
|
||
name = re.sub(r'[\d.\-—=]+$', '', name).strip()
|
||
# 清除开头的破折号
|
||
name = name.lstrip('-—－=')
|
||
|
||
codes.append({
|
||
'code': code_part,
|
||
'name': name,
|
||
'desc': '',
|
||
'debit': 0.0,
|
||
'credit': 0.0,
|
||
'y_pos': cy,
|
||
'x_pos': cx,
|
||
})
|
||
|
||
# 按 y 排序
|
||
codes.sort(key=lambda c: c['y_pos'])
|
||
return codes
|
||
|
||
def _extract_amounts(self, bounds):
|
||
"""提取金额，按借方/贷方分类"""
|
||
y_debit = bounds.get('y_debit_header')
|
||
y_credit = bounds.get('y_credit_header')
|
||
|
||
# 安全默认值
|
||
y_debit_val = y_debit if y_debit is not None else 3000
|
||
y_credit_val = y_credit if y_credit is not None else 3500
|
||
|
||
amounts = []
|
||
skip_texts = {'摘要', '科目', '借方', '贷方', '合计', '经办', '审核', '出纳', '制单'}
|
||
|
||
for text, conf, cx, cy, box in self.lines:
|
||
if conf < 0.5:
|
||
continue
|
||
if text.strip() in skip_texts:
|
||
continue
|
||
if any(k in text for k in ['Manager', 'HONOR', 'Magic6', 'RMB']):
|
||
continue
|
||
|
||
m = re.search(r'([\d,]+\.\d{2})', text.replace(',', ''))
|
||
if not m:
|
||
continue
|
||
|
||
val = float(m.group(1).replace(',', ''))
|
||
# 排除年份
|
||
if 1900 < val < 2100:
|
||
continue
|
||
if val < 0.5:
|
||
continue
|
||
|
||
# 分类借贷：只取借方标题以下的文本作为金额
|
||
if cy < y_debit_val:
|
||
continue # 在借方标题上方，不是金额
|
||
elif cy <= y_credit_val:
|
||
zone = 'debit'
|
||
else:
|
||
zone = 'credit'
|
||
|
||
amounts.append({'val': val, 'x': cx, 'y': cy, 'zone': zone})
|
||
|
||
debit_amts = [a for a in amounts if a['zone'] == 'debit']
|
||
credit_amts = [a for a in amounts if a['zone'] == 'credit']
|
||
|
||
return debit_amts, credit_amts
|
||
|
||
def _deduplicate_amounts(self, amts):
|
||
"""对金额去重：相同值的只保留一个（OCR常把同一金额读两次）"""
|
||
if not amts:
|
||
return []
|
||
|
||
# 按 y 排序后，相同值的只保留第一个
|
||
amts.sort(key=lambda a: a['y'])
|
||
seen_vals = set()
|
||
deduped = []
|
||
for a in amts:
|
||
rounded = round(a['val'], 2)
|
||
if rounded in seen_vals:
|
||
continue # 相同值的金额只保留一个
|
||
seen_vals.add(rounded)
|
||
deduped.append(a)
|
||
|
||
# 进一步：如果某个金额等于其他金额之和，去除（可能是合计行）
|
||
if len(deduped) >= 3:
|
||
total = sum(a['val'] for a in deduped)
|
||
for a in deduped[:]:
|
||
other_sum = total - a['val']
|
||
if other_sum > 0.01 and abs(a['val'] - other_sum) < 0.01:
|
||
deduped = [x for x in deduped if x is not a]
|
||
break
|
||
|
||
return deduped
|
||
|
||
return deduped
|
||
|
||
def _extract_summary(self, bounds):
|
||
"""提取凭证摘要（通常在表格顶部区域）"""
|
||
y_subj = bounds.get('y_subject_header')
|
||
y_subj_val = y_subj if y_subj is not None else 2000
|
||
|
||
candidates = []
|
||
for text, conf, cx, cy, box in self.lines:
|
||
if conf < 0.6:
|
||
continue
|
||
if not re.search(r'[\u4e00-\u9fff]', text):
|
||
continue
|
||
if cy < 0 or cy > y_subj_val - 100:
|
||
continue
|
||
if len(text.strip()) < 6:
|
||
continue
|
||
if any(k in text for k in ['经办', '合计', '审核', '日期', '业务', '附件']):
|
||
continue
|
||
|
||
candidates.append((text, cy))
|
||
|
||
# 按 y 排序，取最靠上且有意义的候选
|
||
candidates.sort(key=lambda x: x[1])
|
||
# 过滤掉日期/期号类文本
|
||
real_summaries = [t for t, y in candidates
|
||
if not re.match(r'\d{3}年第\d+期', t)
|
||
and not re.match(r'[:：]\d+', t)
|
||
and '日期' not in t
|
||
and len(t) >= 6]
|
||
if real_summaries:
|
||
return real_summaries[0]
|
||
return ''
|
||
|
||
def _match_entries(self, result, codes, debit_amts, credit_amts, summary, bounds):
|
||
"""将金额按匹配到科目，构建分录"""
|
||
if not codes:
|
||
result['entries'] = []
|
||
return
|
||
|
||
n_codes = len(codes)
|
||
n_debit = len(debit_amts)
|
||
n_credit = len(credit_amts)
|
||
|
||
# 策略 1：等额借贷
|
||
if n_codes == n_debit == n_credit:
|
||
for i, code in enumerate(codes):
|
||
code['debit'] = debit_amts[i]['val']
|
||
code['credit'] = credit_amts[i]['val']
|
||
|
||
# 策略 2：单行分录
|
||
elif n_codes == 1 and (n_debit > 0 or n_credit > 0):
|
||
codes[0]['debit'] = sum(a['val'] for a in debit_amts)
|
||
codes[0]['credit'] = sum(a['val'] for a in credit_amts)
|
||
|
||
# 策略 3：借贷均足量，按顺序匹配
|
||
elif n_debit >= n_codes and n_credit >= n_codes:
|
||
for i, code in enumerate(codes):
|
||
if i < n_debit:
|
||
code['debit'] = debit_amts[i]['val']
|
||
if i < n_credit:
|
||
code['credit'] = credit_amts[i]['val']
|
||
|
||
# 策略 4：借方多，贷方平均分配
|
||
elif n_debit >= n_codes:
|
||
for i, code in enumerate(codes):
|
||
code['debit'] = debit_amts[i]['val']
|
||
if n_credit > 0:
|
||
per_code = sum(a['val'] for a in credit_amts) / n_codes
|
||
for code in codes:
|
||
code['credit'] = round(per_code, 2)
|
||
|
||
# 策略 5：贷方多，借方平均分配
|
||
elif n_credit >= n_codes:
|
||
for i, code in enumerate(codes):
|
||
code['credit'] = credit_amts[i]['val']
|
||
if n_debit > 0:
|
||
per_code = sum(a['val'] for a in debit_amts) / n_codes
|
||
for code in codes:
|
||
code['debit'] = round(per_code, 2)
|
||
|
||
# 策略 6：都不够
|
||
else:
|
||
for i, code in enumerate(codes):
|
||
if i < n_debit:
|
||
code['debit'] = debit_amts[i]['val']
|
||
if i < n_credit:
|
||
code['credit'] = credit_amts[i]['val']
|
||
|
||
# 设置摘要
|
||
for code in codes:
|
||
code['desc'] = summary or '会计凭证'
|
||
|
||
# 构建 entries
|
||
result['entries'] = []
|
||
for i, code in enumerate(codes):
|
||
result['entries'].append({
|
||
'id': i + 1,
|
||
'description': code['desc'],
|
||
'account_code': code['code'],
|
||
'account_name': code['name'] or '未知科目',
|
||
'debit': code['debit'],
|
||
'credit': code['credit'],
|
||
})
|
||
|
||
# 清洗科目名称
|
||
for e in result['entries']:
|
||
e['account_name'] = e['account_name'].replace('--', '-').strip()
|