ry-ocr/ (新)
本地发票识别微服务 (PaddleOCR 3.x + FastAPI, 8801)
- QR 优先: 扫到二维码即取开票时间/发票号/金额; 没扫到/格式不合法直接判非发票, 不跑 OCR
- 配置 QR_FULL_OCR 控制快路径(false, 0.2s)还是全字段(true, 4.5s)
- /recognize/invoice (multipart) + /recognize/invoice/by-path (本地路径, 白名单) + /recognize/text
- is_invoice / from_qr / qr_raw / qr_error / error_code 字段
- 12 字段发票抽取 (regex + 启发式, 左右主体识别)
- 超时保护 (15s 单页 / 60s 总流程) + PaddleOCR 单例 + ThreadPoolExecutor
ry-api/ruoyi-business/
- pom.xml: 加 hutool-http/json/core 5.8.27, lombok 1.18.30 (OcrClient @Slf4j 所需)
- ocr/: OcrClient + InvoiceResult/Fields/Line + ZipExtractor + InvoiceOcrScheduler
- oss/: OssUploader + OssConfMeta (OCR 识别后重传 OSS)
- config/: OcrConfig + OcrExecutorConfig (后台线程池)
- service/impl/InvoiceOcrService: 后台提交 OCR, ZIP 路径解压识别, 替换场景先清旧
- 会议材料 CRUD 全套 (BizMeetingAuditLog/Executor/Invoice/Material/Supervisor):
controller + service + mapper + domain + xml
ry-vue3/
- MeetingDetail.vue (新建): 会议详情页 (含评分维度章节, 改只读)
- Meetings.vue / OssFileUploader.vue / router / Login.vue: 适配新字段
ry-api/ruoyi-admin/
- RuoYiApplication.java + application.yml: 启用 @Async 异步支持
_self/
- manager_meetings.md / manager_meeting_detail.md: 文档
299 lines
10 KiB
Python
299 lines
10 KiB
Python
"""从 OCR 文本/行里抽取发票字段
|
||
|
||
适配中国大陆 增值税发票(电子普票 / 专票 / 电子专票 / 数电票)
|
||
|
||
关键策略:
|
||
- 主体(销售方/购买方)按 OCR box **坐标**判断归属(左右两栏)
|
||
- 名称提取加 stop word,避免 OCR 单行文本混淆
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from typing import List, Optional, Tuple
|
||
|
||
from app.models import InvoiceFields, OCRLine
|
||
from app.utils import (
|
||
amount_consistent,
|
||
extract_cn_amount,
|
||
extract_num_amount,
|
||
extract_pretax_amount,
|
||
extract_tax_amount,
|
||
extract_total_amount,
|
||
)
|
||
|
||
|
||
# ---------- 发票类型 ----------
|
||
_INVOICE_TYPES = [
|
||
"增值税电子专用发票",
|
||
"增值税电子普通发票",
|
||
"增值税专用发票",
|
||
"增值税普通发票",
|
||
"通用机打发票",
|
||
"数电票",
|
||
"电子发票",
|
||
]
|
||
|
||
# ---------- 发票号码 ----------
|
||
_NO_PATTERN = re.compile(
|
||
r"(?:发\s*票\s*号\s*码|号\s*码|No\.?|号)\s*[::]?\s*(\d{8,20})",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
# ---------- 发票代码 ----------
|
||
_CODE_PATTERN = re.compile(
|
||
r"(?:发\s*票\s*代\s*码|代\s*码)\s*[::]?\s*(\d{10,12}|\d{8,12})",
|
||
)
|
||
|
||
# ---------- 开票日期 ----------
|
||
_DATE_PATTERN = re.compile(
|
||
r"(?:开\s*票\s*日\s*期|日\s*期)\s*[::]?\s*"
|
||
r"(\d{4})\s*[年/\.]\s*(\d{1,2})\s*[月/\.]\s*(\d{1,2})",
|
||
)
|
||
|
||
# ---------- 纳税人识别号(必须至少含 1 个字母,排除纯数字发票号)----------
|
||
_TAX_NO_PATTERN = re.compile(r"((?=[0-9A-Z]*[A-Z])[0-9A-Z]{18})")
|
||
|
||
# ---------- 主体标签 ----------
|
||
_BUYER_LABEL = re.compile(r"购\s*买\s*方\s*(?:信\s*息|名\s*称|)")
|
||
_SELLER_LABEL = re.compile(r"销\s*售\s*方\s*(?:信\s*息|名\s*称|)")
|
||
|
||
# ---------- 名称(带 stop word 截断)----------
|
||
_NAME_STOP = r"(?:销售方|购买方|统一社会信用|纳税人|项目名称|规格型号|^单位$|^数量$|^单价$|^金额|^税率|^税额|备注|收款人|复核|开票人|价税合计|小写|大写)"
|
||
_NAME_PATTERN = re.compile(
|
||
rf"名\s*称\s*[::]\s*"
|
||
rf"((?:(?!{_NAME_STOP})[^\n\r]){{2,60}}?(?:公司|商店|厂|店|部|中心|工作室))"
|
||
)
|
||
|
||
|
||
# ---------- 工具函数 ----------
|
||
|
||
def _norm(text: str) -> str:
|
||
return re.sub(r"\s+", " ", text or "").strip()
|
||
|
||
|
||
def _box_center(box) -> Tuple[float, float]:
|
||
"""box: [[x1,y1], [x2,y2], [x3,y3], [x4,y4]] → (cx, cy)"""
|
||
if not box or len(box) < 4:
|
||
return (0.0, 0.0)
|
||
xs = [p[0] for p in box]
|
||
ys = [p[1] for p in box]
|
||
return ((min(xs) + max(xs)) / 2, (min(ys) + max(ys)) / 2)
|
||
|
||
|
||
def _detect_invoice_type(text: str) -> Optional[str]:
|
||
for t in _INVOICE_TYPES:
|
||
if t in text:
|
||
return t
|
||
return None
|
||
|
||
|
||
def _extract_invoice_no(text: str) -> Optional[str]:
|
||
m = _NO_PATTERN.search(text)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _extract_invoice_code(text: str) -> Optional[str]:
|
||
m = _CODE_PATTERN.search(text)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _extract_date(text: str) -> Optional[str]:
|
||
m = _DATE_PATTERN.search(text)
|
||
if not m:
|
||
return None
|
||
y, mo, d = m.groups()
|
||
return f"{int(y):04d}-{int(mo):02d}-{int(d):02d}"
|
||
|
||
|
||
def _clean_name(name: str) -> str:
|
||
name = re.sub(r"^[\s::,,。、]+", "", name)
|
||
name = name.split("纳税人")[0].split("统一社会")[0]
|
||
name = re.sub(r"[^一-龥A-Za-z0-9()()·\-]", "", name)
|
||
return name.rstrip("::;,,。、 ").strip()
|
||
|
||
|
||
def _extract_parties_from_lines(lines: List[OCRLine]) -> tuple:
|
||
"""基于 box 坐标的主体识别
|
||
|
||
1) 找到 "购买方信息"/"销售方信息" 标签所在位置(box 中心)
|
||
2) 把每个 "名称: xxx" 行按 x 坐标归属到对应栏
|
||
3) 兜底:左右栏无法区分时按 y 坐标(top/bottom)
|
||
"""
|
||
# 找标签坐标
|
||
buyer_label_box = None
|
||
seller_label_box = None
|
||
for line in lines:
|
||
if _BUYER_LABEL.search(line.text) and not buyer_label_box:
|
||
buyer_label_box = line.box
|
||
if _SELLER_LABEL.search(line.text) and not seller_label_box:
|
||
seller_label_box = line.box
|
||
|
||
by_x = buyer_label_box is not None and seller_label_box is not None and \
|
||
abs(_box_center(buyer_label_box)[0] - _box_center(seller_label_box)[0]) > 50
|
||
|
||
# 找所有 "名称: ..." 行的 box
|
||
name_lines = []
|
||
for line in lines:
|
||
m = _NAME_PATTERN.search(line.text)
|
||
if m:
|
||
name_lines.append((line, _clean_name(m.group(1))))
|
||
|
||
# 找所有税号
|
||
tax_lines = []
|
||
for line in lines:
|
||
m = _TAX_NO_PATTERN.search(line.text)
|
||
if m:
|
||
tax_lines.append((line, m.group(1)))
|
||
|
||
seller_name = buyer_name = None
|
||
seller_tax = buyer_tax = None
|
||
|
||
if by_x and buyer_label_box and seller_label_box:
|
||
# 左右栏布局:用 x 坐标归属
|
||
bx, _ = _box_center(buyer_label_box)
|
||
sx, _ = _box_center(seller_label_box)
|
||
mid = (bx + sx) / 2
|
||
|
||
for line, name in name_lines:
|
||
cx, _ = _box_center(line.box)
|
||
if cx < mid and not buyer_name:
|
||
buyer_name = name
|
||
elif cx >= mid and not seller_name:
|
||
seller_name = name
|
||
|
||
for line, tax in tax_lines:
|
||
cx, _ = _box_center(line.box)
|
||
if cx < mid and not buyer_tax:
|
||
buyer_tax = tax
|
||
elif cx >= mid and not seller_tax:
|
||
seller_tax = tax
|
||
else:
|
||
# 上下栏 / 单栏:按 y 坐标(第一个 = 购买方)
|
||
name_lines.sort(key=lambda x: (_box_center(x[0].box)[1], _box_center(x[0].box)[0]))
|
||
tax_lines.sort(key=lambda x: (_box_center(x[0].box)[1], _box_center(x[0].box)[0]))
|
||
if len(name_lines) >= 1:
|
||
buyer_name = name_lines[0][1]
|
||
if len(name_lines) >= 2:
|
||
seller_name = name_lines[1][1]
|
||
if len(tax_lines) >= 1:
|
||
buyer_tax = tax_lines[0][1]
|
||
if len(tax_lines) >= 2:
|
||
seller_tax = tax_lines[1][1]
|
||
|
||
return seller_name, seller_tax, buyer_name, buyer_tax
|
||
|
||
|
||
def _extract_parties_from_text(text: str) -> tuple:
|
||
"""无 box 信息时的 fallback:按显式标签分段"""
|
||
seller_name = buyer_name = None
|
||
seller_tax = buyer_tax = None
|
||
|
||
for m in _BUYER_LABEL.finditer(text):
|
||
chunk = text[m.end():m.end() + 200]
|
||
nm = _NAME_PATTERN.search(chunk)
|
||
tm = _TAX_NO_PATTERN.search(chunk[:100])
|
||
if nm and not buyer_name:
|
||
buyer_name = _clean_name(nm.group(1))
|
||
if tm and not buyer_tax:
|
||
buyer_tax = tm.group(1)
|
||
if buyer_name and buyer_tax:
|
||
break
|
||
|
||
for m in _SELLER_LABEL.finditer(text):
|
||
chunk = text[m.end():m.end() + 200]
|
||
nm = _NAME_PATTERN.search(chunk)
|
||
tm = _TAX_NO_PATTERN.search(chunk[:100])
|
||
if nm and not seller_name:
|
||
seller_name = _clean_name(nm.group(1))
|
||
if tm and not seller_tax:
|
||
seller_tax = tm.group(1)
|
||
if seller_name and seller_tax:
|
||
break
|
||
|
||
if not buyer_name:
|
||
name_positions = [(m.start(), m.group(1)) for m in _NAME_PATTERN.finditer(text)]
|
||
if len(name_positions) >= 1:
|
||
buyer_name = _clean_name(name_positions[0][1])
|
||
if len(name_positions) >= 2:
|
||
seller_name = _clean_name(name_positions[1][1])
|
||
|
||
tax_positions = _TAX_NO_PATTERN.findall(text)
|
||
if not buyer_tax and len(tax_positions) >= 1:
|
||
buyer_tax = tax_positions[0]
|
||
if not seller_tax and len(tax_positions) >= 2:
|
||
seller_tax = tax_positions[1]
|
||
|
||
return seller_name, seller_tax, buyer_name, buyer_tax
|
||
|
||
|
||
def extract(text: str, lines: Optional[List[OCRLine]] = None) -> InvoiceFields:
|
||
raw = _norm(text)
|
||
|
||
fields = InvoiceFields()
|
||
fields.invoice_type = _detect_invoice_type(raw)
|
||
fields.invoice_code = _extract_invoice_code(raw)
|
||
fields.invoice_no = _extract_invoice_no(raw)
|
||
fields.invoice_date = _extract_date(raw)
|
||
|
||
# 金额
|
||
total = extract_total_amount(raw)
|
||
tax = extract_tax_amount(raw)
|
||
pretax = extract_pretax_amount(raw)
|
||
|
||
# 兜底:从所有 .xx 数字中找满足 tax + pretax = total 的组合
|
||
if (tax is None or pretax is None) and total is not None:
|
||
candidates = sorted(
|
||
{float(x) for x in re.findall(r"(\d+\.\d{2})", raw) if float(x) < total},
|
||
reverse=True,
|
||
)
|
||
for i, a in enumerate(candidates):
|
||
for b in candidates[i + 1:]:
|
||
if abs(a + b - total) < 0.011:
|
||
if pretax is None:
|
||
pretax = round(a, 2)
|
||
if tax is None:
|
||
tax = round(b, 2)
|
||
break
|
||
if tax is not None and pretax is not None:
|
||
break
|
||
# 单数字兜底:如果 total 不在候选里,只剩一个候选时
|
||
if (tax is None or pretax is None) and len(candidates) == 1:
|
||
only = round(candidates[0], 2)
|
||
if pretax is None and tax is None:
|
||
pretax = only
|
||
tax = round(total - only, 2)
|
||
elif tax is None:
|
||
tax = only
|
||
elif pretax is None:
|
||
pretax = only
|
||
|
||
# 二次兜底:合计 - 任一 = 另一
|
||
if tax is None and total is not None and pretax is not None:
|
||
tax = round(total - pretax, 2)
|
||
if pretax is None and total is not None and tax is not None:
|
||
pretax = round(total - tax, 2)
|
||
|
||
fields.amount = total
|
||
fields.tax_amount = tax
|
||
fields.amount_pretax = pretax
|
||
fields.amount_cn = extract_cn_amount(raw)
|
||
fields.amount_match = amount_consistent(fields.amount_cn, fields.amount)
|
||
|
||
# 主体
|
||
if lines:
|
||
seller_name, seller_tax, buyer_name, buyer_tax = _extract_parties_from_lines(lines)
|
||
if not (seller_name and buyer_name):
|
||
sn, st, bn, bt = _extract_parties_from_text(raw)
|
||
seller_name = seller_name or sn
|
||
buyer_name = buyer_name or bn
|
||
seller_tax = seller_tax or st
|
||
buyer_tax = buyer_tax or bt
|
||
else:
|
||
seller_name, seller_tax, buyer_name, buyer_tax = _extract_parties_from_text(raw)
|
||
|
||
fields.seller_name = seller_name
|
||
fields.seller_tax_no = seller_tax
|
||
fields.buyer_name = buyer_name
|
||
fields.buyer_tax_no = buyer_tax
|
||
|
||
return fields |