560 lines
23 KiB
Python
560 lines
23 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]))
|
||
self._merge_continuation_lines()
|
||
|
||
def _merge_continuation_lines(self):
|
||
"""合并续行：会计科目名称被OCR分成多行时，合并回一行
|
||
|
||
策略：续行以横杠开头（如 -淇县支行1710020809200182975），
|
||
找到上方最近的一个以横杠结尾的代码行进行合并。
|
||
如果找不到以横杠结尾的代码行，则找上方最近的不以横杠开头的代码行。
|
||
"""
|
||
# 先识别所有代码行
|
||
code_lines = [] # [(idx, text, ends_with_dash, cy, cx)]
|
||
for i, (text, conf, cx, cy, box) in enumerate(self.lines):
|
||
t = text.strip()
|
||
if re.match(r'\d{3,4}\.\d', t):
|
||
code_lines.append((i, t, t.endswith('-'), cy, cx))
|
||
|
||
if not code_lines:
|
||
return
|
||
|
||
merged = list(self.lines)
|
||
removed = set()
|
||
merged_codes = set() # 已接收续行的代码行，不再接收
|
||
|
||
for i, (text, conf, cx, cy, box) in enumerate(self.lines):
|
||
t = text.strip()
|
||
if not (t.startswith('-') or t.startswith('－') or t.startswith('—')):
|
||
continue
|
||
if i in removed:
|
||
continue
|
||
|
||
# 跳过包含日期/表头关键词的续行
|
||
stripped = t.lstrip('－-—')
|
||
if any(k in stripped for k in ['日期', '记账凭证', '记账', '业务日期', '凭证号', '附件数']):
|
||
continue
|
||
|
||
# 匹配代码行
|
||
best_code = None
|
||
best_score = 99999
|
||
|
||
for idx, ct, ends_dash, ccy, ccx in code_lines:
|
||
if idx in merged_codes:
|
||
continue # 已合并过的代码行不再接收续行
|
||
if abs(cy - ccy) > 300:
|
||
continue
|
||
if abs(cx - ccx) > 300:
|
||
continue
|
||
|
||
dist_y = abs(cy - ccy)
|
||
dist_x = abs(cx - ccx)
|
||
|
||
if ends_dash:
|
||
score = dist_y * 2 + dist_x * 2
|
||
else:
|
||
# 不以横杠结尾的代码行：x必须非常接近才接受
|
||
if dist_x > 100:
|
||
continue
|
||
score = 1500 + dist_y + dist_x * 2
|
||
|
||
if score < best_score:
|
||
best_score = score
|
||
best_code = idx
|
||
|
||
if best_code is not None and best_score < 2000:
|
||
orig_text, orig_conf, orig_cx, orig_cy, orig_box = merged[best_code]
|
||
merged[best_code] = (orig_text + text, min(orig_conf, conf), orig_cx, orig_cy, orig_box)
|
||
removed.add(i)
|
||
merged_codes.add(best_code) # 标记已合并
|
||
|
||
self.lines = [merged[i] for i in range(len(merged)) if i not in removed]
|
||
|
||
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})', re.sub(r'[^\d.]', '', t))
|
||
if m:
|
||
result['total_amount'] = float(m.group(1))
|
||
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 = []
|
||
code_map = {} # code_part → 在 codes 列表中的索引
|
||
|
||
# 安全获取边界值
|
||
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
|
||
|
||
# 科目代码区域：摘要标题上方 ~ 借方标题之间
|
||
if cy < y_lower - 100 or cy > y_debit_val:
|
||
continue
|
||
|
||
# x坐标：科目代码通常在表格中间区域
|
||
# 注：不设硬x过滤器，正则 \d{3,4}\.\d 和首位数 1000-9999 校验已足够精准
|
||
# 设 x 过滤器会误杀 PDF 版本（排版偏左）和少部分照片（x=1297 也有代码）
|
||
|
||
t = text.strip()
|
||
# 排除表头文字
|
||
if t in ('摘要', '科目', '借方', '贷方', '合计') or len(t) < 4:
|
||
continue
|
||
|
||
# 匹配科目代码模式：3-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
|
||
|
||
# 提取名称 — 按用户规则：取最后一个横杠之后的文字
|
||
name = t[m.end():].lstrip('－-—= ')
|
||
# 保留长数字尾（银行账号），仅清除短数字尾
|
||
if not re.search(r'\d{6,}', name):
|
||
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()
|
||
|
||
# 去重：优先保留带完整名称的行
|
||
if code_part in code_map:
|
||
existing = codes[code_map[code_part]]
|
||
# 已有名称为空，而新行有名称 → 更新
|
||
if len(name) >= 2 and (not existing['name'] or len(existing['name']) < 2 or len(name) > len(existing['name'])): # 优先长名
|
||
existing['name'] = name
|
||
existing['y_pos'] = cy
|
||
existing['x_pos'] = cx
|
||
continue
|
||
|
||
codes.append({
|
||
'code': code_part,
|
||
'name': name,
|
||
'desc': '',
|
||
'debit': 0.0,
|
||
'credit': 0.0,
|
||
'y_pos': cy,
|
||
'x_pos': cx,
|
||
})
|
||
code_map[code_part] = len(codes) - 1
|
||
|
||
# 按 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 any(k in t for k in ['过账', '经办', '审核', '出纳', '制单', 'Manager']):
|
||
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 # 只找下方的
|
||
if abs(cx - code['x_pos']) > 400:
|
||
continue
|
||
dist = cy - code['y_pos']
|
||
if dist < best_dist and dist < 400:
|
||
best = (t, cy, cx)
|
||
best_dist = dist
|
||
if best:
|
||
t, cy, cx = best
|
||
# 按用户规则提取名称（原始逻辑：先清洗后取最后横杠）
|
||
n = t.lstrip('－-—= ')
|
||
# 保留长数字尾（银行账号）
|
||
if not re.search(r'\d{6,}', n):
|
||
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:
|
||
# 名称质量检查：如果提取结果可疑，试下一个候选
|
||
bad_name = (len(n) < 3 or n.endswith('/')
|
||
or n in ('里C', '里A', '假台', '里C3', '里A3')
|
||
or re.search(r'[里景苑园]$', n)
|
||
or re.search(r'[A-Z]\d', n))
|
||
if bad_name:
|
||
# 移除当前候选，再找下一个
|
||
name_candidates = [(ct, ccy, ccx) for ct, ccy, ccx in name_candidates
|
||
if ct != t or ccy != cy or ccx != cx]
|
||
# 递归尝试下一个（限制深度）
|
||
for _ in range(10):
|
||
next_best = None
|
||
next_dist = 99999
|
||
for nt, ncy, ncx in name_candidates:
|
||
if ncy <= code['y_pos']:
|
||
continue
|
||
if abs(ncx - code['x_pos']) > 400:
|
||
continue
|
||
ndist = ncy - code['y_pos']
|
||
if ndist < next_dist and ndist < 400:
|
||
next_best = (nt, ncy, ncx)
|
||
next_dist = ndist
|
||
if not next_best:
|
||
break
|
||
nt, ncy, ncx = next_best
|
||
nn = nt.lstrip('－-—= ')
|
||
# 保留长数字尾（银行账号）
|
||
if not re.search(r'\d{6,}', nn):
|
||
nn = re.sub(r'[\d.\-—=]+$', '', nn).strip()
|
||
nn = nn.lstrip('-—－=')
|
||
if '\\' in nn:
|
||
nn = nn.split('\\', 1)[0].strip()
|
||
else:
|
||
nidx = max(nn.rfind('－'), nn.rfind('-'), nn.rfind('—'))
|
||
if nidx >= 0:
|
||
nn = nn[nidx+1:].strip()
|
||
if nn and len(nn) >= 2 and not (
|
||
len(nn) < 3 or nn.endswith('/')
|
||
or nn in ('里C', '里A', '假台', '里C3', '里A3')
|
||
or re.search(r'[里景苑园]$', nn)
|
||
or re.search(r'[A-Z]\d', nn)
|
||
):
|
||
code['name'] = nn
|
||
break
|
||
# 这个也不行，继续下一个
|
||
name_candidates = [(ct, ccy, ccx) for ct, ccy, ccx in name_candidates
|
||
if ct != nt or ccy != ncy or ccx != ncx]
|
||
else:
|
||
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
|
||
|
||
# 提取金额：去掉所有非数字字符（逗号/空格等），避免"11, 075.68"被解析为75.68
|
||
cleaned = re.sub(r'[^\d.]', '', text)
|
||
m = re.search(r'(\d+\.\d{2})', cleaned)
|
||
if not m:
|
||
continue
|
||
|
||
val = float(m.group(1))
|
||
# 排除年份
|
||
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])
|
||
# 过滤掉日期/期号类文本（如"J22年第9期"、"022年第9期"）
|
||
real_summaries = [t for t, y in candidates
|
||
if not re.match(r'.*\d{1,4}年第?\d{1,2}期', 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()
|