430 lines
16 KiB
Python
430 lines
16 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('-—－=')
|
||
|
||
if '\\' in name:
|
||
name = name.split('\\', 1)[0].strip()
|
||
else:
|
||
idx = max(name.rfind('－'), name.rfind('-'), name.rfind('—'))
|
||
if idx >= 0:
|
||
name = name[idx+1:].strip()
|
||
|
||
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'])
|
||
|
||
# ---- 名称补位：代码行没有名称时，从附近行补取 ----
|
||
# 收集科目区所有含中文的文本（用于名称补位）
|
||
name_candidates = []
|
||
for text, conf, cx, cy, box in self.lines:
|
||
if conf < 0.5:
|
||
continue
|
||
if cy < y_lower - 100 or cy > y_debit_val:
|
||
continue
|
||
t = text.strip()
|
||
if len(t) < 2 or not re.search(r'[\u4e00-\u9fff]', t):
|
||
continue
|
||
if t in ('摘要', '科目', '借方', '贷方', '合计', '日期', '会计'):
|
||
continue
|
||
# 排除本身就是科目代码的行
|
||
if re.match(r'\d{3,4}\.\d', t):
|
||
continue
|
||
name_candidates.append((t, cy, cx))
|
||
|
||
name_candidates.sort(key=lambda x: x[1])
|
||
|
||
for code in codes:
|
||
if code['name'] and len(code['name']) >= 2:
|
||
continue # 已有名称，跳过
|
||
# 找代码行下方最近的含中文文本
|
||
best = None
|
||
best_dist = 99999
|
||
for t, cy, cx in name_candidates:
|
||
if cy <= code['y_pos']:
|
||
continue # 只找下方的
|
||
# x 坐标接近（左右偏差不超过600px）
|
||
if abs(cx - code['x_pos']) > 600:
|
||
continue
|
||
dist = cy - code['y_pos']
|
||
if dist < best_dist and dist < 600:
|
||
best = (t, cy, cx)
|
||
best_dist = dist
|
||
if best:
|
||
t, cy, cx = best
|
||
# 按用户规则提取名称
|
||
n = t.lstrip('－-—= ')
|
||
n = re.sub(r'[\d.\-—=]+$', '', n).strip()
|
||
n = n.lstrip('-—－=')
|
||
if '\\' in n:
|
||
n = n.split('\\', 1)[0].strip()
|
||
else:
|
||
idx = max(n.rfind('－'), n.rfind('-'), n.rfind('—'))
|
||
if idx >= 0:
|
||
n = n[idx+1:].strip()
|
||
if n and len(n) >= 2:
|
||
code['name'] = n
|
||
|
||
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 重复读取的只保留一个（按 y 坐标邻近去重）
|
||
|
||
不按值去重——同一区域完全可能有多个相同金额（如借方3笔3000.00）。
|
||
只在同一位值（y差<15像素）且值相同时才认为是 OCR 重复。
|
||
"""
|
||
if not amts:
|
||
return []
|
||
|
||
# 按 y 排序
|
||
amts.sort(key=lambda a: a['y'])
|
||
|
||
deduped = []
|
||
for a in amts:
|
||
# 检查是否与上一条金额在同一位置附近（y差<15像素且值相同）
|
||
is_dup = False
|
||
if deduped:
|
||
last = deduped[-1]
|
||
if abs(a['y'] - last['y']) < 15 and abs(a['val'] - last['val']) < 0.01:
|
||
is_dup = True
|
||
if not is_dup:
|
||
deduped.append(a)
|
||
|
||
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:
|
||
summary = real_summaries[0]
|
||
# P05/PO5/P0S → POS 纠错
|
||
summary = re.sub(r'P[O0][5S]', 'POS', summary)
|
||
return summary
|
||
return ''
|
||
|
||
def _match_entries(self, result, codes, debit_amts, credit_amts, summary, bounds):
|
||
"""将金额匹配到科目，构建分录
|
||
|
||
顺序分配 + 多余金额合并到最后一个科目：
|
||
- 第1个借方金额 → 第1个科目，第2个借方 → 第2个科目……
|
||
- 借方金额多于科目数 → 多余金额合并到最后一个科目
|
||
- 贷方同理
|
||
- 不做平均分配（平均分配会产生不存在于原始凭证的金额）
|
||
"""
|
||
if not codes:
|
||
result['entries'] = []
|
||
return
|
||
|
||
n_codes = len(codes)
|
||
|
||
# 初始化
|
||
for code in codes:
|
||
code['debit'] = 0.0
|
||
code['credit'] = 0.0
|
||
|
||
# 借方顺序分配：多余的累加到最后一个科目
|
||
for i, amt in enumerate(debit_amts):
|
||
idx = min(i, n_codes - 1)
|
||
codes[idx]['debit'] += amt['val']
|
||
|
||
# 贷方顺序分配：多余的累加到最后一个科目
|
||
for i, amt in enumerate(credit_amts):
|
||
idx = min(i, n_codes - 1)
|
||
codes[idx]['credit'] += amt['val']
|
||
|
||
# 四舍五入
|
||
for code in codes:
|
||
code['debit'] = round(code['debit'], 2)
|
||
code['credit'] = round(code['credit'], 2)
|
||
|
||
# 设置摘要
|
||
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()
|